# -*- coding: utf-8 -*-
"""
第八批模块单测 (零外部依赖, 不连库不触网)
==========================================
运行: 在 tradingSystem 仓库根目录执行  python scripts/test_batch8_units.py
覆盖: 上游榜单的版本比对 (app/core/plan_diff.py) —— 名册与指纹、三种比对语义、
      进出榜的尾部闸、档位升降与券商覆盖翻转、名次跳变、持仓票视图。

这一批的重点不是"能不能算出差异", 而是**不报假变化**: 榜单是按 top 截断过的, 尾部的进出
是截断噪音而不是上游观点变化。假警报天天响, 提示就等于没有 —— 与 `_capped` 治的是同一个病。
"""
import os
import sys
import traceback

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from app.core import plan_diff as pd                                           # noqa: E402

RESULTS = []


def case(name):
    def deco(fn):
        RESULTS.append((name, fn))
        return fn
    return deco


# ================================================================ fixture
def _r(code, rank, score, tier="强传导", theme="储能", bucket="main", name=None):
    """名册单行 (plan_diff 内部形态)。"""
    return {"c": code, "n": name or code[:6], "r": rank, "s": score,
            "t": tier, "h": theme, "b": bucket}


def roster(*rows) -> dict:
    return {r["c"]: r for r in rows}


def plan(main=(), observe=(), date="2026-07-30", capped_main=False, capped_obs=False) -> dict:
    """plan_feed.parse_plan() 结果的最小形态。

    `returned` / `truncated` 也要带上 —— 快照落库时靠它们记条数与截断标志, 少了的话
    "上一版坏没坏"这类判据在单测里恒为假, 用例看着过了实际什么都没验。
    """
    def row(x, bucket):
        return {"ts_code": x["c"], "name": x["n"], "rank": x["r"], "score": x["s"],
                "tier": x["t"], "theme": x["h"], "bucket": bucket}
    return {"date": date,
            "main": [row(x, "main") for x in main],
            "observe": [row(x, "observe") for x in observe],
            "returned": {"main": len(main), "observe": len(observe)},
            "truncated": {"main": bool(capped_main), "observe": bool(capped_obs)}}


# ================================================================ [A] 名册与指纹
@case("[A1] roster_of 从计划结构取出名册")
def t_a1():
    ro = pd.roster_of(plan(main=[_r("600418.SH", 1, 242.2)],
                          observe=[_r("600877.SH", 1, 101.1, tier=None, bucket="observe")]))
    assert set(ro) == {"600418.SH", "600877.SH"}
    assert ro["600418.SH"]["b"] == "main" and ro["600877.SH"]["b"] == "observe"
    assert ro["600418.SH"]["t"] == "强传导" and ro["600877.SH"]["t"] is None


@case("[A2] 同一只票两档都出现时以主榜为准")
def t_a2():
    p = plan(main=[_r("600418.SH", 3, 240.0)],
             observe=[_r("600418.SH", 1, 99.0, tier=None, bucket="observe")])
    ro = pd.roster_of(p)
    assert len(ro) == 1
    assert ro["600418.SH"]["b"] == "main", "主榜在前, 观察档那条不该覆盖它"


@case("[A3] 指纹稳定: 同一名册两次算结果一致")
def t_a3():
    a = roster(_r("600000.SH", 1, 10.0), _r("600418.SH", 2, 9.0))
    b = roster(_r("600418.SH", 2, 9.0), _r("600000.SH", 1, 10.0))   # 插入次序不同
    assert pd.digest_of(a) == pd.digest_of(b), "指纹不能依赖 dict 的插入次序"


@case("[A4] 指纹不含 score —— 分数抖动但次序没变, 不算新版本")
def t_a4():
    a = roster(_r("600000.SH", 1, 242.2801))
    b = roster(_r("600000.SH", 1, 242.2799))
    assert pd.digest_of(a) == pd.digest_of(b), (
        "/plan 是实时算的, score 末位天天抖; 算进指纹会让快照表白涨而信息量为零")


@case("[A5] 指纹含 rank —— 次序变了就是新版本")
def t_a5():
    assert pd.digest_of(roster(_r("600000.SH", 1, 10.0))) != \
           pd.digest_of(roster(_r("600000.SH", 2, 10.0)))


@case("[A6] 指纹含 tier / theme / bucket")
def t_a6():
    base = roster(_r("600000.SH", 1, 10.0))
    assert pd.digest_of(base) != pd.digest_of(roster(_r("600000.SH", 1, 10.0, tier="弱传导")))
    assert pd.digest_of(base) != pd.digest_of(roster(_r("600000.SH", 1, 10.0, theme="整机制造")))
    assert pd.digest_of(base) != pd.digest_of(roster(_r("600000.SH", 1, 10.0, bucket="observe")))


@case("[A7] 名册 ↔ 落库行 往返不丢字段")
def t_a7():
    ro = roster(_r("600418.SH", 7, 233.3, tier="弱传导", theme="整车"))
    back = pd.roster_from_rows(pd.roster_rows(ro))
    assert back == ro
    assert pd.digest_of(back) == pd.digest_of(ro)


