"""用户案例端到端集成测试（合成数据）：高股息 Top n → 持仓前 x，m/y 双周期。

不连真库，用确定性合成行情 + 内存 SQLite（覆盖 daily_basic 与 adjust_factor
两条新链路），验证：
- dv_ratio 因子参与评分（dividend_yield）
- conditions（dv_ratio <= 30）真正过滤掉特殊分红股
- hfq 复权生效（除权日不再被计为亏损）
- m/y 双周期、x<=n、顺延买入的端到端组合行为
- 最低佣金 min_commission
"""

from __future__ import annotations

from datetime import date
from decimal import Decimal

import pandas as pd
import pytest
from app.infrastructure.persistence.sqlalchemy.base import Base
from app.infrastructure.persistence.sqlalchemy.models.market import (
    AdjustFactorModel,
    DailyBasicModel,
    StockDailyModel,
    StockModel,
)
from app.quant.engine import LocalEngine
from app.quant.service import ResearchService
from sqlalchemy import create_engine
from sqlalchemy.orm import Session

_SYMS = ["600000.SH", "600001.SH", "600002.SH", "600003.SH"]
_START = date(2024, 1, 1)


def _seed(session: Session, *, spike_date: date | None = None) -> None:
    """4 只股票：A/B 高股息，C 中股息，D 低股息；除权日因子在 2024-06-03 跳到 1.1。

    `spike_date`：把 A 股在该日的 dv_ratio 抬到 45%（特殊分红尖峰，
    用于验证 `dv_ratio <= 30` 条件确实把它剔除）。
    """
    session.add_all(
        [
            StockModel(
                symbol=s, name=f"股票{s[:6]}", industry="银行" if i < 2 else "制造",
                market="主板", area="深圳", list_date=date(2000, 1, 1), status="L",
            )
            for i, s in enumerate(_SYMS)
        ]
    )
    # 股息率：A=8% B=6% C=4% D=2%；special 时 A 在首个交易日出现 45% 的异常高值
    dv = {"600000.SH": 8.0, "600001.SH": 6.0, "600002.SH": 4.0, "600003.SH": 2.0}
    dates = pd.bdate_range(_START, periods=130)
    bars, basics, factors = [], [], []
    for i, sym in enumerate(_SYMS):
        price = 10.0 + i
        for d in dates:
            price = price * (1 + 0.0008 + 0.0003 * 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)),
                )
            )
            rate = dv[sym]
            if spike_date is not None and sym == "600000.SH" and d.date() == spike_date:
                rate = 45.0  # 特殊分红尖峰（应按条件在该择股日被剔除）
            basics.append(
                DailyBasicModel(
                    symbol=sym, trade_date=d.date(), source="tushare",
                    close=Decimal(str(price)), dv_ratio=Decimal(str(rate)),
                    dv_ttm=Decimal(str(rate)), pe=Decimal("8"), pb=Decimal("1"),
                    total_mv=Decimal("1e11"),
                )
            )
            # 2024-06-03 起因子 1.1（模拟一次除权）：hfq 价格应整体上移 10%
            f = 1.1 if d.date() >= date(2024, 6, 3) else 1.0
            factors.append(
                AdjustFactorModel(symbol=sym, trade_date=d.date(), factor=Decimal(str(f)))
            )
    session.add_all(bars + basics + factors)
    session.commit()


@pytest.fixture
def service(tmp_path) -> ResearchService:
    engine = create_engine(f"sqlite:///{tmp_path / 'case.db'}", future=True)
    Base.metadata.create_all(engine)
    session = Session(engine)
    _seed(session, spike_date=date(2024, 3, 1))
    from app.infrastructure.persistence.sqlalchemy.repositories.market_impl import (
        SqlAlchemyDailyBarRepository,
        SqlAlchemyDailyBasicRepository,
        SqlAlchemyStockRepository,
    )

    # 复权折算发生在 SqlAlchemyDailyBarRepository 的 SQL 内（price_adjust=...），
    # 业务层无需 adjust_factor 仓储
    yield ResearchService(
        SqlAlchemyStockRepository(session),
        SqlAlchemyDailyBarRepository(session),
        LocalEngine(),
        basic_repo=SqlAlchemyDailyBasicRepository(session),
    )
    session.close()


def _spec(**kw):
    from app.domain.entities.research import (
        ConditionSpec,
        CostSpec,
        FactorSpec,
        ResearchSpec,
        SelectionSpec,
        UniverseSpec,
    )

    base = dict(
        type="backtest",
        universe=UniverseSpec(exclude_st=False, min_listing_days=0),
        price_adjustment="hfq",
        factors=[FactorSpec(name="dividend_yield")],
        conditions=[ConditionSpec(field="dv_ratio", op="lte", value=30)],
        selection=SelectionSpec(
            top_n=3, hold_top_x=2, allow_substitute=False, defer_buy=True
        ),
        rebalance="monthly",
        selection_interval_months=3,
        rebalance_interval_months=3,
        period=(date(2024, 3, 1), date(2024, 6, 28)),
        costs=CostSpec(
            commission_rate=0.0003, stamp_tax_rate=0.0005, slippage_rate=0.001,
            min_commission=5.0,
        ),
    )
    base.update(kw)
    return ResearchSpec(**base)


