通过纯Python完成股票回测的框架。

什么是回溯测试框架？

无论是传统的股票交易还是量化交易，一个不可避免的问题是，我们需要测试我们的交易策略是否可行，最简单的方法是使用历史数据来测试交易策略，回溯测试框架提供了这样一个平台，让交易策略在历史数据中不断交易，最终生成最终的结果。交易策略的可行性可以通过查看策略回报、年化回报和结果的更大回溯测试来评估。

代码地址在末尾。

这个项目不是完美的，但在不断改进。

背面测试框架

回溯测试框架应该至少包含两部分，回溯测试类和事务类。

回测类提供各种挂钩函数放置自己的交易逻辑，交易类用来模拟市场的交易平台。这个类提供了买卖的方法。

代码架构

以您自己的回溯测试框架为例。主要包含以下两个文件

回溯测试/

backtest.py

broker.py

BackTest.py主要提供BackTest，用于提供backtest框架，并公开以下钩子函数。

def初始化(自):

''开始回溯测试前的初始化'''

及格

def before_on_tick(self，tick):

及格

def after_on_tick(self，tick):

及格

def before_trade(self，order):

“”将在事务之前调用此函数

资金管理和风险管理的代码可以放在这里

如果返回真，则允许该事务，否则放弃该事务

'''

返回真

def on_order_ok(自身，订单):

成功执行订单时，调用。

及格

def on_order_timeout(自身，订单):

当订单超时时，调用。

及格

def finish(self):

回溯测试结束后的“调用”。

及格

@abstractmethod

def on_tick(self，bar):

'''

回溯测试实例必须实现的方法，并编写自己的事务逻辑

'''

及格

玩过量化平台的回溯测试框架或者开源框架应该熟悉这些钩子函数，只是名字不同，除了on_tick之外大部分函数都是一样的。

之所以是on_tick而不是on_bar，是因为我希望交易逻辑是某个时间点的交易。在这个时间点，我可以得到当前时间的所有股票和以前的股票数据来判断是否交易，而不是一个时间点一个个参与股票的交易逻辑。

而broker.py主要提供交易的买入和卖出方式。

def buy(self，code，price，shares，ttl=-1):

'''

以限定价格提交购买订单

-

参数：

代码：str

股票代码

价格：浮动或无

能买到的更高价，如果是None，就按市场价买

shares:int

购买的股票数量

ttl:int

默认情况下，订单允许存在的最长时间为-1，并且永不超时

-

返回：

词典

{

类型' :订单类型，'购买'，

代码' :股票代码，

日期' :提交日期，

Ttl':生存时间，当Ttl等于0时，将超时，以后不执行

股份' :目标股份数量，

价格' :目标价，

交易成功的历史数据，如

[{'price':成交价格，

日期' :结束时间，

佣金' :交易费，

股份：周转份额

}]

''

}

'''

如果价格不是：

stock _ info=self . CTX . tick _ data[code]

price=stock _ info[self . deal _ price]

订单={

键入' : '购买'，

代码' :代码，

日期' : self.ctx.now，

ttl': ttl，

:股，

价格' :价格，

deal_lst': []

}

self.submit(订单)

退货单

def sell(self，code，price，shares，ttl=-1):

'''

以限定价格提交卖单

-

参数：

代码：str

股票代码

价格：浮动或无

更低可售价格，如果是无，将按市场价出售

shares:int

出售的股份数量

ttl:int

默认情况下，订单允许存在的最长时间为-1，并且永不超时

-

返回：

词典

{

类型' :订单类型，'销售'，

代码' :股票代码，

日期' :提交日期，

Ttl':生存时间，当Ttl等于0时，将超时，以后不执行

股份' :目标股份数量，

价格' :目标价，

交易成功的历史数据，如

[{'open_price':开盘价，

收盘价' :收盘价，

关闭_日期' :关闭时间，

Open_date':持仓时间，

佣金' :交易费，

股份：周转份额，

利润' :交易收入}]

''

}

'''

如果代码不在自己的位置：

返回

如果价格不是：

stock _ info=self . CTX . tick _ data[code]

price=stock _ info[self . deal _ price]

订单={

类型' : '销售'，

代码' :代码，

日期' : self.ctx.now，

ttl': ttl，

:股，

价格' :价格，

deal_lst': []

}

self.submit(订单)

退货单

因为我讨厌抽象太多的类，抽象太多的类和方法，怕自己忘了自己，所以一直用常用的数据结构，比如list，dict。

在这里，格言代表一种秩序。

以上方法保证了一个回测框架的基本事务逻辑，回测的操作需要一个调度器不断驱动这些方法。这里的调度器如下。

类别调度程序(对象):

