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, 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

策略的基类，其他策略都基于该策略进行编写。

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

最后定义一个回测类，将上述模块串联到一起。回测系统中只是对策略交易的已实现收益(未实现收益)进行回测，并未加入收益率、夏普比率、最大回撤等策略评价指标。

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",'20180101','20200323')

backtester.start_backtest()

Backtesting started...

2019-01-31 收到指令: 卖出 100 600000

2019-02-01 成交: 卖出 100 600000 价格 10.82

2019-02-15 收到指令: 买入 100 600000

2019-02-18 成交: 买入 100 600000 价格 10.75

......

2020-03-09 收到指令: 买入 100 600000

2020-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]. 2015.