from __future__ import annotations from datetime import datetime from dataclasses import dataclass, field from typing import Any def _number(value: Any, kind: type = float) -> Any: try: return kind(value or 0) except (TypeError, ValueError): return kind() @dataclass(slots=True) class OrderItem: """由 QMT 委托明细解析得到的标准订单记录。""" id: str code: str side: str remark: str status: str created_at: datetime | None volume: int local_order_id: str = "" traded_volume: int = 0 remaining_volume: int = 0 exchange_id: str = "" name: str = "" price: float = 0.0 trade_price: float = 0.0 trade_amount: float = 0.0 @classmethod def from_trade_detail(cls, data: dict[str, Any]) -> "OrderItem": """从 TradeDetailData 的 QMT 原始字段创建订单。""" instrument_id = str(data.get("m_strInstrumentID") or "") exchange_id = str(data.get("m_strExchangeID") or "") code = ( f"{instrument_id}.{exchange_id}" if instrument_id and exchange_id else instrument_id ) remaining_volume = _number(data.get("m_nVolumeTotal"), int) traded_volume = _number(data.get("m_nVolumeTraded"), int) remark = str(data.get("m_strRemark") or "") return cls( id=str(data.get("m_strOrderSysID") or ""), code=code, side={"23": "BUY", "24": "SELL", "48": "BUY", "49": "SELL"}.get( str(data.get("m_nOffsetFlag")), "" ), remark=remark, status=str(data.get("m_nOrderStatus") or ""), created_at=_trade_datetime(data), volume=remaining_volume + traded_volume, local_order_id=remark.split("|", 1)[0] if remark else "", traded_volume=traded_volume, remaining_volume=remaining_volume, exchange_id=exchange_id, name=str(data.get("m_strInstrumentName") or ""), price=_number(data.get("m_dPrice")), trade_price=_number(data.get("m_dTradePrice")), trade_amount=_number(data.get("m_dTradeAmount")), ) @dataclass(slots=True) class PositionItem: stock_code: str = "" stock_name: str = "" direction: Any = None volume: int = 0 open_price: float = 0.0 float_profit: float = 0.0 market_value: float = 0.0 stock_holder: str = "" frozen_volume: int = 0 can_use_volume: int = 0 on_road_volume: int = 0 yesterday_volume: int = 0 last_price: float = 0.0 profit_rate: float = 0.0 future_trade_type: Any = None expire_date: str = "" @classmethod def from_dict(cls, data: dict[str, Any], code: str = "") -> "PositionItem": return cls( stock_code=str(data.get("StockCode") or code), stock_name=str(data.get("StockName") or ""), direction=data.get("Direction"), volume=_number(data.get("Volume"), int), open_price=_number(data.get("OpenPrice")), float_profit=_number(data.get("FloatProfit")), market_value=_number(data.get("MarketValue")), stock_holder=str(data.get("StockHolder") or ""), frozen_volume=_number(data.get("FrozenVolume"), int), can_use_volume=_number(data.get("CanUseVolume"), int), on_road_volume=_number(data.get("OnRoadVolume"), int), yesterday_volume=_number(data.get("YesterdayVolume"), int), last_price=_number(data.get("LastPrice")), profit_rate=_number(data.get("ProfitRate")), future_trade_type=data.get("FutureTradeType"), expire_date=str(data.get("ExpireDate") or ""), ) @classmethod def from_trade_detail(cls, data: dict[str, Any]) -> "PositionItem": """从 TradeDetailData/Holding 的 QMT 原始字段创建持仓。""" instrument_id = str(data.get("m_strInstrumentID") or "") exchange_id = str(data.get("m_strExchangeID") or "") stock_code = ( f"{instrument_id}.{exchange_id}" if instrument_id and exchange_id else instrument_id ) return cls( stock_code=stock_code, stock_name=str(data.get("m_strInstrumentName") or ""), direction=data.get("m_nDirection"), volume=_number(data.get("m_nVolume"), int), open_price=_number(data.get("m_dOpenPrice")), float_profit=_number(data.get("m_dFloatProfit")), market_value=_number(data.get("m_dMarketValue")), stock_holder=str(data.get("m_strStockHolder") or ""), frozen_volume=_number(data.get("m_nFrozenVolume"), int), can_use_volume=_number(data.get("m_nCanUseVolume"), int), on_road_volume=_number(data.get("m_nOnRoadVolume"), int), yesterday_volume=_number(data.get("m_nYesterdayVolume"), int), last_price=_number(data.get("m_dLastPrice")), profit_rate=_number(data.get("m_dProfitRate")), future_trade_type=data.get("m_eFutureTradeType"), expire_date=str(data.get("m_strExpireDate") or ""), ) @dataclass(slots=True) class Assets: total: float = 0.0 available: float = 0.0 @classmethod def from_dict(cls, data: dict[str, Any]) -> "Assets": """兼容新版 QMT 原始资金字段及旧版简化字段。""" return cls( total=_number(data.get("m_dBalance", data.get("total"))), available=_number(data.get("m_dAvailable", data.get("available"))), ) def _trade_datetime(data: dict[str, Any]) -> datetime | None: date = str(data.get("m_strInsertDate") or "") clock = str(data.get("m_strInsertTime") or "").replace(":", "").zfill(6) try: return datetime.strptime(date + clock, "%Y%m%d%H%M%S") except ValueError: return None @dataclass(slots=True) class Tick: last_price: float = 0.0 last_close: float = 0.0 raw: dict[str, Any] = field(default_factory=dict) @dataclass(slots=True) class HistoryDataRequest: length: int = 10 period: str = "" field: str = "" dividend_type: int = 0 skip_paused: bool = True @dataclass(slots=True) class MarketDataRequest: fields: list[str] = field(default_factory=list) stocks: list[str] = field(default_factory=list) start_time: str = "" end_time: str = "" period: str = "" dividend_type: str = "" count: int = 0 @dataclass(slots=True) class FinancialDataRequest: tabname: str = ""; colname: str = ""; market: str = ""; code: str = "" report_type: str = ""; barpos: int = 0 field_list: list[str] = field(default_factory=list); stock_list: list[str] = field(default_factory=list) start_date: str = ""; end_date: str = "" @dataclass(slots=True) class FactorDataRequest: field_list: list[str] = field(default_factory=list); stock_list: list[str] = field(default_factory=list) stock_code: str = ""; start_date: str = ""; end_date: str = "" @dataclass(slots=True) class BSMPriceRequest: option_type: str; object_prices: Any; strike_price: float; risk_free: float; sigma: float; days: int; dividend: float @dataclass(slots=True) class BSMIVRequest: option_type: str; object_prices: float; strike_price: float; option_price: float; risk_free: float; days: int; dividend: float @dataclass(slots=True) class LocalDataRequest: stock_code: str; start_time: str = ""; end_time: str = ""; period: str = ""; divid_type: str = ""; count: int = 0