tradingSystem/scripts/test_batch9_units.py

# -*- coding: utf-8 -*-
"""
第九批模块单测 (零外部依赖, 不连库不触网)
==========================================
运行: 在 tradingSystem 仓库根目录执行  python scripts/test_batch9_units.py
覆盖: 接管既有持仓前的成本价体检 (app/core/rebuild_check.py)。

这一批守的是一个**只发生一次、且不可逆**的决定: 账本清空后按「以下游为准」认领真实持仓,
那一刻定死每只票的开仓价 → 摊薄成本 → 安全垫 → 补仓/加仓/保垫减仓的共同判据。
2026-07-29 首次接管 22 只持仓时踩过一次 (全按现价开仓, 安全垫齐刷刷是 0)。

所以用例的重点不是"算得对不对", 而是**该拦的拦不拦得住**, 以及**不该拦的会不会误伤** ——
误伤的代价是多等一天, 漏拦的代价是一本每个数都错、却和真账长得一模一样的账。
"""
import os
import sys
import traceback

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

from app.core import rebuild_check as rb                                       # noqa: E402

RESULTS = []


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


def _p(code, qty=1000, cost=10.0, avail=None, price=None):
    """一行下游持仓 (downstream_repo.fetch_positions 的行形态)。"""
    return {"ts_code": code, "qty": qty, "cost": cost,
            "avail_qty": qty if avail is None else avail, "price": price}


# ================================================================ [A] 单只票的判定
@case("[A1] 成本价正常 → OK, 并算出安全垫")
def t_a1():
    r = rb.check_row(_p("600000.SH", cost=20.0), price=10.0)
    assert r["verdict"] == rb.OK
    assert abs(r["cushion_pct"] - (-0.5)) < 1e-9, "真实成本 20、现价 10 就是实亏 50%"


@case("[A2] 成本价缺失 / 为 0 / 为负 → MISSING")
def t_a2():
    for bad in (None, 0, 0.0, -1.0, ""):
        r = rb.check_row(_p("600000.SH", cost=bad), price=10.0)
        assert r["verdict"] == rb.MISSING, f"cost={bad!r} 应判 MISSING"
    assert "安全垫恒为 0" in rb.check_row(_p("600000.SH", cost=0), price=10.0)["why"]


@case("[A3] 成本 ≈ 现价 → EQ_PRICE (安全垫≈0)")
def t_a3():
    r = rb.check_row(_p("600000.SH", cost=10.02), price=10.0)   # 差 0.2% < 0.5%
    assert r["verdict"] == rb.EQ_PRICE
    r2 = rb.check_row(_p("600000.SH", cost=10.30), price=10.0)  # 差 3% > 0.5%
    assert r2["verdict"] == rb.OK, "正常的小幅浮盈不该被当成'拿现价充数'"


@case("[A4] 成本与现价差两个数量级 → ABSURD (多半是单位错)")
def t_a4():
    assert rb.check_row(_p("600000.SH", cost=1000.0), price=10.0)["verdict"] == rb.ABSURD
    assert rb.check_row(_p("600000.SH", cost=0.05), price=10.0)["verdict"] == rb.ABSURD
    # 分/元 混用是最典型的一种: 成本记成 1002 分而现价是 10.02 元
    r = rb.check_row(_p("600000.SH", cost=1002.0), price=10.02)
    assert r["verdict"] == rb.ABSURD and "单位" in r["why"]


@case("[A5] 可用量 > 总量 或为负 → AVAIL_BAD")
def t_a5():
    assert rb.check_row(_p("600000.SH", qty=1000, avail=1500), price=10.0)["verdict"] \
        == rb.AVAIL_BAD
    assert rb.check_row(_p("600000.SH", qty=1000, avail=-1), price=10.0)["verdict"] \
        == rb.AVAIL_BAD
    # 当日买入是合法的: 可用 < 总量 (成本给个与现价不同的值, 免得撞上 EQ_PRICE)
    assert rb.check_row(_p("600000.SH", qty=1000, avail=0, cost=20.0),
                        price=10.0)["verdict"] == rb.OK


@case("[A6] 取不到现价 → NO_PRICE, 但成本本身仍算可用")
def t_a6():
    r = rb.check_row(_p("600000.SH", cost=20.0), price=None)
    assert r["verdict"] == rb.NO_PRICE and r["cushion_pct"] is None
    assert "成本值本身可用" in r["why"]


@case("[A7] 可用量的判定排在成本之前 —— 两个都坏时先报可用量")
def t_a7():
    r = rb.check_row(_p("600000.SH", qty=100, avail=999, cost=0), price=10.0)
    assert r["verdict"] == rb.AVAIL_BAD


# ================================================================ [B] 整批的阻断判据
@case("[B1] 全部正常 → 不阻断")
def t_b1():
    rows = [_p("600000.SH", cost=20.0), _p("600519.SH", cost=1500.0)]
    out = rb.check_costs(rows, {"600000.SH": 10.0, "600519.SH": 1800.0})
    assert out["blocking"] is False and out["reasons"] == []
    assert "可以建账" in out["hint"]


@case("[B2] 只要有一只成本不可用就阻断 —— 一本账里混一只错的也不行")
def t_b2():
    rows = [_p("600000.SH", cost=20.0), _p("600519.SH", cost=0)]
    out = rb.check_costs(rows, {"600000.SH": 10.0, "600519.SH": 1800.0})
    assert out["blocking"] is True
    assert "600519.SH:MISSING" in out["reasons"][0]


@case("[B3] 整组成本≈现价 → 阻断 (这就是 07-29 那个坑的模样)")
def t_b3():
    rows = [_p(f"60000{i}.SH", cost=10.0) for i in range(1, 6)]
    out = rb.check_costs(rows, {f"60000{i}.SH": 10.0 for i in range(1, 6)})
    assert out["blocking"] is True and out["eq_ratio"] == 1.0
    assert "安全垫会是 0" in out["reasons"][0]


@case("[B4] 少数几只成本≈现价 → 不阻断 (当日买入是正常的)")
def t_b4():
    rows = [_p("600000.SH", cost=20.0), _p("600519.SH", cost=1800.0),
            _p("600036.SH", cost=30.0), _p("601318.SH", cost=10.0)]
    px = {"600000.SH": 10.0, "600519.SH": 1500.0, "600036.SH": 40.0, "601318.SH": 10.0}
    out = rb.check_costs(rows, px)
    assert out["counts"].get(rb.EQ_PRICE) == 1
    assert out["blocking"] is False, "四只里一只当日买入很正常, 不该拦"


@case("[B5] 只有一只票时不按占比阻断 —— 样本太小, 它可能真是当日买的")
def t_b5():
    out = rb.check_costs([_p("600000.SH", cost=10.0)], {"600000.SH": 10.0})
    assert out["counts"].get(rb.EQ_PRICE) == 1
    assert out["blocking"] is False


@case("[B6] 取不到现价的票不稀释 EQ_PRICE 占比")
def t_b6():
    # 两只成本≈现价 + 八只取不到现价。若拿 10 做分母, 占比 20% 不阻断 —— 那是错的:
    # 能判的两只全是 EQ_PRICE, 该阻断。
    rows = [_p("600001.SH", cost=10.0), _p("600002.SH", cost=10.0)] + \
           [_p(f"6001{i:02d}.SH", cost=5.0) for i in range(1, 9)]
    px = {"600001.SH": 10.0, "600002.SH": 10.0}          # 其余取不到价
    out = rb.check_costs(rows, px)
    assert out["counts"].get(rb.NO_PRICE) == 8
    assert out["eq_ratio"] == 1.0 and out["blocking"] is True


