【程式交易實戰 10】打造多策略組合，讓你穩穩睡好覺

歡迎來到 程式交易實戰 的最後ㄧ堂課，在本系列文章中介紹了許多交易策略，可以發現不同的策略會有不同的特性，您還在尋找策略聖杯嗎？也許多個策略組合能夠有效幫助您降低風險，得到更平穩的報酬，我們不妨試著組合多個策略，進一步打造更平穩的報酬吧！

讀完本篇文，您將學會...

• 透過庫存個股進行移動停損停利實單演練

多策略組合介紹

在程式交易實戰系列文章中，介紹了許多交易策略，可以發現不同的策略各有特色，有些策略是獲利高、回檔大、交易次數多，有些策略是獲利普通、回檔小、交易次數少。通常不同策略所帶來的獲利週期往往也不一致。在開發策略時會希望將策略調整到完美的狀態，但同時也相當容易掉入過度擬和（Overfitting）的陷阱中，我們可以試著將多個策略組合起來，觀察策略之間是否有互補性，進而降低單一策略容易遇到的潛在風險。

假設有A、B兩支交易策略，他們的損益曲線如下圖，我們可以觀察到：

• 策略A：絕對獲利大，回檔大
• 策略B：絕對獲利小，回檔小

通常我們會認為策略B是一個較差的策略，因為他最後的報酬不理想，但事實上可以觀察到兩個策略的【獲利分布是不一樣的】，可以將兩個策略疊合起來會看得更清楚。可以發現疊合後的獲利雖然不如策略A，但是它的回檔降低了非常多，如此以來我們就可以較有信心使用這個策略組合。

以下我們就一樣以 0050 為標的，試著建構 2 種策略將它們組合起來並討論組合後的效果。

• 策略1：MACD 柱狀翻紅買進1張、翻黑賣出1張
• 策略2：紅 K 並且收盤突破昨日高點就買進1張、收黑K賣出1張。

首先取得 0050 的日 K 歷史資料，回測期間為 2010/1/1 至 2022/12/22！

策略回測

step1. 載入相關套件

# 載入富果 API 相關套件
from fugle_realtime import (HttpClient,WebSocketClient)

from configparser import ConfigParser
from fugle_trade.sdk import SDK

from fugle_trade.order import OrderObject
from fugle_trade.constant import (APCode, Trade, PriceFlag, BSFlag, Action)

# 載入相關套件
import matplotlib.pyplot as plt
import numpy as np
import requests
import pandas as pd
import time

step2. 取得歷史數據

key = "INPUT_YOUR_API_TOKEN" # 您的行情 API Token
symbol = "0050" # 股票代碼

api_client = HttpClient(api_token=key)

candles_list = []

for y in range(2010,2023):
    
    start_date = str(y)+"-01-01" # 選擇資料起始日期
    end_date = str(y)+"-12-31"  # 選擇資料結束日期

    # 取得歷史資料
    historical_data = api_client.historical.candles(symbol, start_date, end_date, None)['data']

    # 將資料轉為dataframe方便回測
    candles_list = candles_list + historical_data
    
    # 不要抓太快避免造成server負擔
    time.sleep(0.5)
    
df = pd.DataFrame(candles_list).sort_values('date').set_index('date')

step3. 策略回測撰寫

策略 1 使用 MACD 指標，這裡引入 talib 函式庫來計算！

import talib

df['MACD'], df['MACDSignal'], df['MACDhist'] = talib.MACD(df['close'],12,26,9)

MACD 指標計算完畢後就可以開始來撰寫策略 1 的內容了。 首先在第一天先買進一張作為初始部位，方便後面的日期繼續回測。另外因為策略內容需要比較前一天的數據，所以需要從第 2 天開始回測。

buy_price = df.iloc[0]['close']
strategy1_record = []

for d in df[1:].iloc:
    # 如果今日的MACD柱跟昨天不同向
    if d['MACDhist'] * df.loc[d.name-1,'MACDhist'] < 0:
        # 翻紅買進
        if d['MACDhist'] > 0:
            buy_price = d['close']
        
        # 翻黑賣出
        elif d['MACDhist'] < 0:
            strategy1_record.append({'date': d['date'], 'profit': d['close'] - buy_price})

當部位出場時會將每次的損益都記錄在 strategy1_record 裡，有了每一筆損益後就可以畫出累積損益圖，繪製如下：

import matplotlib.pyplot as plt

