tradingSystem/scripts/test_batch28_units.py

# -*- coding: utf-8 -*-
"""三源合议 · 工作包二纯逻辑 (2026-09-11 方案第四节, 台账 007/009)。全部离线, 不连库。
  A 基本面 fund_rules: 质地档映射、质地中整体为好、共识已定价、缺与超期无读数。
  B 择时 timing_rules: 昨夜定性映射、盘中转多升级、超期无读数。
  C 合议 consensus: 多数决方向、平局中性、强弱、附录丙路由表逐条。
"""
import os
import sys
import traceback

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

from app.core import fund_rules as fr          # noqa: E402
from app.core import timing_rules as tm        # noqa: E402
from app.core import consensus as cons         # noqa: E402

RESULTS = []


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


def _cr(overall="好", valuation="便宜", good=8, mid=2, bad=0, groups=None, doubt=False, age=0):
    return {"overall": overall, "valuation": valuation,
            "grade_counts": {"好": good, "中": mid, "差": bad, "证据不足": 0},
            "groups": groups or {"回报与护城河": "好", "盈余质量与财务安全": "好", "成长与含金量": "中"},
            "doubt_hard": doubt, "age_days": age}


# ================================================================ A 基本面
@case("A 基本面: 质地好为看多")
def _():
    assert fr.synthesize(_cr(overall="好"))["stance"] == "看多"


@case("A 基本面: 质地中但三类两好零差、十视角差≤1 → 看多（整体方向为好就够）")
def _():
    r = fr.synthesize(_cr(overall="中", bad=1,
                          groups={"回报与护城河": "好", "盈余质量与财务安全": "好", "成长与含金量": "中"}))
    assert r["stance"] == "看多", r


@case("A 基本面: 质地中只有一类好 → 中性")
def _():
    r = fr.synthesize(_cr(overall="中", bad=0,
                          groups={"回报与护城河": "好", "盈余质量与财务安全": "中", "成长与含金量": "中"}))
    assert r["stance"] == "中性", r


@case("A 基本面: 质地中两好但十视角差>1 → 中性")
def _():
    r = fr.synthesize(_cr(overall="中", bad=2,
                          groups={"回报与护城河": "好", "盈余质量与财务安全": "好", "成长与含金量": "中"}))
    assert r["stance"] == "中性", r


@case("A 基本面: 质地差看空")
def _():
    assert fr.synthesize(_cr(overall="差"))["stance"] == "看空"


@case("A 基本面: 有硬疑点看空（哪怕质地不差）")
def _():
    assert fr.synthesize(_cr(overall="中", doubt=True))["stance"] == "看空"


@case("A 基本面: 缺评析无读数, 绝不折看空")
def _():
    assert fr.synthesize(None)["stance"] == "无读数"
    assert fr.synthesize({})["stance"] == "无读数"


@case("A 基本面: 报告超 120 天无读数")
def _():
    assert fr.synthesize(_cr(overall="好", age=121))["stance"] == "无读数"
    assert fr.synthesize(_cr(overall="好", age=120))["stance"] == "看多"


@case("A 基本面: 十视角好≥8且估值贵 → 共识已定价标记, 立场不变")
def _():
    r = fr.synthesize(_cr(overall="好", good=8, valuation="贵"))
    assert r["stance"] == "看多" and r["mark"] == "共识已定价" and r["priced_in"] is True


@case("A 基本面: 好视角够但估值不贵 → 不打共识已定价")
def _():
    assert fr.synthesize(_cr(overall="好", good=9, valuation="便宜"))["mark"] is None


# ================================================================ B 择时
@case("B 择时: BUY看多 / WATCH中性 / SELL·AVOID·DROPPED看空")
def _():
    assert tm.synthesize("BUY")["stance"] == "看多"
    assert tm.synthesize("WATCH")["stance"] == "中性"
    for v in ("SELL", "AVOID", "DROPPED"):
        assert tm.synthesize(v)["stance"] == "看空", v


@case("B 择时: 缺或超期无读数")
def _():
    assert tm.synthesize(None)["stance"] == "无读数"
    assert tm.synthesize("BUY", fresh=False)["stance"] == "无读数"


