"""买卖理由（数据化）与因子曲线的单测。 用户要求：「所有买卖点详细说明买卖理由，用数据说话」「回测图上增加因子相关曲线」。 因此这里验证的是**数字真的来自引擎当时计算**，而不是后补的文案： 1. 买入理由带名次 / 候选数 / 综合分 / 每个因子当时的原始值； 2. 卖出理由区分「跌出 TopN（第几名）」「被股票池过滤」「持有超 Tmax」； 3. Tmin 保护、涨停未买等未成交点也有结构化理由； 4. `Trade.entry_reason / exit_reason` 跟着成交记录走； 5. `factor_curves` = 持仓权重加权平均的**原始值**，空仓日不落点。 """ from __future__ import annotations from datetime import date, timedelta import pandas as pd import pytest from app.domain.entities.combo import BacktestCombo from app.domain.entities.research import CostSpec from app.quant.combo_engine import HoldingBandRunner from app.quant.factors import get_factor def _days(n: int = 12, start: date = date(2024, 1, 2)) -> list[pd.Timestamp]: out: list[date] = [] d = start while len(out) < n: if d.weekday() < 5: out.append(d) d += timedelta(days=1) return [pd.Timestamp(x) for x in out] def _close(days, series: dict[str, list[float]]) -> pd.DataFrame: return pd.DataFrame(series, index=pd.DatetimeIndex(days)) def _combo(**over) -> BacktestCombo: base = dict( name="理由单测", strategy_ids=["S1"], initial_capital=1_000_000.0, hold_count=1, hold_min_days=0, hold_max_days=None, rebalance_freq="daily", period=(date(2024, 1, 2), date(2024, 1, 31)), ) base.update(over) return BacktestCombo(**base) def _factor_panels(days, values: dict[str, float]): """用一个真实注册因子（momentum_20）承载合成面板：label/方向/单位来自注册表。""" panel = pd.DataFrame( {sym: [v] * len(days) for sym, v in values.items()}, index=pd.DatetimeIndex(days) ) return {"momentum_20": (get_factor("momentum_20")[0], panel)} def _run( *, score_rows: list[dict[str, float]], close_series: dict[str, list[float]], factor_values: dict[str, float] | None = None, eligibility_fn=None, **combo_over, ): days = _days(len(score_rows)) score = pd.DataFrame(score_rows, index=pd.DatetimeIndex(days)) close = _close(days, close_series) runner = HoldingBandRunner( combo=_combo(**combo_over), costs=CostSpec(), score=score, close=close, eligibility_fn=eligibility_fn, factor_panels=_factor_panels(days, factor_values or {"A": 0.1, "B": 0.2}), ) return runner.run(), days # ---------- 买入理由 ---------- def test_buy_reason_has_real_numbers(): """买入成交的理由 = 名次 + 候选数 + 综合分 + 各因子当时的原始值（全部来自引擎）。""" result, _ = _run( score_rows=[{"A": 0.9, "B": 0.5}] * 6, close_series={"A": [100.0] * 6, "B": [100.0] * 6}, factor_values={"A": 0.123, "B": 0.456}, ) buys = [a for a in result.signal_history if a.signal == "BUY" and a.filled] assert len(buys) == 1 r = buys[0].reason assert r is not None assert r.code == "buy_enter_topn" assert r.data["rank"] == 1 assert r.data["total"] == 2 assert r.data["top_n"] == 1 assert r.data["score"] == pytest.approx(0.9) # 因子原始值（面板里 A=0.123）——区间内买入理由必须能对上这个数 assert r.data["factors"]["momentum_20"] == pytest.approx(0.123) assert "第 1" in r.text and "0.9000" in r.text def test_sell_reason_ranks_and_hold_days(): """跌出 TopN 的卖出理由要说出「第几名掉出去」与持有交易日。""" # 第 1 天 A 第一 → 买入 A；第 3 天起 B 第一 → 卖出 A（名次 2/2） rows = [{"A": 0.9, "B": 0.5}, {"A": 0.9, "B": 0.5}, {"A": 0.1, "B": 0.9}] + [ {"A": 0.1, "B": 0.9} ] * 3 result, _ = _run( score_rows=rows, close_series={"A": [100.0] * 6, "B": [100.0] * 6}, hold_min_days=0, ) sells = [a for a in result.signal_history if a.signal == "SELL" and a.filled] assert len(sells) == 1 r = sells[0].reason assert r is not None assert r.code == "sell_drop_topn" assert r.data["rank"] == 2 assert r.data["total"] == 2 assert r.data["hold_days"] == 2 # 第 1 天买、第 3 天卖 → 2 个交易日 assert "第 2/2" in r.text # 成交明细里的买卖理由两端齐全 trade = result.trades[0] assert trade.entry_reason is not None and trade.entry_reason.code == "buy_enter_topn" assert trade.exit_reason is not None and trade.exit_reason.code == "sell_drop_topn" def test_sell_reason_not_in_pool_is_distinct(): """被股票池/条件过滤掉（不在候选池）≠ 排名掉出去：理由要分开写。""" result, _ = _run( score_rows=[{"A": 0.9, "B": 0.5}] * 4, close_series={"A": [100.0] * 4, "B": [100.0] * 4}, # 第 4 天把 B 之外的 A 挡在候选池外（A 持有中，属于「已不在候选池」） eligibility_fn=lambda as_of: {"B"} if as_of >= date(2024, 1, 5) else None, ) sells = [a for a in result.signal_history if a.signal == "SELL" and a.filled] assert sells, "A 不在候选池后应被卖出" r = sells[0].reason assert r is not None and r.code == "sell_drop_topn" assert r.data["in_pool"] is False assert "已不在候选池" in r.text def test_sell_reason_tmax_force_exit(): """Tmax 强制了结：理由说明「持有 N 天 > Tmax」，与排名无关。""" result, _ = _run( score_rows=[{"A": 0.9, "B": 0.5}] * 8, close_series={"A": [100.0] * 8, "B": [100.0] * 8}, hold_max_days=3, ) forced = [ a for a in result.signal_history if a.signal == "SELL" and a.filled and a.reason is not None and a.reason.code == "sell_force_tmax" ] assert forced, "超 Tmax 应有强制了结" r = forced[0].reason assert r is not None assert r.data["hold_days"] > r.data["tmax"] == 3 assert "Tmax=3" in r.text def test_tmin_protection_reason(): """未满 Tmin 掉出 TopN：理由写明「仅持 N 天 < Tmin，暂留」。""" rows = [{"A": 0.9, "B": 0.5}, {"A": 0.1, "B": 0.9}] + [{"A": 0.1, "B": 0.9}] * 4 result, _ = _run( score_rows=rows, close_series={"A": [100.0] * 6, "B": [100.0] * 6}, hold_min_days=3, ) deferred = [ a for a in result.signal_history if not a.filled and a.reason is not None and a.reason.code == "sell_defer_tmin" ] assert deferred, "未满 Tmin 应记录暂留理由" r = deferred[0].reason assert r is not None assert r.data["hold_days"] < r.data["tmin"] == 3 assert "Tmin=3" in r.text and "暂留" in r.text def test_buy_blocked_by_limit_up_reason(): """涨停无法追买：理由里带「收盘 / 前收 = 比值 ≥ 阈值」的真实数字。""" # 第 3 天 A 相对前收涨 10% 以上（600xxx 主板阈值 1.099）→ 当日买不进 a = [100.0, 100.0, 111.0, 111.0] result, _ = _run( score_rows=[{"A": 0.9, "B": 0.5}] * 2 + [{"A": 0.9, "B": 0.5}] * 2, close_series={"A": a, "B": [100.0] * 4}, # 前几天 A 不可选，逼到第 3 天涨停时才想买 eligibility_fn=lambda as_of: {"B"} if as_of < date(2024, 1, 4) else {"A", "B"}, ) blocked = [ a_ for a_ in result.signal_history if a_.signal == "BUY" and not a_.filled and a_.reason is not None and a_.reason.code == "buy_skip_limit_up" ] assert blocked, "涨停日应记录未成交理由" r = blocked[0].reason assert r is not None assert r.data["close_prev_ratio"] == pytest.approx(1.11, abs=1e-3) assert r.data["limit_ratio"] == pytest.approx(1.099) assert "涨停" in r.text # ---------- 因子曲线 ---------- def test_factor_curve_is_holding_weighted_raw_value(): """因子曲线 = 持仓权重加权平均的原始值（有 label/方向/单位），空仓日不落点。""" rows = [{"A": 0.9, "B": 0.5}] * 6 result, days = _run( score_rows=rows, close_series={"A": [100.0] * 6, "B": [200.0] * 6}, factor_values={"A": 0.2, "B": 0.8}, ) assert len(result.factor_curves) == 1 fc = result.factor_curves[0] assert fc.name == "momentum_20" assert fc.direction == "higher_is_better" assert fc.unit == "小数" assert fc.label.startswith("动量") # 只买 A（N=1），因子值恒为 A 的 0.2；第一天调仓在收盘后建仓 → 第一天也落点 assert all(p.value == pytest.approx(0.2) for p in fc.points) assert len(fc.points) == len(days) def test_factor_curve_skips_empty_holding_days(): """空仓日不落点（不插值、不用 0 假填充），曲线点数少于交易日数。""" # 只有第 3 天有股票可选：之前空仓，之后持仓 rows = [{"A": float("nan"), "B": float("nan")}] * 2 + [{"A": 0.9, "B": 0.5}] * 4 result, days = _run( score_rows=rows, close_series={"A": [100.0] * 6, "B": [100.0] * 6}, ) fc = result.factor_curves[0] assert 0 < len(fc.points) < len(days)