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():
    # 既有设计对"某一只没成本价"有逐只回退现价的路径 (build_recon_fixes, 带 price_source
    # 留痕)。闸不该推翻它 —— 闸管的是"一本从头就错的账", 不是替既有设计做二次判断。
    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 False
    assert out["estimated"] == ["600519.SH"]
    assert "改不回来" in out["hint"], "估出来的成本后续对账不会再碰, 这句必须说出来"


@case("[B2b] 全部没成本价 → 阻断 (对端那一列整个没填)")
def t_b2b():
    rows = [_p("600000.SH", cost=0), _p("600519.SH", cost=None)]
    out = rb.check_costs(rows, {"600000.SH": 10.0, "600519.SH": 1800.0})
    assert out["blocking"] is True and out["estimated"] == []
    assert "全部" in out["reasons"][0]


@case("[B2c] 数据是**错的**而不是缺的 → 一只就阻断")
def t_b2c():
    # ABSURD/AVAIL_BAD 说明这批数据的生产方式有问题, 不该只怀疑那一只
    for bad in ({"cost": 1000.0}, {"avail": 9999}):
        rows = [_p("600000.SH", cost=20.0), _p("600519.SH", **bad)]
        out = rb.check_costs(rows, {"600000.SH": 10.0, "600519.SH": 10.0})
        assert out["blocking"] is True, bad
        assert "错的" 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)


# ================================================================ [E] 连续不一致按日推进
# 2026-07-31 实机暴露: 日报关注区写出「连续 175 日不一致」, 而项目 07-14 才开工。
# 原因是每调一次 reconcile() 就 +1, 而盘中轻对账每分钟调一次 —— 设计里「连续 3 日 → ERROR
# 待人工」实际成了「连续 3 分钟」。假警报天天响, 真告警就被埋掉。
from app.core import recon as rc                                               # noqa: E402


@case("[E1] 同一天内反复对账不重复计数")
def t_e1():
    s = rc.advance_streak(0, 0, 20260731, True)
    assert s["streak"] == 1 and s["changed"] is True
    for _ in range(5):                       # 手工点五次「对账」
        s = rc.advance_streak(s["streak"], s["ymd"], 20260731, True)
    assert s["streak"] == 1, "同一天点几次都只算一天 —— 否则三分钟就升 ERROR"
    assert s["changed"] is False


@case("[E2] 跨交易日才 +1")
def t_e2():
    s = rc.advance_streak(1, 20260731, 20260801, True)
    assert s["streak"] == 2 and s["ymd"] == 20260801
    s = rc.advance_streak(s["streak"], s["ymd"], 20260803, True)   # 跨周末
    assert s["streak"] == 3


@case("[E3] 差异消失立刻归零, 不必等下一天")
def t_e3():
    s = rc.advance_streak(7, 20260731, 20260731, False)
    assert s["streak"] == 0 and s["changed"] is True and s["ymd"] == 20260731


@case("[E4] 本来就是 0 且没差异 → 什么都没变, 不必写库")
def t_e4():
    assert rc.advance_streak(0, 20260731, 20260731, False)["changed"] is False


@case("[E5] 三日门槛与 severity 对得上")
def t_e5():
    assert rc.recon_severity(0) == rc.SEV_OK
    assert rc.recon_severity(1) == rc.SEV_WARN and rc.recon_severity(2) == rc.SEV_WARN
    assert rc.recon_severity(3) == rc.SEV_ERROR
    # 走满三个交易日才该到 ERROR —— 这条串起来验, 免得两边各改一半
    s = {"streak": 0, "ymd": 0}
    for i, d in enumerate((20260731, 20260801, 20260803), start=1):
        s = rc.advance_streak(s["streak"], s["ymd"], d, True)
        assert rc.recon_severity(s["streak"]) == (rc.SEV_ERROR if i >= 3 else rc.SEV_WARN)


@case("[E6] 缺日期时退化成每次都推进, 但不会把已有计数弄丢")
def t_e6():
    s = rc.advance_streak(2, 0, 0, True)     # 两个 ymd 都拿不到
    assert s["streak"] == 3, "判不了是不是同一天就按保守走(照常推进), 别把计数清零"



