"""研究服务：把 Research Specification 编排为数据获取 + 引擎执行。

本层是业务入口：API / Agent 只能调用这里的用例（AGENT.md §16/§17），
禁止直接拼接引擎配置。数据一律经 Repository 获取（防未来函数由查询层保证）。

内存优化：大数据面板优先走 Repository 的流式列裁剪查询
（stream_range_many_columns，SQL 侧转 REAL、分批拉取），避免 ORM 对象 /
Decimal 全量物化；老实现回退到 get_range_many 逐实体路径。
"""

from __future__ import annotations

from collections.abc import Iterable
from datetime import date, timedelta
from typing import Any

import pandas as pd

from app.domain.entities.market import (
    DAILY_BAR_NUMERIC_FIELDS,
    DAILY_BASIC_NUMERIC_FIELDS,
    FinancialIndicator,
)
from app.domain.entities.research import (
    BacktestResult,
    FactorCorrelationReport,
    FactorTestReport,
    ResearchSpec,
)
from app.domain.repositories.market import (
    DailyBarRepository,
    StockRepository,
)
from app.quant.composite import build_factor_panels
from app.quant.engine import QuantEngine
from app.quant.evaluation import factor_correlation_report
from app.quant.selection import build_condition_fields, eligible_symbols
from app.quant.universe import (  # noqa: F401 —— 范围过滤
    filter_stocks,
    names_as_of,
    resolve_members,
)

# 流式路径每攒多少行落一个 DataFrame 分片（控制 concat 峰值）
_FRAME_CHUNK_ROWS = 50_000


def bars_to_daily_df(bars) -> pd.DataFrame:
    """DailyBar 列表 → 引擎长表 DataFrame（symbol/trade_date/ohlc/volume/amount）。

    领域实体中的 Decimal 在此转 float，供 pandas 数值运算（保持 DataFrame 全数值列）。
    """
    df = pd.DataFrame([b.model_dump() for b in bars])
    if not df.empty:
        for col in ("open", "high", "low", "close", "volume", "amount"):
            if col in df.columns:
                df[col] = df[col].astype(float)
    return df


def _frame_from_stream(rows: Iterable[tuple], columns: list[str]) -> pd.DataFrame:
    """把流式 (symbol, trade_date_iso, *float_cols) 分批拼成 float 长表。

    全程只保留分批 DataFrame + 最终一份结果，避免整批 tuple/Decimal 同时驻留。
    返回列：symbol, trade_date(datetime64), *columns(float64)。
    """
    cols = ["symbol", "trade_date", *columns]
    buf: list[tuple] = []
    pieces: list[pd.DataFrame] = []
    for row in rows:
        buf.append(row)
        if len(buf) >= _FRAME_CHUNK_ROWS:
            pieces.append(pd.DataFrame(buf, columns=cols))
            buf = []
    if buf:
        pieces.append(pd.DataFrame(buf, columns=cols))
    if not pieces:
        return pd.DataFrame()
    df = pd.concat(pieces, ignore_index=True)
    df["trade_date"] = pd.to_datetime(df["trade_date"])
    for col in columns:  # NULL → NaN，统一 float64
        df[col] = pd.to_numeric(df[col], errors="coerce")
    return df


