291 lines
13 KiB
Python
291 lines
13 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
下游/上游只读表访问 (trading_* 与 strategy_daily_results, 153 代理, 严格单表)
|
|
==============================================================================
|
|
这三张表**归下游系统维护**, PMS 只读 (设计 §9)。
|
|
|
|
列名口径 (2026-07-27 实机 SHOW COLUMNS 确认, 见 check_db.py 输出):
|
|
trading_position: id, stock_code, stock_name, total_quantity, available_quantity,
|
|
frozen_quantity, cost_price, market_price, market_value, profit_loss, ...
|
|
→ **可用数量列 available_quantity 下游已提供**, QMT 需求清单 A1 的核心疑问就此落地,
|
|
PMS 无需自行按 T+1 推算下游可卖量 (账本侧仍按 T+1 自行维护, 对账时以下游为准)。
|
|
探测机制保留: 候选表把已确认列名排在首位, 万一下游改名或换表也能自动兜住, 探测结果经
|
|
「运维 → 导出下游表结构」可见, 也是回填 D1 的现成材料。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from app.db.session import fetch_all, fetch_one
|
|
|
|
# 数量/可用量列名候选 (探测顺序即优先级; 首项为实机确认的真实列名)
|
|
QTY_CANDIDATES = ["total_quantity", "current_qty", "total_qty", "position_qty", "hold_qty",
|
|
"stock_qty", "volume", "quantity", "qty", "position_volume", "hold_volume"]
|
|
AVAIL_CANDIDATES = ["available_quantity", "available_qty", "avail_qty", "can_use_volume",
|
|
"available_volume", "sellable_qty", "enable_amount", "can_sell_qty"]
|
|
COST_CANDIDATES = ["cost_price", "avg_cost", "open_price", "position_cost", "cost"]
|
|
FROZEN_CANDIDATES = ["frozen_quantity", "frozen_qty", "frozen_volume", "freeze_qty"]
|
|
PRICE_CANDIDATES = ["market_price", "last_price", "current_price", "price"]
|
|
|
|
FILLED_STATUSES = ("completed", "filled")
|
|
|
|
|
|
def to_dot(code: str) -> str:
|
|
"""SH600000 / 600000 → 600000.SH (PMS 内部统一点式)。"""
|
|
c = (code or "").strip().upper()
|
|
if not c:
|
|
return ""
|
|
if "." in c:
|
|
return c
|
|
if c[:2] in ("SH", "SZ", "BJ") and c[2:].isdigit():
|
|
return f"{c[2:]}.{c[:2]}"
|
|
if c.isdigit() and len(c) == 6:
|
|
return f"{c}.SH" if c[0] == "6" else (f"{c}.SZ" if c[0] in "03" else f"{c}.BJ")
|
|
return c
|
|
|
|
|
|
def to_prefix(code: str) -> str:
|
|
"""600000.SH → SH600000 (strategy_daily_results 口径)。"""
|
|
c = (code or "").strip().upper()
|
|
if "." in c:
|
|
num, mkt = c.split(".", 1)
|
|
return f"{mkt}{num}"
|
|
return c
|
|
|
|
|
|
def _pick(keys, candidates):
|
|
low = {str(k).lower(): k for k in keys}
|
|
for c in candidates:
|
|
if c in low:
|
|
return low[c]
|
|
return None
|
|
|
|
|
|
def describe(table: str) -> list:
|
|
"""SHOW COLUMNS —— 供页面导出、回填 QMT_INTERFACE_REQUIREMENTS D1。"""
|
|
if table not in ("trading_position", "trading_order", "trading_buy_plan",
|
|
"strategy_daily_results", "gp_stock_category"):
|
|
raise ValueError(f"不允许探测的表: {table}")
|
|
return fetch_all(f"SHOW COLUMNS FROM {table}")
|
|
|
|
|
|
# ================================================================ trading_position
|
|
def fetch_positions() -> dict:
|
|
"""下游持仓快照。返回 {"rows":[{ts_code, qty, avail_qty, cost, frozen}], "columns":{...}}"""
|
|
rows = fetch_all("SELECT * FROM trading_position LIMIT 1000")
|
|
if not rows:
|
|
return {"rows": [], "columns": {"qty": None, "avail": None, "cost": None},
|
|
"raw_count": 0}
|
|
keys = rows[0].keys()
|
|
qty_col = _pick(keys, QTY_CANDIDATES)
|
|
avail_col = _pick(keys, AVAIL_CANDIDATES)
|
|
cost_col = _pick(keys, COST_CANDIDATES)
|
|
frozen_col = _pick(keys, FROZEN_CANDIDATES)
|
|
price_col = _pick(keys, PRICE_CANDIDATES)
|
|
code_col = _pick(keys, ["stock_code", "ts_code", "code", "security_code"]) or "stock_code"
|
|
|
|
out = []
|
|
for r in rows:
|
|
code = to_dot(str(r.get(code_col) or ""))
|
|
if not code:
|
|
continue
|
|
out.append({
|
|
"ts_code": code,
|
|
"qty": _int(r.get(qty_col)) if qty_col else None,
|
|
"avail_qty": _int(r.get(avail_col)) if avail_col else None,
|
|
"cost": _float(r.get(cost_col)) if cost_col else None,
|
|
"frozen": _int(r.get(frozen_col)) if frozen_col else None,
|
|
"price": _float(r.get(price_col)) if price_col else None,
|
|
})
|
|
return {"rows": out, "raw_count": len(rows),
|
|
"columns": {"code": code_col, "qty": qty_col, "avail": avail_col,
|
|
"cost": cost_col, "frozen": frozen_col, "price": price_col}}
|
|
|
|
|
|
def _int(v):
|
|
try:
|
|
return int(float(v or 0))
|
|
except (TypeError, ValueError):
|
|
return 0
|
|
|
|
|
|
def _float(v):
|
|
try:
|
|
return float(v or 0)
|
|
except (TypeError, ValueError):
|
|
return 0.0
|
|
|
|
|
|
# ================================================================ trading_order
|
|
def fetch_filled_orders(*, since_id=None, since_time=None, limit: int = 500) -> list:
|
|
"""已成交单增量拉取 (回放用)。价格口径: filled_price > order_price (下游 filled_amount 不可靠)。
|
|
|
|
order_id 可能是字符串委托号 —— 游标同时支持数值与字符串比较, 取不到则按时间兜底。
|
|
"""
|
|
# 状态枚举为常量 (QMT A2 待正式确认), 直接内联避免 IN 绑定展开
|
|
where = ["order_status IN ('" + "', '".join(FILLED_STATUSES) + "')"]
|
|
p = {"n": int(limit)}
|
|
if since_id not in (None, "", 0):
|
|
where.append("order_id > :sid")
|
|
p["sid"] = since_id
|
|
elif since_time:
|
|
where.append("(filled_time >= :st OR order_time >= :st)")
|
|
p["st"] = since_time
|
|
sql = ("SELECT * FROM trading_order WHERE " + " AND ".join(where) +
|
|
" ORDER BY order_id ASC LIMIT :n")
|
|
rows = fetch_all(sql, p)
|
|
return [_norm_order(r) for r in rows]
|
|
|
|
|
|
def latest_filled_order_id():
|
|
"""当前已成交单里最大的 order_id —— 回放游标冷启动时的锚点。
|
|
|
|
取 MAX 而不是"最新时间", 是为了跟游标推进用同一把尺子: `fetch_filled_orders` 走的是
|
|
`order_id > :sid` + `ORDER BY order_id ASC`, 即**字典序** (order_id 形如
|
|
`SELL_601956.SH_1778549400`, 并非时间序)。锚点若按时间取, 会跟字典序对不上而漏行。
|
|
"""
|
|
r = fetch_one("SELECT MAX(order_id) AS mx FROM trading_order WHERE order_status IN ('"
|
|
+ "', '".join(FILLED_STATUSES) + "')")
|
|
return (r or {}).get("mx")
|
|
|
|
|
|
def _norm_order(r: dict) -> dict:
|
|
side = str(r.get("order_side") or "").strip().lower()
|
|
if side in ("1", "b", "买入", "买"):
|
|
side = "buy"
|
|
elif side in ("2", "s", "卖出", "卖"):
|
|
side = "sell"
|
|
qty = r.get("filled_quantity") or r.get("filled_qty") or r.get("order_quantity")
|
|
price = r.get("filled_price") or r.get("filled_avg_price") or r.get("order_price")
|
|
return {"order_id": r.get("order_id"), "ts_code": to_dot(str(r.get("stock_code") or "")),
|
|
"side": side, "qty": _int(qty), "price": _float(price),
|
|
"done_time": str(r.get("filled_time") or r.get("order_time") or ""),
|
|
"status": r.get("order_status"), "raw": r}
|
|
|
|
|
|
# ================================================================ trading_buy_plan
|
|
def fetch_buy_plans(*, is_active=None, limit: int = 200) -> list:
|
|
"""上游买入计划 (PMS 作为承接方, 替代原 ENTRY_GATE 角色 —— 设计 §6)。
|
|
|
|
is_active: 7=待仲裁 (上游产出待承接), 6=待挂单 (下游取走), 5=盘中拒。
|
|
"""
|
|
sql = ("SELECT id, strategy_id, stock_code, stock_name, target_price, buy_amount, "
|
|
"factor_code, trading_time, create_time, update_time, tp_ratio, sl_ratio, "
|
|
"prob_thresh, hold_days, is_active, approved_by FROM trading_buy_plan")
|
|
p = {"n": int(limit)}
|
|
if is_active is not None:
|
|
sql += " WHERE is_active = :ia"
|
|
p["ia"] = int(is_active)
|
|
sql += " ORDER BY update_time DESC LIMIT :n"
|
|
rows = fetch_all(sql, p)
|
|
out = []
|
|
for r in rows:
|
|
px = _float(r.get("target_price"))
|
|
out.append({"plan_id": r.get("id"), "ts_code": to_dot(str(r.get("stock_code") or "")),
|
|
"name": r.get("stock_name"), "price": px,
|
|
"amount": _float(r.get("buy_amount")),
|
|
"score": _float(r.get("prob_thresh")), "factor": r.get("factor_code"),
|
|
"tp_ratio": _float(r.get("tp_ratio")), "sl_ratio": _float(r.get("sl_ratio")),
|
|
"is_active": r.get("is_active"), "update_time": r.get("update_time")})
|
|
return out
|
|
|
|
|
|
# ================================================================ strategy_daily_results
|
|
def fetch_refs(ts_code: str):
|
|
"""决策系统昨夜结论: 支撑/压力参考位 (主口径; 停更超期由 services 兜底自算)。"""
|
|
r = fetch_one(
|
|
"SELECT stock_code, signal_type, support_level, pressure_level, trade_date "
|
|
"FROM strategy_daily_results WHERE stock_code = :code "
|
|
"ORDER BY trade_date DESC LIMIT 1", {"code": to_prefix(ts_code)})
|
|
if not r:
|
|
return None
|
|
return {"ts_code": ts_code, "signal_type": r.get("signal_type"),
|
|
"support": _float(r.get("support_level")) or None,
|
|
"pressure": _float(r.get("pressure_level")) or None,
|
|
"trade_date": r.get("trade_date")}
|
|
|
|
|
|
INDUSTRY_COL_CANDIDATES = ("industry", "category", "sector", "industry_name", "sw_industry",
|
|
"sw_l1", "sw_l2", "industry_l1", "board_name", "plate_name")
|
|
CODE_COL_CANDIDATES = ("ts_code", "stock_code", "code", "symbol", "sec_code", "secu_code",
|
|
"security_code", "gp_code")
|
|
|
|
_cat_cols = {"at": 0.0, "data": None}
|
|
CAT_COLS_TTL = 600.0
|
|
|
|
|
|
def category_columns(force: bool = False) -> dict:
|
|
"""`SHOW COLUMNS FROM gp_stock_category` → 认出代码列与行业列 (缓存 10 分钟)。
|
|
|
|
先认列再查, 而不是拿一堆候选列名硬撞: 2026-07-31 实测撞出来的是
|
|
`Unknown column 'stock_code'` —— 报错本身没错, 但撞法会把「列不存在」和「这只票不在
|
|
表里」混成一锅, 而且错误串长到没法看。SHOW COLUMNS 一次就说清了。
|
|
"""
|
|
import time
|
|
now = time.time()
|
|
if not force and _cat_cols["data"] is not None and now - _cat_cols["at"] < CAT_COLS_TTL:
|
|
return _cat_cols["data"]
|
|
d = {"columns": None, "code_col": None, "industry_col": None, "error": None}
|
|
try:
|
|
rows = fetch_all("SHOW COLUMNS FROM gp_stock_category")
|
|
cols = [str(r.get("Field") or r.get("field") or r.get("COLUMN_NAME") or "")
|
|
for r in (rows or [])]
|
|
cols = [c for c in cols if c]
|
|
d["columns"] = cols
|
|
low = {c.lower(): c for c in cols}
|
|
d["code_col"] = next((low[k] for k in CODE_COL_CANDIDATES if k in low), None)
|
|
d["industry_col"] = next((low[k] for k in INDUSTRY_COL_CANDIDATES if k in low), None)
|
|
if not cols:
|
|
d["error"] = "SHOW COLUMNS 回了空 —— 表可能不存在或代理不给看"
|
|
elif not d["code_col"]:
|
|
d["error"] = f"没有可识别的代码列; 实际列: {cols}"
|
|
elif not d["industry_col"]:
|
|
d["error"] = f"没有可识别的行业列; 实际列: {cols}"
|
|
except Exception as e:
|
|
d["error"] = f"{type(e).__name__}: {str(e)[:200]}"
|
|
_cat_cols.update({"at": now, "data": d})
|
|
return d
|
|
|
|
|
|
def probe_category(ts_code: str) -> dict:
|
|
"""查 gp_stock_category 并把**过程**带出来。
|
|
|
|
存在的理由: 原来的 fetch_sector_from_category 把异常 `except: continue` 吞掉,
|
|
表不存在 / 列名对不上 / 代理拒绝 一律返回 None, 跟"这只票没有行业"完全无法区分 ——
|
|
于是行业硬拦截会**静默失效**。调用方要判"数据源到底通不通", 必须看得见 error。
|
|
|
|
返回 {value, code_col, industry_col, columns, error}: value=None 且 error=None
|
|
才是真正的"表能用, 只是这只票不在里面"。
|
|
"""
|
|
meta = category_columns()
|
|
out = {"value": None, "code_col": meta.get("code_col"),
|
|
"industry_col": meta.get("industry_col"), "columns": meta.get("columns"),
|
|
"error": meta.get("error")}
|
|
if out["error"] or not out["code_col"]:
|
|
return out
|
|
col = out["code_col"]
|
|
dot = to_dot(ts_code)
|
|
forms = [dot, to_prefix(dot), dot.split(".")[0]] # 点式 / 前缀式 / 纯数字
|
|
for form in dict.fromkeys(forms):
|
|
try:
|
|
r = fetch_one(f"SELECT * FROM gp_stock_category WHERE {col} = :code LIMIT 1",
|
|
{"code": form})
|
|
except Exception as e:
|
|
out["error"] = f"{col}={form}: {type(e).__name__}: {str(e)[:160]}"
|
|
return out
|
|
if r:
|
|
out["matched_form"] = form
|
|
ind = out["industry_col"]
|
|
if ind and r.get(ind):
|
|
out["value"] = str(r[ind])
|
|
else:
|
|
for k in INDUSTRY_COL_CANDIDATES:
|
|
if r.get(k):
|
|
out["value"], out["industry_col"] = str(r[k]), k
|
|
break
|
|
return out
|
|
return out
|
|
|
|
|
|
def fetch_sector_from_category(ts_code: str):
|
|
"""行业适配器 gp_stock_category (映射表就绪前的临时数据源 —— 设计 §5)。"""
|
|
return probe_category(ts_code)["value"]
|