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"]


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