from __future__ import annotations from dataclasses import dataclass from typing import Any from urllib.parse import urlencode from .client import Client as HTTPClient from .models import Assets, OrderItem, PositionItem @dataclass(slots=True) class Portfolio: assets: Assets positions: list[PositionItem] orders: list[OrderItem] class Client_V2(HTTPClient): """QMT ``/api/v2`` synchronous client.""" def portfolio(self) -> Portfolio: data = self._get_json("/api/v2/portfolio", ) or {} positions = data.get("positions", {}) orders = data.get("orders", {}) return Portfolio( assets=Assets.from_dict(data.get("assets", {})), positions=[PositionItem.from_trade_detail(row) for row in positions], orders=[OrderItem.from_trade_detail(value) for value in orders], ) def positions(self) -> list[PositionItem]: data = self._post("/api/v2/positions", {"account": self.account_type}) or {} return [ PositionItem.from_dict(value, code) for code, value in data.get("data", {}).items() ] def assets(self) -> dict[str, Any]: return self._post("/api/v2/assets", {"account": self.account_type}) or {} def context_info(self) -> dict[str, Any]: return self._get("/api/v2/context/info") or {} def stock_name(self, stock_code: str) -> Any: return self._get_ref("stock_name", stock_code) def open_date(self, stock_code: str) -> Any: return self._get_ref("open_date", stock_code) def last_volume(self, stock_code: str) -> Any: return self._get_ref("last_volume", stock_code) def total_share(self, stock_code: str) -> Any: return self._get_ref("total_share", stock_code) def svol(self, stock_code: str) -> Any: return self._get_ref("svol", stock_code) def bvol(self, stock_code: str) -> Any: return self._get_ref("bvol", stock_code) def divid_factors(self, stock_code: str) -> Any: return self._get_ref("divid_factors", stock_code) def etf_info(self, stock_code: str) -> Any: return self._get_ref("etf_info", stock_code) def etf_iopv(self, stock_code: str) -> Any: return self._get_ref("etf_iopv", stock_code) def instrument_detail(self, stock_code: str) -> Any: return self._get_ref("instrumentdetail", stock_code) def his_st_data(self, stock_code: str) -> Any: return self._get_ref("his_st_data", stock_code) def _get_ref(self, endpoint: str, stock_code: str) -> Any: query = urlencode({"stock_code": stock_code}) payload = self._get(f"/api/v2/get/{endpoint}?{query}") or {} return payload.get("ref") Client = Client_V2 __all__ = ["Client", "Client_V2", "Portfolio"]