# ================================================================ [F] 行业占比闸的双判据
# 2026-07-31 实机暴露: 空账本 + 10 只强传导候选 + 一条 60% 升仓命令 → 一条方案都出不来。
# 因为只看"占组合"的话, 第一只票按定义就是组合的 100%, 必然超任何小于 100% 的上限;
# 而它被拒后组合市值不推进, 后面每一只面对的还是 100% —— 哪怕分属十个不同行业。
# 行业源是当天才通的 (此前 sector 恒为 None、整段跳过), 所以这个洞一直藏着。
from app.core.sizer import check_caps                                          # noqa: E402

SCALE = 2_000_000


def _ctx(**kw):
    d = dict(scale=SCALE, portfolio_cap=0.60, stock_cap=0.08, max_names=15,
             portfolio_mv=0.0, names_count=0, stock_mv=0.0, is_new_name=True,
             sector="储能", sector_names=0, sector_mv=0.0,
             sector_max_names=4, sector_max_ratio=0.40)
    d.update(kw)
    return d


@case("[F1] 空账本的第一笔买入不该被行业占比闸拦下")
def t_f1():
    bad = check_caps(ts_code="600000.SH", add_amount=0.06 * SCALE, ctx=_ctx())
    assert bad == [], f"第一只票必然是组合的 100%, 拦它等于建不了仓: {bad}"


@case("[F2] 候选分属不同行业时, 能一路建到总仓上限")
def t_f2():
    c, n = _ctx(), 0
    for i in range(1, 15):
        c["sector"], c["sector_mv"], c["sector_names"] = f"行业{i}", 0.0, 0
        if check_caps(ts_code=f"股{i}", add_amount=0.06 * SCALE, ctx=c):
            break
        c["portfolio_mv"] += 0.06 * SCALE
        c["names_count"] += 1
        n += 1
    assert n == 10 and abs(c["portfolio_mv"] / SCALE - 0.60) < 1e-9, (
        f"6% 一只、总仓上限 60% → 应正好进 10 只, 实际 {n} 只")


@case("[F3] 组合建到接近上限时, 行业占比闸照常拦")
def t_f3():
    # 组合 52%(=104万), 储能已占 38%(=76万), 再买 6% → 组合 58% (未撞总仓闸),
    # 储能占组合 75.9% > 40%, 占规模 44% > 绝对线 24% —— 两条都成立, 该拦
    c = _ctx(portfolio_mv=0.52 * SCALE, sector_mv=0.38 * SCALE, names_count=9, sector_names=3)
    bad = check_caps(ts_code="X", add_amount=0.06 * SCALE, ctx=c)
    assert any("SECTOR_RATIO" in b for b in bad), bad
    assert not any("PORTFOLIO_CAP" in b for b in bad), "这条用例要单独验行业判据"


@case("[F4] 占比超但绝对敞口小 → 不拦 (集中度是风险的放大器, 不是风险本身)")
def t_f4():
    # 组合只有 12%(=24万) 且全在储能: 占组合 100% 超上限, 但只占规模 18% < 绝对线 24%
    c = _ctx(portfolio_mv=0.12 * SCALE, sector_mv=0.12 * SCALE, names_count=2, sector_names=2)
    assert check_caps(ts_code="X", add_amount=0.06 * SCALE, ctx=c) == []


@case("[F5] 绝对线不会单独触发 —— 它只用来豁免建仓初期, 不会额外拦人")
def t_f5():
    """绝对线取 `sector_max_ratio × portfolio_cap` 是有讲究的: 组合在总仓上限以内时,
    "占规模超绝对线" 必然蕴含 "占组合超上限"。所以这条判据**只会放宽、不会收紧** ——
    它把建仓初期那段不可满足的区间豁免掉, 而不改变组合建起来之后的口径。
    取值再大一点 (比如直接用 sector_max_ratio) 就会变成一道独立的、更严的闸。"""
    ratio, cap = 0.40, 0.60
    for port_pct in (0.06, 0.12, 0.24, 0.36, 0.48, 0.60):
        for sec_pct in (0.02, 0.06, 0.12, 0.20, 0.28, 0.36):
            if sec_pct > port_pct:
                continue
            over_scale = sec_pct > ratio * cap + 1e-9
            over_port = (sec_pct / port_pct) > ratio + 1e-9
            assert not (over_scale and not over_port), (
                f"组合 {port_pct:.0%} 行业 {sec_pct:.0%}: 绝对线单独触发了")