strategy1_plotdf = pd.DataFrame(strategy1_record).set_index('date')
strategy1_plotdf['profit_cumsum'] = strategy1_plotdf['profit'].cumsum()
strategy1_plotdf['profit_cumsum'].plot(figsize=(15,8))

我們可以觀察到主要的獲利都是 2020 年至 2022 年之間，過程中也有數次較大的回檔。 以下我們再繼續實作策略 2，一樣在第一天就先買進 1 張，並從第二天開始回測。

buy_price = df.iloc[0]['close']
position = 1
strategy2_record = []

for d in df[1:].iloc:
    # 如果沒有部位 當天是紅K 並且收過昨天高點
    if position == 0 and d['close'] > d['open'] and d['close'] > df.loc[d.name-1,'high'] :
        position = 1
        buy_price = d['close']
        
    # 如果有部位 當天是黑K
    elif position == 1 and d['close'] < d['open']  :
        position = 0
        strategy2_record.append({'date': d['date'], 'profit':   d['close'] - buy_price})

累積損益圖如下：

我們可以觀察到這組策略的獲利不如策略 1，在 2022 年的回檔也很大，但值得注意的是它的獲利分布比較廣，有助於補足策略 1 在 2012-2018 年績效較普通的情況。於是以下我們就將兩組策略的累積損益圖疊加後來進行比較！

由上圖就可以觀察到總合兩個策略可以達成縮小回檔、增加獲利穩定性的目標，較穩定的策略才有助於我們有足夠的信心來繼續執行這個策略。

def calculate_performance(pl_record):
    
    df = pd.DataFrame(pl_record)
    df['profit_cumsum'] = df['profit'].cumsum()
    pl_list = list(df['profit'])
    
    winrate = len([x for x in pl_list if x > 0]) / len([x for x in pl_list if x <= 0])
    odds = ( sum([x for x in pl_list if x > 0]) / len([x for x in pl_list if x > 0]) ) / abs( sum([x for x in pl_list if x <= 0]) / len([x for x in pl_list if x <= 0]) )
    profit_factor = sum([x for x in pl_list if x > 0]) / abs(sum([x for x in pl_list if x <= 0]))
    df['drawdown'] = df['profit_cumsum'] - df['profit_cumsum'].cummax()
    maximum_drawdown = np.min(df["drawdown"])
    
    print("勝率", round(winrate*100,2) ,"%")
    print("賺賠比", round(odds,2))
    print("獲利因子",round(profit_factor,2))
    print('MDD', round(maximum_drawdown,2))

print("策略1：")
calculate_performance(strategy1_record)
print("----------------")
print("策略2：")
calculate_performance(strategy2_record)
print("----------------")
print("策略組合：")
combine_dict_list =[]
for d in combine_df.iloc:
    combine_dict_list.append({'date': d.name, 'profit': d['total_avg']})
    
calculate_performance(combine_dict_list)

Response

策略1：
勝率 89.23 %
賺賠比 2.07
獲利因子 1.84
MDD -11.4
----------------
策略2：
勝率 74.24 %
賺賠比 1.52
獲利因子 1.13
MDD -25.45
----------------
策略組合：
勝率 79.31 %
賺賠比 1.73
獲利因子 1.37
MDD -15.53

以評估指標來檢視三種策略可以發現，組合策略皆是介於策略 1 與策略 2 之間，較可惜的是組合策略的 MDD 還是大於策略 1，由損益圖可以發現主要是在 2022 年受到策略 2 的影響，若能調整策略資金配置比例或許是解決方法之一，這也可以是讀者延伸研究的方向。除此之外在 2012-2018 年組合策略都有較好的穩定度！

交易實作

step1. 策略 function 訂定

def get_last_date_macd_hist():

    api_client = HttpClient(api_token=key)

    start_date = datetime.datetime.now().date() - timedelta(days = 60)
    end_date = datetime.datetime.now().date()- timedelta(days = 1)
    # 取得歷史資料
    historical_data = api_client.historical.candles(symbol, start_date, end_date, None)
    
    close_price_list = []
    
    for ohlc in historical_data['data'][::-1]:
        close_price_list.append(ohlc['close'])

    MACD, MACDSignal,MACDhist = talib.MACD(np.array(close_price_list),12,26,9)
    
    return MACDhist[-1]
    

