From 1fc3e119d3a3e5e1c3045621d44376b269406f25 Mon Sep 17 00:00:00 2001 From: yanweidong Date: Sat, 5 Sep 2026 15:56:39 +0800 Subject: [PATCH] fix state.py --- py-client/strategy/trend/boot.py | 2 +- py-client/strategy/trend/open.py | 3 + py-client/strategy/trend/positions.py | 9 +- py-client/strategy/trend/state.py | 183 ++++++++++---------------- 4 files changed, 78 insertions(+), 119 deletions(-) diff --git a/py-client/strategy/trend/boot.py b/py-client/strategy/trend/boot.py index f84ee0b..1623462 100644 --- a/py-client/strategy/trend/boot.py +++ b/py-client/strategy/trend/boot.py @@ -61,7 +61,7 @@ def StartTrend() -> None: config.account_config.strategy, config.account_config.account_id, ) - order_book = OrderBook(state=storeState) + order_book = OrderBook() order_book.refresh(client, portfolio.orders) storeState.reconcile(positions, portfolio.orders) diff --git a/py-client/strategy/trend/open.py b/py-client/strategy/trend/open.py index 08d9492..83b40df 100644 --- a/py-client/strategy/trend/open.py +++ b/py-client/strategy/trend/open.py @@ -8,6 +8,7 @@ from libs import calc_buy_volume from sdk import OP_BUY from .runtime import Runtime from .order import PlaceOrderRequest +from .state import PendingOrder import logging as log @@ -88,6 +89,8 @@ def do_open(run: Runtime, code: str, volume: int, signal_key: str, price: float) signal_key, kind="base", ) + if not run.state.new_order(PendingOrder(order_id, code, "base", volume)): + raise RuntimeError("已有待确认买单") if not run.orders.place(run.client,request): raise RuntimeError("订单提交失败") diff --git a/py-client/strategy/trend/positions.py b/py-client/strategy/trend/positions.py index 1227adc..77310f7 100644 --- a/py-client/strategy/trend/positions.py +++ b/py-client/strategy/trend/positions.py @@ -10,6 +10,7 @@ from sdk import OP_BUY, OP_SELL, PositionItem, Tick from .order import PlaceOrderRequest from .runtime import Runtime +from .state import PendingOrder import logging as log LOSS_TIERS = (-30.0, -50.0) @@ -124,14 +125,13 @@ def handle_profit( return TradeDecision(False, "无可用整手持仓") order_id = runtime.orders.new_order_id() request = PlaceOrderRequest( - client=runtime.client, op=OP_SELL, code=position.stock_code, volume=volume, order_id=order_id, strategy_name=runtime.account_cfg.strategy, ) - if not runtime.orders.place(request): + if not runtime.orders.place(runtime.client, request): return TradeDecision(False, "止盈委托失败") return TradeDecision(True, f"卖出 {volume} 股，订单={order_id}") @@ -167,7 +167,6 @@ def handle_loss( order_id = runtime.orders.new_order_id() request = PlaceOrderRequest( - client=runtime.client, op=OP_BUY, code=position.stock_code, volume=volume, @@ -175,7 +174,9 @@ def handle_loss( strategy_name=runtime.account_cfg.strategy, kind="add", ) - if not runtime.orders.place(request): + if not runtime.state.new_order(PendingOrder(order_id, position.stock_code, "added", volume)): + return TradeDecision(False, "已有待确认买单") + if not runtime.orders.place(runtime.client, request): reserved = amount if runtime.state.busy(position.stock_code) else 0.0 return TradeDecision(False, "补仓委托失败或待确认", reserved) runtime.add_watch.forget(position.stock_code) diff --git a/py-client/strategy/trend/state.py b/py-client/strategy/trend/state.py index 4abfd29..0faa814 100644 --- a/py-client/strategy/trend/state.py +++ b/py-client/strategy/trend/state.py @@ -1,13 +1,11 @@ -"""简单成交账本：订单明确结束后一次记账，JSON 原子保存。""" +"""底仓、补仓记录与待确认订单的 JSON 存储。""" from __future__ import annotations import json import logging as log import math -import shutil -import time -from dataclasses import asdict, dataclass, field, replace +from dataclasses import asdict, dataclass, replace from pathlib import Path from threading import RLock from typing import Iterable @@ -21,26 +19,20 @@ TERMINAL_STATUSES = {"53", "54", "56", "57"} class StateItem: code: str base_order_id: str = "" - base_pre_qty: int =0 base_qty: int = 0 base_cost: float = 0.0 added_order_id: str = "" - added_pre_qty: int =0 added_num: int = 0 added_qty: int = 0 - added_amount: float = 0.0 - - @property - def added_cost(self) -> float: - return self.added_amount / self.added_qty if self.added_qty else 0.0 + added_cost: float = 0.0 @dataclass(frozen=True, slots=True) class PendingOrder: order_id: str + code: str kind: str - expected_qty: int - submitted_at: float = field(default_factory=time.time) + pre_qty: int class State: @@ -48,154 +40,117 @@ class State: self.path = Path(path) self.lock = RLock() self.items: dict[str, StateItem] = {} - self.pending: dict[str, PendingOrder] = {} # 每只证券最多一个待确认买单 - self._dirty = False - self._next_warning = 0.0 + self.pending: dict[str, PendingOrder] = {} + self.IsModify = False self._load() + def _load(self) -> None: + """启动时读取 JSON 中的持仓记录和待确认订单。""" + if self.path.is_file(): + raw = json.loads(self.path.read_text(encoding="utf-8")) + self.items = {code: StateItem(**item) for code, item in raw["items"].items()} + self.pending = {code: PendingOrder(**item) for code, item in raw["pending"].items()} + @classmethod def for_strategy(cls, data_dir: str | Path, strategy: str, account_id: str) -> State: return cls(Path(data_dir) / f"{strategy}_{account_id}_state.json") - @property - def codes(self) -> list[str]: - with self.lock: - return list(self.items) - def get(self, code: str) -> StateItem: with self.lock: - return replace(self.items[code]) + return self.items[code] def busy(self, code: str) -> bool: with self.lock: return code in self.pending - def begin(self, order_id: str, code: str, kind: str, expected_qty: int) -> bool: - if kind not in {"base", "add"} or type(expected_qty) is not int or expected_qty <= 0: - raise ValueError("订单类型或预开仓数量无效") + def new_order(self, order: PendingOrder) -> bool: + """提交前保存待确认订单，同一证券已有 pending 时跳过。""" with self.lock: - if code in self.pending or (kind == "base" and code in self.items): + if order.code in self.pending: return False - if kind == "add" and code not in self.items: - raise ValueError("缺少底仓记录") - dirty = self._dirty - self.pending[code] = PendingOrder(order_id, kind, expected_qty) - self._dirty = True + was_modified = self.IsModify + self.pending[order.code] = order + self.IsModify = True try: - self.save() # 必须成功落盘后才允许发送请求 + self.save() except Exception: - self.pending.pop(code) - self._dirty = dirty + del self.pending[order.code] + self.IsModify = was_modified raise return True - def reject(self, order_id: str) -> None: - """仅用于已明确未受理的请求。""" - with self.lock: - for code, pending in self.pending.items(): - if pending.order_id == order_id: - del self.pending[code] - self._dirty = True - self.save() - return - def reconcile(self, positions: Iterable[PositionItem], orders: list[OrderItem]) -> None: + """核对 pending 的订单结果，再导入/核对实际持仓。""" by_id: dict[str, dict[str, OrderItem]] = {} for order in orders: - if order.local_order_id and order.id: + if order.local_order_id: by_id.setdefault(order.local_order_id, {})[order.id] = order - with self.lock: - for position in positions: - code = position.stock_code - if (code and code not in self.items and code not in self.pending - and position.volume > 0 and math.isfinite(position.open_price) - and position.open_price > 0): - self.items[code] = StateItem(code, position.volume, position.open_price) - self._dirty = True - now = time.time() + with self.lock: for code, pending in list(self.pending.items()): - rows = [o for o in by_id.get(pending.order_id, {}).values() - if o.code == code and o.side == "BUY"] - filled = _finished_fill(rows, pending.expected_qty) + rows = [order for order in by_id.get(pending.order_id, {}).values() + if order.code == code and order.side == "BUY"] + if not rows: + del self.pending[code] + self.IsModify = True + log.info("[状态] 清理无对应订单的 pending，代码=%s，订单=%s", code, pending.order_id) + continue + filled = _finished_fill(rows, pending.pre_qty) if filled is None: - if now >= self._next_warning and now - pending.submitted_at >= 180: - log.warning("[状态] 订单待核查，代码=%s，订单=%s，计划=%d；保留防重", - code, pending.order_id, pending.expected_qty) continue qty, amount = filled if qty: - item = self.items.setdefault(code, StateItem(code)) + item = self.items.get(code, StateItem(code)) if pending.kind == "base": - item.base_qty, item.base_cost = qty, amount / qty + item = replace(item, base_order_id=pending.order_id, + base_qty=qty, base_cost=amount / qty) else: - item.added_num += 1 - item.added_qty += qty - item.added_amount += amount + total_qty = item.added_qty + qty + total_amount = item.added_qty * item.added_cost + amount + item = replace(item, added_order_id=pending.order_id, + added_num=item.added_num + 1, + added_qty=total_qty, added_cost=total_amount / total_qty) + self.items[code] = item del self.pending[code] - self._dirty = True + self.IsModify = True log.info("[状态] 对账结束，代码=%s，订单=%s，计划=%d，成交=%d", - code, pending.order_id, pending.expected_qty, qty) - if now >= self._next_warning: - self._next_warning = now + 180 - self.save() # 成交记账与移除 pending 在同一文件中原子提交 + code, pending.order_id, pending.pre_qty, qty) + + for position in positions: + code = position.stock_code + if not code or position.volume <= 0: + continue + if code not in self.items and code not in self.pending: + self.items[code] = StateItem(code=code, base_order_id=position.trade_id, + base_qty=position.volume, base_cost=position.open_price) + self.IsModify = True + elif code in self.items: + item = self.items[code] + if item.base_qty + item.added_qty != position.volume: + log.info("[状态] 持仓数量差异，代码=%s，记录=%d，实际=%d", + code, item.base_qty + item.added_qty, position.volume) + self.save() def save(self) -> None: with self.lock: - if not self._dirty: + if not self.IsModify: return self.path.parent.mkdir(parents=True, exist_ok=True) temporary = self.path.with_suffix(self.path.suffix + ".tmp") - payload = dict(version=3, items={k: asdict(v) for k, v in self.items.items()}, - pending={k: asdict(v) for k, v in self.pending.items()}) + payload = dict(items={code: asdict(item) for code, item in self.items.items()}, + pending={code: asdict(item) for code, item in self.pending.items()}) temporary.write_text(json.dumps(payload, ensure_ascii=False, indent=2, allow_nan=False) + "\n", encoding="utf-8") temporary.replace(self.path) - self._dirty = False - - def _load(self) -> None: - try: - raw = json.loads(self.path.read_text(encoding="utf-8")) - except FileNotFoundError: - return - if raw.get("version") not in {2, 3}: - raise ValueError("仅支持 v2/v3 状态文件；旧格式请先核实转换，原文件未修改") - self.items = {k: StateItem(**v) for k, v in raw["items"].items()} - if raw["version"] == 3: - self.pending = {k: PendingOrder(**v) for k, v in raw["pending"].items()} - return - # v2 的未决订单可能已经增量记账，先撤回这部分，结束时再完整记一次。 - for old in raw["pending"].values(): - code = old["code"] - if code in self.pending: - raise ValueError(f"{code} 存在多个旧未决订单，请核查；原文件未修改") - item = self.items[code] - qty, amount = old.get("applied_qty", 0), old.get("applied_amount", 0.0) - if old["kind"] == "add": - item.added_qty -= qty - item.added_amount -= amount - item.added_num -= int(old.get("counted", False)) - else: - base_amount = item.base_qty * item.base_cost - amount - item.base_qty -= qty - item.base_cost = base_amount / item.base_qty if item.base_qty else 0.0 - if not item.base_qty and not item.added_qty: - del self.items[code] - self.pending[code] = PendingOrder(old["order_id"], old["kind"], old["expected_qty"], - old.get("submitted_at", time.time())) - backup = self.path.with_suffix(self.path.suffix + ".v2.bak") - if not backup.exists(): - shutil.copy2(self.path, backup) - self._dirty = True - self.save() + self.IsModify = False -def _finished_fill(orders: list[OrderItem], expected_qty: int) -> tuple[int, float] | None: - """仅完整终态快照可记账；缺项、未成交完或金额未知均继续等待。""" - if not orders or expected_qty <= 0 or sum(o.volume for o in orders) != expected_qty: +def _finished_fill(orders: list[OrderItem], pre_qty: int) -> tuple[int, float] | None: + """订单全部结束且成交金额完整后一次记账，部分成交撤单也按实计入。""" + if sum(order.volume for order in orders) != pre_qty: return None qty, amount = 0, 0.0 for order in orders: - if (order.status not in TERMINAL_STATUSES or not 0 <= order.traded_volume <= order.volume + if (order.status not in TERMINAL_STATUSES or (order.status == "56" and order.traded_volume != order.volume)): return None if order.traded_volume: