fix bug

py-client/strategy/trend/boot.py

@@ -56,15 +56,14 @@ def StartTrend() -> None:
         portfolio = client.portfolio()
         assets = portfolio.assets
         positions = list(portfolio.positions.values())
-        order_book = OrderBook()
-        order_book.refresh(client, portfolio.orders)
-
         storeState = State.for_strategy(
             config.global_config.qmt_data_dir,
             config.account_config.strategy,
             config.account_config.account_id,
         )
-        storeState.reconcile(positions, order_book.data)
+        order_book = OrderBook(state=storeState)
+        order_book.refresh(client, portfolio.orders)
+        storeState.reconcile(positions, portfolio.orders)
 
         # 获取本策略的信号开仓数据
         signals = init_signals(
@@ -180,7 +179,7 @@ def RunOnce(run: Runtime, signals:list[SignalItem]) -> None:
 
     # 6. 更新状态机
     try:
-        run.state.reconcile(positions, run.orders.data)
+        run.state.reconcile(positions, portfolio.orders)
     except Exception:
         log.exception("[状态] 订单状态对账失败")
         return

py-client/strategy/trend/open.py

@@ -8,7 +8,6 @@ from libs import calc_buy_volume
 from sdk import OP_BUY
 from .runtime import Runtime
 from .order import PlaceOrderRequest
-from .state import STATUS_ING, StateItem
 import logging as log
 
 
@@ -88,16 +87,11 @@ def do_open(run: Runtime, code: str, volume: int, signal_key: str, price: float)
         volume,
         order_id,
         signal_key,
+        kind="base",
     )
     if not run.orders.place(request):
         raise RuntimeError("订单提交失败")
 
-    run.state.set(StateItem(
-        code=code,
-        base_order_id=order_id,
-        base_status=STATUS_ING,
-    ))
-    run.state.save()
     run.open_watch.forget(code)
 
 

py-client/strategy/trend/order.py

@@ -3,15 +3,14 @@
 from __future__ import annotations
 
 import secrets
-import logging as log
 from dataclasses import dataclass
 from datetime import datetime, timedelta
 from threading import Lock
-from typing import Any
 from cachelib import SimpleCache
-import httpx
+import logging
 
-from sdk import APIError, ORDER_SIDE_BY_OFFSET, Client, OrderItem
+import httpx
+from sdk import Client,ORDER_SIDE_BY_OFFSET,APIError,OrderItem
 
 # 表示委托仍在处理、可能继续成交的 QMT 状态。
 BUSY_STATUSES = {"48", "49", "50", "51", "52", "55"}
@@ -23,13 +22,12 @@ CANCELABLE_STATUSES = {"49", "50", "51", "52"}
 @dataclass(frozen=True, slots=True)
 class PlaceOrderRequest:
     """``OrderBook.place`` 提交委托所需的全部参数。"""
-
-    client: Any
     op: int
     code: str
     volume: int
     order_id: str
     strategy_name: str
+    kind: str = ""
 
 
 class OrderBook:
@@ -79,30 +77,28 @@ class OrderBook:
             ):
                 client.cancel_by_id(item.id)
                 canceled += 1
-                log.info("[Order] 超时撤单，代码=%s，方向=%s，柜台订单=%s", item.code, item.side, item.id)
+                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
-        log.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, request: PlaceOrderRequest) -> bool:
         """按最新价提交委托，并立即写入本地方向锁。"""
         side = ORDER_SIDE_BY_OFFSET.get(str(request.op), "")
         if not side:
-            log.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):
-                log.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)
 
@@ -115,31 +111,12 @@ class OrderBook:
                 order_id=request.order_id,
             )
         except APIError as exc:
-            log.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):
             # 响应异常不能证明柜台未受理，保留缓存防重，不自动重试。
-            log.exception("[Order] 下单请求或响应异常，代码=%s，本地订单=%s", request.code, request.order_id)
+            logging.exception("[Order] 下单请求或响应异常，代码=%s，本地订单=%s", request.code, request.order_id)
             return False