'''

调度中心在回测的整个过程中，通过tick驱动回测逻辑

所有计划的对象都将与一个名为ctx的上下文对象绑定。因为在整个回溯测试过程中共享所有关键数据，

可用变量包括：

CTX . feed : { code 1: PD . data frame，code2:pd.dataframe}对象

Ctx.now:周期时间

Ctx.tick_data:周期内所有股票报价

Ctx.trade_cal:交易日历

Ctx.broker:代理对象

ctx.bt/ctx.backtest:的回测对象

可用方法：

ctx.get_hist

'''

def __init__(self):

''''''

self.ctx=Context()

自我。_pre_hook_lst=[]

自我。_post_hook_lst=[]

自我。_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))

#绑定ctx对象，并在循环开始前为代理、回溯测试、钩子和其他实例调用它们的初始化方法

对于runner_lst:中的runner

runner.ctx=self.ctx

runner.initialize()

#创建交易日历

如果“trade_cal”不在self.ctx:中

df=list(self . CTX . feed . values())[0]

self.ctx['trade_cal']=df.index

#通过遍历交易日历时间依次呼叫跑步者

#首先调用所有预挂钩的运行方法

#然后调用broker的run方法，backtest

#最后调用后挂钩的run方法

for tick in self . CTX . trade _ cal :

self . CTX . set _ currenet _ time(tick)

对于runner_lst:中的runner

runner.run(滴答)

#循环结束后调用所有流道对象的结束方法

对于runner_lst:中的runner

runner.finish()

Backtest类实例化时，会自动创建一个scheduler对象，然后可以通过Backtest实例的start方法启动scheduler，scheduler会根据历史数据的一个时间戳持续驱动backtest，调用broker实例。

为了处理不同实例之间的数据访问隔离，在Backtest、Broker的实例上绑定了一个Context对象，通过self.ctx访问共享数据，共享数据主要包括feed对象，即历史数据，以及一个数据结构如下的字典对象。

{code1: pd。数据帧，代码2: pd。数据框}

而这个语境对象也绑定了经纪人，回溯测试的实例，这就可以使得数据访问接口统一，但是可能导致数据访问混乱，这就要看策略者的使用了，这样的一个好处就是减少了一堆代理方法，通过添加方法去访问其他的对象的方法，真不嫌麻烦，那些人。

绑定及语境对象代码如下：

类上下文（用户词典):

def __getattr__(self，key):

# 让调用这可以通过索引或者属性引用皆可

返回自我[钥匙]

def set _ current _ time(self，tick):

self['now']=tick

tick_data={}

# 获取当前所有有报价的股票报价

对于代码hist在self['feed']中项目(:)

df=hist[hist.index==tick]

if len(df)=1:

tick_data[code]=df.iloc[-1]

self['tick_data']=tick_data

def get_hist(自身，代码=无):

'''如果不指定代码，获取截至到当前时间的所有股票的历史数据'''

如果代码不是：

hist={}

对于代码hist在self['feed']中项目(:)

hist[code]=hist[hist。索引=自我。现在]

自我反馈中的否则如果代码：

返回{code: self.feed[code]}

返回历史

类别调度程序（对象):

'''

整个回测过程中的调度中心，通过一个个时间刻度（打勾)来驱动回测逻辑

所有被调度的对象都会绑定一个叫做中强的语境对象，由于共享整个回测过程中的所有关键数据，

可用变量包括：

ctx.feed: {code1: pd .数据帧，代码2: pd .数据框}对象

ctx.now:循环所处时间

ctx.tick_data:循环所处时间的所有有报价的股票报价

ctx.trade_cal:交易日历

broker:对象

ctx.bt/ctx.backtest:回溯测试对象

可用方法：

ctx.get_hist

'''

def __init__(self):

''''''

self.ctx=Context()

自我_pre_hook_lst=[]

自我. post_hook_lst=[]

自我_runner_lst=[]

def add_feed(self，feed):

self.ctx['feed']=feed

def add_hook(self，hook，typ='post'):

如果typ=='post '和钩不在自己_post_hook_lst:

自我. post_hook_lst.append(hook)

elif typ=='pre '，挂钩不在自身中_pre_hook_lst:

自我. 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):

如果自己逃跑_runner_lst:

返回

自我. runner_lst.append(runner)

为了使得整个框架可扩展，回测框架中框架中抽象了一个钩类，这个类可以在在每次回测框架调用前或者调用后被调用，这样就可以加入一些处理逻辑，比如统计资产变化等。

这里创建了一个斯达的钩对象，用于统计资产变化。

类别统计（基本):

def __init__(self):

自我_date_hist=[]

自我_cash_hist=[]