@case("[F6] 同一行业的只数上限仍然管用 (占比放宽不等于行业闸失效)")
def t_f6():
    c, n = _ctx(), 0
    for i in range(1, 9):
        if check_caps(ts_code=f"股{i}", add_amount=0.06 * SCALE, ctx=c):
            break
        c["portfolio_mv"] += 0.06 * SCALE
        c["sector_mv"] += 0.06 * SCALE
        c["names_count"] += 1
        c["sector_names"] += 1
        n += 1
    assert n == 4, f"同一行业最多 4 只 (SECTOR_NAMES), 实际 {n}"


@case("[F7] 行业源没配时整段跳过 (约束停用而不是误拦)")
def t_f7():
    assert check_caps(ts_code="X", add_amount=0.06 * SCALE, ctx=_ctx(sector=None)) == []


# ================================================================ [G] 拒绝原因聚合
from app.core.planner import summarize_rejects                                 # noqa: E402


@case("[G1] 把 rejects 聚合成一行人话, 按条数降序")
def t_g1():
    rej = [{"ts_code": f"股{i}", "reasons": ["SECTOR_RATIO: 行业[储能]占组合将达 100.0% > 上限 40%"]}
           for i in range(5)] + \
          [{"ts_code": "股X", "reasons": ["STOCK_CAP: 股X 加后 9.0% > 单股上限 8%"]}]
    s = summarize_rejects(rej)
    assert s.startswith("SECTOR_RATIO 5 只"), s
    assert "STOCK_CAP 1 只" in s


@case("[G2] 一只票撞多条判据时每条都计数")
def t_g2():
    s = summarize_rejects([{"ts_code": "股A", "reasons": ["STOCK_CAP: …", "SECTOR_RATIO: …"]}])
    assert "STOCK_CAP 1 只" in s and "SECTOR_RATIO 1 只" in s


@case("[G3] 没有拒绝时返回空串 (调用方据此区分'拒了'和'本来就没候选')")
def t_g3():
    assert summarize_rejects([]) == "" and summarize_rejects(None) == ""


@case("[G4] 没有判据码的原因 (一手不可行那类) 也归得了类")
def t_g4():
    s = summarize_rejects([{"ts_code": "股A",
                            "reasons": ["目标金额 3000 元按现价 45.00 买不足一手"]}])
    assert "1 只" in s and "股A" in s


@case("[G5] 无码原因**先抹掉数字再归类** —— 价格各不相同也要聚成一组")
def t_g5():
    # 2026-08-03 实机: 8 只票同一个原因、8 个不同价格, 原来按 text[:20] 硬截,
    # 价格落在截断范围内 → 聚成 8 组一组一只 (聚合等于没做), 而且截断落在词中间,
    # 印出「…按现价 18.68 买」「…按现价 6.47 买不」这种半句话。
    rej = [{"ts_code": c, "reasons": [f"目标金额 60 元按现价 {p} 买不足一手 "
                                      f"(已尝试: CUSTOM → BASE/FILL → BASE)"]}
           for c, p in (("600764.SH", "18.68"), ("000800.SZ", "6.47"),
                        ("600104.SH", "11.22"), ("002594.SZ", "95.77"))]
    s = summarize_rejects(rej)
    assert s.count("只") == 1, f"没聚成一组: {s}"
    assert "4 只" in s, s
    assert "买不足一手" in s, f"截断落在词中间了: {s}"
    assert "18.68" not in s and "95.77" not in s, f"具体价格不该出现在聚合标签里: {s}"
    assert "已尝试" not in s, f"括号尾注是排查细节, 不该进分类标签: {s}"


@case("[G6] 有码与无码混在一起时各归各的, 仍按条数降序")
def t_g6():
    rej = [{"ts_code": "A", "reasons": ["目标金额 60 元按现价 1.00 买不足一手"]},
           {"ts_code": "B", "reasons": ["目标金额 60 元按现价 2.00 买不足一手"]},
           {"ts_code": "C", "reasons": ["目标金额 60 元按现价 3.00 买不足一手"]},
           {"ts_code": "D", "reasons": ["SECTOR_RATIO: 行业[白酒]占组合将达 55.0% > 上限 40%"]},
           {"ts_code": "E", "reasons": ["STOCK_CAP: 单股占规模将达 9.1% > 上限 8%"]}]
    s = summarize_rejects(rej)
    assert s.index("3 只") < s.index("SECTOR_RATIO"), f"没按条数降序: {s}"
    assert "SECTOR_RATIO 1 只" in s and "STOCK_CAP 1 只" in s, s