@case("[A8] 落库行里的坏数据跳过而不是整份废掉")
def t_a8():
    back = pd.roster_from_rows([{"c": "600000.SH", "r": 1}, "坏行", None, {}, {"r": 5}])
    assert list(back) == ["600000.SH"]


@case("[A9] 名册行按代码序输出 —— 落库文本稳定")
def t_a9():
    rows = pd.roster_rows(roster(_r("600418.SH", 2, 9.0), _r("600000.SH", 1, 10.0)))
    assert [r["c"] for r in rows] == ["600000.SH", "600418.SH"]


# ================================================================ [B] 档位序
@case("[B1] 档位强弱: 强传导 > 弱传导 > 无传导")
def t_b1():
    assert pd.tier_rank("强传导") > pd.tier_rank("弱传导") > pd.tier_rank("无传导")


@case("[B2] 未知档位返回 None (判不了就不判)")
def t_b2():
    assert pd.tier_rank("超强传导") is None and pd.tier_rank(None) is None


@case("[B3] 未知档位不参与升降判定 —— 宁可不报, 不能报反")
def t_b3():
    d = pd.diff(roster(_r("600000.SH", 1, 10.0, tier="超强传导")),
                roster(_r("600000.SH", 1, 10.0, tier="强传导")),
                prev_date="2026-07-29", curr_date="2026-07-30")
    assert d["counts"]["tier_up"] == 0 and d["counts"]["tier_down"] == 0


# ================================================================ [C] 首次
@case("[C1] 没有上一版时一条变化都不报")
def t_c1():
    d = pd.diff(None, roster(*[_r(f"60000{i}.SH", i, 100 - i) for i in range(1, 9)]),
                curr_date="2026-07-30")
    assert d["kind"] == pd.KIND_FIRST
    assert sum(d["counts"].values()) == 0, "首次落快照若报'新进榜', 会是整整一榜的假变化"
    assert d["roster_size"]["curr"] == 8


@case("[C2] 首次的 note 说明从下一次开始才有比对")
def t_c2():
    d = pd.diff({}, roster(_r("600000.SH", 1, 10.0)))
    assert "首次" in d["note"] and d["prev_broken"] is False


@case("[C3] 上一版读不出来要说成故障, 不能长得跟'首次'一样")
def t_c3():
    # 快照的 roster_json 坏了 → 解析成空。若与"首次"同形, 一条真实变化都不报却显示正常。
    d = pd.diff({}, roster(_r("600000.SH", 1, 10.0)),
                prev_date="2026-07-29", curr_date="2026-07-30", prev_broken=True)
    assert d["prev_broken"] is True
    assert "读不出来" in d["note"] and "比不了" in d["note"]
    assert "首次" not in d["note"]


@case("[C4] 首次也要回显调用方给的参数, 不能写死默认值")
def t_c4():
    d = pd.diff(None, roster(_r("600000.SH", 1, 10.0)), rank_jump=80, tail_guard=0.2)
    assert d["params"] == {"rank_jump": 80, "tail_guard": 0.2}, "写死会让页面显示错的口径"


# ================================================================ [D] 三种比对语义
@case("[D1] 计划日不同 = 跨日的正常升降档")
def t_d1():
    d = pd.diff(roster(_r("600000.SH", 1, 10.0)), roster(_r("600000.SH", 1, 10.0)),
                prev_date="2026-07-29", curr_date="2026-07-30")
    assert d["kind"] == pd.KIND_CROSS_DAY


@case("[D2] 同一计划日的不同版本要单独成一类")
def t_d2():
    d = pd.diff(roster(_r("600000.SH", 1, 10.0)), roster(_r("600000.SH", 2, 10.0)),
                prev_date="2026-07-30", curr_date="2026-07-30")
    assert d["kind"] == pd.KIND_REVISION


@case("[D3] 同日新版本的结论要点明'候选池与此前不是同一份'")
def t_d3():
    d = pd.diff(roster(_r("600000.SH", 1, 10.0)), roster(_r("600418.SH", 1, 11.0)),
                prev_date="2026-07-30", curr_date="2026-07-30")
    assert "重算" in d["note"] and "候选池" in d["note"]


# ================================================================ [E] 进出榜与尾部闸
@case("[E1] 新进榜与掉榜各归各位")
def t_e1():
    prev = roster(_r("600000.SH", 1, 10.0), _r("600418.SH", 2, 9.0))
    curr = roster(_r("600000.SH", 1, 10.0), _r("600519.SH", 2, 9.5))
    d = pd.diff(prev, curr, prev_date="2026-07-29", curr_date="2026-07-30")
    assert [x["ts_code"] for x in d["entered"]] == ["600519.SH"]
    assert [x["ts_code"] for x in d["exited"]] == ["600418.SH"]


@case("[E2] 掉榜那条带的是**上一版**的排名与档位")
def t_e2():
    prev = roster(_r("600418.SH", 7, 233.3, tier="弱传导"))
    d = pd.diff(prev, roster(_r("600000.SH", 1, 10.0)),
                prev_date="2026-07-29", curr_date="2026-07-30")
    x = d["exited"][0]
    assert x["rank"] == 7 and x["tier"] == "弱传导", "掉榜的票在这一版没有数据, 只能示上一版"


