333 lines
13 KiB
Python
333 lines
13 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""三源合议 · 工作包二纯逻辑 (2026-09-11 方案第四节, 台账 007/009)。全部离线, 不连库。
|
||
A 基本面 fund_rules: 质地档映射、质地中整体为好、共识已定价、缺与超期无读数。
|
||
B 择时 timing_rules: 昨夜定性映射、盘中转多升级、超期无读数。
|
||
C 合议 consensus: 多数决方向、平局中性、强弱、附录丙路由表逐条。
|
||
"""
|
||
import unit_test_guard # noqa: F401，单独运行也禁止联网
|
||
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"] == "不建"
|
||
|
||
|
||
# ================================================================ F 加固包 (2026-09-14)
|
||
@case("C 路由·弱表态开着 (weak_confirm 真): 方向看多但只有一方表态 → 交人")
|
||
def _():
|
||
r = cons.decide("看多", "无读数", "无读数", weak_confirm=True)
|
||
assert r["direction"] == "看多" and r["route"] == "交人"
|
||
|
||
|
||
@case("C 路由·弱表态关掉 (weak_confirm 假): 方向看多但只有一方表态 → 放行")
|
||
def _():
|
||
r = cons.decide("看多", "无读数", "无读数", weak_confirm=False)
|
||
assert r["direction"] == "看多" and r["route"] == "放行"
|
||
|
||
|
||
@case("D 装配·持仓行昨夜中性当日转多 → 择时票升看多 (decorate_positions 带 flip_map)")
|
||
def _():
|
||
from app.services import consensus_service as csv
|
||
from app.core import tradedays as _td
|
||
rows = [{"ts_code": "600000.SH"}]
|
||
csv.decorate_positions(
|
||
rows, nightly={"600000.SH": {"verdict": "WATCH", "trade_date": _td.ymd()}},
|
||
tech_states={}, company_reviews={"600000.SH": _cr(overall="好")},
|
||
flip_map={"600000.SH": "10:30"}, params=_ASM_P)
|
||
tmb = rows[0]["consensus_blocks"]["timing"]
|
||
assert tmb["stance"] == "看多" and tmb["intraday_flip_at"] == "10:30"
|
||
|
||
|
||
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())
|