"""Targeted offline study of a gate-conditioned initial-stop refresh.

The study replays production entries only for the requested period. Whenever
an initial stop is touched, it rebuilds that ticker's setup using bars through
the previous close and recomputes the production momentum gate across the whole
historical universe. If the gate still passes and the new setup has a lower
valid stop, the simulator adopts it and checks it against the same day's low.

This is causal: no value from the stop day's eventual close is used to cancel
an intraday stop. The snapshot is read-only and no live settings are changed.
"""

from __future__ import annotations

import argparse
import asyncio
import bisect
import json
import multiprocessing
import os
import sys
from collections import Counter
from concurrent.futures import ProcessPoolExecutor, as_completed
from datetime import date, datetime
from pathlib import Path
from types import SimpleNamespace
from typing import Any

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))


def _sqlite_url(path: Path) -> str:
    return f"sqlite+aiosqlite:///{path.resolve().as_posix()}"


def _parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("snapshot")
    parser.add_argument("--start-date", default="2024-07-01")
    parser.add_argument("--workers", type=int, default=6)
    parser.add_argument("--out", default=None)
    parser.add_argument("--quiet", action="store_true")
    return parser.parse_args()


def _default_output_path() -> Path:
    stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
    return Path("reports") / f"gate-protected-stop-{stamp}.json"


class GateStopRefresher:
    """Point-in-time gate and replacement-stop calculator for stop events."""

    def __init__(
        self,
        prices: dict[str, tuple],
        recommendation_config: dict,
        activation: dict,
        benchmark_closes: dict[date, float],
    ) -> None:
        from app.services import backtest_service as bt

        self.bt = bt
        self.prices = prices
        self.recommendation_config = recommendation_config
        self.activation = activation
        self.benchmark_closes = benchmark_closes
        self.threshold = float(activation.get("min_momentum_percentile", 80.0))
        self.dates = {
            symbol: [date.fromordinal(value) for value in columns[0]]
            for symbol, columns in prices.items()
        }
        self.index_of = {
            symbol: {value: index for index, value in enumerate(columns[0])}
            for symbol, columns in prices.items()
        }
        self.percentile_cache: dict[int, dict[str, float]] = {}
        self.setup_cache: dict[tuple[str, int], dict | None] = {}
        self.events: list[dict[str, Any]] = []

    def _momentum_percentiles(self, asof_ord: int) -> dict[str, float]:
        cached = self.percentile_cache.get(asof_ord)
        if cached is not None:
            return cached

        values: dict[str, float] = {}
        for symbol, columns in self.prices.items():
            idx = bisect.bisect_right(columns[0], asof_ord) - 1
            if idx < 252:
                continue
            closes = columns[4]
            value = self.bt._residual_momentum_12_1(
                self.dates[symbol], closes, idx, self.benchmark_closes
            )
            if value is None and closes[idx - 252] > 0:
                value = closes[idx - 21] / closes[idx - 252] - 1.0
            if value is not None:
                values[symbol] = float(value)

        ordered = sorted(values, key=lambda symbol: values[symbol])
        denominator = len(ordered) - 1
        percentiles = {
            symbol: (rank / denominator * 100.0) if denominator > 0 else 100.0
            for rank, symbol in enumerate(ordered)
        }
        self.percentile_cache[asof_ord] = percentiles
        return percentiles

    def _long_setup(self, symbol: str, asof_idx: int) -> dict | None:
        columns = self.prices[symbol]
        asof_ord = columns[0][asof_idx]
        key = (symbol, asof_ord)
        if key in self.setup_cache:
            return self.setup_cache[key]

        records = [
            SimpleNamespace(
                date=date.fromordinal(o),
                open=op,
                high=high,
                low=low,
                close=close,
                volume=volume,
            )
            for o, op, high, low, close, volume in zip(
                columns[0][: asof_idx + 1],
                columns[1][: asof_idx + 1],
                columns[2][: asof_idx + 1],
                columns[3][: asof_idx + 1],
                columns[4][: asof_idx + 1],
                columns[5][: asof_idx + 1],
            )
        ]
        setups = self.bt._window_setups(
            records, self.recommendation_config, self.activation
        )
        setup = next((row for row in setups if row["direction"] == "long"), None)
        self.setup_cache[key] = setup
        return setup

    def __call__(
        self,
        symbol: str,
        stop_ord: int,
        active_stop: float,
        position: dict,
        bar: Any,
    ) -> float | None:
        columns = self.prices[symbol]
        stop_idx = self.index_of[symbol].get(stop_ord)
        if stop_idx is None:
            stop_idx = bisect.bisect_left(columns[0], stop_ord)
        asof_idx = stop_idx - 1
        if asof_idx < self.bt.MIN_LOOKBACK - 1:
            return None

        asof_ord = columns[0][asof_idx]
        setup = self._long_setup(symbol, asof_idx)
        momentum_pct = self._momentum_percentiles(asof_ord).get(symbol)
        gate_passed = bool(
            setup is not None
            and self.bt._momentum_qualifies(
                {
                    "meets_core": setup["meets_core"],
                    "direction": "long",
                    self.bt.PRODUCTION_PERCENTILE_KEY: momentum_pct,
                },
                self.threshold,
            )
        )
        new_stop = float(setup["stop"]) if gate_passed and setup is not None else None
        lower_stop = bool(new_stop is not None and new_stop < active_stop - 1e-9)
        original_risk = float(position["entry"] - position["initial_stop"])
        replacement_risk_r = (
            (float(position["entry"]) - new_stop) / original_risk
            if lower_stop and original_risk > 0 and new_stop is not None
            else None
        )
        self.events.append({
            "symbol": symbol,
            "stop_date": date.fromordinal(stop_ord).isoformat(),
            "gate_asof_date": date.fromordinal(asof_ord).isoformat(),
            "momentum_percentile": round(momentum_pct, 2)
            if momentum_pct is not None
            else None,
            "gate_core_passed": bool(setup and setup["meets_core"]),
            "gate_passed": gate_passed,
            "active_stop": round(active_stop, 4),
            "replacement_stop": round(new_stop, 4) if new_stop is not None else None,
            "lower_stop": lower_stop,
            "same_bar_survives": bool(lower_stop and bar.low > new_stop),
            "replacement_risk_r": round(replacement_risk_r, 3)
            if replacement_risk_r is not None
            else None,
        })
        return new_stop


