01引言
使用矢量化方法(pandas)建立的基于研究的量化回测框架，不考虑交易的委托成交行为，与真实市场情况差距比较大。今天为大家介绍的是基于事件驱动的回测框架，这是一种十分复杂的回测系统，力图模拟实盘交易，搭建一种仿真的回测环境。
与矢量化方法相比，事件驱动的系统具有许多优点，一是事件驱动回测可以用于历史回测和实时交易，而矢量化的回测必须一次获得所有数据才能进行统计分析；二是使用事件驱动的回测不会出现前瞻性偏见，因为将市场数据接收视为“事件”，可以用市场数据“滴灌”来复制订单管理和投资组合系统的行为方式；三是事件驱动的回测允许对如何执行订单和产生交易成本进行定制。由于可以构建自定义交易处理程序，因此可以轻松处理基本的市场和限价订单。
尽管事件驱动的回测系统具有许多优点，但与简单的矢量化系统相比，两大缺点也比较突出：一是实施和测试要复杂得多，有更多的“活动组建”(模块)，导致引入错误的机会更大；二是执行速度较慢，进行数学计算时，无法利用最佳的矢量化运算。下面仍然与均值回归交易策略为例，为大家展示Python基于事件驱动回测框架的构建思路，回测代码主要参考了《Mastering Python for Finance》Chapter 9 Backtesting，对市场数据获取使用了tushare作为替代。
02 回测框架与Python代码
基于事件驱动的回测框架一般包括以下几个模块，(1)数据采集，数据采集模块通过接口获取行情数据和历史数据(这里使用tushare)，产生市场数据事件。(2)事件模块，一般是设定一个事件基类，然后在事件的基类下面生成很多子事件，如市场数据事件、交易信号事件、委托下单事件和订单成交事件等。(3)策略模块，一般先设定一个策略基类，然后通过基类衍生很多子策略，该模块通过输入数据，生成交易信号(signal)，即产生信号事件。(4)交易执行模块，接收信号事件，确定需要开仓和平仓的头寸数量，输出委托下单事件，根据委托下单事件进行模拟或者真实的交易，当订单成交事件完成时更新持有资产头寸以及其他相关数据。(5)资产头寸，记录资金、仓位、仓位市值等信息。最后，所有事件通过事件队列进行管理，当一个事件完成后，由下一个事件开始任务，不断循环。
Python基于事件驱动的回测系统主要使用面向对象(class)来编写，因此需要对类的基础要求比较高。
#先引入后面可能用到的包(package)
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
%matplotlib inline
#正常显示画图时出现的中文和负号
from pylab import mpl
mpl.rcParams['font.sans-serif']=['SimHei']
mpl.rcParams['axes.unicode_minus']=False
将每一个时间戳(timestamp)内的数据作为输入参数，构建类TickData。
class TickData:
def __init__(self, symbol, timestamp,last_price=0, total_volume=0):
self.symbol = symbol
self.timestamp = timestamp
self.open_price = 0
self.last_price = last_price
self.total_volume = total_volume
根据数据要求，生成市场数据事件，这里主要获取收盘价、开盘价、成交量和时间戳。
class MarketData:
def __init__(self):
self.__recent_ticks__ = dict()
def add_last_price(self, time, symbol, price, volume):
tick_data = TickData(symbol, time, price, volume)
self.__recent_ticks__[symbol] = tick_data
def add_open_price(self, time, symbol, price):
tick_data = self.get_existing_tick_data(symbol, time)
tick_data.open_price = price
def get_existing_tick_data(self, symbol, time):
if not symbol in self.__recent_ticks__:
tick_data = TickData(symbol, time)
self.__recent_ticks__[symbol] = tick_data
return self.__recent_ticks__[symbol]
def get_last_price(self, symbol):
return self.__recent_ticks__[symbol].last_price
def get_open_price(self, symbol):
return self.__recent_ticks__[symbol].open_price
def get_timestamp(self, symbol):
return self.__recent_ticks__[symbol].timestamp
获取市场数据并搭建市场模拟交易的状态。
#获取数据
import tushare as ts
class MarketDataSource:
def __init__(self):
self.event_tick = None
self.ticker = None
self.autype='qfq'
self.start, self.end = None, None
self.md = MarketData()
def start_market_simulation(self):
data = ts.get_k_data(self.ticker,autype=self.autype,
start=self.start, end=self.end)
data.index=pd.to_datetime(data.date)
data=data.sort_index()
for time, row in data.iterrows():
self.md.add_last_price(time, self.ticker,
row["close"], row["volume"])
self.md.add_open_price(time, self.ticker, row["open"])
if not self.event_tick is None:
self.event_tick(self.md)
交易指令和头寸管理。
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 = 0
class Position:
def __init__(self):
self.symbol = None
self.buys, self.sells, = 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.buys - self.sells
changed_value = qty * price * (-1 if is_buy else 1)
self.position_value += changed_value
if == 0:
self.realized_pnl = self.position_value
def update_unrealized_pnl(self, price):
if == 0:
self.unrealized_pnl = 0
else:
self.unrealized_pnl = price * + self.position_value
return self.unrealized_pnl
策略的基类，其他策略都基于该策略进行编写。
class Strategy:
def __init__(self):
self.event_sendorder = None
def event_tick(self, market_data):
pass
def event_order(self, order):
pass
def event_position(self, positions):
pass
def send_market_order(self, symbol, qty, is_buy, timestamp):
if not self.event_sendorder is None:
order = Order(timestamp, symbol, qty, is_buy, True)
self.event_sendorder(order)
下面以均值回归模型为例进行回测，关于均值回归模型更详细的内容可参考推文：
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 > 0 else False
self.is_short = True if < 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)
最后定义一个回测类，将上述模块串联到一起。回测系统中只是对策略交易的已实现收益(未实现收益)进行回测，并未加入收益率、夏普比率、最大回撤等策略评价指标。
import datetime as dt
import pandas as pd
class 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.")
开始回测
backtester = Backtester("600000",'0101','')
backtester.start_backtest()
Backtesting started...
-01-31 收到指令: 卖出 100 600000
-02-01 成交: 卖出 100 600000 价格 10.82
-02-15 收到指令: 买入 100 600000
-02-18 成交: 买入 100 600000 价格 10.75
......
-03-09 收到指令: 买入 100 600000
-03-10 成交: 买入 100 600000 价格 10.71
Completed.
backtester.rpnl.plot(figsize=(16,6))
plt.show()
策略已实现收益：
03 结语
本文以均值回归模型为例，展示了基于事件驱动回测系统的Python实现过程。当然，上述回测系统仍然是一个简化版的系统，还存在很多需要完善的地方，比如没有加入关于策略的量化评价指标和可视化模块，没有考虑交易手续费等，这些都留待读者自己去思考和进一步完善。现实的市场交易比回测系统要复杂更多，因此回测系统再怎么完美也很难完全复现真实交易的场景。量化回测是量化交易的重要组成部分，回测系统的好坏会直接影响对策略的评估。目前很多量化平台和Python量化回测开源框架都提供了相应的回测系统，大家也没必要都自己去重新造轮子，但是对于Python基础比较扎实，从事个人量化交易的来说，了解回测系统的运作过程，构建自己的量化交易系统还是很有必要的。后续推文将会介绍Python量化回测开源框架的应用。
参考资料：Weiming J M, Weiming J M. Mastering Python for Finance[M]. .
手把手教你python实现量价形态选股知乎_【手把手教你】Python实现基于事件驱动的量化回测...
