"""PARKED — no live consumer; its Jobs (fxhnt-backtest-ingest-job, fxhnt-backtest-verdict-job) and PVC
(fxhnt-backtest-pvc) are decommissioned. Revivable research: plugs into `backtest_refs` when an equity
strat is added to the live registry.

Walk-forward equity-factor backtest runner. Reads the dedicated DuckDB
backtest warehouse (never live Tiingo), reconstructs the point-in-time universe
each month, scores names with the SAME domain factor math as the live sleeves,
holds weights between monthly rebalances, and produces a daily net-return series
per construction."""
from __future__ import annotations

import datetime as dt
from dataclasses import dataclass
from typing import Any

import numpy as np

from fxhnt.adapters.warehouse.duckdb_feature_store import DuckDbFeatureStore
from fxhnt.domain.backtest import compute_stats
from fxhnt.domain.gauntlet.core import evaluate, per_period_sharpe
from fxhnt.domain.equity_backtest import (
    epoch_day,
    daily_returns_for_weights,
    month_end_rebalance_dates,
    pit_universe_by_month,
    universe_asof,
)
from fxhnt.domain.strategies.equity_factor import (
    composite_price,
    construction_weights,
    lowvol_score,
    momentum_12_1_from_closes,
    momentum_score,
    realized_vol_from_closes,
)

_CONSTRUCTIONS = ("long", "ls", "tilt")
_DOLLAR_VOL_WINDOW_DAYS = 63
_TRADING_DAYS_YEAR = 252


@dataclass(frozen=True)
class BacktestRunResult:
    returns_by_construction: dict[str, np.ndarray]
    rebalance_dates: list[str]
    n_names_avg: float


def _iso(epoch_day: int) -> str:
    return (dt.date(1970, 1, 1) + dt.timedelta(days=epoch_day)).isoformat()


def select_backtest_universe(
    store: DuckDbFeatureStore,
    *,
    min_history: int = 300,
    top_k: int = 2500,
    min_price: float = 1.0,
) -> dict[str, float]:
    """Bounded, survivorship-safe candidate universe for the backtest.

    Thin wrapper that delegates to the store adapter's `select_candidate_universe`
    (keeps the SQL inside the warehouse adapter, respecting the hexagonal boundary).

    Returns {symbol: peak_yearly_avg_dollar_volume} for the `top_k` most-liquid
    symbols that have at least `min_history` bars where BOTH close and volume are
    present (the self-joined set, NOT adjclose-only bars). Liquidity is ranked by
    PEAK calendar-year average dollar-volume (close*volume), NOT lifetime average,
    so names that were highly liquid only while alive (later delisted) or only
    recently (recent IPOs) are still included — avoiding survivorship bias in the
    candidate set. This caps memory: the runner then loads only these symbols.

    `min_price` is the data-hygiene floor on a symbol's median close (and the
    adapter additionally drops NASDAQ test-ticker symbol patterns); see
    `DuckDbFeatureStore.select_candidate_universe`.
    """
    return store.select_candidate_universe(
        min_history=min_history, top_k=top_k, min_price=min_price
    )


class EquityBacktestRunner:
    def __init__(
        self,
        store: DuckDbFeatureStore,
        *,
        n: int = 150,
        cost_bps_per_turnover: float = 15.0,
        borrow_annual: float = 0.0,
        momentum_min_history: int = 252,
        candidate_top_k: int | None = None,
        candidate_min_history: int = 300,
        candidate_min_price: float = 1.0,
        max_day_ret: float = 10.0,
        min_day_ret: float = -0.9,
        delisting_return: float = -0.30,
        delist_gap_days: int = 30,
        pit: bool = False,
        trailing_window_months: int = 12,
        min_bars_window: int = 60,
    ) -> None:
        self._store = store
        self._n = n
        self._cost = cost_bps_per_turnover / 1e4
        self._borrow_daily = borrow_annual / _TRADING_DAYS_YEAR
        self._mom_min = momentum_min_history
        self._candidate_top_k = candidate_top_k
        self._candidate_min_history = candidate_min_history
        self._candidate_min_price = candidate_min_price
        self._max_day_ret = max_day_ret
        self._min_day_ret = min_day_ret
        self._delisting_return = delisting_return
        self._delist_gap_days = delist_gap_days
        self._pit = pit
        self._trailing_window_months = trailing_window_months
        self._min_bars_window = min_bars_window

    @staticmethod
    def _ym_of(iso_date: str) -> int:
        y, m, _ = iso_date.split("-")
        return int(y) * 12 + int(m) - 1

    def _run_pit(self) -> BacktestRunResult:
        """POINT-IN-TIME path (default for the real run): rank/select each month's
        universe by TRAILING dollar-volume using ONLY data <= the rebalance date —
        removing the peak-full-history lookahead bias. Two-pass + memory-bounded:
        Pass 1 is one SQL aggregation (monthly dollar-volume); the union of all
        per-month top-N names is the ONLY symbol set whose adjclose panel is loaded."""
        monthly = self._store.monthly_dollar_volume()
        days = self._store.trading_days_epoch()
        all_iso = [_iso(d) for d in days]
        rebals = month_end_rebalance_dates(all_iso)
        rebal_yms = [self._ym_of(r) for r in rebals]

        uni_by_ym = pit_universe_by_month(
            monthly, rebal_yms,
            window_months=self._trailing_window_months, top_n=self._n,
            min_price=self._candidate_min_price, min_bars=self._min_bars_window,
        )

        union = sorted({s for syms in uni_by_ym.values() for s in syms})
        adj = self._store.read_panel(union, "adjclose")

        # Holding-period calendar uses the FULL trading calendar (`days` from
        # trading_days_epoch(), the same calendar the rebalance dates are built
        # from), NOT the union-only adjclose panel which is sparser. A day where
        # the union has prices for OTHER names but a held name has a gap is still
        # a real trading day: daily_returns_for_weights already hold-flats the
        # gapped name, so iterating the full calendar matches the non-PIT path.
        all_days = days

        # Delisting detection (return-booking survivorship fix), same as the other
        # paths: a name whose adjclose TERMINATES more than `delist_gap_days` before
        # the dataset's last day delisted; its terminus is its delisting day.
        delist_day: dict[str, int] = {}
        if all_days:
            cutoff = all_days[-1] - self._delist_gap_days
            for s, ser in adj.items():
                if not ser:
                    continue
                terminus = max(ser)
                if terminus < cutoff:
                    delist_day[s] = terminus

        series: dict[str, list[float]] = {c: [] for c in _CONSTRUCTIONS}
        prev_weights: dict[str, dict[str, float]] = {c: {} for c in _CONSTRUCTIONS}
        names_seen = 0
        n_rebals = 0
        traded_dates: list[str] = []

        for k, rebal in enumerate(rebals):
            rebal_day = epoch_day(rebal)
            uni = uni_by_ym.get(rebal_yms[k], [])

            mom: list[float | None] = []
            vols: list[float | None] = []
            kept: list[str] = []
            for s in uni:
                if s not in adj:
                    continue
                hist = [adj[s][d] for d in sorted(adj[s]) if d <= rebal_day]
                if len(hist) < self._mom_min:
                    continue
                m = momentum_12_1_from_closes(hist)
                v = realized_vol_from_closes(hist)
                if m is None or v is None or rebal_day not in adj[s]:
                    continue
                kept.append(s)
                mom.append(m)
                vols.append(v)
            if len(kept) < 2:
                continue
            names_seen += len(kept)
            n_rebals += 1
            traded_dates.append(rebal)

            scores = composite_price(momentum_score(mom), lowvol_score(vols))

            end_day = epoch_day(rebals[k + 1]) if k + 1 < len(rebals) else all_days[-1]
            hold_days = [d for d in all_days if rebal_day <= d <= end_day]

            for c in _CONSTRUCTIONS:
                w_list = construction_weights(scores, c)
                weights = {kept[i]: w_list[i] for i in range(len(kept)) if w_list[i] != 0.0}
                turnover = _turnover(prev_weights[c], weights)
                rets = daily_returns_for_weights(
                    weights, adj, hold_days, self._borrow_daily,
                    min_day_ret=self._min_day_ret, max_day_ret=self._max_day_ret,
                    delist_day=delist_day, delisting_return=self._delisting_return,
                )
                if rets:
                    rets[0] -= turnover * self._cost
                series[c].extend(rets)
                prev_weights[c] = weights

        return BacktestRunResult(
            returns_by_construction={c: np.asarray(series[c], dtype=float) for c in _CONSTRUCTIONS},
            rebalance_dates=traded_dates,
            n_names_avg=(names_seen / n_rebals) if n_rebals else 0.0,
        )

    def run(self) -> BacktestRunResult:
        if self._pit:
            return self._run_pit()

        members = self._store.read_membership()

        # Low-memory candidate path: pre-filter to a bounded, survivorship-safe set
        # of liquid names, load ONLY the adjclose panel for them (not close/volume),
        # and rank rebalances by a STATIC peak-yearly-liquidity dict. The alive-gate
        # stays point-in-time (membership tuples), only the liquidity ranking is
        # static. None-path = unchanged full-panel trailing-dollar-vol behavior.
        liq: dict[str, float] | None = None
        if self._candidate_top_k is not None:
            liq = select_backtest_universe(
                self._store,
                min_history=self._candidate_min_history,
                top_k=self._candidate_top_k,
                min_price=self._candidate_min_price,
            )
            members = [m for m in members if m[0] in liq]
            cand_syms = [m[0] for m in members]
            adj = self._store.read_panel(cand_syms, "adjclose")
            close: dict[str, dict[int, float]] = {}
            vol: dict[str, dict[int, float]] = {}
            syms = cand_syms
        else:
            syms = [m[0] for m in members]
            adj = self._store.read_panel(syms, "adjclose")
            close = self._store.read_panel(syms, "close")
            vol = self._store.read_panel(syms, "volume")

        all_days = sorted({d for s in adj.values() for d in s})
        all_iso = [_iso(d) for d in all_days]
        rebals = month_end_rebalance_dates(all_iso)

        # Delisting detection (return-booking survivorship fix): a candidate whose
        # adjclose series TERMINATES more than `delist_gap_days` before the dataset's
        # last day delisted (bankruptcy / M&A / going-private) — its terminus is its
        # delisting day, and a held position in it must realize a one-time terminal
        # return. A name whose data merely ends because the backtest ends (terminus
        # within `delist_gap_days` of all_days[-1]) is NOT a delisting (still trading).
        delist_day: dict[str, int] = {}
        if all_days:
            cutoff = all_days[-1] - self._delist_gap_days
            for s, ser in adj.items():
                if not ser:
                    continue
                terminus = max(ser)
                if terminus < cutoff:
                    delist_day[s] = terminus

        series: dict[str, list[float]] = {c: [] for c in _CONSTRUCTIONS}
        prev_weights: dict[str, dict[str, float]] = {c: {} for c in _CONSTRUCTIONS}
        names_seen = 0
        n_rebals = 0
        traded_dates: list[str] = []

        for k, rebal in enumerate(rebals):
            rebal_day = epoch_day(rebal)
            dollar_vol: dict[str, float]
            if liq is not None:
                # static peak-yearly-liquidity ranking; alive-gate is still PIT
                dollar_vol = liq
            else:
                lo = rebal_day - _DOLLAR_VOL_WINDOW_DAYS
                dollar_vol = {}
                for s in syms:
                    cl, vl = close.get(s, {}), vol.get(s, {})
                    acc = [cl[d] * vl[d] for d in cl if lo <= d <= rebal_day and d in vl]
                    if acc:
                        dollar_vol[s] = float(np.mean(acc))
            uni = universe_asof(members, dollar_vol, rebal, self._n)

            mom: list[float | None] = []
            vols: list[float | None] = []
            kept: list[str] = []
            for s in uni:
                hist = [adj[s][d] for d in sorted(adj[s]) if d <= rebal_day]
                if len(hist) < self._mom_min:
                    continue
                m = momentum_12_1_from_closes(hist)
                v = realized_vol_from_closes(hist)
                if m is None or v is None or rebal_day not in adj[s]:
                    continue
                kept.append(s)
                mom.append(m)
                vols.append(v)
            if len(kept) < 2:
                continue
            names_seen += len(kept)
            n_rebals += 1
            traded_dates.append(rebal)

            scores = composite_price(momentum_score(mom), lowvol_score(vols))

            end_day = epoch_day(rebals[k + 1]) if k + 1 < len(rebals) else all_days[-1]
            hold_days = [d for d in all_days if rebal_day <= d <= end_day]

            for c in _CONSTRUCTIONS:
                w_list = construction_weights(scores, c)
                weights = {kept[i]: w_list[i] for i in range(len(kept)) if w_list[i] != 0.0}
                turnover = _turnover(prev_weights[c], weights)
                rets = daily_returns_for_weights(
                    weights, adj, hold_days, self._borrow_daily,
                    min_day_ret=self._min_day_ret, max_day_ret=self._max_day_ret,
                    delist_day=delist_day, delisting_return=self._delisting_return,
                )
                if rets:
                    rets[0] -= turnover * self._cost
                series[c].extend(rets)
                prev_weights[c] = weights

        return BacktestRunResult(
            returns_by_construction={c: np.asarray(series[c], dtype=float) for c in _CONSTRUCTIONS},
            rebalance_dates=traded_dates,
            n_names_avg=(names_seen / n_rebals) if n_rebals else 0.0,
        )


