Shopping Cart

購物車內沒有任何商品。

[程式碼分享]用Python寫台股交易策略，基礎的策略架構範例｜股票量化交易從零開始（五）

Masterlink 6 101307181123
目錄 Content 隱藏
用 Python 撰寫量化交易策略
步驟一：資料抓取調整
步驟二：策略撰寫前準備
步驟三：策略撰寫

用 Python 撰寫量化交易策略

在接下來的文章，我們會開始撰寫我們的第一支策略，並且在頁面上顯示你的回測結果。那麼在開始撰寫策略之前，我們會有一些事前準備，因為我們需要準備歷史的股票數據才能分析回測。

立即訂閱電子報，掌握最新資訊

立即訂閱電子報，掌握最新資訊！

    稱呼

    電子郵件

    以下非必填，但若您願意分享，我們將能推送更精準的內容給您

    投資經驗

    是否為理工科背景、工程師或有寫程式的經驗？

    有興趣的主題
    量化交易台股期貨海外期貨虛擬貨幣美股

    有興趣的量化交易軟體/平台
    不清楚MultiChartsTradingViewPythonXQ

    想透過量化交易達成甚麼目的？
    不確定自動交易選股回測投資績效量化自己的投資方法想找現成的策略套用

    還有什麼想詢問的？

    20241111 1920 400
    20241111 978 258

    【2024】台灣虛擬貨幣交易所排名，手續費折扣整理
    【2024】全球虛擬貨幣交易所排名，手續費折扣整理

    因此，請先回到我們上一篇下載歷史資料的地方，將開源程式碼修改為以下範例：

    from tech_analysis_api_v2.api import TechAnalysis
from tech_analysis_api_v2.model import *


import threading
import pandas as pd
from datetime import datetime, timedelta
import mplfinance as mpf


def OnDigitalSSOEvent(aIsOK, aMsg):
    print(f'OnDigitalSSOEvent: {aIsOK} {aMsg}')


def OnTAConnStuEvent(aIsOK):
    print(f'OnTAConnStuEvent: {aIsOK}')
    if aIsOK:
        event.set()


def OnUpdate(ta_Type: eTA_Type, aResultPre, aResultLast):
   
    if aResultPre != None:
        if ta_Type == eTA_Type.SMA:
            print(f'前K {str(aResultPre)}')
        if ta_Type == eTA_Type.EMA:
            print(f'前K {str(aResultPre)}')
        if ta_Type == eTA_Type.WMA:
            print(f'前K {str(aResultPre)}')
        if ta_Type == eTA_Type.SAR:
            print(f'前K {str(aResultPre)}')
        if ta_Type == eTA_Type.RSI:
            print(f'前K {str(aResultPre)}')
        if ta_Type == eTA_Type.MACD:
            print(f'前K {str(aResultPre)}')
        if ta_Type == eTA_Type.KD:
            print(f'前K {str(aResultPre)}')
        if ta_Type == eTA_Type.CDP:
            print(f'前K {str(aResultPre)}')
    if aResultLast != None:
        if ta_Type == eTA_Type.SMA:
            print(f'最新 Time:{aResultLast.KBar.TimeSn_Dply}, SMA:{aResultLast.Value}')
        if ta_Type == eTA_Type.EMA:
            print(f'最新 Time:{aResultLast.KBar.TimeSn_Dply}, EMA:{aResultLast.Value}')
        if ta_Type == eTA_Type.WMA:
            print(f'最新 Time:{aResultLast.KBar.TimeSn_Dply}, EMA:{aResultLast.Value}')
        if ta_Type == eTA_Type.SAR:
            print(f'最新 Time:{aResultLast.KBar.TimeSn_Dply}, SAR:{aResultLast.SAR}, EPh:{aResultLast.EPh}, EPl:{aResultLast.EPl}, AF:{aResultLast.AF}, RaiseFall:{aResultLast.RaiseFall}')
        if ta_Type == eTA_Type.RSI:
            print(f'最新 Time:{aResultLast.KBar.TimeSn_Dply}, RSI:{aResultLast.RSI}, UpDn:{aResultLast.UpDn}, UpAvg:{aResultLast.UpAvg}, DnAvg:{aResultLast.DnAvg}')
        if ta_Type == eTA_Type.MACD:
            print(f'最新 Time:{aResultLast.KBar.TimeSn_Dply}, DIF:{aResultLast.DIF}, OSC:{aResultLast.OSC}')
        if ta_Type == eTA_Type.KD:
            print(f'最新 Time:{aResultLast.KBar.TimeSn_Dply}, K:{aResultLast.K}, D:{aResultLast.D}')
        if ta_Type == eTA_Type.CDP:
            print(f'最新 Time:{aResultLast.KBar.TimeSn_Dply}, CDP:{aResultLast.CDP}, AH:{aResultLast.AH}, NH:{aResultLast.NH}, AL:{aResultLast.AL}, NL:{aResultLast.NL}')
       
