通過純Python完成股票回測(cè)框架的搭建。
什么是回測(cè)框架?
無(wú)論是傳統(tǒng)股票交易還是量化交易，無(wú)法避免的一個(gè)問題是我們需要檢驗(yàn)自己的交易策略是否可行，而最簡(jiǎn)單的方式就是利用歷史數(shù)據(jù)檢驗(yàn)交易策略，而回測(cè)框架就是提供這樣的一個(gè)平臺(tái)讓交易策略在歷史數(shù)據(jù)中不斷交易，最終生成最終結(jié)果，通過查看結(jié)果的策略收益，年化收益，最大回測(cè)等用以評(píng)估交易策略的可行性。
代碼地址在最后。
本項(xiàng)目并不是一個(gè)已完善的項(xiàng)目, 還在不斷的完善。
回測(cè)框架
回測(cè)框架應(yīng)該至少包含兩個(gè)部分, 回測(cè)類, 交易類.
回測(cè)類提供各種鉤子函數(shù)，用于放置自己的交易邏輯，交易類用于模擬市場(chǎng)的交易平臺(tái)，這個(gè)類提供買入，賣出的方法。
代碼架構(gòu)
以自己的回測(cè)框架為例。主要包含下面兩個(gè)文件
backtest/
backtest.py
broker.py
backtest.py主要提供BackTest這個(gè)類用于提供回測(cè)框架，暴露以下鉤子函數(shù).
def initialize(self):
"""在回測(cè)開始前的初始化"""
pass
def before_on_tick(self, tick):
pass
def after_on_tick(self, tick):
pass
def before_trade(self, order):
"""在交易之前會(huì)調(diào)用此函數(shù)
可以在此放置資金管理及風(fēng)險(xiǎn)管理的代碼
如果返回True就允許交易，否則放棄交易
"""
return True
def on_order_ok(self, order):
"""當(dāng)訂單執(zhí)行成功后調(diào)用"""
pass
def on_order_timeout(self, order):
"""當(dāng)訂單超時(shí)后調(diào)用"""
pass
def finish(self):
"""在回測(cè)結(jié)束后調(diào)用"""
pass
@abstractmethod
def on_tick(self, bar):
"""
回測(cè)實(shí)例必須實(shí)現(xiàn)的方法，并編寫自己的交易邏輯
"""
pass
玩過量化平臺(tái)的回測(cè)框架或者開源框架應(yīng)該對(duì)這些鉤子函數(shù)不陌生，只是名字不一樣而已，大多數(shù)功能是一致的，除了on_tick.
之所以是on_tick而不是on_bar, 是因?yàn)槲蚁Ｍ灰走壿嬍且粋€(gè)一個(gè)時(shí)間點(diǎn)的參與交易，在這個(gè)時(shí)間點(diǎn)我可以獲取所有當(dāng)前時(shí)間的所有股票以及之前的股票數(shù)據(jù)，用于判斷是否交易，而不是一個(gè)時(shí)間點(diǎn)的一個(gè)一個(gè)股票參與交易邏輯。
而broker.py主要提供buy,sell兩個(gè)方法用于交易。
def buy(self, code, price, shares, ttl=-1):
"""
限價(jià)提交買入訂單
---------
Parameters:
code:str
股票代碼
price:float or None
最高可買入的價(jià)格, 如果為None則按市價(jià)買入
shares:int
買入股票數(shù)量
ttl:int
訂單允許存在的最大時(shí)間，默認(rèn)為-1，永不超時(shí)
---------
return:
dict
{
"type": 訂單類型, "buy",
"code": 股票代碼,
"date": 提交日期,
"ttl": 存活時(shí)間, 當(dāng)ttl等于0時(shí)則超時(shí)，往后不會(huì)在執(zhí)行
"shares": 目標(biāo)股份數(shù)量,
"price": 目標(biāo)價(jià)格,
"deal_lst": 交易成功的歷史數(shù)據(jù)，如
[{"price": 成交價(jià)格,
"date": 成交時(shí)間,
"commission": 交易手續(xù)費(fèi),
"shares": 成交份額
}]
""
}
"""
if price is None:
stock_info = self.ctx.tick_data[code]
price = stock_info[self.deal_price]
order = {
"type": "buy",
"code": code,
"date": self.ctx.now,
"ttl": ttl,
"shares": shares,
"price": price,
"deal_lst": []
}
self.submit(order)
return order
def sell(self, code, price, shares, ttl=-1):
"""
限價(jià)提交賣出訂單
---------
Parameters:
code:str
股票代碼
price:float or None
最低可賣出的價(jià)格, 如果為None則按市價(jià)賣出
shares:int
賣出股票數(shù)量
ttl:int
訂單允許存在的最大時(shí)間，默認(rèn)為-1，永不超時(shí)
---------
return:
dict
{
"type": 訂單類型, "sell",
"code": 股票代碼,
"date": 提交日期,
"ttl": 存活時(shí)間, 當(dāng)ttl等于0時(shí)則超時(shí)，往后不會(huì)在執(zhí)行
"shares": 目標(biāo)股份數(shù)量,
"price": 目標(biāo)價(jià)格,
"deal_lst": 交易成功的歷史數(shù)據(jù)，如
[{"open_price": 開倉(cāng)價(jià)格,
"close_price": 成交價(jià)格,
"close_date": 成交時(shí)間,
"open_date": 持倉(cāng)時(shí)間,
"commission": 交易手續(xù)費(fèi),
"shares": 成交份額,
"profit": 交易收益}]
""
}
"""
if code not in self.position:
return
if price is None:
stock_info = self.ctx.tick_data[code]
price = stock_info[self.deal_price]
order = {
"type": "sell",
"code": code,
"date": self.ctx.now,
"ttl": ttl,
"shares": shares,
"price": price,
"deal_lst": []
}
self.submit(order)
return order
由于我很討厭抽象出太多類，抽象出太多類及方法，我怕我自己都忘記了，所以對(duì)于對(duì)象的選擇都是盡可能的使用常用的數(shù)據(jù)結(jié)構(gòu)，如list, dict.
這里用一個(gè)dict代表一個(gè)訂單。
上面的這些方法保證了一個(gè)回測(cè)框架的基本交易邏輯，而回測(cè)的運(yùn)行還需要一個(gè)調(diào)度器不斷的驅(qū)動(dòng)這些方法，這里的調(diào)度器如下。
class Scheduler(object):
"""
整個(gè)回測(cè)過程中的調(diào)度中心, 通過一個(gè)個(gè)時(shí)間刻度(tick)來(lái)驅(qū)動(dòng)回測(cè)邏輯
所有被調(diào)度的對(duì)象都會(huì)綁定一個(gè)叫做ctx的Context對(duì)象,由于共享整個(gè)回測(cè)過程中的所有關(guān)鍵數(shù)據(jù),
可用變量包括:
ctx.feed: {code1: pd.DataFrame, code2: pd.DataFrame}對(duì)象
ctx.now: 循環(huán)所處時(shí)間
ctx.tick_data: 循環(huán)所處時(shí)間的所有有報(bào)價(jià)的股票報(bào)價(jià)
ctx.trade_cal: 交易日歷
ctx.broker: Broker對(duì)象
ctx.bt/ctx.backtest: Backtest對(duì)象
可用方法:
ctx.get_hist
"""
def __init__(self):
""""""
self.ctx = Context()
self._pre_hook_lst = []
self._post_hook_lst = []
self._runner_lst = []
def run(self):
# runner指存在可調(diào)用的initialize, finish, run(tick)的對(duì)象
runner_lst = list(chain(self._pre_hook_lst, self._runner_lst, self._post_hook_lst))
# 循環(huán)開始前為broker, backtest, hook等實(shí)例綁定ctx對(duì)象及調(diào)用其initialize方法
for runner in runner_lst:
runner.ctx = self.ctx
runner.initialize()
# 創(chuàng)建交易日歷
if "trade_cal" not in self.ctx:
df = list(self.ctx.feed.values())[0]
self.ctx["trade_cal"] = df.index
# 通過遍歷交易日歷的時(shí)間依次調(diào)用runner
# 首先調(diào)用所有pre-hook的run方法
# 然后調(diào)用broker,backtest的run方法
# 最后調(diào)用post-hook的run方法
for tick in self.ctx.trade_cal:
self.ctx.set_currnet_time(tick)
for runner in runner_lst:
runner.run(tick)
# 循環(huán)結(jié)束后調(diào)用所有runner對(duì)象的finish方法
for runner in runner_lst:
runner.finish()
在Backtest類實(shí)例化的時(shí)候就會(huì)自動(dòng)創(chuàng)建一個(gè)調(diào)度器對(duì)象，然后通過Backtest實(shí)例的start方法就能啟動(dòng)調(diào)度器，而調(diào)度器會(huì)根據(jù)歷史數(shù)據(jù)的一個(gè)一個(gè)時(shí)間戳不斷驅(qū)動(dòng)Backtest, Broker實(shí)例被調(diào)用。
為了處理不同實(shí)例之間的數(shù)據(jù)訪問隔離，所以通過一個(gè)將一個(gè)Context對(duì)象綁定到Backtest, Broker實(shí)例上，通過self.ctx訪問共享的數(shù)據(jù)，共享的數(shù)據(jù)主要包括feed對(duì)象，即歷史數(shù)據(jù)，一個(gè)數(shù)據(jù)結(jié)構(gòu)如下的字典對(duì)象。
{code1: pd.DataFrame, code2: pd.DataFrame}
而這個(gè)Context對(duì)象也綁定了Broker, Backtest的實(shí)例, 這就可以使得數(shù)據(jù)訪問接口統(tǒng)一，但是可能導(dǎo)致數(shù)據(jù)訪問混亂，這就要看策略者的使用了，這樣的一個(gè)好處就是減少了一堆代理方法，通過添加方法去訪問其他的對(duì)象的方法，真不嫌麻煩，那些人。
綁定及Context對(duì)象代碼如下:
class Context(UserDict):
def __getattr__(self, key):
# 讓調(diào)用這可以通過索引或者屬性引用皆可
return self[key]
def set_currnet_time(self, tick):
self["now"] = tick
tick_data = {}
# 獲取當(dāng)前所有有報(bào)價(jià)的股票報(bào)價(jià)
for code, hist in self["feed"].items():
df = hist[hist.index == tick]
if len(df) == 1:
tick_data[code] = df.iloc[-1]
self["tick_data"] = tick_data
def get_hist(self, code=None):
"""如果不指定code, 獲取截至到當(dāng)前時(shí)間的所有股票的歷史數(shù)據(jù)"""
if code is None:
hist = {}
for code, hist in self["feed"].items():
hist[code] = hist[hist.index <= self.now]
elif code in self.feed:
return {code: self.feed[code]}
return hist
class Scheduler(object):
"""
整個(gè)回測(cè)過程中的調(diào)度中心, 通過一個(gè)個(gè)時(shí)間刻度(tick)來(lái)驅(qū)動(dòng)回測(cè)邏輯
所有被調(diào)度的對(duì)象都會(huì)綁定一個(gè)叫做ctx的Context對(duì)象,由于共享整個(gè)回測(cè)過程中的所有關(guān)鍵數(shù)據(jù),
可用變量包括:
ctx.feed: {code1: pd.DataFrame, code2: pd.DataFrame}對(duì)象
ctx.now: 循環(huán)所處時(shí)間
ctx.tick_data: 循環(huán)所處時(shí)間的所有有報(bào)價(jià)的股票報(bào)價(jià)
ctx.trade_cal: 交易日歷
ctx.broker: Broker對(duì)象
ctx.bt/ctx.backtest: Backtest對(duì)象
可用方法:
ctx.get_hist
"""
def __init__(self):
""""""
self.ctx = Context()
self._pre_hook_lst = []
self._post_hook_lst = []
self._runner_lst = []
def add_feed(self, feed):
self.ctx["feed"] = feed
def add_hook(self, hook, typ="post"):
if typ == "post" and hook not in self._post_hook_lst:
self._post_hook_lst.append(hook)
elif typ == "pre" and hook not in self._pre_hook_lst:
self._pre_hook_lst.append(hook)
def add_broker(self, broker):
self.ctx["broker"] = broker
def add_backtest(self, backtest):
self.ctx["backtest"] = backtest
# 簡(jiǎn)寫
self.ctx["bt"] = backtest
def add_runner(self, runner):
if runner in self._runner_lst:
return
self._runner_lst.append(runner)
為了使得整個(gè)框架可擴(kuò)展，回測(cè)框架中框架中抽象了一個(gè)Hook類，這個(gè)類可以在在每次回測(cè)框架調(diào)用前或者調(diào)用后被調(diào)用，這樣就可以加入一些處理邏輯，比如統(tǒng)計(jì)資產(chǎn)變化等。
這里創(chuàng)建了一個(gè)Stat的Hook對(duì)象，用于統(tǒng)計(jì)資產(chǎn)變化。
class Stat(Base):
def __init__(self):
self._date_hist = []
self._cash_hist = []
self._stk_val_hist = []
self._ast_val_hist = []
self._returns_hist = []
def run(self, tick):
self._date_hist.append(tick)
self._cash_hist.append(self.ctx.broker.cash)
self._stk_val_hist.append(self.ctx.broker.stock_value)
self._ast_val_hist.append(self.ctx.broker.assets_value)
@property
def data(self):
df = pd.DataFrame({"cash": self._cash_hist,
"stock_value": self._stk_val_hist,
"assets_value": self._ast_val_hist}, index=self._date_hist)
df.index.name = "date"
return df
而通過這些統(tǒng)計(jì)的數(shù)據(jù)就可以計(jì)算最大回撤年化率等。
def get_dropdown(self):
high_val = -1
low_val = None
high_index = 0
low_index = 0
dropdown_lst = []
dropdown_index_lst = []
for idx, val in enumerate(self._ast_val_hist):
if val >= high_val:
if high_val == low_val or high_index >= low_index:
high_val = low_val = val
high_index = low_index = idx
continue
dropdown = (high_val - low_val) / high_val
dropdown_lst.append(dropdown)
dropdown_index_lst.append((high_index, low_index))
high_val = low_val = val
high_index = low_index = idx
if low_val is None:
low_val = val
low_index = idx
if val < low_val:
low_val = val
low_index = idx
if low_index > high_index:
dropdown = (high_val - low_val) / high_val
dropdown_lst.append(dropdown)
dropdown_index_lst.append((high_index, low_index))
return dropdown_lst, dropdown_index_lst
@property
def max_dropdown(self):
"""最大回車率"""
dropdown_lst, dropdown_index_lst = self.get_dropdown()
if len(dropdown_lst) > 0:
return max(dropdown_lst)
else:
return 0
@property
def annual_return(self):
"""
年化收益率
y = (v/c)^(D/T) - 1
v: 最終價(jià)值
c: 初始價(jià)值
D: 有效投資時(shí)間(365)
注: 雖然投資股票只有250天，但是持有股票后的非交易日也沒辦法投資到其他地方，所以這里我取365
參考: https://wiki.mbalib.com/zh-tw/%E5%B9%B4%E5%8C%96%E6%94%B6%E7%9B%8A%E7%8E%87
"""
D = 365
c = self._ast_val_hist[0]
v = self._ast_val_hist[-1]
days = (self._date_hist[-1] - self._date_hist[0]).days
ret = (v / c) ** (D / days) - 1
return ret
至此一個(gè)筆者需要的回測(cè)框架形成了。
交易歷史數(shù)據(jù)
在回測(cè)框架中我并沒有集成各種獲取數(shù)據(jù)的方法，因?yàn)檫@并不是回測(cè)框架必須集成的部分，規(guī)定數(shù)據(jù)結(jié)構(gòu)就可以了，數(shù)據(jù)的獲取通過查看數(shù)據(jù)篇，
回測(cè)報(bào)告
回測(cè)報(bào)告我也放在了回測(cè)框架之外，這里寫了一個(gè)Plottter的對(duì)象用于繪制一些回測(cè)指標(biāo)等。結(jié)果如下:
![用Python徒手?jǐn)]一個(gè)股票回測(cè)框架_第1張圖片](https://img.it610.com/image/info3/e2cbe7b2ad1a4afd9aa9d29384bfff52.jpg)
回測(cè)示例
下面是一個(gè)回測(cè)示例。
import json
from backtest import BackTest
from reporter import Plotter
class MyBackTest(BackTest):
def initialize(self):
self.info("initialize")
def finish(self):
self.info("finish")
def on_tick(self, tick):
tick_data = self.ctx["tick_data"]
for code, hist in tick_data.items():
if hist["ma10"] > 1.05 * hist["ma20"]:
self.ctx.broker.buy(code, hist.close, 500, ttl=5)
if hist["ma10"] < hist["ma20"] and code in self.ctx.broker.position:
self.ctx.broker.sell(code, hist.close, 200, ttl=1)
if __name__ == '__main__':
from utils import load_hist
feed = {}
for code, hist in load_hist("000002.SZ"):
# hist = hist.iloc[:100]
hist["ma10"] = hist.close.rolling(10).mean()
hist["ma20"] = hist.close.rolling(20).mean()
feed[code] = hist
mytest = MyBackTest(feed)
mytest.start()
order_lst = mytest.ctx.broker.order_hist_lst
with open("report/order_hist.json", "w") as wf:
json.dump(order_lst, wf, indent=4, default=str)
stats = mytest.stat
stats.data.to_csv("report/stat.csv")
print("策略收益： {:.3f}%".format(stats.total_returns * 100))
print("最大回徹率: {:.3f}% ".format(stats.max_dropdown * 100))
print("年化收益: {:.3f}% ".format(stats.annual_return * 100))
print("夏普比率: {:.3f} ".format(stats.sharpe))
plotter = Plotter(feed, stats, order_lst)
plotter.report("report/report.png")
項(xiàng)目地址
https://github.com/youerning/stock_playground
更多文章、技術(shù)交流、商務(wù)合作、聯(lián)系博主
微信掃碼或搜索：z360901061
微信掃一掃加我為好友
QQ號(hào)聯(lián)系： 360901061
您的支持是博主寫作最大的動(dòng)力，如果您喜歡我的文章，感覺我的文章對(duì)您有幫助，請(qǐng)用微信掃描下面二維碼支持博主2元、5元、10元、20元等您想捐的金額吧，狠狠點(diǎn)擊下面給點(diǎn)支持吧，站長(zhǎng)非常感激您！手機(jī)微信長(zhǎng)按不能支付解決辦法：請(qǐng)將微信支付二維碼保存到相冊(cè)，切換到微信，然后點(diǎn)擊微信右上角掃一掃功能，選擇支付二維碼完成支付。
【本文對(duì)您有幫助就好】元
