Optimize

py-client/libs/__init__.py

@@ -1,6 +1,6 @@
 from .calc import calc_buy_volume, trading_time
 from .lockfile import is_lock, write_lockfile
-from .market import market_allow_open, status
+from .market import market_allow_open, refresh_market, status
 from .signal import SignalItem, SignalResult, fetch_signal, init_signals
 
 __all__ = [
@@ -9,6 +9,7 @@ __all__ = [
     "is_lock",
     "write_lockfile",
     "market_allow_open",
+    "refresh_market",
     "status",
     "SignalItem",
     "SignalResult",

py-client/libs/market.py

@@ -1,10 +1,13 @@
 import logging
 import secrets
+from threading import Lock
 
 from .http import get_json
 
 API_HOST = "http://139.224.247.176:13499"
 MARKET_URL, PERIOD, HTTP_TIMEOUT = "/a/market", "60m", 5.0
+_market_lock = Lock()
+_market_status = "UNKNOWN"
 
 
 def status(payload) -> str:
@@ -15,10 +18,22 @@ def status(payload) -> str:
     return result if result in {"UP", "DOWN", "NEUTRAL"} else "UNKNOWN"
 
 
-def market_allow_open(api_host: str = API_HOST) -> bool:
+def refresh_market(api_host: str = API_HOST) -> str:
+    """由后台调度线程刷新大盘状态；请求失败时缓存为 UNKNOWN。"""
+    global _market_status
     url = f"{api_host}{MARKET_URL}?period={PERIOD}&t={secrets.token_urlsafe(12)}"
-    try: result = status(get_json(url, HTTP_TIMEOUT))
+    try:
+        result = status(get_json(url, HTTP_TIMEOUT))
     except Exception as exc:
-        logging.error("获取大盘指数失败: %s %s", url, exc); return False
+        result = "UNKNOWN"
+        logging.error("获取大盘指数失败: %s %s", url, exc)
+    with _market_lock:
+        _market_status = result
     logging.info("大盘信号: url=%s status=%s", url, result)
-    return result == "UP"
+    return result
+
+
+def market_allow_open() -> bool:
+    """读取最近一次后台刷新得到的大盘缓存；未知状态时禁止开仓。"""
+    with _market_lock:
+        return _market_status == "UP"

py-client/main.py

@@ -4,6 +4,7 @@
 import logging
 import os
 import sys
+from datetime import datetime
 from apscheduler.schedulers.background import BackgroundScheduler
 import config
 from dataclasses import dataclass
@@ -22,6 +23,7 @@ logging.basicConfig(
 )
 
 from sdk import APIError, Client
+from libs.market import refresh_market
 from strategy.trend.boot import StartTrend
 from strategy.zt.boot import StartZT
 from strategy.ipo import AutoBuyIpo
@@ -122,8 +124,18 @@ def main() -> int:
             id="auto_buy_ipo",
             replace_existing=True,
         )
+        scheduler.add_job(
+            refresh_market,
+            trigger="interval",
+            minutes=1,
+            args=[config.global_config.api_host],
+            id="market_refresh",
+            replace_existing=True,
+            next_run_time=datetime.now(),
+        )
         scheduler.start()
         logging.info("IPO 自动打新定时任务已启动：每日 10:00、14:00")
+        logging.info("大盘信号后台刷新已启动：每分钟一次")
 
         STRATEGIES[config.account_config.strategy].start_strategy()
         logging.info("%s 策略启动成功",config.account_config.strateg)

py-client/strategy/trend/boot.py

@@ -7,6 +7,7 @@ from __future__ import annotations
 
 import logging
 import time
+from concurrent.futures import Future, ThreadPoolExecutor
 from datetime import datetime
 
 import config
@@ -25,9 +26,6 @@ from .positions import manage_positions
 
 def Overview(assets, positions, account_cfg=None) -> None:
     """打印策略启动时的账户、资金和持仓概览。
-
-    该函数对应 Go 客户端 ``logic.Overview``。为便于单独测试，可以
-    显式传入账户配置；未传入时使用 ``config.account_config``。
     """
     account_cfg = account_cfg or config.account_config
 
@@ -99,20 +97,24 @@ def StartTrend() -> None:
         open_watch=DipWatch(),
         add_watch=DipWatch(),
         profit_tracker=GridTrailingTracker(config.account_config.grid_step_pct),
+        executor=ThreadPoolExecutor(max_workers=2, thread_name_prefix="trend"),
     )
 
     Overview(assets, positions, config.account_config)
 
