价差交易策略模板（SpreadStrategyTemplate）

价差交易策略模板提供完整的信号生成和委托管理功能，用户可以基于该模板(位于vnpy.app.spread_trading.template中)自行开发策略。

用户自行开发的策略可以放在用户运行文件夹下的strategies文件夹内。

请注意：

1. 策略文件命名采用下划线模式，如statistical_arbitrage_strategy.py，而策略类命名采用驼峰式，如StatisticalArbitrageStrategy。

2. 自建策略的类名不要与示例策略的类名重合。如果重合了，图形界面上只会显示一个策略类名。

目前，vnpy官方提供两个价差策略，即BasicSpreadStrategy和StatisticalArbitrageStrategy。下面通过StatisticalArbitrageStrategy示例，来展示策略开发的具体步骤：

在基于价差交易策略模板编写策略逻辑之前，需要在策略文件的顶部载入需要用到的内部组件，如下方代码所示：

from vnpy.trader.utility import BarGenerator, ArrayManager
from vnpy_spreadtrading import (
    SpreadStrategyTemplate,
    SpreadAlgoTemplate,
    SpreadData,
    OrderData,
    TradeData,
    TickData,
    BarData
)

其中，SpreadStrategyTemplate和SpreadAlgoTemplate是价差交易策略模板和价差算法模板，SpreadData、OrderData、TickData、TradeData和BarData是储存对应信息的数据容器，BarGenerator是K线生成模块，ArrayManager是K线时间序列管理模块。

策略参数与变量

在策略类的下方，可以设置策略的作者（author），参数（parameters）以及变量（variables），如下方代码所示：

    author = "用Python的交易员"
    boll_window = 20
    boll_dev = 2
    max_pos = 10
    payup = 10
    interval = 5
    spread_pos = 0.0
    boll_up = 0.0
    boll_down = 0.0
    boll_mid = 0.0
    parameters = [
        "boll_window",
        "boll_dev",
        "max_pos",
        "payup",
        "interval"
    ]
    variables = [
        "spread_pos",
        "boll_up",
        "boll_down",
        "boll_mid"
    ]

虽然策略的参数和变量都从属于策略类，但策略参数是固定的（由交易员从外部指定），而策略变量则在交易的过程中随着策略的状态变化，所以策略变量一开始只需要初始化为对应的基础类型。例如：整数设为0，浮点数设为0.0，而字符串则设为””。

如果需要价差交易模块引擎在运行过程中，将策略参数和变量显示在UI界面上，并在数据刷新、停止策略时保存其数值，则需把参数和变量的名字（以字符串的数据类型）添加进parameters和variables列表里。

请注意，该列表只能接受参数和变量以str、int、float和bool四种数据类型传入。如果策略里需要用到其他数据类型的参数与变量，请把该参数或变量的定义放到__init__函数下。

类的初始化

入参：strategy_engine, strategy_name: str, spread: SpreadData, setting: dict

出参：无

__init__函数是策略类的构造函数，需要与继承的SpreadStrategyTemplate保持一致。

在这个继承的策略类里，初始化一般分三步，如下方代码所示：

    def __init__(self, cta_engine, strategy_name, vt_symbol, setting):
        """"""
        super().__init__(
            strategy_engine, strategy_name, spread, setting
        )
        self.bg = BarGenerator(self.on_spread_bar)
        self.am = ArrayManager()

1 . 通过super( )的方法继承SpreadStrategyTemplate，在__init__( )函数中传入策略引擎、策略名称、价差以及参数设置（以上参数均由策略引擎在使用策略类创建策略实例时自动传入，用户无需进行设置）；

2 . 调用K线生成模块（BarGenerator）：通过时间切片将Tick数据合成1分钟K线数据。如有需求，还可合成更长的时间周期数据。

3 . 调用K线时间序列管理模块（ArrayManager）：基于K线数据将其转化为便于向量化计算的时间序列数据结构，并在内部支持使用talib库来计算相应的技术指标。

价差策略引擎调用的函数

SpreadStrategyTemplate中的update_setting函数、该函数后面四个以get开头的函数和后面两个以update开头的函数，都是价差策略引擎去负责调用的函数，一般在策略编写的时候是不需要调用的。

策略的回调函数

SpreadStrategyTemplate中以on开头的函数称为回调函数，在编写策略的过程中能够用来接收价差行情或者接收状态更新。回调函数的作用是当某一个事件发生的时候，策略里的这类函数会被价差交易策略引擎自动调用（无需在策略中主动操作）。回调函数按其功能可分为以下三类：

