首先需要保证vt_symbol的唯一,我使用的vt_symbol格式是self.vtsymbol = f"{self.symbol}{self.exchange.value}/{self.gateway_name}"。vnpy\trader\object.py里面的所有类都要继承BaseData,然后就是添加gateway_name参数到其他使用到object.py里面的类。举个例子
@dataclass class HistoryRequest(BaseData): """ Request sending to specific gateway for querying history data. """ symbol: str exchange: Exchange start: datetime end: datetime = None interval: Interval = None #-------------------------------------------------------------------------------------------------- def __post_init__(self): """""" self.vt_symbol = f"{self.symbol}_{self.exchange.value}/{self.gateway_name}"
然后其他地方引用到HistoryRequest的地方添加gateway_name参数
#------------------------------------------------------------------------------------------------- def query_history(self,event): """查询合约历史数据""" if len(self.history_contract) > 0: symobl,exchange,gateway_name = extract_vt_symbol(self.history_contract[0]) req = HistoryRequest( symbol = symobl, exchange = Exchange(exchange), interval = Interval.MINUTE, start = datetime.now() - timedelta(days = 1), gateway_name = self.gateway_name ) self.rest_api.query_history(req) self.history_contract.pop(0)
需要分解vt_symbol建议使用extract_vt_symbol函数
#------------------------------------------------------------------------------------ def extract_vt_symbol(vt_symbol: str): """ 返回(symbol:str, exchange: Exchange,gateway_name:str) """ *symbol_1, exchange_gateway_name = vt_symbol.split('_') exchange,gateway_name = exchange_gateway_name.split('/') if len(symbol_1) == 1: symbol = symbol_1[0] elif len(symbol_1) == 2: symbol = "_".join([symbol_1[0],symbol_1[1]]) elif len(symbol_1) == 3: symbol = "_".join([symbol_1[0],symbol_1[1],symbol_1[2]]) elif len(symbol_1) == 4: symbol = "_".join([symbol_1[0],symbol_1[1],symbol_1[2],symbol_1[3]]) return symbol, Exchange(exchange),gateway_name