-    while True:
+    try:
-        started_at = time.monotonic()
+        while True:
-        try:
+            started_at = time.monotonic()
-            RunOnce(run, signals)
+            try:
-        except Exception:
+                RunOnce(run, signals)
-            # 单轮错误只记录日志，下一轮仍继续运行。
+            except Exception:
-            logging.exception("趋势策略本轮执行失败")
+                # 单轮错误只记录日志，下一轮仍继续运行。
+                logging.exception("趋势策略本轮执行失败")
 
-        elapsed = time.monotonic() - started_at
+            elapsed = time.monotonic() - started_at
-        time.sleep(max(0.0, 30.0 - elapsed))
+            time.sleep(max(0.0, 30.0 - elapsed))
+    finally:
+        run.executor.shutdown(wait=True, cancel_futures=True)
 
 
 def RunOnce(run: Runtime, signals:list[SignalItem]) -> None:
@@ -148,14 +150,7 @@ def RunOnce(run: Runtime, signals:list[SignalItem]) -> None:
         logging.exception("获取持仓失败")
         return
 
-    # 5. 更新状态机
+    # 5. 验证有效开仓信号：排除已有持仓和未决订单。
-    try:
-        run.state.reconcile(positions, run.orders.data)
-    except Exception:
-        logging.exception("订单状态对账失败，本轮禁止自动交易")
-        return
-
-    # 6. 验证有效开仓信号：排除已有持仓和未决订单。
     allow_open: list[SignalItem] = []
     allow_codes: list[str] = []
     for signal in signals:
@@ -164,16 +159,48 @@ def RunOnce(run: Runtime, signals:list[SignalItem]) -> None:
             allow_codes.append(signal.code)
 
     # 6. 获取持仓和待开仓证券的实时行情 tick。
-    all_codes = allow_codes + position_codes
+    all_codes = list(dict.fromkeys(position_codes + allow_codes))
     try:
         ticks = run.client.full_tick(all_codes)
     except Exception:
         logging.exception("获取行情失败")
         return
 
-    # 7. 执行开仓：必须同时存在有效信号且大盘允许开仓。
+    # 7. 更新状态机
-    if allow_open and market_ok and allow_open_by_cash:
+    try:
-        open_signal(run, ticks, allow_open)
+        run.state.reconcile(positions, run.orders.data)
+    except Exception:
+        logging.exception("订单状态对账失败，本轮禁止自动交易")
+        return
 
-    # 8. 持仓计算。当前 Go 版本的 managePositions 为空，保留扩展入口。
+    # 启动线程，开始计算
-    manage_positions(run, ticks, positions, market_ok,assets.available)
+    # 9. 持仓计算。
+    futures: list[tuple[str, Future]] = [
+        (
+            "持仓计算",
+            run.executor.submit(
+                manage_positions,
+                run,
+                ticks,
+                positions,
+                market_ok,
+                assets.available,
+            ),
+        )
+    ]
+
+    # 10. 开仓计算：必须同时存在有效信号且大盘允许开仓。
+    if allow_open and market_ok and allow_open_by_cash:
+        futures.append(("开仓计算", run.executor.submit(open_signal, run, ticks, allow_open)))
+
+    # 11. 开始执行
+    for name, future in futures:
+        _wait_worker(name, future)
+
+
+def _wait_worker(name: str, future: Future) -> None:
+    """保留单轮继续运行的语义，分别记录工作线程异常。"""
+    try:
+        future.result()
+    except Exception:
+        logging.exception("趋势策略%s线程失败", name)

py-client/strategy/trend/open.py

@@ -1,4 +1,4 @@
-"""趋势策略开仓逻辑，对应 Go 版本的 ``logic/open.go``。"""
+"""趋势策略开仓逻辑。"""
 
 from __future__ import annotations
 

py-client/strategy/trend/runtime.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from concurrent.futures import ThreadPoolExecutor
 from dataclasses import dataclass, field
 
 from config import AccountConfig, GlobalConfig
@@ -42,3 +43,4 @@ class Runtime:
     open_watch: DipWatch
     add_watch: DipWatch
     profit_tracker: GridTrailingTracker
+    executor: ThreadPoolExecutor

py-client/strategy/zt/boot.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import logging
 import time
+from concurrent.futures import Future, ThreadPoolExecutor
 from datetime import datetime, time as clock_time
 
 import config
@@ -15,7 +16,7 @@ from sdk import Client
 from strategy.trend.order import OrderBook
 from strategy.trend.watch import DipWatch
 
