"""公司质地接入（2026-09-09 接入方案第四节）：取数、第五路信号与合成、判决收敛只降不升、卡上文字。离线，不连库。""" import datetime as dt import json import sys import types for _n in ("pandas", "pymysql", "psycopg"): if _n not in sys.modules: try: __import__(_n) except Exception: # noqa: BLE001 _m = types.ModuleType(_n) if _n == "pandas": _m.DataFrame = type("DataFrame", (), {}) sys.modules[_n] = _m import card # noqa: E402 import logic_state as ls # noqa: E402 import sources # noqa: E402 def t(name, cond): print((" ok " if cond else " FAIL") + name) assert cond, name class _DF: def __init__(self, rows): self._rows = rows def to_dict(self, _kind): return self._rows def _review(overall="好", doubt=False, ran="2026-09-08", groups=None): return {"overall": overall, "groups": groups or {"回报与护城河": "好", "盈余质量与财务安全": "好", "成长与含金量": "好"}, "doubt_hard": doubt, "ran_at": ran, "age_days": 1, "period": "2026Q1", "grade_counts": {"好": 5, "中": 3, "差": 1, "证据不足": 1}, "valuation": "贵", "thesis": "质地好但估值贵", "confidence": "中", "invalidation": "下一期看现金流", "quality_reasons": ["回报类好"], "report_url": "http://192.168.16.178:8000/company/601126.SH/review"} def test_sources(): calls = [] def rp(sql, params=None): calls.append(params) return _DF([{"ts_code": "601126.SH", "period": "2026Q1", "review_version": "2026.09-2", "grades": json.dumps({"penman": "好"}), "summary": json.dumps({"overall": "好", "groups": {"回报与护城河": "好"}, "valuation": "贵", "doubt_hard": False, "thesis": "x", "confidence": "中", "grade_counts": {"好": 5, "中": 3, "差": 1, "证据不足": 1}, "report_path": "/company/601126.SH/review"}, ensure_ascii=False), "brain_used": True, "brain_status": "ok", "ran_at": dt.datetime(2026, 9, 8, 10, 0)}, {"ts_code": "688008.SH", "period": "2026Q1", "review_version": "2026.09-2", "grades": {"buffett": "差"}, "summary": None, "brain_used": False, "brain_status": None, "ran_at": dt.datetime(2026, 5, 1, 10, 0)}]) out = sources.company_reviews(["SH601126", "SH688008", "SZ000001"], "2026-09-09", read_pg=rp) t("按点后缀码查", calls[0][0] == ["601126.SH", "688008.SH", "000001.SZ"]) t("摘要解析与报告链接", out["SH601126"]["overall"] == "好" and out["SH601126"]["report_url"].endswith("/company/601126.SH/review")) t("运行日距数据日", out["SH601126"]["age_days"] == 1 and out["SH688008"]["age_days"] == 131) t("没摘要的老行只有评价计数", out["SH688008"]["overall"] is None and out["SH688008"]["grade_counts"]["差"] == 1) t("没报告的票不在字典里", "SZ000001" not in out) def boom(*a, **k): raise RuntimeError("no table") t("表读不到整体缺席", sources.company_reviews(["SH601126"], "2026-09-09", read_pg=boom) == {}) def test_fifth_path(): up = ls.from_company(_review("好"), "2026-09-09") t("质地好向好且带截止日", up["signal"] == ls.SIG_UP and up["as_of"] == "2026-09-08" and up["path"] == ls.PATH_COMPANY) flat = ls.from_company(_review("中"), "2026-09-09") t("质地中平稳", flat["signal"] == ls.SIG_FLAT) down = ls.from_company(_review("差", doubt=True), "2026-09-09") t("质地差且盈余差：转弱且达到存疑条件", down["signal"] == ls.SIG_DOWN and down["hard"]) soft = ls.from_company(_review("差", doubt=False), "2026-09-09") t("质地差但盈余不差：转弱不硬", soft["signal"] == ls.SIG_DOWN and not soft["hard"]) t("没报告缺失", ls.from_company(None, "2026-09-09")["signal"] == ls.SIG_NONE) stale = _review("好"); stale["age_days"] = 200 t("报告陈旧缺失", ls.from_company(stale, "2026-09-09")["signal"] == ls.SIG_NONE) t("证据不足缺失", ls.from_company(_review("证据不足"), "2026-09-09")["signal"] == ls.SIG_NONE) # 合成：质地好单独不能定态（慢信号）；配研报论断向好则强化；质地差硬转弱单独就存疑 alone = ls.compose([ls.signal(ls.PATH_CLAIM, ls.SIG_NONE, why="缺"), up]) t("只剩公司质地这一路慢信号不能单独定态", alone["state"] == ls.STATE_UNKNOWN) with_claim = ls.compose([ls.signal(ls.PATH_CLAIM, ls.SIG_UP, as_of="2026-09-08", coverage=3, why="利好"), up]) t("研报论断向好加质地好为逻辑强化", with_claim["state"] == ls.STATE_STRONG) doubt = ls.compose([ls.signal(ls.PATH_CLAIM, ls.SIG_UP, as_of="2026-09-08", coverage=3, why="利好"), down]) t("一路向好一路转弱是证据矛盾交人", doubt["state"] == ls.STATE_UNKNOWN) doubt2 = ls.compose([ls.signal(ls.PATH_CLAIM, ls.SIG_NONE, why="缺"), down]) t("质地差达到进入条件则逻辑存疑", doubt2["state"] == ls.STATE_DOUBT) t("存疑触发表里有第五路", ls.PATH_COMPANY in ls.DOUBT_TRIGGERS and ls.SPEED[ls.PATH_COMPANY] == "慢") def test_card_gate_and_views(): ev = {"pointed": True, "theme": "端侧AI芯片", "n_sources": 3, "pct0": 4.0, "covered": True, "upside": 0.2, "risk_name": False, "accum_state": "明确吸筹·量在价先", "accum_score": 80, "accum_age": 1, "y_signal": "WATCH", "stale_snapshot": False, "logic": []} base = card.judge(ev) t("底子是候选", base["verdict"] == "候选") down = card.judge(ev, company_review=_review("差", doubt=True)) t("质地差候选降为关注且依据写明", down["verdict"] == "关注" and "公司质地评为差" in down["basis"] and down["company_gate"]) t("质地中不动", card.judge(ev, company_review=_review("中"))["verdict"] == "候选") t("质地好不升档、证据不足不动", card.judge(ev, company_review=_review("好"))["verdict"] == "候选" and card.judge(ev, company_review=_review("证据不足"))["verdict"] == "候选") t("开关关掉不降", card.judge(ev, company_review=_review("差"), quality_gate=False)["verdict"] == "候选") watch_ev = {**ev, "accum_state": ""} t("本来就是关注的不再变", card.judge(watch_ev, company_review=_review("差"))["verdict"] == "关注") v = card.company_review_view(_review("好")) t("整句带质地三类、十视角计数、论点、失效条件、链接", all(x in v for x in ("质地好", "回报好", "好5/中3/差1", "买方论点", "失效条件", "报告 http"))) t("短句与分组标签", card.company_review_short(_review("好")).startswith("好·贵") and card.company_review_label(_review("证据不足")) == "无" and card.company_review_label(None) == "无" and card.company_review_view(None).startswith("公司深度：暂无")) def main(): test_sources() test_fifth_path() test_card_gate_and_views() print("ALL OK — 公司质地：取数 / 第五路 / 判决收敛 / 卡上文字") if __name__ == "__main__": main()