# -*- coding: utf-8 -*-
"""
第三十七批模块单测 (一字板方向按昨收定, 2026-10-08, 零外部依赖)
================================================================
运行: 在 tradingSystem 仓库根目录执行  python scripts/test_batch37_units.py

病根: executor._limit_up / _limit_down 拿 day_chg_from_open 的正负定一字板方向。这个涨幅是
现价比窗口首根分钟线的开盘价, 一字板上两者相等, 涨幅恒为 0, 同时满足「>= 0 算涨停」与
「<= 0 算跌停」, 两个标记一起亮。于是一字跌停的买单以「涨停一字板」跳过, 一字涨停的卖单
在非末日被当跌停顺延, 规则闸还给卖单挂上跌停告警。这两个函数此前没有任何单测钉着。

本批覆盖:
  A  et.limit_board 纯函数: 一字涨停/一字跌停/正常 K 线/三个板块的限制/风险警示 5%/
     低价股取整/除权除息日落在带外回退/昨收缺失与旧口径逐字相同/涨跌幅过小回退。
  B  接线 (executor._limit_flags 加三处消费方): 一字跌停买入跳过且理由说对、规则闸拦
     LIMIT_DOWN 不拦 LIMIT_UP; 一字涨停卖出非末日不再顺延、规则闸不挂跌停告警;
     正常 K 线不查昨收; 昨收缺失或取数异常与修之前逐字相同; 网格即时成交两个方向。
  C  market.get_prev_close 新鲜度: 只认上一交易日那一根/隔夜没跑回 None/今天的行不算/
     三种日期格式/取数异常不缓存/长假后第一天。
"""
import unit_test_guard  # noqa: F401，单独运行也禁止联网
import os
import sys
import traceback
from datetime import date, datetime

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

from app.core import exec_timing as et                                          # noqa: E402
from app.core import rule_gate as rg                                            # noqa: E402
from app.core import tradedays as td                                            # noqa: E402
from app.services import exec_advisor, executor, market                         # noqa: E402

RESULTS = []


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


PRM = {"sell_avoid_open_min": 30, "buy_halt_dayup": 0.05, "eod_force_time": "14:45",
       "eod_force_discount": 0.998, "urgent_sell_discount": 0.995, "sell_bucket_times": ""}


def one_word(px, bars=30):
    """一字板的当日聚合: 最高 = 最低 = 开盘 = 现价, 于是 day_chg_from_open 恰为 0。"""
    return {"price": px, "open": px, "high": px, "low": px, "vwap": px,
            "day_chg_from_open": 0.0, "bars": bars}


def normal(px=10.3):
    return {"price": px, "open": 10.0, "high": 10.5, "low": 9.9, "vwap": 10.2,
            "day_chg_from_open": px / 10.0 - 1, "bars": 60}


def _old_up(day):
    """修之前 executor._limit_up 的原文, 拿来比「回退时逐字相同」。"""
    if not day:
        return False
    hi, lo, px = day.get("high"), day.get("low"), day.get("price")
    chg = day.get("day_chg_from_open")
    return bool(hi and lo and px and hi == lo == px and (chg or 0) >= 0 and day.get("bars", 0) > 3)


def _old_down(day):
    if not day:
        return False
    hi, lo, px = day.get("high"), day.get("low"), day.get("price")
    chg = day.get("day_chg_from_open")
    return bool(hi and lo and px and hi == lo == px and (chg or 0) <= 0 and day.get("bars", 0) > 3)


def _flags(b):
    return (b["limit_up"], b["limit_down"], b["basis"])


# ================================================================ A 纯函数
@case("A 一字涨停 (主板 +10%) → 只亮涨停, 按昨收定")
def _():
    assert _flags(et.limit_board(one_word(11.0), prev_close=10.0, ts_code="600000.SH")) \
        == (True, False, "prev_close")


@case("A 一字跌停 (主板 -10%) → 只亮跌停, 按昨收定 (旧口径两个都亮, 这就是病根)")
def _():
    d = one_word(9.0)
    assert _old_up(d) and _old_down(d), "旧口径在一字板上两个都亮 —— 若这条不成立说明病根描述错了"
    assert _flags(et.limit_board(d, prev_close=10.0, ts_code="600000.SH")) \
        == (False, True, "prev_close")


