01 引言手把手教你用Python搭建自己的量化回测框架【均值回归策略】 是使用矢量化方法（pandas）建立的基于研究的量化回测框架，不考虑交易的委托成交行为，与真实市场情况差距比较大。今天为大家介绍的是基于事件驱动的回测框架，这是一种十分复杂的回测系统，力图模拟实盘交易，搭建一种仿真的回测环境。 与矢量化方法相比，事件驱动的系统具有许多优点，一是事件驱动回测可以用于历史回测和实时交易，而矢量化的回测必须一次获得所有数据才能进行统计分析；二是使用事件驱动的回测不会出现前瞻性偏见，因为将市场数据接收视为“事件”，可以用市场数据“滴灌”来复制订单管理和投资组合系统的行为方式；三是事件驱动的回测允许对如何执行订单和产生交易成本进行定制。由于可以构建自定义交易处理程序，因此可以轻松处理基本的市场和限价订单。 尽管事件驱动的回测系统具有许多优点，但与简单的矢量化系统相比，两大缺点也比较突出：一是实施和测试要复杂得多，有更多的“活动组建”（模块），导致引入错误的机会更大；二是执行速度较慢，进行数学计算时，无法利用最佳的矢量化运算。下面仍然与均值回归交易策略为例，为大家展示Python基于事件驱动回测框架的构建思路，回测代码主要参考了《Mastering Python for Finance》Chapter 9 Backtesting，对市场数据获取使用了tushare作为替代，公众号后台回复'finance'可获取本书的电子书（英文版）。 02 回测框架与Python代码基于事件驱动的回测框架一般包括以下几个模块，（1）数据采集，数据采集模块通过接口获取行情数据和历史数据（这里使用tushare），产生市场数据事件。（2）事件模块，一般是设定一个事件基类，然后在事件的基类下面生成很多子事件，如市场数据事件、交易信号事件、委托下单事件和订单成交事件等。（3）策略模块，一般先设定一个策略基类，然后通过基类衍生很多子策略，该模块通过输入数据，生成交易信号（signal），即产生信号事件。（4）交易执行模块，接收信号事件，确定需要开仓和平仓的头寸数量，输出委托下单事件，根据委托下单事件进行模拟或者真实的交易，当订单成交事件完成时更新持有资产头寸以及其他相关数据。（5）资产头寸，记录资金、仓位、仓位市值等信息。最后，所有事件通过事件队列进行管理，当一个事件完成后，由下一个事件开始任务，不断循环。 Python基于事件驱动的回测系统主要使用面向对象（class）来编写，因此需要对类的基础要求比较高，关于Python的面向对象编程可以参考推文：【手把手教你】Python面向对象编程入门及股票数据管理应用实例。 #先引入后面可能用到的包（package）import pandas as pd import numpy as npimport matplotlib.pyplot as plt%matplotlib inline #正常显示画图时出现的中文和负号from pylab import mplmpl.rcParams['font.sans-serif']=['SimHei']mpl.rcParams['axes.unicode_minus']=False将每一个时间戳（timestamp）内的数据作为输入参数，构建类TickData。 根据数据要求，生成市场数据事件，这里主要获取收盘价、开盘价、成交量和时间戳。 import datetime as dtimport pandas as pdclass Backtester: def __init__(self, symbol, start_date, end_date): self.target_symbol = symbol self.start_dt = start_date self.end_dt = end_date self.strategy = None self.unfilled_orders = [] self.positions = dict() self.current_prices = None self.rpnl, self.upnl = pd.DataFrame(), pd.DataFrame() def get_timestamp(self): return self.current_prices.get_timestamp(self.target_symbol) def get_trade_date(self): timestamp = self.get_timestamp() return timestamp.strftime('%Y-%m-%d') def update_filled_position(self, symbol, qty, is_buy,price, timestamp): position = self.get_position(symbol) position.event_fill(timestamp, is_buy, qty, price) self.strategy.event_position(self.positions) self.rpnl.loc[timestamp, 'rpnl'] = position.realized_pnl print (self.get_trade_date(), \ '成交:', '买入' if is_buy else '卖出', \ qty, symbol, '价格', price) def get_position(self, symbol): if symbol not in self.positions: position = Position() position.symbol = symbol self.positions[symbol] = position return self.positions[symbol] def evthandler_order(self, order): self.unfilled_orders.append(order) print (self.get_trade_date(), \ '收到指令:', \ '买入' if order.is_buy else '卖出', order.qty, \ order.symbol) def match_order_book(self, prices): if len(self.unfilled_orders) > 0: self.unfilled_orders = \ [order for order in self.unfilled_orders if self.is_order_unmatched(order, prices)] def is_order_unmatched(self, order, prices): symbol = order.symbol timestamp = prices.get_timestamp(symbol) if order.is_market_order and timestamp > order.timestamp: # Order is matched and filled. order.is_filled = True open_price = prices.get_open_price(symbol) order.filled_timestamp = timestamp order.filled_price = open_price self.update_filled_position(symbol, order.qty, order.is_buy, open_price, timestamp) self.strategy.event_order(order) return False return True def evthandler_tick(self, prices): self.current_prices = prices self.strategy.event_tick(prices) self.match_order_book(prices) def start_backtest(self): self.strategy = MeanRevertingStrategy(self.target_symbol) self.strategy.event_sendorder = self.evthandler_order mds = MarketDataSource() mds.event_tick = self.evthandler_tick mds.ticker = self.target_symbol mds.start, mds.end = self.start_dt, self.end_dt print ('Backtesting started...') mds.start_market_simulation() print ('Completed.')获取市场数据并搭建市场模拟交易的状态。 交易指令和头寸管理。 class Order: def __init__(self, timestamp, symbol, qty, is_buy, is_market_order, price=0): self.timestamp = timestamp self.symbol = symbol self.qty = qty self.price = price self.is_buy = is_buy self.is_market_order = is_market_order self.is_filled = False self.filled_price = 0 self.filled_time = None self.filled_qty = 0class Position: def __init__(self): self.symbol = None self.buys, self.sells, self.net = 0, 0, 0 self.realized_pnl = 0 self.unrealized_pnl = 0 self.position_value = 0 def event_fill(self, timestamp, is_buy, qty, price): if is_buy: self.buys += qty else: self.sells += qty self.net = self.buys - self.sells changed_value = qty * price * (-1 if is_buy else 1) self.position_value += changed_value if self.net == 0: self.realized_pnl = self.position_value def update_unrealized_pnl(self, price): if self.net == 0: self.unrealized_pnl = 0 else: self.unrealized_pnl = price * self.net + self.position_value return self.unrealized_pnl下面以均值回归模型为例进行回测，关于均值回归模型更详细的内容可参考推文：手把手教你用Python搭建自己的量化回测框架【均值回归策略】 。 class MeanRevertingStrategy(Strategy): def __init__(self, symbol, lookback_intervals=20, buy_threshold=-1.5, sell_threshold=1.5): Strategy.__init__(self) self.symbol = symbol self.lookback_intervals = lookback_intervals self.buy_threshold = buy_threshold self.sell_threshold = sell_threshold self.prices = pd.DataFrame() self.is_long, self.is_short = False, False def event_position(self, positions): if self.symbol in positions: position = positions[self.symbol] self.is_long = True if position.net > 0 else False self.is_short = True if position.net < 0 else False def event_tick(self, market_data): self.store_prices(market_data) if len(self.prices) < self.lookback_intervals: return signal_value = self.calculate_z_score() timestamp = market_data.get_timestamp(self.symbol) if signal_value < self.buy_threshold: self.on_buy_signal(timestamp) elif signal_value > self.sell_threshold: self.on_sell_signal(timestamp) def store_prices(self, market_data): timestamp = market_data.get_timestamp(self.symbol) self.prices.loc[timestamp, 'close'] = \ market_data.get_last_price(self.symbol) self.prices.loc[timestamp, 'open'] = \ market_data.get_open_price(self.symbol) def calculate_z_score(self): self.prices = self.prices[-self.lookback_intervals:] returns = self.prices['close'].pct_change().dropna() z_score = ((returns-returns.mean())/returns.std())[-1] return z_score def on_buy_signal(self, timestamp): if not self.is_long: self.send_market_order(self.symbol, 100, True, timestamp) def on_sell_signal(self, timestamp): if not self.is_short: self.send_market_order(self.symbol, 100, False, timestamp)最后定义一个回测类，将上述模块串联到一起。回测系统中只是对策略交易的已实现收益（未实现收益）进行回测，并未加入收益率、夏普比率、最大回撤等策略评价指标，关于这方面内容可以参考手把手教你用Python搭建自己的量化回测框架【均值回归策略】 。 开始回测 backtester = Backtester('600000','20180101','20200323')backtester.start_backtest()backtester.rpnl.plot(figsize=(16,6))plt.show()策略已实现收益： 03 结语本文以均值回归模型为例，展示了基于事件驱动回测系统的Python实现过程。当然，上述回测系统仍然是一个简化版的系统，还存在很多需要完善的地方，比如没有加入关于策略的量化评价指标和可视化模块，没有考虑交易手续费等，这些都留待读者自己去思考和进一步完善。现实的市场交易比回测系统要复杂更多，因此回测系统再怎么完美也很难完全复现真实交易的场景。量化回测是量化交易的重要组成部分，回测系统的好坏会直接影响对策略的评估。目前很多量化平台和Python量化回测开源框架都提供了相应的回测系统，大家也没必要都自己去重新造轮子，但是对于Python基础比较扎实，从事个人量化交易的来说，了解回测系统的运作过程，构建自己的量化交易系统还是很有必要的。后续推文将会介绍Python量化回测开源框架的应用。 参考资料：Weiming J M, Weiming J M. Mastering Python for Finance[M]. 2015. |
|
|
