"""日频行业观点快照的离线单测（不连库）。 覆盖：从研判结论视图正常取数并归一（只要产业研判与环节评析两类、自我校验三值、校验问题条数、 指纹与生成日期）；视图读失败返回空列表不抛错；同一计划日重跑覆盖当日行不追加重复行；材料指纹 未变时陈旧天数累加、采信倾向的上一版值带出来；材料指纹变化时算作一次迁移且陈旧天数归零； 第一次见到的主题不判迁移；版本史里摘迁移的入口函数。 取数、读上一版、落库三处都接受注入的函数，落库这一路用一张内存里的假表接住真实的 SQL （DELETE 当日行 + 批插），所以幂等测的是真语句不是桩。 开发机没有 pandas 与数据库驱动时，只给缺席的模块装最小桩（与 test_market_context.py 同一约定： 仅在模块缺席时装桩，不覆盖真实模块）。 跑法：python3 test_judgement_snapshot.py 或 pytest test_judgement_snapshot.py """ import datetime as dt import sys import types _STUBS = ("pandas", "psycopg", "pymysql", "dotenv") for _n in _STUBS: if _n not in sys.modules: try: __import__(_n) except ImportError: _m = types.ModuleType(_n) if _n == "pandas": # db.py 的函数签名在定义时引用这两个名字 _m.DataFrame = type("DataFrame", (), {}) _m.Series = type("Series", (), {}) sys.modules[_n] = _m import config # noqa: E402 import judgement # noqa: E402 import sources # noqa: E402 def t(name, cond): assert cond, name print(" ok", name) DAY = "2026-09-03" TABLE = "t_akg_judgement_snapshot_test" # ---------------------------------------------------------------- 假数据 def _view_rows(fp_a="fp-a-1", leaning_a="偏多"): """研判结论视图的两行产业研判、一行环节评析，外加两行该被挡掉的（个股评析、簇键为空）。""" return [ {"scope": "industry", "subject_name": "液冷散热", "segment_name": None, "cluster_key": "Industry::液冷散热", "leaning": leaning_a, "n_bull": 4, "n_bear": 2, "n_flags": 1, "verified": True, "verify_problems": [], "n_materials": 37, "model": "m1", "rounds": 3, "review_date": dt.date(2026, 8, 19), "reviewed_at": dt.datetime(2026, 8, 19, 10, 30, 5), "input_version": fp_a, "review_id": "rv1"}, {"scope": "industry", "subject_name": "固态电池", "segment_name": None, "cluster_key": "Industry::固态电池", "leaning": "证据不足", "n_bull": "1", "n_bear": 0, "n_flags": 0, "verified": "false", "verify_problems": '["缺少产能口径", "价格来源不一致"]', "n_materials": 12, "model": "m1", "rounds": 3, "review_date": "2026-08-11", "reviewed_at": "2026-08-11T09:00:00", "input_version": "fp-b-1", "review_id": "rv2"}, {"scope": "segment", "subject_name": "正极材料", "segment_name": "正极材料", "cluster_key": "Segment:正极材料", "leaning": "中性", "n_bull": 2, "n_bear": 2, "n_flags": None, "verified": None, "verify_problems": None, "n_materials": 8, "model": "m1", "rounds": 2, "review_date": "2026-08-20", "reviewed_at": None, "input_version": None, "review_id": "rv3"}, {"scope": "industry", "subject_name": None, "segment_name": None, "cluster_key": "Industry::", "leaning": "偏多", "n_bull": 1, "n_bear": 0, "n_flags": 0, "verified": True, "verify_problems": None, "n_materials": 3, "model": "m1", "rounds": 3, "review_date": "2026-08-01", "reviewed_at": None, "input_version": "fp-x", "review_id": "rv4"}, ] def _pg_ok(sql, params=None): s = " ".join(sql.split()) assert "v_factor_judgement" in s, s assert set(params) == {"industry", "segment"}, params # 只要这两类 return _view_rows() def _boom(*a, **k): raise OSError("connection refused") class _FakeCursor: """只认本模块会发的三种语句的假游标：建表、按计划日删、批插。""" def __init__(self, store): self.store = store def __enter__(self): return self def __exit__(self, *a): return False def execute(self, sql, params=None): s = " ".join(sql.split()) if s.upper().startswith("CREATE TABLE"): self.store["created"] += 1 assert TABLE in s, s return if s.upper().startswith("DELETE"): assert "WHERE plan_date = %s" in s, s day = params[0] self.store["deleted"].append(day) self.store["rows"] = [r for r in self.store["rows"] if r[0] != day] return raise AssertionError(f"意外的语句: {s}") def executemany(self, sql, rows): s = " ".join(sql.split()) assert s.upper().startswith("INSERT INTO") and TABLE in s, s assert s.count("%s") == len(judgement.COLUMNS), s for r in rows: assert len(r) == len(judgement.COLUMNS) self.store["rows"].extend(list(rows)) class _FakeConn: def __init__(self, store): self.store = store def __enter__(self): return self def __exit__(self, *a): return False def cursor(self): return _FakeCursor(self.store) def commit(self): self.store["commits"] += 1 def _fake_store(): return {"rows": [], "deleted": [], "created": 0, "commits": 0} def _as_dicts(store): return [dict(zip(judgement.COLUMNS, r)) for r in store["rows"]] # ---------------------------------------------------------------- 用例 def test_fetch(): config.JUDGEMENT_SCOPES = {"industry", "segment"} rows = sources.judgement_rows(read_pg=_pg_ok) t("只出簇键与主题名齐全的行（空主题那行被挡掉）", len(rows) == 3) a = next(r for r in rows if r["cluster_key"] == "Industry::液冷散热") t("产业研判字段齐：采信倾向、多空条数、材料指纹、生成日期", a["scope"] == "industry" and a["subject_name"] == "液冷散热" and a["leaning"] == "偏多" and a["n_bull"] == 4 and a["n_bear"] == 2 and a["input_version"] == "fp-a-1" and a["review_date"] == "2026-08-19" and a["reviewed_at"] == "2026-08-19 10:30:05") t("自我校验为真、校验问题零条", a["verified"] is True and a["n_verify_problems"] == 0) b = next(r for r in rows if r["cluster_key"] == "Industry::固态电池") t("字符串形态的数字与真假值都归一、JSON 串的校验问题按条数", b["n_bull"] == 1 and b["verified"] is False and b["n_verify_problems"] == 2 and b["reviewed_at"] == "2026-08-11 09:00:00") c = next(r for r in rows if r["scope"] == "segment") t("环节评析带环节名；空的自我校验保持为空不当成假", c["segment_name"] == "正极材料" and c["verified"] is None and c["n_verify_problems"] is None and c["input_version"] is None) t("scope 传参可覆盖旋钮", sources.judgement_rows(scopes=("industry", "segment"), read_pg=_pg_ok) == rows) def test_fetch_failure(): t("视图读失败返回空列表不抛错", sources.judgement_rows(read_pg=_boom) == []) config.JUDGEMENT_SCOPES = set() t("一类都不抄时不查库", sources.judgement_rows(read_pg=_boom) == []) config.JUDGEMENT_SCOPES = {"industry", "segment"} def test_prev_read_failure(): config.JUDGEMENT_SNAPSHOT_TABLE = TABLE t("表还不存在时上一版为空字典、不抛错", judgement.load_previous(DAY, read_mysql=_boom) == {}) t("计划日不合法时也只是没有上一版", judgement.load_previous("不是日期", read_mysql=_boom) == {}) def test_same_day_idempotent(): config.JUDGEMENT_SNAPSHOT_TABLE = TABLE store = _fake_store() def _write(day, rows): judgement.save(day, rows, conn_factory=lambda: _FakeConn(store)) def _fetch(): return sources.judgement_rows(read_pg=_pg_ok) r1 = judgement.snapshot(DAY, fetch=_fetch, load_prev=lambda d: {}, write=_write) t("首日写三行、每个主题一行", r1["rows"] == 3 and len(store["rows"]) == 3) t("按 scope 计数：产业研判两个、环节评析一个", r1["by_scope"] == {"industry": 2, "segment": 1}) judgement.snapshot(DAY, fetch=_fetch, load_prev=lambda d: {}, write=_write) t("同一计划日重跑仍是三行，不追加重复行", len(store["rows"]) == 3) t("重跑先删当日行（删的正是这个计划日）", store["deleted"] == [DAY, DAY]) keys = [r["cluster_key"] for r in _as_dicts(store)] t("三个簇键各一行、无重复", sorted(keys) == sorted(set(keys)) and len(keys) == 3) t("每行都带计划日与写入时刻", all(r["plan_date"] == DAY and len(r["snapshot_at"] or "") == 19 for r in _as_dicts(store))) t("首次见到不判迁移、陈旧天数从零起算", all(r["migrated"] is None and r["stale_days"] == 0 for r in _as_dicts(store)) and r1["first_seen"] == 3 and r1["migrated"] == 0) # 视图这天一行都读不到：当日行照样先删，不留上一次重跑的残行。 judgement.snapshot(DAY, fetch=lambda: [], load_prev=lambda d: {}, write=_write) t("视图无行时当日行被清空，不留残行", store["rows"] == []) def test_stale_days_accumulate(): """指纹未变：陈旧天数按距上一个计划日的自然日数累加，采信倾向的上一版值带出来。""" prev = {"Industry::液冷散热": {"plan_date": "2026-09-01", "cluster_key": "Industry::液冷散热", "leaning": "偏多", "input_version": "fp-a-1", "stale_days": 5}} rows = judgement.build_rows(DAY, sources.judgement_rows(read_pg=_pg_ok), prev, now="x") a = next(r for r in rows if r["cluster_key"] == "Industry::液冷散热") t("指纹未变：不算迁移", a["migrated"] == 0) t("陈旧天数 = 上一版 5 天 + 两个计划日相隔 2 天 = 7", a["stale_days"] == 7) t("上一版采信倾向带出来", a["leaning_prev"] == "偏多" and a["leaning"] == "偏多") other = next(r for r in rows if r["cluster_key"] == "Industry::固态电池") t("上一版里没有的主题按第一次见到", other["migrated"] is None and other["stale_days"] == 0) # 指纹本身缺失（环节评析那行没有指纹）：不判迁移，但日子照样变老。 prev2 = {"Segment:正极材料": {"plan_date": "2026-09-02", "leaning": "中性", "input_version": None, "stale_days": 3}} rows = judgement.build_rows(DAY, sources.judgement_rows(read_pg=_pg_ok), prev2, now="x") seg = next(r for r in rows if r["cluster_key"] == "Segment:正极材料") t("指纹缺失：不判迁移，陈旧天数照常累加", seg["migrated"] is None and seg["stale_days"] == 4) # 计划日之间隔了一整个周末也照样按自然日数累加。 prev3 = {"Industry::液冷散热": {"plan_date": "2026-08-28", "leaning": "偏多", "input_version": "fp-a-1", "stale_days": 0}} rows = judgement.build_rows("2026-08-31", sources.judgement_rows(read_pg=_pg_ok), prev3, now="x") a = next(r for r in rows if r["cluster_key"] == "Industry::液冷散热") t("跨周末按自然日数累加 3 天", a["stale_days"] == 3) def test_migration_on_fingerprint_change(): """指纹变化才算一次迁移：陈旧天数归零，上一版采信倾向留在 leaning_prev 里。""" prev = {"Industry::液冷散热": {"plan_date": "2026-09-02", "cluster_key": "Industry::液冷散热", "leaning": "偏多", "input_version": "fp-a-1", "stale_days": 9}} def _fetch_changed(sql=None, params=None): return _view_rows(fp_a="fp-a-2", leaning_a="偏空") rows = judgement.build_rows(DAY, sources.judgement_rows(read_pg=_fetch_changed), prev, now="x") a = next(r for r in rows if r["cluster_key"] == "Industry::液冷散热") t("指纹变化：算一次迁移", a["migrated"] == 1) t("迁移时陈旧天数归零", a["stale_days"] == 0) t("采信倾向从偏多变成偏空，两版都在行上", a["leaning_prev"] == "偏多" and a["leaning"] == "偏空") # 同一份材料指纹没变、只是采信倾向的文本被重新生成时，不算迁移（口径就是只认指纹）。 def _fetch_same_fp(sql=None, params=None): return _view_rows(fp_a="fp-a-1", leaning_a="偏空") rows = judgement.build_rows(DAY, sources.judgement_rows(read_pg=_fetch_same_fp), prev, now="x") a = next(r for r in rows if r["cluster_key"] == "Industry::液冷散热") t("指纹没变就不算迁移，哪怕采信倾向的文本变了", a["migrated"] == 0 and a["stale_days"] == 10) hist = [{"plan_date": "2026-09-01", "migrated": None, "leaning": "偏多", "leaning_prev": None, "input_version": "fp-a-1", "review_date": "2026-08-19"}, {"plan_date": "2026-09-02", "migrated": 0, "leaning": "偏多", "leaning_prev": "偏多", "input_version": "fp-a-1", "review_date": "2026-08-19"}, {"plan_date": DAY, "migrated": 1, "leaning": "偏空", "leaning_prev": "偏多", "input_version": "fp-a-2", "review_date": "2026-09-02"}] got = judgement.migrations(hist) t("版本史里只摘出被判为迁移的那一天，带方向与出处", len(got) == 1 and got[0]["plan_date"] == DAY and got[0]["leaning_from"] == "偏多" and got[0]["leaning_to"] == "偏空" and got[0]["review_date"] == "2026-09-02") t("空版本史不报错", judgement.migrations([]) == [] and judgement.migrations(None) == []) def test_recent_rows_and_holiday_guard(): print("近日行读取与非交易日守卫（审查 2026-09-07 第 9、10 条）") rows = [ {"plan_date": "2026-09-01", "cluster_key": "Segment::甲", "leaning": "偏多", "migrated": 0}, {"plan_date": "2026-09-02", "cluster_key": "Segment::甲", "leaning": "偏空", "migrated": 1, "leaning_prev": "偏多"}, {"plan_date": "2026-09-03", "cluster_key": "Segment::甲", "leaning": "偏空", "migrated": 0}, {"plan_date": "2026-09-03", "cluster_key": "Segment::乙", "leaning": "偏多", "migrated": None, "n_bull": float("nan")}, ] got = judgement.recent_rows("2026-09-04", days=2, read_mysql=lambda src, sql, params: rows) t("按簇索引、只留最近 days 行、按日升序", [r["plan_date"] for r in got["Segment::甲"]] == ["2026-09-02", "2026-09-03"]) t("NaN 归一成 None", got["Segment::乙"][0]["n_bull"] is None) t("读失败返回空字典不抛错", judgement.recent_rows("2026-09-04", read_mysql=_boom) == {}) t("日期认不出返回空字典", judgement.recent_rows("不是日期", read_mysql=lambda *a, **k: rows) == {}) wrote = [] r = judgement.snapshot(None, fetch=lambda: [], load_prev=lambda d: {}, write=lambda d, rs: wrote.append((d, rs)), calendar=["2000-01-03"]) t("今天不在交易日历里 -> 不落行、返回跳过", r.get("skipped") == "非交易日" and wrote == []) import datetime as _dt today = _dt.date.today().isoformat() r = judgement.snapshot(None, fetch=lambda: [], load_prev=lambda d: {}, write=lambda d, rs: wrote.append((d, rs)), calendar=["2000-01-03", today]) t("今天是交易日 -> 照常落行", r.get("rows") == 0 and wrote and wrote[0][0] == today) wrote.clear() r = judgement.snapshot(None, fetch=lambda: [], load_prev=lambda d: {}, write=lambda d, rs: wrote.append((d, rs)), calendar=[]) t("日历读不到 -> 不拦，按今天落行（不让守卫本身成为断产点）", wrote and wrote[0][0] == today) wrote.clear() r = judgement.snapshot("2026-09-01", fetch=lambda: [], load_prev=lambda d: {}, write=lambda d, rs: wrote.append((d, rs)), calendar=["2000-01-03"]) t("显式给了计划日 -> 不查日历，照写", wrote and wrote[0][0] == "2026-09-01") def main(): test_fetch() test_recent_rows_and_holiday_guard() test_fetch_failure() test_prev_read_failure() test_same_day_idempotent() test_stale_days_accumulate() test_migration_on_fingerprint_change() print("ALL OK — 行业观点快照：取数归一 / 读失败为空 / 同日重跑幂等 / 陈旧天数累加 / " "指纹变化算迁移 / 版本史摘迁移 全部通过") if __name__ == "__main__": main()