# -*- coding: utf-8 -*-
"""观察读数包 (2026-09-14 方案第三节, 台账 013)。全部离线, 不连库。

把每轮扫描里合议相关的判定落进 pms_consensus_stat, 让台账 006 到 011 的复核有据。
  A 动作引擎打标签: 合议判跳过分 fund/bear、判观察打 wait_tech、增持门打 gate_block;
    开关关掉时标签不出现 (逐字回旧); tag 值与 consensus_stats 认的 kind 对齐。
  B 归类落表 record_round: 检查点写路由与装配两类、转空实弹随时写、试算不写、被闸拒不记、
    确认与否据动作名; daily_summary 取数失败返回空不抛。
  C 映射覆盖 record_map_cover: 记 map_cover 与逐只 tech_noread。
  D 仓库 upsert: 更新子句只用 VALUES 列不带绑定参数 (pymysql 批量陷阱), SQL 单表合规。
  E 参数登记与检查点窗口。
"""
import json
import os
import sys
import traceback
from datetime import datetime

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

from app.core import action_engine as ae                    # noqa: E402
from app.services import consensus_stats as cst             # noqa: E402
from app.repo import consensus_stat_repo as csr             # noqa: E402

RESULTS = []


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


# ================================================================ 公共桩
def _copen(code, route, rr, direction="中性"):
    """一条挂了合议块的候选 (scan_open 用)。"""
    return {"ts_code": code, "score": 100, "rank": 1,
            "consensus": {"route": route, "route_reason": rr, "direction": direction}}


def _caps():
    return {"max_names": 20, "sector_source_ready": True}


def _pos_con(code, direction):
    """一条挂了合议块的持仓行 (scan 增持门用)。"""
    return {"ts_code": code, "total_qty": 6000, "avail_qty": 6000, "price": 10.0,
            "avg_cost": 10.0, "base_qty": 6000, "add_qty": 0, "dca_qty": 0,
            "cushion_pct": 0.0, "cushion_peak": 0.0, "frozen_reason": "NONE",
            "price_ok": True, "target_pct": 0.06,
            "consensus_blocks": {"consensus": {"direction": direction},
                                 "tech": {"stance": direction}, "fund": {"stance": direction}}}


def _sp(**kw):
    p = {"cushion_solid": 0.03, "trim_peak": 0.06, "trim_giveback": 0.5,
         "dca_triggers": (-0.08, -0.15), "dca_deep_confirm": -0.15, "dca_max_ratio": 0.5,
         "no_chase_ma5": 0.06, "build_window_tdays": 10, "fill_max_loss": -0.03,
         "batch_split": (0.5, 0.25, 0.25), "stock_target_default": 0.06, "scale": 2_000_000,
         "logic_state_route": False, "tech_gate_increase": True}
    p.update(kw)
    return p


class _patched:
    """把 record_round 会碰的两处外部依赖换成桩: param_store.get 返回固定检查点串,
    consensus_stat_repo.upsert_many 捕获落表行 (不连库)。with 结束自动还原。"""

    def __init__(self, times="0935,1030,1330,1445"):
        self.times = times
        self.captured = []

    def __enter__(self):
        self._o_get = cst.param_store.get
        self._o_up = cst.consensus_stat_repo.upsert_many
        self._o_del = cst.consensus_stat_repo.delete_day_kind
        cst.param_store.get = lambda k, d=None: (self.times if k == cst.STAT_TIMES_KEY
                                                 else self._o_get(k, d))
        cst.consensus_stat_repo.upsert_many = lambda rows: (self.captured.extend(rows)
                                                            or len(rows))
        cst.consensus_stat_repo.delete_day_kind = lambda d, k: 0   # 离线不打库
        return self.captured

    def __exit__(self, *a):
        cst.param_store.get = self._o_get
        cst.consensus_stat_repo.upsert_many = self._o_up
        cst.consensus_stat_repo.delete_day_kind = self._o_del
        return False


def _by_kind(rows):
    out = {}
    for r in rows:
        out.setdefault(r["kind"], []).append(r)
    return out


