原创文章第101篇,专注“个人成长与财富自由、世界运作的逻辑, AI量化投资”。 1000天行动计划之第101天,开启第2个100天计划进程! 我们回到主线——backtrader。
AI量化框架的主体是backtrader的事件驱动回测以及实盘能力。外加wxPython的GUI,支持csv、hdf5,Arctic(mongo),qlib的数据存储能力——这里也会把qlib框架整合进来,但不会用它的全流程——我把它的“因子表达式”能力独立拆分出来,不依赖它自有的数据格式,可以兼容csv,hdf5,mongo等。 01 backtrader策略
为了与“向量化回测”整合:qlib的AI导向的框架与backtrader的事件驱动与实盘做对比,使用backtrader来走一下“双均线”的策略流程。
cerebro = bt.Cerebro(maxcpus=1) # 在notebook里运作,maxcpus=1很重要 cerebro.broker.setcash(100000.0) #
print('初始总市值: %.2f' % cerebro.broker.getvalue())
cerebro.run()
print('期末总市值: %.2f' % cerebro.broker.getvalue()) 添加数据源,使用pandasdata import pandas as pd from datetime import datetime
symbol = 'SPX' start = datetime(2010,1,1) end = datetime(2022,9,23)
with pd.HDFStore('../../data/hdf5/index.h5') as store: df = store[symbol]
df['openinterest'] = 0 df = df[['open', 'high', 'low', 'close', 'volume', 'openinterest']]
data = bt.feeds.PandasData(dataname=df, name=symbol, fromdate=start, todate=end) cerebro.adddata(data) 双均线策略: class SMAStrategy(bt.Strategy):
def __init__(self): self.sma_short = bt.indicators.SMA(period=42) self.sma_long = bt.indicators.SMA(period=252) #self.sma = btind.SimpleMovingAverage(period=15)
def next(self): #print(self.data0[0], self.sma_short[0], self.sma_long[0]) if self.sma_short[0] > self.sma_long[0]: self.buy() else: self.close() ![](http://image109.360doc.com/DownloadImg/2023/10/1215/273692764_1_20231012033920617_wm.png)
02 backtrader通用模板 # encoding:utf8 from datetime import datetime import backtrader as bt import pandas as pd from loguru import logger
class BacktraderEngine: def __init__(self, init_cash=1000000.0, benchmark='000300.SH', start=datetime(2010, 1, 1), end=datetime.now().date()): self.init_cash = init_cash self.start = start self.end = end self.benchmark = benchmark
cerebro = bt.Cerebro() cerebro.broker.setcash(init_cash)
# 设置手续费 cerebro.broker.setcommission(0.0001) # 滑点:双边各 0.0001 cerebro.broker.set_slippage_perc(perc=0.0001)
self.cerebro = cerebro self.cerebro.addanalyzer(bt.analyzers.PyFolio, _name='_PyFolio') 添加数据:
def add_data(self, code): # 加载数据 df = self.feed.get_df(code) df = to_backtrader_dataframe(df) data = bt.feeds.PandasData(dataname=df, name=code, fromdate=self.start, todate=self.end)
self.cerebro.adddata(data) self.cerebro.addobserver(bt.observers.Benchmark, data=data) self.cerebro.addobserver(bt.observers.TimeReturn) 策略模板: 策略里,notify_order, notify_trade两个函数,一个是订阅提交时触发,一个是交易产生时触发,这个在策略调试时特别有用,而且是通用的,直接写在基类里,另外就是基数实现log函数 # encoding: utf8 import backtrader as bt from loguru import logger import pandas as pd
class StrategyBase(bt.Strategy): def log(self, txt, dt=None): dt = dt or self.datas[0].datetime.date(0) logger.info('%s, %s' % (dt.isoformat(), txt))
def get_current_dt(self): return self.datas[0].datetime.date(0)
# 打印订单日志 def notify_order(self, order):
order_status = ['Created', 'Submitted', 'Accepted', 'Partial', 'Completed', 'Canceled', 'Expired', 'Margin', 'Rejected'] # 未被处理的订单 if order.status in [order.Submitted, order.Accepted]: self.log('未处理订单:订单号:%.0f, 标的: %s, 状态状态: %s' % (order.ref, order.data._name, order_status[order.status])) # 已经处理的订单 if order.status in [order.Partial, order.Completed]: if order.isbuy(): self.log( 'BUY EXECUTED, 状态: %s, 订单号:%.0f, 标的: %s, 数量: %.2f, 价格: %.2f, 成本: %.2f, 手续费 %.2f' % (order_status[order.status], # 订单状态 order.ref, # 订单编号 order.data._name, # 股票名称 order.executed.size, # 成交量 order.executed.price, # 成交价 order.executed.value, # 成交额 order.executed.comm)) # 佣金 else: # Sell self.log( 'SELL EXECUTED, status: %s, ref:%.0f, name: %s, Size: %.2f, Price: %.2f, Cost: %.2f, Comm %.2f' % (order_status[order.status], order.ref, order.data._name, order.executed.size, order.executed.price, order.executed.value, order.executed.comm))
elif order.status in [order.Canceled, order.Margin, order.Rejected, order.Expired]: # order.Margin资金不足,订单无法成交 # 订单未完成 self.log('未完成订单,订单号:%.0f, 标的 : %s, 订单状态: %s' % ( order.ref, order.data._name, order_status[order.status]))
self.order = None
def notify_trade(self, trade): logger.debug('trade......',trade.status) # 交易刚打开时 if trade.justopened: self.log('开仓, 标的: %s, 股数: %.2f,价格: %.2f' % ( trade.getdataname(), trade.size, trade.price)) # 交易结束 elif trade.isclosed: self.log('平仓, 标的: %s, GROSS %.2f, NET %.2f, 手续费 %.2f' % ( trade.getdataname(), trade.pnl, trade.pnlcomm, trade.commission)) # 更新交易状态 else: self.log('交易更新, 标的: %s, 仓位: %.2f,价格: %.2f' % ( trade.getdataname(), trade.size, trade.price)) 这样我们在写一个新策略时,直接继承使用,代码就非常简洁:
class StrategyPickTime(StrategyBase): params = ( ('long', 252), ('short', 42), )
def __init__(self): # bt.ind.EMA(self.data, period=self.p.ema_period) self.long = bt.ind.SMA(period=self.p.long) self.short = bt.ind.SMA(period=self.p.short)
def next(self): if self.short[0] > self.long[0]: self.buy() else: self.close() 运行时,可以打印出所有的日志: ![](http://image109.360doc.com/DownloadImg/2023/10/1215/273692764_2_20231012033921179_wm.png)
03 “模块化算子”
这个才是“积木式”策略开发的精华。
我们连上面的代码都不想写,想直接从“仓库”里复用。 我们把策略逻辑拆解成:选标的、按信号筛选,排序,仓位分配等等。 先从最简单的“等权重买入并持有,每个季度动态再平衡”,这个经典的资本配置策略开始。 这里拆分成“算子”会是: 1、每个季度运行一次(含首次)
2、选标的(选择所有) 3、仓位分配(等权)
我们在策略模板的基础上,再继承实现”算子策略模板“。 传入一个算子列表,在策略next函数里,逐个执行每个algo,若遇到某个algo返回True,则本次运行结果。 class StratgeyAlgoBase(StrategyBase): def __init__(self, algo_list=None): self.algos = algo_list
def next(self): context = { 'strategy': self }
for algo in self.algos: if algo(context) is True: # 如果algo返回True,直接不运行,本次不调仓 return 有了这个模板,结合前面说的三个步骤: 我们不需要写代码就可以完成这个策略! e.cerebro.addstrategy(StratgeyAlgoBase, algo_list=[ RunQuarterly(), SelectAll(), WeightEqually ]) ![](http://image109.360doc.com/DownloadImg/2023/10/1215/273692764_3_20231012033921492_wm.png)
我们只需要持续封装算子库,可以最大化减少开发策略所需要的时间,把时间多多用在策略的思想上吧。 后续我们再来介绍每个“算子”如何设计。 所有代码、数据请前往星球下载。 整合:qlib的AI导向的框架与backtrader的事件驱动与实盘
【研报复现】基于半衰主成分风险平价模型的全球资产配置策略研究
知识星球
|