开始收录

  • 在“本地代码”选择输入需要订阅的行情，如rb1905.SHFE；

  • 然后点击后边“K线记录”或者“Tick记录”中的“添加”选项，会把记录特定品种任务添加到data_recorder_setting.json上，并且显示到“K线记录列表”或者“Tick记录列表”中，如图。

  • 通过queue.put()与queue.get()异步完成收录行情信息任务。

https://vnpy-community.oss-cn-shanghai.aliyuncs.com/forum_experience/yazhang/data_recoder/start.png

下面介绍行情收录的具体原理：若无合约记录的历史，用户需要先添加行情记录任务，如连接CTP接口后记录rb1905.SHFE的tick数据，然后调用add_tick_recording()函数执行下面工作：

  • 先创建tick_recordings字典；

  • 调用接口的suscribe()函数订阅行情；

  • 保存该tick_recordings字典到json文件上；

  • 推送行情记录事件。

  1.     def add_tick_recording(self, vt_symbol: str):
  2.         """"""
  3.         if vt_symbol in self.tick_recordings:
  4.             self.write_log(f"已在Tick记录列表中：{vt_symbol}")
  5.             return
  6.  
  7.         contract = self.main_engine.get_contract(vt_symbol)
  8.         if not contract:
  9.             self.write_log(f"找不到合约：{vt_symbol}")
  10.             return
  11.  
  12.         self.tick_recordings[vt_symbol] = {}
  13.             "symbol": contract.symbol,
  14.             "exchange": contract.exchange.value,
  15.             "gateway_name": contract.gateway_name
  16.         }
  17.  
  18.         self.subscribe(contract)
  19.         self.save_setting()
  20.         self.put_event()
  21.  
  22.         self.write_log(f"添加Tick记录成功：{vt_symbol}")

下面对add_tick_recording()函数里面调用的子函数进行扩展：

订阅行情

调用main_engine的suscribe()函数来订阅行情，需要填入的信息为symbol、exchange、gateway_name

  1.     def subscribe(self, contract: ContractData):
  2.         """"""
  3.         req = SubscribeRequest(
  4.             symbol=contract.symbol,
  5.             exchange=contract.exchange
  6.         )
  7.         self.main_engine.subscribe(req, contract.gateway_name)

将订阅信息保存到json文件

  • 主要把tick_recordings或者bar_recordings通过save_json()函数保存到C:\Users\Administrator.vntrader文件夹内的data_recorder_setting.json上。

  • 该json文件用于存放行情记录的任务，当每次启动行情模块后，会调用load_setting()函数来得到tick_recordings和bar_recordings字典，进而开始记录的任务。

  1. setting_filename = "data_recorder_setting.json"
  2.     def save_setting(self):
  3.         """"""
  4.         setting = {
  5.             "tick": self.tick_recordings,
  6.             "bar": self.bar_recordings
  7.         }
  8.         save_json(self.setting_filename, setting)
  9.  
  10.     def load_setting(self):
  11.         """"""
  12.         setting = load_json(self.setting_filename)
  13.         self.tick_recordings = setting.get("tick", {})
  14.         self.bar_recordings = setting.get("bar", {})

推送行情记录事件

  • 创建行情记录列表tick_symbols和bar_symbols，并且缓存在data字典里；

  • 创建evnte对象，其类型是EVENT_RECORDER_UPDATE, 内容是data字典；

  • 调用event_engine的put()函数推送event事件。

  1.     def put_event(self):
  2.         """"""
  3.         tick_symbols = list(self.tick_recordings.keys())
  4.         tick_symbols.sort()
  5.  
  6.         bar_symbols = list(self.bar_recordings.keys())
  7.         bar_symbols.sort()
  8.  
  9.         data = {
  10.             "tick": tick_symbols,
  11.             "bar": bar_symbols
  12.         }
  13.  
  14.         event = Event(
  15.             EVENT_RECORDER_UPDATE,
  16.             data
  17.         )
  18.         self.event_engine.put(event)

注册行情记录事件

register_event()函数分别注册2种事件：EVENT_CONTRACT、EVENT_TICK

  • EVENT_CONTRACT事件，调用的是process_contract_event()函数: 从tick_recordings和bar_recordings字典获取需要订阅的合约品种；然后使用subscribe()函数进行订阅行情。

  • EVENT_TICK事件，调用的是process_tick_event()函数：从tick_recordings和bar_recordings字典获取需要订阅的合约品种；然后使用record_tick()和record_bar()函数，把行情记录任务推送到queue队列中等待执行。

  1.     def register_event(self):
  2.         """"""
  3.         self.event_engine.register(EVENT_TICK, self.process_tick_event)
  4.         self.event_engine.register(EVENT_CONTRACT, self.process_contract_event)
  5.  
  6.     def process_tick_event(self, event: Event):
  7.         """"""
  8.         tick = event.data
  9.  
  10.         if tick.vt_symbol in self.tick_recordings:
  11.             self.record_tick(tick)
  12.  
  13.         if tick.vt_symbol in self.bar_recordings:
  14.             bg = self.get_bar_generator(tick.vt_symbol)
  15.             bg.update_tick(tick)
  16.  
  17.     def process_contract_event(self, event: Event):
  18.         """"""
  19.         contract = event.data
  20.         vt_symbol = contract.vt_symbol
  21.  
  22.         if (vt_symbol in self.tick_recordings or vt_symbol in self.bar_recordings):
  23.             self.subscribe(contract)
  24.  
  25.     def record_tick(self, tick: TickData):
  26.         """"""
  27.         task = ("tick", copy(tick))
  28.         self.queue.put(task)
  29.  
  30.     def record_bar(self, bar: BarData):
  31.         """"""
  32.         task = ("bar", copy(bar))
  33.         self.queue.put(task)
  34.  
  35.     def get_bar_generator(self, vt_symbol: str):
  36.         """"""
  37.         bg = self.bar_generators.get(vt_symbol, None)
  38.  
  39.         if not bg:
  40.             bg = BarGenerator(self.record_bar)
  41.             self.bar_generators[vt_symbol] = bg
  42.  
  43.         return bg

执行记录行情任务

在while循环中，从queue队列读取任务，调用save_tick_data()或者save_bar_data()函数来记录数据，并且载入到数据库中。

  1.     def run(self):
  2.         """"""
  3.         while self.active:
  4.             try:
  5.                 task = self.queue.get(timeout=1)
  6.                 task_type, data = task
  7.  
  8.                 if task_type == "tick":
  9.                     database_manager.save_tick_data([data])
  10.                 elif task_type == "bar":
  11.                     database_manager.save_bar_data([data])
  12.  
  13.             except Empty:
  14.                 continue