"""趋势策略委托簿,对应 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(frozen=True, 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() -> str: """生成 ``trend-xxxxxxxx`` 格式的本地订单号。""" return f"trend-{secrets.token_hex(12)}" 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