caution
內含 fugle_trade 下單套件，該套件已不再更新，下單語法請改用 合作券商 之 SDK
歡迎來到 程式交易實戰 的最後ㄧ堂課，在本系列文章中介紹了許多交易策略，可以發現不同的策略會有不同的特性，您還在尋找策略聖杯嗎？也許多個策略組合能夠有效幫助您降低風險，得到更平穩的報酬，我們不妨試著組合多個策略，進一步打造更平穩的報酬吧！
讀完本篇文，您將學會...
- 透過庫存個股進行移動停損停利實單演練
多策略組合介紹
在程式交易實戰系列文章中，介紹了許多交易策略，可以發現不同的策略各有特色，有些策略是獲利高、回檔大、交易次數多，有些策略是獲利普通、回檔小、交易次數少。通常不同策略所帶來的獲利週期往往也不一致。在開發策略時會希望將策略調整到完美的狀態，但同時也相當容易掉入過度擬和（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()
結論
開發策略時除了精雕細琢一支策略，也應考慮過度擬合的陷阱，不妨考慮以多策略組合的方式來補足單策略的缺點。這種交易方式不僅更簡單，在獲利穩定度、風險都能夠較單一策略有更好的表現，若是能進一步考慮動態調整策略資金配會是更好的解決方案！