@case("A 正常 K 线 (最高≠最低) → 两个都不亮, 给了昨收也一样")
def _():
    for pc in (None, 10.0, 9.0):
        assert _flags(et.limit_board(normal(), prev_close=pc, ts_code="600000.SH")) \
            == (False, False, "none")
    # 三根以内的分钟线不认一字板 (开盘头几分钟没成交时价格一条线, 不是封板)
    assert _flags(et.limit_board(one_word(11.0, bars=3), prev_close=10.0)) == (False, False, "none")
    assert _flags(et.limit_board({}, prev_close=10.0)) == (False, False, "none")
    assert _flags(et.limit_board(None, prev_close=10.0)) == (False, False, "none")


@case("A 板块限制: 创业板/科创板 20%, 北交所 30%, 主板 10%")
def _():
    assert et.board_limit_ratio("600000.SH") == 0.10
    assert et.board_limit_ratio("000001.SZ") == 0.10
    assert et.board_limit_ratio("300750.SZ") == 0.20
    assert et.board_limit_ratio("301001.SZ") == 0.20
    assert et.board_limit_ratio("688981.SH") == 0.20
    assert et.board_limit_ratio("830799.BJ") == 0.30
    assert _flags(et.limit_board(one_word(120.0), prev_close=100.0, ts_code="300750.SZ"))[:2] \
        == (True, False)
    assert _flags(et.limit_board(one_word(80.0), prev_close=100.0, ts_code="688981.SH"))[:2] \
        == (False, True)
    assert _flags(et.limit_board(one_word(13.0), prev_close=10.0, ts_code="830799.BJ"))[:2] \
        == (True, False)
    # 同样 +20% 放在主板上不可能是一字板, 落在带外 → 方向不明回退
    assert _flags(et.limit_board(one_word(12.0), prev_close=10.0, ts_code="600000.SH"))[2] \
        == "fallback"


@case("A 风险警示 5% 与低价股取整都认得出")
def _():
    assert _flags(et.limit_board(one_word(10.5), prev_close=10.0, ts_code="600000.SH"))[:2] \
        == (True, False)
    assert _flags(et.limit_board(one_word(9.5), prev_close=10.0, ts_code="600000.SH"))[:2] \
        == (False, True)
    # 1.05 × 1.1 = 1.155, 交易所四舍五入到 1.16, 实际涨幅 +10.48%, 仍在上沿 11% 之内
    assert _flags(et.limit_board(one_word(1.16), prev_close=1.05, ts_code="600000.SH"))[:2] \
        == (True, False)
    # 1.05 × 0.95 = 0.9975 → 1.00, 实际 -4.76%, 仍在下沿 4% 之外
    assert _flags(et.limit_board(one_word(1.0), prev_close=1.05, ts_code="600000.SH"))[:2] \
        == (False, True)


@case("A 除权除息日: 小额分红仍判得对, 大比例送转落在带外回退")
def _():
    # 昨收 10, 每股派 0.3, 除息参考价 9.7, 涨停 10.67: 比未折算的昨收 +6.7%, 仍在带里 → 涨停
    assert _flags(et.limit_board(one_word(10.67), prev_close=10.0, ts_code="600000.SH"))[:2] \
        == (True, False)
    # 10 送 3: 参考价 7.69, 涨停 8.46, 比未折算的昨收 -15.4%。主板跌停不可能跌这么多 → 回退,
    # 而不是把一字涨停判成跌停
    assert _flags(et.limit_board(one_word(8.46), prev_close=10.0, ts_code="600000.SH")) \
        == (True, True, "fallback")
    # 10 转 10: 涨停 5.5, -45%, 哪个板块都不可能 → 回退
    assert _flags(et.limit_board(one_word(5.5), prev_close=10.0, ts_code="300750.SZ"))[2] \
        == "fallback"