@case("[E3] 两版都没吃满 top 时不设尾部闸 —— 消失就是真消失")
def t_e3():
    prev = roster(*[_r(f"6000{i:02d}.SH", i, 100 - i) for i in range(1, 11)])
    curr = roster(*[_r(f"6000{i:02d}.SH", i, 100 - i) for i in range(1, 10)])   # 第 10 名没了
    d = pd.diff(prev, curr, prev_date="2026-07-30", curr_date="2026-07-30",
                prev_capped=False, curr_capped=False)
    assert d["counts"]["exited"] == 1 and d["tail_churn"]["exited"] == 0
    # tail_guarded 报的是各档闸位在第几名, None = 该档没设闸 (让"为什么这条没报"可查)
    assert d["tail_guarded"] == {"entered": {"main": None, "observe": None},
                                 "exited": {"main": None, "observe": None}}


@case("[E4] 这一版吃满 top 时, 榜尾掉榜归 tail_churn 不列名")
def t_e4():
    prev = roster(*[_r(f"6000{i:02d}.SH", i, 100 - i) for i in range(1, 11)])
    curr = roster(*[_r(f"6000{i:02d}.SH", i, 100 - i) for i in range(1, 10)])   # 原第 10 名没了
    d = pd.diff(prev, curr, prev_date="2026-07-30", curr_date="2026-07-30",
                curr_capped=True)                    # 榜长 9, 闸位 4.5, 上一版 rank 10 在闸外
    assert d["counts"]["exited"] == 0
    assert d["tail_churn"]["exited"] == 1, "它可能只是掉到我们要的条数之外, 不是上游摘了它"


@case("[E5] 吃满 top 也照报榜首掉榜")
def t_e5():
    prev = roster(*[_r(f"6000{i:02d}.SH", i, 100 - i) for i in range(1, 11)])
    curr = roster(*[_r(f"6000{i:02d}.SH", i, 100 - i) for i in range(2, 11)])   # 榜首没了
    d = pd.diff(prev, curr, prev_date="2026-07-30", curr_date="2026-07-30", curr_capped=True)
    assert [x["ts_code"] for x in d["exited"]] == ["600001.SH"]
    assert d["tail_churn"]["exited"] == 0


@case("[E6] 新进榜的尾部闸看的是**上一版**吃没吃满")
def t_e6():
    prev = roster(*[_r(f"6000{i:02d}.SH", i, 100 - i) for i in range(1, 11)])
    curr = dict(prev)
    curr["600099.SH"] = _r("600099.SH", 9, 91.5)     # 新面孔排在榜尾
    d = pd.diff(prev, curr, prev_date="2026-07-30", curr_date="2026-07-30", prev_capped=True)
    assert d["counts"]["entered"] == 0
    assert d["tail_churn"]["entered"] == 1, "上一版吃满了, 这张脸当时可能就在榜上只是没露面"


@case("[E7] 上一版吃满时, 榜首的新面孔照报")
def t_e7():
    prev = roster(*[_r(f"6000{i:02d}.SH", i, 100 - i) for i in range(1, 11)])
    curr = dict(prev)
    curr["600099.SH"] = _r("600099.SH", 1, 200.0)    # 新面孔直接冲到第一
    d = pd.diff(prev, curr, prev_date="2026-07-30", curr_date="2026-07-30", prev_capped=True)
    assert [x["ts_code"] for x in d["entered"]] == ["600099.SH"]


@case("[E8] 两道闸互不干扰: 这一版截断不该影响新进榜的判定")
def t_e8():
    prev = roster(*[_r(f"6000{i:02d}.SH", i, 100 - i) for i in range(1, 11)])
    curr = dict(prev)
    curr["600099.SH"] = _r("600099.SH", 9, 91.5)
    d = pd.diff(prev, curr, prev_date="2026-07-30", curr_date="2026-07-30",
                prev_capped=False, curr_capped=True)
    assert d["counts"]["entered"] == 1, "上一版没吃满, 那这张新面孔就是真的新"
    assert d["tail_churn"]["entered"] == 0


@case("[E9] 整档消失不被尾部闸吞掉")
def t_e9():
    prev = roster(_r("600877.SH", 1, 101.0, tier=None, bucket="observe"),
                  _r("600000.SH", 1, 10.0))
    curr = roster(_r("600000.SH", 1, 10.0))          # 观察档整档没了
    d = pd.diff(prev, curr, prev_date="2026-07-30", curr_date="2026-07-30", curr_capped=True)
    assert [x["ts_code"] for x in d["exited"]] == ["600877.SH"], (
        "这一版观察档一条都没有, 谈不上被 top 截断 —— 整档消失本身就是要报的大事")


