您当前的位置:python

陈枫燕清羽最新章节免费阅读-用Python徒手撸一个股票回测框架

发布日期:2021-10-13编辑:python分类:python

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

    什么是回测框架?

    无论是传统股票陈枫燕清羽最新章节免费阅读还是量化陈枫燕清羽最新章节免费阅读,无法避免的一个问题是我们需要检验自己的陈枫燕清羽最新章节免费阅读策略是否可行,而最简单的方式就是利用历史数据检验陈枫燕清羽最新章节免费阅读策略,而回测框架就是提供这样的一个平台让陈枫燕清羽最新章节免费阅读策略在历史数据中不断陈枫燕清羽最新章节免费阅读,最终生成最终结果,通过查看结果的策略收益,年化收益,最大回测等用以评估陈枫燕清羽最新章节免费阅读策略的可行性。

    代码地址在最后。

    本项目并不是一个已完善的项目,还在不断的完善。

    回测框架

    回测框架应该至少包含两个部分,回测类,陈枫燕清羽最新章节免费阅读类.

    回测类提供各种钩子函数,用于放置自己的陈枫燕清羽最新章节免费阅读逻辑,陈枫燕清羽最新章节免费阅读类用于模拟市场的陈枫燕清羽最新章节免费阅读平台,这个类提供买入,卖出的方法。

    代码架构

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

    backtest/

    backtest.py

    broker.py

    backtest.py主要提供BackTest这个类用于提供回测框架,暴露以下钩子函数.

    definitialize(self):

    """在回测开始前的初始化"""

    pass

    defbefore_on_tick(self,tick):

    pass

    defafter_on_tick(self,tick):

    pass

    defbefore_trade(self,order):

    """在陈枫燕清羽最新章节免费阅读之前会调用此函数

    可以在此放置资金管理及风险管理的代码

    如果返回True就允许陈枫燕清羽最新章节免费阅读,否则放弃陈枫燕清羽最新章节免费阅读

    """

    returnTrue

    defon_order_ok(self,order):

    """当订单执行成功后调用"""

    pass

    defon_order_timeout(self,order):

    """当订单超时后调用"""

    pass

    deffinish(self):

    """在回测结束后调用"""

    pass

    @abstractmethod

    defon_tick(self,bar):

    """

    回测实例必须实现的方法,并编写自己的陈枫燕清羽最新章节免费阅读逻辑

    """

    pass

    玩过量化平台的回测框架或者开源框架应该对这些钩子函数不陌生,只是名字不一样而已,大多数功能是一致的,除了on_tick.

    之所以是on_tick而不是on_bar,是因为我希望陈枫燕清羽最新章节免费阅读逻辑是一个一个时间点的参与陈枫燕清羽最新章节免费阅读,在这个时间点我可以获取所有当前时间的所有股票以及之前的股票数据,用于判断是否陈枫燕清羽最新章节免费阅读,而不是一个时间点的一个一个股票参与陈枫燕清羽最新章节免费阅读逻辑。

    而broker.py主要提供buy,sell两个方法用于陈枫燕清羽最新章节免费阅读。

    defbuy(self,code,price,shares,ttl=-1):

    """

    限价提交买入订单

    ---------

    Parameters:

    code:str

    股票代码

    price:floatorNone

    最高可买入的价格,如果为None则按市价买入

    shares:int

    买入股票数量

    ttl:int

    订单允许存在的最大时间,默认为-1,永不超时

    ---------

    return:

    dict

    {

    "type":订单类型,"buy",

    "code":股票代码,

    "date":提交日期,

    "ttl":存活时间,当ttl等于0时则超时,往后不会在执行

    "shares":目标股份数量,

    "price":目标价格,

    "deal_lst":陈枫燕清羽最新章节免费阅读成功的历史数据,如

    [{"price":成交价格,

    "date":成交时间,

    "commission":陈枫燕清羽最新章节免费阅读手续费,

    "shares":成交份额

    }]

    ""

    }

    """

    ifpriceisNone:

    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)

    returnorder

    defsell(self,code,price,shares,ttl=-1):

    """

    限价提交卖出订单

    ---------

    Parameters:

    code:str

    股票代码

    price:floatorNone

    最低可卖出的价格,如果为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":陈枫燕清羽最新章节免费阅读收益}]

    ""

    }

    """

    ifcodenotinself.position:

    return

    ifpriceisNone:

    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)

    returnorder

    由于我很讨厌抽象出太多类,抽象出太多类及方法,我怕我自己都忘记了,所以对于陈枫燕清羽最新章节免费阅读的选择都是尽可能的使用常用的数据结构,如list,dict.

    这里用一个dict代表一个订单。

    上面的这些方法保证了一个回测框架的基本陈枫燕清羽最新章节免费阅读逻辑,而回测的运行还需要一个调度器不断的驱动这些方法,这里的调度器如下。

    classScheduler(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=[]

    defrun(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方法

    forrunnerinrunner_lst:

    runner.ctx=self.ctx

    runner.initialize()

    #创建陈枫燕清羽最新章节免费阅读日历

    if"trade_cal"notinself.ctx:

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

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

    #通过遍历陈枫燕清羽最新章节免费阅读日历的时间依次调用runner

    #首先调用所有pre-hook的run方法

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

    #最后调用post-hook的run方法

    fortickinself.ctx.trade_cal:

    self.ctx.set_currnet_time(tick)

    forrunnerinrunner_lst:

    runner.run(tick)

    #循环结束后调用所有runner陈枫燕清羽最新章节免费阅读的finish方法

    forrunnerinrunner_lst:

    runner.finish()

    在Backtest类实例化的时候就会自动创建一个调度器陈枫燕清羽最新章节免费阅读,然后通过Backtest实例的start方法就能启动调度器,而调度器会根据历史数据的一个一个时间戳不断驱动Backtest,Broker实例被调用。

    为了处理不同实例之间的数据访问隔离,所以通过一个将一个Context陈枫燕清羽最新章节免费阅读绑定到Backtest,Broker实例上,通过self.ctx访问共享的数据,共享的数据主要包括feed陈枫燕清羽最新章节免费阅读,即历史数据,一个数据结构如下的字典陈枫燕清羽最新章节免费阅读。

    {code1:pd.DataFrame,code2:pd.DataFrame}

    而这个Context陈枫燕清羽最新章节免费阅读也绑定了Broker,Backtest的实例,这就可以使得数据访问接口统一,但是可能导致数据访问混乱,这就要看策略者的使用了,这样的一个好处就是减少了一堆代理方法,通过添加方法去访问其他的陈枫燕清羽最新章节免费阅读的方法,真不嫌麻烦,那些人。

    绑定及Context陈枫燕清羽最新章节免费阅读代码如下:

    classContext(UserDict):

    def__getattr__(self,key):

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

    returnself[key]

    defset_currnet_time(self,tick):

    self["now"]=tick

    tick_data={}

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

    forcode,histinself["feed"].items():

    df=hist[hist.index==tick]

    iflen(df)==1:

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

    self["tick_data"]=tick_data

    defget_hist(self,code=None):

    """如果不指定code,获取截至到当前时间的所有股票的历史数据"""

    ifcodeisNone:

    hist={}

    forcode,histinself["feed"].items():

    hist[code]=hist[hist.index<=self.now]

    elifcodeinself.feed:

    return{code:self.feed[code]}

    returnhist

    classScheduler(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=[]

    defadd_feed(self,feed):

    self.ctx["feed"]=feed

    defadd_hook(self,hook,typ="post"):

    iftyp=="post"andhooknotinself._post_hook_lst:

    self._post_hook_lst.append(hook)

    eliftyp=="pre"andhooknotinself._pre_hook_lst:

    self._pre_hook_lst.append(hook)

    defadd_broker(self,broker):

    self.ctx["broker"]=broker

    defadd_backtest(self,backtest):

    self.ctx["backtest"]=backtest

    #简写

    self.ctx["bt"]=backtest

    defadd_runner(self,runner):

    ifrunnerinself._runner_lst:

    return

    self._runner_lst.append(runner)

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

    这里创建了一个Stat的Hook陈枫燕清羽最新章节免费阅读,用于统计资产变化。

    classStat(Base):

    def__init__(self):

    self._date_hist=[]

    self._cash_hist=[]

    self._stk_val_hist=[]

    self._ast_val_hist=[]

    self._returns_hist=[]

    defrun(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

    defdata(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"

    returndf

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

    defget_dropdown(self):

    high_val=-1

    low_val=None

    high_index=0

    low_index=0

    dropdown_lst=[]

    dropdown_index_lst=[]

    foridx,valinenumerate(self._ast_val_hist):

    ifval>=high_val:

    ifhigh_val==low_valorhigh_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

    iflow_valisNone:

    low_val=val

    low_index=idx

    ifval

    low_val=val

    low_index=idx

    iflow_index>high_index:

    dropdown=(high_val-low_val)/high_val

    dropdown_lst.append(dropdown)

    dropdown_index_lst.append((high_index,low_index))

    returndropdown_lst,dropdown_index_lst

    @property

    defmax_dropdown(self):

    """最大回车率"""

    dropdown_lst,dropdown_index_lst=self.get_dropdown()

    iflen(dropdown_lst)>0:

    returnmax(dropdown_lst)

    else:

    return0

    @property

    defannual_return(self):

    """

    年化收益率

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

    v:最终价值

    c:初始价值

    D:有效投资时间(365)

    注:虽然投资股票只有250天,但是持有股票后的非陈枫燕清羽最新章节免费阅读日也没办法投资到其他地方,所以这里我取365

    参考:%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

    returnret

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

    陈枫燕清羽最新章节免费阅读历史数据

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

    回测报告

    回测报告我也放在了回测框架之外,这里写了一个Plottter的陈枫燕清羽最新章节免费阅读用于绘制一些回测指标等。结果如下:

    用Python徒手撸一个股票回测框架

    回测示例

    下面是一个回测示例。

    importjson

    frombacktestimportBackTest

    fromreporterimportPlotter

    classMyBackTest(BackTest):

    definitialize(self):

    "initialize")

    deffinish(self):

    "finish")

    defon_tick(self,tick):

    tick_data=self.ctx["tick_data"]

    forcode,histintick_data.items():

    ifhist["ma10"]>1.05*hist["ma20"]:

    self.ctx.broker.buy(code,hist.close,500,ttl=5)

    ifhist["ma10"]

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

    if__name__=='__main__':

    fromutilsimportload_hist

    feed={}

    forcode,histinload_hist(".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

    withopen("report/order_hist.json","w")aswf:

    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")

    顶一下

    (0)

    0%

    踩一下

    (0)

    0%