【程式交易實戰 07】策略精進 - 均線加碼實作

歡迎來到 程式交易實戰 的第七堂課，還記得上堂課實作的定期定額加碼策略嗎？在股價來回震盪時，定期定額搭配加碼策略能夠更有效地提升投資績效！相信大家對於加碼機制已有更深的了解，今天將延續加碼機制的部分，讓讀者更進一步了解，如果將加碼機制運用到波段策略中結果會如何？

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

• 回測均線策略搭配加碼機制
• 均線策略搭配加碼機制的實單演練

均線策略加碼

均線策略是許多投資人會使用的策略之一，代表著過去一段時間內投資人買進的平均成本，當股價上漲穿過均線時代表這一段期間的投資人解套了，若標的具有題材性、多頭走勢明顯，通常會是不錯的初始進場點，若再搭配加碼策略機制，也許會在多頭走勢中再更加提升獲利績效表現！

績效回測

我們延續 上一篇 的研究標的，一樣使用 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 requests
import datetime
import json
import time

key = "YOUR_TOKEN" # 請輸入您的行情 API Token
symbol = "0050" # 股票代碼

candles_list = []

for y in range(2010,2023):
    start_date = str(y)+"-01-01" # 選擇資料起始日期
    end_date = str(y)+"-12-31"  # 選擇資料結束日期
    # API URL
    url = f"https://api.fugle.tw/marketdata/v0.3/candles?symbolId={str(symbol)}&apiToken={key}&from={start_date}&to={end_date}&fields=open,high,low,close,volume,turnover,change"

    # 取得歷史資料
    historical_data = requests.get(url).json()
    # 將資料轉為 dataframe 方便回測
    candles_list = candles_list + historical_data['data']
    time.sleep(0.5)
    
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_realtime import (HttpClient, WebSocketClient)

# 輸入行情 API Token
key = 'YOUR_TOKEN'
api_client = HttpClient(api_token=key)

# 讀取設定檔
config = ConfigParser()
config.read('./config.simulation.ini') # 使用模擬金鑰設定檔
# 密碼輸入及登入
sdk = SDK(config)
sdk.login()

step2. 計算 MA 數值

首先取得前 N 日的收盤價和當日開盤價並計算，並包成 function 方便我們後續運用

from datetime import timedelta

def cal_ma(timeperiod):
    symbol = "0050"
    # 為了取足夠遠
    start_date = datetime.datetime.now().date() - timedelta(days = timeperiod*2)
    end_date = datetime.datetime.now().date()

    url = f"https://api.fugle.tw/marketdata/v0.3/candles?symbolId={str(symbol)}&apiToken={key}&from={str(start_date)}&to={str(end_date)}"

    # 取得歷史資料
    historical_data = requests.get(url).json()

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

    # 取當日開盤價
    today_open_price = api_client.intraday.quote(symbolId=symbol)['data']['quote']['priceOpen']['price']
    ma_price_list.append(today_open_price)

    MA = sum(ma_price_list) / timeperiod

    return MA

測試看看 5MA、10MA 目前是多少

print("5MA", cal_ma(5))
print("10MA", cal_ma(10))

取得當日的 5MA 及 10MA 後即可開始，我們一樣以回測使用的邏輯來進行交易，因為我們需要知道目前已加碼幾次，所以這邊以整股的方式來示範，每次加碼就買進 1 張。若資金有限，可以將交易盤別（APCode）改為盤中零股（IntradayOdd）。

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, symbol):
        
        self.symbol = symbol
        self.create_ws_quote()
        self.ma_5 = cal_ma(5)
        self.ma_10 = cal_ma(10)
        # 今日是否已買進
        self.buy = False
    
    # 取得相同股票的庫存 當出清部位的訊號產生時就全部清空
    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 _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']

            # 突破 5 日線 並且沒有部位(代表是首次) 買進
            if now_price > self.ma_5 and len(self.get_symbol_inventories()) == 0:

                order = OrderObject(
                    buy_sell=Action.Buy,
                    price_flag=PriceFlag.Market, # 市價買 
                    price='',
                    stock_no=self.symbol,
                    quantity=1,
                    ap_code=APCode.Common,
                    trade=Trade.Cash
                )
                sdk.place_order(order)
                self.buy = True
                
            
            # 跌破 10 日線 並且有部位 賣出
            if now_price < self.ma_10 and len(self.get_symbol_inventories()) > 0:
                order = OrderObject(
                    buy_sell=Action.Sell,
                    price_flag=PriceFlag.Market,
                    price='',
                    stock_no=self.symbol,
                    quantity= len(self.get_symbol_inventories()),
                    ap_code=APCode.Common,
                    trade=Trade.Cash
                )
                sdk.place_order(order)
                
            # 時間是收盤.今日尚未買進.股價創新高.庫存未達到10張 就加碼1張
            if datetime.time(13, 26) >= datetime.datetime.now().time() >= datetime.time(13, 25) and \
                self.buy is False and\
                len(self.get_symbol_inventories()) < 10 and \
                now_price > self.cal_max_cost():  # 計算上次買進的最大成本(最高價)
                
                order = OrderObject(
                    buy_sell=Action.Buy,
                    price_flag=PriceFlag.LimitUp,
                    price='',
                    stock_no=self.symbol,
                    quantity=1,
                    ap_code=APCode.Common,
                    trade=Trade.Cash
                )
                sdk.place_order(order)
                self.buy = True

    def create_ws_quote(self):

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

t = Trade(symbol = "0050")

結論

本週我們實作了均線加碼策略的回測及下單程式，在回測中我們也驗證了策略加入加碼機制確實能夠提升績效結果，也示範若要將回測結果進行實單測試應該如何進行！讀者也可藉由此篇文章進一步思考，是否有更適合的加碼策略可以運用或結合到您的原始策略中。加碼機制的介紹也將告一段落，接下來預計會帶大家了解策略精進的另一個大主題：停損停利機制的部分，請大家敬請期待！