量化交易初探（三）——择股与回测

显示全部楼层 · 2023-9-24 20:02:48

量化交易初探（三）——择股与回测

在之前的两篇文章中，我们分别展示了基于baostock模块的A股数据获取脚本和基于backtrader模块的简单择时回测框架，详见：
在之前代码的基础上，我们将要添加择股回测的功能，这样就把量化交易的范围拓展到全市场，进一步丰富了策略选择的空间。
与之前重复的代码不再展示，这里仅放上修改的部分供参考。为了方便，这次把数据获取的部分也写到回测脚本中了，从代码规整的角度还是建议分开：
首先是主程序部分：
if __name__ == '__main__':

# 主程序其他部分省略
...
dataset = []
codeset = []
for item in alternative_code:  # 获取备选股池中个股数据
      rs = get_share_info(item, date_start, date_end, adjustflag="3")
      data, code = create_dataset(rs)
      dataset.append(data)
      codeset.append(code)

trade_info = pd.read_csv("trade_info.csv", parse_dates=['trade_date'])

cerebro, results = run_backtest(Strategy_MA, dataset, codeset, startcash, date_start, date_end)
数据获取和预处理：
def get_share_info(code, start_date, end_date, adjustflag="1"):
# 读取日频K线数据

fields = "date,code,open,high,low,close,volume"

rs = bs.query_history_k_data_plus(code=code,
      fields=fields, start_date=start_date, end_date=end_date, # 为空默认从2015-01-01到最近一个交易日
      frequency='d', adjustflag=adjustflag)
      # adjustflag 复权类型，不修改则默认不复权
print('query_history_k_data_plus respond error_code:'+rs.error_code)
print('query_history_k_data_plus respond error_msg:'+rs.error_msg)
data_list = []
while (rs.error_code == '0') & rs.next():
      # 获取一条记录，将记录合并在一起
      data_list.append(rs.get_row_data())
result = pd.DataFrame(data_list, columns=rs.fields)

return result

def create_dataset(data_bef):
# 将所有备选股票数据导入同一数据集

# 日期对齐
date = pd.DataFrame(index=data_bef.index.unique()) # 获取回测区间内所有交易日
code = data_bef.loc[0, 'code']
df = data_bef[['date','open','high','low','close','volume']].copy()
df.loc[:,['open','high','low','close']] = df.loc[:,['open','high','low','close']].copy().astype(float)
df.loc[:,['volume']] = df.loc[:,['volume']].copy().astype(int)
data = pd.merge(date, df, left_index=True, right_index=True, how='left')
data['date'] = pd.to_datetime(data['date']) # 这是新添加的行
data.set_index(['date'],inplace=True)
# 缺失值处理：
data.loc[:,['volume']] = data.loc[:,['volume']].fillna(0)
data.loc[:,['open','high','low','close']] = data.loc[:,['open','high','low','close']].fillna(method='pad')
data.loc[:,['open','high','low','close']] = data.loc[:,['open','high','low','close']].fillna(0)

return data, code
回测模块：
class Strategy_MA(bt.Strategy):
# 省略其他部分
...
def next(self):

      dt = self.datas[0].datetime.date(0) # 获取当前的回测时间点
      # 如果是调仓日，则进行调仓操作
      if dt in self.trade_dates:
         print("--------------{} 调仓----------".format(dt))
         if len(self.order_list) > 0:
            for od in self.order_list:
                  self.cancel(od) # 撤销未完成订单
            self.order_list = [] # 重置订单列表
         # 提取当前调仓日的持仓列表
         buy_stocks_data = self.buy_stock.query(f"trade_date=='{dt}'")
         long_list = buy_stocks_data['sec_code'].tolist()
         print('long_list', long_list) # 打印持仓列表
         # 不再继续持有的股票进行平仓
         sell_stock = [i for i in self.buy_stocks_pre if i not in long_list]
         print('sell_stock', sell_stock) # 打印平仓列表
         if len(sell_stock) > 0:
            print("-----------平仓--------------")
            for stock in sell_stock:
                  data = self.getdatabyname(stock)
                  if self.getposition(data).size > 0 :
                     od = self.close(data=data)
                     self.order_list.append(od) # 记录卖出订单
         print("-----------买入此次调仓的股票--------------")
         for stock in long_list:
            w = buy_stocks_data.query(f"sec_code=='{stock}'")['weight'].iloc[0] # 提取持仓权重
            data = self.getdatabyname(stock)
            order = self.order_target_percent(data=data, target=w*0.95) # 为减少可用资金不足的情况，留 5% 的现金做备用，该函数以多退少补的方式进行购买/卖出
            print()
            self.order_list.append(order)

         self.buy_stocks_pre = long_list # 保存此次调仓的股票列表
回测执行：
def run_backtest(strategy, dataset, codeset, startcash, start, end):
# 省略其他部分
...
for item in range(len(dataset)):
      code = codeset[item]
      cerebro.adddata(bt.feeds.PandasData(dataname=dataset[item],fromdate=datetime.strptime(start, '%Y-%m-%d'),todate=datetime.strptime(end, '%Y-%m-%d')), name=code)
# 添加策略
cerebro.addstrategy(strategy, end_date=datetime.strptime(end, '%Y-%m-%d'))
results = cerebro.run()

return cerebro, results
可以看到，这次代码中并没有直接编写择股策略，而是导入已经确定好的调仓信息"trade_info.csv"进行回测，这样我们就可以将相对复杂的策略推理部分放在另一个脚本中编写，以提升回测程序的执行效率。关于择股策略编写的部分将在之后的文章中分享，欢迎感兴趣的朋友多多批评指正。