自我. stk_val_hist=[]

自我. ast_val_hist=[]

自我. returns_hist=[]

def run(self，tick):

自我_date_hist.append(tick)

自我_ cash _ hist .追加(自我。CTX。经纪人。现金)

自我_ STK _瓦尔_史。追加(自我。CTX。经纪人。股票价值)

自我. ast _ val _ hist。追加(自我。CTX。经纪人。资产_价值)

@属性

def data(self):

df=pd .DataFrame({'cash': self ._现金_历史，

stock_value': self ._stk_val_hist，

assets_value': self ._ast_val_hist}，index=self ._日期_历史)

df.index.name='date '

返回df

而通过这些统计的数据就可以计算更大回撤年化率等。

def get_dropdown(self):

high_val=-1

低_值=无

high_index=0

low_index=0

dropdown_lst=[]

dropdown_index_lst=[]

对于idx，val在列举(自我._ast_val_hist):

if val=high_val:

如果high_val==low_val或high_index=low_index:

高_阀=低_阀=阀

高指数=低指数=idx

继续

下拉=(高_ val-低_ val)/高_ val

dropdown_lst.append(dropdown)

下拉_ index _ lst。追加((高_索引，低_索引))

高_阀=低_阀=阀

高指数=低指数=idx

如果low_val为None:

low_val=val

low_index=idx

if val low_val:

low_val=val

low_index=idx

if low_index high_index:

下拉=(高_ val-低_ val)/高_ val

dropdown_lst.append(dropdown)

下拉_ index _ lst。追加((高_索引，低_索引))

返回dropdown_lst，dropdown_index_lst

@属性

def max_dropdown(self):

'''更大回车率'''

dropdown_lst，down _ index _ lst=self。get _ drop down()

if len(dropdown_lst) 0:

返回更大值(dropdown_lst)

else:

返回0

@属性

极好的年度回报（自我):

'''

年化收益率

y=(v/c)^(D/T) - 1

v:最终价值

c:初始价值

D:有效投资时间(365)

注： 虽然投资股票只有250天，但是持有股票后的非交易日也没办法投资到其他地方，所以这里我取365

参考： https://wiki.mbalib.com/zh-tw/年化收益率

'''

D=365

c=自我. ast_val_hist[0]

v=自我. ast_val_hist[-1]

天数=(自我._date_hist[-1] - self ._date_hist[0]).天

ret=(v/c) ** (D /天)- 1

返回浸水使柔软

至此一个笔者需要的回测框架形成了。

交易历史数据

在回测框架中我并没有集成各种获取数据的方法，因为这并不是回测框架必须集成的部分，规定数据结构就可以了，数据的获取通过查看数据篇，

回测报告

回测报告我也放在了回测框架之外，这里写了一个绘图仪的对象用于绘制一些回测指标等。结果如下：

用计算机编程语言徒手撸一个股票回测框架

回测示例

下面是一个回测示例。

导入数据

从回溯测试导入BackTest

从报告器导入绘图仪

类别我的回溯测试(回溯测试):

极好的初始化（自):

self.info('初始化)

def finish(self):

self.info('finish ')

def on_tick(self，tick):

tick _ data=self。CTX[' tick _ data ']

有关代码，请参见tick_data.items()中的hist :

if hist[' ma10 ']1.05 * hist[' ma20 ']:

self.ctx.broker.buy(代码，hist.close，500，ttl=5)

if hist['ma10'] hist['ma20']和自我编码。CTX。经纪人。位置：

self.ctx.broker.sell(code，hist.close，200，ttl=1)

if __name__=='__main__':

从实用工具导入加载历史记录

feed={}

有关代码，请参见加载历史中的hist('000002 .SZ'):

# hist=hist.iloc[:100]

hist[' ma10 ']=hist。关闭。滚动(10 ).平均值()

hist[' ma20 ']=hist。关闭。滚动(20 ).平均值()

feed[code]=hist

mytest=MyBackTest(提要)

mytest.start()

order _ lst=我的测试。CTX。经纪人。订单_历史_ lst

打开(' report/order_hist.json '，' w ')为wf:

json.dump(order_lst，wf，indent=4，default=str)

stats=mytest.stat

统计数据。数据。to _ CSV(报告/统计。CSV)

"打印("策略收益：{:3f}% ' .format(stats.total_returns * 100))

"打印("更大回彻率： {:3f}% ' .格式(stats.max_dropdown * 100))

"打印("年化收益： {:3f}% ' .格式(stats.annual_return * 100))

"打印("夏普比率： {:3f} ' .格式(stats.sharpe))

绘图仪=绘图仪（进给、统计、顺序_ 1)

绘图仪。report('report/report.png ')