策略实例状态控制（所有策略都需要）

on_init

• 入参：无

• 出参：无

初始化策略时on_init函数会被调用，默认写法是调用write_log函数输出“策略初始化”日志，再调用load_bar函数加载历史数据，如下方代码所示：

    def on_init(self):
        """
        Callback when strategy is inited.
        """
        self.write_log("策略初始化")
        self.load_bar(10)

请注意，如果是基于Tick数据回测，请在此处调用load_tick函数。

策略初始化时，策略的inited和trading状态都为【False】，此时只是调用ArrayManager计算并缓存相关的计算指标，不能发出交易信号。调用完on_init函数之后，策略的inited状态才变为【True】，策略初始化才完成。

on_start

• 入参：无

• 出参：无

启动策略时on_start函数会被调用，默认写法是调用write_log函数输出“策略启动”日志，如下方代码所示：

    def on_start(self):
        """
        Callback when strategy is started.
        """
        self.write_log("策略启动")

调用策略的on_start函数启动策略后，策略的trading状态变为【True】，此时策略才能够发出交易信号。

on_stop

• 入参：无

• 出参：无

停止策略时on_stop函数会被调用，默认写法是调用write_log函数输出“策略停止”日志，同时还原策略的变量，如下方代码所示：

    def on_stop(self):
        """
        Callback when strategy is stopped.
        """
        self.write_log("策略停止")
        self.put_event()

调用策略的on_stop函数停止策略后，策略的trading状态变为【False】，此时策略就不会发出交易信号了。

接收数据、计算指标、发出交易信号

on_spread_data

• 入参：无

• 出参：无

当价差数据更新的时候on_spread_data函数会被调用（因本次示例策略类StatisticalArbitrageStrategy不是基于on_spread_data交易，故不作示例讲解。基于on_spread_data交易的示例代码可参考示例策略BasicSpreadStrategy）。StatisticalArbitrageStrategy的写法是先调用get_spread_tick获取价差Tick数据，然后推进on_spread_tick函数中，如下方代码所示：

    def on_spread_data(self):
        """
        Callback when spread price is updated.
        """
        tick = self.get_spread_tick()
        self.on_spread_tick(tick)

on_spread_tick

• 入参：tick: TickData

• 出参：无

当策略收到最新的价差Tick数据的行情时，on_spread_tick函数会被调用。默认写法是通过BarGenerator的update_tick函数把收到的Tick数据推进前面创建的bg实例中以便合成1分钟的K线，如下方代码所示：

    def on_spread_tick(self, tick: TickData):
        """
        Callback when new spread tick data is generated.
        """
        self.bg.update_tick(tick)

on_spread_bar

• 入参：bar: BarData

• 出参：无

当策略收到最新的价差K线数据时（实盘时默认推进来的是基于Tick合成的一分钟的K线，回测时则取决于选择参数时填入的K线数据频率），on_spread_bar函数就会被调用。

如果策略基于on_spread_bar推进来的K线交易，那么请把交易请求类函数都写在on_spread_bar函数下。示例策略类StatisticalArbitrageStrategy是通过1分钟K线数据回报来生成CTA信号的。一共有三部分，如下方代码所示：

    def on_spread_bar(self, bar: BarData):
        """
        Callback when spread bar data is generated.
        """
        self.stop_all_algos()
        self.am.update_bar(bar)
        if not self.am.inited:
            return
        self.boll_mid = self.am.sma(self.boll_window)
        self.boll_up, self.boll_down = self.am.boll(
            self.boll_window, self.boll_dev)
        if not self.spread_pos:
            if bar.close_price >= self.boll_up:
                self.start_short_algo(
                    bar.close_price - 10,
                    self.max_pos,
                    payup=self.payup,
                    interval=self.interval
                )
            elif bar.close_price <= self.boll_down:
                self.start_long_algo(
                    bar.close_price + 10,
                    self.max_pos,
                    payup=self.payup,
                    interval=self.interval
                )
        elif self.spread_pos < 0:
            if bar.close_price <= self.boll_mid:
                self.start_long_algo(
                    bar.close_price + 10,
                    abs(self.spread_pos),
                    payup=self.payup,
                    interval=self.interval
                )
        else:
            if bar.close_price >= self.boll_mid:
                self.start_short_algo(
                    bar.close_price - 10,
                    abs(self.spread_pos),
                    payup=self.payup,
                    interval=self.interval
                )
        self.put_event()