def load_daily_df(
    daily_repo,
    symbols: list[str],
    start: date,
    end: date,
    columns: list[str],
    adjust: str = "none",
    price_adjust: str = "none",
) -> pd.DataFrame:
    """从 Repository 装配行情长表（供研究/选股共用）。

    两个 adjust 语义不同（v3 §20.5）：
    - `adjust`：行集过滤（stock_daily.adjust 主口径 none / 新浪兜底 qfq 行）
    - `price_adjust`：复权折算（none/qfq/hfq，按 adjust_factor 在 SQL 侧折算价格列）

    优先走流式列裁剪（stream_range_many_columns，SQL 侧转 REAL、分批），
    失败或实现缺失时回退 get_range_many / 逐只 get_range（回退路径不支持 price_adjust，
    此时由调用方保证 price_adjust=none，避免静默混入未折算价格）。
    """
    if not symbols:
        return pd.DataFrame()
    streamer = getattr(daily_repo, "stream_range_many_columns", None)
    if streamer is not None:
        try:
            return _frame_from_stream(
                streamer(
                    symbols, start, end, sorted(columns),
                    adjust=adjust, price_adjust=price_adjust,
                ),
                sorted(columns),
            )
        except TypeError:  # 实现未升级（无 price_adjust 形参）→ 回退
            if price_adjust != "none":
                raise
            return _frame_from_stream(
                streamer(symbols, start, end, sorted(columns), adjust=adjust), sorted(columns)
            )
        except Exception:  # noqa: BLE001 —— 流式路径失败回退旧路径（兼容非 SQL 实现）
            if price_adjust != "none":
                raise  # 复权路径失败必须显式报错，禁止静默退回不复权
            pass
    elif price_adjust != "none":
        raise ValueError(
            f"行情仓储不支持流式列裁剪，无法按 {price_adjust} 复权装配（拒绝静默退回不复权）"
        )
    get_many = getattr(daily_repo, "get_range_many", None)
    if get_many is not None:
        bars = list(get_many(symbols, start, end, adjust=adjust))
    else:  # 兜底：逐只查询
        bars = []
        for sym in symbols:
            bars.extend(daily_repo.get_range(sym, start, end))
    return bars_to_daily_df(bars)


def split_factor_columns(columns: Iterable[str]) -> tuple[set[str], set[str]]:
    """把因子所需列拆成 (行情列, 每日指标列)。

    行情列来自 stock_daily（含 close 等）；每日指标列来自 daily_basic（dv_ratio 等）。
    未识别的列按行情列处理（由仓储的列白名单兜底报错，错误信息更贴近调用点）。
    """
    bars: set[str] = set()
    basics: set[str] = set()
    for col in columns:
        if col in DAILY_BASIC_NUMERIC_FIELDS and col not in DAILY_BAR_NUMERIC_FIELDS:
            basics.add(col)
        else:
            bars.add(col)
    return bars, basics


def load_basic_df(
    basic_repo,
    symbols: list[str],
    start: date,
    end: date,
    columns: list[str],
) -> pd.DataFrame:
    """装配每日指标长表（daily_basic）—— 仅取所需列。

    与 load_daily_df 同构（流式优先、回退批量/逐只）。
    """
    if not symbols or not columns:
        return pd.DataFrame()
    streamer = getattr(basic_repo, "stream_range_many_columns", None)
    if streamer is not None:
        try:
            return _frame_from_stream(
                streamer(symbols, start, end, sorted(columns)), sorted(columns)
            )
        except Exception:  # noqa: BLE001 —— 回退旧路径
            pass
    get_many = getattr(basic_repo, "get_range_many", None)
    if get_many is not None:
        rows = [
            {"symbol": r.symbol, "trade_date": r.trade_date, **{c: getattr(r, c) for c in columns}}
            for r in get_many(symbols, start, end)
        ]
    else:  # 兜底：逐只查询
        rows = []
        for sym in symbols:
            for r in basic_repo.get_range(sym, start, end):
                rows.append(
                    {
                        "symbol": r.symbol,
                        "trade_date": r.trade_date,
                        **{c: getattr(r, c) for c in columns},
                    }
                )
    df = pd.DataFrame(rows)
    if df.empty:
        return df
    df["trade_date"] = pd.to_datetime(df["trade_date"])
    for col in columns:
        df[col] = pd.to_numeric(df[col], errors="coerce")
    return df


def merge_basic_into_daily(daily: pd.DataFrame, basic: pd.DataFrame) -> pd.DataFrame:
    """把每日指标列并入行情长表（按 symbol + trade_date 左连接）。

    命名冲突保护：daily_basic 也有 close 列，若与行情 close 冲突则**丢弃指标侧**列
    （成交/估值口径以 stock_daily 不复权 close 为准，避免静默改口径，AGENT.md §8）。
    """
    if basic is None or basic.empty:
        return daily
    if daily is None or daily.empty:
        return daily
    add_cols = [
        c for c in basic.columns if c not in ("symbol", "trade_date") and c not in daily.columns
    ]
    if not add_cols:
        return daily
    left = daily.copy()
    left["trade_date"] = pd.to_datetime(left["trade_date"])
    right = basic[["symbol", "trade_date", *add_cols]].copy()
    right["trade_date"] = pd.to_datetime(right["trade_date"])
    merged = left.merge(right, on=["symbol", "trade_date"], how="left")
    return merged


