From baaf5dfc01ff1f4d6a58b9a57280996fb277bc3a Mon Sep 17 00:00:00 2001 From: zlt Date: Tue, 18 Aug 2026 16:01:40 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A4=84=E7=90=86pms=E7=B3=BB=E7=BB=9F?= =?UTF-8?q?=E5=AF=B9=E6=8E=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- api.py | 39 +++++++++++++ plan_reconcile.py | 125 +++++++++++++++++++++++++++++++----------- test_plan_verdict.py | 127 +++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 260 insertions(+), 31 deletions(-) create mode 100644 test_plan_verdict.py diff --git a/api.py b/api.py index 6177670..9bd7f8b 100644 --- a/api.py +++ b/api.py @@ -19,6 +19,7 @@ from fastapi.responses import PlainTextResponse import db import plan +import plan_reconcile from xxl import router as xxl_router app = FastAPI(title="akg-factor-bridge · 每日选股计划", version="0.1") @@ -65,3 +66,41 @@ def refresh(date: str | None = None): except SystemExit as e: raise HTTPException(status_code=404, detail=str(e)) return {"ok": True, "file": out} + + +@app.get("/plan/verdict") +def plan_verdict(codes: str | None = None, code: str | None = None, + date: str | None = None): + """逐票『计划判决』（只读）——今日页 / 机会线索页对齐用。 + + GET /plan/verdict?codes=300750,600438,SH688041 多只（逗号分隔） + GET /plan/verdict?code=300750&date=2026-08-04 单只 + 指定档位日 + + 每只票返回 decision（main 主榜 / observe 观察档 / reject 不采纳 / absent 无此票）、 + 与命令行 plan_reconcile 完全同口径的 verdict_text（页面直接展示的一行解释）， + 以及 score/rank/tier/upside/热度/传导/赛道/图谱证据等明细。 + date 缺省=当日档位日。三种代码形态都收（600000.SH / SH600000 / 600000）。纯 SELECT，不写任何库。 + """ + raw = (codes or code or "").strip() + want = [c.strip() for c in raw.split(",") if c.strip()] + if not want: + raise HTTPException(status_code=400, detail="缺少 code / codes 参数") + d = date or plan_reconcile.latest_date() + if not d: + raise HTTPException(status_code=404, detail="档位表为空——先跑当日构建") + try: + L = plan_reconcile._load(d) # noqa: SLF001 —— 桥内自用只读加载 + except Exception as e: # noqa: BLE001 —— 数据层异常统一收成 500 + raise HTTPException(status_code=500, detail=f"加载档位数据失败: {e!r}") + verdicts = [] + for c in want: + try: + k = plan_reconcile._norm_code(c) # noqa: SLF001 + except SystemExit as e: # _norm_code 认不出的形态会 raise SystemExit + verdicts.append({"input": c, "error": str(e)}) + continue + v = plan_reconcile.verdict(k, L) + v["input"] = c + verdicts.append(v) + return {"date": d, "stale": L.get("stale", ""), + "count": len(verdicts), "verdicts": verdicts} diff --git a/plan_reconcile.py b/plan_reconcile.py index 6c0e0d6..0c10ef4 100644 --- a/plan_reconcile.py +++ b/plan_reconcile.py @@ -133,7 +133,22 @@ def _fmt(v, pat="{:.2f}"): return "—" if v is None else pat.format(v) -def explain(k: str, L: dict) -> None: +def latest_date() -> str | None: + """当日档位日 = t_factor_akg_gate 的 MAX(trade_date)；空表返回 None。只读。""" + row = db.read_mysql("factor", "SELECT MAX(trade_date) d FROM t_factor_akg_gate") + v = None if row.empty else row.iloc[0, 0] + if v is None or pd.isna(v): + return None + return pd.Timestamp(v).date().isoformat() + + +def verdict(k: str, L: dict) -> dict: + """把 explain() 的『判决』拆成结构化只读结果（供 /plan/verdict、今日页调用）。 + + 不打印、不写库。`verdict_text` 与命令行 `explain()`『▶ 判决：』之后的文案 + 逐字一致——页面上的一行解释与命令行对账因此完全同口径（explain 现在也复用本函数）。 + decision 取值：main（主榜）/ observe（观察档）/ reject（不采纳，gate=0）/ absent（当日档位表无此票）。 + """ nm = L["names"].get(k, "") g = L["gate"].get(k) up = L["upside"].get(k) @@ -145,6 +160,83 @@ def explain(k: str, L: dict) -> None: e = L["ev"].get(k, {"segs": set(), "chains": set()}) risk = bool(_RISK_RE.match(nm.replace(" ", ""))) if nm else False + decision = tier = reason = None + rank = rank_total = None + + if g == 2.0: + decision = "main" + rank = L["rank_main"].get(k) + rank_total = len(L["main"]) + # 档界：score=200+档×20+组内分(±9.9) → 强≥230.1、弱≥210.1、无≤209.9 + tier = "强传导" if sc and sc >= 230 else ("弱传导" if sc and sc >= 210 else "无传导") + text = (f"主榜第 {rank}/{rank_total} 名（score={sc:.1f}，{tier}档）" + + ("——计划默认只显示前 20，名次靠后不等于不在计划里" + if rank and rank > 20 else "")) + elif g == 1.0: + decision = "observe" + rank = L["rank_obs"].get(k) + rank_total = len(L["obs"]) + text = (f"观察档第 {rank}/{rank_total} 名（score={sc:.1f}，" + f"无券商覆盖、低置信）") + elif g == 0.0: + decision = "reject" + import config as _cfg + _tol = float(getattr(_cfg, "UPSIDE_NEG_TOLERANCE", 0.0) or 0.0) + if risk: + reason = "重大风险闸（ST/退市族），07-31 拍板默认杜绝" + elif up is not None and up < -_tol: + reason = (f"有覆盖但 upside={up:+.1%} < {-_tol:+.0%}——贵了不买是绝对下限" + + ("" if _tol == 0 else f"（负容忍 {_tol:.0%} 已启用仍不够）")) + elif up is not None and hits: + reason = "覆盖、upside、赛道三者都过——按机制不该是 0，把本行发我核（疑似口径错位）" + elif up is not None: + reason = ("有覆盖、upside≥0，但不在十条赛道（赛道闸拦下）" + + ("——图上有环节证据，属映射够不着的错杀池，看首批锚能否接住" + if e["segs"] else "")) + elif not hits and (trv or 0) <= 0: + reason = "无覆盖、无传导、不在赛道——两锚皆无" + else: + reason = "无覆盖且当日不在传导链上（观察档条件差最后一步）" + text = f"不采纳（gate=0）。原因：{reason}" + else: + decision = "absent" + text = ("当日档位表无此票" + + ("（可能不在覆盖池 universe，或当日未出行）" if not hits + else "——在赛道却无档位行，把本行发我核")) + + return { + "code": k, + "name": nm, + "gate": g, + "decision": decision, + "verdict_text": text, + "reason": reason, + "score": (round(float(sc), 1) if sc is not None else None), + "rank": rank, + "rank_total": rank_total, + "tier": tier, + "upside": (round(float(up), 4) if up is not None else None), + "heat": (round(float(ht), 2) if ht is not None else None), + "transmission": (round(float(trv), 2) if trv is not None else None), + "tracks": [{"track": t, "key": key, "rule": rule} + for t, key, rule in hits], + "transmission_targets": list(paths), + "graph_segments": sorted(e["segs"]), + "graph_chains": sorted(e["chains"]), + "risk": risk, + } + + +def explain(k: str, L: dict) -> None: + """CLI 打印：明细（赛道/覆盖/传导/图谱）+ 判决行。判决行复用 verdict()，与 API 同口径。""" + nm = L["names"].get(k, "") + up = L["upside"].get(k) + ht = L["heat"].get(k) + trv = L["tr"].get(k) + hits = L["hits"].get(k, []) + paths = L["tmap"].get(k, []) + e = L["ev"].get(k, {"segs": set(), "chains": set()}) + print(f"\n◆ {k} {nm}") if hits: shown = "；".join(f"{t}←{key}({rule})" for t, key, rule in hits[:4]) @@ -160,36 +252,7 @@ def explain(k: str, L: dict) -> None: + ("…" if len(e["segs"]) > 4 else "") + "）" + (f"，链名 {'、'.join(sorted(e['chains'])[:4])}" if e["chains"] else "，边上无链名")) - if g == 2.0: - n = L["rank_main"].get(k) - # 档界：score=200+档×20+组内分(±9.9) → 强≥230.1、弱≥210.1、无≤209.9 - tier = "强传导" if sc and sc >= 230 else ("弱传导" if sc and sc >= 210 else "无传导") - print(f" ▶ 判决：主榜第 {n}/{len(L['main'])} 名（score={sc:.1f}，{tier}档）" - + ("——计划默认只显示前 20，名次靠后不等于不在计划里" if n and n > 20 else "")) - elif g == 1.0: - print(f" ▶ 判决：观察档第 {L['rank_obs'].get(k)}/{len(L['obs'])} 名（score={sc:.1f}，" - f"无券商覆盖、低置信）") - elif g == 0.0: - import config as _cfg - _tol = float(getattr(_cfg, "UPSIDE_NEG_TOLERANCE", 0.0) or 0.0) - if risk: - reason = "重大风险闸（ST/退市族），07-31 拍板默认杜绝" - elif up is not None and up < -_tol: - reason = (f"有覆盖但 upside={up:+.1%} < {-_tol:+.0%}——贵了不买是绝对下限" - + ("" if _tol == 0 else f"（负容忍 {_tol:.0%} 已启用仍不够）")) - elif up is not None and hits: - reason = "覆盖、upside、赛道三者都过——按机制不该是 0，把本行发我核（疑似口径错位）" - elif up is not None: - reason = ("有覆盖、upside≥0，但不在十条赛道（赛道闸拦下）" - + ("——图上有环节证据，属映射够不着的错杀池，看首批锚能否接住" if e["segs"] else "")) - elif not hits and (trv or 0) <= 0: - reason = "无覆盖、无传导、不在赛道——两锚皆无" - else: - reason = "无覆盖且当日不在传导链上（观察档条件差最后一步）" - print(f" ▶ 判决：不采纳（gate=0）。原因：{reason}") - else: - print(" ▶ 判决：当日档位表无此票" - + ("（可能不在覆盖池 universe，或当日未出行）" if not hits else "——在赛道却无档位行，把本行发我核")) + print(f" ▶ 判决：{verdict(k, L)['verdict_text']}") def audit(L: dict) -> None: diff --git a/test_plan_verdict.py b/test_plan_verdict.py new file mode 100644 index 0000000..899d9e1 --- /dev/null +++ b/test_plan_verdict.py @@ -0,0 +1,127 @@ +"""plan_reconcile.verdict() 纯逻辑单测（无需 DB / 无需 pandas 实体）。 + +verdict() 只读 L 字典、不碰库，故可离线跑通。校验四种 decision 分支 + tier 三档 ++ 前20外提示，且 verdict_text 与命令行 explain()『▶ 判决：』之后的文案逐字一致 +（页面一行解释 == 命令行对账，同口径）。 + +隔离约定：仅对"当前尚未导入"的依赖装临时桩，导入后立即还原，绝不覆盖真实模块—— +因此与 test_pool_logic.py / test_xxl_trigger.py 同进程 pytest 收集也不会互相污染。 + +跑法：python3 test_plan_verdict.py 或 pytest test_plan_verdict.py +""" +import importlib.util +import os +import sys +import types + +_NAMES = ("pandas", "common", "db", "tracks", "psycopg", "pymysql") +_HERE = os.path.dirname(os.path.abspath(__file__)) +_PR_PATH = os.path.join(_HERE, "plan_reconcile.py") + + +def _load_isolated(): + """装最小桩→加载 plan_reconcile→还原 import-only 桩。 + + config 留到 run() 结束再还原：verdict() 的 gate=0 分支在运行期才 `import config`， + 需要 UPSIDE_NEG_TOLERANCE 可读（真实环境用真 config，离线用桩）。 + """ + saved = {n: sys.modules.get(n) for n in _NAMES + ("config",)} + for n in _NAMES: + if sys.modules.get(n) is None: + sys.modules[n] = types.ModuleType(n) # 仅在缺席时装桩，不覆盖真实模块 + if sys.modules.get("config") is None: + cfg = types.ModuleType("config") + cfg.UPSIDE_NEG_TOLERANCE = 0.0 + sys.modules["config"] = cfg + spec = importlib.util.spec_from_file_location("plan_reconcile_uut", _PR_PATH) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + for n in _NAMES: # 还原 import-only 桩 + if saved[n] is None: + sys.modules.pop(n, None) + else: + sys.modules[n] = saved[n] + return mod, saved + + +def _restore_config(saved): + if saved["config"] is None: + sys.modules.pop("config", None) + else: + sys.modules["config"] = saved["config"] + + +def _L(): + """合成档位数据（主榜 30 只 / 观察档 40 只），覆盖四种判决分支。""" + return { + "names": { + "SH600519": "贵州茅台", "SH600000": "浦发银行", "SZ000002": "万科A", + "SZ000111": "*ST锐电", "SZ000625": "长安汽车", "SH601111": "中国国航", + }, + "gate": { + "SH600519": 2.0, "SH601111": 2.0, "SZ000625": 2.0, + "SH600000": 1.0, "SZ000002": 0.0, "SZ000111": 0.0, + # SZ300999 故意缺席 → absent + }, + "score": { + "SH600519": 235.0, "SH601111": 232.0, "SZ000625": 215.0, + "SH600000": 105.0, + }, + "upside": {"SH600519": 0.25, "SZ000002": -0.05}, + "heat": {"SH600519": 0.9}, + "tr": {"SH600519": 55.0}, + "hits": {"SH600519": [("光伏", "CATL", "graph_segment")]}, + "tmap": {"SH600519": ["宁德时代(源2×空间30%)"]}, + "ev": {"SH600519": {"segs": {"电池"}, "chains": {"锂电链"}}}, + "main": ["x"] * 30, + "obs": ["y"] * 40, + "rank_main": {"SH600519": 3, "SH601111": 25, "SZ000625": 5}, + "rank_obs": {"SH600000": 7}, + } + + +def test_verdict(): + pr, saved = _load_isolated() + try: + L = _L() + + v = pr.verdict("SH600519", L) # 主榜·强传导 + assert v["decision"] == "main" and v["tier"] == "强传导", v + assert v["verdict_text"] == "主榜第 3/30 名（score=235.0，强传导档）", v["verdict_text"] + assert v["score"] == 235.0 and v["rank"] == 3 and v["rank_total"] == 30, v + assert v["upside"] == 0.25 and v["tracks"][0]["rule"] == "graph_segment", v + assert v["graph_segments"] == ["电池"] and v["transmission"] == 55.0, v + + v = pr.verdict("SH601111", L) # 主榜但名次>20 → 带前20提示 + assert v["decision"] == "main" and v["tier"] == "强传导", v + assert v["verdict_text"] == ( + "主榜第 25/30 名（score=232.0，强传导档）" + "——计划默认只显示前 20，名次靠后不等于不在计划里"), v["verdict_text"] + + v = pr.verdict("SZ000625", L) # score 215 → 弱传导 + assert v["decision"] == "main" and v["tier"] == "弱传导", v + assert v["verdict_text"] == "主榜第 5/30 名（score=215.0，弱传导档）", v["verdict_text"] + + v = pr.verdict("SH600000", L) # 观察档 + assert v["decision"] == "observe", v + assert v["verdict_text"] == "观察档第 7/40 名（score=105.0，无券商覆盖、低置信）", v["verdict_text"] + + v = pr.verdict("SZ000002", L) # gate=0，有覆盖 upside<0 → 贵了不买 + assert v["decision"] == "reject", v + assert "贵了不买是绝对下限" in v["reason"], v + assert v["verdict_text"] == f"不采纳（gate=0）。原因：{v['reason']}", v["verdict_text"] + + v = pr.verdict("SZ000111", L) # gate=0，名字 *ST → 风险闸 + assert v["decision"] == "reject" and v["risk"] is True, v + assert "风险闸" in v["reason"], v + + v = pr.verdict("SZ300999", L) # 不在 gate 表 → absent + assert v["decision"] == "absent", v + assert v["verdict_text"] == "当日档位表无此票（可能不在覆盖池 universe，或当日未出行）", v["verdict_text"] + finally: + _restore_config(saved) + + +if __name__ == "__main__": + test_verdict() + print("ALL OK — verdict() 四分支 / tier 三档 / 前20提示 全部通过")