歡迎來到 程式交易實戰 的第六堂課，還記得上堂課實作的定期定額策略嗎？在股價來回震盪時，定期定額操作無疑是最便於持續參與市場的策略之一，但有沒有可能可以再提升定期定額的績效呢？今天將繼續透過定期定額的實例應用跟大家分享，如果將定期定額策略加入加減碼機制結果會如何？
讀完本篇文，您將學會...
- 了解加減碼重要性
- 定期定額搭配加減碼實單演練
加減碼的重要性
在實作之前，我們先來介紹加減碼如何進行：
加減碼主要分為順勢加碼及逆勢加碼，順勢加碼是指在股票上漲時，持續分批買進股票，屬於動能交易者常使用的策略之ㄧ。逆勢加碼則是在股票下跌時，持續分批買進，通常是看好該公司的長期成長性，只是短時間受系統性風險影響。讀者可以根據自己的交易習慣來選擇最適合自己的方式。
定期定額搭配加碼策略
在上週的課程中我們實作了簡單的定期定額策略，發現只要有紀律的定期買進就能在上漲行情發生時加速獲利。那若長期持有的股票，在行情下跌時加碼買進，長期來看是否會有更高的獲利呢? 以下我們就以台灣 50 ETF（0050.tw）做為測試的標的，比較兩種策略的報酬情況：
- A策略: 原始策略，每月 5 日買進 5000 元
- B策略: 除了原始策略外，若該交易日漲跌幅超過 1%，再另外以收盤價加碼 5000 元，考量資金量有限，一個月最多加碼三次。
績效回測
我們延續 上一篇 的實驗結果，使用研究標的 0050 每月 5 日進行實驗：
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. 因 0050 是會固定配息的 ETF，在計算報酬時，也把股利納入計算，以下為取得股利資料，因目前尚未提供股利資料 API，您可以下載 此資料 進行實作！
dividend_df = pd.read_csv(r'0050歷史股利.csv')
for i in dividend_df.iloc:
dividend_df.loc[i.name, 'date'] = str(datetime.datetime.strptime(i['date'], "%Y/%m/%d").date())
dividend_df = dividend_df.set_index('date')
step3. 跑回測結果，可自行調整回測參數，包含加碼上限次數 (max_overweight_limit)、預計單筆買進金額 (each_overweight_amount)、跌幅 (decline_ratio)
# 回測
def run_backtest(max_overweight_limit, each_overweight_amount, decline_ratio):
temp_date_list = []
# 產生每月 5 日的列表
for y in range(2010,2023,1):
for m in range(1,13,1):
temp_date_list.append(str(y).zfill(4)+"-"+str(m).zfill(2)+"-05")
buy_date_list = []
# 先計算預定買進的日期
for date in temp_date_list:
if date not in df.index:
count = 0
while date not in df.index and count <30:
date = str(datetime.datetime.strptime(date, "%Y-%m-%d").date() + timedelta(days=1))
# 如果連續找了 30 天沒找到，代表資料已經到底了
count = count + 1
# 如果順利找到
if count < 30:
buy_date_list.append(date)
else:
buy_date_list.append(date)
# print(buy_date_list)
# 本月是否已加碼 N 次
overweight_count = 0
# 計算累積持有股數
total_position = 0
# 計算累積成本
total_cost = 0
# 買賣紀錄
buy_record_list = []
# 報酬率比較
compare_return = []
for price in df.iloc:
# 5 號買進
if price.name in buy_date_list:
# 重置加碼計數器
overweight_count = 0
total_position = total_position + each_overweight_amount/price['close']
total_cost = total_cost + each_overweight_amount
buy_record_list.append({"date":datetime.datetime.strptime(price.name, "%Y-%m-%d").date(),
"buy_qty":each_overweight_amount/price['close'],
'price':price['close'],
'total_position': total_position,
'total_cost': total_cost})
# 本月還沒加碼到 3 次
elif overweight_count < max_overweight_limit and price['close'] < (price['close']-price['change']) * ( 1 -decline_ratio):
total_position = total_position + each_overweight_amount/price['close']
total_cost = total_cost + each_overweight_amount
buy_record_list.append({"date":datetime.datetime.strptime(price.name, "%Y-%m-%d").date(),
"buy_qty":each_overweight_amount/price['close'],
'price':price['close'],
'total_position': total_position,
'total_cost': total_cost })
overweight_count = overweight_count + 1
# 如果是除息時間
if price.name in dividend_df.index:
# 這裡將總成本扣掉配息的金額，但也有其他算法是不把股利的部分計算在報酬率中，可依讀者自行決定
total_cost = total_cost - total_position*dividend_df.loc[price.name,'dividend']
buy_record_df = pd.DataFrame(buy_record_list).sort_index().set_index('date')
# 計算持有股數市值
buy_record_df['market_value'] = buy_record_df['total_position'] * buy_record_df['price']
# 計算報酬率
buy_record_df['return_rate'] = (buy_record_df['market_value'] / buy_record_df['total_cost'] - 1 ) * 100
compare_return.append(buy_record_df['return_rate'])
return compare_return
run_backtest(0,5000,0.01)[0].plot(label='normal')
run_backtest(3,5000,0.01)[0].plot(label='overweight 3 Times * 5000')
plt.legend()
我們可以發現加碼是有效果的!尤其是在 2016-2021 年走勢屬於多頭的時期，加碼的報酬皆領先一般定期定額投入，雖然今年看似不適用加碼機制，但相信幾年後回來看，這段時間點，也許就是讓報酬提升的關鍵點！
透過實驗結果進行定期定額自動下單
首先請將設定檔及憑證檔移到相對應的路徑並輸入登入密碼及憑證密碼！
from configparser import ConfigParser
from fugle_realtime import HttpClient
from fugle_trade.sdk import SDK
from fugle_trade.order import OrderObject
from fugle_trade.constant import (APCode, Trade, PriceFlag, BSFlag, Action)
import time
# 輸入行情 API Token
key = 'YOUR_TOKEN'
api_client = HttpClient(api_token=key)
# 讀取設定檔
config = ConfigParser()
config.read('./config.simulation.ini') # 使用模擬金鑰設定檔
# 密碼輸入及登入
sdk = SDK(config)
sdk.login()
接下來為交易邏輯的部分，在觸發加碼時計算可以買進的股數
# 取得目前即時股價 並推算可以買進的股數
def cal_buy_qty(symbol_id, invest_amount):
quote_data = api_client.intraday.quote(symbolId=symbol_id)
last_price = quote_data['data']['quote']['trade']['price']
return int(invest_amount / last_price)
# 取得開盤參考價 並計算現在是否已下跌 1%
def cal_strategy(symbol_id):
quote_data = api_client.intraday.quote(symbolId=symbol_id)
meta_data = api_client.intraday.meta(symbolId=symbol_id) # 取得開盤參考價
return meta_data['data']['meta']['priceReference'] * (1-0.01) >= quote_data['data']['quote']['trade']['price']
透過帳務 API 檢查本月是否已經買進 3 次
def check_inventories(symbol_id, year_month):
inventories = sdk.get_inventories()
year_month_count = 0
for inventory in inventories:
if inventory['stk_no'] == symbol_id:
for stk in inventory['stk_dats']:
if stk['t_date'][:6] == year_month:
year_month_count = year_month_count + 1
break
return year_month_count
每日執行程式，如果今日的日期為 5 號 就直接買進，其他日期則檢查本月是否已買進超過 3 次，如果還沒到達 3 次則檢查今日收盤前是否跌幅超過 1%，若有則買進一單位。
# 取得台股市場交易日套件，若尚未安裝請先執行以下註解
#!pip install pandas-market-calendars
import pandas_market_calendars as mcal
# 使用台股市場，並預計執行到明年底
tsx = mcal.get_calendar('TSX')
trading_list = tsx.schedule(start_date='2022-10-01', end_date='2023-12-31')['market_open'].index.tolist()
# 取得每月 5 號或 5 號後的第一個交易日期
specific_list = list(filter(lambda x:x.day >= 5, trading_list))
data_dict = {"date":specific_list,"year":list(map(lambda x:x.year, specific_list)), "month":list(map(lambda x:x.month, specific_list))}
# 取得定期定額預計買進日
buy_date_list = pd.DataFrame(data_dict).drop_duplicates(subset=['year','month'])['date'].tolist()
# 如果今天並非定期定額預計買進日就會檢查本月是否買了 3 次，如果沒有則檢查收盤前是否比開盤下跌 1%
symbol = '0050'
# 如果不是定期買進日
if pd.Timestamp(datetime.datetime.today().date()) not in buy_date_list:
# 先檢查本月是否買到 3 次了
year_month = str(datetime.datetime.today().date()).replace('-','')[:6]
if check_inventories(symbol, year_month) <= 3:
# 還沒到 3 次，就開始等待到收盤時檢查是否下跌 1%
while True:
# 時間即將收盤 且已下跌 1%
if datetime.time(13, 29) >= datetime.datetime.now().time() >= datetime.time(13, 28) and cal_strategy(symbol) is True:
# 計算要買的股數
qty = cal_buy_qty(symbol, 5000)
order = OrderObject(
buy_sell=Action.Buy,
price_flag=PriceFlag.LimitUp, # 漲停買進
price = '',
stock_no= symbol,
quantity=qty,
ap_code=APCode.IntradayOdd,
trade=Trade.Cash,
)
sdk.place_order(order)
break
else:
while True:
# 時間是收盤
if datetime.time(13, 29) >= datetime.datetime.now().time() >= datetime.time(13, 28):
# 計算要買的股數
qty = cal_buy_qty(symbol, 5000)
order = OrderObject(
buy_sell=Action.Buy,
price_flag=PriceFlag.LimitUp, # 漲停買進
price ='',
stock_no=symbol,
quantity=qty,
ap_code=APCode.IntradayOdd,
trade=Trade.Cash,
)
sdk.place_order(order)
break
結語
本週我們實作了定期定額加碼版的回測及自動下單程式，若標的波動較小將會加碼在較為接近的價位使報酬率無法非常突出，對於波動較大的商品加碼的機制可能會有更大的助益。下一篇文章將介紹屬於順勢加碼的均線策略，順勢加碼將會有更明顯的差異，請大家持續追蹤!