def _fill_names(result: BacktestResult, stocks) -> BacktestResult:
    """把股票池的 `symbol → name` 回填进回测结果的各展示结构（就地修改并返回）。

    为什么放在服务层而不是引擎里：名称只是展示增强，两个引擎（LocalEngine / QlibEngine）
    都经本方法返回，改一处即同时覆盖；引擎保持「只认 symbol」的纯计算职责，不引入
    名称查询/IO，也就不必在引擎内部为零散字段各写一次映射。

    `stocks` 为空（universe 过滤后无标的、或调用方未装配）时**静默跳过**，不抛错：
    名称缺失只影响展示，不应让一个已经算完的回测失败（AGENT.md §24 的诚实标注
    由引擎的 unimplemented 承担，不由名称缺失承担）。
    """
    if not stocks:
        return result
    name_map = {s.symbol: s.name for s in stocks if getattr(s, "name", None)}
    if not name_map:
        return result

    def _fill(rows) -> None:
        for row in rows:
            if row.name is None:
                row.name = name_map.get(row.symbol)

    _fill(result.symbol_curves)
    # marks 与 signal_history 在 LocalEngine 中共享同一批 ActionRecord 对象，
    # 这里仍显式回填一次：对独立实现（如 Qlib 适配器）也成立，幂等无副作用。
    for curve in result.symbol_curves:
        _fill(curve.marks)
    _fill(result.positions)
    _fill(result.trades)
    _fill(result.signal_history)
    _fill(result.fills)
    _fill(result.selection_history)
    return result