-        if not isinstance(result, dict):
-            log.warning("[Order] 下单失败，代码=%s，本地订单=%s，原因=响应格式无效", request.code, request.order_id)
-            return False
-        order_ref = str(result.get("order_ref") or "").strip().lower()
-        if result.get("status") != "success" or order_ref in {"", "unknown", "none"}:
-            log.warning("[Order] 下单被拒绝，代码=%s，本地订单=%s，状态=%s，柜台订单=%s", request.code, request.order_id, result.get("status"), order_ref)
-            return False
-
-        pending = OrderItem(
-            id=order_ref,
-            code=request.code,
-            side=side,
-            remark=request.order_id,
-            status="48",
-            created_at=datetime.now(),
-            volume=request.volume,
-            local_order_id=request.order_id,
-        )
-        with self.mutex:
-            self.data.append(pending)
-        log.info("[Order] 下单已受理，代码=%s，方向=%s，数量=%d，本地订单=%s，柜台订单=%s", request.code, side, request.volume, request.order_id, order_ref)
+        
+        logging.info("[Order] 下单已受理，代码=%s，方向=%s，数量=%d，本地订单=%s，返回=%s", request.code, side, request.volume, request.order_id, result)
         return True

py-client/strategy/trend/positions.py

@@ -10,7 +10,6 @@ from sdk import OP_BUY, OP_SELL, PositionItem, Tick
 
 from .order import PlaceOrderRequest
 from .runtime import Runtime
-from .state import STATUS_ING
 import logging as log
 
 LOSS_TIERS = (-30.0, -50.0)
@@ -181,14 +180,11 @@ def handle_loss(
         volume=volume,
         order_id=order_id,
         strategy_name=runtime.account_cfg.strategy,
+        kind="add",
     )
     if not runtime.orders.place(request):
-        return TradeDecision(False, "补仓委托失败")
-
-    state.added_status = STATUS_ING
-    state.added_order_id = order_id
-    runtime.state.set(state)
-    runtime.state.save()
+        reserved = amount if runtime.state.busy(position.stock_code) else 0.0
+        return TradeDecision(False, "补仓委托失败或待确认", reserved)
     runtime.add_watch.forget(position.stock_code)
     return TradeDecision(True, f"买入 {volume} 股，订单={order_id}", amount)
 

py-client/strategy/trend/state.py

@@ -1,239 +1,207 @@
-"""趋势策略持仓状态的内存管理与 JSON 持久化。"""
+"""简单成交账本：订单明确结束后一次记账，JSON 原子保存。"""
 
 from __future__ import annotations
 
 import json
 import logging as log
-from dataclasses import asdict, dataclass
+import math
+import shutil
+import time
+from dataclasses import asdict, dataclass, field, replace
 from pathlib import Path
-from threading import Lock
+from threading import RLock
 from typing import Iterable
 
 from sdk import OrderItem, PositionItem
-from .order import BUSY_STATUSES, COMPLETED_STATUSES
 
-
-# 委托状态：无操作、处理中、已完成。
-STATUS_NONE = ""
-STATUS_ING = "ING"
-STATUS_OK = "OK"
+TERMINAL_STATUSES = {"53", "54", "56", "57"}
 
 
 @dataclass(slots=True)
 class StateItem:
-    """单只证券的底仓和补仓状态。"""
-
-    # 证券代码。
     code: str
-
-    # 底仓订单、数量、成本和处理状态。
     base_order_id: str = ""
+    base_pre_qty: int =0
     base_qty: int = 0
     base_cost: float = 0.0
-    base_status: str = STATUS_NONE
-
-    # 补仓订单、补仓次数、数量、成本和处理状态。
     added_order_id: str = ""
+    added_pre_qty: int =0
     added_num: int = 0
     added_qty: int = 0
-    added_cost: float = 0.0
-    added_status: str = STATUS_NONE
+    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
+
+
+@dataclass(slots=True)
+class PendingOrder:
+    order_id: str
+    kind: str
+    expected_qty: int
+    submitted_at: float = field(default_factory=time.time)
 
 
 class State:
-    """线程安全的策略状态存储。
-
-    状态以内存字典提供快速访问，并通过临时文件替换的方式写入 JSON，
-    防止程序在写入过程中退出而破坏原状态文件。
-    """
-
     def __init__(self, path: str | Path) -> None:
         self.path = Path(path)