• 清空未成交委托：为了防止之前下的单子在上1分钟没有成交，但是下1分钟可能已经调整了价格，就用stop_all_algos()方法立刻撤销之前未成交的所有委托，保证策略在当前这1分钟开始时的整个状态是清晰和唯一的；

• 调用K线时间序列管理模块：基于最新的1分钟价差K线数据来计算相应的技术指标，如布林带通道上下轨等。首先获取ArrayManager对象，然后将收到的K线推送进去，检查ArrayManager的初始化状态，如果还没初始化成功就直接返回，没有必要去进行后续的交易相关的逻辑判断。因为很多技术指标计算对最少K线数量有要求，如果数量不够的话计算出来的指标会出现错误或无意义。反之，如果没有return，就可以开始计算技术指标了；

• 信号计算：通过持仓的判断以及结合布林带通道在通道突破点挂出委托，同时设置离场点。

  请注意，如果需要在图形界面刷新指标数值，请不要忘记调用put_event()函数。

委托状态更新

以下函数在策略中几乎都可以直接pass，其具体逻辑应用交给回测/实盘引擎负责。

on_spread_pos

• 入参：无

• 出参：无

收到持有仓位更新时on_spread_pos函数会被调用。与CTA策略模块访问策略逻辑持仓不同，价差交易模块访问的是账户底层持仓。所以默认写法是通过调用get_spread_pos函数获取价差持仓，以供策略进行逻辑判断，如下方代码所示：

    def on_spread_pos(self):
        """
        Callback when spread position is updated.
        """
        self.spread_pos = self.get_spread_pos()
        self.put_event()

on_spread_algo

• 入参：algo: SpreadAlgoTemplate

• 出参：无

收到算法状态更新时on_spread_algo函数会被调用。

on_order

• 入参：order: OrderData

• 出参：无

收到策略委托回报时on_order函数会被调用。

on_trade

• 入参：trade: TradeData

• 出参：无

收到策略成交回报时on_trade函数会被调用。

主动函数

start_long_algo

• 入参：direction: Direction, price: float, volume: float, payup: int, interval: int, lock: bool = False, extra: dict = None

• 出参：algoid: str

start_short_algo

• 入参：direction: Direction, price: float, volume: float, payup: int, interval: int, lock: bool = False, extra: dict = None

• 出参：algoid: str

与CTA策略模块不同，价差交易的示例策略都是通过调用star_long_algo/start_short_algo函数（针对价差）而不是buy/sell/short/cover函数（针对特定合约）来发出委托的。在价差交易模块中，算法负责价差交易的执行，策略负责价差算法的调度。价差算法将价差交易简化为了普通委托，封装掉了所有主动腿下单和被动腿对冲的细节。

以下方star_long_algo函数的代码为例，可以看到，价格、数量、超价的数值、时间间隔是必填的参数，锁仓转换和开平方向则分别默认为False和Offset.NONE。也可以看到，函数内部收到传进来的参数之后就调用了SpreadStrategyTemplate里的start_algo函数来发单（因为是long指令，则自动把方向填成了LONG）

     def start_long_algo(
        self,
        price: float,
        volume: float,
        payup: int,
        interval: int,
        lock: bool = False,
        extra: dict = None
    ) -> str:
        """"""
        if not extra:
            extra = None
        return self.start_algo(
            Direction.SHORT, price, volume,
            payup, interval, lock, extra
        )

start_algo

• 入参：direction: Direction, price: float, volume: float, payup: int, interval: int, lock: bool, extra: dict

• 出参：algoid: str

start_algo函数是价差策略引擎调用的启动新的价差交易算法的函数。一般在策略编写的时候不需要单独调用，通过start_long_algo/start_short_algo函数发送委托即可。

请注意，要在策略启动之后，也就是策略的trading状态变为【True】之后，才能发出交易委托。如果策略的Trading状态为【False】时调用了该函数，只会返回[]。

stop_algo

• 入参：algoid: str

• 出参：无

stop_all_algos

• 入参：无

• 出参：无

stop_algo和stop_all_algos都是负责停止价差算法的交易请求类函数。stop_algo是停止策略内指定的价差算法，stop_all_algos是停止策略所有的活动价差算法。

请注意，要在策略启动之后，也就是策略的trading状态变为【True】之后，才能撤单。

buy：买入开仓（Direction：LONG，Offset：OPEN）

sell：卖出平仓（Direction：SHORT，Offset：CLOSE）

short：卖出开仓（Direction：SHORT，Offset：OPEN）

