diff --git a/py-client/strategy/zt/order.py b/py-client/libs/order.py similarity index 60% rename from py-client/strategy/zt/order.py rename to py-client/libs/order.py index 1c39e49..8a66d41 100644 --- a/py-client/strategy/zt/order.py +++ b/py-client/libs/order.py @@ -1,4 +1,4 @@ -"""做 T 策略委托簿,对应 Go 客户端的 ``logic/order.go``。""" +"""策略共用委托簿。""" from __future__ import annotations @@ -10,7 +10,7 @@ from cachelib import SimpleCache import logging import httpx -from sdk import Client,ORDER_SIDE_BY_OFFSET,APIError,OrderItem +from sdk import Client, ORDER_SIDE_BY_OFFSET, APIError, OrderItem # 表示委托仍在处理、可能继续成交的 QMT 状态。 BUSY_STATUSES = {"48", "49", "50", "51", "52", "55"} @@ -22,6 +22,7 @@ CANCELABLE_STATUSES = {"49", "50", "51", "52"} @dataclass(slots=True) class PlaceOrderRequest: """``OrderBook.place`` 提交委托所需的全部参数。""" + op: int code: str volume: int @@ -33,18 +34,22 @@ class PlaceOrderRequest: class OrderBook: """线程安全的活动委托缓存。""" - def __init__(self, lock_timeout_sec: int = 180, cancel_timeout_sec: float = 10) -> None: + def __init__( + self, order_prefix: str, lock_timeout_sec: int = 180, cancel_timeout_sec: float = 10 + ) -> None: + self.order_prefix = order_prefix self.lock_timeout_sec = max(1, lock_timeout_sec) self.cancel_timeout_sec = timedelta(seconds=cancel_timeout_sec) self.data: list[OrderItem] = [] self.busy_keys: set[str] = set() - self.busy_cache = SimpleCache(threshold=10_000, default_timeout=self.lock_timeout_sec) + self.busy_cache = SimpleCache( + threshold=10_000, default_timeout=self.lock_timeout_sec + ) self.mutex = Lock() - @staticmethod - def new_order_id(side:str) -> str: - """生成 ``zt-xxxxxxxx`` 格式的本地订单号。""" - return f"zt-{side}-{secrets.token_hex(10)}" + def new_order_id(self, side: str) -> str: + """生成带策略前缀的本地订单号。""" + return f"{self.order_prefix}-{side}-{secrets.token_hex(10)}" def busy(self, code: str, side: str) -> bool: """判断证券是否存在仍在处理中的同方向委托。""" @@ -72,16 +77,23 @@ class OrderBook: # 清理过期的 if ( item.created_at is not None - and item.local_order_id.startswith("zt-") + and item.local_order_id.startswith(f"{self.order_prefix}-") and item.status in CANCELABLE_STATUSES and current - item.created_at > self.cancel_timeout_sec ): try: client.cancel_by_id(item.id) canceled += 1 - logging.info("[Order] 超时撤单,代码=%s,方向=%s,柜台订单=%s", item.code, item.side, item.id) + logging.info( + "[Order] 超时撤单,代码=%s,方向=%s,柜台订单=%s", + item.code, + item.side, + item.id, + ) except Exception: - logging.exception("[Order] 撤单失败,保留在途状态,订单=%s", item.id) + logging.exception( + "[Order] 撤单失败,保留在途状态,订单=%s", item.id + ) # 缓存本次有效订单 data.append(item) @@ -89,19 +101,30 @@ class OrderBook: with self.mutex: self.data = data self.busy_keys = busy_keys - logging.info("[Order] 刷新完成,跟踪=%d,处理中=%d,撤销=%d", len(data), len(busy_keys), canceled) + logging.info( + "[Order] 刷新完成,跟踪=%d,处理中=%d,撤销=%d", + len(data), + len(busy_keys), + canceled, + ) def place(self, client: Client, request: PlaceOrderRequest) -> bool: """按最新价提交委托,并立即写入本地方向锁。""" side = ORDER_SIDE_BY_OFFSET.get(str(request.op), "") if not side: - logging.warning("[Order] 下单失败,代码=%s,原因=未知买卖方向(%s)", request.code, request.op) + logging.warning( + "[Order] 下单失败,代码=%s,原因=未知买卖方向(%s)", + request.code, + request.op, + ) return False key = self._busy_key(side, request.code) with self.mutex: if key in self.busy_keys or self.busy_cache.has(key): - logging.info("[Order] 跳过重复下单,代码=%s,方向=%s", request.code, side) + logging.info( + "[Order] 跳过重复下单,代码=%s,方向=%s", request.code, side + ) return False self.busy_cache.set(key, True, timeout=self.lock_timeout_sec) @@ -114,12 +137,29 @@ class OrderBook: order_id=request.order_id, ) except APIError as exc: - logging.exception("[Order] 下单失败,代码=%s,本地订单=%s,HTTP状态=%d,错误=%s", request.code, request.order_id, exc.status_code, exc.message or str(exc)) + logging.exception( + "[Order] 下单失败,代码=%s,本地订单=%s,HTTP状态=%d,错误=%s", + request.code, + request.order_id, + exc.status_code, + exc.message or str(exc), + ) return False except (httpx.RequestError, ValueError): # 响应异常不能证明柜台未受理,保留缓存防重,不自动重试。 - logging.exception("[Order] 下单请求或响应异常,代码=%s,本地订单=%s", request.code, request.order_id) + logging.exception( + "[Order] 下单请求或响应异常,代码=%s,本地订单=%s", + request.code, + request.order_id, + ) return False - - logging.info("[Order] 下单已受理,代码=%s,方向=%s,数量=%d,本地订单=%s,返回=%s", request.code, side, request.volume, request.order_id, result) + + logging.info( + "[Order] 下单已受理,代码=%s,方向=%s,数量=%d,本地订单=%s,返回=%s", + request.code, + side, + request.volume, + request.order_id, + result, + ) return True diff --git a/py-client/libs/overview.py b/py-client/libs/overview.py new file mode 100644 index 0000000..e56b7b2 --- /dev/null +++ b/py-client/libs/overview.py @@ -0,0 +1,39 @@ +"""账户启动概览日志。""" + +import logging as log + +import config + + +def Overview(assets, positions, account_cfg=None) -> None: + """记录策略启动时的账户、资金和持仓概览。""" + account_cfg = account_cfg or config.account_config + + if account_cfg is not None: + log.info( + "[启动] 账户=%s,主机=%s,单笔金额=%.2f", + account_cfg.account_id, + account_cfg.host_key, + account_cfg.buy_value, + ) + + if assets is not None: + log.info("[启动] 总资产=%.2f,可用资金=%.2f", assets.total, assets.available) + else: + log.warning("[启动] 获取资金概览失败") + + for position in positions: + if position.volume <= 0: + continue + log.info( + "[启动] %s %s %s,持仓=%d,可用=%d,成本=%.2f(%.2f),现价=%.2f,盈亏=%.2f%%", + position.trade_id, + position.stock_code, + position.stock_name, + position.volume, + position.can_use_volume, + position.open_price, + position.open_cost, + position.last_price, + position.profit_rate * 100, + ) diff --git a/py-client/libs/runtime.py b/py-client/libs/runtime.py new file mode 100644 index 0000000..bb8a607 --- /dev/null +++ b/py-client/libs/runtime.py @@ -0,0 +1,26 @@ +"""策略单次运行所需的公共上下文对象。""" + +from __future__ import annotations + +from concurrent.futures import ThreadPoolExecutor +from dataclasses import dataclass + +from config import AccountConfig, GlobalConfig +from sdk import Client +from libs.grid_take_profit import GridTrailingTracker +from libs.order import OrderBook +from libs.watch import DipWatch + + +@dataclass(slots=True) +class Runtime: + """集中保存策略运行期间共享的客户端、配置和内存组件。""" + + client: Client + global_cfg: GlobalConfig + account_cfg: AccountConfig + orders: OrderBook + open_watch: DipWatch + add_watch: DipWatch + profit_tracker: GridTrailingTracker + executor: ThreadPoolExecutor | None = None diff --git a/py-client/strategy/trend/watch.py b/py-client/libs/watch.py similarity index 100% rename from py-client/strategy/trend/watch.py rename to py-client/libs/watch.py diff --git a/py-client/strategy/trend/boot.py b/py-client/strategy/trend/boot.py index 55c8074..dec5ab4 100644 --- a/py-client/strategy/trend/boot.py +++ b/py-client/strategy/trend/boot.py @@ -13,36 +13,18 @@ from datetime import datetime 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 libs.collector import collector_push from sdk import Client from libs.grid_take_profit import GridTrailingTracker -from .order import OrderBook -from .watch import DipWatch -from .runtime import Runtime +from libs.order import OrderBook +from libs.watch import DipWatch +from libs.runtime import Runtime from .open import open_signal from .positions import manage_positions -def Overview(assets, positions, account_cfg=None) -> None: - """记录策略启动时的账户、资金和持仓概览。""" - account_cfg = account_cfg or config.account_config - - if account_cfg is not None: - log.info("[启动] 账户=%s,主机=%s,单笔金额=%.2f", account_cfg.account_id, account_cfg.host_key, account_cfg.buy_value) - - if assets is not None: - log.info("[启动] 总资产=%.2f,可用资金=%.2f", assets.total, assets.available) - else: - log.warning("[启动] 获取资金概览失败") - - for position in positions: - if position.volume <= 0: - continue - log.info("[启动] %s %s %s,持仓=%d,可用=%d,成本=%.2f(%.2f),现价=%.2f,盈亏=%.2f%%",position.trade_id, position.stock_code, position.stock_name, position.volume, position.can_use_volume, position.open_price,position.open_cost, position.last_price, position.profit_rate * 100) - - - def StartTrend() -> None: """初始化趋势策略,并以 30 秒间隔持续执行。""" client = Client( @@ -55,7 +37,7 @@ def StartTrend() -> None: portfolio = client.portfolio() assets = portfolio.assets positions = list(portfolio.positions.values()) - order_book = OrderBook() + order_book = OrderBook("trend") order_book.refresh(client, portfolio.orders) # 获取本策略的信号开仓数据 @@ -63,7 +45,12 @@ def StartTrend() -> None: config.global_config, config.account_config.signal_allow, ) - log.info("[启动] 趋势策略已启动,账户=%s,信号=%d,持仓=%d", config.account_config.account_id, len(signals), len(positions)) + log.info( + "[启动] 趋势策略已启动,账户=%s,信号=%d,持仓=%d", + config.account_config.account_id, + len(signals), + len(positions), + ) executor = ThreadPoolExecutor(max_workers=3, thread_name_prefix="trend") run = Runtime( client=client, @@ -101,7 +88,9 @@ def StartTrend() -> None: try: RunOnce(run, signals) except Exception as e: - log.error(f"[Trend] 本 tick 执行失败,下一 tick 继续: {e}", exc_info=True) + log.error( + f"[Trend] 本 tick 执行失败,下一 tick 继续: {e}", exc_info=True + ) finally: try: if executor is not None: @@ -110,13 +99,15 @@ def StartTrend() -> None: client.close() -def RunOnce(run: Runtime, signals:list[SignalItem]) -> None: +def RunOnce(run: Runtime, signals: list[SignalItem]) -> None: """按固定步骤执行一轮趋势策略, ``RunOnce``。""" if not trading_time(datetime.now()): return - print("=" * 40 + f" Ticker {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} " +"=" * 40) - + print( + "=" * 40 + f" Ticker {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} " + "=" * 40 + ) + started_at = time.monotonic() # 1. 一次获取资产、持仓和订单,并清理过期订单。 @@ -131,21 +122,27 @@ def RunOnce(run: Runtime, signals:list[SignalItem]) -> None: return futures: list[tuple[str, Future]] = [ - ( - "数据提交", - run.executor.submit( - collector_push, - run.account_cfg.account_id, - assets, - positions, - ), - ) - ] + ( + "数据提交", + run.executor.submit( + collector_push, + run.account_cfg.account_id, + assets, + positions, + ), + ) + ] # 2. 验证可用资金;低于资金安全线时禁止开新仓。 - allow_open_by_cash = assets.available >= assets.total * run.account_cfg.min_cash_ratio + allow_open_by_cash = ( + assets.available >= assets.total * run.account_cfg.min_cash_ratio + ) if not allow_open_by_cash: - log.info("[Status] 禁止开仓:可用资金不足,可用=%.2f,总资产=%.2f", assets.available, assets.total) + log.info( + "[Status] 禁止开仓:可用资金不足,可用=%.2f,总资产=%.2f", + assets.available, + assets.total, + ) # 3. 获取大盘状态,只有大盘信号允许时才执行开仓。 market_ok = market_allow_open() @@ -169,20 +166,37 @@ def RunOnce(run: Runtime, signals:list[SignalItem]) -> None: log.exception("[行情] 获取行情失败,代码数量=%d", len(all_codes)) return - log.info("[RunOnce] 本轮就绪,持仓=%d,候选=%d,大盘允许=%s,资金允许=%s", len(positions), len(allow_open), market_ok, allow_open_by_cash) - + log.info( + "[RunOnce] 本轮就绪,持仓=%d,候选=%d,大盘允许=%s,资金允许=%s", + len(positions), + len(allow_open), + market_ok, + allow_open_by_cash, + ) + # 启动线程,开始计算 # 7. 持仓计算。 - futures.append(("持仓计算",run.executor.submit(manage_positions,run,ticks,positions,market_ok,assets.available))) + futures.append( + ( + "持仓计算", + run.executor.submit( + manage_positions, run, ticks, positions, market_ok, assets.available + ), + ) + ) # 8. 开仓计算:必须同时存在有效信号且大盘允许开仓。 if allow_open and market_ok and allow_open_by_cash: - futures.append(("开仓计算", run.executor.submit(open_signal, run, ticks, allow_open))) + futures.append( + ("开仓计算", run.executor.submit(open_signal, run, ticks, allow_open)) + ) # 9. 开始执行 for name, future in futures: _wait_worker(name, future) - log.info("[RunOnce] 本轮完成,耗时=%d毫秒", int((time.monotonic() - started_at) * 1000)) + log.info( + "[RunOnce] 本轮完成,耗时=%d毫秒", int((time.monotonic() - started_at) * 1000) + ) def _wait_worker(name: str, future: Future) -> None: diff --git a/py-client/strategy/trend/open.py b/py-client/strategy/trend/open.py index af89773..89da94f 100644 --- a/py-client/strategy/trend/open.py +++ b/py-client/strategy/trend/open.py @@ -6,16 +6,18 @@ from datetime import datetime from libs import calc_buy_volume from sdk import OP_BUY -from .runtime import Runtime -from .order import PlaceOrderRequest +from libs.runtime import Runtime +from libs.order import PlaceOrderRequest import logging as log -def open_signal(run:Runtime, ticks, open_signals) -> None: +def open_signal(run: Runtime, ticks, open_signals) -> None: """逐个验证开仓信号并提交买入委托。""" for item in open_signals: if item.code in run.account_cfg.excluded_codes: - log.info("[Open] %s 信号=%s,跳过:已配置为排除股票", item.code, item.signal_key) + log.info( + "[Open] %s 信号=%s,跳过:已配置为排除股票", item.code, item.signal_key + ) continue # 1. 验证信号配置允许开仓的时间区间。 signal_config = run.global_cfg.signals.get(item.signal_key) @@ -36,8 +38,10 @@ def open_signal(run:Runtime, ticks, open_signals) -> None: continue # 2. 检查该证券是否已有买入委托锁,防止重复下单。 - if run.orders.busy(item.code,"BUY"): - log.info("[Open] %s 信号=%s,跳过:买入委托处理中", item.code, item.signal_key) + if run.orders.busy(item.code, "BUY"): + log.info( + "[Open] %s 信号=%s,跳过:买入委托处理中", item.code, item.signal_key + ) continue # 3. 验证行情和最新价格是否有效。 @@ -54,14 +58,34 @@ def open_signal(run:Runtime, ticks, open_signals) -> None: continue # 当前价高于昨收价可开仓 - if signal_config.gt_last_price_is_open and item.last_close>0 and price>item.last_close: + if ( + signal_config.gt_last_price_is_open + and item.last_close > 0 + and price > item.last_close + ): try: do_open(run, item.code, volume, item.signal_key, price) - log.info("[Open] %s 信号=%s,买入=%d股,原因=现价高于昨收", item.code, item.signal_key, volume) + log.info( + "[Open] %s 信号=%s,买入=%d股,原因=现价高于昨收", + item.code, + item.signal_key, + volume, + ) except RuntimeError as exc: - log.info("[Open] %s 信号=%s,买入=%d股失败:%s", item.code, item.signal_key, volume, exc) + log.info( + "[Open] %s 信号=%s,买入=%d股失败:%s", + item.code, + item.signal_key, + volume, + exc, + ) except Exception: - log.exception("[Open] %s 信号=%s,买入=%d股异常", item.code, item.signal_key, volume) + log.exception( + "[Open] %s 信号=%s,买入=%d股异常", + item.code, + item.signal_key, + volume, + ) continue # 4. 等待价格从观察低点反弹,防止直接接下跌中的“飞刀”。 @@ -70,14 +94,29 @@ def open_signal(run:Runtime, ticks, open_signals) -> None: try: do_open(run, item.code, volume, item.signal_key, price) - log.info("[Open] %s 信号=%s,买入=%d股,原因=反弹已确认", item.code, item.signal_key, volume) + log.info( + "[Open] %s 信号=%s,买入=%d股,原因=反弹已确认", + item.code, + item.signal_key, + volume, + ) except RuntimeError as exc: - log.warning("[Open] %s 信号=%s,买入=%d股失败:%s", item.code, item.signal_key, volume, exc) + log.warning( + "[Open] %s 信号=%s,买入=%d股失败:%s", + item.code, + item.signal_key, + volume, + exc, + ) except Exception: - log.exception("[Open] %s 信号=%s,买入=%d股异常", item.code, item.signal_key, volume) - + log.exception( + "[Open] %s 信号=%s,买入=%d股异常", item.code, item.signal_key, volume + ) -def do_open(run: Runtime, code: str, volume: int, signal_key: str, price: float) -> None: + +def do_open( + run: Runtime, code: str, volume: int, signal_key: str, price: float +) -> None: """生成本地订单号并按最新价提交开仓委托。""" order_id = run.orders.new_order_id("BUY") request = PlaceOrderRequest( @@ -89,9 +128,7 @@ def do_open(run: Runtime, code: str, volume: int, signal_key: str, price: float) kind="base", ) - #run.state.new_order(PendingOrder(order_id, code, "base", volume)) - - if not run.orders.place(run.client,request): + if not run.orders.place(run.client, request): raise RuntimeError("订单提交失败") run.open_watch.forget(code) diff --git a/py-client/strategy/trend/order.py b/py-client/strategy/trend/order.py deleted file mode 100644 index ef0065d..0000000 --- a/py-client/strategy/trend/order.py +++ /dev/null @@ -1,122 +0,0 @@ -"""趋势策略委托簿,对应 Go 客户端的 ``logic/order.go``。""" - -from __future__ import annotations - -import secrets -from dataclasses import dataclass -from datetime import datetime, timedelta -from threading import Lock -from cachelib import SimpleCache -import logging - -import httpx -from sdk import Client,ORDER_SIDE_BY_OFFSET,APIError,OrderItem - -# 表示委托仍在处理、可能继续成交的 QMT 状态。 -BUSY_STATUSES = {"48", "49", "50", "51", "52", "55"} -COMPLETED_STATUSES = {"56"} -TRACKED_STATUSES = BUSY_STATUSES | COMPLETED_STATUSES -CANCELABLE_STATUSES = {"49", "50", "51", "52"} - - -@dataclass(slots=True) -class PlaceOrderRequest: - """``OrderBook.place`` 提交委托所需的全部参数。""" - op: int - code: str - volume: int - order_id: str - strategy_name: str - kind: str = "" - - -class OrderBook: - """线程安全的活动委托缓存。""" - - def __init__(self, lock_timeout_sec: int = 180, cancel_timeout_sec: float = 10) -> None: - self.lock_timeout_sec = max(1, lock_timeout_sec) - self.cancel_timeout_sec = timedelta(seconds=cancel_timeout_sec) - self.data: list[OrderItem] = [] - self.busy_keys: set[str] = set() - self.busy_cache = SimpleCache(threshold=10_000, default_timeout=self.lock_timeout_sec) - self.mutex = Lock() - - @staticmethod - def new_order_id(side:str) -> str: - """生成 ``trend-xxxxxxxx`` 格式的本地订单号。""" - return f"trend-{side}-{secrets.token_hex(10)}" - - def busy(self, code: str, side: str) -> bool: - """判断证券是否存在仍在处理中的同方向委托。""" - with self.mutex: - key = self._busy_key(side, code) - return key in self.busy_keys or self.busy_cache.has(key) - - @staticmethod - def _busy_key(side: str, code: str) -> str: - return f"{side}-{code}" - - def refresh(self, client: Client, orders: list[OrderItem]) -> None: - """用账户快照刷新委托,并撤销超时的活动委托。""" - current = datetime.now() - data: list[OrderItem] = [] - busy_keys: set[str] = set() - canceled = 0 - - for item in orders: - # 不处理状态不对的 - if item.status not in TRACKED_STATUSES: - continue - if item.status in BUSY_STATUSES: - busy_keys.add(self._busy_key(item.side, item.code)) - # 清理过期的 - if ( - item.created_at is not None - and item.status in CANCELABLE_STATUSES - and current - item.created_at > self.cancel_timeout_sec - ): - client.cancel_by_id(item.id) - canceled += 1 - logging.info("[Order] 超时撤单,代码=%s,方向=%s,柜台订单=%s", item.code, item.side, item.id) - continue - - # 缓存本次有效订单 - data.append(item) - - with self.mutex: - self.data = data - self.busy_keys = busy_keys - logging.info("[Order] 刷新完成,跟踪=%d,处理中=%d,撤销=%d", len(data), len(busy_keys), canceled) - - def place(self, client: Client, request: PlaceOrderRequest) -> bool: - """按最新价提交委托,并立即写入本地方向锁。""" - side = ORDER_SIDE_BY_OFFSET.get(str(request.op), "") - if not side: - logging.warning("[Order] 下单失败,代码=%s,原因=未知买卖方向(%s)", request.code, request.op) - return False - - key = self._busy_key(side, request.code) - with self.mutex: - if key in self.busy_keys or self.busy_cache.has(key): - logging.info("[Order] 跳过重复下单,代码=%s,方向=%s", request.code, side) - return False - self.busy_cache.set(key, True, timeout=self.lock_timeout_sec) - - try: - result = client.passorder( - op_type=request.op, - stock_code=request.code, - volume=request.volume, - strategy_name=request.strategy_name, - order_id=request.order_id, - ) - except APIError as exc: - logging.exception("[Order] 下单失败,代码=%s,本地订单=%s,HTTP状态=%d,错误=%s", request.code, request.order_id, exc.status_code, exc.message or str(exc)) - return False - except (httpx.RequestError, ValueError): - # 响应异常不能证明柜台未受理,保留缓存防重,不自动重试。 - logging.exception("[Order] 下单请求或响应异常,代码=%s,本地订单=%s", request.code, request.order_id) - return False - - logging.info("[Order] 下单已受理,代码=%s,方向=%s,数量=%d,本地订单=%s,返回=%s", request.code, side, request.volume, request.order_id, result) - return True diff --git a/py-client/strategy/trend/positions.py b/py-client/strategy/trend/positions.py index d5e4cdd..16dea12 100644 --- a/py-client/strategy/trend/positions.py +++ b/py-client/strategy/trend/positions.py @@ -8,8 +8,8 @@ from libs.calc import calc_buy_volume, calculate_min_profit_rate from libs.grid_take_profit import GridState from sdk import OP_BUY, OP_SELL, PositionItem, Tick -from .order import PlaceOrderRequest -from .runtime import Runtime +from libs.order import PlaceOrderRequest +from libs.runtime import Runtime import logging as log LOSS_TIERS = [-50.0] @@ -34,11 +34,15 @@ def manage_positions( # 遍历处理每个持仓 for position in positions: try: - available = max(0,0,available) + available = max(0, 0, available) code = position.stock_code tick = ticks.get(code) if code in runtime.account_cfg.excluded_codes: - log.info("[Position - ] 代码=%s,名称=%s,止盈=跳过,补仓=跳过,原因=已配置为排除股票", code, position.stock_name) + log.info( + "[Position - ] 代码=%s,名称=%s,止盈=跳过,补仓=跳过,原因=已配置为排除股票", + code, + position.stock_name, + ) continue if ( not code @@ -47,7 +51,11 @@ def manage_positions( or tick is None or tick.last_price <= 0 ): - log.warning("[Position - ] 代码=%s,名称=%s,止盈=跳过,补仓=跳过,原因=持仓或行情数据无效", code or "未知", position.stock_name) + log.warning( + "[Position - ] 代码=%s,名称=%s,止盈=跳过,补仓=跳过,原因=持仓或行情数据无效", + code or "未知", + position.stock_name, + ) continue pnl_rate = round( @@ -77,18 +85,29 @@ def manage_positions( elif runtime.account_cfg.enable_loss_add_position: loss_add_action = "大盘信号不允许" - if pnl_rate>=0: + if pnl_rate >= 0: log.info( "[Position ↑ ] 代码=%s,名称=%s,盈亏=%.2f%%,止盈=%s,补仓=%s", - code, position.stock_name, pnl_rate, profit_action, loss_add_action, + code, + position.stock_name, + pnl_rate, + profit_action, + loss_add_action, ) else: log.info( "[Position ↓ ] 代码=%s,名称=%s,盈亏=%.2f%%,止盈=%s,补仓=%s", - code, position.stock_name, pnl_rate, profit_action, loss_add_action, + code, + position.stock_name, + pnl_rate, + profit_action, + loss_add_action, ) except Exception: - log.exception("[Position] 持仓处理异常,代码=%s,继续处理后续持仓", position.stock_code) + log.exception( + "[Position] 持仓处理异常,代码=%s,继续处理后续持仓", + position.stock_code, + ) def handle_profit( @@ -132,8 +151,7 @@ def handle_profit( ) if not runtime.orders.place(runtime.client, request): return TradeDecision(False, "止盈委托失败") - - + return TradeDecision(True, f"卖出 {volume} 股,订单={order_id}") @@ -142,10 +160,12 @@ def handle_loss( position: PositionItem, tick: Tick, pnl_rate: float, - available: float + available: float, ) -> TradeDecision: """按亏损档位、反弹确认和本轮剩余预算提交补仓。""" - add_num = get_add_num(hands=int(position.volume/100),market_value=position.market_value) + add_num = get_add_num( + hands=int(position.volume / 100), market_value=position.market_value + ) if add_num >= len(LOSS_TIERS) or add_num < 0: return TradeDecision(False, f"补仓次数无效:{add_num}") if pnl_rate > LOSS_TIERS[add_num]: @@ -171,10 +191,10 @@ def handle_loss( strategy_name=runtime.account_cfg.strategy, kind="add", ) - + if not runtime.orders.place(runtime.client, request): return TradeDecision(False, "补仓订单委托失败") - + runtime.add_watch.forget(position.stock_code) return TradeDecision(True, f"买入 {volume} 股,订单={order_id}", amount) @@ -182,9 +202,10 @@ def handle_loss( def _position_key(runtime: Runtime, code: str) -> str: return f"{runtime.account_cfg.account_id}:{code}" -def get_add_num(hands:int,market_value:float) -> int: - if market_value>10000: + +def get_add_num(hands: int, market_value: float) -> int: + if market_value > 10000: return -1 if hands < 2: return 0 - return -1 \ No newline at end of file + return -1 diff --git a/py-client/strategy/trend/runtime.py b/py-client/strategy/trend/runtime.py deleted file mode 100644 index f650c15..0000000 --- a/py-client/strategy/trend/runtime.py +++ /dev/null @@ -1,44 +0,0 @@ -"""趋势策略单次运行所需的上下文对象。""" - -from __future__ import annotations - -from concurrent.futures import ThreadPoolExecutor -from dataclasses import dataclass, field - -from config import AccountConfig, GlobalConfig -from sdk import Client -from libs.grid_take_profit import GridTrailingTracker - -from .order import OrderBook -from .watch import DipWatch - - -@dataclass(slots=True) -class Runtime: - """集中保存趋势策略运行期间共享的依赖和状态。 - - 将这些对象集中到一个 dataclass 后,开仓、持仓管理和单轮调度函数 - 只需接收一个 ``Runtime``,无需重复传递大量参数。 - - Attributes: - client: QMT HTTP 客户端,用于查询账户、行情和提交委托。 - global_cfg: 公共配置,包含 QMT、外部 API 和信号配置。 - account_cfg: 当前主机的账户及交易策略配置。 - state: 策略持仓状态的本地持久化存储。 - orders: 当前活动委托和证券方向锁。 - open_watch: 新开仓使用的价格反弹观察器。 - add_watch: 亏损补仓使用的价格反弹观察器。 - profit_tracker: 跨轮保存的账户持仓最高盈利网格跟踪器。 - """ - - # 外部服务与账户配置。 - client: Client - global_cfg: GlobalConfig - account_cfg: AccountConfig - - # 策略运行过程中共享的状态组件。 - orders: OrderBook - open_watch: DipWatch - add_watch: DipWatch - profit_tracker: GridTrailingTracker - executor: ThreadPoolExecutor diff --git a/py-client/strategy/zt/boot.py b/py-client/strategy/zt/boot.py index bedda59..fff2674 100644 --- a/py-client/strategy/zt/boot.py +++ b/py-client/strategy/zt/boot.py @@ -16,9 +16,9 @@ from libs.signal import init_signals from libs.collector import collector_push from libs.grid_take_profit import GridTrailingTracker from sdk import Client -from .order import OrderBook -from .watch import DipWatch -from .runtime import Runtime +from libs.order import OrderBook +from libs.watch import DipWatch +from libs.runtime import Runtime from .state import TState, SOLD from .open import open_signal from .positions import manage_positions @@ -26,20 +26,39 @@ from .positions import manage_positions def StartZT() -> None: """初始化做 T 策略,并以 30 秒间隔持续执行。""" - with Client(config.global_config.qmt_base_url, config.global_config.qmt_token, - config.HTTP_TIMEOUT) as client: - state = TState.for_strategy(config.global_config.qmt_data_dir, "zt", config.account_config.account_id) - run = Runtime(client, config.global_config, config.account_config, state, - OrderBook(), DipWatch(), DipWatch(), - GridTrailingTracker(config.account_config.grid_step_pct)) - log.info("[ZT 启动] 账户=%s,底仓信号=dcm,状态文件=%s", run.account_cfg.account_id, state.path) + with Client( + config.global_config.qmt_base_url, + config.global_config.qmt_token, + config.HTTP_TIMEOUT, + ) as client: + state = TState.for_strategy( + config.global_config.qmt_data_dir, "zt", config.account_config.account_id + ) + run = Runtime( + client=client, + global_cfg=config.global_config, + account_cfg=config.account_config, + orders=OrderBook("zt"), + open_watch=DipWatch(), + add_watch=DipWatch(), + profit_tracker=GridTrailingTracker(config.account_config.grid_step_pct), + ) + log.info( + "[ZT 启动] 账户=%s,底仓信号=dcm,状态文件=%s", + run.account_cfg.account_id, + state.path, + ) while True: now = datetime.now() if now.time() >= clock_time(15): # 收盘前最后一次只读对账,不发新单;未完成买回继续持久保存。 try: portfolio = client.portfolio() - state.reconcile(list(portfolio.positions.values()), portfolio.orders, now.date().isoformat()) + state.reconcile( + list(portfolio.positions.values()), + portfolio.orders, + now.date().isoformat(), + ) except Exception: log.exception("[ZT] 收盘对账失败,保留本地待确认记录") for item in state.items.values(): @@ -48,14 +67,14 @@ def StartZT() -> None: return # 单轮失败不能杀死唯一的交易定时线程。 try: - RunOnce(run) + RunOnce(run, state) except Exception: log.exception("[ZT] 本 tick 执行失败,下一个 tick 继续") # 计算距离下一个目标时间点(0秒或30秒)的等待时间。 time.sleep(30 - datetime.now().second % 30) -def RunOnce(run: Runtime) -> None: +def RunOnce(run: Runtime, state: TState) -> None: """账户快照 → 成交对账 → 做 T 管理 → dcm 建仓,共用一份资金预算。""" now = datetime.now() if not trading_time(now) or now.time() >= clock_time(15): @@ -68,7 +87,7 @@ def RunOnce(run: Runtime) -> None: positions = list(portfolio.positions.values()) run.orders.refresh(run.client, portfolio.orders) # 对账使用完整原始订单列表,不能丢弃撤单和废单的部分成交。 - run.state.reconcile(positions, portfolio.orders, today) + state.reconcile(positions, portfolio.orders, today) # 2. 获取本策略的信号开仓数据;信号失败不阻断已有做 T 买回。 try: @@ -76,13 +95,23 @@ def RunOnce(run: Runtime) -> None: except Exception: log.exception("[ZT] 获取 dcm 信号失败,本轮只管理已有底仓") signals = [] - position_codes = {position.stock_code for position in positions if position.volume > 0} - candidates = [signal for signal in signals - if signal.signal_key == "dcm" and signal.code not in position_codes] + position_codes = { + position.stock_code for position in positions if position.volume > 0 + } + candidates = [ + signal + for signal in signals + if signal.signal_key == "dcm" and signal.code not in position_codes + ] # 3. 获取持仓和待开仓证券的实时行情 tick,零持仓的待买回证券也包含在内。 - codes = list(dict.fromkeys(list(position_codes) + list(run.state.items) - + [signal.code for signal in candidates])) + codes = list( + dict.fromkeys( + list(position_codes) + + list(state.items) + + [signal.code for signal in candidates] + ) + ) ticks = run.client.full_tick(codes) if codes else {} now = datetime.now() # 网络请求可能跨过尾盘边界,提交前重新判断。 if not trading_time(now) or now.time() >= clock_time(15): @@ -91,7 +120,7 @@ def RunOnce(run: Runtime) -> None: # 4. 先完成买回,避免开底仓抢占资金;交易逻辑串行,状态无需多线程写入。 available = max(0.0, portfolio.assets.available) # 未确认买单可能尚未反映在资金快照中,保守预留,宁可少买也不重复使用。 - for pending in run.state.pending.values(): + for pending in state.pending.values(): if pending.kind != "sell": tick = ticks.get(pending.code) if tick is None or tick.last_price <= 0: @@ -99,16 +128,22 @@ def RunOnce(run: Runtime) -> None: break available = max(0.0, available - pending.qty * tick.last_price * 1.01) force = now.time() >= clock_time(14, 50) - available = manage_positions(run, ticks, positions, available, today, force) + available = manage_positions(run, state, ticks, positions, available, today, force) # 5. 验证可用资金;低于资金安全线时禁止开新仓,尾盘只完成做 T 买回。 reserve = max(0.0, portfolio.assets.total * run.account_cfg.min_cash_ratio) # 未完成的卖出/买回可能继续占用资金,不再额外开底仓。 - outstanding = bool(run.state.pending) or any(item.phase == SOLD for item in run.state.items.values()) + outstanding = bool(state.pending) or any( + item.phase == SOLD for item in state.items.values() + ) if not force and not outstanding and market_allow_open() and available > reserve: - open_signal(run, ticks, candidates, available - reserve) + open_signal(run, state, ticks, candidates, available - reserve) # 6. 数据采集不与交易逻辑争用状态;采集函数自身隔离传输异常。 collector_push(run.account_cfg.account_id, portfolio.assets, positions) - log.info("[ZT] 本轮完成,底仓=%d,待确认=%d,耗时=%d毫秒", - len(run.state.items), len(run.state.pending), int((time.monotonic() - started_at) * 1000)) + log.info( + "[ZT] 本轮完成,底仓=%d,待确认=%d,耗时=%d毫秒", + len(state.items), + len(state.pending), + int((time.monotonic() - started_at) * 1000), + ) diff --git a/py-client/strategy/zt/open.py b/py-client/strategy/zt/open.py index 7b832a1..df685b7 100644 --- a/py-client/strategy/zt/open.py +++ b/py-client/strategy/zt/open.py @@ -8,12 +8,12 @@ import math from libs.calc import calc_buy_volume from sdk import OP_BUY -from .runtime import Runtime -from .order import PlaceOrderRequest -from .state import PendingOrder +from libs.runtime import Runtime +from libs.order import PlaceOrderRequest +from .state import PendingOrder, TState -def open_signal(run: Runtime, ticks, signals, available: float) -> float: +def open_signal(run: Runtime, state: TState, ticks, signals, available: float) -> float: """逐个验证开仓信号并提交买入委托,返回本轮剩余资金。""" for item in signals: try: @@ -22,20 +22,28 @@ def open_signal(run: Runtime, ticks, signals, available: float) -> float: break if item.signal_key != "dcm" or item.code in run.account_cfg.excluded_codes: continue - state = run.state.items.get(item.code) - if state is not None and state.base_qty > 0: + item_state = state.items.get(item.code) + if item_state is not None and item_state.base_qty > 0: continue # 1. 验证信号配置允许开仓的时间区间。 signal_config = run.global_cfg.signals.get("dcm") if signal_config is None or not check_timezone(signal_config.timezone): continue # 2. 检查该证券是否已有买入委托锁,防止重复下单。 - if run.state.busy(item.code) or run.orders.busy(item.code, "BUY") or run.orders.busy(item.code, "SELL"): + if ( + state.busy(item.code) + or run.orders.busy(item.code, "BUY") + or run.orders.busy(item.code, "SELL") + ): continue # 3. 验证行情和最新价格是否有效。 tick = ticks.get(item.code) price = tick.last_price if tick else 0.0 - if not math.isfinite(price) or price <= 0 or price > run.account_cfg.zt_max_price: + if ( + not math.isfinite(price) + or price <= 0 + or price > run.account_cfg.zt_max_price + ): continue # 4. 根据单笔买入金额计算整手开仓数量,预留少量价差和费用。 budget = min(run.account_cfg.buy_value, available) @@ -47,8 +55,18 @@ def open_signal(run: Runtime, ticks, signals, available: float) -> float: if not run.open_watch.triggered("ZT 建仓", item.code, price): continue order_id = run.orders.new_order_id("base") - request = PlaceOrderRequest(OP_BUY, item.code, volume, order_id, "zt", kind="base") - run.state.new_order(PendingOrder(order_id, item.code, "base", volume, datetime.now().date().isoformat())) + request = PlaceOrderRequest( + OP_BUY, item.code, volume, order_id, "zt", kind="base" + ) + state.new_order( + PendingOrder( + order_id, + item.code, + "base", + volume, + datetime.now().date().isoformat(), + ) + ) # 即使响应丢失,也保留资金预算和 pending,不能继续使用这笔钱。 available -= amount if run.orders.place(run.client, request): diff --git a/py-client/strategy/zt/positions.py b/py-client/strategy/zt/positions.py index 8e5ac59..fc7de99 100644 --- a/py-client/strategy/zt/positions.py +++ b/py-client/strategy/zt/positions.py @@ -8,22 +8,29 @@ import math from libs.grid_take_profit import GridState from sdk import OP_BUY, OP_SELL, PositionItem -from .order import PlaceOrderRequest -from .runtime import Runtime -from .state import PendingOrder, READY, SOLD +from libs.order import PlaceOrderRequest +from libs.runtime import Runtime +from .state import PendingOrder, READY, SOLD, TState -def manage_positions(run: Runtime, ticks, positions: list[PositionItem], available: float, - today: str, force_buy_back: bool = False) -> float: +def manage_positions( + run: Runtime, + state_store: TState, + ticks, + positions: list[PositionItem], + available: float, + today: str, + force_buy_back: bool = False, +) -> float: """遍历本地底仓记录;全部卖出后即使持仓快照为空,也必须处理买回。""" by_code = {position.stock_code: position for position in positions} - for code, state in list(run.state.items.items()): + for code, state in list(state_store.items.items()): try: now = datetime.now() if now.hour >= 15: break force_buy_back = force_buy_back or (now.hour, now.minute) >= (14, 50) - if code in run.account_cfg.excluded_codes or run.state.busy(code): + if code in run.account_cfg.excluded_codes or state_store.busy(code): continue if run.orders.busy(code, "BUY") or run.orders.busy(code, "SELL"): continue @@ -36,39 +43,65 @@ def manage_positions(run: Runtime, ticks, positions: list[PositionItem], availab expected_qty = state.base_qty - state.sell_qty + state.buy_qty # 快照延迟或手动增减仓不能当作新的做 T 信号,先核对数量差异。 if actual_qty != expected_qty: - log.warning("[ZT 持仓] %s 数量不符,记录=%d,实际=%d,暂停交易", code, expected_qty, actual_qty) + log.warning( + "[ZT 持仓] %s 数量不符,记录=%d,实际=%d,暂停交易", + code, + expected_qty, + actual_qty, + ) continue if state.phase == SOLD: - available = _try_buy_back(run, state, price, available, today, force_buy_back) + available = _try_buy_back( + run, state_store, state, price, available, today, force_buy_back + ) elif state.phase == READY and position and not force_buy_back: if price <= run.account_cfg.zt_max_price: - _try_sell(run, state, position, price, today) + _try_sell(run, state_store, state, position, price, today) except Exception: log.exception("[ZT 持仓] %s 处理异常,继续后续证券", code) return available -def _try_sell(run: Runtime, state, position: PositionItem, price: float, today: str) -> None: +def _try_sell( + run: Runtime, + state_store: TState, + state, + position: PositionItem, + price: float, + today: str, +) -> None: """基于独立保存的底仓成本,用跨轮最高盈利网格判断做 T 卖出。""" if state.base_cost <= 0: return pnl_rate = (price - state.base_cost) / state.base_cost * 100 key = f"{run.account_cfg.account_id}:{state.code}:{today}" - observation = run.sell_tracker.observe(key, pnl_rate) + observation = run.profit_tracker.observe(key, pnl_rate) if observation.state != GridState.RETREAT: return - volume = min(position.can_use_volume, int(state.base_qty * run.account_cfg.zt_sell_ratio)) + volume = min( + position.can_use_volume, int(state.base_qty * run.account_cfg.zt_sell_ratio) + ) volume = volume // 100 * 100 if volume <= 0: return order_id = run.orders.new_order_id("t-sell") - request = PlaceOrderRequest(OP_SELL, state.code, volume, order_id, "zt", kind="sell") - run.state.new_order(PendingOrder(order_id, state.code, "sell", volume, today)) + request = PlaceOrderRequest( + OP_SELL, state.code, volume, order_id, "zt", kind="sell" + ) + state_store.new_order(PendingOrder(order_id, state.code, "sell", volume, today)) if run.orders.place(run.client, request): log.info("[ZT 卖出] %s %d 股,等待成交后确定买回数量和价格", state.code, volume) -def _try_buy_back(run: Runtime, state, price: float, available: float, today: str, force: bool) -> float: +def _try_buy_back( + run: Runtime, + state_store: TState, + state, + price: float, + available: float, + today: str, + force: bool, +) -> float: """按实际卖出均价下跌后反弹买回;尾盘不再受下跌幅度、反弹及价格上限限制。""" target = state.sell_price * (1 - run.account_cfg.zt_buy_fall_pct / 100) if not force and (price > target or price > run.account_cfg.zt_max_price): @@ -78,17 +111,22 @@ def _try_buy_back(run: Runtime, state, price: float, available: float, today: st if volume <= 0 or amount > available: log.warning("[ZT 买回] %s 买回资金不足或数量无效,保留未完成轮次", state.code) return available - if not force and not run.buy_watch.triggered("ZT 买回", state.code, price): + if not force and not run.add_watch.triggered("ZT 买回", state.code, price): return available order_id = run.orders.new_order_id("t-buy") request = PlaceOrderRequest(OP_BUY, state.code, volume, order_id, "zt", kind="buy") - run.state.new_order(PendingOrder(order_id, state.code, "buy", volume, today)) + state_store.new_order(PendingOrder(order_id, state.code, "buy", volume, today)) # pending 已落盘,任何请求结果都预留资金;下一轮再从柜台快照确认。 available -= amount try: if run.orders.place(run.client, request): - run.buy_watch.forget(state.code) - log.info("[ZT 买回] %s %d 股,%s", state.code, volume, "尾盘强制买回" if force else "下跌后反弹") + run.add_watch.forget(state.code) + log.info( + "[ZT 买回] %s %d 股,%s", + state.code, + volume, + "尾盘强制买回" if force else "下跌后反弹", + ) except Exception: log.exception("[ZT 买回] %s 请求结果未知,保留 pending 和预算", state.code) return available diff --git a/py-client/strategy/zt/runtime.py b/py-client/strategy/zt/runtime.py deleted file mode 100644 index 33831f0..0000000 --- a/py-client/strategy/zt/runtime.py +++ /dev/null @@ -1,31 +0,0 @@ -"""做 T 策略单次运行所需的上下文对象。""" - -from dataclasses import dataclass - -from config import AccountConfig, GlobalConfig -from libs.grid_take_profit import GridTrailingTracker -from sdk import Client -from .order import OrderBook -from .watch import DipWatch -from .state import TState - - -@dataclass(slots=True) -class Runtime: - """集中保存策略运行期间共享的依赖和状态。 - - 将这些对象集中到一个 dataclass 后,开仓、持仓管理和单轮调度函数 - 只需接收一个 Runtime,无需重复传递大量参数。 - """ - - # 外部服务与账户配置。 - client: Client - global_cfg: GlobalConfig - account_cfg: AccountConfig - - # 策略运行过程中共享的状态组件。 - state: TState - orders: OrderBook - open_watch: DipWatch - buy_watch: DipWatch - sell_tracker: GridTrailingTracker diff --git a/py-client/strategy/zt/state.py b/py-client/strategy/zt/state.py index 538b4aa..558f781 100644 --- a/py-client/strategy/zt/state.py +++ b/py-client/strategy/zt/state.py @@ -35,7 +35,7 @@ class TStateItem: class PendingOrder: order_id: str code: str - kind: str # base:底仓;sell:做 T 卖出;buy:做 T 买回 + kind: str # base:底仓;sell:做 T 卖出;buy:做 T 买回 qty: int trade_date: str submit_at: float = field(default_factory=time) @@ -52,7 +52,9 @@ class TState: self._load() @classmethod - def for_strategy(cls, data_dir: str | Path, strategy: str, account_id: str) -> TState: + def for_strategy( + cls, data_dir: str | Path, strategy: str, account_id: str + ) -> TState: return cls(Path(data_dir) / f"{strategy}_{account_id}_state.json") def busy(self, code: str) -> bool: @@ -69,7 +71,9 @@ class TState: del self.pending[order.order_id] raise - def reconcile(self, positions: list[PositionItem], orders: list[OrderItem], today: str) -> None: + def reconcile( + self, positions: list[PositionItem], orders: list[OrderItem], today: str + ) -> None: """先按实际成交记账,再接管未知持仓;不覆盖已记录的底仓成本。""" # 同一本地委托可能拆单;按券商订单号去重,数量齐全且全部结束才记账。 by_id: dict[str, dict[str, OrderItem]] = {} @@ -78,8 +82,11 @@ class TState: by_id.setdefault(order.local_order_id, {})[order.id] = order for order_id, pending in list(self.pending.items()): side = "SELL" if pending.kind == "sell" else "BUY" - rows = [row for row in by_id.get(order_id, {}).values() - if row.code == pending.code and row.side == side] + rows = [ + row + for row in by_id.get(order_id, {}).values() + if row.code == pending.code and row.side == side + ] if not rows: log.warning("[ZT 状态] 委托暂未查到,保留待确认:%s", order_id) continue @@ -87,11 +94,20 @@ class TState: continue if any(row.status not in TERMINAL_STATUSES for row in rows): continue - if any(row.status == "56" and row.traded_volume != row.volume for row in rows): + if any( + row.status == "56" and row.traded_volume != row.volume for row in rows + ): continue qty = sum(row.traded_volume for row in rows) - amounts = [row.trade_amount if row.trade_amount > 0 else row.trade_price * row.traded_volume - for row in rows if row.traded_volume > 0] + amounts = [ + ( + row.trade_amount + if row.trade_amount > 0 + else row.trade_price * row.traded_volume + ) + for row in rows + if row.traded_volume > 0 + ] if any(not math.isfinite(amount) or amount <= 0 for amount in amounts): continue amount = sum(amounts) @@ -108,31 +124,59 @@ class TState: item.phase = SOLD if qty else READY else: total = item.buy_qty + qty - item.buy_cost = (item.buy_qty * item.buy_cost + amount) / total if total else 0.0 + item.buy_cost = ( + (item.buy_qty * item.buy_cost + amount) / total if total else 0.0 + ) item.buy_qty = total item.buy_order_id = order_id item.phase = DONE if total >= item.sell_qty else SOLD if item.phase == DONE: item.trade_date = today # 零成交撤单也记录,保留计划、实际数量、均价和柜台终态。 - self.records.append({**asdict(pending), "confirmed_date": today, - "filled_qty": qty, "filled_cost": cost, - "amount": amount, "statuses": [row.status for row in rows]}) + self.records.append( + { + **asdict(pending), + "confirmed_date": today, + "filled_qty": qty, + "filled_cost": cost, + "amount": amount, + "statuses": [row.status for row in rows], + } + ) del self.pending[order_id] for position in positions: code = position.stock_code if position.volume <= 0 or self.busy(code): continue - if code not in self.items and math.isfinite(position.open_price) and position.open_price > 0: - self.items[code] = TStateItem(code, position.volume, position.open_price) - self.records.append({"kind": "import", "code": code, "date": today, - "filled_qty": position.volume, "filled_cost": position.open_price}) - log.warning("[ZT 底仓] 首次接管 %s,使用当前均价,无法还原历史成本", code) + if ( + code not in self.items + and math.isfinite(position.open_price) + and position.open_price > 0 + ): + self.items[code] = TStateItem( + code, position.volume, position.open_price + ) + self.records.append( + { + "kind": "import", + "code": code, + "date": today, + "filled_qty": position.volume, + "filled_cost": position.open_price, + } + ) + log.warning( + "[ZT 底仓] 首次接管 %s,使用当前均价,无法还原历史成本", code + ) for item in self.items.values(): # 未买回的轮次跨日继续,不删除零持仓的做 T 债务。 - if item.trade_date != today and item.phase == DONE and not self.busy(item.code): + if ( + item.trade_date != today + and item.phase == DONE + and not self.busy(item.code) + ): item.phase, item.trade_date = READY, "" item.sell_qty = item.buy_qty = 0 item.sell_price = item.buy_cost = 0.0 @@ -142,10 +186,15 @@ class TState: def save(self) -> None: self.path.parent.mkdir(parents=True, exist_ok=True) temporary = self.path.with_suffix(self.path.suffix + ".tmp") - payload = {"items": {key: asdict(item) for key, item in self.items.items()}, - "pending": {key: asdict(item) for key, item in self.pending.items()}, - "records": self.records} - temporary.write_text(json.dumps(payload, ensure_ascii=False, indent=2, allow_nan=False) + "\n", encoding="utf-8") + payload = { + "items": {key: asdict(item) for key, item in self.items.items()}, + "pending": {key: asdict(item) for key, item in self.pending.items()}, + "records": self.records, + } + temporary.write_text( + json.dumps(payload, ensure_ascii=False, indent=2, allow_nan=False) + "\n", + encoding="utf-8", + ) temporary.replace(self.path) def _load(self) -> None: @@ -153,12 +202,18 @@ class TState: return raw = json.loads(self.path.read_text(encoding="utf-8")) # 兼容原 ZT 文件,保留原底仓成本;没有额外版本字段。 - self.items = {code: TStateItem(**item) for code, item in raw.get("items", raw).items()} - self.pending = {key: PendingOrder(**item) for key, item in raw.get("pending", {}).items()} + self.items = { + code: TStateItem(**item) for code, item in raw.get("items", raw).items() + } + self.pending = { + key: PendingOrder(**item) for key, item in raw.get("pending", {}).items() + } self.records = raw.get("records", []) if "items" not in raw: # 旧记录只有提交行情价,不把它伪装成真实成交历史。 self.records.append({"kind": "legacy_import", "items": raw}) for item in self.items.values(): if item.phase in {"SELLING", "BUYING", SOLD}: - raise ValueError(f"[ZT 状态] {item.code} 旧做 T 轮次未结束,需先核对成交再迁移") + raise ValueError( + f"[ZT 状态] {item.code} 旧做 T 轮次未结束,需先核对成交再迁移" + ) diff --git a/py-client/strategy/zt/watch.py b/py-client/strategy/zt/watch.py deleted file mode 100644 index b5f3626..0000000 --- a/py-client/strategy/zt/watch.py +++ /dev/null @@ -1,104 +0,0 @@ -import logging as log -from dataclasses import dataclass -from datetime import datetime, timedelta -from threading import Lock - - -@dataclass(slots=True) -class _Entry: - last_close: float - expires_at: datetime - - -class DipWatch: - """观察价格低点,并在价格达到指定反弹幅度时触发。""" - - def __init__( - self, - expire_seconds: float = 300, - rebound_threshold: float = 1.5, # 反弹力度 1.5% - ) -> None: - self.expire_seconds = expire_seconds - self.rebound_threshold = rebound_threshold - self.data: dict[str, _Entry] = {} - self.lock = Lock() - - def triggered( - self, - tag: str, - code: str, - price: float, - now: datetime | None = None, - ) -> bool: - """更新观察价格;达到反弹阈值时返回 ``True``。""" - if price <= 0: - log.warning("[%s Watch] %s 价格无效:%.2f", tag, code, price) - return False - - current = now or datetime.now() - with self.lock: - watch = self.data.get(code) - - if watch is None: - self._start(code, price, current) - log.info( - "[%s Watch] %s 开始观察,收盘价=%.2f", - tag, - code, - price, - ) - return False - - if current >= watch.expires_at: - self._start(code, price, current) - log.info("[%sWatch] %s 观察已过期,重新观察,收盘价=%.2f", tag, code, price) - return False - - if price < watch.last_close: - old_price = watch.last_close - self._start(code, price, current) - log.info( - "[%s Watch] %s 刷新低点,原收盘价=%.2f,新收盘价=%.2f", - tag, - code, - old_price, - price, - ) - return False - - rebound = (price - watch.last_close) / watch.last_close * 100 - if rebound < self.rebound_threshold: - log.info( - "[%s Watch] %s 等待反弹,收盘价=%.2f,现价=%.2f,反弹=%.2f%%,阈值=%.2f%%", - tag, - code, - watch.last_close, - price, - rebound, - self.rebound_threshold, - ) - return False - - del self.data[code] - log.info( - "[%s Watch] %s 反弹触发,收盘价=%.2f,现价=%.2f,反弹=%.2f%%", - tag, - code, - watch.last_close, - price, - rebound, - ) - return True - - def forget(self, code: str) -> None: - """清除指定股票的价格观察状态。""" - with self.lock: - removed = self.data.pop(code, None) - if removed is not None: - log.info("[Watch] %s 已清除观察状态", code) - - def _start(self, code: str, price: float, now: datetime) -> None: - self.data[code] = _Entry( - last_close=price, - expires_at=now + timedelta(seconds=self.expire_seconds), - )