104 lines
3.7 KiB
Python
104 lines
3.7 KiB
Python
"""做 T 策略的底仓与日内轮次状态。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from dataclasses import asdict, dataclass
|
|
from pathlib import Path
|
|
from threading import Lock
|
|
from typing import Iterable
|
|
|
|
from sdk import OrderItem, PositionItem
|
|
|
|
READY = "READY"
|
|
SELLING = "SELLING"
|
|
SOLD = "SOLD"
|
|
BUYING = "BUYING"
|
|
DONE = "DONE"
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class TStateItem:
|
|
code: str
|
|
base_qty: int = 0
|
|
base_cost: float = 0.0
|
|
trade_date: str = ""
|
|
phase: str = READY
|
|
sell_order_id: str = ""
|
|
sell_qty: int = 0
|
|
sell_price: float = 0.0
|
|
buy_order_id: str = ""
|
|
|
|
|
|
class TState:
|
|
"""持久化 dcm 底仓和每只证券每日一次的做 T 进度。"""
|
|
|
|
def __init__(self, path: str | Path) -> None:
|
|
self.path = Path(path)
|
|
self.lock = Lock()
|
|
self.items = self._load()
|
|
|
|
@classmethod
|
|
def for_strategy(cls, data_dir: str | Path, strategy: str, account_id: str) -> "TState":
|
|
return cls(Path(data_dir) / f"{strategy}_{account_id}_state.json")
|
|
|
|
def get(self, code: str) -> TStateItem:
|
|
with self.lock:
|
|
return self.items[code]
|
|
|
|
def set(self, item: TStateItem) -> None:
|
|
with self.lock:
|
|
self.items[item.code] = item
|
|
|
|
def reconcile(self, positions: Iterable[PositionItem], orders: list[OrderItem], today: str) -> None:
|
|
position_list = [item for item in positions if item.stock_code and item.volume > 0]
|
|
position_codes = {item.stock_code for item in position_list}
|
|
by_local_id: dict[str, list[OrderItem]] = {}
|
|
for order in orders:
|
|
if order.local_order_id:
|
|
by_local_id.setdefault(order.local_order_id, []).append(order)
|
|
|
|
for position in position_list:
|
|
if position.stock_code not in self.items and position.open_price > 0:
|
|
self.set(TStateItem(position.stock_code, position.volume, position.open_price))
|
|
|
|
for code in list(self.items):
|
|
item = self.get(code)
|
|
if code not in position_codes:
|
|
with self.lock:
|
|
self.items.pop(code, None)
|
|
continue
|
|
if item.trade_date and item.trade_date != today and item.phase in {DONE, READY}:
|
|
item.trade_date, item.phase = "", READY
|
|
item.sell_order_id = item.buy_order_id = ""
|
|
item.sell_qty = 0
|
|
item.sell_price = 0.0
|
|
if item.phase == SELLING and _completed(by_local_id.get(item.sell_order_id)):
|
|
item.phase = SOLD
|
|
elif item.phase == BUYING and _completed(by_local_id.get(item.buy_order_id)):
|
|
item.phase = DONE
|
|
self.set(item)
|
|
self.save()
|
|
|
|
def save(self) -> None:
|
|
with self.lock:
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary = self.path.with_suffix(self.path.suffix + ".tmp")
|
|
temporary.write_text(json.dumps({key: asdict(value) for key, value in self.items.items()}, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
temporary.replace(self.path)
|
|
|
|
def _load(self) -> dict[str, TStateItem]:
|
|
try:
|
|
raw = json.loads(self.path.read_text(encoding="utf-8"))
|
|
except FileNotFoundError:
|
|
return {}
|
|
except (OSError, json.JSONDecodeError) as exc:
|
|
raise ValueError(f"[ZT 状态] 读取失败: {exc}") from exc
|
|
if not isinstance(raw, dict):
|
|
raise ValueError("[ZT 状态] 根节点必须是对象")
|
|
return {code: TStateItem(**value) for code, value in raw.items()}
|
|
|
|
|
|
def _completed(orders: list[OrderItem] | None) -> bool:
|
|
return bool(orders) and all(order.status == "56" for order in orders)
|