cover：买入平仓（Direction：LONG，Offset：CLOSE）

• 入参：vt_symbol: str, price: float, volume: float, lock: bool = False

• 出参：vt_orderids: List[vt_orderid] / 无

buy/sell/short/cover都是策略内部的负责针对特定合约发出底层交易委托的请求类函数。策略可以通过这些函数给价差策略引擎发送交易信号来达到下单的目的。

以下方buy函数的代码为例，可以看到，本地代码、价格和数量是必填的参数，锁仓转换则默认为False。也可以看到，函数内部收到传进来的参数之后就调用了SpreadStrategyTemplate里的send_order函数来发单（因为是buy指令，则自动把方向填成了LONG，开平填成了OPEN）

如果lock设置为True，那么该笔订单则会进行锁仓委托转换（在有今仓的情况下，如果想平仓，则会先平掉所有的昨仓，然后剩下的部分都进行反向开仓来代替平今仓，以避免平今的手续费惩罚）。

    def buy(self, vt_symbol: str, price: float, volume: float, lock: bool = False) -> List[str]:
        """"""
        return self.send_order(vt_symbol, price, volume, Direction.LONG, Offset.OPEN, lock)

send_order

• 入参：vt_symbol: str, price: float, volume: float, direction: Direction, offset: Offset, lock: bool = False

• 出参：vt_orderids / 无

send_order函数是价差策略引擎调用的针对特定合约（而不是价差）发送委托的函数。一般在策略编写的时候不需要单独调用，通过buy/sell/short/cover函数发送委托即可。

请注意，要在策略启动之后，也就是策略的trading状态变为【True】之后，才能发出交易委托。如果策略的Trading状态为【False】时调用了该函数，只会返回[]。

cancel_order

• 入参：vt_orderid: str

• 出参：无

cancel_all

• 入参：无

• 出参：无

cancel_order和cancel_all都是负责撤单的交易请求类函数。cancel_order是撤掉策略内指定的活动委托，cancel_all是撤掉策略所有的活动委托。

请注意，要在策略启动之后，也就是策略的trading状态变为【True】之后，才能撤单。

功能函数

以下为策略以外的功能函数：

put_event

• 入参：无

• 出参：无

在策略中调用put_event函数，可以通知图形界面刷新策略状态相关显示。

请注意，要策略初始化完成，inited状态变为【True】之后，才能刷新界面。

write_log

• 入参：msg: str

• 出参：无

在策略中调用write_log函数，可以进行指定内容的日志输出。

get_spread_tick

• 入参：无

• 出参：tick: TickData

在策略里调用get_spread_tick函数，可以获取价差Tick数据。

get_spread_pos

• 入参：无

• 出参：spread_pos: float

在策略里调用get_spread_pos函数，可以获取价差净持仓数据。

get_leg_tick

• 入参：vt_symbol: str

• 出参：leg.tick: TickData / None

在策略里调用get_leg_tick函数，可以获取特定合约的Tick数据。

get_leg_pos

• 入参：vt_symbol: str, direction: Direction = Direction.NET

• 出参：leg.net_pos: float / leg.long_pos: float /leg.short_pos: float / None

在策略里调用get_leg_pos函数，可以获取特定合约的持仓数据，用于处理瘸腿后的细粒度调整。

send_email

• 入参：msg: str

• 出参：无

配置好邮箱相关信息之后（配置方法详见基本使用篇的全局配置部分），在策略中调用send_email函数，可以发送指定内容的邮件到自己的邮箱。

请注意，要策略初始化完成，inited状态变为【True】之后，才能发送邮件。

load_bar

• 入参：days: int, interval: Interval = Interval.MINUTE, callback: Callable = None

• 出参：无

在策略中调用load_bar函数，可以在策略初始化时加载K线数据。

如下方代码所示，load_bar函数调用时，默认加载的天数是10，频率是一分钟，对应也就是加载10天的1分钟K线数据。在回测时，10天指的是10个交易日，而在实盘时，10天则是指的是自然日，因此建议加载的天数宁可多一些也不要太少。

    def load_bar(
        self,
        days: int,
        interval: Interval = Interval.MINUTE,
        callback: Callable = None,
    ):
        """
        Load historical bar data for initializing strategy.
        """
        if not callback:
            callback = self.on_spread_bar
        self.strategy_engine.load_bar(self.spread, days, interval, callback)

load_tick

• 入参：days: int

• 出参：无

在策略中调用load_tick函数，可以在策略初始化时加载Tick数据。