From 72361233a0f5df9fd214149c57012387533fe5ee Mon Sep 17 00:00:00 2001 From: dekun Date: Thu, 25 Jun 2026 16:30:35 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20CTP=E9=87=8D=E8=BF=9E=E5=89=8D=E6=8E=A2?= =?UTF-8?q?=E6=B5=8B=E5=89=8D=E7=BD=AE=E5=8F=AF=E8=BE=BE=E6=80=A7=EF=BC=8C?= =?UTF-8?q?=E5=A4=B1=E8=B4=A5=E6=97=B6=E5=85=B3=E9=97=AD=E7=BD=91=E5=85=B3?= =?UTF-8?q?=E5=B9=B6=E6=98=8E=E7=A1=AE=E6=8A=A5=E9=94=99?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Cursor --- deploy.sh | 2 +- docs/SIMNOW.md | 10 ++-- static/js/trade.js | 46 ++++++++++++--- vnpy_bridge.py | 141 ++++++++++++++++++++++++++++++++++++--------- 4 files changed, 157 insertions(+), 42 deletions(-) diff --git a/deploy.sh b/deploy.sh index 1e78317..37b98e2 100644 --- a/deploy.sh +++ b/deploy.sh @@ -18,8 +18,8 @@ SERVICE_NAME="qihuo" # SimNow 前置候选（按优先级；部署时自动 nc 探测） SIMNOW_FRONTS=( - "182.254.243.31:30001:30011" "180.168.146.187:10201:10211" + "180.168.146.187:10202:10212" "180.168.146.187:10130:10131" "218.202.237.33:10203:10213" ) diff --git a/docs/SIMNOW.md b/docs/SIMNOW.md index 8b02d36..5afc6dc 100644 --- a/docs/SIMNOW.md +++ b/docs/SIMNOW.md @@ -139,15 +139,15 @@ tcp://IP:端口 pm2 restart qihuo ``` -**云服务器若 `180.168.146.187` 端口超时**，可改用备用前置（交易时段）： +**云服务器网络说明**：`182.254.243.31` 段前置已停用（Connection refused），请勿再使用。官方前置为 `180.168.146.187:10201/10211`。若服务器 `nc -zv 180.168.146.187 10201` 超时，属于**出网/防火墙**问题，需联系云厂商放行或换能访问 SimNow 的网络，无法仅靠改代码解决。 + +旧文档备用地址（已失效，仅作排查参考）： ```env -SIMNOW_TD_ADDRESS=tcp://182.254.243.31:30001 -SIMNOW_MD_ADDRESS=tcp://182.254.243.31:30011 +# 勿用 — 已 dead +# SIMNOW_TD_ADDRESS=tcp://182.254.243.31:30001 ``` -服务器上用 `nc -zv 182.254.243.31 30001` 验证连通后再配置。 - ### 3. 网页端连接 CTP 1. 登录本系统 diff --git a/static/js/trade.js b/static/js/trade.js index ec957c9..0089c2f 100644 --- a/static/js/trade.js +++ b/static/js/trade.js @@ -19,7 +19,9 @@ var lastQuotePrice = null; var priceType = 'limit'; var lastCtpReconnectAt = 0; + var lastCtpUnreachableAt = 0; var ctpReconnecting = false; + var ctpConnectInflight = false; var isTradingSession = false; var hasSlTpMonitoring = false; var ctpConnected = false; @@ -121,6 +123,15 @@ } catch (e) { /* quota */ } } + function showCtpError(msg) { + var hint = document.querySelector('.ctp-install-hint'); + if (hint) hint.textContent = msg || ''; + } + + function isCtpUnreachableError(msg) { + return !!(msg && (msg.indexOf('不可达') >= 0 || msg.indexOf('Connection refused') >= 0 || msg.indexOf('timed out') >= 0)); + } + function applyPositionsData(data) { if (!list || !data) return; var cap = document.getElementById('cap-display'); @@ -130,6 +141,12 @@ ctpConnected = !!connected; isTradingSession = !!data.trading_session; updateCtpBadge(!!connected, !!connecting); + if (!connected && !connecting && data.ctp_status && data.ctp_status.last_error) { + showCtpError(data.ctp_status.last_error); + if (isCtpUnreachableError(data.ctp_status.last_error)) { + lastCtpUnreachableAt = Date.now(); + } + } var riskBadge = document.getElementById('risk-badge'); if (riskBadge && data.risk_status) { riskBadge.textContent = data.risk_status.status_label || ''; @@ -300,7 +317,7 @@ } function waitForCtpConnected(maxMs) { - var deadline = Date.now() + (maxMs || 35000); + var deadline = Date.now() + (maxMs || 70000); function tick() { return fetch('/api/ctp/status') .then(function (r) { return r.json(); }) @@ -308,6 +325,7 @@ var st = d.status || {}; if (st.connected) { updateCtpBadge(true, false); + showCtpError(''); if (d.account && d.account.available != null) { var avail = document.getElementById('avail-display'); if (avail) avail.textContent = Number(d.account.available).toFixed(2); @@ -323,8 +341,10 @@ } updateCtpBadge(false, false); if (st.last_error) { - var hint = document.querySelector('.ctp-install-hint'); - if (hint) hint.textContent = st.last_error; + showCtpError(st.last_error); + if (isCtpUnreachableError(st.last_error)) { + lastCtpUnreachableAt = Date.now(); + } } return false; }) @@ -334,6 +354,10 @@ } function requestCtpConnect(force) { + if (!force && ctpConnectInflight) { + return Promise.resolve({}); + } + ctpConnectInflight = true; updateCtpBadge(false, true); return fetch('/api/ctp/connect', { method: 'POST', @@ -344,24 +368,29 @@ .then(function (d) { if (d.status && d.status.connected) { updateCtpBadge(true, false); + showCtpError(''); pollPositions(); return d; } if (d.connecting || (d.status && d.status.connecting)) { - return waitForCtpConnected(35000).then(function (ok) { - if (!ok && d.error) alert(d.error); - else if (!ok && d.status && d.status.last_error) alert(d.status.last_error); + return waitForCtpConnected(70000).then(function (ok) { + if (!ok && d.error) showCtpError(d.error); + else if (!ok && d.status && d.status.last_error) showCtpError(d.status.last_error); return d; }); } if (!d.ok) { updateCtpBadge(false, false); - alert(d.error || (d.status && d.status.last_error) || '连接失败'); + var err = d.error || (d.status && d.status.last_error) || '连接失败'; + showCtpError(err); } return d; }) .catch(function () { updateCtpBadge(false, false); + }) + .finally(function () { + ctpConnectInflight = false; }); } @@ -465,9 +494,10 @@ } function tryAutoCtpReconnect() { - if (ctpReconnecting) return; + if (ctpReconnecting || ctpConnectInflight) return; var now = Date.now(); if (now - lastCtpReconnectAt < 60000) return; + if (lastCtpUnreachableAt && now - lastCtpUnreachableAt < 300000) return; lastCtpReconnectAt = now; ctpReconnecting = true; requestCtpConnect(false).finally(function () { diff --git a/vnpy_bridge.py b/vnpy_bridge.py index f2aedf1..8d0592a 100644 --- a/vnpy_bridge.py +++ b/vnpy_bridge.py @@ -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]: