akg-factor-bridge/test_plan_logic_state.py

"""计划装配接入逻辑状态四态的离线单测（不连库）。

钉住三件事：
  一，四态随每张卡一起产出，并按约定的形状发给下游（每路带截止日与一句话说明）。
  二，四态**不改判决**。候选卡的三门槛判决与逻辑四态是正交的两维，收敛规则是单调的：
      逻辑强化只能提前卡内序，永远不能把判决往上升一档；逻辑存疑只能改分流通道，
      不能把仅展示变成可执行。
  三，取数任一路读不到都不让计划断产，那一路记缺失。

跑法：python3 test_plan_logic_state.py   或   pytest test_plan_logic_state.py
"""
import logic_state as ls
import plan

DS = "2026-09-03"


def t(name, cond):
    assert cond, name
    print("  ok", name)


def claim(date, direction="利好"):
    return {"disclosure_date": date, "direction": direction, "mechanism": "机制",
            "doc_title": "研报", "claim_id": "c" + date.replace("-", "")}


def jrow(leaning="偏多", verified=1, migrated=0, stale=3, seg="固态电解质"):
    return {"leaning": leaning, "leaning_prev": None, "migrated": migrated,
            "stale_days": stale, "verified": verified, "review_date": "2026-09-03",
            "n_materials": 43, "subject_name": seg, "segment_name": seg,
            "n_bull": 3, "n_bear": 3}


def main():
    print("四态随卡产出")
    evd = {"logic": [claim("2026-08-25")]}
    seg_of = {"SH600000": ["固态电解质"]}
    seg_view = {"固态电解质": jrow()}
    broker = {"SH600000": ls.signal(ls.PATH_BROKER, ls.SIG_FLAT, as_of=DS,
                                    coverage=6, why="预测中位数变化 +1%，在阈值之内")}
    st = plan._logic_state_of("SH600000", evd, seg_of, seg_view, broker, DS)  # noqa: SLF001
    t("三路都在且无负面 -> 逻辑成立", st["state"] == ls.STATE_HOLD)
    t("五路都记了名，缺的两路写明是公司事件与公司质地（没报告）",
      len(st["paths"]) == 5 and ls.PATH_EVENT in st["missing"] and ls.PATH_COMPANY in st["missing"])

    print("取数缺席时不断产")
    st = plan._logic_state_of("SH600000", {}, {}, {}, {}, DS)  # noqa: SLF001
    t("四路全缺 -> 无法判断加证据不足，不抛异常",
      st["state"] == ls.STATE_UNKNOWN and st["why"] == ls.WHY_THIN)
    t("缺失名单写全了五路", len(st["missing"]) == 5)

    st = plan._logic_state_of("SH999999", evd, seg_of, seg_view, broker, DS)  # noqa: SLF001
    t("这只票不在任何被指向环节上 -> 产业研判缺席，其余照算",
      ls.PATH_JUDGE in st["missing"] and ls.PATH_CLAIM in st["usable"])

    print("一票挂多个环节")
    st = plan._logic_state_of(  # noqa: SLF001
        "SH600000", evd, {"SH600000": ["没评过的环节", "固态电解质"]}, seg_view, broker, DS)
    t("取第一个有行业观点的那个环节", ls.PATH_JUDGE in st["usable"])

    print("发给下游的形状")
    out = plan._state_out(st)  # noqa: SLF001
    t("状态、子因、截止日都在", set(out) >= {"state", "why", "as_of", "usable", "missing",
                                            "reasons", "paths"})
    t("每一路都带自己的截止日与一句话说明",
      all(set(p) == {"path", "signal", "as_of", "why"} for p in out["paths"]))
    t("不发内部中间量（硬触发标记、出处原值不外泄）",
      all("hard" not in p and "refs" not in p for p in out["paths"]))
    t("没有状态时发 None，不硬拼一个空壳", plan._state_out(None) is None)  # noqa: SLF001
    t("没做过落定的结果：原始态就是 state 本身、落定说明为空",
      out["raw_state"] == out["state"] and out["settle_note"] is None and out["prev_state"] is None)

    print("落定态随卡发出（2026-09-07 第三件桥侧前置）")
    import logic_state_daily as lsd
    settled = lsd.settle_one("SH600000", st, [{"trade_date": "2026-09-02", "raw_state": ls.STATE_DOUBT,
                                               "state": ls.STATE_DOUBT}])
    out2 = plan._state_out(settled)  # noqa: SLF001
    t("原始态成立、上一次落定存疑、只有一天不存疑 -> 发出的 state 仍是存疑",
      out2["raw_state"] == st["state"] and out2["state"] == ls.STATE_DOUBT
      and out2["prev_state"] == ls.STATE_DOUBT and "现在 1 天" in out2["settle_note"])
    t("落定不改每路的截止日与说明", out2["paths"] == out["paths"])

    print("四态不改判决（收敛规则单调）")
    for state in (ls.STATE_STRONG, ls.STATE_HOLD, ls.STATE_UNKNOWN, ls.STATE_DOUBT):
        for verdict in ("候选", "关注", "仅展示"):
            r = ls.apply_to_card(verdict, state)
            assert r["verdict"] == verdict, (verdict, state, r)
    t("十二种组合逐个扫过，判决一次都没被四态改动", True)
    t("逻辑强化最多提前卡内序", ls.apply_to_card("关注", ls.STATE_STRONG)["rank_bonus"] == 1)
    t("逻辑存疑不把仅展示变成可执行",
      not ls.apply_to_card("仅展示", ls.STATE_DOUBT)["force_confirm"])
    t("候选加逻辑存疑是唯一需要新语义的一格：强制人工确认",
      ls.apply_to_card("候选", ls.STATE_DOUBT)["force_confirm"])

    print("计划日与数据日差一天")
    t("次日推算正确", plan._next_day("2026-09-03") == "2026-09-04")  # noqa: SLF001
    t("认不出的日期原样返回，不抛异常", plan._next_day("不是日期") == "不是日期")  # noqa: SLF001

    print("乙路历史传到计划装配（审查 2026-09-07 第 9 条）")
    cur = jrow(leaning="偏空")
    cur["cluster_key"] = "Segment::固态电解质"
    hist = {"Segment::固态电解质": [
        {"plan_date": "2026-09-01", "leaning": "偏空", "leaning_prev": "偏多", "migrated": 1},
        {"plan_date": "2026-09-02", "leaning": "偏空", "leaning_prev": "偏空", "migrated": 0}]}
    st = plan._logic_state_of("SH600000", {"logic": []}, {"SH600000": ["固态电解质"]},  # noqa: SLF001
                              {"固态电解质": cur}, {}, DS, seg_hist=hist)
    t("近日翻空且一直偏空 -> 硬触发维持 -> 逻辑存疑", st["state"] == ls.STATE_DOUBT)
    st = plan._logic_state_of("SH600000", {"logic": []}, {"SH600000": ["固态电解质"]},  # noqa: SLF001
                              {"固态电解质": cur}, {}, DS)
    t("不传历史时退回只看当天一行（不断产）", st["state"] != ls.STATE_DOUBT)
    full = [claim("2026-07-10"), claim("2026-07-20"), claim("2026-08-28", "利空")]
    st = plan._logic_state_of("SH600000", {"logic": full[-1:]}, {}, {}, {}, DS, full_logic=full)  # noqa: SLF001
    t("甲路吃全量论断：三条里判得出跨期翻转", any(p["path"] == ls.PATH_CLAIM and p["signal"] == ls.SIG_DOWN
                                          for p in st["paths"]) and st["state"] == ls.STATE_DOUBT)

    print("ALL OK — 四态随卡产出 / 缺席不断产 / 下游形状 / 判决不被改动 / 计划日推算 / 乙路历史与甲路全量 全部通过")


if __name__ == "__main__":
    main()