def OnRcvDone(ta_Type: eTA_Type, aResult):
    if ta_Type == eTA_Type.SMA:
        for x in aResult:
            print(f'回補 {x}')
    if ta_Type == eTA_Type.EMA:
        for x in aResult:
            print(f'回補 {x}')
    if ta_Type == eTA_Type.WMA:
        for x in aResult:
            print(f'回補 {x}')
    if ta_Type == eTA_Type.SAR:
        for x in aResult:
            print(f'回補 {x}')
    if ta_Type == eTA_Type.RSI:
        for x in aResult:
            print(f'回補 {x}')
    if ta_Type == eTA_Type.MACD:
        for x in aResult:
            print(f'回補 {x}')
    if ta_Type == eTA_Type.KD:
        for x in aResult:
            print(f'回補 {x}')
    if ta_Type == eTA_Type.CDP:
        for x in aResult:
            print(f'回補 {x}')
       
def option():
    ProdID = input("商品代號: ")
    SNK = input("分K(1/3/5): ")
    STA_Type = input("指標(SMA/EMA/WMA/SAR/RSI/MACD/KD/CDP): ")
    DateBegin = input("日期(ex: 20230619): ")


    NK = eNK_Kind.K_1m
    if SNK == '1':
        NK = eNK_Kind.K_1m
    elif SNK == '3':
        NK = eNK_Kind.K_3m
    elif SNK == '5':
        NK = eNK_Kind.K_5m


    TA_Type = eTA_Type.SMA
    if STA_Type == 'SMA':
        TA_Type = eTA_Type.SMA
    elif STA_Type == 'EMA':
        TA_Type = eTA_Type.EMA
    elif STA_Type == 'WMA':
        TA_Type = eTA_Type.WMA
    elif STA_Type == 'SAR':
        TA_Type = eTA_Type.SAR
    elif STA_Type == 'RSI':
        TA_Type = eTA_Type.RSI
    elif STA_Type == 'MACD':
        TA_Type = eTA_Type.MACD
    elif STA_Type == 'KD':
        TA_Type = eTA_Type.KD
    elif STA_Type == 'CDP':
        TA_Type = eTA_Type.CDP


    return TechAnalysis.get_k_setting(ProdID, TA_Type, NK, DateBegin)

event = threading.Event()

def fetch_historical_data(prod_id, start_date, end_date):
    """
    根據提供的日期範圍抓取歷史成交資料。
    """
    ta = TechAnalysis(OnDigitalSSOEvent, OnTAConnStuEvent, OnUpdate, OnRcvDone)
    ta.Login('H124418422', 'steven123')  # Replace with your credentials
    event.wait()


    all_data = []
   
    # 生成日期範圍
    for single_date in pd.date_range(start=start_date, end=end_date):
        formatted_date = single_date.strftime("%Y%m%d")
        lsBS, sErrMsg = ta.GetHisBS_Stock(prod_id, formatted_date)
        if sErrMsg:
            print(f"Error on {formatted_date}: {sErrMsg}")
        else:
            for x in lsBS:
                data_point = {
                    'ProdID': x.Prod,
                    'Match_Time': x.Match_Time,
                    'Match_Price': x.Match_Price,
                    'Match_Quantity': x.Match_Quantity,
                    'Is_TryMatch': x.Is_TryMatch,
                    'BS': x.BS,
                    'Date': single_date  # 在每條資料中新增日期
                }
                all_data.append(data_point)
   
    df = pd.DataFrame(all_data)
    df['Formatted_Time'] = df.apply(lambda row: convert_time(row['Match_Time'], row['Date']), axis=1)
    return df