# ================================================================ [H] 一手取整的记账口径
# 2026-07-31 实机: 命令进度写着 planned_amount=600,000 / gap=0 ("完全满足"), 而 15 条方案
# 的金额合计只有 573,883 —— 少买 26,117 元 (4.4%) 且账面上完全看不出来。
# 原因是 plan_increase_exposure 的②分支按**理论目标**累计, 而①分支按取整后的实际金额累计,
# 同一个函数隔二十行两种写法。
from app.core import planner as pl                                             # noqa: E402


def _cx(**kw):
    d = dict(scale=SCALE, portfolio_cap=0.60, stock_cap=0.08, max_names=15,
             portfolio_mv=0.0, names_count=0, stock_mv=0.0, is_new_name=True,
             sector=None, sector_names=0, sector_mv=0.0,
             sector_max_names=4, sector_max_ratio=0.40)
    d.update(kw)
    return d


@case("[H1] planned_amount 必须等于各条方案金额之和")
def t_h1():
    r = pl.plan_increase_exposure(
        add_amount=0.30 * SCALE, positions=[],
        candidates=[{"ts_code": f"股{i}.SH", "price": p, "score": 1.0 - i * 0.01}
                    for i, p in enumerate([24.83, 25.22, 35.21, 136.73, 35.04])],
        ctx=_cx(), params={"stock_target_default": 0.06})
    assert abs(sum(i["amount"] for i in r["items"]) - r["planned_amount"]) < 1e-6, (
        f"账面 {r['planned_amount']:,.0f} vs 方案合计 "
        f"{sum(i['amount'] for i in r['items']):,.0f}")


@case("[H2] 高价股的取整损失要记进 rounding_gap, 不能被吞掉")
def t_h2():
    # 136.73 元一股, 一手就要 13,673 元 —— 目标 12 万只能买到 800 股 = 109,384
    r = pl.plan_increase_exposure(
        add_amount=0.06 * SCALE, positions=[],
        candidates=[{"ts_code": "002353.SZ", "price": 136.73, "score": 1.0}],
        ctx=_cx(), params={"stock_target_default": 0.06})
    assert abs(r["planned_amount"] - 109_384) < 1.0, r["planned_amount"]
    assert abs(r["rounding_gap"] - (120_000 - 109_384)) < 1.0, r["rounding_gap"]
    assert any("一手取整" in n for n in r["notes"]), r["notes"]


@case("[H3] 取整零头不算命令失败, 候选不够才算")
def t_h3():
    # 候选够: 目标全分配出去了, 只是取整少买一点 → ok
    ok_r = pl.plan_increase_exposure(
        add_amount=0.06 * SCALE, positions=[],
        candidates=[{"ts_code": "002353.SZ", "price": 136.73, "score": 1.0}],
        ctx=_cx(), params={"stock_target_default": 0.06})
    assert ok_r["ok"] is True and ok_r["shortfall"] == 0 and ok_r["rounding_gap"] > 0
    # 候选不够: 要 30% 却只有一只票能给 6% → 不 ok
    bad_r = pl.plan_increase_exposure(
        add_amount=0.30 * SCALE, positions=[],
        candidates=[{"ts_code": "002353.SZ", "price": 136.73, "score": 1.0}],
        ctx=_cx(), params={"stock_target_default": 0.06})
    assert bad_r["ok"] is False and bad_r["shortfall"] > 0
    assert any("候选与补仓空间不足" in n for n in bad_r["notes"])


@case("[H4] 组合市值按实际金额推进, 不按理论目标 —— 否则后面几只被虚高的仓位拦掉")
def t_h4():
    r = pl.plan_increase_exposure(
        add_amount=0.60 * SCALE, positions=[],
        candidates=[{"ts_code": f"股{i}.SH", "price": 136.73, "score": 1.0 - i * 0.01}
                    for i in range(10)],
        ctx=_cx(), params={"stock_target_default": 0.06})
    names = {i["ts_code"] for i in r["items"]}
    assert len(names) == 10, (
        f"十只高价股每只 6%, 按实际金额推进才装得下 10 只; 按理论目标会提前收敛, "
        f"实际装了 {len(names)} 只")


@case("[H5] 个股建仓命令的 gap 也要如实反映取整损失")
def t_h5():
    r = pl.plan_open_target(ts_code="002353.SZ", price=136.73, target_pct=0.06,
                            ctx=_cx(), params={})
    assert r["ok"] and r["gap"] > 0, f"原来写死 gap=0, 等于说'完全建到目标了': {r}"
    assert abs(r["planned_amount"] - sum(i["amount"] for i in r["items"])) < 1e-6



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