在最新版的vnpy_xt的基础上,做以下3处修改,就可以支持券商版:
1、修改XtGateway类的connect()方法:
def connect(self, setting: dict) -> None:
"""连接交易接口"""
# if self.thread:
# return
# self.thread = Thread(target=self._connect, args=(setting,))
# self.thread.start()
self._connect(setting)
2、修改XtMdApi类的connect()方法:
def connect(
self,
token: str,
stock_active: bool,
futures_active: bool,
option_active: bool
) -> None:
"""连接"""
self.gateway.write_log("开始启动行情服务,请稍等")
self.token = token
self.stock_active = stock_active
self.futures_active = futures_active
self.option_active = option_active
if self.inited:
self.gateway.write_log("行情接口已经初始化,请勿重复操作")
return
try:
# 使用xt版
# self.init_xtdc()
# 尝试查询合约信息,确认连接成功
xtdata.get_instrument_detail("000001.SZ")
except Exception as ex:
self.gateway.write_log(f"迅投研数据服务初始化失败,发生异常:{ex}")
return
self.inited = True
self.gateway.write_log("行情接口连接成功")
self.query_contracts()
3、修改XtTdApi类的connect()方法:
def connect(self, path: str, accountid: str, account_type: str, session: int = 0) -> int:
"""发起连接"""
self.inited = True
self.account_id = accountid
self.path = path
self.account_type = account_type
# 创建客户端和账号实例
if not session:
session = int(float(datetime.now().strftime("%H%M%S.%f")) * 1000)
# self.xt_client = XtQuantTrader(self.path, session)
# self.xt_account = StockAccount(self.account_id, account_type=self.account_type)
# # 注册回调接口
# self.xt_client.register_callback(self)
# # 启动交易线程
# self.xt_client.start()
# # 建立交易连接,返回0表示连接成功
# connect_result: int = self.xt_client.connect()
# if connect_result:
# self.gateway.write_log("交易接口连接失败")
# return connect_result
self.xt_client = XtQuantTrader(path=self.path,
session=session,
callback=self)
self.xt_client.start()
res = self.xt_client.connect()
if res != 0:
self.gateway.write_log(f'连接失败,请确认miniqmt登录后再重试')
return False
# 订阅账户
self.xt_account = StockAccount(self.account_id)
res = self.xt_client.subscribe(account=self.xt_account)
if res != 0:
self.gateway.write_log(f'连接账户失败')
return -1
self.inited = True
self.gateway.write_log(f'连接账户成功')
self.connected = True
self.gateway.write_log("交易接口连接成功")
# 订阅交易回调推送
subscribe_result: int = self.xt_client.subscribe(self.xt_account)
if subscribe_result:
self.gateway.write_log("交易推送订阅失败")
return -1
self.gateway.write_log("交易推送订阅成功")
# 初始化数据查询
self.query_account()
self.query_position()
self.query_order()
self.query_trade()
return res