@case("A 昨收缺失/为零/认不出 → 回退, 两个标记与旧口径逐字相同")
def _():
    days = [one_word(11.0), one_word(9.0), one_word(11.0, bars=3), normal(), {},
            {**one_word(11.0), "day_chg_from_open": None},
            {**one_word(11.0), "day_chg_from_open": 0.02},
            {**one_word(11.0), "day_chg_from_open": -0.02},
            {**one_word(11.0), "high": 0}]
    for pc in (None, 0, "", "abc", -1):
        for d in days:
            b = et.limit_board(d, prev_close=pc, ts_code="600000.SH")
            assert (b["limit_up"], b["limit_down"]) == (_old_up(d), _old_down(d)), (pc, d, b)


@case("A 涨跌幅过小 (停牌行情冻在昨收、或一条线但没封板) → 回退旧口径")
def _():
    for px in (10.0, 10.1, 9.9, 10.39):
        assert _flags(et.limit_board(one_word(px), prev_close=10.0, ts_code="600000.SH")) \
            == (True, True, "fallback"), px


# ================================================================ B 接线
class _Stub:
    """临时替换 market.get_prev_close; 退出时还原。raise_on_call 用来证明「不该查」。"""
    def __init__(self, value=None, exc=None, raise_on_call=False):
        self.value, self.exc, self.raise_on_call, self.calls = value, exc, raise_on_call, []

    def __call__(self, code, *a, **k):
        self.calls.append(code)
        if self.raise_on_call:
            raise AssertionError(f"不该为 {code} 取昨收")
        if self.exc:
            raise self.exc
        return self.value

    def __enter__(self):
        self._orig = market.get_prev_close
        market.get_prev_close = self
        return self

    def __exit__(self, *a):
        market.get_prev_close = self._orig


def _ctx_day(code, day):
    up, down = executor._limit_flags(code, day)
    return {**day, "limit_up": up, "limit_down": down, "halted": False}


def _gate(side, day_ctx, qty=1000):
    ctx = {"ts_code": "600000.SH",
           "position": {"total_qty": 6000, "avail_qty": 6000, "frozen_reason": "NONE"},
           "day": {**day_ctx, "ma5": day_ctx["price"]},
           "params": {"no_chase_ma5": 0.06, "buy_halt_dayup": 0.05},
           "flags": {"buy_halt": False, "exec_halt": False, "brake_active": False,
                     "blacklisted": False, "is_command": True},
           "caps": dict(scale=2_000_000, portfolio_cap=0.60, stock_cap=0.08, max_names=15,
                        portfolio_mv=800_000, names_count=5, stock_mv=60_000, is_new_name=False,
                        sector=None, sector_names=0, sector_mv=0.0, sector_max_names=4,
                        sector_max_ratio=0.40, cash_reserve=0.0, sector_source_ready=True)}
    return rg.check(side=side, action="ADD" if side == "buy" else "TRIM", qty=qty,
                    price=day_ctx["price"], ctx=ctx)


def _codes(lst):
    return {x.split(":")[0] for x in lst}


@case("B 一字跌停·买入: 跳过且理由是跌停不是涨停; 规则闸拦 LIMIT_DOWN 不拦 LIMIT_UP")
def _():
    with _Stub(value=10.0) as s:
        dc = _ctx_day("600000.SH", one_word(9.0))
    assert s.calls == ["600000.SH"]
    assert (dc["limit_up"], dc["limit_down"]) == (False, True)
    r = et.decide(side="buy", now="10:30", day=dc, params=PRM, is_last_day=False, quota=1000)
    assert r["action"] == et.ACT_SKIP and "跌停一字板" in r["reason"] \
        and "涨停" not in r["reason"], r
    # 末日 14:50 也不接 (修之前同样是跳过, 只是理由错)
    r = et.decide(side="buy", now="14:50", day=dc, params=PRM, is_last_day=True, quota=1000)
    assert r["action"] == et.ACT_SKIP and "跌停一字板" in r["reason"], r
    kw = dict(side="buy", now="10:30", day=dc, params=PRM, is_last_day=False,
              is_command=True, quota=1000)
    assert et.hard_gate(**kw) == et.decide(**kw)          # 硬闸与 decide 不拆漂
    g = _gate("buy", dc)
    assert not g["passed"] and "LIMIT_DOWN" in _codes(g["failed"]) \
        and "LIMIT_UP" not in _codes(g["failed"]), g