def _sample_out():
    return {"dry_run": False, "rejected": [],
            "consensus_seen": [
                {"ts_code": "600000.SH", "fund": "看多", "tech": "看多", "timing": "看多",
                 "direction": "看多", "route": "放行", "phase": "趋势多"},
                {"ts_code": "600001.SH", "fund": "无读数", "tech": "中性", "timing": "看多",
                 "direction": "看多", "route": "交人", "phase": None},
                {"ts_code": "600010.SH", "fund": "看多", "tech": "看多", "timing": "无读数",
                 "direction": "看多", "route": "放行", "phase": "趋势多"}],
            "skipped": [
                {"ts_code": "600002.SH", "action": "OPEN", "why": "没有买方评析",
                 "tag": "open_skip_fund"},
                {"ts_code": "600003.SH", "action": "OPEN", "why": "方向看空（…）",
                 "tag": "open_skip_bear"},
                {"ts_code": "600004.SH", "action": "OPEN", "why": "等技术面开口",
                 "tag": "open_wait_tech", "disp": "wait_tech"},
                {"ts_code": "600005.SH", "action": "FILL", "why": "技术面看空",
                 "tag": "gate_block", "detail": {"gate_kind": "fill"}},
                {"ts_code": "600009.SH", "why": "宏观偏热闸生效"}]}   # 无 tag, 应被忽略


def _sample_scanned():
    return {"candidates": [
        {"ts_code": "600006.SH", "action": "EXIT", "source": "tech_exit", "qty": 1000,
         "reason": "确认转空清仓"},
        {"ts_code": "600007.SH", "action": "TRIM", "source": "tech_exit", "qty": 300,
         "reason": "未确认减三分之一"},
        {"ts_code": "600008.SH", "action": "FILL", "source": "engine", "qty": 500}]}


# ================================================================ A 动作引擎打标签
@case("A scan_open·合议判跳过·没有买方评析 → open_skip_fund")
def _():
    res = ae.scan_open(candidates=[_copen("600000.SH", "跳过", "没有买方评析")],
                       params={"consensus_route": True}, caps=_caps(),
                       room_amt=1e6, slots=10, skip={})
    assert len(res["skipped"]) == 1
    assert res["skipped"][0]["tag"] == "open_skip_fund"
    assert res["skipped"][0]["ts_code"] == "600000.SH"


@case("A scan_open·合议判跳过·方向看空 → open_skip_bear")
def _():
    res = ae.scan_open(candidates=[_copen("600000.SH", "跳过", "方向看空（基本面看空、技术面看空、择时中性）")],
                       params={"consensus_route": True}, caps=_caps(),
                       room_amt=1e6, slots=10, skip={})
    assert res["skipped"][0]["tag"] == "open_skip_bear"


@case("A scan_open·合议判观察 → open_wait_tech 且 disp wait_tech")
def _():
    res = ae.scan_open(candidates=[_copen("600000.SH", "观察", "等技术面开口")],
                       params={"consensus_route": True}, caps=_caps(),
                       room_amt=1e6, slots=10, skip={})
    assert res["skipped"][0]["tag"] == "open_wait_tech"
    assert res["skipped"][0]["disp"] == ae.DISP_WAIT_TECH


@case("A scan_open·合议开关关掉·跳过项不带 tag (开着才有, 逐字回旧)")
def _():
    cand = _copen("600000.SH", "跳过", "没有买方评析")
    on = ae.scan_open(candidates=[dict(cand)], params={"consensus_route": True},
                      caps=_caps(), room_amt=1e6, slots=10, skip={})
    # 关掉合议: con 为 None, 那只候选不再被合议分流; 用 skip 在 eval_open 之前接住它, 免起重设备
    off = ae.scan_open(candidates=[dict(cand)], params={"consensus_route": False},
                       caps=_caps(), room_amt=1e6, slots=10,
                       skip={("600000.SH", "OPEN"): "已在途"})
    assert on["skipped"][0].get("tag") == "open_skip_fund"
    assert all("tag" not in s for s in off["skipped"])


@case("A scan·增持门拦下 → gate_block 且 detail.gate_kind ∈ {fill,add,dca}")
def _():
    res = ae.scan(positions=[_pos_con("600000.SH", "看空")], params=_sp(tech_gate_increase=True),
                  market={"600000.SH": {}}, skip={})
    gb = [s for s in res["skipped"] if s.get("tag") == "gate_block"]
    assert gb, "增持门看空应打 gate_block 标签"
    assert all((s.get("detail") or {}).get("gate_kind") in ("fill", "add", "dca") for s in gb)


@case("A scan·增持门开关关掉 → 无 gate_block tag (逐字回旧)")
def _():
    res = ae.scan(positions=[_pos_con("600000.SH", "看空")], params=_sp(tech_gate_increase=False),
                  market={"600000.SH": {}}, skip={})
    assert all(s.get("tag") != "gate_block" for s in res["skipped"])


