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

from __future__ import annotations

from typing import Any


OP_BUY = 23
OP_SELL = 24
ORDER_TYPE_VOLUME = 1101
PR_TYPE_LATEST = 5
QUICK_TRADE_NOW = 2
ORDER_SIDE_BY_OFFSET = {"23": "BUY", "24": "SELL", "48": "BUY", "49": "SELL"}


class TradeMixin:
    def passorder(
        self,
        op_type: int,
        stock_code: str = "",
        volume: int = 0,
        order_type: int = ORDER_TYPE_VOLUME,
        pr_type: int = PR_TYPE_LATEST,
        price: float = -1,
        quick_trade: int = QUICK_TRADE_NOW,
        strategy_name: str = "",
        order_id: str = "",
        stock: str = "",
    ) -> dict[str, Any]:
        # ``stock`` is retained for compatibility with the original IPO client.
        stock_code = str(stock_code or stock).strip()
        if not stock_code:
            raise ValueError("stock_code cannot be empty")
        return self._post_json(
            "/api/trade/passorder",
            {
                "opType": op_type,
                "orderType": order_type,
                "stockCode": stock_code,
                "prType": pr_type,
                "price": price,
                "volume": volume,
                "quickTrade": quick_trade,
                "strategyName": strategy_name,
                "orderId": order_id,
            },
        )

    def ipo_data(self, ipo_type: str = "STOCK") -> list[dict[str, Any]]:
        """Return today's IPO candidates from the QMT REST service."""
        response = self._post_json(
            "/api/trade/ipo_data",
            {"type": str(ipo_type).strip().upper()},
        ) or []
        return response if isinstance(response, list) else []


    def cancel_by_id(self, order_id: str) -> dict[str, Any]:
        return self._post_json("/api/trade/cancel_by_id", {"order_id": order_id})