@case("B 一字涨停·卖出: 非末日不再当跌停顺延, 规则闸不挂跌停告警; 买入照旧跳过")
def _():
    with _Stub(value=10.0):
        dc = _ctx_day("600000.SH", one_word(11.0))
    assert (dc["limit_up"], dc["limit_down"]) == (True, False)
    r = et.decide(side="sell", now="10:30", day=dc, params=PRM, is_last_day=False, quota=1000)
    assert r["action"] == et.ACT_FIRE, r                    # 现价 = 均价, 过了开盘半小时, 出手
    r = et.decide(side="sell", now="10:30", day=dc, params=PRM, is_last_day=False, quota=1000,
                  urgent=True)
    assert r["action"] == et.ACT_FIRE and "紧急" in r["reason"], r
    g = _gate("sell", dc)
    assert g["passed"] and "LIMIT_DOWN" not in _codes(g["warnings"]), g
    r = et.decide(side="buy", now="10:30", day=dc, params=PRM, is_last_day=False, quota=1000)
    assert r["action"] == et.ACT_SKIP and "涨停一字板" in r["reason"], r
    assert "LIMIT_UP" in _codes(_gate("buy", dc)["failed"])


@case("B 正常 K 线: 两个标记都不亮, 而且根本不去取昨收 (平常这一跳不多查库)")
def _():
    with _Stub(raise_on_call=True) as s:
        assert executor._limit_flags("600000.SH", normal()) == (False, False)
        assert executor._limit_flags("600000.SH", {}) == (False, False)
    assert s.calls == []


@case("B 昨收缺失或取数异常: 两个都亮, 买卖的结局与修之前逐字相同")
def _():
    for stub in (_Stub(value=None), _Stub(exc=RuntimeError("库抖"))):
        with stub:
            dc = _ctx_day("600000.SH", one_word(9.0))
        assert (dc["limit_up"], dc["limit_down"]) == (True, True)
        old = {**one_word(9.0), "limit_up": True, "limit_down": True, "halted": False}
        for side in ("buy", "sell"):
            kw = dict(side=side, now="10:30", params=PRM, is_last_day=False, quota=1000)
            assert et.decide(day=dc, **kw) == et.decide(day=old, **kw)
        r = et.decide(side="buy", now="10:30", day=dc, params=PRM, is_last_day=False, quota=1000)
        assert r["action"] == et.ACT_SKIP and "涨停一字板" in r["reason"], r     # 旧理由原样
        g = _gate("buy", dc)
        assert _codes(g["failed"]) & {"LIMIT_UP", "LIMIT_DOWN"} == {"LIMIT_UP"}, g  # 只记一条


@case("B 网格即时成交: 一字跌停不接买, 一字涨停可卖")
def _():
    with _Stub(value=10.0):
        down = _ctx_day("600000.SH", one_word(9.0))
        up = _ctx_day("600000.SH", one_word(11.0))
    now = datetime(2026, 9, 16, 10, 30)
    r = exec_advisor._strategy_immediate(side="buy", now=now, day=down, quota=1000, fired_today=0)
    assert r["action"] == et.ACT_WAIT and "跌停" in r["reason"], r
    r = exec_advisor._strategy_immediate(side="sell", now=now, day=up, quota=1000, fired_today=0)
    assert r["action"] == et.ACT_FIRE, r


# ================================================================ C 昨收新鲜度
class _Rows:
    def __init__(self, rows=None, exc=None):
        self.rows, self.exc, self.calls = rows or [], exc, 0

    def __call__(self, ts_code, days=70):
        self.calls += 1
        if self.exc:
            raise self.exc
        return list(self.rows)


def _with_rows(fake, fn):
    orig = market._factor_rows
    market._factor_rows = fake
    market._ma_cache.update({"day": None, "data": {}})
    try:
        return fn()
    finally:
        market._factor_rows = orig
        market._ma_cache.update({"day": None, "data": {}})


TODAY = date(2026, 9, 16)          # 周三, 前后都不是节假日, 不受日历库有没有当年数据影响
PREV = date(2026, 9, 15)