@case("[B7] 零持仓行不参与体检")
def t_b7():
    rows = [_p("600000.SH", qty=0, cost=0), _p("600519.SH", cost=1800.0)]
    out = rb.check_costs(rows, {"600519.SH": 1500.0})
    assert out["n"] == 1 and out["blocking"] is False, "qty=0 的行是历史残留, 不该拖累建账"


@case("[B8] 下游一只持仓都没有 → 不阻断但说清是没账可建")
def t_b8():
    out = rb.check_costs([], {})
    assert out["n"] == 0 and out["blocking"] is False
    assert "无账可建" in out["hint"]


@case("[B9] 阻断时的 hint 要给出可执行的下一步")
def t_b9():
    out = rb.check_costs([_p("600000.SH", cost=0)], {"600000.SH": 10.0})
    assert "cost_price" in out["hint"], "得说清要对端改哪一列, 不是只说'数据有问题'"
    assert "长得一模一样" in out["hint"], "得说清为什么不能将就着建"


# ================================================================ [C] 情形覆盖 (不阻断)
@case("[C1] 四种情形齐全 → enough")
def t_c1():
    rows = [_p("600000.SH", cost=10.0),                        # 浮盈 +50%
            _p("600519.SH", cost=20.0),                        # 浮亏 −25%
            _p("600036.SH", cost=10.0, qty=1000, avail=0),     # 当日买入
            _p("601318.SH", cost=10.0)]
    px = {"600000.SH": 15.0, "600519.SH": 15.0, "600036.SH": 11.0, "601318.SH": 11.0}
    cov = rb.coverage(rb.check_costs(rows, px)["rows"])
    assert cov["enough"] is True and cov["missing"] == []


@case("[C2] 全是不赚不亏 → 报出这轮验不到哪些纪律")
def t_c2():
    rows = [_p(f"60000{i}.SH", cost=10.0) for i in range(1, 4)]
    cov = rb.coverage(rb.check_costs(rows, {f"60000{i}.SH": 10.05 for i in range(1, 4)})["rows"])
    assert cov["enough"] is False
    assert any("浮盈" in m for m in cov["missing"])
    assert any("浮亏" in m for m in cov["missing"])


@case("[C3] 只有一只持仓 → 报持仓不足 3 只")
def t_c3():
    cov = rb.coverage(rb.check_costs([_p("600000.SH", cost=20.0)], {"600000.SH": 10.0})["rows"])
    assert cov["enough"] is False and any("不足 3 只" in m for m in cov["missing"])


@case("[C4] 覆盖不全绝不阻断建账 —— 数据是真的就该建")
def t_c4():
    rows = [_p("600000.SH", cost=20.0), _p("600519.SH", cost=3000.0)]
    out = rb.check_costs(rows, {"600000.SH": 10.0, "600519.SH": 1500.0})
    cov = rb.coverage(out["rows"])
    assert out["blocking"] is False and cov["enough"] is False


# ================================================================ [D] 健壮性
@case("[D1] 脏数据不炸")
def t_d1():
    for bad in ({"ts_code": "600000.SH"}, {"ts_code": None, "qty": "x", "cost": "y"},
                {"ts_code": "600000.SH", "qty": "1000", "cost": "20.0", "avail_qty": "1000"}):
        rb.check_row(bad, price=10.0)
    out = rb.check_costs([{"ts_code": "600000.SH", "qty": "1000", "cost": "20.0"}],
                         {"600000.SH": "10.0"})
    assert out["n"] == 1, "字符串数字要能吃进去 —— 库里 DECIMAL 列取出来常是字符串"


@case("[D2] rows 为 None 不炸")
def t_d2():
    assert rb.check_costs(None, None)["n"] == 0
    assert rb.coverage(None)["enough"] is False


@case("[D3] 判定常量互不相同 —— 别把两种故障混成一个码")
def t_d3():
    vs = [rb.OK, rb.MISSING, rb.EQ_PRICE, rb.ABSURD, rb.AVAIL_BAD, rb.NO_PRICE]
    assert len(set(vs)) == len(vs)


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()