[開源程式碼分享]用Python寫台股交易策略｜股票量化交易從零開始（五）

目錄 Content 隱藏

用 Python 撰寫量化交易策略

步驟一：資料抓取調整

步驟二：策略撰寫前準備

步驟三：策略撰寫

用 Python 撰寫量化交易策略

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

點我了解更多資訊

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

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」檔案。

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

步驟二：策略撰寫前準備

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

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

步驟三：策略撰寫

首先，我們先安裝「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()）。

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

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

同時，在「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...

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

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

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

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

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

量化通粉絲社群，一起討論程式交易！

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

加入臉書社團「程式交易 Taiwan」即時獲取實用的資源！

[開源程式碼分享]用Python寫台股交易策略｜股票量化交易從零開始（五）

用 Python 撰寫量化交易策略

步驟一：資料抓取調整

步驟二：策略撰寫前準備

步驟三：策略撰寫

量化通

MultiCharts教學懶人包，5 步驟無痛上手程式交易

MT4 教學懶人包，避開詐騙，小白也能看懂的EA資源分享

加密貨幣教學懶人包，手把手帶你搞懂加密貨幣如何投資

Python程式交易教學懶人包，學Python量化交易必須知道的5件事

MultiCharts 加密貨幣課程｜比特幣實戰【第三梯次】

MultiCharts 中階課程｜進階短線技術分析模組【第三梯次】

TradingView 懶人包整理，完整教學，指標和策略分享

MultiCharts優惠整理！軟體費用特價與折扣代碼｜TOUCHANCE折扣、程式交易費用比較、中文終身版、券商版、免費試用教學

[2024]TradingView黑色星期五特價隨時更新，快閃限時3折優惠，折扣70%+額外領15美元獎勵與省10%稅金

發佈留言取消回覆

用 Python 撰寫量化交易策略

步驟一：資料抓取調整

步驟二：策略撰寫前準備

步驟三：策略撰寫

量化通

相關文章

發佈留言取消回覆