diff --git a/py-client/__pycache__/main.cpython-311.pyc b/py-client/__pycache__/main.cpython-311.pyc index b99eac3..bb08f26 100644 Binary files a/py-client/__pycache__/main.cpython-311.pyc and b/py-client/__pycache__/main.cpython-311.pyc differ diff --git a/py-client/libs/__init__.py b/py-client/libs/__init__.py index c899b38..4ee0cac 100644 --- a/py-client/libs/__init__.py +++ b/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", diff --git a/py-client/libs/__pycache__/__init__.cpython-311.pyc b/py-client/libs/__pycache__/__init__.cpython-311.pyc index d082834..542a67a 100644 Binary files a/py-client/libs/__pycache__/__init__.cpython-311.pyc and b/py-client/libs/__pycache__/__init__.cpython-311.pyc differ diff --git a/py-client/libs/__pycache__/market.cpython-311.pyc b/py-client/libs/__pycache__/market.cpython-311.pyc index 9f8ad48..b7f5170 100644 Binary files a/py-client/libs/__pycache__/market.cpython-311.pyc and b/py-client/libs/__pycache__/market.cpython-311.pyc differ diff --git a/py-client/libs/market.py b/py-client/libs/market.py index 21192e9..1ae2107 100644 --- a/py-client/libs/market.py +++ b/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" diff --git a/py-client/main.py b/py-client/main.py index 580eb84..cc40e97 100644 --- a/py-client/main.py +++ b/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) diff --git a/py-client/strategy/ipo/__pycache__/boot.cpython-311.pyc b/py-client/strategy/ipo/__pycache__/boot.cpython-311.pyc index 141828a..0e1d348 100644 Binary files a/py-client/strategy/ipo/__pycache__/boot.cpython-311.pyc and b/py-client/strategy/ipo/__pycache__/boot.cpython-311.pyc differ diff --git a/py-client/strategy/trend/__pycache__/boot.cpython-311.pyc b/py-client/strategy/trend/__pycache__/boot.cpython-311.pyc index 1aa5c8f..643655c 100644 Binary files a/py-client/strategy/trend/__pycache__/boot.cpython-311.pyc and b/py-client/strategy/trend/__pycache__/boot.cpython-311.pyc differ diff --git a/py-client/strategy/trend/__pycache__/open.cpython-311.pyc b/py-client/strategy/trend/__pycache__/open.cpython-311.pyc index d2e6070..4218111 100644 Binary files a/py-client/strategy/trend/__pycache__/open.cpython-311.pyc and b/py-client/strategy/trend/__pycache__/open.cpython-311.pyc differ diff --git a/py-client/strategy/trend/__pycache__/runtime.cpython-311.pyc b/py-client/strategy/trend/__pycache__/runtime.cpython-311.pyc index e81d92c..5659c4d 100644 Binary files a/py-client/strategy/trend/__pycache__/runtime.cpython-311.pyc and b/py-client/strategy/trend/__pycache__/runtime.cpython-311.pyc differ diff --git a/py-client/strategy/trend/boot.py b/py-client/strategy/trend/boot.py index 529cbe0..eaaf29d 100644 --- a/py-client/strategy/trend/boot.py +++ b/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: - started_at = time.monotonic() - try: - RunOnce(run, signals) - except Exception: - # 单轮错误只记录日志，下一轮仍继续运行。 - logging.exception("趋势策略本轮执行失败") + try: + while True: + started_at = time.monotonic() + try: + RunOnce(run, signals) + except Exception: + # 单轮错误只记录日志，下一轮仍继续运行。 + logging.exception("趋势策略本轮执行失败") - elapsed = time.monotonic() - started_at - time.sleep(max(0.0, 30.0 - elapsed)) + elapsed = time.monotonic() - started_at + 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. 更新状态机 - try: - run.state.reconcile(positions, run.orders.data) - except Exception: - logging.exception("订单状态对账失败，本轮禁止自动交易") - return - - # 6. 验证有效开仓信号：排除已有持仓和未决订单。 + # 5. 验证有效开仓信号：排除已有持仓和未决订单。 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. 执行开仓：必须同时存在有效信号且大盘允许开仓。 - if allow_open and market_ok and allow_open_by_cash: - open_signal(run, ticks, allow_open) + # 7. 更新状态机 + try: + 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) diff --git a/py-client/strategy/trend/open.py b/py-client/strategy/trend/open.py index bc19391..e8040be 100644 --- a/py-client/strategy/trend/open.py +++ b/py-client/strategy/trend/open.py @@ -1,4 +1,4 @@ -"""趋势策略开仓逻辑，对应 Go 版本的 ``logic/open.go``。""" +"""趋势策略开仓逻辑。""" from __future__ import annotations diff --git a/py-client/strategy/trend/runtime.py b/py-client/strategy/trend/runtime.py index 40b0a18..fcf3ce7 100644 --- a/py-client/strategy/trend/runtime.py +++ b/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 diff --git a/py-client/strategy/zt/__pycache__/boot.cpython-311.pyc b/py-client/strategy/zt/__pycache__/boot.cpython-311.pyc index 0fb171f..4810a0e 100644 Binary files a/py-client/strategy/zt/__pycache__/boot.cpython-311.pyc and b/py-client/strategy/zt/__pycache__/boot.cpython-311.pyc differ diff --git a/py-client/strategy/zt/__pycache__/open.cpython-311.pyc b/py-client/strategy/zt/__pycache__/open.cpython-311.pyc index 0dfea26..01a2f16 100644 Binary files a/py-client/strategy/zt/__pycache__/open.cpython-311.pyc and b/py-client/strategy/zt/__pycache__/open.cpython-311.pyc differ diff --git a/py-client/strategy/zt/boot.py b/py-client/strategy/zt/boot.py index 01d2733..c64b364 100644 --- a/py-client/strategy/zt/boot.py +++ b/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: - open_base(run, ticks, signals) - manage_positions( - run, - ticks, - positions, - assets.available, - today, - force_buy_back=datetime.now().time() >= clock_time(14, 50), - ) + can_open = market_ok and assets.available >= assets.total * run.account_cfg.min_cash_ratio + force_buy_back = datetime.now().time() >= clock_time(14, 50) + + # 状态对账与开仓判断并行。持仓线程在自己的线程中等待对账完成， + # 以保证它读取到最新的底仓和做 T 轮次状态，避免并发写 State。 + with ThreadPoolExecutor(max_workers=3, thread_name_prefix="zt") as executor: + state_future = executor.submit(run.state.reconcile, positions, run.orders.data, today) + open_future = executor.submit(_run_open_signal, state_future, run, ticks, signals, can_open) + 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) diff --git a/py-client/strategy/zt/open.py b/py-client/strategy/zt/open.py index 5749f7c..6cdfed8 100644 --- a/py-client/strategy/zt/open.py +++ b/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 diff --git a/py-client/tests/__pycache__/test_market.cpython-311.pyc b/py-client/tests/__pycache__/test_market.cpython-311.pyc new file mode 100644 index 0000000..9f6a22e Binary files /dev/null and b/py-client/tests/__pycache__/test_market.cpython-311.pyc differ diff --git a/py-client/tests/__pycache__/test_trend.cpython-311.pyc b/py-client/tests/__pycache__/test_trend.cpython-311.pyc index ea67ad1..87fb86f 100644 Binary files a/py-client/tests/__pycache__/test_trend.cpython-311.pyc and b/py-client/tests/__pycache__/test_trend.cpython-311.pyc differ diff --git a/py-client/tests/test_market.py b/py-client/tests/test_market.py new file mode 100644 index 0000000..244f2de --- /dev/null +++ b/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() diff --git a/py-client/tests/test_trend.py b/py-client/tests/test_trend.py index 91e54f6..fb077eb 100644 --- a/py-client/tests/test_trend.py +++ b/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( - client=client, - account_cfg=SimpleNamespace(min_cash_ratio=0.1), - global_cfg=SimpleNamespace(api_host="http://example"), - orders=SimpleNamespace(refresh=lambda _client: None), - state=SimpleNamespace( - codes=["A"], - unresolved_codes=[], - reconcile=lambda *_args: None, - ), - ) - 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") as manage_mock, - ): - RunOnce(runtime, []) + with ThreadPoolExecutor(max_workers=2) as executor: + runtime = SimpleNamespace( + client=client, + account_cfg=SimpleNamespace(min_cash_ratio=0.1), + global_cfg=SimpleNamespace(api_host="http://example"), + orders=SimpleNamespace(refresh=lambda _client: None, data=[]), + state=SimpleNamespace( + codes=["A"], + reconcile=lambda *_args: 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") as manage_mock, + ): + 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 ( - 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]) + with ThreadPoolExecutor(max_workers=2) as executor: + 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), + 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() - self.assertTrue(state.has_unresolved_order("A")) + open_mock.assert_called_once() + self.assertEqual(state.codes, []) if __name__ == "__main__":