基于VNPY实现网格策略实盘（币圈）

vnpy事件驱动框架

首先我们要利用到vnpy的事件驱动框架，是一个消息队列。其中，交易所gateway就是事件的生产者，算法模块AlgotradingApp是队列的消费者。为了能够获取行情，我们要让网格策略算法监听委托、成交回报、订单回报、行情tick这几种事件。

tick、trade、order事件
在交易所gateway中，on_tick方法可以将tick数据推送到事件引擎中，其他事件以此类推：

    def on_tick(self, tick: TickData) -> None:"""Tick event push.Tick event of a specific vt_symbol is also pushed."""self.on_event(EVENT_TICK, tick)self.on_event(EVENT_TICK + tick.vt_symbol, tick)

在事件的消费者算法引擎中，需要注册此事件即可获得tick的数据，并给此事件准备好handler函数：

def register_event(self):""""""self.event_engine.register(EVENT_TICK, self.process_tick_event)self.event_engine.register(EVENT_TIMER, self.process_timer_event)self.event_engine.register(EVENT_ORDER, self.process_order_event)self.event_engine.register(EVENT_TRADE, self.process_trade_event)

tick事件的handler函数如下：

    def process_tick_event(self, event: Event):""""""tick = event.dataalgos = self.symbol_algo_map.get(tick.vt_symbol, None)if algos:for algo in algos:algo.update_tick(tick)

首先它会取出data，然后将此数据映射到相应的算法中。

交易所gateway

交易所的gateway就是程序与交易数据收发的入口，里面集成了rest、websocket、行情、交易的接口。本文使用的okex交易所的gateway。交易所的各种数据就是通过各种on开头的回调函数传递到事件引擎中，再被算法引擎监听获得。

    def __init__(self, event_engine: EventEngine, gateway_name: str = "OKEX") -> None:"""构造函数"""super().__init__(event_engine, gateway_name)self.rest_api: "OkexRestApi" = OkexRestApi(self)self.ws_public_api: "OkexWebsocketPublicApi" = OkexWebsocketPublicApi(self)self.ws_private_api: "OkexWebsocketPrivateApi" = OkexWebsocketPrivateApi(self)

vnpy算法引擎

算法引擎负责处理各种事件的数据并映射到对应的算法中，首先来看看其初始化函数：

class AlgoEngine(BaseEngine):""""""setting_filename = "algo_trading_setting.json"def __init__(self, main_engine: MainEngine, event_engine: EventEngine):"""Constructor"""super().__init__(main_engine, event_engine, APP_NAME)self.algos = {}self.symbol_algo_map = {}self.orderid_algo_map = {}self.algo_templates = {}self.algo_settings = {}self.load_algo_template()self.register_event()

其中有两个比较重要的字典，用来映射订单、行情到对应的algo类中，每个algo类就是不同的算法。

vnpy数据格式

vnpy的所有事件数据都有它自己的dataclass，比如交易数据TradeData：

@dataclass
class TradeData(BaseData):"""Trade data contains information of a fill of an order. One ordercan have several trade fills."""symbol: strexchange: Exchangeorderid: strtradeid: strdirection: Direction = Noneoffset: Offset = Offset.NONEprice: float = 0volume: float = 0datetime: datetime = Nonedef __post_init__(self):""""""self.vt_symbol = f"{self.symbol}.{self.exchange.value}"self.vt_orderid = f"{self.gateway_name}.{self.orderid}"self.vt_tradeid = f"{self.gateway_name}.{self.tradeid}"

可以看到我们可以通过trade数据得知这个交易明细的交易所，订单号，价格，数量，时间，这些都是算法类需要利用到的数据

algo类和算法模板template

为了实现自己编写的算法，我们需要写一个算法类，所有编写的算法都需要继承算法模板AlgoTemplate类，从而与算法引擎进行交互。

class GridAlgo(AlgoTemplate):""""""display_name = "Grid 网格交易"default_setting = {"vt_symbol": "","upper_limit": "","lower_limit": "","investment": "","grid_step": 0.0,"grid_levels":0,"interval": 10,"stop_loss": 0,"trailing_up":False,}