-        self.lock = Lock()
-        self.items = self._load()
+        self.lock = RLock()
+        self.items: dict[str, StateItem] = {}
+        self.pending: dict[str, PendingOrder] = {}  # 每只证券最多一个待确认买单
+        self._dirty = False
+        self._next_warning = 0.0
+        self._load()
 
     @classmethod
-    def for_strategy(
-        cls,
-        data_dir: str | Path,
-        strategy: str,
-        account_id: str,
-    ) -> "State":
-        """根据数据目录、策略名称和账户生成独立状态文件。"""
-        state_path = Path(data_dir) / f"{strategy}_{account_id}_state.json"
-        return cls(state_path)
+    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:
-        """获取指定证券的状态；不存在时抛出 KeyError。"""
         with self.lock:
-            return self.items[code]
+            return replace(self.items[code])
 
-    def set(self, item: StateItem) -> None:
-        """新增或覆盖一只证券的状态。"""
+    def busy(self, code: str) -> bool:
         with self.lock:
-            self.items[item.code] = item
+            return code in self.pending
 
-    def delete(self, code: str) -> bool:
-        """删除已终结的证券状态，并返回是否实际删除。"""
+    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("订单类型或预开仓数量无效")
         with self.lock:
-            return self.items.pop(code, None) is not None
+            if code in self.pending or (kind == "base" and code in self.items):
+                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
+            try:
+                self.save()  # 必须成功落盘后才允许发送请求
+            except Exception:
+                self.pending.pop(code)
+                self._dirty = dirty
+                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 sync_positions(self, positions: Iterable[PositionItem]) -> None:
-        """把尚未接管的真实持仓初始化为已完成底仓。
-
-        无证券代码、无持仓数量或成本无效的记录会被忽略。同步结束后
-        立即保存，确保首次接管的持仓在程序重启后仍可恢复。
-        """
-        known_codes = set(self.codes)
-        imported = 0
-        for position in positions:
-            if (
-                not position.stock_code
-                or position.volume <= 0
-                or position.open_price <= 0
-                or position.stock_code in known_codes
-            ):
-                continue
-
-            self.set(
-                StateItem(
-                    base_order_id=position.trade_id,
-                    code=position.stock_code,
-                    base_qty=position.volume,
-                    base_cost=round(position.open_price, 2),
-                    base_status=STATUS_OK,
-                )
-            )
-            known_codes.add(position.stock_code)
-            imported += 1
-
-        self.save()
-        if imported:
-            log.info("[状态] 导入持仓=%d，状态总数=%d", imported, len(known_codes))
-
-    def reconcile(
-        self,
-        positions: Iterable[PositionItem],
-        orders: list[OrderItem],
-    ) -> None:
-        """用真实持仓和委托恢复本地状态；拆分订单全部完成才算完成。"""
-        position_list = list(positions)
-        self.sync_positions(position_list)
-        position_codes = {
-            item.stock_code for item in position_list if item.volume > 0
-        }
-        orders_by_local_id: dict[str, list[OrderItem]] = {}
+    def reconcile(self, positions: Iterable[PositionItem], orders: list[OrderItem]) -> None:
+        by_id: dict[str, dict[str, OrderItem]] = {}
         for order in orders:
-            if order.local_order_id:
-                orders_by_local_id.setdefault(order.local_order_id, []).append(order)
+            if order.local_order_id and 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
 
-        for code in list(self.codes):
-            item = self.get(code)
-            for order_id_attr, status_attr, qty_attr, cost_attr in (
-                ("base_order_id", "base_status", "base_qty", "base_cost"),
-                ("added_order_id", "added_status", "added_qty", "added_cost"),
-            ):
-                local_order_id = getattr(item, order_id_attr)
-                current_status = getattr(item, status_attr)
-                if current_status != STATUS_ING or not local_order_id:
+            now = time.time()
+            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)
+                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
-
-                matching_orders = orders_by_local_id.get(local_order_id)
-                status = _order_status(matching_orders)
-                if status != current_status:
-                    log.info("[状态] %s 订单=%s，状态=%s->%s", code, local_order_id, current_status, status)
-                setattr(item, status_attr, status)
-                if status == STATUS_OK:
-                    quantity, cost = _filled_order(matching_orders)
-                    if quantity > 0:
-                        setattr(item, qty_attr, quantity)
-                    if cost > 0:
-                        setattr(item, cost_attr, cost)
-                    if status_attr == "added_status":
+                qty, amount = filled
+                if qty:
+                    item = self.items.setdefault(code, StateItem(code))
+                    if pending.kind == "base":
+                        item.base_qty, item.base_cost = qty, amount / qty
+                    else:
                         item.added_num += 1