-from .open import open_base
+from .open import open_signal
 from .positions import manage_positions
 from .runtime import Runtime
 from .state import TState
@@ -50,11 +51,10 @@ def RunOnce(run: Runtime) -> None:
         return
     today = datetime.now().date().isoformat()
     try:
-        run.state.reconcile(positions, run.orders.data, today)
+        signals = init_signals(run.global_cfg, run.account_cfg.signal_allow)
     except Exception:
-        logging.exception("[ZT] 状态对账失败")
+        logging.exception("[ZT] 获取 dcm 信号失败")
         return
-    signals = init_signals(run.global_cfg, run.account_cfg.signal_allow)
     candidate_codes = [item.code for item in signals if item.code not in position_codes]
     codes = list(dict.fromkeys(position_codes + candidate_codes))
     try:
@@ -63,13 +63,50 @@ def RunOnce(run: Runtime) -> None:
         logging.exception("[ZT] 获取行情失败")
         return
     market_ok = market_allow_open(run.global_cfg.api_host)
-    if market_ok and assets.available >= assets.total * run.account_cfg.min_cash_ratio:
+    can_open = market_ok and assets.available >= assets.total * run.account_cfg.min_cash_ratio
-        open_base(run, ticks, signals)
+    force_buy_back = datetime.now().time() >= clock_time(14, 50)
-    manage_positions(
+
-        run,
+    # 状态对账与开仓判断并行。持仓线程在自己的线程中等待对账完成，
-        ticks,
+    # 以保证它读取到最新的底仓和做 T 轮次状态，避免并发写 State。
-        positions,
+    with ThreadPoolExecutor(max_workers=3, thread_name_prefix="zt") as executor:
-        assets.available,
+        state_future = executor.submit(run.state.reconcile, positions, run.orders.data, today)
-        today,
+        open_future = executor.submit(_run_open_signal, state_future, run, ticks, signals, can_open)
-        force_buy_back=datetime.now().time() >= clock_time(14, 50),
+        positions_future = executor.submit(
-    )
+            _run_manage_positions,
+            state_future,
+            run,
+            ticks,
+            positions,
+            assets.available,
+            today,
+            force_buy_back,
+        )
+        _wait_worker("状态对账", state_future)
+        _wait_worker("开仓", open_future)
+        _wait_worker("持仓管理", positions_future)
+
+
+def _run_open_signal(state_future: Future, run: Runtime, ticks, signals, can_open: bool) -> None:
+    state_future.result()
+    if can_open:
+        open_signal(run, ticks, signals)
+
+
+def _run_manage_positions(
+    state_future: Future,
+    run: Runtime,
+    ticks,
+    positions,
+    available: float,
+    today: str,
+    force_buy_back: bool,
+) -> None:
+    state_future.result()
+    manage_positions(run, ticks, positions, available, today, force_buy_back)
+
+
+def _wait_worker(name: str, future: Future) -> None:
+    try:
+        future.result()
+    except Exception:
+        logging.exception("[ZT] %s线程失败", name)

py-client/strategy/zt/open.py

@@ -9,7 +9,7 @@ from sdk import OP_BUY
 from strategy.trend.order import PlaceOrderRequest
 
 
-def open_base(run, ticks, signals) -> None:
+def open_signal(run, ticks, signals) -> None:
     """仅处理 dcm 信号，使用趋势策略同款反弹确认建立底仓。"""
     for signal in signals:
         if signal.signal_key != "dcm" or run.orders.busy(signal.code, "BUY"):
@@ -25,3 +25,7 @@ def open_base(run, ticks, signals) -> None:
         if run.orders.place(request):
             run.buy_watch.forget(signal.code)
             logging.info("[ZT 建仓] %s 买入 %d 股", signal.code, volume)
+
+
+# 与 trend 策略的开仓函数命名保持一致。
+open_base = open_signal

py-client/tests/test_market.py

@@ -0,0 +1,22 @@
+from __future__ import annotations
+
+import unittest
+from unittest.mock import patch
+
+from libs.market import market_allow_open, refresh_market
+
+
+class MarketCacheTests(unittest.TestCase):
+    def test_refresh_updates_open_cache(self):
+        with patch("libs.market.get_json", return_value={"data": {"action": "UP"}}):
+            self.assertEqual(refresh_market("http://example"), "UP")
+        self.assertTrue(market_allow_open())
+
+    def test_refresh_failure_blocks_open(self):
+        with patch("libs.market.get_json", side_effect=OSError("offline")):
+            self.assertEqual(refresh_market("http://example"), "UNKNOWN")
+        self.assertFalse(market_allow_open())
+
+
+if __name__ == "__main__":
+    unittest.main()