算法类再继承AlgoTemplate之后，就可以开始编写策略逻辑。在算法引擎中收到的Trade、Order事件数据都可以在算法类中利用on_trade，on_order进行处理。
更新算法的仓位数据时，利用on_trade方法：

    def on_trade(self, trade: TradeData):""""""if trade.direction == Direction.LONG:self.pos += trade.volumeelse:self.pos -= trade.volume

多单LONG仓位增加，空单SHORT仓位减小。至于策略逻辑可以写在on_timer函数中，定时根据行情检测买卖条件。

网格交易策略逻辑

创建好了algo类接下来就可以编写策略逻辑，为了实现网格交易首先我们需要下限价单，根据网格上界与下界，以及制定好的间隔下单：

    def send_all_limit_orders(self, tick):# 先购买足够多的币用来下网格卖单grid_price = []for price in np.arange(self.lower_limit, self.upper_limit,(self.upper_limit - self.lower_limit) / self.grid_levels):grid_price.append(price)# sell_order_count = 0sell_order_amount = 0for price in grid_price:if price > tick.last_price:sell_order_amount += self.grid_usdt_amounts# 为了能够刚好下完所有限价，多买入2%的币buy_amount = sell_order_amount*1self.buy(self.vt_symbol,price=tick.ask_price_1,volume=round(buy_amount, 5),order_type=OrderType.MARKET)time.sleep(3)print(f"{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'):}买入初始仓位完成")# 开始下限价单if len(self.active_orders) < self.grid_levels:for price in np.arange(self.lower_limit, self.upper_limit, (self.upper_limit-self.lower_limit)/self.grid_levels):if price < tick.last_price:volume = self.grid_usdt_amountsorderid = self.buy(volume=round(volume/price, 5),price=round(price,5),vt_symbol=self.vt_symbol)else:volume = self.grid_usdt_amountsorderid = self.sell(volume=round(volume/price, 5),price=round(price,5),vt_symbol=self.vt_symbol)self.order_book[orderid] = round(price,5)if tick.gateway_name == 'BINANCE':# 币安10秒内最多50个委托time.sleep(0.3)if tick.gateway_name == 'OKEX':# 币安10秒内最多50个委托time.sleep(0.1)return True

成功下单之后我们就需要利用on_timer函数定时检测是否有网格成交并且填充网格：

    def on_timer(self):""""""tick = self.get_tick(self.vt_symbol)if not tick:print(f"{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}:暂未收到行情")returnif not self.is_init:if self.is_sbot:self.send_all_limit_orders_sbot(tick)else:self.send_all_limit_orders(tick)self.is_init = True# 最多容忍2个订单差if not self.is_all_sent:if len(self.order_book) - len(self.active_orders) >= 2:print(f"本地订单簿长度{len(self.order_book)}，在线订单{len(self.active_orders)}")print(f"{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}:所有限价单订单回报未完全更新")returnelse:print(f"本地订单簿长度{len(self.order_book)}，在线订单{len(self.active_orders)}")self.is_all_sent = Trueif tick.last_price <= self.stop_loss:# 到达止损价，即使止损print(f"达到止损价{self.stop_loss}, 撤销所有限价单，卖出现有仓位")self.cancel_all()self.sell(volume=self.pos,price=tick.ask_price_5,vt_symbol=self.vt_symbol)if not self.is_sbot:if self.is_all_sent:self.check_grid_count(tick)else:if self.is_all_sent:self.check_grid_count_sbot(tick)

