58 lines
1.7 KiB
Python
58 lines
1.7 KiB
Python
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})
|