qlib/backend/tests/test_combo_service.py

"""回测组合服务集成测试（合成数据 + 内存 SQLite，不连真库）。

验证 ComboService.run 端到端：装配行情 → 多策略 Borda → 持仓区间 runner → BacktestResult。
覆盖：
- 两策略打分合并后选出并集 TopN；
- 持仓天数区间 [Tmin, Tmax] 在真实数据装配路径下生效；
- 公共配置的成本/复权被采用并写进 config_snapshot（可复现）。
"""

from __future__ import annotations

from datetime import date
from decimal import Decimal

import pandas as pd
import pytest
from app.application.services.combo_service import ComboService
from app.domain.entities.combo import BacktestCombo, GlobalConfig
from app.domain.entities.strategy import SelectionStrategy
from app.infrastructure.persistence.sqlalchemy.base import Base
from app.infrastructure.persistence.sqlalchemy.models.market import (
    DailyBasicModel,
    StockDailyModel,
    StockModel,
)
from sqlalchemy import create_engine
from sqlalchemy.orm import Session

# 5 只股票，股息率梯度：A 最高 … E 最低
_SYMS = ["600000.SH", "600001.SH", "600002.SH", "600003.SH", "600004.SH"]
_DV = {"600000.SH": 9.0, "600001.SH": 7.0, "600002.SH": 5.0, "600003.SH": 3.0, "600004.SH": 1.0}
_START = date(2024, 1, 1)


def _seed(session: Session) -> None:
    session.add_all(
        [
            StockModel(symbol=s, name=f"股票{s[:6]}", industry="银行", market="主板",
                       area="深圳", list_date=date(2000, 1, 1), status="L")
            for s in _SYMS
        ]
    )
    dates = pd.bdate_range(_START, periods=80)
    bars, basics = [], []
    for i, sym in enumerate(_SYMS):
        price = 10.0 + i
        for d in dates:
            price *= 1 + 0.0006 + 0.0002 * i
            bars.append(StockDailyModel(
                symbol=sym, trade_date=d.date(), source="tushare", adjust="none",
                open=Decimal(str(price)), high=Decimal(str(price)),
                low=Decimal(str(price)), close=Decimal(str(price)),
                volume=Decimal("1000000"), amount=Decimal(str(price * 1e6)),
            ))
            basics.append(DailyBasicModel(
                symbol=sym, trade_date=d.date(), source="tushare",
                close=Decimal(str(price)), dv_ratio=Decimal(str(_DV[sym])),
                dv_ttm=Decimal(str(_DV[sym])), pe=Decimal("8"), pb=Decimal("1"),
                total_mv=Decimal("1e11"),
            ))
    session.add_all(bars + basics)
    session.commit()


@pytest.fixture
def service(tmp_path) -> ComboService:
    engine = create_engine(f"sqlite:///{tmp_path / 'combo.db'}", future=True)
    Base.metadata.create_all(engine)
    session = Session(engine)
    _seed(session)
    from app.infrastructure.persistence.sqlalchemy.repositories.market_impl import (
        SqlAlchemyDailyBarRepository,
        SqlAlchemyDailyBasicRepository,
        SqlAlchemyStockRepository,
    )
    yield ComboService(
        SqlAlchemyStockRepository(session),
        SqlAlchemyDailyBarRepository(session),
        basic_repo=SqlAlchemyDailyBasicRepository(session),
    )
    session.close()


def _strategy(sid: str, name: str, *, conditions=None) -> SelectionStrategy:
    return SelectionStrategy(
        id=sid, name=name,
        factors=[{"name": "dividend_yield", "weight": 1.0}],
        conditions=conditions or [],
    )


def test_combo_run_produces_backtest_result_with_config_snapshot(service: ComboService) -> None:
    """两策略（一个带 dv_ratio 条件、一个不带）→ 组合跑出 BacktestResult，
    且 config_snapshot 固化了当时的成本/复权与策略定义（可复现）。"""
    combo = BacktestCombo(
        id="CMB-T1", name="双策略高股息", strategy_ids=["S1", "S2"],
        initial_capital=1_000_000.0, hold_count=2, hold_min_days=0, hold_max_days=None,
        rebalance_freq="monthly", period=(date(2024, 1, 2), date(2024, 4, 15)),
    )
    strategies = [
        _strategy("S1", "纯高股息"),
        _strategy("S2", "高股息+过滤", conditions=[{"field": "dv_ratio", "op": "lte", "value": 30}]),
    ]
    config = GlobalConfig(commission_rate=0.0003, stamp_tax_rate=0.0005,
                          slippage_rate=0.001, min_commission=5.0, price_adjustment="hfq")

    result = service.run(combo, strategies, config)

    assert result.summary.initial_capital == 1_000_000.0
    assert result.equity_curve, "应产出净值曲线"
    assert result.trades or result.positions, "应有成交或持仓"
    # 可复现快照：含组合参数 + 两策略定义 + 当时成本/复权
    snap = result.config_snapshot
    assert snap["combo"]["hold_count"] == 2
    assert {s["id"] for s in snap["strategies"]} == {"S1", "S2"}
    assert snap["costs"]["min_commission"] == 5.0
    assert snap["price_adjustment"] == "hfq"


def test_hold_max_days_limits_holding_in_real_run(service: ComboService) -> None:
    """日频 + Tmax=8：任何一笔交易的持有交易日数不超过 Tmax+1。"""
    combo = BacktestCombo(
        id="CMB-T2", name="短持", strategy_ids=["S1"],
        hold_count=1, hold_min_days=0, hold_max_days=8,
        rebalance_freq="daily", period=(date(2024, 1, 2), date(2024, 4, 15)),
    )
    strategies = [_strategy("S1", "纯高股息")]
    config = GlobalConfig(min_commission=0.0, price_adjustment="none")
    result = service.run(combo, strategies, config)

    assert result.trades, "日频短持应产生多次换手"
    # 用结果的 signal_history 重建交易日序列来按交易日计跨度
    trade_dates = sorted({pd.Timestamp(p.date) for p in result.equity_curve})
    pos = {d: i for i, d in enumerate(trade_dates)}
    for t in result.trades:
        span = pos[pd.Timestamp(t.exit_date)] - pos[pd.Timestamp(t.entry_date)]
        assert span <= 8 + 1, f"持仓跨 {span} 个交易日 > Tmax+1：{t}"


def test_missing_strategy_raises_clear_error(service: ComboService) -> None:
    """组合引用了 S-GONE，但只传入了 S-OTHER → 明确报出缺失的 id（不静默跳过）。"""
    combo = BacktestCombo(
        id="CMB-T3", name="缺策略", strategy_ids=["S-GONE"],
        hold_count=1, rebalance_freq="monthly", period=(date(2024, 1, 2), date(2024, 2, 1)),
    )
    other = _strategy("S-OTHER", "别的")
    with pytest.raises(ValueError, match="S-GONE"):
        service.run(combo, [other], GlobalConfig())