易经量化交易系统之回测系统2

Posted 2022-12-03 最老程序员闫涛

订单类

当回测系统运行时，我们所开发的策略，会根据市场行情数据，生成相应的买入、卖出操作，这时就会产生订单。订单包括时间戳、交易标的代码、买卖操作、数量、价格、头寸等信息，产生订单后，会将订单提交给broker来进行处理，会进一步添加执行时间、交易价格、交易数量信息。我们这里先只处理市场价订单，其他类型订单，如限制、终止订单等，可以在这个基本系统上进行扩展。

class Order(object):
    # 订单类型定义
    OT_MARKET_ORDER = 1001 # 市价类型订单
    # 订单状态定义
    OS_ISSUED = 2001
    OS_FILLED = 2002

    def __init__(self, timestamp, symbol, quant, is_buy, 
                order_type=OT_MARKET_ORDER, price=0.0):
        self.name = 'fas.bktr.Order'
        self.timestamp = timestamp
        self.symbol = symbol
        self.quant = quant
        self.is_buy = is_buy
        self.order_type = order_type
        self.price = price
        self.order_state = Order.OS_ISSUED
        self.filled_time = None
        self.filled_price = 0.0
        self.filled_quant = 0

• 第3行：定义订单类型常量之市价订单，后面如果我们要扩展系统，可以定义更多的订单类型；
• 第5、6行：定义订单状态常量，分别为订单生成和订单完成；

订单包括下单时间，是否是购买订单，下单价格（在具体执行时可能是市场的当前价格），下单数量（可能是市场上最大可满足的数量），当订单执行完成时，需要有订单完成时间，实际执行价格，实际完成数量等信息。

仓位

仓位维护我们当前的仓位和账户余额，已实现损益和未实现损益：

class Position(object):
    def __init__(self):
        self.name = 'fas.bktr.Position'
        self.symbol = None
        self.buy_quants, self.sell_quants, self.net_quants = 0, 0, 0
        self.realized_pnl = 0.0
        self.unrealized_pnl = 0.0
        self.position_value = 0.0

    def event_fill(self, timestamp, is_buy, quant, price):
        if is_buy:
            self.buy_quants += quant
        else:
            self.sell_quants += quant
        self.net_quants = self.buy_quants - self.sell_quants
        changed_value = quant * price * (-1 if is_buy else 1)
        self.position_value += changed_value
        if self.net_quants == 0:
            self.realized_pnl = self.position_value

    def update_unrealized_pnl(self, price):
        if self.net_quants == 0:
            self.unrealized_pnl = 0
        else:
            self.unrealized_pnl = price * self.net_quants + \\
                        self.position_value
        return self.unrealized_pnl

• 第5行：定义累计买入数量（股），累计卖出数量和持有数量；
• 第6行：已实现的损益，pnl代表profile and loss；
• 第7行：未实现的损益，通常是由于持有标的产生的；
• 第10~17行：以买入为例，增加累计买入数量和持有数量，资金变化量为负值，代表资金将减少；若为卖出时，增加累计卖出数量，减少持有数量，资金会增加；如果持有量为0时，则资金（position_value）就全部是已实现损益；
• 第21~27行：当持有标的价格变动时，未实现损益会发生变化。公式为当前资金再加上当前标的价格乘以持有数量；

策略

策略基类

我们编写回测平台的目的，就是要尽可能真实的测试我们的策略在市场环境下的表现，因此策略是回测系统的重要核心功能。我们在实际中，会根据不同的理论，开发不同的策略，因此我们先定一个策略的基类：

class Strategy(object):
    def __init__(self):
        self.name = 'fas.bktr.Strategy'
        self.event_send_order = None

    def event_tick(self, market_data):
        pass

    def event_order(self, order):
        pass

    def event_position(self, positions):
        pass

    def send_order(self, timestamp, symbol, is_buy, quant):
        if self.event_send_order is not None:
            order = Order(timestamp, symbol, quant, is_buy, 
                        order_type=Order.OT_MARKET_ORDER, 
                        price=0.0)
            self.event_send_order(order)

• 第4行：定义订单发送事件响应函数；
• 第6行：还记得在MarketDataSource.start_market_simulation方法中，每个Tick时间点都会调用一个event_tick函数，这里就是这个函数的定义，因为这里是策略类的虚基类，因此该函数为空，需要具体的策略类给出具体的实现；
• 第9行：当订单状态发生变化时，回测系统会调用这个方法；
• 第12行：当仓位发生变化时，回测系统会调用这个方法；
• 第15~20行：生成订单，并将订单事件发布到系统中去；

具体策略类

具体的策略在具体的策略类中实现，在这里我们实现一个均值回归策略：

class MeanRevertingStrategy(Strategy):
    def __init__(self, symbol, lookback_intervals=20, 
                buy_threshold=-1.5, sell_threshold=1.5):
        super(MeanRevertingStrategy, self).__init__()
        self.symbol = symbol
        self.lookback_intervals = lookback_intervals
        self.buy_threshold = buy_threshold
        self.sell_threshold = sell_threshold
        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_quants > 0 else False
            self.is_short = True if position.net_quants <= 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_tick_data().timestamp
        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):
        tick_data = market_data.get_tick_data(self.symbol)
        timestamp = tick_data.timestamp
        self.prices.loc[timestamp, 'open'] = tick_data.open
        self.prices.loc[timestamp, 'high'] = tick_data.high
        self.prices.loc[timestamp, 'low'] = tick_data.low
        self.prices.loc[timestamp, 'close'] = tick_data.close
        self.prices.loc[timestamp, 'volume'] = tick_data.volume
        self.prices.loc[timestamp, 'outstanding_share'] = tick_data.outstanding_share
        self.prices.loc[timestamp, 'turnover'] = tick_data.turnover

    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_order(self.symbol, 100, True, timestamp)

    def on_sell_signal(self, timestamp):
        if not self.is_short:
            self.send_order(self.symbol, 100, False, timestamp)

• 第2、3行：定义构造函数，lookback_intervals是向前看多少个时间点，buy_threshold为当收益率降低50%时，我们就买入该股票；sell_thrshold为当收益率涨50%时，我们就卖出该股票；
• 第11~15行：根据仓位类，当持有股票时，is_long为真，当不持有股票时，is_short为真；
• 第17~26行：保存价格信息，如果价格信息不够向前看的时间点数，则直接返回，什么都不做。否则计算信号量，当信号量跌破购买阈值buy_threshold时买入股票，当信号量涨破卖出阈值sell_threshold时，卖出股票；
• 第28~37行：保存价格信息；
• 第39~43行：取lookback_intervals长度的价格数据，用当天的收盘价减去昨日的收盘价，得到收益率序列returns，然后根据如下公式计算信号量：
  $$z=\frac{r-\mu }{\sigma }$$
  其中r为收益率，$\mu$为收益率均值，$\sigma$为收益率标准差；
• 第45~47行：当触发买入条件时，发送买入订单；
• 第49~51行：当触发卖出条件时，发送卖出订单；

在有了上述基础类之后，我们将在下一篇博文中完善我们的回测引擎类，执行一次完整的回测过程。我们目前的回测系统，实际上还不是很真实，因为在实际交易中，还需要交手续费、过户费、印花税等，我们将在后面博文中引入Broker类，来处理这些业务逻辑。由于我们的策略都是有风险的，因此我们需要在我们回测系统中，加入风控模块，这将在我们最后一篇博文中加以介绍。