歡迎來到 程式交易實戰 的第七堂課，還記得上堂課實作的定期定額加碼策略嗎？在股價來回震盪時，定期定額搭配加碼策略能夠更有效地提升投資績效！相信大家對於加碼機制已有更深的了解，今天將延續加碼機制的部分，讓讀者更進一步了解，如果將加碼機制運用到波段策略中結果會如何？
讀完本篇文，您將學會...
- 回測均線策略搭配加碼機制
- 均線策略搭配加碼機制的實單演練
均線策略加碼
均線策略是許多投資人會使用的策略之一，代表著過去一段時間內投資人買進的平均成本，當股價上漲穿過均線時代表這一段期間的投資人解套了，若標的具有題材性、多頭走勢明顯，通常會是不錯的初始進場點，若再搭配加碼策略機制，也許會在多頭走勢中再更加提升獲利績效表現！
績效回測
我們延續 上一篇 的研究標的，一 樣使用 0050 這檔個股，初始本金為 100 萬，策略邏輯如下：
- 第一次進場：當股價首次突破 5 日均線時先買入 50 萬元的部位
- 加碼：當日收盤價創第一次進場以來的新高就加碼 10 萬元
- 出場：跌破 10 日均線時清空所有部位
step1. 取得歷史數據，如下：
# 載入相關套件
import matplotlib.pyplot as plt
from datetime import timedelta
import pandas as pd
import numpy as np
import datetime
import time
from fugle_marketdata import (WebSocketClient, RestClient)
# 輸入行情 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')
step2. 訂定衡量績效的指標公式
def calculate_performance(pl_record):
df = pd.DataFrame(pl_record)
pl_list = list(df['pl'])
return_rate = sum(pl_list) / 1000000
winrate = len([x for x in pl_list if x > 0]) / len(pl_list)
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]))
print("報酬率", round(return_rate*100,2) ,"%")
print("勝率", round(winrate*100,2) ,"%")
print("賺賠比", round(odds,2))
print("獲利因子",round(profit_factor,2))
step3. 撰寫策略回測邏輯
這邊使用 talib 套件來計算均線。當策略出場時就記錄該筆交易的損益和持有部位的數量，之後即可藉由這些資訊來做為損益分析、優化的依據。
from talib.abstract import *
# define backtest function
def run_backtest(plus_position):
df['5MA'] = SMA(df['close'], timeperiod=5)
df['10MA'] = SMA(df['close'], timeperiod=10)
df['last_close'] = df['close'].shift(1) # 前一日 收盤價
df['last_5MA'] = df['5MA'].shift(1) # 前一日 5MA
# 紀錄買進
buy_record_list = []
funds = 1000000
last_funds = 1000000
new_high_price = 0
position = []
pl_record = []
for d in df.iloc:
# 如果突破5日線 且 沒有部位
if d['last_close'] <= d['last_5MA'] and d['close'] > d['5MA'] and len(position) == 0:
# 首次站上5日線就買進10萬
position.append( 500000 / d['close'])
new_high_price = d['close']
funds = funds - 500000
# 加碼機制：如果已經第一次進場了 且還有剩餘的資金
if plus_position == 'yes' and len(position) != 0 and funds >= 100000:
# 條件：檢查是否有創首次買進的創新高
if d['close'] > new_high_price:
position.append( 100000 / d['close'])
new_high_price= d['close']
funds = funds - 100000
# 如果股價跌破 10 日線就全出
if d['close'] < d['10MA'] and len(position) != 0:
funds = funds + d['close'] * sum(position)
pl_record.append({'date': d.name, 'accu_pf': funds ,'pl': funds - last_funds , 'max_position': len(position)})
position.clear()
last_funds = funds
plot_df = pd.DataFrame(pl_record).set_index('date')
plot_df['accu_pf'].plot(figsize=(10,5))
return calculate_performance(pl_record)
回測結果
print('有加碼：')
run_backtest(plus_position='yes')
print('----')
print('無加碼：')
run_backtest(plus_position='no')
print('----')
Response
有加碼：
報酬率 33.84 %
勝率 17.21 %
賺賠比 6.65
獲利因子 1.38
----
無加碼：
報酬率 20.73 %
勝率 18.1 %
賺賠比 5.9
獲利因子 1.3
----
根據上圖的累計損益圖以及回測結果顯示，加入加碼機制後（藍色線圖）的報酬率、賺賠比以及獲利因子都有顯著的提升！此範例為順勢的加碼機制策略，因走勢的確立而提升信心進行加碼， 實驗結果也證實加碼機制的運用能夠有效提升績效！因此，我們進一步進入均線的實單演練。
實作均線自動下單程式
以下我們就將改進版的加碼策略寫成能夠實際下單的程式吧！
step1. 登入富果 API
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)
from fugle_marketdata import (WebSocketClient, RestClient)
# 輸入行情 API KEY
key = 'INPUT_YOUR_API_KEY'
client = RestClient(api_key = key)
stock = client.stock # Stock REST API client
# 讀取設定檔
config = ConfigParser()
config.read('./config.simulation.ini') # 使用模擬金鑰設定檔
# 密碼輸入及登入
sdk = SDK(config)
sdk.login()
step2. 計算 MA 數值
首先取得前 N 日的收盤價和當日開盤價並計算，並包成 function 方便我們後續運用
from datetime import timedelta
def cal_ma(symbol, timeperiod):
# 為了取足夠遠的歷史數據
start_date = datetime.datetime.now().date() - timedelta(days = timeperiod*2)
end_date = datetime.datetime.now().date()
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)
ma_price_list = []
for ohlc in historical_data['data'][:timeperiod-1]:
ma_price_list.append(ohlc['close'])
# 取當日開盤價
today_open_price = stock.intraday.quote(symbol = symbol)['openPrice']
ma_price_list.append(today_open_price)
MA = sum(ma_price_list) / timeperiod
return MA
測試看看 5MA、10MA 目前是多少
print("5MA", cal_ma("0050", 5))
print("10MA", cal_ma("0050", 10))
取得當日的 5MA 及 10MA 後即可開始，我們一樣以回測使用的邏輯來進行交易，因為我們需要知道目前已加碼幾次，所以這邊以整股的方式來示範，每次加碼就買進 1 張。若資金有限，可以將交易盤別（APCode）改為盤中零股（IntradayOdd）。
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 ma_strategy:
def __init__(self, symbol):
self.API_KEY = "INPUT_YOUR_KEY"
self.symbol = symbol
self.ma_5 = cal_ma(symbol, 5)
self.ma_10 = cal_ma(symbol, 10)
# 今日是否已買進
self.buy = False
self.position = 0
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_symbol_inventories(self):
inventories = sdk.get_inventories()
for inventor in inventories:
if inventor['stk_no'] == self.symbol:
return inventor['stk_dats']
return []
# 取得上次買進最高的價格 判斷是否創新高
def cal_max_cost(self):
pos_list = self.get_symbol_inventories()
newlist = sorted(pos_list, key=lambda d: d['price']) # 依據價格由低到高排序
return newlist[-1]['price']
# 取得最新報價
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']
# 突破 5 日線 並且沒有部位(代表是首次) 買進
if now_price > self.ma_5 and self.buy == False:
place_stock_order("Buy", 1, "LimitUp", "Common")
self.buy = True
self.position = 1
# 今日已買進 + 股價創新高 + 就加碼 1 張（考量資金，部位以不超過五張為限）
if self.buy == True and now_price > self.cal_max_cost() and self.position <= 5: # 計算上次買進的最大成本(最高價)
place_stock_order("Buy", 1, "LimitUp", "Common")
self.buy = True
self.position = self.position + 1
# 跌破 10 日線 並且有部位 賣出
if now_price < self.ma_10 and self.position > 0:
place_stock_order("Sell", self.position, "LimitDown", "Common")
self.buy = False
self.position = 0
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 = ma_strategy(symbolId)
strategy.main()
結論
本週我們實作了均線加碼策略的回測及下單程式，在回測中我們也驗證了策略加入加碼機制確實能夠提升績效結果，也示範若要將回測結果進行實單測試應該如何進行！讀者也可藉由此篇文章進一步思考，是否有更適合的加碼策略可以運用或結合到您的原始策略中。加碼機制的介紹也將告一段落，接下來預計會帶大家了解策略精進的另一個大主題：停損停利機制的部分，請大家敬請期待！