dekun/qihuo
1
0
Fork 0
Code Issues Actions Packages Projects Releases Wiki Activity

fix: CTP重连前探测前置可达性，失败时关闭网关并明确报错

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-06-25 16:30:35 +08:00
parent 240fbe7994
commit 72361233a0
4 changed files with 157 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
vnpy_bridge.py
+113 -28
View File
@@ -19,6 +19,9 @@ logger = logging.getLogger(__name__)
GATEWAY_NAME = "CTP"
CONNECT_WAIT_SEC = 60
CONNECT_POLL_INTERVAL_SEC = 0.5
_position_refresh_callback: Optional[Callable[[], None]] = None
@@ -79,8 +82,38 @@ def _mode_label(mode: str) -> str:
return "SimNow" if mode == "simulation" else "期货公司实盘"
def _format_ctp_failure(ctp_logs: list[str]) -> str:
def _parse_tcp_address(address: str) -> tuple[str, int]:
raw = (address or "").strip()
if raw.startswith("tcp://"):
raw = raw[6:]
if ":" not in raw:
raise ValueError(f"无效 TCP 地址: {address}")
host, port_s = raw.rsplit(":", 1)
return host, int(port_s)
def probe_tcp_address(address: str, timeout: float = 5.0) -> tuple[bool, str]:
"""探测 CTP 前置 TCP 是否可达。"""
import socket
try:
host, port = _parse_tcp_address(address)
with socket.create_connection((host, port), timeout=timeout):
return True, ""
except Exception as exc:
return False, str(exc)
def _format_ctp_failure(ctp_logs: list[str], *, td_address: str = "") -> str:
"""根据 CTP 网关日志拼出可读错误。"""
if td_address:
ok, err = probe_tcp_address(td_address, timeout=4.0)
if not ok:
return (
f"SimNow 交易前置不可达：{td_address}（{err}）。"
"182.254.243.31 已停用，请改 .env 为官方前置 "
"tcp://180.168.146.187:10201 / 10211，并确认服务器能访问该地址。"
)
text = "\n".join(ctp_logs)
if "4097" in text or "Decrypt handshake" in text or "shake hand" in text.lower():
return (
@@ -160,6 +193,35 @@ class CtpBridge:
def connect_in_progress(self) -> bool:
return self._connect_in_progress
def _close_gateway(self) -> None:
"""关闭 CTP 网关，避免半连接状态下重连卡在「连接登录」。"""
if not self._engine:
return
try:
gw = self._engine.get_gateway(GATEWAY_NAME)
if gw:
gw.close()
except Exception as exc:
logger.debug("gateway close: %s", exc)
self._connected_mode = None
time.sleep(0.6)
def _wait_connected(self, mode: str) -> bool:
"""等待账户回报或交易通道登录成功。"""
if not self._engine:
return False
loops = max(1, int(CONNECT_WAIT_SEC / CONNECT_POLL_INTERVAL_SEC))
for _ in range(loops):
try:
if self._engine.get_all_accounts():
return True
except Exception:
pass
if self._td_logged_in():
return True
time.sleep(CONNECT_POLL_INTERVAL_SEC)
return False
def status(self, mode: str) -> dict[str, Any]:
if self._connected_mode == mode:
self.ping()
@@ -199,14 +261,7 @@ class CtpBridge:
try:
with self._connect_lock:
if force and self._connected_mode:
try:
gw = self._engine.get_gateway(GATEWAY_NAME)
if gw:
gw.close()
except Exception:
pass
self._connected_mode = None
time.sleep(0.8)
self._close_gateway()
elif self._connected_mode and self._connected_mode != mode:
try:
self._engine.close()
@@ -214,6 +269,8 @@ class CtpBridge:
pass
self._connected_mode = None
time.sleep(1)
elif not (self._connected_mode == mode and self.ping()):
self._close_gateway()
ctp_logs: list[str] = []
from vnpy.trader.event import EVENT_LOG
@@ -222,7 +279,7 @@ class CtpBridge:
msg = getattr(event.data, "msg", "") or str(event.data)
if msg:
ctp_logs.append(str(msg))
if len(ctp_logs) > 20:
if len(ctp_logs) > 40:
ctp_logs.pop(0)
logger.info("CTP | %s", msg)
@@ -236,27 +293,36 @@ class CtpBridge:
setting.get("交易服务器"),
setting.get("柜台环境", "实盘"),
)
td_addr = setting.get("交易服务器", "")
ok, err = probe_tcp_address(td_addr, timeout=5.0)
if not ok:
raise RuntimeError(
f"SimNow 交易前置不可达：{td_addr}（{err}）。"
"请更新 .env 中 SIMNOW_TD_ADDRESS 为官网最新地址，"
"并在服务器执行 nc -zv 验证出网。"
)
self._engine.connect(setting, GATEWAY_NAME)
for _ in range(60):
accounts = self._engine.get_all_accounts()
if accounts:
self._connected_mode = mode
self._last_error = ""
logger.info("CTP 已连接 [%s] account=%s", mode, len(accounts))
self._install_position_margin_hook()
self._schedule_fee_sync(mode)
try:
self.refresh_positions()
except Exception as exc:
logger.debug("initial position query: %s", exc)
_fire_position_refresh_callback()
return
time.sleep(0.5)
if self._wait_connected(mode):
self._connected_mode = mode
self._last_error = ""
logger.info("CTP 已连接 [%s] td_login=%s accounts=%s",
mode, self._td_logged_in(),
len(self._engine.get_all_accounts() or []))
self._install_position_margin_hook()
self._schedule_fee_sync(mode)
try:
self.refresh_positions()
except Exception as exc:
logger.debug("initial position query: %s", exc)
_fire_position_refresh_callback()
return
finally:
self._ee.unregister(EVENT_LOG, _on_log)
hint = _format_ctp_failure(ctp_logs)
self._close_gateway()
hint = _format_ctp_failure(ctp_logs, td_address=setting.get("交易服务器", ""))
self._last_error = hint
logger.warning("CTP 连接失败 [%s]: %s | logs=%s", mode, hint, ctp_logs[-5:])
raise RuntimeError(hint)
finally:
self._connect_in_progress = False
@@ -1040,12 +1106,20 @@ def ctp_try_auto_reconnect(mode: str) -> bool:
if not b.available():
return False
if b.connect_in_progress():
return True
return False
st = _setting_for_mode(mode)
if not st.get("用户名") or not st.get("密码") or not st.get("交易服务器"):
return False
if b.connected_mode == mode and b.ping():
return True
td = st.get("交易服务器", "")
ok, err = probe_tcp_address(td, timeout=4.0)
if not ok:
b._last_error = (
f"SimNow 交易前置不可达：{td}（{err}）。"
"请更新 SIMNOW_TD_ADDRESS 并确认服务器出网。"
)
return False
info = b.start_connect_async(mode, force=False)
return bool(
info.get("connected")
@@ -1055,7 +1129,18 @@ def ctp_try_auto_reconnect(mode: str) -> bool:
def ctp_status(mode: str) -> dict[str, Any]:
return get_bridge().status(mode)
st = get_bridge().status(mode)
if not st.get("connected") and not st.get("connecting"):
setting = _setting_for_mode(mode)
td = setting.get("交易服务器", "")
if td:
ok, err = probe_tcp_address(td, timeout=3.0)
st["td_reachable"] = ok
if not ok and not st.get("last_error"):
st["last_error"] = (
f"SimNow 交易前置不可达：{td}（{err}）"
)
return st
def ctp_get_account(mode: str) -> dict[str, Any]: