"""card.judge() 纯逻辑单测（无需 DB、无需 pandas）。 覆盖（2026-09-03 台账 013 的关注定义细化）：候选 / 关注只剩"确认线缺失或陈旧"两种 / 只差覆盖与潜在吸筹归仅展示 / 仅展示（未启动、硬风险否决）/ 缺失文案与判决依据 / 因果论断证据线只展示不进判决 / 论断质量标注（陈旧、单一来源、疑似误抽、未结分歧， 四项都只提示不作门槛）/ 旋钮 / 卡内序 / 坏信号集合与 pool 同源。 跑法：python3 test_card.py 或 pytest test_card.py """ import card BASE = dict(pointed=True, theme="散热器件", n_sources=2, pct0=4.1, covered=True, upside=0.25, risk_name=False, accum_state="明确吸筹·量在价先", accum_score=78, accum_age=3, y_signal="WATCH", stale_snapshot=False) CLAIM = {"direction": "利好", "mechanism": "液冷散热渗透率提升带动订单", "horizon": "半年内", "condition": None, "doc_title": "散热行业深度", "disclosure_date": "2026-08-20", "claim_id": "clm_001"} def t(name, cond): assert cond, name print(" ok", name) def main(): r = card.judge(dict(BASE)) t("三门槛全过且明确吸筹 -> 候选", r["verdict"] == "候选" and r["confirm"] and not r["risk"]) t("候选理由四条齐", len(r["reasons"]) == 4 and "已启动" in r["reasons"][1]) t("候选的判决依据一句话", "明确吸筹" in r["basis"] and r["failed_gates"] == []) t("无论断时 logic 为空列表", r["logic"] == []) # ---- 关注只剩两种：确认线缺失（无评分）、确认线陈旧（明确吸筹但日龄超限或未知）---- r = card.judge({**BASE, "accum_state": ""}) t("缺评分 -> 仍关注且缺失文案", r["verdict"] == "关注" and "还没有吸筹评分" in r["missing"][0]) t("关注的判决依据写明判断不了、交人裁决", "判断不了" in r["basis"] and "交人裁决" in r["basis"]) r = card.judge({**BASE, "accum_age": 45}) t("评分陈旧 -> 仍关注、不确认、缺失文案", r["verdict"] == "关注" and not r["confirm"] and any("45 个交易日前" in m for m in r["missing"])) r = card.judge({**BASE, "accum_age": None}) t("明确吸筹但日龄未知 -> 关注（陈旧一类）", r["verdict"] == "关注" and any("有多旧算不出来" in m for m in r["missing"])) # ---- 只差覆盖与潜在吸筹不再升格关注（台账 013）---- r = card.judge({**BASE, "covered": False, "upside": None}) # 「不升格关注、归仅展示」是设计规则，不是给人看的话，2026-09-04 从文案里去掉了。 # 所以这里钉的是判决本身与缺失项的实质内容，不再钉那句解释规则的话。 t("只差覆盖 -> 仅展示，缺失项写明没有券商覆盖", r["verdict"] == "仅展示" and "covered" in r["failed_gates"] and "没有券商覆盖" in r["missing"]) r = card.judge({**BASE, "accum_state": "潜在吸筹·低位企稳"}) t("潜在吸筹 -> 仅展示，failed_gates 附加 confirm", r["verdict"] == "仅展示" and not r["confirm"] and "confirm" in r["failed_gates"] and any("潜在吸筹" in m and "没到明确吸筹这条线" in m for m in r["missing"])) t("潜在吸筹的判决依据写明没到明确吸筹这条线", "没到明确吸筹这条线" in r["basis"]) r = card.judge({**BASE, "accum_state": "信号不明"}) t("信号不明等非明确状态 -> 仅展示", r["verdict"] == "仅展示" and "confirm" in r["failed_gates"]) r = card.judge({**BASE, "upside": -0.05}) t("负容忍默认 0：轻微为负不过覆盖门槛 -> 仅展示", r["verdict"] == "仅展示" and "covered" in r["failed_gates"] and any("预期空间 -5%" in m and "容忍线" in m for m in r["missing"])) # ---- 其余仅展示与硬风险 ---- r = card.judge({**BASE, "pct0": 1.2}) t("未启动 -> 仅展示，判决依据带上实际涨幅与门槛值", r["verdict"] == "仅展示" and "started" in r["failed_gates"] and "当日涨幅 +1.2%" in r["basis"] and "3% 的启动线" in r["basis"]) r = card.judge({**BASE, "pointed": False}) t("未被指向 -> 仅展示且标无传导", r["verdict"] == "仅展示" and "所在环节当日没有被传导指向" in r["missing"]) r = card.judge({**BASE, "y_signal": "sell"}) t("昨夜卖出信号 -> 仅展示，信号代号翻成中文", r["verdict"] == "仅展示" and r["risk"] and r["basis"] == "择时决策系统昨夜给出卖出信号") r = card.judge({**BASE, "stale_snapshot": True}) t("快照日不符 -> 硬风险", r["verdict"] == "仅展示" and "用的传导快照不是计划当天的" in r["risk"]) r = card.judge({**BASE, "accum_state": "⚠️ 高位派发"}) t("高位派发 -> 硬风险，且不重复写进缺失", r["verdict"] == "仅展示" and "资金在高位派发" in r["risk"] and not any("派发" in m for m in r["missing"])) r = card.judge({**BASE, "risk_name": True}) t("ST 族 -> 仅展示", r["verdict"] == "仅展示" and "clean_name" in r["failed_gates"]) # ---- 因果论断：只展示，不改判决 ---- r = card.judge({**BASE, "logic": [CLAIM, {**CLAIM, "claim_id": "clm_002", "doc_title": None}]}) t("带论断仍是候选，logic 两行带出处", r["verdict"] == "候选" and len(r["logic"]) == 2 and "《散热行业深度》" in r["logic"][0] and "2026-08-20" in r["logic"][0] and "clm_001" in r["logic"][0]) t("论断不进理由", len(r["reasons"]) == 4) r = card.judge({**BASE, "pct0": 1.2, "logic": [CLAIM]}) t("论断不能把未启动拉成候选", r["verdict"] == "仅展示" and r["logic"]) t("坏论断条目被跳过", card.logic_lines([None, "x", {}]) == ["因果论断"]) # ---- 旋钮 ---- r = card.judge({**BASE, "pct0": 2.5}, start_pct=2.0) t("启动阈值旋钮生效", r["verdict"] == "候选") r = card.judge({**BASE, "accum_age": 45}, accum_max_age=60) t("评分日龄旋钮生效", r["verdict"] == "候选") r = card.judge({**BASE, "upside": -0.05}, neg_tol=0.10) t("负容忍线传入生效", r["verdict"] == "候选") rows = [{"verdict": "关注", "pct0": 9.0}, {"verdict": "候选", "pct0": 3.5}, {"verdict": "候选", "pct0": 7.2}, {"verdict": "仅展示", "pct0": None}] rows.sort(key=card.sort_key) t("卡内序：候选在前、同判决涨幅降序、缺行情垫底", [x["pct0"] for x in rows] == [7.2, 3.5, 9.0, None]) # ---- 论断质量标注（只展示、只提示，不进判决）---- # 质量画像由取数层在截断之前算好挂在每条论断上，这里按那个形状造数据。 def q(**kw): base = dict(n=1, age_days=10, latest_date="2026-08-25", n_docs=1, n_flagged=0, n_disputed=0, n_via_segment=0) base.update(kw) return base def claims(n, quality): return [{**CLAIM, "quality": quality} for _ in range(n)] r = card.judge({**BASE, "logic": claims(1, q())}) t("论断新鲜且单条：不产生任何质量提示", r["verdict"] == "候选" and not [m for m in r["missing"] if "论断" in m]) t("质量画像随判决一起出", r["logic_quality"]["age_days"] == 10) r = card.judge({**BASE, "logic": claims(1, q(age_days=200, latest_date="2026-02-16"))}) t("论断超陈旧线：写明过了多少天与最新一条的日期", "研报论断已过 200 天（最新一条 2026-02-16）" in r["missing"]) t("论断陈旧不改判决（只提示不作门槛）", r["verdict"] == "候选") r = card.judge({**BASE, "logic": claims(1, q(age_days=91))}, logic_stale_days=120) t("陈旧线旋钮生效：放宽到 120 天后 91 天不再提示", not [m for m in r["missing"] if "研报论断已过" in m]) r = card.judge({**BASE, "logic": claims(1, q(age_days=90))}) t("边界：正好等于陈旧线不算陈旧（超过才提示）", not [m for m in r["missing"] if "研报论断已过" in m]) r = card.judge({**BASE, "logic": claims(3, q(n=3, n_docs=1))}) t("多条论断出自同一份研报：单一来源提示", "3 条论断全部出自同一份研报" in r["missing"]) r = card.judge({**BASE, "logic": claims(3, q(n=3, n_docs=2))}) t("来源多于一份就不提示", not [m for m in r["missing"] if "同一份研报" in m]) r = card.judge({**BASE, "logic": claims(1, q(n=1, n_docs=1))}) t("只有一条论断时不算单一来源（那是条数少不是来源集中）", not [m for m in r["missing"] if "同一份研报" in m]) r = card.judge({**BASE, "logic": claims(4, q(n=4, n_docs=2, n_flagged=2, n_disputed=1))}) t("疑似误抽与未结分歧各写一条", "其中 2 条被标为疑似误抽" in r["missing"] and "其中 1 条处于未结的方向分歧" in r["missing"]) r = card.judge(dict(BASE)) t("没有论断时不产生质量提示、画像为空", r["logic_quality"] is None and not [m for m in r["missing"] if "论断" in m]) r = card.judge({**BASE, "logic": [CLAIM]}) t("上游没给质量画像时不报错、也不瞎提示", r["logic_quality"] is None and not [m for m in r["missing"] if "论断" in m]) t("论断为空列表时不产生提示", card.logic_notes([]) == [] and card.logic_quality([]) is None) # ---- 启动涨幅线的整条开关（2026-09-04 拍板关闭）---- r = card.judge({**BASE, "pct0": -3.2}, require_started=False) t("门槛关掉后，当日下跌也能判候选", r["verdict"] == "候选" and r["failed_gates"] == []) t("关掉后理由里写明这条线已关，不再说「当日已启动」", any("启动线已关闭" in x for x in r["reasons"]) and not any("当日已启动" in x for x in r["reasons"])) r = card.judge({**BASE, "pct0": None}, require_started=False) t("关掉后连没有当日行情也不拦（这是整条关掉，不是把阈值调成负数）", r["verdict"] == "候选") t("但没有行情这件事仍写进缺失项", "没有当日行情" in r["missing"]) # 关掉这一条不许顺带放松别的门槛与风险 r = card.judge({**BASE, "pct0": -3.2, "pointed": False, "theme": None, "n_sources": None}, require_started=False) t("关掉启动线不影响传导门槛", r["verdict"] == "仅展示" and "pointed" in r["failed_gates"]) r = card.judge({**BASE, "pct0": -3.2, "covered": False, "upside": None}, require_started=False) t("关掉启动线不影响覆盖门槛", r["verdict"] == "仅展示" and "covered" in r["failed_gates"]) r = card.judge({**BASE, "pct0": -3.2, "accum_state": "潜在吸筹"}, require_started=False) t("关掉启动线不影响吸筹确认线", r["verdict"] == "仅展示" and not r["confirm"]) r = card.judge({**BASE, "pct0": -3.2, "y_signal": "SELL"}, require_started=False) t("关掉启动线不影响硬风险否决", r["verdict"] == "仅展示" and r["risk"]) r = card.judge({**BASE, "pct0": -3.2, "accum_state": ""}, require_started=False) t("关掉启动线后，判断不了的仍是关注", r["verdict"] == "关注") r = card.judge({**BASE, "pct0": -3.2}) t("默认仍是开着的（改口径要靠配置，不靠默认值悄悄变）", r["verdict"] == "仅展示" and "started" in r["failed_gates"]) # ---- 给人看的文字里不许出现代码名（2026-09-04）---- # 起因：页面上出现过「选股系统判为仅展示，不出提议：门槛未过：started」。 # started 是门槛在代码里的变量名，SELL 是决策系统的枚举值，两者都只该在系统之间 # 传递。判决依据、缺失项、风险这三个字段会被 PMS 原样显示给人读，所以它们里面 # 一个代码名都不许有。failed_gates 不在此列——那是给程序判断用的机器字段。 CODE_WORDS = ("started", "pointed", "covered", "confirm", "SELL", "AVOID", "DROPPED", "BUY", "WATCH", "HOLD") probes = [ dict(BASE), # 候选 {**BASE, "accum_state": ""}, # 关注·没有评分 {**BASE, "accum_age": 99}, # 关注·评分太旧 {**BASE, "accum_age": None}, # 关注·日龄算不出 {**BASE, "accum_state": "潜在吸筹·低位企稳"}, # 仅展示·没到确认线 {**BASE, "pct0": 1.2}, # 仅展示·没启动 {**BASE, "pct0": None}, # 仅展示·没有行情 {**BASE, "pointed": False, "theme": None, "n_sources": None}, # 仅展示·没传导 {**BASE, "covered": False, "upside": None}, # 仅展示·没覆盖 {**BASE, "upside": -0.05}, # 仅展示·预期空间为负 {**BASE, "y_signal": "SELL"}, # 仅展示·昨夜卖出信号 {**BASE, "y_signal": "AVOID"}, # 仅展示·昨夜回避信号 {**BASE, "stale_snapshot": True}, # 仅展示·快照日不符 {**BASE, "accum_state": "高位派发"}, # 仅展示·资金派发 {**BASE, "pointed": False, "pct0": None, "covered": False, "upside": None, "accum_state": ""}, # 全都不满足 ] for ev in probes: r = card.judge(ev) texts = [r["basis"], *r["missing"], *r["risk"], *r["reasons"]] for txt in texts: for w in CODE_WORDS: assert w not in txt, f"给人看的文字里出现了代码名 {w}：{txt}" assert r["basis"] and len(r["basis"]) >= 6, f"判决依据太短，说不清原因：{r['basis']!r}" t(f"给人看的三个字段里没有代码名（{len(probes)} 种情形逐个扫过）", True) t("门槛翻译带上实际数字", card.gate_reasons(["started"], pct0=-3.21, start_pct=3.0)[0] == "当日涨幅 -3.2%，没到 3% 的启动线") t("认不出的门槛名原样带上，不吞掉", card.gate_reasons(["新门槛"]) == ["门槛 新门槛 未过"]) t("坏信号集合口径", card.BAD_SIGNALS == {"SELL", "AVOID", "DROPPED"}) print("ALL OK — 候选卡判决 / 关注两种（缺失、陈旧）/ 只差覆盖与潜在吸筹归仅展示 / 硬风险三种 / " "因果论断只展示 / 论断质量标注四项 / 缺失文案与依据 / 旋钮 / 卡内序 全部通过") if __name__ == "__main__": main()