326 lines
16 KiB
Python
326 lines
16 KiB
Python
"""日频行业观点快照的离线单测（不连库）。
|
||
|
||
覆盖：从研判结论视图正常取数并归一（只要产业研判与环节评析两类、自我校验三值、校验问题条数、
|
||
指纹与生成日期）；视图读失败返回空列表不抛错；同一计划日重跑覆盖当日行不追加重复行；材料指纹
|
||
未变时陈旧天数累加、采信倾向的上一版值带出来；材料指纹变化时算作一次迁移且陈旧天数归零；
|
||
第一次见到的主题不判迁移；版本史里摘迁移的入口函数。
|
||
|
||
取数、读上一版、落库三处都接受注入的函数，落库这一路用一张内存里的假表接住真实的 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()
|