def get_today_macd_hist(now_price):

    api_client = HttpClient(api_token=key)

    start_date = datetime.datetime.now().date() - timedelta(days = 60)
    end_date = datetime.datetime.now().date()- timedelta(days = 1)

    # 取得歷史資料
    historical_data = api_client.historical.candles(symbol, start_date, end_date, None)

    close_price_list = []
    
    for ohlc in historical_data['data'][::-1]:
        close_price_list.append(ohlc['close'])
    close_price_list.append(now_price)

    MACD, MACDSignal,MACDhist = talib.MACD(np.array(close_price_list),12,26,9)
    
    return MACDhist[-1]

step2. 下單實作

def get_last_ohlc():

    api_client = HttpClient(api_token=key)

    start_date = datetime.datetime.now().date() - timedelta(days = 1)
    end_date = datetime.datetime.now().date()- timedelta(days = 1)
    
    # 取得歷史資料
    historical_data = api_client.historical.candles(symbol, start_date, end_date, None)

    return historical_data['data'][0]

from fugle_realtime import WebSocketClient
import json
from fugle_trade.constant import *
from fugle_trade.order import OrderObject
from fugle_trade.sdk import SDK

class Trade:
    def __init__(self):
        
        self.create_ws_quote()
        # 今日是否已買進
        self.buy = False
        self.last_macd = get_last_date_macd_hist()
        self.last_ohlc = get_last_ohlc()

        
    # 取得相同股票的庫存 當出清部位的訊號產生時就全部清空
    def get_inventories(self):
        inventories = sdk.get_inventories()
        
        for inventor in inventories:
            if inventor['stk_no'] == symbol:
                return inventor['stk_dats']
        return []
    
    def _on_new_price(self, message):
        json_data = json.loads(message)

        if json_data['data']['info']['type'] == "EQUITY":
            # 更新目前價格
            now_price = json_data['data']['quote']['trade']['price']
            today_open = json_data['data']['quote']['priceOpen']['price']
  
                
            # 時間是收盤.今日尚未買進
            if datetime.time(13, 26) >= datetime.datetime.now().time() >= datetime.time(13, 25) and \
                self.buy is False:

                today_macd = get_today_macd_hist(now_price)

                # 翻紅買進
                if self.today_macd * self.last_macd < 0 and self.today_macd > 0 and len(self.get_inventories()) == 0:

                    order = OrderObject(
                        buy_sell=Action.Buy,
                        price_flag=PriceFlag.LimitUp,
                        price='',
                        stock_no=symbol,
                        quantity=1,
                        ap_code=APCode.Common,
                        trade=Trade.Cash
                    )
                    sdk.place_order(order)
                    self.buy = True


                # 翻黑賣出
                elif self.today_macd * self.last_macd < 0 and self.today_macd < 0 and len(self.get_inventories()) != 0:
                    order = OrderObject(
                        buy_sell=Action.Sell,
                        price_flag=PriceFlag.LimitDown,
                        price='',
                        stock_no=symbol,
                        quantity= len(self.get_inventories()),
                        ap_code=APCode.Common,
                        trade=Trade.Cash
                    )
                    sdk.place_order(order)

            if now_price > self.last_ohlc['high'] and now_price > today_open and len(self.get_inventories()) == 0:
                                    order = OrderObject(
                        buy_sell=Action.Buy,
                        price_flag=PriceFlag.LimitUp,
                        price='',
                        stock_no=symbol,
                        quantity=1,
                        ap_code=APCode.Common,
                        trade=Trade.Cash
                    )
                    sdk.place_order(order)
                    self.buy = True

            elif now_price < today_open and len(self.get_inventories()) != 0 and len(self.get_inventories()) != 0:
                    order = OrderObject(
                        buy_sell=Action.Sell,
                        price_flag=PriceFlag.LimitDown,
                        price='',
                        stock_no=symbol,
                        quantity= len(self.get_inventories()),
                        ap_code=APCode.Common,
                        trade=Trade.Cash
                    )
                    sdk.place_order(order)

    def create_ws_quote(self):

        ws_client = WebSocketClient(api_token=key)
        ws = ws_client.intraday.quote(symbolId=symbol, on_message=self._on_new_price)
        ws.run_async()
        time.sleep(1)

t = Trade()

結論

開發策略時除了精雕細琢一支策略，也應考慮過度擬合的陷阱，不妨考慮以多策略組合的方式來補足單策略的缺點。這種交易方式不僅更簡單，在獲利穩定度、風險都能夠較單一策略有更好的表現，若是能進一步考慮動態調整策略資金配會是更好的解決方案！