def convert_time(time_val, date_val):
    # 解析日期
    year = date_val.year
    month = date_val.month
    day = date_val.day
    # 轉成STR比較好處理
    time_str = str(time_val)
     
    time_str = time_str[:6]  
    if '.' in time_str:
        time_str = time_str.replace('.', '')
        time_str = '0' + time_str


    # 解析時間 小時、分鐘、秒
    hours = int(time_str[:2])
    minutes = int(time_str[2:4])
    seconds = int(time_str[4:6])
    # 返回datetime物件
    return datetime(year, month, day, hours, minutes, seconds)


def aggregate_to_ohlc(df, time_frame='5T'):
    """
    將給定的DataFrame轉換為指定時間幀的OHLC數據，並使用最後一個有效時間點的數據填充空白時間段。


    參數:
    df -- 原始的DataFrame。
    time_frame -- 要聚合的時間幀，默認為'5T'（五分鐘）。


    返回:
    轉換後的DataFrame，包含OHLC和總交易量。
    """
    df['Formatted_Time'] = pd.to_datetime(df['Formatted_Time'])
    df.set_index('Formatted_Time', inplace=True)


    # 定義聚合成OHLC的規則
    ohlc_dict = {
        'Match_Price': 'ohlc',
        'Match_Quantity': 'sum'
    }


    # 聚合資料
    df_ohlc = df.resample(time_frame).apply(ohlc_dict)
    df_ohlc.columns = df_ohlc.columns.droplevel(0)  # 移除多级列名


    # 13:25 並不會有交易，所以使用前一筆的資料填補
    df_ohlc.fillna(method='pad', inplace=True)
   
    # Remove data not between 9:00-13:30
    df_ohlc = df_ohlc.between_time('09:00', '13:30')
    # remove data from holidays
    df_ohlc = df_ohlc[df_ohlc.index.dayofweek < 5]


    return df_ohlc




def main():
    end_date = pd.to_datetime("today")
    start_date = end_date - pd.DateOffset(months=1)


    df = fetch_historical_data(prod_id='2454',start_date=start_date, end_date=end_date)
    if df is not None:
        df_ohlc = aggregate_to_ohlc(df, time_frame='5T')
        print("---------印出OHLC資料---------")
        print(df_ohlc)
    if df_ohlc is not None:
        # 確保你的 DataFrame 索引是日期時間型別(datetime)
        df_ohlc.index = pd.to_datetime(df_ohlc.index)


        # 交易量欄位重新命名為 volume (mplfinance 預設欄位名稱)
        df_ohlc.rename(columns={'Match_Quantity': 'volume'}, inplace=True)
        # 儲存為CSV檔案
        # df_ohlc.to_csv("2330_ohlc.csv")
        df_ohlc.to_csv("2454_ohlc.csv")
        print("OHLC資料已儲存為CSV檔案")
        mpf.plot(df_ohlc, type='candle', style='charles',
                title='245', volume=True)
main()

    步驟一：資料抓取調整

    關於上述的程式碼改動，基本上就是抓取比更多天的資料，並且去除掉不必要的資料，以及將資料輸出成 CSV 檔案。

    所以開始運行後，我們會有一個「2454_ohlc.csv」檔案。

    元富5 1

    這個檔案是過去一個月，聯發科每五分鐘開高收低（ohlc）的價格，整理在一起的資料。

    Py 101209161710
    Py 101209161711

    步驟二：策略撰寫前準備

    接下我們回到第一篇和第二篇文章所建立的環境底下，建立一個名為「first_strategy」的資料夾，並且在底下新增一個「data」供我們放置資料，以及新建一個「strategy.py」的檔案。

    新增完成後，待會我們會在這裡撰寫我們的第一支策略。

    元富5 2

    步驟三：策略撰寫

    首先，我們先安裝「backtesting library」。

    pip install backtesting

    接下來我們打開「strategy.py」貼入以下程式碼，寫一隻交叉均線的策略。

    import pandas as pd
