206 lines
7.3 KiB
Python
206 lines
7.3 KiB
Python
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
|