py-client/tests/test_trend.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import unittest
+from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime, timedelta
 from tempfile import TemporaryDirectory
 from types import SimpleNamespace
@@ -151,28 +152,29 @@ class TrendTests(unittest.TestCase):
             deals=lambda: [],
             full_tick=lambda _codes: {"A": Tick(last_price=11)},
         )
-        runtime = SimpleNamespace(
+        with ThreadPoolExecutor(max_workers=2) as executor:
-            client=client,
+            runtime = SimpleNamespace(
-            account_cfg=SimpleNamespace(min_cash_ratio=0.1),
+                client=client,
-            global_cfg=SimpleNamespace(api_host="http://example"),
+                account_cfg=SimpleNamespace(min_cash_ratio=0.1),
-            orders=SimpleNamespace(refresh=lambda _client: None),
+                global_cfg=SimpleNamespace(api_host="http://example"),
-            state=SimpleNamespace(
+                orders=SimpleNamespace(refresh=lambda _client: None, data=[]),
-                codes=["A"],
+                state=SimpleNamespace(
-                unresolved_codes=[],
+                    codes=["A"],
-                reconcile=lambda *_args: None,
+                    reconcile=lambda *_args: None,
-            ),
+                ),
-        )
+                executor=executor,
-        with (
+            )
-            patch("strategy.trend.boot.trading_time", return_value=True),
+            with (
-            patch("strategy.trend.boot.market_allow_open", return_value=True),
+                patch("strategy.trend.boot.trading_time", return_value=True),
-            patch("strategy.trend.boot.open_signal") as open_mock,
+                patch("strategy.trend.boot.market_allow_open", return_value=True),
-            patch("strategy.trend.boot.manage_positions") as manage_mock,
+                patch("strategy.trend.boot.open_signal") as open_mock,
-        ):
+                patch("strategy.trend.boot.manage_positions") as manage_mock,
-            RunOnce(runtime, [])
+            ):
+                RunOnce(runtime, [])
         open_mock.assert_not_called()
         manage_mock.assert_called_once()
 
-    def test_unknown_order_without_position_blocks_reopen(self):
+    def test_state_without_broker_order_allows_reopen(self):
         with TemporaryDirectory() as directory:
             state = State.for_strategy(directory, "trend", "A")
             state.set(StateItem(
@@ -189,26 +191,28 @@ class TrendTests(unittest.TestCase):
                 deals=lambda: [],
                 full_tick=lambda _codes: {"A": Tick(last_price=10)},
             )
-            runtime = SimpleNamespace(
-                client=client,
-                account_cfg=SimpleNamespace(min_cash_ratio=0.1),
-                global_cfg=SimpleNamespace(api_host="http://example"),
-                orders=OrderBook(),
-                state=state,
-                open_watch=SimpleNamespace(forget=lambda _code: None),
-                add_watch=SimpleNamespace(forget=lambda _code: None),
-            )
             signal = SimpleNamespace(code="A", signal_key="morning")
-            with (
+            with ThreadPoolExecutor(max_workers=2) as executor:
-                patch("strategy.trend.boot.trading_time", return_value=True),
+                runtime = SimpleNamespace(
-                patch("strategy.trend.boot.market_allow_open", return_value=True),
+                    client=client,
-                patch("strategy.trend.boot.open_signal") as open_mock,
+                    account_cfg=SimpleNamespace(min_cash_ratio=0.1),
-                patch("strategy.trend.boot.manage_positions"),
+                    global_cfg=SimpleNamespace(api_host="http://example"),
-            ):
+                    orders=OrderBook(),
-                RunOnce(runtime, [signal])
+                    state=state,
+                    open_watch=SimpleNamespace(forget=lambda _code: None),
+                    add_watch=SimpleNamespace(forget=lambda _code: None),
+                    executor=executor,
+                )
+                with (
+                    patch("strategy.trend.boot.trading_time", return_value=True),
+                    patch("strategy.trend.boot.market_allow_open", return_value=True),
+                    patch("strategy.trend.boot.open_signal") as open_mock,
+                    patch("strategy.trend.boot.manage_positions"),
+                ):
+                    RunOnce(runtime, [signal])
 
-            open_mock.assert_not_called()
+            open_mock.assert_called_once()
-            self.assertTrue(state.has_unresolved_order("A"))
+            self.assertEqual(state.codes, [])
 
 
 if __name__ == "__main__":