@case("B 择时: 盘中转多把无读数与中性升看多")
def _():
    assert tm.synthesize(None, intraday_flip_at="10:05")["stance"] == "看多"
    assert tm.synthesize("WATCH", intraday_flip_at="10:05")["stance"] == "看多"


@case("B 择时: 盘中转多与昨夜看空相遇归中性; 昨夜看多维持看多")
def _():
    assert tm.synthesize("SELL", intraday_flip_at="10:05")["stance"] == "中性"
    assert tm.synthesize("BUY", intraday_flip_at="10:05")["stance"] == "看多"


# ================================================================ C 合议
@case("C 合议·方向: 多数决, 无读数弃权, 平局中性")
def _():
    assert cons.decide("看多", "看多", "看空")["direction"] == "看多"
    assert cons.decide("看空", "看空", "看多")["direction"] == "看空"
    assert cons.decide("看多", "看空", "无读数")["direction"] == "中性"
    assert cons.decide("无读数", "无读数", "无读数", fund_required=False)["direction"] == "中性"


@case("C 合议·强弱: 一方表态弱, 三方一致强, 两方为中")
def _():
    assert cons.decide("看多", "无读数", "无读数")["strength"] == "弱"
    assert cons.decide("看多", "看多", "看多")["strength"] == "强"
    assert cons.decide("看多", "看多", "无读数")["strength"] == "中"


@case("C 路由: 基本面无读数且开关开 → 跳过没有买方评析")
def _():
    r = cons.decide("无读数", "看多", "看多", fund_required=True)
    assert r["route"] == "跳过" and "买方评析" in r["route_reason"]


@case("C 路由: 关掉 fund_required 后无评析不因此跳过")
def _():
    assert cons.decide("无读数", "看多", "看多", fund_required=False)["route"] != "跳过"


@case("C 路由: 方向看空 → 跳过")
def _():
    assert cons.decide("看空", "看空", "中性")["route"] == "跳过"


@case("C 路由: 中性且技术面收口等待且基本面非看空 → 观察等开口")
def _():
    r = cons.decide("中性", "中性", "中性", tech_phase="收口等待")
    assert r["route"] == "观察" and "开口" in r["route_reason"]


@case("C 路由: 中性且基本面看多技术面看空 → 观察等技术面转向")
def _():
    r = cons.decide("看多", "看空", "无读数")
    assert r["direction"] == "中性" and r["route"] == "观察" and "转向" in r["route_reason"]


@case("C 路由: 基本面看空技术面看多 → 交人试探仓")
def _():
    r = cons.decide("看空", "看多", "中性")
    assert r["route"] == "交人" and "试探" in r["route_reason"]


@case("C 路由: 看多但异议在择时系统 → 交人")
def _():
    r = cons.decide("看多", "看多", "看空")
    assert r["direction"] == "看多" and r["route"] == "交人" and "择时" in r["route_reason"]


@case("C 路由: 看多但只有一方表态 → 交人")
def _():
    r = cons.decide("看多", "无读数", "无读数")
    assert r["route"] == "交人" and "一方" in r["route_reason"]


@case("C 路由: 看多且多方表态、异议只在技术面 → 放行")
def _():
    r = cons.decide("看多", "看空", "看多")
    assert r["direction"] == "看多" and r["route"] == "放行"


@case("C 路由: 三方都无读数 → 放行机械方案")
def _():
    r = cons.decide("无读数", "无读数", "无读数", fund_required=False)
    assert r["route"] == "放行" and "机械" in r["route_reason"]


# ================================================================ D 装配服务 (脱库, 桩参数与映射)
_ASM_P = {"fund_required": True, "fund_stale_days": 120, "fund_consensus_good_min": 8,
          "consensus_route": True, "timing_stale_tdays": 3}


@case("D 装配: 三方看多 → 合议看多强放行, 六键齐全")
def _():
    from app.services import consensus_service as csv
    row = {"ts_code": "600000.SH", "company_review": _cr(overall="好")}
    r = csv.assemble(row, nightly={"verdict": "BUY", "trade_date": 20260910},
                     tech_states={"600000.SH": {"stance": "看多", "phase": "趋势多"}}, params=_ASM_P)
    assert r["consensus"]["direction"] == "看多" and r["consensus"]["route"] == "放行"
    assert set(csv.hard_keys(r).keys()) == {"fund_stance", "tech", "tech_text",
                                            "timing", "consensus", "consensus_text"}


