以下內容皆參考 Backtrader 官網
昨天使用了 backtrader 將 shioaji 的歷史資料載入,今天我們就做一個很簡單的策略來測試實際回測看看
class TestStrategy(bt.Strategy):
def log(self, txt, dt=None):
dt = dt or self.datas[0].datetime.date(0)
print('%s, %s' % (dt.isoformat(), txt))
def __init__(self):
self.dataclose = self.datas[0].close
def next(self):
self.log('收盤價: %.2f' % dataclose[0])
我們使用 pandas.DataFrame 將 shioaji 的資料載入到 backtrader 的 datas 裡,所以 self.datas[0] 就是我們載入的資料,載入的資料有 close, high, low, open, volume,這 5 條線,如果把 datetime 也算進來的話,那就是 6 條線。
在 init 裡,我們把 close 這一條線特別設了一個變數叫 self.dataclose,然後在 next 這個函數裡,我們將 dataclose[0] 的數值輸出來。
回測就是從最舊的資料,一筆一筆往最新的資料跑,所以這個 dataclose[0] 就是目前程式跑到的那一筆,如果要看前一筆,就是 dataclose[-1], 以此類推,可以是 -2, -3 前兩筆,前三筆。
以上這段程式就是讓它一天一天跑,然後把收盤價輸出來
from datetime import datetime
import pandas as pd
import shioaji as sj
import backtrader as bt
api = sj.Shiaji()
api.login(person_id="身分證字號", passwd="密碼")
stock2330 = api.Contracts.Stocks["2330"]
kbar2330 = api.kbars(stock2330, start='2020-01-01', end='2020-12-31')
dts = list(map(lambda x:dateime.utcformattimestamp(x/10**9), kbr2330.ts))
df = pd.DataFrame(
{
"open": pd.Series(kbar2330.Open),
"high": pd.Series(kbar2330.High),
"low": pd.Series(kbar2330.Low),
"close": pd.Series(kbar2330.close),
"volume": pd.Series(kbar2330.Volume),
}
)
df.index = pd.Index(dts)
以上為取得資料的程式碼,如果是用 jupyter notebook 的話,可以放在一個獨立的程式區塊,這樣就不用一直去抓資料下來了,執行起來會快一些
cerebro = bt.Cerebro()
# 載入要執行的策略
cerebro.addstrategy(TestStrategy)
# 因為台積的股價有點高,所以我們初始資金設 100 萬
cerebro.broker.setcash(10**6)
# 券商的手續費
cerebro.broker.setcommission(commission=0.1425/100)
data = bt.feeds.PandasData(dataname=df, timeframe=bt.TimeFrame.Minutes)
cerebro.resampledata(data, timeframe=bt.TimeFrame.Days)
# 如果跑到最後,還有股票沒有賣出的話,會以當天收盤價的價格來加總
print("初始資金: %.2f" % cerebro.broker.getvalue())
# 執行策略
cerebro.run()
print("結束資金:%.2f" % cerebro.broker.getvalue())
以上程式,執行之後可以看到輸出每天的收盤價,接下來我們來賣加一個買入的策略
class TestStrategy(bt.Strategy):
def log(self, txt, dt=None):
#...略
def __init__(self):
#...略
def next(self):
self.log('收盤價: %.2f' % dataclose[0])
# 當天收盤價比昨天低 -> 跌
if self.dataclose[0] < self.dataclose[-1]:
# 昨天的收盤價比前天低 -> 連跌 2 天
if self.dataclose[-1] < self.dataclose[-2]:
self.log("買入: %.2f" % self.dataclose[0])
# 一張股票是1000股
self.buy(size=1000)
這個策略就是,如果股票連跌 2 天,就買入 1 張 (1000 股),執行完的個策略後,再去跑回測,就可以看到我們買進股票,再來我們要增加賣出的策略
class TestStrategy(bt.Strategy):
def log(self, txt, dt=None):
# ...略
def __init__(self):
# ...略
# 建立一個變數儲存我們己經買的股票
self.order = None
def notify_order(self, order):
if order.status in [order.Submitted, order.Accepted]:
# 買賣單傳送到券商,券商接收,無需處理
return
if order.status in [order.Completed]:
# 買賣單交易完成
if order.isbuy():
# 買單
self.log("已買入:%.2f" % order.executed.price)
elif order.issell():
# 賣單
self.log("已賣出:%.2f" % order.executed.price)
# 紀錄是在第幾筆資料進行交易
self.bar_executed = len(self)
elif order.status in [order.Canceled, order.Margin, order.Rejected]:
self.log("交易失敗")
# 沒有待處理訂單
self.order = None
def next(self):
self.log("收盤價:%.2f" % self.dataclose[0])
# 還有未處理的訂單,直接跳過
if self.order:
return
# 沒有持股 -> 進行買入策略
if not self.position:
if self.dataclose[-1] < self.dataclose[0]:
if self.dataclose[-2] < self.dataclose[-1]:
self.log("買入:%.2f" % self.dataclose[0])
self.order = self.buy(size=1000)
# 有持股 -> 進行賣出策略
else:
if len(self) >= (self.bar_executed + 5):
self.log("賣出:%.2f" % sef.dataclose[0])
self.order = self.sell(size=1000)
這樣就是一個簡單的買入,賣出策略回測,執行之後,就可以看到以這樣子的策略,我們的在 2020 年可以賺到多少,當然因為這個是歷史資料,所以只是一個評估,並不代表未來用這個策略去執行也可以獲得同樣的收益喔