125 lines
4.4 KiB
Python
125 lines
4.4 KiB
Python
"""趋势策略委托簿，对应 Go 客户端的 ``logic/order.go``。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import secrets
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timedelta
|
|
from threading import Lock
|
|
from typing import Any
|
|
|
|
from sdk import ORDER_SIDE_BY_OFFSET, Client, OrderItem
|
|
|
|
# 表示委托仍在处理、可能继续成交的 QMT 状态。
|
|
BUSY_STATUSES = {"48", "49", "50", "51", "52", "55"}
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class PlaceOrderRequest:
|
|
"""``OrderBook.place`` 提交委托所需的全部参数。"""
|
|
|
|
client: Any
|
|
op: int
|
|
code: str
|
|
volume: int
|
|
order_id: str
|
|
strategy_name: str
|
|
|
|
|
|
class OrderBook:
|
|
"""线程安全的活动委托缓存。"""
|
|
|
|
def __init__(self, lock_timeout_sec: float = 180, cancel_timeout_sec: float = 10) -> None:
|
|
self.lock_timeout_sec = max(0.0, float(lock_timeout_sec))
|
|
self.cancel_timeout_sec = timedelta(seconds=cancel_timeout_sec)
|
|
self.data: dict[str, OrderItem] = {}
|
|
self.lock: dict[str, float] = {}
|
|
self.mutex = Lock()
|
|
|
|
@staticmethod
|
|
def new_order_id(leg: str) -> str:
|
|
"""生成短订单号，为 QMT 备注中的信号键预留空间。"""
|
|
return f"zt-{leg[:1]}-{secrets.token_hex(4)}"
|
|
|
|
def busy(self, code: str, side: str) -> bool:
|
|
"""判断证券是否存在仍在处理中的同方向委托。"""
|
|
with self.mutex:
|
|
self._clear_expired_locks(datetime.now().timestamp())
|
|
key = f"{side}-{code}"
|
|
return key in self.lock
|
|
|
|
def refresh(self, client: Client) -> None:
|
|
"""从 QMT 刷新当前委托明细和方向索引。"""
|
|
orders = client.trade_detail_data("order")
|
|
parsed_orders = [(f"{item.side}-{item.code}", item) for item in orders]
|
|
now_timestamp = datetime.now().timestamp()
|
|
with self.mutex:
|
|
self.data = {key: item for key, item in parsed_orders}
|
|
self.lock = {
|
|
key: (
|
|
item.created_at.timestamp()
|
|
if item.created_at is not None
|
|
else now_timestamp
|
|
)
|
|
for key, item in parsed_orders
|
|
if item.status in BUSY_STATUSES
|
|
}
|
|
self._clear_expired_locks(now_timestamp)
|
|
|
|
def cancel_expired(self, client: Any, now: datetime | None = None) -> None:
|
|
"""尝试撤销超过有效期且具有委托编号的订单。"""
|
|
self.refresh(client)
|
|
current = now or datetime.now()
|
|
|
|
# 使用快照遍历，避免网络调用期间长期持有互斥锁。
|
|
for order in list(self.data.values()):
|
|
if (
|
|
order.created_at is not None
|
|
and order.status in {"49", "50", "51", "52"}
|
|
and current - order.created_at > self.cancel_timeout_sec
|
|
and order.id
|
|
):
|
|
client.cancel_by_id(order.id)
|
|
|
|
def place(self, request: PlaceOrderRequest) -> bool:
|
|
"""按最新价提交委托，并立即写入本地方向锁。"""
|
|
result = request.client.passorder_latest_tagged(
|
|
request.op,
|
|
request.code,
|
|
request.volume,
|
|
request.strategy_name,
|
|
request.order_id,
|
|
)
|
|
if not isinstance(result, dict):
|
|
return False
|
|
order_ref = str(result.get("order_ref") or "").strip().lower()
|
|
if result.get("status") != "success" or order_ref in {"", "unknown", "none"}:
|
|
return False
|
|
|
|
side = ORDER_SIDE_BY_OFFSET.get(str(request.op), "")
|
|
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:
|
|
key = f"{side}-{request.code}"
|
|
self.data[key] = pending
|
|
self.lock[key] = pending.created_at.timestamp()
|
|
return True
|
|
|
|
def _clear_expired_locks(self, now_timestamp: float) -> None:
|
|
"""清理过期方向锁；调用方必须已持有 ``mutex``。"""
|
|
expired = [
|
|
key
|
|
for key, created_at in self.lock.items()
|
|
if now_timestamp - created_at >= self.lock_timeout_sec
|
|
]
|
|
for key in expired:
|
|
self.lock.pop(key, None)
|