diff --git a/py-client/strategy/trend/open.py b/py-client/strategy/trend/open.py index 83b40df..cc585ca 100644 --- a/py-client/strategy/trend/open.py +++ b/py-client/strategy/trend/open.py @@ -80,7 +80,7 @@ def open_signal(run:Runtime, ticks, open_signals) -> None: def do_open(run: Runtime, code: str, volume: int, signal_key: str, price: float) -> None: """生成本地订单号并按最新价提交开仓委托。""" - order_id = run.orders.new_order_id() + order_id = run.orders.new_order_id("BUY") request = PlaceOrderRequest( OP_BUY, code, @@ -89,8 +89,9 @@ 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("已有待确认买单") + + #run.state.new_order(PendingOrder(order_id, code, "base", volume)) + if not run.orders.place(run.client,request): raise RuntimeError("订单提交失败") diff --git a/py-client/strategy/trend/order.py b/py-client/strategy/trend/order.py index 57130d6..de8bef6 100644 --- a/py-client/strategy/trend/order.py +++ b/py-client/strategy/trend/order.py @@ -42,9 +42,9 @@ class OrderBook: self.mutex = Lock() @staticmethod - def new_order_id() -> str: + def new_order_id(side:str) -> str: """生成 ``trend-xxxxxxxx`` 格式的本地订单号。""" - return f"trend-{secrets.token_hex(12)}" + return f"trend-{side}-{secrets.token_hex(10)}" def busy(self, code: str, side: str) -> bool: """判断证券是否存在仍在处理中的同方向委托。""" diff --git a/py-client/strategy/trend/positions.py b/py-client/strategy/trend/positions.py index 77310f7..f3fe661 100644 --- a/py-client/strategy/trend/positions.py +++ b/py-client/strategy/trend/positions.py @@ -123,7 +123,7 @@ def handle_profit( volume = position.can_use_volume - position.can_use_volume % 100 if volume <= 0: return TradeDecision(False, "无可用整手持仓") - order_id = runtime.orders.new_order_id() + order_id = runtime.orders.new_order_id("SELL") request = PlaceOrderRequest( op=OP_SELL, code=position.stock_code, @@ -165,7 +165,7 @@ def handle_loss( if volume <= 0 or amount > available: return TradeDecision(False, "本轮可用资金不足") - order_id = runtime.orders.new_order_id() + order_id = runtime.orders.new_order_id("BUY") request = PlaceOrderRequest( op=OP_BUY, code=position.stock_code, @@ -174,8 +174,9 @@ def handle_loss( strategy_name=runtime.account_cfg.strategy, kind="add", ) - if not runtime.state.new_order(PendingOrder(order_id, position.stock_code, "added", volume)): - return TradeDecision(False, "已有待确认买单") + + #runtime.state.new_order(PendingOrder(order_id, position.stock_code, "added", volume)) + if not runtime.orders.place(runtime.client, request): reserved = amount if runtime.state.busy(position.stock_code) else 0.0 return TradeDecision(False, "补仓委托失败或待确认", reserved) diff --git a/py-client/strategy/trend/state.py b/py-client/strategy/trend/state.py index 0faa814..267230a 100644 --- a/py-client/strategy/trend/state.py +++ b/py-client/strategy/trend/state.py @@ -4,15 +4,15 @@ from __future__ import annotations import json import logging as log -import math -from dataclasses import asdict, dataclass, replace +import time +from dataclasses import asdict, dataclass, field, replace from pathlib import Path from threading import RLock -from typing import Iterable +from typing import Any from sdk import OrderItem, PositionItem -TERMINAL_STATUSES = {"53", "54", "56", "57"} +PENDING_TIME_OUT = 3600 @dataclass(frozen=True, slots=True) @@ -33,14 +33,15 @@ class PendingOrder: code: str kind: str pre_qty: int + submit_at: int = field(default_factory=lambda: int(time.time())) class State: - def __init__(self, path: str | Path) -> None: - self.path = Path(path) + def __init__(self, data_dir: str | Path, strategy: str, account_id: str) -> None: + self.path = Path(data_dir) / f"{strategy}_{account_id}_state.json" self.lock = RLock() self.items: dict[str, StateItem] = {} - self.pending: dict[str, PendingOrder] = {} + self.pending: list[PendingOrder] = [] self.IsModify = False self._load() @@ -49,86 +50,131 @@ class State: 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()} + self.pending = [PendingOrder(**item) for item in raw["pending"]] + now = int(time.time()) + for pending in list(self.pending): + if now - pending.submit_at >= PENDING_TIME_OUT: + self.pending.remove(pending) + self.IsModify = True + log.info("[状态] 清理超时 pending，代码=%s，订单=%s", pending.code, pending.order_id) + self.save() - @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") def get(self, code: str) -> StateItem: with self.lock: return self.items[code] - def busy(self, code: str) -> bool: - with self.lock: - return code in self.pending - - def new_order(self, order: PendingOrder) -> bool: + def new_order(self, order: PendingOrder) -> None: """提交前保存待确认订单，同一证券已有 pending 时跳过。""" with self.lock: - if order.code in self.pending: - return False - was_modified = self.IsModify - self.pending[order.code] = order - self.IsModify = True - try: + if order: + self.pending.append(order) + self.IsModify = True self.save() - except Exception: - del self.pending[order.code] - self.IsModify = was_modified - raise - return True - def reconcile(self, positions: Iterable[PositionItem], orders: list[OrderItem]) -> None: - """核对 pending 的订单结果，再导入/核对实际持仓。""" - by_id: dict[str, dict[str, OrderItem]] = {} + def merged_order(self, orders: list[OrderItem]) -> dict[str, dict[str, Any]]: + """按证券代码、本地订单号合并，返回成交数量、金额、均价和状态。""" + merged: dict[str, dict[str, Any]] = {} for order in orders: - if order.local_order_id: - by_id.setdefault(order.local_order_id, {})[order.id] = order + if not order.local_order_id: + continue + + if not order.local_order_id in merged.keys(): + strStatus = "ING" + if order.status == "56": + strStatus = "OK" + merged[order.local_order_id] = { + "code":order.code, + "qty": order.traded_volume, + "cost": order.trade_price, + "status": order.status, + "merged_status":strStatus + } + continue + + old = merged[order.local_order_id] + strStatus = "ING" + if old["status"] == order.status == "56": + strStatus = "OK" + + totalQty = old["qty"]+order.traded_volume + # 合计成交数量为零，跳过，避免除零。 + if totalQty == 0: + continue + # 有成交数量但缺少成交金额，跳过，避免拉低成本。 + if order.traded_volume > 0 and not order.trade_amount: + continue + cost = ((old["qty"]*old["cost"])+order.trade_amount) / totalQty + merged[order.local_order_id]["qty"] = totalQty + merged[order.local_order_id]["cost"] = cost + merged[order.local_order_id]["merged_status"] = strStatus + + return merged + + def reconcile(self, positions: list[PositionItem], orders: list[OrderItem]) -> None: + """合并订单 → 对齐 pending 和持仓 → 保存 JSON。""" + # 1. 合并同一本地订单的成交数据。 + merged = self.merged_order(orders) with self.lock: - for code, pending in list(self.pending.items()): - 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] + # 2. 按本地订单号查找合并结果，核对证券代码后写入底仓或补仓。 + # 只留下尚未完成确认的订单，避免循环中反复查找、删除列表元素。 + remaining: list[PendingOrder] = [] + for pending in self.pending: + code = pending.code + result = merged.get(pending.order_id) + if result is None or result["code"] != code: + # 按现有规则：本轮快照中没有对应订单，就清理 pending。 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 result["merged_status"] != "OK" or result["qty"] != pending.pre_qty: + # 未全部成功，或拆单快照的数量尚未齐全，留到下轮确认。 + remaining.append(pending) continue - qty, amount = filled + qty, cost = result["qty"], result["cost"] + # 只记录实际成交。 if qty: item = self.items.get(code, StateItem(code)) if pending.kind == "base": + # 底仓：记录本次成交数量和实际均价。 item = replace(item, base_order_id=pending.order_id, - base_qty=qty, base_cost=amount / qty) + base_qty=qty, base_cost=cost) else: + # 补仓：次数加一，数量累加，成本按成交数量加权。 total_qty = item.added_qty + qty - total_amount = item.added_qty * item.added_cost + amount + total_amount = item.added_qty * item.added_cost + qty * cost 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.IsModify = True log.info("[状态] 对账结束，代码=%s，订单=%s，计划=%d，成交=%d", code, pending.order_id, pending.pre_qty, qty) + self.pending = remaining + + # 未记录且没有待确认订单的持仓，作为首次接管的底仓导入。 + pending_codes = {pending.code for pending in remaining} 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) + item = self.items.get(code) + if item is None: + if code not in pending_codes: + self.items[code] = StateItem(code=code, base_order_id=position.trade_id, + base_qty=position.volume, base_cost=position.open_price) + self.IsModify = True + continue + + # 已有记录只报告数量差异，不用总持仓覆盖底仓/补仓的划分。 + recorded_qty = item.base_qty + item.added_qty + if recorded_qty != position.volume: + log.info("[状态] 持仓数量差异，代码=%s，记录=%d，实际=%d", + code, recorded_qty, position.volume) + + # 3. 本轮统一保存；没有修改时 save() 不写文件。 self.save() def save(self) -> None: @@ -138,25 +184,7 @@ class State: self.path.parent.mkdir(parents=True, exist_ok=True) temporary = self.path.with_suffix(self.path.suffix + ".tmp") payload = dict(items={code: asdict(item) for code, item in self.items.items()}, - pending={code: asdict(item) for code, item in self.pending.items()}) + pending=[asdict(item) for item in self.pending]) temporary.write_text(json.dumps(payload, ensure_ascii=False, indent=2, allow_nan=False) + "\n", encoding="utf-8") temporary.replace(self.path) self.IsModify = False - - -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 (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