class TestUserCaseEndToEnd:
    def test_backtest_runs_with_all_new_knobs(self, service) -> None:
        result = service.run_backtest(_spec())
        assert result.summary.total_trades >= 1
        # 配置可溯源（v3 §20.5）
        snap = result.config_snapshot
        assert snap["price_adjustment"] == "hfq"
        assert snap["price_basis"]["execution_price_basis"] == "close_adj"
        assert snap["selection_interval_months"] == 3
        assert snap["selection"]["hold_top_x"] == 2
        assert result.symbol_curves, "应输出个股收益曲线"

    def test_condition_excludes_special_dividend_spike(self, service) -> None:
        """A 股在 2024-03-01 的 dv_ratio=45% > 30% → 该日不得进入候选池；
        其它择股日 A 股息率仍最高（8%）→ 应正常入选（证明过滤是按日求值而非一刀切）。"""
        result = service.run_backtest(
            _spec(period=(date(2024, 3, 1), date(2024, 6, 28)),
                  selection_interval_months=3, rebalance_interval_months=3)
        )
        by_day: dict = {}
        for p in result.selection_history:
            by_day.setdefault(p.date, []).append(p.symbol)
        assert date(2024, 3, 1) in by_day
        assert "600000.SH" not in by_day[date(2024, 3, 1)], "尖峰日应按条件剔除"
        later = [d for d in by_day if d > date(2024, 3, 1)]
        assert later, "应存在后续择股日"
        assert all("600000.SH" in by_day[d] for d in later), "尖峰解除后应恢复入选"

    def test_top_x_cap_and_pool_recorded(self, service) -> None:
        result = service.run_backtest(_spec())
        # 候选池 = n = 3（4 只中剔除 1 只）
        per_day: dict = {}
        for p in result.selection_history:
            per_day.setdefault(p.date, []).append(p.symbol)
        assert per_day and all(len(v) <= 3 for v in per_day.values())
        # 持仓 ≤ x = 2
        held: dict = {}
        for pos in result.positions:
            held.setdefault(pos.date, []).append(pos.symbol)
        assert held and all(len(v) <= 2 for v in held.values())

    def test_hfq_reflects_dividend_jump(self, service) -> None:
        """hfq 下 2024-06-03 不复权价与复权价之比应为 1.1（无横截面跳变计入收益）。"""
        from app.quant.service import load_daily_df

        daily = load_daily_df(
            service._daily_repo, ["600000.SH"], date(2024, 5, 31), date(2024, 6, 4),
            ["close"], price_adjust="hfq",
        )
        raw = load_daily_df(
            service._daily_repo, ["600000.SH"], date(2024, 5, 31), date(2024, 6, 4),
            ["close"], price_adjust="none",
        )
        adj = daily.sort_values("trade_date")["close"].tolist()
        rawp = raw.sort_values("trade_date")["close"].tolist()
        # 6/3 之前因子 1.0；之后 1.1 → 复权价整体上移
        assert adj[0] == pytest.approx(rawp[0])
        assert adj[-1] == pytest.approx(rawp[-1] * 1.1, rel=1e-9)

    def test_min_commission_increases_cost(self, service) -> None:
        """小资金 + x=2 → 单笔约 1 万元，佣金 3 元低于最低 5 元 → 成本上升。"""
        free = service.run_backtest(
            _spec(
                initial_capital=20_000.0,
                costs=_spec().costs.model_copy(update={"min_commission": 0.0}),
            )
        )
        costly = service.run_backtest(_spec(initial_capital=20_000.0))  # min_commission=5.0
        assert costly.summary.final_equity < free.summary.final_equity

    def test_unimplemented_notes_surfaced(self, service) -> None:
        result = service.run_backtest(_spec())
        joined = " | ".join(result.unimplemented)
        assert "后复权" in joined
        assert "顺延买入" in joined
        assert "退市" in joined  # 幸存者偏差显式标注


class TestSelectionBacktestConsistencyWithConditions:
    """v3 §28：回测择股日候选池 == 同日 /api/selections（同一求值器）。"""

    def test_pool_matches_selection_service(self, service) -> None:
        from app.application.services.selection_service import SelectionService
        from app.domain.entities.selection import SelectionQuery

        result = service.run_backtest(
            _spec(selection_interval_months=3, rebalance_interval_months=3)
        )
        sel_service = SelectionService(
            service._stock_repo, service._daily_repo, basic_repo=service._basic_repo
        )
        first_day = min(p.date for p in result.selection_history)
        query = SelectionQuery(
            method="condition",
            price_adjustment="hfq",
            conditions=[{"field": "dv_ratio", "op": "lte", "value": 30}],
            as_of=first_day,
        )
        sel = sel_service.select(query)
        eligible = {c.symbol for c in sel.candidates}
        # 回测当日候选池（记为条件通过者中的前 n）必是条件合格集合的子集
        pool = {p.symbol for p in result.selection_history if p.date == first_day}
        assert pool <= eligible, f"候选池 {pool} 不在条件合格集 {eligible} 内"
        assert eligible, "条件合格集不应为空（B/C/D 股 dv_ratio ≤ 30）"