feat(bridge): bridge稳定性完善—自动重连miniQMT+health探活+交易日判断
- xt_gateway: _open_session抽离, reconnect(stop旧trader+重建), is_alive(query探活), _retry_with_reconnect(query/order失败重连重试一次) - query_account/positions/place_order包重试: 断线(异常/None)→reconnect→重试, broker拒单(order_id<=0)不重连 - bridge /health: is_alive真实探活(5s缓存)+断线后台reconnect(不阻塞), 不再假阳性 - trade_calendar: is_trading_day(周一-周五), /order非交易日加warning(120141提示) - test_gateway15+test_trade_calendar6=NAS21passed, 回归bridge_client/d4a 10绿 - 修复Issue#4运维发现: miniQMT重启后bridge自动重连(无需手动重启)
This commit is contained in:
+157
-74
@@ -58,6 +58,7 @@ class XtGateway:
|
||||
self._xt: Any = None
|
||||
self._account: Any = None
|
||||
self._connected: bool = False
|
||||
self._last_query_ok: bool = False # 上次 query 是否成功（供 /health 探活参考）
|
||||
|
||||
@property
|
||||
def connected(self) -> bool:
|
||||
@@ -67,68 +68,143 @@ class XtGateway:
|
||||
def connect(self) -> bool:
|
||||
"""连接 miniQMT 客户端。失败记日志不崩溃，返回 False。"""
|
||||
try:
|
||||
from xtquant import xttrader
|
||||
from xtquant.xttype import StockAccount
|
||||
return self._open_session()
|
||||
except ImportError as e:
|
||||
logger.error("xtquant import 失败（检查 site-packages）: %s", e)
|
||||
self._connected = False
|
||||
return False
|
||||
|
||||
try:
|
||||
xt = xttrader.XtQuantTrader(MINIQMT_USERDATA, SESSION_ID)
|
||||
xt.start()
|
||||
ret = xt.connect()
|
||||
if ret != 0:
|
||||
logger.error(
|
||||
"xtquant connect 返回 %s（miniQMT 未登录或路径错误）", ret,
|
||||
)
|
||||
self._connected = False
|
||||
return False
|
||||
|
||||
account = StockAccount(ACCOUNT_ID)
|
||||
try:
|
||||
xt.subscribe(account)
|
||||
except Exception as e:
|
||||
logger.warning("subscribe 异常（可忽略，继续）: %s", e)
|
||||
|
||||
self._xt = xt
|
||||
self._account = account
|
||||
self._connected = True
|
||||
logger.info("xtquant 连接成功 account=%s", ACCOUNT_ID)
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error("xtquant 连接异常: %s", e)
|
||||
self._connected = False
|
||||
return False
|
||||
|
||||
def _open_session(self) -> bool:
|
||||
"""建 XtQuantTrader + start + connect + subscribe（延迟 import xtquant）。
|
||||
|
||||
从 connect() 抽出，供 reconnect() 复用。raise 异常由调用方捕获。
|
||||
"""
|
||||
from xtquant import xttrader
|
||||
from xtquant.xttype import StockAccount
|
||||
|
||||
xt = xttrader.XtQuantTrader(MINIQMT_USERDATA, SESSION_ID)
|
||||
xt.start()
|
||||
ret = xt.connect()
|
||||
if ret != 0:
|
||||
logger.error(
|
||||
"xtquant connect 返回 %s（miniQMT 未登录或路径错误）", ret,
|
||||
)
|
||||
self._connected = False
|
||||
return False
|
||||
|
||||
account = StockAccount(ACCOUNT_ID)
|
||||
try:
|
||||
xt.subscribe(account)
|
||||
except Exception as e:
|
||||
logger.warning("subscribe 异常（可忽略，继续）: %s", e)
|
||||
|
||||
self._xt = xt
|
||||
self._account = account
|
||||
self._connected = True
|
||||
logger.info("xtquant 连接成功 account=%s", ACCOUNT_ID)
|
||||
return True
|
||||
|
||||
def reconnect(self) -> bool:
|
||||
"""重连：stop 旧 trader（若有），再 _open_session 重建连接。
|
||||
|
||||
miniQMT 客户端重启后旧 XtQuantTrader 连接失效，必须重建。
|
||||
"""
|
||||
if self._xt is not None and hasattr(self._xt, "stop"):
|
||||
try:
|
||||
self._xt.stop()
|
||||
except Exception as e:
|
||||
logger.warning("旧 trader stop 异常（忽略，继续重建）: %s", e)
|
||||
self._xt = None
|
||||
self._account = None
|
||||
self._connected = False
|
||||
logger.info("开始重连 miniQMT ...")
|
||||
return self.connect()
|
||||
|
||||
def is_alive(self) -> bool:
|
||||
"""轻量探活：query_stock_asset 返回非空 = True。
|
||||
|
||||
供 /health 调用检测真实连接状态（_connected 标志可能假阳性）。
|
||||
不触发 reconnect（保持轻量），由调用方决定是否重连。
|
||||
"""
|
||||
try:
|
||||
if not self._connected:
|
||||
return False
|
||||
asset = self._xt.query_stock_asset(self._account)
|
||||
alive = asset is not None
|
||||
self._last_query_ok = alive
|
||||
return alive
|
||||
except Exception as e:
|
||||
logger.debug("is_alive 探活异常: %s", e)
|
||||
self._last_query_ok = False
|
||||
return False
|
||||
|
||||
def _retry_with_reconnect(self, fn: Any, fail_msg: str) -> Any:
|
||||
"""执行 fn()，失败（异常）时 reconnect 一次再重试。
|
||||
|
||||
重连仍失败则 raise（由调用方按原逻辑降级返回空/False）。
|
||||
|
||||
Args:
|
||||
fn: 无参可调用，执行实际 query/order。
|
||||
fail_msg: 日志标识（如 "query_account"）。
|
||||
"""
|
||||
try:
|
||||
result = fn()
|
||||
self._last_query_ok = True
|
||||
return result
|
||||
except Exception as e:
|
||||
self._last_query_ok = False
|
||||
logger.warning("%s 首次失败，尝试重连: %s", fail_msg, e)
|
||||
if not self.reconnect():
|
||||
raise RuntimeError(f"重连失败，放弃 {fail_msg}: {e}") from e
|
||||
result = fn() # 重试一次（不再重连）
|
||||
self._last_query_ok = True
|
||||
return result
|
||||
|
||||
def query_account(self) -> dict[str, float]:
|
||||
"""查资金：{cash, frozen, market_value, total}。"""
|
||||
if not self._connected:
|
||||
raise RuntimeError("xtquant 未连接")
|
||||
asset = self._xt.query_stock_asset(self._account)
|
||||
if asset is None:
|
||||
raise RuntimeError("query_stock_asset 返回空（账户ID/权限问题）")
|
||||
return {
|
||||
"cash": float(asset.cash),
|
||||
"frozen": float(asset.frozen_cash),
|
||||
"market_value": float(asset.market_value),
|
||||
"total": float(asset.total_asset),
|
||||
}
|
||||
"""查资金：{cash, frozen, market_value, total}（断线自动重连重试一次）。"""
|
||||
|
||||
def _do() -> dict[str, float]:
|
||||
if not self._connected:
|
||||
raise RuntimeError("xtquant 未连接")
|
||||
asset = self._xt.query_stock_asset(self._account)
|
||||
if asset is None:
|
||||
raise RuntimeError("query_stock_asset 返回空（账户ID/权限问题或断线）")
|
||||
return {
|
||||
"cash": float(asset.cash),
|
||||
"frozen": float(asset.frozen_cash),
|
||||
"market_value": float(asset.market_value),
|
||||
"total": float(asset.total_asset),
|
||||
}
|
||||
|
||||
return self._retry_with_reconnect(_do, "query_account")
|
||||
|
||||
def query_positions(self) -> list[dict[str, Any]]:
|
||||
"""查持仓：[{code, volume, can_use, avg_price}]，code 已转 sanguo 格式。"""
|
||||
if not self._connected:
|
||||
raise RuntimeError("xtquant 未连接")
|
||||
positions = self._xt.query_stock_positions(self._account) or []
|
||||
return [
|
||||
{
|
||||
"code": to_sanguo_code(p.stock_code),
|
||||
"volume": int(p.volume),
|
||||
"can_use": int(p.can_use_volume),
|
||||
"avg_price": float(p.avg_price),
|
||||
}
|
||||
for p in positions
|
||||
]
|
||||
"""查持仓：[{code, volume, can_use, avg_price}]，code 已转 sanguo 格式。
|
||||
|
||||
None（断线）触发重连重试，空列表 []（真没持仓）是正常结果。
|
||||
"""
|
||||
|
||||
def _do() -> list[dict[str, Any]]:
|
||||
if not self._connected:
|
||||
raise RuntimeError("xtquant 未连接")
|
||||
positions = self._xt.query_stock_positions(self._account)
|
||||
if positions is None:
|
||||
raise RuntimeError("query_stock_positions 返回 None（疑似断线）")
|
||||
return [
|
||||
{
|
||||
"code": to_sanguo_code(p.stock_code),
|
||||
"volume": int(p.volume),
|
||||
"can_use": int(p.can_use_volume),
|
||||
"avg_price": float(p.avg_price),
|
||||
}
|
||||
for p in positions
|
||||
]
|
||||
|
||||
return self._retry_with_reconnect(_do, "query_positions")
|
||||
|
||||
def place_order(
|
||||
self,
|
||||
@@ -140,6 +216,9 @@ class XtGateway:
|
||||
) -> int:
|
||||
"""下单，返回 order_id（>0 = 报单成功）。
|
||||
|
||||
断线（调用抛异常）自动重连重试一次；order_id<=0 是 broker 拒单
|
||||
（如非交易日/资金不足），不重试直接返回。
|
||||
|
||||
Args:
|
||||
code: sanguo 格式(sh600000/sz000001)，内部转 xtquant 格式。
|
||||
action: "buy" / "sell"。
|
||||
@@ -149,33 +228,37 @@ class XtGateway:
|
||||
|
||||
Raises:
|
||||
ValueError: action/price_type/code 不合法。
|
||||
RuntimeError: xtquant 未连接。
|
||||
RuntimeError: xtquant 未连接 / 重连失败。
|
||||
"""
|
||||
if not self._connected:
|
||||
raise RuntimeError("xtquant 未连接")
|
||||
from xtquant import xtconstant
|
||||
|
||||
order_type = (
|
||||
xtconstant.STOCK_BUY if action == "buy" else xtconstant.STOCK_SELL
|
||||
)
|
||||
xt_price_type = (
|
||||
xtconstant.FIX_PRICE
|
||||
if price_type == "limit"
|
||||
else xtconstant.LATEST_PRICE
|
||||
)
|
||||
xt_code = to_xtquant_code(code)
|
||||
def _do() -> int:
|
||||
if not self._connected:
|
||||
raise RuntimeError("xtquant 未连接")
|
||||
from xtquant import xtconstant
|
||||
|
||||
order_id = self._xt.order_stock(
|
||||
self._account,
|
||||
xt_code,
|
||||
order_type,
|
||||
volume,
|
||||
xt_price_type,
|
||||
price,
|
||||
"sanguo_bridge",
|
||||
"",
|
||||
)
|
||||
return int(order_id)
|
||||
order_type = (
|
||||
xtconstant.STOCK_BUY if action == "buy" else xtconstant.STOCK_SELL
|
||||
)
|
||||
xt_price_type = (
|
||||
xtconstant.FIX_PRICE
|
||||
if price_type == "limit"
|
||||
else xtconstant.LATEST_PRICE
|
||||
)
|
||||
xt_code = to_xtquant_code(code)
|
||||
|
||||
order_id = self._xt.order_stock(
|
||||
self._account,
|
||||
xt_code,
|
||||
order_type,
|
||||
volume,
|
||||
xt_price_type,
|
||||
price,
|
||||
"sanguo_bridge",
|
||||
"",
|
||||
)
|
||||
return int(order_id)
|
||||
|
||||
return self._retry_with_reconnect(_do, "place_order")
|
||||
|
||||
|
||||
# 模块级单例（bridge.py 启动时调 connect）
|
||||
|
||||
Reference in New Issue
Block a user