如果符合条件就填充网格

    def check_grid_count_sbot(self, tick):print(f"{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}:现在活跃订单簿数量：{len(self.active_orders)}，本地订单数量{len(self.order_book)}")# 每五分钟检测一次网格是否未填充if self.stuck:returnself.stuck = Truediff = self.order_book.keys() - self.active_orders.keys()# 找出离lasttick最近的网格的index，对其暂不填充price_diff = []for order in diff:price = self.order_book[order]price_diff.append(abs(price-tick.last_price))if len(price_diff) == 0:returnindex_min = price_diff.index(min(price_diff))index_count = 0if len(self.order_book.keys()) > len(self.active_orders):for order in diff:if index_count == index_min:print(f"{self.order_book[order]}此价格网格离lastprice过近，不进行委托")index_count += 1continueindex_count += 1price = self.order_book[order]del self.order_book[order]if tick.last_price > price:# print(f"{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'):}触碰了价格为{price}委托单")volume = self.grid_usdt_amountsorderid = self.buy(volume=round(volume/price, 5),price=price,vt_symbol=self.vt_symbol)else:# print(f"{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'):}触碰了价格为{price}委托单")volume = self.grid_usdt_amountsvolume = min(self.pos, round(volume/price, 5))orderid = self.sell(volume=volume,price=price,vt_symbol=self.vt_symbol)self.order_book[orderid] = price

程序入口

首先实例化事件引擎以及主引擎：

    event_engine = EventEngine()main_engine = MainEngine(event_engine)

首先我们需要往主引擎中添加gateway。

    main_engine.add_gateway(OkexGateway)main_engine.add_gateway(HuobiGateway)main_engine.add_gateway(BinanceGateway)

为了实时获取行情我们需要订阅symbol，这里我们选择XRP/USDT币对，oms引擎会储存一个gateway中所有的contract，于是我们遍历这个contract列表，找出xrp现货。

   contracts = main_engine.engines["oms"].contractsspotsymbol = []for contract_name in contracts:contract = \contracts[contract_name]vt_symbol = contract.vt_symbolif contract.product == Product.SPOT:if (arbsymbol in vt_symbol and 'usdt' in vt_symbol) or (arbsymbol.upper() in vt_symbol and 'USDT' in vt_symbol) :spotsymbol.append(vt_symbol)

手动设置好算法的参数：

    up = 0.85down = 0.71grid_levels = 180grid_step = (up-down)/down/grid_levelsgrid_amounts = 45grid_usdt_amount = 14investment = grid_levels*grid_usdt_amountis_sbot = Truedefault_setting = {"vt_symbol": vt_symbol,"upper_limit": up,"lower_limit": down,"investment": investment,"grid_step": grid_step,"grid_levels":grid_levels,"grid_amounts":grid_amounts,"grid_usdt_amounts":grid_usdt_amount,"is_sbot":is_sbot,"interval": 0,"stop_loss": 0.20,"trailing_up":False,}

最后添加算法引擎并初始化运行。

    algoengine = main_engine.add_app(AlgoTradingApp)algoengine.init_engine()algoengine.start_algo(default_setting)

策略实战

参数选择：
选择了btc现货，底部31000顶部40000，180个网格的参数，每个限价单价值11usdt。

    vt_symbol = 'BTC-USDT.OKEX'up = 40000down = 31000grid_levels = 180grid_step = (up-down)/down/grid_levelsgrid_amounts = 45grid_usdt_amount = 11investment = grid_levels*grid_usdt_amount

下所有限价单：

策略输出

2021-06-21 15:09:26:现在活跃订单簿数量：178，本地订单数量180
2021-06-21 15:09:27:现在活跃订单簿数量：178，本地订单数量180
多单成交，数量：0.00033,价格：33000.0
有不同的网格成交了，将stuck取消
2021-06-21 15:09:28:现在活跃订单簿数量：177，本地订单数量180
GridAlgo_1：委托卖出BTC-USDT.OKEX：0.00033@33050.0
33000.0此价格网格离lastprice过近，不进行委托
GridAlgo_1：委托卖出BTC-USDT.OKEX：0.00033@33100.0
2021-06-21 15:09:29:现在活跃订单簿数量：179，本地订单数量180
2021-06-21 15:09:30:现在活跃订单簿数量：179，本地订单数量180
......

将所有成交连点成线：

可以看出策略实现了逢低买入逢高卖出。