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

feat: CTP/SimNow 配置迁入系统设置，登录失败即时报错

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-06-25 16:46:01 +08:00
parent 72361233a0
commit 5a6c89c662
7 changed files with 267 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
vnpy_bridge.py
+12 -27
View File
@@ -12,6 +12,7 @@ from locale_fix import ensure_process_locale
ensure_process_locale()
from ctp_settings import live_setting_dict, simnow_setting_dict
from ctp_symbol import ths_to_vnpy_symbol, to_vnpy_exchange
from contract_specs import get_contract_spec
@@ -43,35 +44,13 @@ _bridge: Optional["CtpBridge"] = None
_bridge_lock = threading.Lock()
def _env(key: str, default: str = "") -> str:
return (os.getenv(key) or default).strip()
def _simnow_setting() -> dict[str, str]:
"""SimNow 仿真前置（可在 .env 覆盖）。看穿式前置需「柜台环境=实盘」。"""
return {
"用户名": _env("SIMNOW_USER"),
"密码": _env("SIMNOW_PASSWORD"),
"经纪商代码": _env("SIMNOW_BROKER_ID", "9999"),
"交易服务器": _env("SIMNOW_TD_ADDRESS", "tcp://180.168.146.187:10201"),
"行情服务器": _env("SIMNOW_MD_ADDRESS", "tcp://180.168.146.187:10211"),
"产品名称": _env("SIMNOW_APP_ID", "simnow_client_test"),
"授权编码": _env("SIMNOW_AUTH_CODE", "0000000000000000"),
"柜台环境": _env("SIMNOW_ENV", "实盘"),
}
"""SimNow 仿真前置（系统设置优先，.env 兜底）。"""
return simnow_setting_dict()
def _live_setting() -> dict[str, str]:
return {
"用户名": _env("CTP_LIVE_USER"),
"密码": _env("CTP_LIVE_PASSWORD"),
"经纪商代码": _env("CTP_LIVE_BROKER_ID"),
"交易服务器": _env("CTP_LIVE_TD_ADDRESS"),
"行情服务器": _env("CTP_LIVE_MD_ADDRESS"),
"产品名称": _env("CTP_LIVE_APP_ID"),
"授权编码": _env("CTP_LIVE_AUTH_CODE"),
"柜台环境": _env("CTP_LIVE_ENV", "实盘"),
}
return live_setting_dict()
def _setting_for_mode(mode: str) -> dict[str, str]:
@@ -206,12 +185,18 @@ class CtpBridge:
self._connected_mode = None
time.sleep(0.6)
def _wait_connected(self, mode: str) -> bool:
def _login_rejected(self, ctp_logs: list[str]) -> bool:
return any("登录失败" in m or "不合法的登录" in m for m in ctp_logs)
def _wait_connected(self, mode: str, ctp_logs: list[str] | None = None) -> bool:
"""等待账户回报或交易通道登录成功。"""
if not self._engine:
return False
logs = ctp_logs or []
loops = max(1, int(CONNECT_WAIT_SEC / CONNECT_POLL_INTERVAL_SEC))
for _ in range(loops):
if self._login_rejected(logs):
return False
try:
if self._engine.get_all_accounts():
return True
@@ -302,7 +287,7 @@ class CtpBridge:
"并在服务器执行 nc -zv 验证出网。"
)
self._engine.connect(setting, GATEWAY_NAME)
if self._wait_connected(mode):
if self._wait_connected(mode, ctp_logs):
self._connected_mode = mode
self._last_error = ""
logger.info("CTP 已连接 [%s] td_login=%s accounts=%s",