def _arm(label: str, sim: dict) -> dict:
    trade_details = sim.pop("trade_details", None)
    row = {"arm": label, **sim}
    if trade_details is not None:
        row["trade_details"] = trade_details
    return row


def _rescued_trade_summary(trades: list[dict]) -> dict:
    rescued = [trade for trade in trades if trade.get("stop_refreshes", 0) > 0]
    rs = [float(trade["r"]) for trade in rescued]
    return {
        "trades": len(rescued),
        "wins": sum(value > 0 for value in rs),
        "win_rate": round(sum(value > 0 for value in rs) / len(rs) * 100.0, 1)
        if rs
        else None,
        "avg_r": round(sum(rs) / len(rs), 3) if rs else None,
        "total_r": round(sum(rs), 2) if rs else None,
        "worst_r": round(min(rs), 2) if rs else None,
        "best_r": round(max(rs), 2) if rs else None,
        "exit_reasons": dict(Counter(trade["reason"] for trade in rescued)),
    }


async def _main() -> None:
    args = _parse_args()
    snapshot = Path(args.snapshot)
    if not snapshot.exists():
        raise SystemExit(f"Snapshot not found: {snapshot}")
    try:
        start_date = date.fromisoformat(args.start_date)
    except ValueError as exc:
        raise SystemExit("--start-date must use YYYY-MM-DD") from exc

    os.environ["BACKTEST_SNAPSHOT_OFFLINE"] = "1"
    os.environ["BACKTEST_ALLOW_SPAWN"] = "1"

    from app.models.ticker import Ticker
    from app.services import backtest_service as bt
    from app.services.admin_service import get_activation_config
    from app.services.paper_trade_service import get_exit_policy
    from app.services.recommendation_service import get_recommendation_config

    engine = create_async_engine(_sqlite_url(snapshot), pool_pre_ping=True)
    Session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
    try:
        async with Session() as db:
            recommendation_config = await get_recommendation_config(db)
            activation = await get_activation_config(db)
            exit_config = await get_exit_policy(db)
            benchmark_closes = await bt._load_benchmark_closes_for_backtest(
                db, days=None, refresh=False
            )
            ticker_result = await db.execute(select(Ticker).order_by(Ticker.symbol))
            symbols = [ticker.symbol for ticker in ticker_result.scalars().all()]
            prices: dict[str, tuple] = {}
            for index, symbol in enumerate(symbols, 1):
                columns = await bt._fetch_columns(db, symbol)
                if columns is not None:
                    prices[symbol] = columns
                if not args.quiet and index % 50 == 0:
                    print(f"loaded prices: {index}/{len(symbols)}", flush=True)
    finally:
        await engine.dispose()

    candidates: list[dict] = []
    workers = max(1, min(int(args.workers), multiprocessing.cpu_count() - 1))
    context = multiprocessing.get_context("spawn")
    with ProcessPoolExecutor(max_workers=workers, mp_context=context) as pool:
        futures = {
            pool.submit(
                bt._replay_candidates_for_period,
                symbol,
                columns,
                recommendation_config,
                activation,
                benchmark_closes,
                start_date,
            ): symbol
            for symbol, columns in prices.items()
        }
        for index, future in enumerate(as_completed(futures), 1):
            candidates.extend(future.result())
            if not args.quiet and index % 25 == 0:
                print(f"replayed tickers: {index}/{len(futures)}", flush=True)

    bt._assign_momentum_percentiles(candidates)
    bt._assign_residual_momentum_percentiles(candidates)
    bt._assign_low_volatility_percentiles(candidates)
    bt._assign_activation_momentum_percentiles(candidates)
    bt._assign_residual_high_vol_blend(candidates)
    threshold = float(activation.get("min_momentum_percentile", 80.0))
    for candidate in candidates:
        candidate["qualified"] = bt._momentum_qualifies(candidate, threshold)

    strategy = next(
        row for row in bt.PORTFOLIO_MONITOR_STRATEGIES if row.get("is_production")
    )
    entry_config = bt._entry_variant_config(str(strategy["entry_variant"]))
    if entry_config is None:
        raise RuntimeError("Production entry configuration missing")
    exit_policy = bt.LIVE_EXIT_MODE_TO_SIM.get(
        str(exit_config.get("mode", "atr_trailing")), "atr_trail3"
    )
    hold_days = int(exit_config.get("hold_days", max(bt.TIME_EXIT_DAYS)))
    trail_multiplier = float(
        exit_config.get("atr_multiplier", bt.ATR_TRAIL_MULTIPLIER)
    )
    sim_kwargs = {
        "qualified_fn": None,
        "ranking_key": str(
            entry_config.get("ranking_key") or entry_config["percentile_key"]
        ),
        "max_positions": int(entry_config["max_positions"]),
        "risk_per_trade": float(entry_config["risk_per_trade"]),
        "atr_trail_multiplier": trail_multiplier,
        "start_date": start_date,
    }

    baseline = bt._simulate_portfolio(
        candidates, prices, benchmark_closes, exit_policy, hold_days, **sim_kwargs
    )
    cooldown_5 = bt._simulate_portfolio(
        candidates,
        prices,
        benchmark_closes,
        exit_policy,
        hold_days,
        reentry_cooldown_sessions=5,
        **sim_kwargs,
    )
    cooldown_10 = bt._simulate_portfolio(
        candidates,
        prices,
        benchmark_closes,
        exit_policy,
        hold_days,
        reentry_cooldown_sessions=10,
        **sim_kwargs,
    )
    refresher = GateStopRefresher(
        prices, recommendation_config, activation, benchmark_closes
    )
    gate_protected = bt._simulate_portfolio(
        candidates,
        prices,
        benchmark_closes,
        exit_policy,
        hold_days,
        initial_stop_refresh_fn=refresher,
        include_trades=True,
        **sim_kwargs,
    )
    if any(row is None for row in (baseline, cooldown_5, cooldown_10, gate_protected)):
        raise RuntimeError("A study arm produced no trades")

    gate_trades = list(gate_protected.get("trade_details") or [])
    event_counts = Counter()
    for event in refresher.events:
        event_counts["stop_touches"] += 1
        if event["gate_passed"]:
            event_counts["gate_passed"] += 1
        if event["lower_stop"]:
            event_counts["lower_stop"] += 1
        if event["same_bar_survives"]:
            event_counts["same_bar_survives"] += 1

    report = {
        "generated_at": datetime.now().astimezone().isoformat(),
        "snapshot": str(snapshot.resolve()),
        "period_start": start_date.isoformat(),
        "tickers": len(prices),
        "entry_candidates": len(candidates),
        "qualified_candidates": sum(bool(row["qualified"]) for row in candidates),
        "params": {
            "entry_cadence_days": bt.STEP_DAYS,
            "setup_stop_atr_multiplier": bt.ATR_MULTIPLIER,
            "exit_policy": exit_policy,
            "exit_atr_multiplier": trail_multiplier,
            "hold_days": hold_days,
            "momentum_percentile_floor": threshold,
            "gate_refresh_information_cutoff": "previous close",
        },
        "arms": [
            _arm("baseline", baseline),
            _arm("cooldown_5", cooldown_5),
            _arm("cooldown_10", cooldown_10),
            _arm("gate_protected_stop", gate_protected),
        ],
        "gate_stop_events": {
            **dict(event_counts),
            "unique_symbols": len({event["symbol"] for event in refresher.events}),
            "rescued_trade_outcomes": _rescued_trade_summary(gate_trades),
            "events": refresher.events,
        },
        "note": (
            "The gate-protected arm recalculates the gate at an initial-stop touch "
            "using only data available through the previous close. It accepts only "
            "a lower stop from a newly valid long setup and checks that replacement "
            "against the same bar. It does not cancel stops using the later same-day close."
        ),
    }
    output = Path(args.out) if args.out else _default_output_path()
    output.parent.mkdir(parents=True, exist_ok=True)
    output.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")

    print(f"Report written: {output}")
    for arm in report["arms"]:
        print(
            f"{arm['arm']}: Sharpe {arm['sharpe']}, CAGR {arm['cagr_pct']}%, "
            f"DD {arm['max_drawdown_pct']}%, trades {arm['trades']}"
        )
    print(f"gate stop events: {dict(event_counts)}")
    print(f"rescued outcomes: {report['gate_stop_events']['rescued_trade_outcomes']}")


if __name__ == "__main__":
    asyncio.run(_main())