big-qmt/py-client/sdk/models.py

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:
    """Fields match format_orders() exactly."""

    stock_code: str = ""
    order_sys_id: str = ""
    ref: int = 0
    order_ref: str = ""
    direction: int = 0
    offset_flag: int = 0
    limit_price: float = 0.0
    volume_total_original: int = 0
    volume_traded: int = 0
    volume_total: int = 0
    traded_price: float = 0.0
    trade_amount: float = 0.0
    insert_date: str = ""
    insert_time: str = ""
    remark: str = ""
    order_status: int = 0

    @property
    def side(self) -> str:
        return _side(self.offset_flag)

    @property
    def get_local_order_id(self) -> str:
        return self.remark.split("|", 1)[0]

    @property
    def created_at(self) -> datetime | None:
        return _parse_datetime(self.insert_date, self.insert_time)


@dataclass(slots=True)
class DealItem:
    """Fields match format_deals() exactly."""

    stock_code: str = ""
    order_sys_id: str = ""
    ref: int = 0
    order_ref: str = ""
    direction: int = 0
    offset_flag: int = 0
    price: float = 0.0
    volume: int = 0
    trade_amount: float = 0.0
    trade_date: str = ""
    trade_time: str = ""
    remark: str = ""
    close_profit: float = 0.0

    @property
    def side(self) -> str:
        return _side(self.offset_flag)

    @property
    def get_local_order_id(self) -> str:
        return self.remark.split("|", 1)[0]


@dataclass(slots=True)
class PositionItem:
    """Fields match format_holding() exactly."""

    stock_code: str = ""
    stock_name: str = ""
    direction: Any = None
    volume: int = 0
    open_price: float = 0.0
    open_cost: 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 = ""


@dataclass(slots=True)
class Assets:
    """Fields match format_assets() exactly."""

    total: float = 0.0
    available: float = 0.0


@dataclass(slots=True)
class Portfolio:
    assets: Assets
    positions: dict[str, PositionItem]
    orders: list[OrderItem]


def _side(offset_flag: int) -> str:
    return {"23": "BUY", "24": "SELL", "48": "BUY", "49": "SELL"}.get(str(offset_flag), "")


def _parse_datetime(date: str, clock: str) -> datetime | None:
    clock = clock.replace(":", "").zfill(6)
    try:
        return datetime.strptime(date.replace("-", "") + 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)

    @classmethod
    def from_dict(cls, data: Any) -> "Tick":
        if not isinstance(data, dict):
            return cls()
        return cls(
            last_price=_number(
                data.get("lastPrice", data.get("last_price", data.get("LastPrice")))
            ),
            last_close=_number(
                data.get("lastClose", data.get("last_close", data.get("LastClose")))
            ),
            raw=data,
        )


@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