diff --git a/py-client/libs/collector.py b/py-client/libs/collector.py index ec01858..8e7b3c8 100644 --- a/py-client/libs/collector.py +++ b/py-client/libs/collector.py @@ -3,7 +3,7 @@ from datetime import date, datetime from enum import Enum import json from typing import Any -from strategy.trend.boot import get_collector_snapshot +from libs.snapshot import get_collector_snapshot import httpx @@ -34,7 +34,7 @@ def _json_value(value: Any) -> Any: return str(value) -def collector_push(account_id: str, assets: Any, positions: Any) -> None: +def collector_push(account_id: str, assets: Any, positions: Any, deals: Any = None) -> None: """[暂停] 数据收集提交,太耗时,超过200毫秒.""" try: payload = _json_value( @@ -42,6 +42,7 @@ def collector_push(account_id: str, assets: Any, positions: Any) -> None: "account_id": account_id, "assets": assets, "positions": positions, + **({"deals": deals} if deals is not None else {}), } ) httpx.post(COLLECTOR_URL, json=payload, timeout=3.0) diff --git a/py-client/libs/snapshot.py b/py-client/libs/snapshot.py new file mode 100644 index 0000000..5e9a08b --- /dev/null +++ b/py-client/libs/snapshot.py @@ -0,0 +1,29 @@ +"""策略账户快照,供采集线程读取。""" + +from copy import deepcopy +from threading import Lock + +from sdk import Assets, DealItem, PositionItem + + +_collector_lock = Lock() +_collector_snapshot: tuple[str, Assets, list[PositionItem], list[DealItem]] | None = None + + +def cache_portfolio( + account_id: str, + assets: Assets, + positions: list[PositionItem], + deals: list[DealItem], +) -> None: + """整体替换最新快照,策略线程不执行序列化和网络上报。""" + global _collector_snapshot + with _collector_lock: + _collector_snapshot = (account_id, assets, positions, deals) + + +def get_collector_snapshot() -> tuple[str, Assets, list[PositionItem], list[DealItem]] | None: + """供 scheduler 读取;复制在锁外执行,不阻塞下一轮缓存更新。""" + with _collector_lock: + snapshot = _collector_snapshot + return deepcopy(snapshot) diff --git a/py-client/strategy/trend/boot.py b/py-client/strategy/trend/boot.py index 0048671..3074d99 100644 --- a/py-client/strategy/trend/boot.py +++ b/py-client/strategy/trend/boot.py @@ -6,16 +6,15 @@ import time import logging as log from concurrent.futures import Future, ThreadPoolExecutor -from copy import deepcopy from datetime import datetime -from threading import Lock import config from libs.calc import trading_time from libs.market import market_allow_open from libs.overview import Overview from libs.signal import init_signals, SignalItem -from sdk import Assets, Client, PositionItem +from sdk import Client +from libs.snapshot import cache_portfolio from libs.grid_take_profit import GridTrailingTracker from libs.order import OrderBook from libs.watch import DipWatch @@ -23,23 +22,6 @@ from libs.runtime import Runtime from .open import open_signal from .positions import manage_positions -_collector_lock = Lock() -_collector_snapshot: tuple[str, Assets, list[PositionItem]] | None = None - - -def _cache_portfolio(account_id: str, assets: Assets, positions: list[PositionItem]) -> None: - """整体替换最新快照,策略线程不执行序列化和网络上报。""" - global _collector_snapshot - with _collector_lock: - _collector_snapshot = (account_id, assets, positions) - - -def get_collector_snapshot() -> tuple[str, Assets, list[PositionItem]] | None: - """供 scheduler 读取;复制在锁外执行,不阻塞下一轮缓存更新。""" - with _collector_lock: - snapshot = _collector_snapshot - return deepcopy(snapshot) - def StartTrend() -> None: """初始化趋势策略,并以 30 秒间隔持续执行。""" @@ -53,7 +35,7 @@ def StartTrend() -> None: portfolio = client.portfolio() assets = portfolio.assets positions = list(portfolio.positions.values()) - _cache_portfolio(config.account_config.account_id, assets, positions) + cache_portfolio(config.account_config.account_id, assets, positions, client.deals()) order_book = OrderBook("trend") order_book.refresh(client, portfolio.orders) @@ -133,7 +115,7 @@ def RunOnce(run: Runtime, signals: list[SignalItem]) -> None: assets = portfolio.assets position_codes = list(portfolio.positions) positions = list(portfolio.positions.values()) - _cache_portfolio(run.account_cfg.account_id, assets, positions) + cache_portfolio(run.account_cfg.account_id, assets, positions, run.client.deals()) run.orders.refresh(run.client, portfolio.orders) except Exception: log.exception("[Portfolio] 刷新账户快照失败")