117 lines
4.1 KiB
Python
117 lines
4.1 KiB
Python
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"}
|
|
|
|
# QMT 的发行数据按市场再分一层时使用的市场键。
|
|
_IPO_MARKETS = {"SH", "SZ", "BJ"}
|
|
|
|
|
|
def _ipo_stock_code(code: Any, info: dict[str, Any]) -> str:
|
|
"""把 QMT 的证券代码键补成全码 ``600000.SH``。"""
|
|
text = str(code or "").strip()
|
|
if not text or "." in text:
|
|
return text
|
|
market = str(
|
|
info.get("market") or info.get("exchange") or info.get("ExchangeID") or ""
|
|
).strip().upper()
|
|
return f"{text}.{market}" if market else text
|
|
|
|
|
|
def _ipo_entry(code: Any, info: dict[str, Any]) -> dict[str, Any]:
|
|
"""保留原始发行信息，并补出候选列表使用的 ``stock`` 字段。"""
|
|
entry = dict(info)
|
|
stock = _ipo_stock_code(code, entry)
|
|
if stock:
|
|
entry["stock"] = stock
|
|
return entry
|
|
|
|
|
|
def _ipo_candidates(response: Any) -> list[dict[str, Any]]:
|
|
"""把 ``/api/trade/ipo_data`` 的响应规范化为候选列表。
|
|
|
|
QMT 的 ``get_ipo_data(type)`` 返回 ``{证券代码: 发行信息}`` 字典，部分
|
|
版本再按市场分一层 ``{市场: {证券代码: 发行信息}}``，旧版服务端还会包
|
|
一层 ``{"data": ...}``。空响应表示当日没有可申购标的；无法识别的结构抛
|
|
``ValueError``，避免把接口异常静默当成“今日无新股”。
|
|
"""
|
|
if isinstance(response, dict) and len(response) == 1 and "data" in response:
|
|
response = response["data"]
|
|
if response is None:
|
|
return []
|
|
if isinstance(response, list):
|
|
# 列表逐项交给策略层校验，单条异常不影响其他候选。
|
|
return list(response)
|
|
if not isinstance(response, dict):
|
|
raise ValueError(f"unsupported IPO response type: {type(response).__name__}")
|
|
if not response:
|
|
return []
|
|
if not all(isinstance(value, dict) for value in response.values()):
|
|
raise ValueError("IPO response values must be objects")
|
|
|
|
candidates: list[dict[str, Any]] = []
|
|
for key, value in response.items():
|
|
market = str(key).strip().upper()
|
|
if market in _IPO_MARKETS and all(isinstance(item, dict) for item in value.values()):
|
|
for code, info in value.items():
|
|
entry = dict(info)
|
|
entry.setdefault("market", market)
|
|
candidates.append(_ipo_entry(code, entry))
|
|
else:
|
|
candidates.append(_ipo_entry(key, value))
|
|
return candidates
|
|
|
|
|
|
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.
|
|
|
|
QMT 按证券代码返回字典，这里统一成候选列表，字段名保持不变。
|
|
"""
|
|
response = self._post_json(
|
|
"/api/trade/ipo_data",
|
|
{"type": str(ipo_type).strip().upper()},
|
|
)
|
|
return _ipo_candidates(response)
|
|
|
|
|
|
def cancel_by_id(self, order_id: str) -> dict[str, Any]:
|
|
return self._post_json("/api/trade/cancel_by_id", {"order_id": order_id})
|