@case("C 最近一根正是上一交易日 → 返回它的收盘")
def _():
    f = _Rows([{"trade_date": date(2026, 9, 14), "close_qfq": 9.8},
               {"trade_date": PREV, "close_qfq": 10.0}])
    assert _with_rows(f, lambda: market.get_prev_close("600000.SH", today=TODAY)) == 10.0


@case("C 因子桥隔夜没跑 (最近一根是前天) → None, 不拿隔天的收盘判方向")
def _():
    f = _Rows([{"trade_date": date(2026, 9, 11), "close_qfq": 9.6},
               {"trade_date": date(2026, 9, 14), "close_qfq": 9.8}])
    assert _with_rows(f, lambda: market.get_prev_close("600000.SH", today=TODAY)) is None


@case("C 表里已有今天的行 → 不算, 仍取上一交易日那根")
def _():
    f = _Rows([{"trade_date": PREV, "close_qfq": 10.0},
               {"trade_date": TODAY, "close_qfq": 11.0}])
    assert _with_rows(f, lambda: market.get_prev_close("600000.SH", today=TODAY)) == 10.0


@case("C 日期列几种写法都认: date / datetime / 横线串 / 八位串")
def _():
    for td_val in (PREV, "2026-09-15", "20260915", datetime(2026, 9, 15, 0, 0)):
        f = _Rows([{"trade_date": td_val, "close_qfq": 10.0}])
        assert _with_rows(f, lambda: market.get_prev_close("600000.SH", today=TODAY)) == 10.0, td_val


@case("C 没有行 → None; 取数异常 → None 且不缓存, 下一次重查")
def _():
    assert _with_rows(_Rows([]), lambda: market.get_prev_close("600000.SH", today=TODAY)) is None
    f = _Rows(exc=RuntimeError("代理库抖"))

    def run():
        a = market.get_prev_close("600000.SH", today=TODAY)
        f.exc, f.rows = None, [{"trade_date": PREV, "close_qfq": 10.0}]
        b = market.get_prev_close("600000.SH", today=TODAY)
        c = market.get_prev_close("600000.SH", today=TODAY)       # 这次走缓存
        return a, b, c
    assert _with_rows(f, run) == (None, 10.0, 10.0)
    assert f.calls == 2, f.calls


@case("C 长假后第一天 (10-08 接 09-30): 日历有当年数据时认得出")
def _():
    today = date(2026, 10, 8)
    f = _Rows([{"trade_date": date(2026, 9, 30), "close_qfq": 10.0}])
    got = _with_rows(f, lambda: market.get_prev_close("600000.SH", today=today))
    if td.calendar_degraded(today):
        # 日历退化成周一到周五, 上一交易日算成 10-07, 对不上 → None, 一字板回退旧口径 (守成)
        assert got is None, got
        print("      (本机日历缺 2026 年数据, 走的是退化分支)")
    else:
        assert td.prev_trade_day(today) == date(2026, 9, 30)
        assert got == 10.0, got


@case("C 暂无昨收或日期陈旧不缓存，数据恢复后同日可重新识别")
def _():
    for rows in ([], [{"trade_date": date(2026, 9, 14), "close_qfq": 9.0}]):
        f = _Rows(rows)
        def run():
            assert market.get_prev_close("600000.SH", today=TODAY) is None
            f.rows = [{"trade_date": PREV, "close_qfq": 10.0}]
            assert market.get_prev_close("600000.SH", today=TODAY) == 10.0
            assert market.get_prev_close("600000.SH", today=TODAY) == 10.0
        _with_rows(f, run)
        assert f.calls == 2


@case("C 非正数与非有限昨收不缓存，不作为方向依据")
def _():
    for value in (0, -10, float("nan"), float("inf")):
        f = _Rows([{"trade_date": PREV, "close_qfq": value}])
        def run():
            assert market.get_prev_close("600000.SH", today=TODAY) is None
            f.rows = [{"trade_date": PREV, "close_qfq": 10.0}]
            assert market.get_prev_close("600000.SH", today=TODAY) == 10.0
        _with_rows(f, run)
        assert f.calls == 2


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