from backtesting import Backtest, Strategy
from backtesting.lib import crossover
from backtesting.test import SMA


class SmaCross(Strategy):
    n1 = 2 # 短期均線天數
    n2 = 10 # 長期均線天數
    stop_loss_percentage = 0.02  # 停損點設為入場價格的 2%
    take_profit_percentage = 0.04  # 利潤目標設為入場價格的 4%
    entry_price = 0  # 用於紀錄進場價格


    def init(self):
        close = self.data.Close
        self.sma1 = self.I(SMA, close, self.n1)
        self.sma2 = self.I(SMA, close, self.n2)


    def next(self):
        if not self.position:
            if crossover(self.sma1, self.sma2):
                self.buy()
                self.entry_price = self.data.Close[-1]  # 更新進場價格
            elif crossover(self.sma2, self.sma1):
                self.sell()
                self.entry_price = self.data.Close[-1]  # 更新進場價格
        else:
            stop_loss = self.entry_price * (1 - self.stop_loss_percentage)
            take_profit = self.entry_price * (1 + self.take_profit_percentage)


            # 檢查並執行停損或達到利潤目標
            if self.data.Close[-1] <= stop_loss or self.data.Close[-1] >= take_profit:
                self.position.close()
                self.entry_price = 0  # 重設進場價格


# 載入資料
df = pd.read_csv('data/2454_ohlc.csv', index_col='Formatted_Time', parse_dates=True)
df.columns = [col.capitalize() for col in df.columns]


# 設定 Backtest df是資料，SmaCross是策略, cash是初始資金, commission是手續費
bt = Backtest(df, SmaCross, cash=1000000, commission=.002)


# 執行你的策略
output = bt.run()
print(output)


# 繪圖展示執行結果
bt.plot()

    關於上述的程式碼，它是一個基於簡單移動平均線（SMA）交叉的交易策略。

    移動平均線是一種常用的技術分析指標，用於平滑價格數據並識別趨勢的方向。而在這個策略中，使用了兩條不同時間範圍的移動平均線，具體進出場邏輯如下：

    移動平均線的設定：

    • n1 和 n2 分別是兩條移動平均線的時間範圍。n1 為 2 期，n2 為 5 期。
    • self.sma1 是短期移動平均線（2 期 SMA），self.sma2 是長期移動平均線（5 期 SMA）。

    進場邏輯:

    • 當短期移動平均線（self.sma1）從下方穿越長期移動平均線（self.sma2）時，表示短期趨勢向上，策略會執行買入操作（self.buy()）。

    出場邏輯:

    • 當短期移動平均線（self.sma1）從上方穿越長期移動平均線（self.sma2）時，表示短期趨勢向下，策略會執行賣出操作（self.sell()）。

    這種基於移動平均線交叉的策略，是假設市場的趨勢將會延續，並試圖通過捕捉這些趨勢變化來賺錢。

    一般來說，短期移動平均線相對於長期移動平均線更敏感，能夠更快地反映價格變動。所以當短期線穿越長期線時，它可能表明市場趨勢的轉變，從而觸發買入或賣出信號。

    元富5 3

    同時，在「terminal」 也可以看到這隻策略相關的交易報告：

    Start                     2023-10-24 09:00:00
End                       2023-11-23 13:30:00
Duration                     30 days 04:30:00
Exposure Time [%]                   93.517787
Equity Final [$]                  1113824.842
Equity Peak [$]                   1118620.842
Return [%]                          11.382484
Buy & Hold Return [%]               13.970588
Return (Ann.) [%]                  207.144368
Volatility (Ann.) [%]               94.571445
Sharpe Ratio                         2.190348
Sortino Ratio                       13.515254
Calmar Ratio                        36.809133
Max. Drawdown [%]                   -5.627526
Avg. Drawdown [%]                   -0.780897
Max. Drawdown Duration       17 days 23:50:00
Avg. Drawdown Duration        0 days 22:37:00

