手把手教你用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 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)

下面以均值回归模型为例进行回测，关于均值回归模型更详细的内容可参考推文：手把手教你用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搭建自己的量化回测框架【均值回归策略】 。

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.
————————————————