@case("[E10] 尾部闸按档各算各的 (闸位取自各自那一档的长度)")
def t_e10():
    prev = roster(*([_r(f"6000{i:02d}.SH", i, 100 - i) for i in range(1, 11)]
                    + [_r("600877.SH", 1, 101.0, tier=None, bucket="observe")]))
    curr = roster(*([_r(f"6000{i:02d}.SH", i, 100 - i) for i in range(1, 10)]
                    + [_r(f"6009{i:02d}.SH", i, 90 - i, tier=None, bucket="observe")
                       for i in range(1, 11)]))
    d = pd.diff(prev, curr, prev_date="2026-07-30", curr_date="2026-07-30",
                curr_capped={"main": True, "observe": True})
    # 主榜榜尾那只走 churn; 观察档那只 rank=1、这一版观察档有 10 条(闸位 5), 在闸内 → 照报
    assert [x["ts_code"] for x in d["exited"]] == ["600877.SH"]
    assert d["tail_churn"]["exited"] == 1


@case("[E11] 主榜吃满不代表观察档也吃满 —— 别拿主榜的截断解释观察档的摘牌")
def t_e11():
    # 上游两个独立参数: top=10 吃满了, obs_top=100 只回了 4 条 (没吃满)。
    # 观察档那只真被摘了 —— 若两档共用主榜那一个标志, 它会被记成"截断噪音"永远看不见。
    prev = roster(*([_r(f"6000{i:02d}.SH", i, 100 - i) for i in range(1, 11)]
                    + [_r(f"6009{i:02d}.SH", i, 90 - i, tier=None, bucket="observe")
                       for i in range(1, 5)]))
    curr = roster(*([_r(f"6000{i:02d}.SH", i, 100 - i) for i in range(1, 11)]
                    + [_r(f"6009{i:02d}.SH", i, 90 - i, tier=None, bucket="observe")
                       for i in range(1, 4)]))          # 观察档第 4 名被摘
    d = pd.diff(prev, curr, prev_date="2026-07-30", curr_date="2026-07-30",
                curr_capped={"main": True, "observe": False})
    assert [x["ts_code"] for x in d["exited"]] == ["600904.SH"], (
        "观察档没吃满就不该设闸 —— 拿主榜的截断解释观察档, 会吞掉一次真实摘牌")
    assert d["tail_churn"]["exited"] == 0
    # 反过来: 两档都按吃满算, 这条就会被吞掉 (证明上面那条断言不是白写的)
    d2 = pd.diff(prev, curr, prev_date="2026-07-30", curr_date="2026-07-30", curr_capped=True)
    assert d2["counts"]["exited"] == 0 and d2["tail_churn"]["exited"] == 1


@case("[E12] 持仓票不受榜尾闸约束 —— 漏报一条就是一个仓位")
def t_e12():
    prev = roster(*[_r(f"6000{i:02d}.SH", i, 100 - i) for i in range(1, 11)])
    curr = roster(*[_r(f"6000{i:02d}.SH", i, 100 - i) for i in range(1, 10)])
    # 非持仓: 上一版 rank 10 在闸外 → 吞掉
    d0 = pd.diff(prev, curr, prev_date="2026-07-30", curr_date="2026-07-30", curr_capped=True)
    assert d0["counts"]["exited"] == 0
    # 同一只票是持仓: 照报, 且记一笔豁免
    d1 = pd.diff(prev, curr, prev_date="2026-07-30", curr_date="2026-07-30",
                 curr_capped=True, held=["600010.SH"])
    assert [x["ts_code"] for x in d1["exited"]] == ["600010.SH"]
    assert d1["tail_churn"]["exited"] == 0 and d1["tail_churn"]["held_exempt"] == 1
    assert d1["held"]["n_watch"] == 1


@case("[E13] rank 判不出来时照报, 不当噪音吞掉")
def t_e13():
    prev = roster(_r("600000.SH", None, 10.0), _r("600001.SH", 1, 99.0))
    curr = roster(_r("600001.SH", 1, 99.0))
    d = pd.diff(prev, curr, prev_date="2026-07-30", curr_date="2026-07-30", curr_capped=True)
    assert [x["ts_code"] for x in d["exited"]] == ["600000.SH"], (
        "这个函数的'是'代表吞掉一条变化, 所以拿不准时必须放行")


# ================================================================ [F] 档位与券商覆盖
@case("[F1] 升档与降档")
def t_f1():
    prev = roster(_r("600000.SH", 1, 10.0, tier="弱传导"), _r("600418.SH", 2, 9.0, tier="强传导"))
    curr = roster(_r("600000.SH", 1, 10.0, tier="强传导"), _r("600418.SH", 2, 9.0, tier="弱传导"))
    d = pd.diff(prev, curr, prev_date="2026-07-29", curr_date="2026-07-30")
    assert [x["ts_code"] for x in d["tier_up"]] == ["600000.SH"]
    assert d["tier_up"][0]["tier_from"] == "弱传导" and d["tier_up"][0]["tier_to"] == "强传导"
    assert [x["ts_code"] for x in d["tier_down"]] == ["600418.SH"]


