怎么写交易策略？

有人问“怎么写交易策略”，这对于新手而言，确实是一个非常普遍的问题，今天就来解答一下这个问题。

交易就是通过期货/证券公司与交易所交互信息的一个过程。回想一下手动交易的过程：

①行情实时变化

②下单

③订单状态实时变化

④订单成交

⑤持仓变化

在手动交易的信息流转过程中，每个节点都依赖人的参与：人看到行情的变化，经过决策下单，交易所返回订单状态实时状态，未成交前，人可以决策是否撤单重报，成交之后，人可以决策是否加仓/平仓，等。

量化交易就是将这个信息流中的全部人工节点，或者部分，置换成程序。

下单是一个决策，这个决策必定由某些信息的变化引起，如果没有任何变化，就不会下单。根据变化信息的不同，我们将策略分为三种：行情驱动型、订单状态驱动型、外部驱动型。

不同驱动型的量化策略实现起来略有不同。

首先，行情驱动型策略，根据实时行情变化判断是否需要交易。

实现这种策略，就要将逻辑写在行情的通知中。以CTP为例，实时行情由MdApi的回调函数OnRtnDepthMarketData通知，所以，我们需要将报单的逻辑写在该回调函数中。

例如，一个双均线交易的逻辑：首先需要将行情快照处理成K线数据，然后分别计算短周期均线和长周期均线的数值，当短周期均线上穿长周期均线时报买单，反之，报卖单。

def OnRtnDepthMarketData(self, snapshot):
    is_new_10s = (snapshot['UpdateTime'][:-1] != self.bar["UpdateTime"][:-1])

    # 新K线开始
    if is_new_10s:
        self.bar_count += 1
        if self.bar_count > 1:
            self.bar_close_list.append(self.bar["LastPrice"])
            print(snapshot['UpdateTime'], snapshot['InstrumentID'], snapshot['LastPrice'], self.bar_close_list[-self.long_n:])

            if self.bar_count > self.long_n:
                short_ma = sum(self.bar_close_list[-self.short_n:]) / self.short_n
                long_ma = sum(self.bar_close_list[-self.long_n:]) / self.long_n
                
                if short_ma < long_ma and self.short_ma >= self.long_ma:
                    g_trader.sell_close("SHFE", "rb2310", snapshot['BidPrice1'], 1, True)
                elif short_ma > long_ma and self.short_ma <= self.long_ma:
                    g_trader.buy_open("SHFE", "rb2310", snapshot['AskPrice1'], 1, True)

                print(short_ma, long_ma, self.short_ma, self.long_ma)

                self.long_ma = long_ma
                self.short_ma = short_ma

    # 将Tick池化为Bar
    tick_to_bar(self.bar, snapshot, is_new_10s)

完整代码可以从
https://gitee.com/AlgoPlus/AlgoPlus项目examples中long_short_ma_global.py找到。

相应的，订单状态驱动型策略也不难理解，就是当订单状态发生变化时判断是否需要交易。

以CTP为例，订单状态和成交信息分别是由TraderApi的回调函数OnRtnOrder和OnRtnTrade通知。所以，报单的逻辑需要写在这两个回调函数中。

例如，当我们需要把大单子拆成小单子来执行时，就需要前一个订单全部成交之后，再报入下一个订单。也就是，当OnRtnOrder通知订单状态为全部成交时，发起新的报单：

def OnRtnOrder(self, pOrder):
    if pOrder['OrderStatus'] == b"0":
        if self.order_ref < 100:
            self.buy_open(self.parameter_dict['ExchangeID'], self.parameter_dict['InstrumentID'], self.parameter_dict['UpperLimitPrice'], self.parameter_dict['Volume'])
        else:
            self.write_log("滚动交易（100笔）已经全部完成！")

这是
https://gitee.com/AlgoPlus/AlgoPlus项目examples中rolling_trade.py的例子。

最后，行情和订单状态之外的，都可以认为是外部驱动型策略，例如twap算法根据时间间隔报单。这类策略只需要定义一个轮询函数，在创建TraderApi实例之后调用该函数就可以了。

例如，在看穿式认证的例子authenticate.py中，按照顺序做5次买、卖、开、平交易，就是定义了Join轮询函数，创建TraderApi实例之后调用了Join函数，执行完成之后退出。

def Join(self):
    while True:
        if self.status >= 0 and isinstance(self.parameter_dict, dict):
            # ############################################################################# #
            # 连续5次买开 - 卖平
            ikk = 0
            while ikk < 5:
                ikk += 1
                self.buy_open(self.parameter_dict['ExchangeID'], self.parameter_dict['InstrumentID'], self.parameter_dict['UpperLimitPrice'], self.parameter_dict['Volume'])
                self.write_log(f"=>{ikk}=>发出涨停买开仓请求！")

                time.sleep(3)

                # 跌停卖平仓
                self.sell_close(self.parameter_dict['ExchangeID'], self.parameter_dict['InstrumentID'], self.parameter_dict['LowerLimitPrice'], self.parameter_dict['Volume'], True)
                self.write_log(f"=>{ikk}=>发出跌停卖平仓请求！")

            #省略其他操作代码
            time.sleep(3)

            # ############################################################################# #
            print("看穿式监管认证仿真交易已经完成！可联系期货公司！")
            break

        time.sleep(1)