@case("A 哨兵·动作引擎 tag 与 consensus_stats 认的 kind 对齐")
def _():
    assert cst.K_OPEN_SKIP_FUND == ae.TAG_OPEN_SKIP_FUND == "open_skip_fund"
    assert cst.K_OPEN_SKIP_BEAR == ae.TAG_OPEN_SKIP_BEAR == "open_skip_bear"
    assert cst.K_OPEN_WAIT_TECH == ae.TAG_OPEN_WAIT_TECH == "open_wait_tech"
    assert cst.K_GATE_BLOCK == ae.TAG_GATE_BLOCK == "gate_block"


# ================================================================ B 归类落表 record_round
@case("B record_round·检查点写全部类别 (放行/交人/无读数/跳过/观察/增持门/转空实弹)")
def _():
    with _patched() as cap:
        r = cst.record_round(_sample_out(), _sample_scanned(), {}, datetime(2026, 9, 15, 9, 35))
    assert r["ok"]
    bk = _by_kind(cap)
    assert {x["ts_code"] for x in bk["open_pass"]} == {"600000.SH", "600010.SH"}
    assert {x["ts_code"] for x in bk["open_confirm"]} == {"600001.SH"}
    assert {x["ts_code"] for x in bk["fund_noread"]} == {"600001.SH"}
    assert {x["ts_code"] for x in bk["timing_noread"]} == {"600010.SH"}
    assert {x["ts_code"] for x in bk["open_skip_fund"]} == {"600002.SH"}
    assert {x["ts_code"] for x in bk["open_skip_bear"]} == {"600003.SH"}
    assert {x["ts_code"] for x in bk["open_wait_tech"]} == {"600004.SH"}
    assert {x["ts_code"] for x in bk["gate_block"]} == {"600005.SH"}
    assert {x["ts_code"] for x in bk["tech_exit"]} == {"600006.SH", "600007.SH"}
    assert all(x["stat_date"] == 20260915 for x in cap)          # 北京日期


@case("B record_round·gate_block 带上 detail.gate_kind; 无 tag 的跳过项被忽略")
def _():
    with _patched() as cap:
        cst.record_round(_sample_out(), _sample_scanned(), {}, datetime(2026, 9, 15, 10, 30))
    bk = _by_kind(cap)
    assert json.loads(bk["gate_block"][0]["detail"])["gate_kind"] == "fill"
    assert "600009.SH" not in {x["ts_code"] for x in cap}         # 无 tag 的宏观闸跳过不落


@case("B record_round·转空实弹据动作名记 confirm (EXIT 真 / TRIM 假)")
def _():
    with _patched() as cap:
        cst.record_round(_sample_out(), _sample_scanned(), {}, datetime(2026, 9, 15, 9, 35))
    ex = {x["ts_code"]: json.loads(x["detail"]) for x in _by_kind(cap)["tech_exit"]}
    assert ex["600006.SH"]["confirm"] is True and ex["600006.SH"]["qty"] == 1000
    assert ex["600007.SH"]["confirm"] is False and ex["600007.SH"]["qty"] == 300


@case("B record_round·非检查点只写转空实弹 (路由与装配两类不写)")
def _():
    with _patched() as cap:
        cst.record_round(_sample_out(), _sample_scanned(), {}, datetime(2026, 9, 15, 10, 0))
    bk = _by_kind(cap)
    assert set(bk) == {"tech_exit"}
    assert {x["ts_code"] for x in bk["tech_exit"]} == {"600006.SH", "600007.SH"}


@case("B record_round·试算 (dry_run) 一律不写")
def _():
    out = {**_sample_out(), "dry_run": True}
    with _patched() as cap:
        r = cst.record_round(out, _sample_scanned(), {}, datetime(2026, 9, 15, 9, 35))
    assert cap == [] and r["rows"] == 0


@case("B record_round·转空离场被闸拒的不记 tech_exit")
def _():
    out = {"dry_run": False, "consensus_seen": [], "skipped": [],
           "rejected": [{"ts_code": "600006.SH", "action": "EXIT"}]}
    with _patched() as cap:
        cst.record_round(out, _sample_scanned(), {}, datetime(2026, 9, 15, 10, 0))
    assert {x["ts_code"] for x in _by_kind(cap).get("tech_exit", [])} == {"600007.SH"}


@case("B record_round·空串检查点 → 只写转空实弹")
def _():
    with _patched(times="") as cap:
        cst.record_round(_sample_out(), _sample_scanned(), {}, datetime(2026, 9, 15, 9, 35))
    assert set(_by_kind(cap)) == {"tech_exit"}


@case("B daily_summary·取数失败返回空 dict 不抛")
def _():
    o = cst.consensus_stat_repo.count_by_kind

    def _boom(a, b):
        raise RuntimeError("db down")
    cst.consensus_stat_repo.count_by_kind = _boom
    try:
        assert cst.daily_summary(20260915) == {}
    finally:
        cst.consensus_stat_repo.count_by_kind = o