@case("[F2] 主榜 ↔ 观察档 互换 = 券商覆盖翻转, 单独一段")
def t_f2():
    prev = roster(_r("600000.SH", 1, 10.0))
    curr = roster(_r("600000.SH", 1, 10.0, tier=None, bucket="observe"))
    d = pd.diff(prev, curr, prev_date="2026-07-30", curr_date="2026-07-30")
    assert len(d["bucket_moved"]) == 1
    m = d["bucket_moved"][0]
    assert m["moved_from"] == "main" and m["moved_to"] == "observe"
    assert d["counts"]["exited"] == 0, "换档不是掉榜 —— 票还在, 只是没券商覆盖了"


@case("[F3] 换了档就不算名次跳变 —— 两档的 rank 不可比")
def t_f3():
    prev = roster(_r("600000.SH", 300, 10.0))
    curr = roster(_r("600000.SH", 1, 10.0, tier=None, bucket="observe"))
    d = pd.diff(prev, curr, prev_date="2026-07-30", curr_date="2026-07-30", rank_jump=10)
    assert d["counts"]["rank_jump"] == 0
    assert d["counts"]["bucket_moved"] == 1


# ================================================================ [G] 名次跳变
@case("[G1] 跳变到阈值才报")
def t_g1():
    prev = roster(_r("600000.SH", 100, 10.0), _r("600418.SH", 20, 20.0))
    curr = roster(_r("600000.SH", 20, 30.0), _r("600418.SH", 15, 22.0))
    d = pd.diff(prev, curr, prev_date="2026-07-29", curr_date="2026-07-30", rank_jump=50)
    assert [x["ts_code"] for x in d["rank_jump"]] == ["600000.SH"]
    assert d["rank_jump"][0]["rank_from"] == 100 and d["rank_jump"][0]["rank_delta"] == -80


@case("[G2] 按跳变幅度降序")
def t_g2():
    prev = roster(_r("600000.SH", 200, 1.0), _r("600418.SH", 60, 2.0))
    curr = roster(_r("600000.SH", 100, 5.0), _r("600418.SH", 1, 9.0))
    d = pd.diff(prev, curr, prev_date="2026-07-29", curr_date="2026-07-30", rank_jump=50)
    assert [x["ts_code"] for x in d["rank_jump"]] == ["600000.SH", "600418.SH"]


@case("[G3] rank 缺失时不算跳变")
def t_g3():
    prev = roster(_r("600000.SH", None, 1.0))
    curr = roster(_r("600000.SH", 1, 9.0))
    d = pd.diff(prev, curr, prev_date="2026-07-29", curr_date="2026-07-30", rank_jump=1)
    assert d["counts"]["rank_jump"] == 0


# ================================================================ [H] 持仓票视图
@case("[H1] 持仓票掉榜进 n_watch")
def t_h1():
    prev = roster(_r("600000.SH", 1, 10.0), _r("600418.SH", 2, 9.0))
    curr = roster(_r("600418.SH", 1, 9.0))
    d = pd.diff(prev, curr, prev_date="2026-07-29", curr_date="2026-07-30",
                held=["600000.SH"])
    assert [x["ts_code"] for x in d["held"]["exited"]] == ["600000.SH"]
    assert d["held"]["n_watch"] == 1
    assert d["exited"][0]["held"] is True


@case("[H2] 持仓票降档也进 n_watch")
def t_h2():
    d = pd.diff(roster(_r("600000.SH", 1, 10.0, tier="强传导")),
                roster(_r("600000.SH", 1, 10.0, tier="弱传导")),
                prev_date="2026-07-29", curr_date="2026-07-30", held=["600000.SH"])
    assert d["held"]["n_watch"] == 1 and len(d["held"]["tier_down"]) == 1


@case("[H2b] 持仓票掉进观察档 = 丢了券商覆盖, 也要进 n_watch")
def t_h2b():
    # 观察档的行没有 tier, 所以这既不算 exited 也不算 tier_down —— 不显式算进来的话,
    # 一只持仓票丢了估值锚, 页面红条不亮、探活脚本还会打一句"没被摘也没降档"。
    d = pd.diff(roster(_r("600000.SH", 3, 240.0, tier="强传导")),
                roster(_r("600000.SH", 5, 99.0, tier=None, bucket="observe")),
                prev_date="2026-07-29", curr_date="2026-07-30", held=["600000.SH"])
    assert d["counts"]["exited"] == 0 and d["counts"]["tier_down"] == 0
    assert [x["ts_code"] for x in d["held"]["coverage_lost"]] == ["600000.SH"]
    assert d["held"]["n_watch"] == 1
    assert "券商覆盖" in d["note"]


@case("[H2c] 掉进主榜(拿到券商覆盖)不算要盯的事")
def t_h2c():
    d = pd.diff(roster(_r("600000.SH", 5, 99.0, tier=None, bucket="observe")),
                roster(_r("600000.SH", 3, 240.0, tier="强传导")),
                prev_date="2026-07-29", curr_date="2026-07-30", held=["600000.SH"])
    assert d["counts"]["bucket_moved"] == 1
    assert d["held"]["coverage_lost"] == [] and d["held"]["n_watch"] == 0


