from __future__ import annotations from typing import Any from .models import Assets, 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[dict[str, Any]]: data = self._get_json("/api/portfolio/deal") or {} return data.get("deals", []) 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