qlib/backend/tests/test_backtest_history.py

"""M9-2 回测历史测试：selection_history（意图=select 同源）、signal_history（BUY/SELL
与成交与否）、fills（Signal↔Fill 区分）——含「涨停导致 BUY 信号未成交」场景。
"""

from __future__ import annotations

from datetime import date
from decimal import Decimal

import pandas as pd
import pytest
from app.application.services.selection_service import SelectionService
from app.domain.entities.market import DailyBar, Stock
from app.domain.entities.research import BacktestResult, ResearchSpec
from app.domain.entities.selection import SelectionQuery
from app.quant.engine import LocalEngine

from conftest_quant import synthetic_daily

_SYMS = ["600000.SH", "600001.SH", "600002.SH", "600003.SH", "600004.SH"]


class _MemStockRepo:
    def __init__(self, stocks):
        self._stocks = stocks

    def list(self):
        return self._stocks

    def get_by_symbol(self, symbol):
        return next((s for s in self._stocks if s.symbol == symbol), None)


class _MemDailyRepo:
    def __init__(self, df: pd.DataFrame) -> None:
        self._bars = [
            DailyBar(
                symbol=r.symbol, trade_date=r.trade_date,
                open=Decimal(str(r.open)), high=Decimal(str(r.high)),
                low=Decimal(str(r.low)), close=Decimal(str(r.close)),
                volume=Decimal(str(r.volume)), amount=Decimal(str(r.amount)),
            )
            for r in df.itertuples()
        ]

    def get_range(self, symbol, start, end):
        return [b for b in self._bars if b.symbol == symbol and start <= b.trade_date <= end]

    def get_range_many(self, symbols, start, end, adjust="none"):
        syms = set(symbols)
        return [b for b in self._bars if b.symbol in syms and start <= b.trade_date <= end]


def _stocks() -> list[Stock]:
    return [
        Stock(symbol=s, name=f"测试{i}", list_date=date(1999, 1, 1)) for i, s in enumerate(_SYMS)
    ]


def _spec(**kw) -> ResearchSpec:
    base = dict(
        type="backtest",
        universe={"exclude_st": False, "min_listing_days": 0, "symbols": _SYMS},
        factors=[{"name": "momentum_60", "weight": 1.0}],
        selection={"top_n": 2},
        rebalance="monthly",
        period=(date(2024, 5, 1), date(2024, 12, 31)),
    )
    base.update(kw)
    return ResearchSpec(**base)


@pytest.fixture()
def daily_df() -> pd.DataFrame:
    return synthetic_daily({s: 0.006 - 0.0015 * i for i, s in enumerate(_SYMS)}, n=320)


class TestBacktestHistory:
    def test_selection_history_matches_select(self, daily_df) -> None:
        spec = _spec()
        result = LocalEngine().run_backtest(daily_df, spec)
        svc = SelectionService(_MemStockRepo(_stocks()), _MemDailyRepo(daily_df))

        rebal_dates = sorted({p.date for p in result.positions})  # 有持仓的调仓日近似
        assert result.selection_history
        hist_by_date: dict[date, list] = {}
        for pick in result.selection_history:
            hist_by_date.setdefault(pick.date, []).append(pick)

        sample = [d for d in hist_by_date if d in rebal_dates][:3]
        for d in sample:
            res = svc.select(
                SelectionQuery(
                    universe=spec.universe,
                    factors=[{"name": "momentum_60", "weight": 1.0}],
                    top_n=2,
                    as_of=d,
                )
            )
            picked = {c.symbol: c.rank for c in res.candidates}
            hist = {p.symbol: p.rank for p in hist_by_date[d]}
            assert hist == picked, f"as_of={d}：意图 {hist} ≠ select {picked}"

    def test_signal_and_fills_consistency(self, daily_df) -> None:
        result = LocalEngine().run_backtest(daily_df, _spec())
        fills = result.fills
        assert fills == [a for a in result.signal_history if a.filled]
        # 每个 SELL fill 对应一笔 Trade 平仓；每个 BUY fill 在交易中体现
        assert all(a.filled for a in fills)
        buy_fills = [a for a in fills if a.signal == "BUY"]
        assert len(buy_fills) > 0
        # fills 价格与滑点一致（buy price >= close）
        assert all(a.price is not None for a in fills)

    def test_limit_up_buy_rejected_visible(self) -> None:
        """把最高动量股在某调仓日设为涨停：意图仍在 selection_history，
        但 signal_history 记录 BUY 未成交（涨停），fills/持仓无该笔。"""
        df = synthetic_daily({s: 0.006 - 0.0015 * i for i, s in enumerate(_SYMS)}, n=320)
        # 找首个 >= 2024-05-01 的交易日作为首个调仓日
        dates = sorted(pd.to_datetime(df["trade_date"].unique()))
        d0 = next(x for x in dates if x.date() >= date(2024, 5, 1))
        top = _SYMS[0]  # 最高漂移股
        prev_date = dates[dates.index(d0) - 1]
        prev_close = float(
            df[(df["symbol"] == top) & (df["trade_date"] == prev_date.date())]["close"].iloc[0]
        )
        # 把 d0 该股 close 抬高到 +10%（近似涨停）
        mask = (df["symbol"] == top) & (df["trade_date"] == d0.date())
        df.loc[mask, "close"] = prev_close * 1.10
        df.loc[mask, "high"] = prev_close * 1.10

        spec = _spec()
        result = LocalEngine().run_backtest(df, spec)

        picks_at_d0 = [p for p in result.selection_history if p.date == d0.date()]
        assert any(p.symbol == top for p in picks_at_d0), "涨停股仍应在选股意图中"
        # 涨停买入被拒
        rejects = [
            a for a in result.signal_history
            if a.date == d0.date() and a.symbol == top and a.signal == "BUY" and not a.filled
        ]
        assert rejects and any("涨停" in (r.reject_reason or "") for r in rejects)
        # 该日无该股成交
        assert not any(
            a.date == d0.date() and a.symbol == top and a.signal == "BUY" and a.filled
            for a in result.fills
        )

    def test_result_serializes_history(self, daily_df) -> None:
        result = LocalEngine().run_backtest(daily_df, _spec())
        js = result.model_dump(mode="json")
        assert "selection_history" in js and "signal_history" in js and "fills" in js
        BacktestResult.model_validate(js)  # roundtrip