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

from __future__ import annotations

from typing import Any

from .models import Assets, DealItem, OrderItem, Portfolio, PositionItem


class PortfolioMixin:
    def org(self, datatype: str) -> list[dict[str, Any]]:
        """查询 account、order、deal 或 position，返回原始字段字典列表。"""
        datatype = str(datatype).strip().lower()
        if datatype not in {"account", "order", "deal", "position"}:
            raise ValueError(f"unsupported org datatype: {datatype}")
        response = self._get_json(f"/api/portfolio/org/{datatype}")
        return response["data"]

    def portfolio(self) -> Portfolio:
        data = self._get_json("/api/portfolio") or {}
        positions = {
            code: PositionItem.from_dict(value, code)
            for code, value in data.get("positions", {}).items()
        }
        return Portfolio(
            assets=Assets.from_dict(data.get("assets", {})),
            positions=positions,
            orders=[OrderItem.from_trade_detail(row) for row in data.get("orders", [])],
        )

    def positions(self) -> tuple[list[str], list[PositionItem]]:
        data = self._get_json("/api/portfolio/positions") or {}
        positions = [
            PositionItem.from_dict(value, code)
            for code, value in data.get("data", {}).items()
        ]
        return [item.stock_code for item in positions], positions

    def assets(self) -> Assets:
        return Assets.from_dict(self._get_json("/api/portfolio/assets") or {})

    def orders(self) -> list[OrderItem]:
        data = self._get_json("/api/portfolio/order") or []
        return [OrderItem.from_trade_detail(row) for row in data]

    def deals(self) -> list[DealItem]:
        """查询原始 Deal 成交对象并转换为标准成交记录。"""
        return [DealItem.from_trade_detail(row) for row in self.org("deal")]

    def trade_detail_data(self, datatype: str) -> Any:
        datatype = str(datatype).strip().lower()
        handlers = {
            "account": self.assets,
            "position": lambda: self.positions()[1],
            "order": self.orders,
            "deal": self.deals,
        }
        handler = handlers.get(datatype)
        if handler is None:
            raise ValueError(f"unsupported trade detail datatype: {datatype}")
        result = handler()
        return [result] if datatype == "account" else result