@case("[H3] 持仓票升档只进 n_total, 不进 n_watch")
def t_h3():
    d = pd.diff(roster(_r("600000.SH", 1, 10.0, tier="弱传导")),
                roster(_r("600000.SH", 1, 10.0, tier="强传导")),
                prev_date="2026-07-29", curr_date="2026-07-30", held=["600000.SH"])
    assert d["held"]["n_watch"] == 0 and d["held"]["n_total"] == 1


@case("[H4] 非持仓票不进持仓视图")
def t_h4():
    d = pd.diff(roster(_r("600000.SH", 1, 10.0)), roster(_r("600418.SH", 1, 10.0)),
                prev_date="2026-07-29", curr_date="2026-07-30", held=["600519.SH"])
    assert d["held"]["n_total"] == 0 and d["counts"]["exited"] == 1


@case("[H5] 持仓票掉榜要在结论句里显眼")
def t_h5():
    d = pd.diff(roster(_r("600000.SH", 1, 10.0)), roster(_r("600418.SH", 1, 10.0)),
                prev_date="2026-07-29", curr_date="2026-07-30", held=["600000.SH"])
    assert "持仓票" in d["note"] and "掉榜" in d["note"]


# ================================================================ [I] 健壮性
@case("[I1] 空对空不炸")
def t_i1():
    d = pd.diff({}, {}, prev_date="2026-07-30", curr_date="2026-07-30")
    assert d["kind"] == pd.KIND_FIRST and sum(d["counts"].values()) == 0


@case("[I2] 名册全没了也报得出来")
def t_i2():
    d = pd.diff(roster(_r("600000.SH", 1, 10.0)), {},
                prev_date="2026-07-29", curr_date="2026-07-30")
    assert d["counts"]["exited"] == 1 and d["roster_size"]["curr"] == 0


@case("[I3] tail_guard 越界值被夹回 [0,1]")
def t_i3():
    prev = roster(*[_r(f"6000{i:02d}.SH", i, 100 - i) for i in range(1, 11)])
    curr = roster(*[_r(f"6000{i:02d}.SH", i, 100 - i) for i in range(1, 11) if i != 5])
    for given, clamped, reported in ((5.0, 1.0, 1), (-1.0, 0.0, 0)):
        d = pd.diff(prev, curr, prev_date="2026-07-30", curr_date="2026-07-30",
                    curr_capped=True, tail_guard=given)
        assert d["params"]["tail_guard"] == clamped
        # 闸开到底(1.0): 榜内的第 5 名掉了要报; 闸关到底(0.0): 一律当截断噪音
        assert d["counts"]["exited"] == reported


@case("[I4] counts 与各段长度始终一致")
def t_i4():
    prev = roster(_r("600000.SH", 1, 10.0, tier="强传导"), _r("600418.SH", 200, 2.0))
    curr = roster(_r("600000.SH", 1, 10.0, tier="弱传导"), _r("600519.SH", 3, 8.0))
    d = pd.diff(prev, curr, prev_date="2026-07-29", curr_date="2026-07-30")
    for k, n in d["counts"].items():
        assert len(d[k]) == n, f"{k} 的 counts 与实际条数对不上"


# ================================================================ [J] 服务层落库往返
# 纯逻辑之外唯一容易藏 bug 的地方: 名册 → JSON 落库 → 读回来 → 比对 这条往返。
# 单测不连库 —— 拿内存桩顶掉 pms_repo 与 param_store (plan_feed 是函数内 import, 装在
# sys.modules 上就生效; 与 test_batch7 用假 requests 是同一套路)。
def _with_fake_repo(params=None):
    import json as _j
    import types
    store = {"rows": [], "seq": 0}

    repo = types.ModuleType("app.repo.pms_repo")

    def insert_plan_snapshot(*, plan_date, digest, roster, meta=None, n_main=0, n_observe=0,
                             capped_main=False, capped_obs=False, fetched_at=None):
        # 与生产同一条判据: 只跟**最新那行**比, 不是"曾经出现过的任何一行"。
        # 早先桩里写的是后者 (模拟唯一键), 于是 A→B→A 这个 case 在桩上永远复现不出来 ——
        # 桩比生产更严, 单测就成了给错实现背书。
        last = max(store["rows"], key=lambda x: x["id"]) if store["rows"] else None
        if last and last["plan_date"] == plan_date and last["digest"] == digest:
            return 0
        store["seq"] += 1
        store["rows"].append({
            "id": store["seq"], "plan_date": plan_date, "digest": digest,
            "fetched_ymd": 20260731, "fetched_at": f"t{store['seq']}",
            "n_main": n_main, "n_observe": n_observe,
            "capped_main": bool(capped_main), "capped_obs": bool(capped_obs),
            # 真表是 MEDIUMTEXT, 落进去的是 JSON 文本、读回来才转对象。桩必须照做 ——
            # 否则"名册里塞了不可序列化的东西"这类事故单测里根本看不见。
            "roster_json": _j.dumps(roster, ensure_ascii=False), "meta": meta or {}})
        return 1

    def latest_plan_snapshots(limit=2):
        out = []
        for r in sorted(store["rows"], key=lambda x: -x["id"])[:limit]:
            r2 = dict(r)
            r2["roster"] = _j.loads(r2.pop("roster_json"))
            out.append(r2)
        return out

    def list_plan_snapshots(*, plan_date=None, limit=50):
        rows = [dict(r) for r in sorted(store["rows"], key=lambda x: -x["id"])]
        if plan_date:
            rows = [r for r in rows if r["plan_date"] == plan_date]
        for r in rows:
            r.pop("roster_json", None)
        return rows[:limit]

    repo.insert_plan_snapshot = insert_plan_snapshot
    repo.latest_plan_snapshots = latest_plan_snapshots
    repo.list_plan_snapshots = list_plan_snapshots
    repo.prune_plan_snapshots = lambda keep=200: 0
    repo.list_positions = lambda only_open=False: []

    ps = types.ModuleType("app.services.param_store")
    vals = dict(params or {})
    ps.get = lambda k, d=None: vals.get(k, d)
    ps.get_int = lambda k, d=0: int(vals.get(k, d))
    ps.get_float = lambda k, d=0.0: float(vals.get(k, d))
    ps.get_bool = lambda k, d=False: bool(vals.get(k, d))
    ps.get_list = lambda k, d=(): list(vals.get(k, d))

    saved = {n: sys.modules.get(n) for n in ("app.repo.pms_repo", "app.services.param_store")}
    sys.modules["app.repo.pms_repo"] = repo
    sys.modules["app.services.param_store"] = ps

    def restore():
        for n, m in saved.items():
            if m is None:
                sys.modules.pop(n, None)
            else:
                sys.modules[n] = m
    return store, restore


