溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
這篇文章主要為大家展示了“怎么用Python完成股票回測框架”，內容簡而易懂，條理清晰，希望能夠幫助大家解決疑惑，下面讓小編帶領大家一起研究并學習一下“怎么用Python完成股票回測框架”這篇文章吧。
無論是傳統股票交易還是量化交易，無法避免的一個問題是我們需要檢驗自己的交易策略是否可行，而最簡單的方式就是利用歷史數據檢驗交易策略，而回測框架就是提供這樣的一個平臺讓交易策略在歷史數據中不斷交易，最終生成最終結果，通過查看結果的策略收益，年化收益，最大回測等用以評估交易策略的可行性。
代碼地址在最后。
本項目并不是一個已完善的項目, 還在不斷的完善。
回測框架應該至少包含兩個部分, 回測類, 交易類.
回測類提供各種鉤子函數，用于放置自己的交易邏輯，交易類用于模擬市場的交易平臺，這個類提供買入，賣出的方法。
以自己的回測框架為例。主要包含下面兩個文件
backtest/ backtest.py broker.py
backtest.py主要提供BackTest這個類用于提供回測框架，暴露以下鉤子函數.
def initialize(self): """在回測開始前的初始化""" pass def before_on_tick(self, tick): pass def after_on_tick(self, tick): pass def before_trade(self, order): """在交易之前會調用此函數 可以在此放置資金管理及風險管理的代碼 如果返回True就允許交易，否則放棄交易 """ return True def on_order_ok(self, order): """當訂單執行成功后調用""" pass def on_order_timeout(self, order): """當訂單超時后調用""" pass def finish(self): """在回測結束后調用""" pass @abstractmethod def on_tick(self, bar): """ 回測實例必須實現的方法，并編寫自己的交易邏輯 """ pass
玩過量化平臺的回測框架或者開源框架應該對這些鉤子函數不陌生，只是名字不一樣而已，大多數功能是一致的，除了on_tick.
之所以是on_tick而不是on_bar, 是因為我希望交易邏輯是一個一個時間點的參與交易，在這個時間點我可以獲取所有當前時間的所有股票以及之前的股票數據，用于判斷是否交易，而不是一個時間點的一個一個股票參與交易邏輯。
而broker.py主要提供buy,sell兩個方法用于交易。
def buy(self, code, price, shares, ttl=-1):
"""
限價提交買入訂單
---------
Parameters:
code:str
股票代碼
price:float or None
最高可買入的價格, 如果為None則按市價買入
shares:int
買入股票數量
ttl:int
訂單允許存在的最大時間，默認為-1，永不超時
---------
return:
dict
{
"type": 訂單類型, "buy",
"code": 股票代碼,
"date": 提交日期,
"ttl": 存活時間, 當ttl等于0時則超時，往后不會在執行
"shares": 目標股份數量,
"price": 目標價格,
"deal_lst": 交易成功的歷史數據，如
[{"price": 成交價格,
"date": 成交時間,
"commission": 交易手續費,
"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):
"""
限價提交賣出訂單
---------
Parameters:
code:str
股票代碼
price:float or None
最低可賣出的價格, 如果為None則按市價賣出
shares:int
賣出股票數量
ttl:int
訂單允許存在的最大時間，默認為-1，永不超時
---------
return:
dict
{
"type": 訂單類型, "sell",
"code": 股票代碼,
"date": 提交日期,
"ttl": 存活時間, 當ttl等于0時則超時，往后不會在執行
"shares": 目標股份數量,
"price": 目標價格,
"deal_lst": 交易成功的歷史數據，如
[{"open_price": 開倉價格,
"close_price": 成交價格,
"close_date": 成交時間,
"open_date": 持倉時間,
"commission": 交易手續費,
"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由于我很討厭抽象出太多類，抽象出太多類及方法，我怕我自己都忘記了，所以對于對象的選擇都是盡可能的使用常用的數據結構，如list, dict.
這里用一個dict代表一個訂單。
上面的這些方法保證了一個回測框架的基本交易邏輯，而回測的運行還需要一個調度器不斷的驅動這些方法，這里的調度器如下。
class Scheduler(object):
"""
整個回測過程中的調度中心, 通過一個個時間刻度(tick)來驅動回測邏輯
所有被調度的對象都會綁定一個叫做ctx的Context對象,由于共享整個回測過程中的所有關鍵數據,
可用變量包括:
ctx.feed: {code1: pd.DataFrame, code2: pd.DataFrame}對象
ctx.now: 循環所處時間
ctx.tick_data: 循環所處時間的所有有報價的股票報價
ctx.trade_cal: 交易日歷
ctx.broker: Broker對象
ctx.bt/ctx.backtest: Backtest對象
可用方法:
ctx.get_hist
"""
def __init__(self):
""""""
self.ctx = Context()
self._pre_hook_lst = []
self._post_hook_lst = []
self._runner_lst = []
def run(self):
# runner指存在可調用的initialize, finish, run(tick)的對象
runner_lst = list(chain(self._pre_hook_lst, self._runner_lst, self._post_hook_lst))
# 循環開始前為broker, backtest, hook等實例綁定ctx對象及調用其initialize方法
for runner in runner_lst:
runner.ctx = self.ctx
runner.initialize()
# 創建交易日歷
if "trade_cal" not in self.ctx:
df = list(self.ctx.feed.values())[0]
self.ctx["trade_cal"] = df.index
# 通過遍歷交易日歷的時間依次調用runner
# 首先調用所有pre-hook的run方法
# 然后調用broker,backtest的run方法
# 最后調用post-hook的run方法
for tick in self.ctx.trade_cal:
self.ctx.set_currnet_time(tick)
for runner in runner_lst:
runner.run(tick)
# 循環結束后調用所有runner對象的finish方法
for runner in runner_lst:
runner.finish()在Backtest類實例化的時候就會自動創建一個調度器對象，然后通過Backtest實例的start方法就能啟動調度器，而調度器會根據歷史數據的一個一個時間戳不斷驅動Backtest, Broker實例被調用。
為了處理不同實例之間的數據訪問隔離，所以通過一個將一個Context對象綁定到Backtest, Broker實例上，通過self.ctx訪問共享的數據，共享的數據主要包括feed對象，即歷史數據，一個數據結構如下的字典對象。
{code1: pd.DataFrame, code2: pd.DataFrame}而這個Context對象也綁定了Broker, Backtest的實例, 這就可以使得數據訪問接口統一，但是可能導致數據訪問混亂，這就要看策略者的使用了，這樣的一個好處就是減少了一堆代理方法，通過添加方法去訪問其他的對象的方法，真不嫌麻煩，那些人。
綁定及Context對象代碼如下:
class Context(UserDict):
def __getattr__(self, key):
# 讓調用這可以通過索引或者屬性引用皆可
return self[key]
def set_currnet_time(self, tick):
self["now"] = tick
tick_data = {}
# 獲取當前所有有報價的股票報價
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, 獲取截至到當前時間的所有股票的歷史數據"""
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):
"""
整個回測過程中的調度中心, 通過一個個時間刻度(tick)來驅動回測邏輯
所有被調度的對象都會綁定一個叫做ctx的Context對象,由于共享整個回測過程中的所有關鍵數據,
可用變量包括:
ctx.feed: {code1: pd.DataFrame, code2: pd.DataFrame}對象
ctx.now: 循環所處時間
ctx.tick_data: 循環所處時間的所有有報價的股票報價
ctx.trade_cal: 交易日歷
ctx.broker: Broker對象
ctx.bt/ctx.backtest: Backtest對象
可用方法:
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
# 簡寫
self.ctx["bt"] = backtest
def add_runner(self, runner):
if runner in self._runner_lst:
return
self._runner_lst.append(runner)為了使得整個框架可擴展，回測框架中框架中抽象了一個Hook類，這個類可以在在每次回測框架調用前或者調用后被調用，這樣就可以加入一些處理邏輯，比如統計資產變化等。
這里創建了一個Stat的Hook對象，用于統計資產變化。
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而通過這些統計的數據就可以計算最大回撤年化率等。
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: 最終價值 c: 初始價值 D: 有效投資時間(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
至此一個筆者需要的回測框架形成了。
在回測框架中我并沒有集成各種獲取數據的方法，因為這并不是回測框架必須集成的部分，規定數據結構就可以了，數據的獲取通過查看數據篇，
回測報告我也放在了回測框架之外，這里寫了一個Plottter的對象用于繪制一些回測指標等。結果如下:
下面是一個回測示例。
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")以上是“怎么用Python完成股票回測框架”這篇文章的所有內容，感謝各位的閱讀！相信大家都有了一定的了解，希望分享的內容對大家有所幫助，如果還想學習更多知識，歡迎關注億速云行業資訊頻道！
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