@case("D 装配: 无评析 → 基本面无读数, 合议跳过没有买方评析")
def _():
    from app.services import consensus_service as csv
    r = csv.assemble({"ts_code": "000001.SZ"}, nightly=None, tech_states={}, params=_ASM_P)
    assert r["fund"]["stance"] == "无读数" and r["consensus"]["route"] == "跳过"


@case("D 装配: 昨夜定性超期 → 择时无读数, 绝不折看空")
def _():
    from app.services import consensus_service as csv
    row = {"ts_code": "600000.SH", "company_review": _cr(overall="好")}
    r = csv.assemble(row, nightly={"verdict": "BUY", "trade_date": 20260101},
                     tech_states={}, params=_ASM_P)
    assert r["timing"]["stance"] == "无读数"


# ================================================================ E advise_v2 仓位矩阵 (附录乙)
_MP = {"stock_target_default": 0.06, "batch_split": (0.5, 0.25, 0.25)}


def _av2(fs, ts, **kw):
    from app.services import advice_service as a
    c = {"valuation": kw.get("val"), "logic_state": ({"state": kw["ls"]} if kw.get("ls") else None),
         "pricing_state": ({"state": kw["ps"]} if kw.get("ps") else None)}
    fund = {"stance": fs, "priced_in": kw.get("priced", False), "valuation": kw.get("vlabel")}
    return a.advise_v2(c, _MP, fund=fund, tech={"stance": ts, "phase": kw.get("phase")})


@case("E 矩阵: 基本面看多×技术面看多 → 标准仓 5/2.5/2.5 放行")
def _():
    r = _av2("看多", "看多")
    assert r["tier"] == "标准仓" and r["splits"] == (0.5, 0.25, 0.25) and r["route"] == "放行"


@case("E 矩阵: 基本面看多×技术面看空 → 观察")
def _():
    assert _av2("看多", "看空")["route"] == "观察"


@case("E 矩阵: 基本面看多估值贵 → 减半仓 5/5")
def _():
    r = _av2("看多", "看多", vlabel="贵")
    assert r["tier"] == "减半仓" and r["splits"] == (0.5, 0.5)


@case("E 矩阵: 共识已定价 → 减半且首批推迟")
def _():
    r = _av2("看多", "看多", priced=True)
    assert r["tier"] == "减半仓" and r["delay_first"] is True


@case("E 矩阵: 基本面中性×技术面看多 → 标准仓 3/3.5/3.5")
def _():
    r = _av2("中性", "看多")
    assert r["tier"] == "标准仓" and r["splits"] == (0.3, 0.35, 0.35)


@case("E 矩阵: 基本面看空×技术面看多 → 试探仓1%交人配紧止盈")
def _():
    r = _av2("看空", "看多")
    assert (r["tier"] == "试探仓" and abs(r["target_pct"] - 0.01) < 1e-9
            and r["route"] == "交人" and r["tight_trail"] is True)


@case("E 矩阵: 基本面看空技术面非多 / 无读数 → 不建跳过")
def _():
    assert _av2("看空", "看空")["tier"] == "不建"
    assert _av2("无读数", "看多")["route"] == "跳过"


@case("E 矩阵: 逻辑存疑 → 不建（既有硬拦保留）")
def _():
    assert _av2("看多", "看多", ls="逻辑存疑")["tier"] == "不建"


def main():
    ok = 0
    for name, fn in RESULTS:
        try:
            fn()
            ok += 1
            print("  ok  " + name)
        except Exception:
            print("  FAIL " + name)
            traceback.print_exc()
    print("-" * 60)
    if ok == len(RESULTS):
        print("ALL PASS (%d cases)" % ok)
        return 0
    print("FAILED %d/%d" % (len(RESULTS) - ok, len(RESULTS)))
    return 1


if __name__ == "__main__":
    sys.exit(main())