@case("[J1] 落库往返: 名册经 JSON 存取后指纹不变")
def t_j1():
    import json as _j
    from app.services import plan_feed as pf
    store, restore = _with_fake_repo({"PMS_PLAN_SNAPSHOT": True})
    try:
        p = plan(main=[_r("600418.SH", 1, 242.24, theme="整车"),
                       _r("600000.SH", 2, 241.55, tier="弱传导", theme=None)],
                 observe=[_r("600877.SH", 1, 101.1, tier=None, bucket="observe")],
                 date="2026-07-30")
        out = pf.snapshot(p)
        assert out["stored"] is True and out["rows"] == 3
        back = pd.roster_from_rows(_j.loads(store["rows"][0]["roster_json"]))
        assert pd.digest_of(back) == out["digest"], "存取一轮指纹就变了, 比对全废"
        assert back["600000.SH"]["h"] is None, "theme 为空的票也要原样存回来"
    finally:
        restore()


@case("[J2] 同一份榜重复落只留一行")
def t_j2():
    from app.services import plan_feed as pf
    store, restore = _with_fake_repo({"PMS_PLAN_SNAPSHOT": True})
    try:
        p = plan(main=[_r("600418.SH", 1, 242.24)], date="2026-07-30")
        assert pf.snapshot(p)["stored"] is True
        assert pf.snapshot(p)["stored"] is False, "内容没变还新增一行, 表会白涨"
        assert len(store["rows"]) == 1
    finally:
        restore()


@case("[J3] score 变了但次序没变, 不算新版本")
def t_j3():
    from app.services import plan_feed as pf
    store, restore = _with_fake_repo({"PMS_PLAN_SNAPSHOT": True})
    try:
        pf.snapshot(plan(main=[_r("600418.SH", 1, 242.2801)], date="2026-07-30"))
        pf.snapshot(plan(main=[_r("600418.SH", 1, 242.2799)], date="2026-07-30"))
        assert len(store["rows"]) == 1
    finally:
        restore()


@case("[J4] 只读预览不写库, 且认得出'与库里那份相同'")
def t_j4():
    from app.services import plan_feed as pf
    store, restore = _with_fake_repo({"PMS_PLAN_SNAPSHOT": True})
    try:
        p = plan(main=[_r("600418.SH", 1, 242.24)], date="2026-07-30")
        pf.snapshot(p)
        n = len(store["rows"])
        out = pf.preview_changes(p)
        assert out["ok"] and out["same_as_stored"] is True
        assert len(store["rows"]) == n, "preview 必须只读 —— 探活脚本的只读契约靠它"
    finally:
        restore()


@case("[J5] 只读预览能比出手上这份与库里那份的差异")
def t_j5():
    from app.services import plan_feed as pf
    store, restore = _with_fake_repo({"PMS_PLAN_SNAPSHOT": True})
    try:
        pf.snapshot(plan(main=[_r("600418.SH", 1, 242.24)], date="2026-07-30"))
        out = pf.preview_changes(plan(main=[_r("600519.SH", 1, 250.0)], date="2026-07-30"))
        d = out["diff"]
        assert d["kind"] == pd.KIND_REVISION
        assert [x["ts_code"] for x in d["entered"]] == ["600519.SH"]
        assert [x["ts_code"] for x in d["exited"]] == ["600418.SH"]
        assert len(store["rows"]) == 1
    finally:
        restore()