# ================================================================ C 映射覆盖
@case("C record_map_cover·记 map_cover (ts_code=*) 与逐只 tech_noread")
def _():
    with _patched() as cap:
        cst.record_map_cover(relevant_codes=["600000.SH", "600001.SH", "600002.SH"],
                             state_codes={"600000.SH", "600001.SH"},
                             now=datetime(2026, 9, 15, 8, 40))
    bk = _by_kind(cap)
    mc = bk["map_cover"][0]
    assert mc["ts_code"] == "*"
    assert json.loads(mc["detail"]) == {"codes": 3, "states": 2}
    assert {x["ts_code"] for x in bk["tech_noread"]} == {"600002.SH"}


@case("C record_map_cover·重建先清当日 tech_noread (以最新一次为准, 评审第三条)")
def _():
    deleted, upserted = [], []
    o_del, o_up = csr.delete_day_kind, csr.upsert_many
    csr.delete_day_kind = lambda d, k: (deleted.append((d, k)), 0)[1]
    csr.upsert_many = lambda rows: (upserted.extend(rows), len(rows))[1]
    try:
        cst.record_map_cover(relevant_codes=["600000.SH"], state_codes=set(),
                             now=datetime(2026, 9, 15, 8, 40))
    finally:
        csr.delete_day_kind, csr.upsert_many = o_del, o_up
    assert (20260915, cst.K_TECH_NOREAD) in deleted        # 先清当日 tech_noread
    assert any(r["kind"] == cst.K_MAP_COVER for r in upserted)  # 再写覆盖行


# ================================================================ D 仓库 upsert
@case("D upsert_many·更新子句只用 VALUES 列不带绑定参数; SQL 单表合规")
def _():
    from app.db.session import assert_single_table
    grabbed = {}
    o = csr.execute_many
    csr.execute_many = lambda sql, payload, **kw: (grabbed.update(sql=sql, payload=payload)
                                                   or len(payload))
    try:
        n = csr.upsert_many([{"stat_date": 20260915, "ts_code": "600000.SH", "kind": "open_pass",
                              "reason": "看多", "detail": None,
                              "first_at": datetime.now(), "last_at": datetime.now()}])
    finally:
        csr.execute_many = o
    sql = grabbed["sql"]
    upd = sql.split("ON DUPLICATE KEY UPDATE", 1)[1]
    assert ":" not in upd, "更新子句不许有绑定参数 (:col) —— 2026-09-11 pymysql 批量陷阱"
    assert "VALUES(reason)" in upd and "rounds = rounds + 1" in upd
    assert_single_table(sql)          # 不抛即单表合规 (守卫会摘掉 ON DUPLICATE 尾巴)
    assert n == 1


@case("D upsert_many·缺主键三件套的行被丢掉")
def _():
    grabbed = {}
    o = csr.execute_many
    csr.execute_many = lambda sql, payload, **kw: (grabbed.update(payload=payload)
                                                   or len(payload))
    try:
        # 只有一行齐全, 另两行分别缺 kind 与 ts_code
        n = csr.upsert_many([
            {"stat_date": 20260915, "ts_code": "600000.SH", "kind": "open_pass",
             "first_at": datetime.now(), "last_at": datetime.now()},
            {"stat_date": 20260915, "ts_code": "600001.SH"},
            {"stat_date": 20260915, "kind": "open_pass"}])
    finally:
        csr.execute_many = o
    assert n == 1 and len(grabbed["payload"]) == 1


# ================================================================ E 参数与检查点窗口
@case("E 参数·PMS_CONSENSUS_STAT_TIMES 登记, 默认四检查点, 类型 str")
def _():
    from app.services import param_store as ps
    assert cst.STAT_TIMES_KEY == "PMS_CONSENSUS_STAT_TIMES"
    dv, typ, _desc = ps.RUNTIME_EXTRA[cst.STAT_TIMES_KEY]
    assert dv == "0935,1030,1330,1445" and typ is str


@case("E _at_checkpoint·命中检查点及其下一分钟, 其余与空串不命中")
def _():
    t = "0935,1030,1330,1445"
    assert cst._at_checkpoint(datetime(2026, 9, 15, 9, 35), t) is True
    assert cst._at_checkpoint(datetime(2026, 9, 15, 9, 36), t) is True    # 下一分钟
    assert cst._at_checkpoint(datetime(2026, 9, 15, 9, 37), t) is False
    assert cst._at_checkpoint(datetime(2026, 9, 15, 14, 45), t) is True
    assert cst._at_checkpoint(datetime(2026, 9, 15, 9, 35), "") is False  # 空串无检查点


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