【程式交易實戰 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_marketdata import (WebSocketClient, RestClient)
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. 取得歷史數據

# 輸入行情 API Token
key = 'INPUT_YOUR_API_KEY'
client = RestClient(api_key = key)
stock = client.stock  # Stock REST API client

symbol = "0050" # 股票代碼

candles_list = []

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

    history_options = {"symbol": symbol, "from": start_date, "to": end_date,
                       "timeframe":"D", "fields":"open,high,low,close,volume,change,turnover"}

    historical_data = stock.historical.candles(**history_options)
    # 將資料轉為 dataframe 方便回測
    candles_list = candles_list + historical_data['data']
    
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():

    client = RestClient(api_key = key)
    stock = client.stock  # Stock REST API client

    start_date = datetime.datetime.now().date() - timedelta(days = 60)
    end_date = datetime.datetime.now().date()- timedelta(days = 1)
    
    # 取得歷史資料
    history_options = {"symbol": symbol, "from": start_date, "to": end_date,
                       "timeframe":"D", "fields":"open,high,low,close,volume,change,turnover"}

    historical_data = stock.historical.candles(**history_options)
    
    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):

    client = RestClient(api_key = key)
    stock = client.stock  # Stock REST API client

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

    # 取得歷史資料
    history_options = {"symbol": symbol, "from": start_date, "to": end_date,
                       "timeframe":"D", "fields":"open,high,low,close,volume,change,turnover"}

    historical_data = stock.historical.candles(**history_options)

    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():

    client = RestClient(api_key = key)
    stock = client.stock  # Stock REST API client

    start_date = datetime.datetime.now().date() - timedelta(days = 1)
    end_date = datetime.datetime.now().date()- timedelta(days = 1)
    
    # 取得歷史資料
    history_options = {"symbol": symbol, "from": start_date, "to": end_date,
                       "timeframe":"D", "fields":"open,high,low,close,volume,change,turnover"}

    historical_data = stock.historical.candles(**history_options)

    return historical_data['data'][0]

import json
from fugle_trade.constant import *
from fugle_trade.order import OrderObject
from fugle_trade.sdk import SDK
from datetime import timedelta
from fugle_marketdata import (WebSocketClient, RestClient)

class portfolio_strategy:
    def __init__(self, symbol):

        self.API_KEY = "INPUT_YOUR_KEY"
        self.symbol = symbol
        # 今日是否已買進
        self.buy = False
        self.position = 0
        self.last_macd = get_last_date_macd_hist()
        self.last_ohlc = get_last_ohlc()
        self.open_price = self.get_open_price()

    def get_open_price(self):
        
        client = RestClient(api_key = self.API_KEY)
        stock = client.stock  # Stock REST API client
        return stock.intraday.quote(symbol = self.symbol)["openPrice"]
    
    def place_stock_order(self, buy_sell, qty, PriceFlag_type, APCode_type):

        order = OrderObject(
            buy_sell=Action[buy_sell],
            price_flag=PriceFlag[PriceFlag_type],
            price = '',
            stock_no=self.symbol,
            quantity=qty,
            ap_code=APCode[APCode_type],
            trade=Trade.Cash   # 現股賣出 的交易類別
        )

        self.sdk.place_order(order)

    
    # 取得最新報價
    def get_latest_price(self, message):

        json_data = json.loads(message)

        if json_data['event']=="data" and json_data.get('data',{}).get('isTrial') == None: # 避免用到試撮資料
            # 更新目前價格
            now_price = json_data['data']['price']
            today_open = self.open_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 today_macd * self.last_macd < 0 and today_macd > 0 and self.position == 0:

                    place_stock_order("Buy", 1, "LimitUp", "Common")

                    self.buy = True

                # 翻黑賣出
                elif today_macd * self.last_macd < 0 and today_macd < 0 and self.position != 0:

                    place_stock_order("Sell", inv_len, "LimitDown", "Common")
                

            if now_price > self.last_ohlc['high'] and now_price > today_open and self.position == 0:

                place_stock_order("Buy", 1, "LimitUp", "Common")

                self.buy = True
                
        
            elif now_price < today_open and len(self.get_inventories()) != 0 and self.position != 0:
                    
                place_stock_order("Sell", inv_len, "LimitDown", "Common")
    
    
    def handle_disconnect(self, code, message):
        print(f'disconnect: {code}, {message}')

    def main(self):

        client = WebSocketClient(api_key=self.API_KEY)
        self.stock = client.stock
        self.stock.on('message', self.get_latest_price)
        self.stock.on("disconnect", self.handle_disconnect)

        self.stock.connect()

        self.stock.subscribe({
            'channel': 'trades',
            'symbol': self.symbol
        })
        
if __name__ == '__main__':

    symbolId = "2330"  # 取得股票代碼

    strategy = portfolio_strategy(symbolId)
    strategy.main()

結論

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