-            self.set(item)
-
-        # Opening orders normally have no position until their first fill.  Order
-        # reconciliation must therefore happen before stale state is removed.
-        for code in list(self.codes):
-            if code not in position_codes:
-                item = self.get(code)
-                if any(
-                    order_id and order_id in orders_by_local_id
-                    for order_id in (item.base_order_id, item.added_order_id)
-                ):
-                    continue
-                if self.delete(code):
-                    log.info("[状态] 已移除持仓状态，代码=%s", code)
-        self.save()
+                        item.added_qty += qty
+                        item.added_amount += amount
+                del self.pending[code]
+                self._dirty = 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 在同一文件中原子提交
 
     def save(self) -> None:
-        """将内存状态格式化写入 JSON，并原子替换正式文件。"""
         with self.lock:
+            if not self._dirty:
+                return
             self.path.parent.mkdir(parents=True, exist_ok=True)
-            temporary_path = self.path.with_suffix(self.path.suffix + ".tmp")
-            payload = {
-                code: asdict(item)
-                for code, item in self.items.items()
-            }
-            temporary_path.write_text(
-                json.dumps(payload, ensure_ascii=False, indent=2) + "\n",
-                encoding="utf-8",
-            )
-            temporary_path.replace(self.path)
+            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()})
+            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) -> dict[str, StateItem]:
-        """读取已有状态文件；文件不存在时从空状态开始。"""
+    def _load(self) -> None:
         try:
             raw = json.loads(self.path.read_text(encoding="utf-8"))
         except FileNotFoundError:
-            return {}
-        except (OSError, json.JSONDecodeError) as exc:
-            raise ValueError(f"[状态] 读取或解析失败: {exc}") from exc
-
-        if not isinstance(raw, dict):
-            raise ValueError("[状态] 状态文件根节点必须是 JSON 对象")
-
-        try:
-            return {
-                code: StateItem(**item)
-                for code, item in raw.items()
-            }
-        except (TypeError, ValueError) as exc:
-            raise ValueError(f"[状态] 状态字段无效: {exc}") from exc
+            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()
 
 
-def _order_status(orders: list[OrderItem] | None) -> str:
-    """将柜台订单简化为无状态、处理中或已成交。"""
-    if not orders:
-        return STATUS_NONE
-    statuses = {order.status for order in orders}
-    if statuses <= COMPLETED_STATUSES:
-        return STATUS_OK
-    if statuses <= BUSY_STATUSES | COMPLETED_STATUSES:
-        return STATUS_ING
-    return STATUS_NONE
-
-
-def _filled_order(orders: list[OrderItem] | None) -> tuple[int, float]:
-    """汇总已成交订单的实际数量和加权成交价。"""
-    quantity = 0
-    amount = 0.0
-    for order in orders or []:
-        filled = order.traded_volume if order.traded_volume > 0 else order.volume
-        if filled <= 0:
-            continue
-        quantity += filled
-        if order.trade_amount > 0:
-            amount += order.trade_amount
-        elif order.trade_price > 0:
-            amount += order.trade_price * filled
-    cost = round(amount / quantity, 4) if quantity > 0 and amount > 0 else 0.0
-    return quantity, cost
+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:
+        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
+                or (order.status == "56" and order.traded_volume != order.volume)):
+            return None
+        if order.traded_volume:
+            value = order.trade_amount if order.trade_amount > 0 else order.trade_price * order.traded_volume
+            if not math.isfinite(value) or value <= 0:
+                return None
+            qty += order.traded_volume
+            amount += value
+    return qty, amount