class ResearchService:
    """研究用例入口（因子测试 / 回测）。依赖注入 Repository 与引擎。"""

    def __init__(
        self,
        stock_repo: StockRepository,
        daily_repo: DailyBarRepository,
        engine: QuantEngine,
        index_repo=None,
        basic_repo=None,
        financial_repo=None,
        name_repo=None,
    ) -> None:
        self._stock_repo = stock_repo
        self._daily_repo = daily_repo
        self._engine = engine
        self._index_repo = index_repo
        # 名称变更历史仓储：universe.exclude_st 的**时点**口径依据。
        # 未注入 → 回退 stock.name 最新快照（旧行为，结果页会如实标注残余偏差）。
        self._name_repo = name_repo
        # 每日指标仓储（daily_basic）：仅当 spec 因子引用 dv_ratio 等列时使用；
        # 未注入而在 spec 中引用 → 明确报错，不做静默降级（否则因子会全是 NaN）。
        self._basic_repo = basic_repo
        # 注：复权（qfq/hfq）折算在行情仓储 `stream_range_many_columns` 的 SQL 内完成
        # （与成交价同源，v3 §20.5），业务层不需要 adjust_factor 仓储；缺口统计走
        # `self._daily_repo.count_price_adjust_gaps`。
        # 财务仓储：spec.conditions 引用 fundamental.* 时使用
        self._financial_repo = financial_repo
        # 最近一次装配的复权因子缺口报告（未复权时为 None），用于结果如实标注
        self.last_adjust_gaps: dict | None = None
        # 最近一次装配的 exclude_st 名称口径（None = 未启用 ST 过滤）
        self.last_name_basis: dict | None = None
        # 最近一次装配的 universe 股票列表（条件求值需要 static.* 元数据）
        self._last_stocks: list = []

    def run_factor_test(
        self, spec: ResearchSpec, horizon_days: int = 21, on_stage=None
    ) -> FactorTestReport:
        """on_stage(str)：执行阶段回调（data_loading / factor_calculation / analysis），
        供 Job 状态机上报 stage（v3 §23）。"""
        if spec.type != "factor_test":
            raise ValueError("factor_test 用例需要 spec.type=factor_test")
        _stage(on_stage, "data_loading")
        daily = self._load_daily(spec)
        _stage(on_stage, "factor_calculation")
        report = self._engine.run_factor_test(daily, spec, horizon_days=horizon_days)
        _stage(on_stage, "analysis")
        return report

    def run_backtest(self, spec: ResearchSpec, on_stage=None) -> BacktestResult:
        if spec.type != "backtest":
            raise ValueError("backtest 用例需要 spec.type=backtest")
        _stage(on_stage, "data_loading")
        daily = self._load_daily(spec)
        _stage(on_stage, "backtesting")
        eligibility = self._build_eligibility(spec, daily)
        result = self._engine.run_backtest(daily, spec, eligibility_fn=eligibility)
        _stage(on_stage, "analysis")
        self._annotate_price_basis(result, spec)
        # 两个引擎都在此汇合：名称回填只做一次（`_last_stocks` 在 _load_daily 中装配）
        return _fill_names(result, self._last_stocks)

    def run_factor_correlation(self, spec: ResearchSpec) -> FactorCorrelationReport:
        """多因子两两相关（v3 §12）：同 universe/period 装配 → 横截面相关矩阵。"""
        daily = self._load_daily(spec)
        panels = {fs.name: build_factor_panels(daily, [fs])[0][1] for fs in spec.factors}
        return factor_correlation_report(panels)

    # ---- 数据装配 ----

    def _load_daily(self, spec: ResearchSpec) -> pd.DataFrame:
        start, end = spec.period
        # 回测前预留因子 warmup（lookback≤120 交易日，取 300 自然日余量）
        data_start = start - timedelta(days=300)
        all_stocks = self._stock_repo.list()
        name_at, applied = names_as_of(all_stocks, start, self._name_repo)
        stocks = filter_stocks(
            all_stocks, spec.universe, as_of=start,
            members=resolve_members(self._index_repo, spec.universe, start),
            name_at=name_at,
        )
        # 供结果如实标注：exclude_st 是否用了时点名称、覆盖了多少只
        self.last_name_basis = (
            {"point_in_time": applied[0], "covered": applied[1], "total": len(all_stocks)}
            if spec.universe.exclude_st
            else None
        )
        self._last_stocks = stocks
        # 引擎所需列裁剪（LocalEngine 只取 close + 因子字段；Qlib 回测取全 OHLCV）
        required = self._engine.required_columns(spec)
        bar_cols, basic_cols = split_factor_columns(required)
        symbols = [s.symbol for s in stocks]
        # 行集口径恒为 none（主口径）；复权折算由 price_adjust 承担（v3 §20.5 严格分离）
        daily = load_daily_df(
            self._daily_repo,
            symbols,
            data_start,
            end,
            sorted(bar_cols),
            adjust="none",
            price_adjust=spec.price_adjustment,
        )
        if spec.price_adjustment != "none":
            self.last_adjust_gaps = self._adjust_gap_report(symbols, data_start, end)
        if basic_cols:
            daily = self._attach_basic(daily, symbols, data_start, end, sorted(basic_cols))
        return daily

    def _adjust_gap_report(self, symbols: list[str], start: date, end: date) -> dict:
        """复权因子缺口统计（如实上报「按 1.0 兜底未折算」的行占比）。"""
        counter = getattr(self._daily_repo, "count_price_adjust_gaps", None)
        if counter is None:
            return {"checked": False, "reason": "仓储未实现 count_price_adjust_gaps"}
        total, missing = counter(symbols, start, end)
        return {
            "checked": True,
            "rows": total,
            "rows_without_factor": missing,
            "missing_pct": round(missing / total * 100, 4) if total else 0.0,
        }

    def _build_eligibility(self, spec: ResearchSpec, daily: pd.DataFrame):
        """按 spec.conditions 构造「择股日 → 合格股票集合」的求值闭包（可选）。

        与选股路径共用 `build_condition_fields` / `eligible_symbols`，因此
        「历史某日 /api/selections 的候选」与「回测在该日的候选池」由同一实现产出
        （v2 §25 / v3 §28 一致性要求）。逐择股日结果缓存，避免重复计算。

        未来函数红线：条件字段只取 <= as_of 的截面；fundamental.* 只取
        announce_date <= as_of 的已公告值（由仓储 list_announced_many 保证）。
        """
        if not self._last_stocks:
            if not spec.conditions:
                return None
            raise ValueError("universe 过滤结果为空，无法构造选股条件求值器")
        statics = {s.symbol: s.model_dump() for s in self._last_stocks}
        candidates = sorted(statics)
        # 时点 ST 过滤：exclude_st 必须在**每个择股日**按当时名称重判，否则
        # 「2020 年入池、2022 年才变 ST」的标的会一直被当作合格候选（与
        # /api/selections 的单时点语义不一致，v2 §25）。名称历史缺失时降级为
        # 池子基准日口径（旧行为），由结果 name_basis 如实标注。
        st_fn = self._build_st_filter(spec, candidates)
        if not spec.conditions:
            if st_fn is None:
                return None
            # 只有 ST 约束时，eligible 必须是「候选 − 该日 ST」，不能直接返回 st_fn
            # （st_fn 返回的是**不合格**集合，直接返回会把语义取反）
            allowed_cache: dict[date, set[str]] = {}

            def _st_only(as_of: date) -> set[str]:
                if as_of not in allowed_cache:
                    allowed_cache[as_of] = set(candidates) - st_fn(as_of)
                return allowed_cache[as_of]

            return _st_only
        uses_fundamental = any(
            f.startswith("fundamental.")
            for c in spec.conditions
            for f in (c.field, c.ref or "")
        )
        cache: dict[date, set[str]] = {}

        def _fn(as_of: date) -> set[str]:
            if as_of in cache:
                return cache[as_of]
            financial = (
                self._load_financial(candidates, as_of) if uses_fundamental else {}
            )
            fields = build_condition_fields(daily, spec.conditions, pd.Timestamp(as_of))
            if not fields:
                cache[as_of] = set()
                return cache[as_of]
            passed = set(
                eligible_symbols(candidates, spec.conditions, statics, fields, financial)
            )
            if st_fn is not None:
                passed -= st_fn(as_of)  # 该日名称含 ST → 不合格
            cache[as_of] = passed
            return cache[as_of]

        return _fn

    def _build_st_filter(self, spec: ResearchSpec, candidates: list[str]):
        """「择股日 → 该日名称含 ST 的股票集合」；不可用（未启用/未注入）时返回 None。

        与 `filter_stocks` 共用 `names_as_of`，因此回测在任一择股日的
        `exclude_st` 结果与该日 `/api/selections` 的候选池同口径（v2 §25）。
        逐日缓存；每次查询只取该日生效的名称区间。
        """
        if not spec.universe.exclude_st or self._name_repo is None:
            return None
        cache: dict[date, set[str]] = {}

        def _fn(as_of: date) -> set[str]:
            if as_of not in cache:
                name_at, applied = names_as_of(self._last_stocks, as_of, self._name_repo)
                if not applied[0]:
                    # 无时点名称可用（表为空/查询失败）：整段时间都退回池子基准日口径，
                    # 不逐日过滤 —— 与 filter_stocks 的回退一致，且 name_basis 会标注 false
                    cache[as_of] = set()
                else:
                    # 逐股优先时点名称、缺失回退最新名称快照 —— 必须与
                    # filter_stocks 的 `(name_at or {}).get(sym) or s.name` 完全同口径，
                    # 否则同一择股日「回测入选」与「/api/selections 排除」会打架（v2 §25）
                    st_syms: set[str] = set()
                    for st in self._last_stocks:
                        nm = (name_at or {}).get(st.symbol) or st.name
                        if nm and "ST" in nm.upper():
                            st_syms.add(st.symbol)
                    cache[as_of] = st_syms
            return cache[as_of]

        return _fn

    def _load_financial(self, symbols: list[str], as_of: date) -> dict[str, Any]:
        """按 announce_date <= as_of 批量取财务，每 symbol 保留最新一版。"""
        if self._financial_repo is None:
            raise ValueError(
                "回测条件引用了 fundamental.* 字段，但未注入 FinancialRepository。"
                "请检查 API/CLI 的依赖装配。"
            )
        getter = getattr(self._financial_repo, "list_announced_many", None)
        if getter is not None:
            rows = list(getter(symbols, as_of))
        else:  # 兜底：逐只取最新已公告
            rows = []
            for sym in symbols:
                latest = self._financial_repo.latest_announced(sym, as_of)
                if latest is not None:
                    rows.append(latest)
        out: dict[str, FinancialIndicator] = {}
        for r in rows:  # 约定升序返回 → 后者覆盖前者即「最新一版」
            out[r.symbol] = r
        return out

    def _annotate_price_basis(self, result, spec: ResearchSpec) -> None:
        """把价格口径写入结果（v3 §20.5：adjust_mode / price_basis / execution_price_basis）。

        同时如实标注复权因子缺口（AGENT.md §24：未覆盖部分必须显式说明，禁止假装支持）。
        """
        mode = spec.price_adjustment
        result.config_snapshot["price_basis"] = {
            "adjust_mode": mode,
            "price_basis": "adjust_factor" if mode != "none" else "raw_close",
            "execution_price_basis": "close_adj" if mode != "none" else "close_raw",
            "adjust_gaps": self.last_adjust_gaps,
            # exclude_st 的名称口径：时点（stock_name_history）还是最新快照
            "name_basis": self.last_name_basis,
        }
        nb = self.last_name_basis
        if nb is not None:
            if nb["point_in_time"]:
                result.unimplemented.append(
                    f"exclude_st 采用**时点名称**（stock_name_history，覆盖 {nb['covered']} 只生效名称）："
                    "「曾为高股息、后变 ST」的标的在其非 ST 期间会被正常纳入（股息陷阱可见）"
                )
            else:
                result.unimplemented.append(
                    "exclude_st 回退**最新名称快照**（名称变更历史表未同步/未注入）："
                    "「曾为高股息、后变 ST/退市」的标的会被整段排除，收益可能被高估；"
                    "修复：python -m app.cli.sync namechange"
                )
        if mode == "none":
            return
        label = "后复权（hfq）" if mode == "hfq" else "前复权（qfq）"
        result.unimplemented.append(
            f"价格口径为{label}：价格列 = 原始价 × adjust_factor"
            + ("；现金分红按复权因子隐含再投资处理" if mode == "hfq" else "（以前复权归一）")
            + "；volume/amount 不做复权折算"
        )
        gaps = self.last_adjust_gaps or {}
        if gaps.get("checked") and gaps.get("rows_without_factor", 0) > 0:
            result.unimplemented.append(
                f"复权因子覆盖不全：{gaps['rows_without_factor']} / {gaps['rows']} 行"
                f"（{gaps['missing_pct']}%）无对应 adjust_factor，已按系数 1.0 兜底（未折算）"
            )

    def _attach_basic(
        self,
        daily: pd.DataFrame,
        symbols: list[str],
        start: date,
        end: date,
        columns: list[str],
    ) -> pd.DataFrame:
        """把 daily_basic 列并入行情长表（dv_ratio 等因子依赖）。"""
        if self._basic_repo is None:
            raise ValueError(
                f"因子需要每日指标列 {columns}（daily_basic），但未注入 DailyBasicRepository。"
                "请检查 API/CLI 的依赖装配。"
            )
        basic = load_basic_df(self._basic_repo, symbols, start, end, columns)
        if basic.empty:
            raise ValueError(
                f"daily_basic 表在 {start}~{end} 无数据，无法计算需要 {columns} 的因子。"
                "请先运行：python -m app.cli.sync daily_basic --start 20200101"
            )
        return merge_basic_into_daily(daily, basic)


def _stage(cb, name: str) -> None:
    if cb is not None:
        cb(name)