@case("[J6] 快照关掉时 changes 明说不可用, 而不是装作没变化")
def t_j6():
    from app.services import plan_feed as pf
    _, restore = _with_fake_repo({"PMS_PLAN_SNAPSHOT": False})
    try:
        out = pf.changes()
        assert out["ok"] is False and out["diff"] is None
        assert "PMS_PLAN_SNAPSHOT" in out["hint"], "得说清是关掉了, 不是没有变化"
    finally:
        restore()


@case("[J7] 快照落库失败不阻断候选池, 但要把错带出来")
def t_j7():
    from app.services import plan_feed as pf
    _, restore = _with_fake_repo({"PMS_PLAN_SNAPSHOT": True})
    try:
        def _boom(**kw):
            raise RuntimeError("DB 挂了")
        sys.modules["app.repo.pms_repo"].insert_plan_snapshot = _boom
        out = pf._snapshot_quiet(plan(main=[_r("600418.SH", 1, 1.0)], date="2026-07-30"))
        assert out["stored"] is False and "DB 挂了" in out["error"]
    finally:
        restore()


@case("[J7b] 上游 A→B→A 改回去时, 比对方向不能反")
def t_j7b():
    # 曾经用 (plan_date, digest) 唯一键去重, 于是第三次的 A 落不下去、库里最新仍是 B,
    # 页面把 B→A 这次真实回退显示成 A→B —— 方向整个反了, 还顺带诬告一句"落库失败"。
    from app.services import plan_feed as pf
    store, restore = _with_fake_repo({"PMS_PLAN_SNAPSHOT": True})
    try:
        a = plan(main=[_r("600418.SH", 1, 242.24)], date="2026-07-30")
        b = plan(main=[_r("600519.SH", 1, 250.00)], date="2026-07-30")
        pf.snapshot(a)
        pf.snapshot(b)
        assert pf.snapshot(a)["stored"] is True, "改回去也是一次真实变化, 必须落得下去"
        assert len(store["rows"]) == 3
        out = pf.preview_changes(a)
        assert out["same_as_stored"] is True
        d = pf.changes()["diff"]
        assert [x["ts_code"] for x in d["entered"]] == ["600418.SH"]
        assert [x["ts_code"] for x in d["exited"]] == ["600519.SH"]
    finally:
        restore()


@case("[J7c] 上一版名册坏了要报故障, 不能显示成'首次'")
def t_j7c():
    import json as _j
    from app.services import plan_feed as pf
    store, restore = _with_fake_repo({"PMS_PLAN_SNAPSHOT": True})
    try:
        pf.snapshot(plan(main=[_r("600418.SH", 1, 1.0)], date="2026-07-30"))
        store["rows"][0]["roster_json"] = _j.dumps([])      # 名册没了, n_main 还写着 1
        out = pf.preview_changes(plan(main=[_r("600519.SH", 1, 2.0)], date="2026-07-31"))
        assert out["diff"]["prev_broken"] is True
        assert "读不出来" in out["hint"]
    finally:
        restore()


@case("[J7d] 持仓代码是前缀式也要认得出来")
def t_j7d():
    from app.services import plan_feed as pf
    _, restore = _with_fake_repo({"PMS_PLAN_SNAPSHOT": True})
    try:
        sys.modules["app.repo.pms_repo"].list_positions = \
            lambda only_open=False: [{"ts_code": "SH600418"}]      # 前缀式
        pf.snapshot(plan(main=[_r("600418.SH", 1, 1.0)], date="2026-07-30"))
        out = pf.preview_changes(plan(main=[_r("600519.SH", 1, 2.0)], date="2026-07-31"))
        assert out["diff"]["held"]["n_watch"] == 1, "不归一的后果是持仓视图静默为空"
    finally:
        restore()


@case("[J8] 快照台账数得出同一计划日有几版")
def t_j8():
    from app.services import plan_feed as pf
    _, restore = _with_fake_repo({"PMS_PLAN_SNAPSHOT": True})
    try:
        pf.snapshot(plan(main=[_r("600418.SH", 1, 1.0)], date="2026-07-30"))
        pf.snapshot(plan(main=[_r("600418.SH", 2, 1.0)], date="2026-07-30"))
        pf.snapshot(plan(main=[_r("600418.SH", 1, 1.0)], date="2026-07-29"))
        log = pf.snapshot_log()
        assert log["ok"] and log["versions_per_date"]["2026-07-30"] == 2
        assert log["revised"] == {"2026-07-30": 2}, "同一 date 多版是 §7.3 那个现象, 要数出来"
    finally:
        restore()


def main():
    import logging
    logging.disable(logging.CRITICAL)
    passed, failed = 0, 0
    for name, fn in RESULTS:
        try:
            fn()
            print(f"  PASS  {name}")
            passed += 1
        except Exception:
            print(f"  FAIL  {name}")
            traceback.print_exc()
            failed += 1
    print("-" * 60)
    if failed:
        print(f"FAILED: {failed} / {passed + failed}")
        sys.exit(1)
    print(f"ALL PASS ({passed} cases)")


if __name__ == "__main__":
    main()