dekun
40 lines
1.1 KiB
Python
40 lines
1.1 KiB
Python
"""CTP 断线自动重连（后台线程）。"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
import threading
|
|
import time
|
|
from typing import Callable
|
|
|
|
from vnpy_bridge import ctp_try_auto_reconnect
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
RECONNECT_INTERVAL_SEC = 10
|
|
|
|
|
|
def _auto_reconnect_enabled() -> bool:
|
|
return (os.getenv("CTP_AUTO_RECONNECT", "true") or "true").strip().lower() in (
|
|
"1",
|
|
"true",
|
|
"yes",
|
|
)
|
|
|
|
|
|
def start_ctp_reconnect_worker(*, get_mode_fn: Callable[[], str], interval: int = RECONNECT_INTERVAL_SEC) -> None:
|
|
"""定时检测 CTP 连接，断线后自动重连。"""
|
|
|
|
def _loop() -> None:
|
|
while True:
|
|
try:
|
|
if _auto_reconnect_enabled():
|
|
mode = get_mode_fn()
|
|
if ctp_try_auto_reconnect(mode):
|
|
logger.debug("CTP 连接正常 [%s]", mode)
|
|
except Exception as exc:
|
|
logger.warning("CTP reconnect worker: %s", exc)
|
|
time.sleep(max(5, interval))
|
|
|
|
threading.Thread(target=_loop, daemon=True, name="ctp-reconnect-worker").start()
|