Trades 7 
Win Rate [%]                        71.428571
Best Trade [%]                       3.989552
Worst Trade [%]                     -4.341864
Avg. Trade [%]                       1.552014
Max. Trade Duration          10 days 23:20:00
Avg. Trade Duration           4 days 06:28:00
Profit Factor                        3.511277
Expectancy [%]                       1.590944
SQN                                  1.335834
_strategy                            SmaCross
_equity_curve                             ...
_trades                      Size  EntryBa...

    關於交易報告提供相關解釋如下：

    1. Start/End/Duration: 這顯示了策略測試的時間範圍。從2023年10月24日開始，到2023年11月23日結束，持續了30天4小時30分鐘。
    2. Exposure Time [%]: 這表示在整個測試期間中，您的資金有多少時間是處於市場中的（即開倉狀態）。在這個例子中，大約93.52%的時間您的資金是處於投資狀態的。
    3. Equity Final [$]: 在測試期結束時，您的帳戶價值是1,113,824.84。
    4. Equity Peak [$]: 在測試期間內，帳戶價值達到的最高點是1,118,620.84。
    5. Return [%]: 您的總回報率是11.38%。
    6. Buy & Hold Return [%]: 如果您在測試期間開始時買入並持有到結束，您的回報率會是13.97%。
    7. Return (Ann.) [%]: 這是年化回報率，如果這種表現持續一年，預計回報率會是207.14%。
    8. Volatility (Ann.) [%]: 年化波動率是94.57%，表明策略的收益波動相對較大。
    9. Sharpe Ratio: 夏普比率是2.19，這是衡量每單位風險所獲得超額回報的指標。大於1通常被認為是良好的。
    10. Sortino Ratio, Calmar Ratio: 這些都是風險調整後的回報率指標，數值越高表示策略風險越低。
    11. Max./Avg. Drawdown [%]: 最大回撤是-5.63%，平均回撤是-0.78%。這表示策略在達到峰值後可能會虧損的最大比例。
    12. Max./Avg. Drawdown Duration: 最大回撤持續時間約18天，平均回撤持續時間約1天。
    13. Trades: 總共進行了7次交易。
    14. Win Rate [%]: 勝率為71.43%，表示大約71%的交易是盈利的。
    15. Best/Worst Trade [%]: 最好的一筆交易盈利約3.99%，而最差的一筆虧損約4.34%。
    16. Avg. Trade [%], Max./Avg. Trade Duration: 平均每筆交易盈利1.55%，最長的一筆交易持續了接近11天，平均每筆交易持續約4天。
    17. Profit Factor: 利潤因子是3.51，表示獲利交易與虧損交易的比例。一般大於1表示策略是盈利的。
    18. Expectancy [%]: 預期每筆交易平均可獲得1.59%的回報。
    19. SQN: 系統品質數字（System Quality Number）是1.34，這是一個衡量交易系統性能的指標，通常大於1.5被認為是好的。

    總結來說，以過去一個月的表現來看，這可以算是一隻不錯的策略。

    透過上述的步驟，這樣我們就完成交易策略的撰寫與回測了！但請注意，如果要將此策略套用在實戰交易中，建議還是要把回測區間拉大以及嘗試各種參數！

    【系列閱讀】元富證券系列文章

    加入LINE社群「量化交易討論群」無壓力討論與分享！

    加入Discord 「量化交易討論群」即時獲取實用的資源！

    Write Together 101306261122
    Write Together 101306261121
    量化通
    量化通

    量化通是個致力於全民量化金融教育的社群，我們希望透過由淺入深的內容，帶領大家以正確觀念來實踐自動化的金融投資研究分析。

    文章: 209

    相關文章

    發佈留言

    發佈留言必須填寫的電子郵件地址不會公開。 必填欄位標示為 *