def evaluate_constructions(
    result: BacktestRunResult,
    *,
    oos_fraction: float = 0.40,
    dsr_min: float = 0.95,
    oos_min_sharpe: float = 0.0,
    max_is_oos_decay: float = 0.50,
) -> dict[str, Any]:
    """Stack each construction's daily returns, split IS/OOS, run the gauntlet.
    n_trials = the number of constructions tested (multiple-testing correction)."""
    n_trials = len(result.returns_by_construction)
    out: dict[str, Any] = {
        "constructions": {},
        "rebalance_dates": result.rebalance_dates,
        "n_names_avg": result.n_names_avg,
        "settings": {
            "oos_fraction": oos_fraction, "dsr_min": dsr_min,
            "oos_min_sharpe": oos_min_sharpe, "max_is_oos_decay": max_is_oos_decay,
            "n_trials": n_trials,
        },
    }
    # DSR's sr_variance is the variance of the per-construction IS Sharpe estimates ACROSS the
    # trials (the constructions) — the cross-trial Sharpe dispersion, computed ONCE and shared by
    # every evaluate() call. Mirrors fxhnt.application.discovery.search (per_period_sharpe over the
    # IS slice; np.var when ≥2 trials else 0.0). per_period_sharpe is NaN-safe (0.0 for len<3).
    is_sharpes = [
        per_period_sharpe(r[: int((1.0 - oos_fraction) * len(r))])
        for r in result.returns_by_construction.values()
        if len(r[: int((1.0 - oos_fraction) * len(r))]) >= 3
    ]
    sr_variance = float(np.var(is_sharpes)) if len(is_sharpes) > 1 else 0.0

    for c, r in result.returns_by_construction.items():
        stats = compute_stats(r)
        if len(r) < 3:
            out["constructions"][c] = {
                "stats": stats.model_dump(),
                "verdict": {"passed": False, "dsr": 0.0, "is_sharpe": 0.0,
                            "oos_sharpe": 0.0, "n_trials": n_trials,
                            "reasons": ["insufficient data"], "pvalue": 1.0},
            }
            continue
        split = int((1.0 - oos_fraction) * len(r))
        is_r, oos_r = r[:split], r[split:]
        verdict = evaluate(
            is_r, oos_r, n_trials=n_trials, sr_variance=sr_variance,
            dsr_min=dsr_min, oos_min_sharpe=oos_min_sharpe,
            max_is_oos_decay=max_is_oos_decay,
            has_economic_rationale=True,
        )
        out["constructions"][c] = {
            "stats": stats.model_dump(),
            "verdict": verdict.model_dump(),
        }
    return out


def _turnover(prev: dict[str, float], cur: dict[str, float]) -> float:
    """TWO-WAY (round-trip) turnover: the full sum of absolute weight changes
    across all names. A complete A->B switch (sell 1.0 of A, buy 1.0 of B) scores
    2.0, not 1.0. `cost_bps_per_turnover` is therefore charged per unit of two-way
    turnover — a deliberately conservative cost model that overstates transaction
    cost (each side billed) and so yields a stricter verdict."""
    names = set(prev) | set(cur)
    return float(sum(abs(cur.get(s, 0.0) - prev.get(s, 0.0)) for s in names))
