ForexBot2/backtest.py

"""
ForexBot2 — Walk-Forward Backtester + Parameter Optimizer
==========================================================
Replays Twelve Data USD/JPY history bar-by-bar through the same
SignalAggregator that drives live mode. The same code path that
trades USD/JPY in production is what produces these metrics — so
the backtest validates the engine, not a parallel implementation.
"""

from __future__ import annotations

import copy
import itertools
import logging
import multiprocessing as mp
import os
from collections import deque
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import replace
from typing import Deque, Dict, List, Optional, Tuple

import pandas as pd

from config import PlatformConfig, Timeframe, USDJPY, PIP_VALUE
from data_provider import ForexDataProvider
from metrics import BacktestResult, Trade, compute_metrics, trades_to_dataframe
from signal_aggregator import SignalAggregator, SignalReport, TradeSetup

logger = logging.getLogger(__name__)


# Period in minutes for each timeframe (used for close-time computation)
TF_MINUTES: Dict[Timeframe, int] = {
    Timeframe.M1:  1,
    Timeframe.M5:  5,
    Timeframe.M15: 15,
    Timeframe.M30: 30,
    Timeframe.H1:  60,
    Timeframe.H4:  240,
    Timeframe.D1:  1440,
    Timeframe.W1:  10080,
}

GRADE_ORDER = ["A+", "A", "B", "C", "D", "F"]


def _grade_passes(grade: str, min_grade: str) -> bool:
    """True if `grade` is at least as good as `min_grade`."""
    if grade not in GRADE_ORDER or min_grade not in GRADE_ORDER:
        return False
    return GRADE_ORDER.index(grade) <= GRADE_ORDER.index(min_grade)


# ══════════════════════════════════════════════════════════════════
#  WALK-FORWARD BACKTESTER
# ══════════════════════════════════════════════════════════════════

class WalkForwardBacktester:
    """Single-pair, single-position event-driven backtester."""

    def __init__(self, config: PlatformConfig):
        self.config = config
        self.aggregator = SignalAggregator(config)
        self.data_provider = self.aggregator.data_provider
        self._regime_adx: Optional[pd.Series] = None
        self._regime_tf: Optional[Timeframe] = None

    # ── Public ─────────────────────────────────────────────────────

    def load_data(self) -> Dict[Timeframe, pd.DataFrame]:
        """Pull (or read from disk cache) all needed timeframes."""
        timeframes = [self.config.primary_timeframe] + list(
            self.config.confirmation_timeframes
        )
        bt = self.config.backtest
        if bt.regime_filter and bt.regime_filter_timeframe not in timeframes:
            timeframes.append(bt.regime_filter_timeframe)

        dfs: Dict[Timeframe, pd.DataFrame] = {}
        for tf in timeframes:
            df = self.data_provider.fetch_full_history(
                symbol=self.config.symbol,
                timeframe=tf,
                max_bars=bt.max_bars,
                start_date=bt.start_date,
                end_date=bt.end_date,
            )
            df = self._apply_window(df)
            dfs[tf] = df
        return dfs

    def _precompute_all_signals(
        self,
        dfs: Dict[Timeframe, pd.DataFrame],
        primary_df: pd.DataFrame,
        warmup: int,
        progress: bool = True,
    ) -> Dict[int, "SignalReport"]:
        """Pre-compute signals for every bar in parallel using all CPU threads.

        TechnicalEngine.analyze() is stateless (pure function of the df slice),
        so concurrent calls are safe. NumPy/pandas C extensions release the GIL,
        giving real parallelism across threads without pickling overhead.
        Returns bar-index → SignalReport so the sequential state machine can
        do O(1) lookups instead of recomputing indicators per bar.
        """
        primary_tf = self.config.primary_timeframe
        # Use all logical threads; cap at 2× physical cores to avoid over-subscription
        n_workers = min(os.cpu_count() or 4, 16)
        bar_indices = list(range(warmup, len(primary_df) - 1))

        def _compute_bar(t: int) -> Tuple[int, "SignalReport"]:
            decision_time = self._bar_close_time(primary_df.index[t], primary_tf)
            sliced = self._slice_dfs(dfs, decision_time)
            report = self.aggregator.analyze_from_dataframes(
                symbol=self.config.symbol,
                dfs=sliced,
                use_news=False,
            )
            return t, report

        precomputed: Dict[int, "SignalReport"] = {}
        pbar = None
        if progress:
            try:
                from tqdm import tqdm  # type: ignore
                pbar = tqdm(
                    total=len(bar_indices),
                    desc=f"Signals  ({n_workers} threads)",
                    unit="bar",
                    leave=False,
                )
            except ImportError:
                pass

        with ThreadPoolExecutor(max_workers=n_workers) as executor:
            futures = {executor.submit(_compute_bar, t): t for t in bar_indices}
            for future in as_completed(futures):
                t, report = future.result()
                precomputed[t] = report
                if pbar:
                    pbar.update(1)

        if pbar:
            pbar.close()

        return precomputed

    def run(
        self,
        dfs: Optional[Dict[Timeframe, pd.DataFrame]] = None,
        progress: bool = True,
    ) -> BacktestResult:
        if dfs is None:
            dfs = self.load_data()

        self._precompute_regime_filter(dfs)

        primary_tf = self.config.primary_timeframe
        primary_df = dfs.get(primary_tf)
        if primary_df is None or primary_df.empty:
            return BacktestResult(initial_balance=self.config.backtest.initial_balance)

        # Warmup: enough bars for ema_trend & ichimoku_senkou_b on the primary TF
        warmup = max(
            self.config.technical.ema_trend,
            self.config.technical.ichimoku_senkou_b,
            200,
        ) + 30

        if len(primary_df) <= warmup + 1:
            logger.warning(
                f"Insufficient data: {len(primary_df)} primary bars, "
                f"need >{warmup}."
            )
            return BacktestResult(initial_balance=self.config.backtest.initial_balance)

        # Pre-compute all bar signals in parallel before the sequential replay
        precomputed = self._precompute_all_signals(
            dfs, primary_df, warmup, progress=progress
        )

        bt = self.config.backtest
        balance = bt.initial_balance
        equity_points: List[Tuple[pd.Timestamp, float]] = [
            (primary_df.index[warmup], balance)
        ]
        trades: List[Trade] = []

        # Auto-session feedback: rolling pnl_pct over last K closed trades.
        # When mean falls below threshold, the session filter activates;
        # once expectancy recovers, it releases.
        self._recent_pnl_pcts: Deque[float] = deque(
            maxlen=max(1, bt.auto_session_lookback_trades)
        )

        open_position: Optional[Dict] = None

        # Optional progress bar for the state-machine replay pass
        iterator = range(warmup, len(primary_df) - 1)
        if progress:
            try:
                from tqdm import tqdm  # type: ignore
                iterator = tqdm(iterator, desc="Replaying bars", unit="bar", leave=False)
            except ImportError:
                pass

        for t in iterator:
            decision_time = self._bar_close_time(primary_df.index[t], primary_tf)

            # ── Step 1: manage any open position ───────────────────
            if open_position is not None:
                next_bar = primary_df.iloc[t]
                exit_info = self._check_exit(open_position, next_bar)
                if exit_info is not None:
                    closed = self._close_trade(
                        open_position, next_bar.name, exit_info, balance
                    )
                    trades.append(closed)
                    balance += closed.pnl_currency
                    equity_points.append((next_bar.name, balance))
                    self._recent_pnl_pcts.append(float(closed.pnl_pct))
                    open_position = None

            # ── Step 2: maybe open a new position ──────────────────
            if open_position is None or not bt.one_position_only:
                report = precomputed[t]

                if (
                    _grade_passes(report.signal_quality, bt.min_trade_grade)
                    and abs(float(report.final_score)) >= bt.min_abs_score
                    and report.trade_setup is not None
                    and t + 1 < len(primary_df)
                    and self._regime_passes(decision_time)
                    and self._session_passes(decision_time)
                ):
                    next_bar = primary_df.iloc[t + 1]
                    open_position = self._open_trade(report, next_bar, balance)

        # ── Force-close any still-open position at end of data ────
        if open_position is not None:
            last_bar = primary_df.iloc[-1]
            closed = self._close_trade(
                open_position,
                last_bar.name,
                {"price": float(last_bar["Close"]), "reason": "end_of_data"},
                balance,
            )
            trades.append(closed)
            balance += closed.pnl_currency
            equity_points.append((last_bar.name, balance))

        # Build equity curve
        eq_index = [p[0] for p in equity_points]
        eq_values = [p[1] for p in equity_points]
        equity_curve = pd.Series(eq_values, index=pd.DatetimeIndex(eq_index))

        elapsed_days: Optional[float] = None
        if len(equity_curve) >= 2:
            elapsed_days = (equity_curve.index[-1] - equity_curve.index[0]).total_seconds() / 86400

        metrics = compute_metrics(
            trades, equity_curve, bt.initial_balance, elapsed_days=elapsed_days
        )

        return BacktestResult(
            trades=trades,
            equity_curve=equity_curve,
            initial_balance=bt.initial_balance,
            final_balance=balance,
            metrics=metrics,
        )

    # ── Private: regime gate ───────────────────────────────────────

    def _precompute_regime_filter(self, dfs: Dict[Timeframe, pd.DataFrame]) -> None:
        """Pre-compute ADX on the regime-filter TF; cached for the whole run."""
        bt = self.config.backtest
        if not bt.regime_filter:
            self._regime_adx = None
            self._regime_tf = None
            return

        tf = bt.regime_filter_timeframe
        df = dfs.get(tf)
        if df is None or df.empty:
            logger.warning(
                f"Regime filter requested ({tf.value}) but no data available; "
                f"all signals will be skipped."
            )
            self._regime_adx = pd.Series(dtype=float)
            self._regime_tf = tf
            return

        self._regime_tf = tf
        self._regime_adx = self._compute_adx_series(df, self.config.technical.adx_period)

    def _regime_passes(self, decision_time: pd.Timestamp) -> bool:
        bt = self.config.backtest
        if not bt.regime_filter:
            return True
        if self._regime_adx is None or self._regime_adx.empty:
            return False
        period_min = TF_MINUTES[self._regime_tf]
        close_times = self._regime_adx.index + pd.Timedelta(minutes=period_min)
        valid = self._regime_adx[close_times <= decision_time]
        if valid.empty:
            return False
        last = valid.iloc[-1]
        if pd.isna(last):
            return False
        return float(last) >= bt.regime_filter_min_adx

    def _session_passes(self, decision_time: pd.Timestamp) -> bool:
        """UTC hour gate — decision_time = entry bar's open time.

        Filter is active when EITHER manual `session_filter` is on, OR
        `auto_session_filter` is on and the rolling mean pnl_pct over the
        last N closed trades is below threshold.
        """
        bt = self.config.backtest
        active = bt.session_filter
        if not active and bt.auto_session_filter:
            recent = getattr(self, "_recent_pnl_pcts", None)
            if recent is not None and len(recent) >= bt.auto_session_lookback_trades:
                mean_pnl = sum(recent) / len(recent)
                active = mean_pnl < bt.auto_session_threshold_pct
        if not active:
            return True
        h = int(decision_time.hour)
        return bt.session_start_hour <= h < bt.session_end_hour

    @staticmethod
    def _compute_adx_series(df: pd.DataFrame, period: int = 14) -> pd.Series:
        """Wilder's ADX series — same algorithm as engine._compute_adx."""
        import numpy as np
        high, low, close = df["High"], df["Low"], df["Close"]
        tr = pd.concat([
            high - low,
            (high - close.shift()).abs(),
            (low - close.shift()).abs(),
        ], axis=1).max(axis=1)
        plus_dm = high.diff()
        minus_dm = -low.diff()
        plus_dm = plus_dm.where((plus_dm > minus_dm) & (plus_dm > 0), 0.0)
        minus_dm = minus_dm.where((minus_dm > plus_dm) & (minus_dm > 0), 0.0)
        atr_s = tr.ewm(alpha=1/period, min_periods=period).mean()
        plus_di = 100 * (plus_dm.ewm(alpha=1/period, min_periods=period).mean() / atr_s)
        minus_di = 100 * (minus_dm.ewm(alpha=1/period, min_periods=period).mean() / atr_s)
        dx = 100 * (plus_di - minus_di).abs() / (plus_di + minus_di).replace(0, np.nan)
        return dx.ewm(alpha=1/period, min_periods=period).mean()

    # ── Private: data slicing ──────────────────────────────────────

    def _apply_window(self, df: pd.DataFrame) -> pd.DataFrame:
        bt = self.config.backtest
        if bt.start_date:
            start = pd.Timestamp(bt.start_date, tz="UTC")
            df = df[df.index >= start]
        if bt.end_date:
            end = pd.Timestamp(bt.end_date, tz="UTC")
            df = df[df.index <= end]
        return df

    @staticmethod
    def _bar_close_time(open_time: pd.Timestamp, tf: Timeframe) -> pd.Timestamp:
        return open_time + pd.Timedelta(minutes=TF_MINUTES[tf])

    def _slice_dfs(
        self, dfs: Dict[Timeframe, pd.DataFrame], decision_time: pd.Timestamp
    ) -> Dict[Timeframe, pd.DataFrame]:
        """Return only bars whose close time ≤ decision_time (no lookahead)."""
        sliced: Dict[Timeframe, pd.DataFrame] = {}
        for tf, df in dfs.items():
            close_times = df.index + pd.Timedelta(minutes=TF_MINUTES[tf])
            sliced[tf] = df[close_times <= decision_time]
        return sliced

    # ── Private: trade execution ───────────────────────────────────

    def _open_trade(
        self, report: SignalReport, next_bar: pd.Series, balance: float
    ) -> Dict:
        """Open at next bar's open, applying spread + slippage."""
        ts: TradeSetup = report.trade_setup  # type: ignore[assignment]
        bt = self.config.backtest

        raw_entry = float(next_bar["Open"])

        # Sanity-test inversion: flip direction and mirror SL/TP around entry
        direction = ts.direction
        sl = ts.stop_loss
        tp = ts.take_profit_1
        if bt.invert_signal:
            direction = "SELL" if direction == "BUY" else "BUY"
            sl, tp = (2 * raw_entry - sl), (2 * raw_entry - tp)

        # Cost in price units = (half-spread + slippage) × pip_value
        half_spread_pips = 0.5 if bt.use_spread_cost else 0.0
        cost_pips = half_spread_pips + bt.slippage_pips
        cost_price = cost_pips * PIP_VALUE

        if direction == "BUY":
            entry_price = raw_entry + cost_price  # buy higher (worse)
        else:
            entry_price = raw_entry - cost_price  # sell lower (worse)

        # Position sizing — risk a fixed % of current balance
        risk_pct = bt.risk_per_trade_pct / 100.0
        risk_amount = balance * risk_pct
        risk_pips = max(ts.risk_pips, 1e-9)
        pip_currency_value = risk_amount / risk_pips  # $ per pip

        # ── Diagnostic snapshot at entry ─────────────────────────
        analysis = report.primary_analysis
        adx_val = 0.0
        atr_pct = 0.0
        rsi_val = 50.0
        ema_dist = 0.0
        if analysis is not None:
            adx_sig = analysis.indicators.get("adx")
            if adx_sig is not None:
                adx_val = float(adx_sig.value)
            atr_pct = float(analysis.key_levels.get("atr_pct", 0.0))
            rsi_sig = analysis.indicators.get("rsi")
            if rsi_sig is not None:
                rsi_val = float(rsi_sig.value)
            ema_trend = analysis.key_levels.get("ema_trend")
            if ema_trend and ema_trend > 0:
                ema_dist = (raw_entry - ema_trend) / ema_trend * 100.0

        # Per-timeframe scores from MTF map (keys are tf.value strings)
        m15_s = float(report.mtf_scores.get("15min", 0.0))
        h1_s = float(report.mtf_scores.get("1h", 0.0))
        h4_s = float(report.mtf_scores.get("4h", 0.0))

        entry_ts = next_bar.name
        entry_hour = int(entry_ts.hour) if hasattr(entry_ts, "hour") else -1
        entry_dow = int(entry_ts.dayofweek) if hasattr(entry_ts, "dayofweek") else -1

        return {
            "direction": direction,
            "entry_time": next_bar.name,
            "entry_price": entry_price,
            "stop_loss": sl,
            "take_profit": tp,
            "risk_pips": risk_pips,
            "pip_currency_value": pip_currency_value,
            "balance_at_open": balance,
            "grade": report.signal_quality,
            "score_at_entry": float(report.final_score),
            "confidence_at_entry": float(report.final_confidence),
            "bar_index_open": next_bar.name,
            # Diagnostics
            "entry_hour": entry_hour,
            "entry_dow": entry_dow,
            "adx_at_entry": adx_val,
            "atr_pct_at_entry": atr_pct,
            "market_regime": report.market_regime or "",
            "volatility_regime": report.volatility_regime or "",
            "mtf_agreement": bool(report.mtf_agreement),
            "m15_score": m15_s,
            "h1_score": h1_s,
            "h4_score": h4_s,
            "rsi_at_entry": rsi_val,
            "ema_distance_pct": ema_dist,
        }

    def _check_exit(
        self, position: Dict, bar: pd.Series
    ) -> Optional[Dict]:
        """Return {price, reason} if SL or TP hit, else None.

        On the entry bar (bar with same timestamp as position's entry_time),
        we already filled at the open — only check subsequent bars' OHLC.
        """
        if bar.name == position["bar_index_open"]:
            return None  # entry bar — no exit check

        bt = self.config.backtest
        high = float(bar["High"])
        low = float(bar["Low"])

        sl = position["stop_loss"]
        tp = position["take_profit"]
        slip = bt.slippage_pips * PIP_VALUE
        half_spread = (0.5 if bt.use_spread_cost else 0.0) * PIP_VALUE

        if position["direction"] == "BUY":
            sl_hit = low <= sl
            tp_hit = high >= tp
            if sl_hit and tp_hit:
                # Conservative: SL first
                return {"price": sl - slip - half_spread, "reason": "sl"} if bt.sl_first_on_tie else \
                       {"price": tp - slip - half_spread, "reason": "tp"}
            if sl_hit:
                return {"price": sl - slip - half_spread, "reason": "sl"}
            if tp_hit:
                return {"price": tp - slip - half_spread, "reason": "tp"}
        else:  # SELL
            sl_hit = high >= sl
            tp_hit = low <= tp
            if sl_hit and tp_hit:
                return {"price": sl + slip + half_spread, "reason": "sl"} if bt.sl_first_on_tie else \
                       {"price": tp + slip + half_spread, "reason": "tp"}
            if sl_hit:
                return {"price": sl + slip + half_spread, "reason": "sl"}
            if tp_hit:
                return {"price": tp + slip + half_spread, "reason": "tp"}
        return None

    def _close_trade(
        self, position: Dict, exit_time: pd.Timestamp, exit_info: Dict, balance: float
    ) -> Trade:
        exit_price = exit_info["price"]
        reason = exit_info["reason"]

        if position["direction"] == "BUY":
            pnl_pips = (exit_price - position["entry_price"]) / PIP_VALUE
        else:
            pnl_pips = (position["entry_price"] - exit_price) / PIP_VALUE

        pnl_currency = pnl_pips * position["pip_currency_value"]
        pnl_pct = pnl_currency / position["balance_at_open"] * 100

        # Duration in primary-TF bars (timestamp diff / period)
        period_min = TF_MINUTES[self.config.primary_timeframe]
        duration_minutes = (exit_time - position["entry_time"]).total_seconds() / 60
        duration_bars = int(round(duration_minutes / period_min))

        return Trade(
            entry_time=position["entry_time"],
            exit_time=exit_time,
            direction=position["direction"],
            entry_price=position["entry_price"],
            exit_price=exit_price,
            stop_loss=position["stop_loss"],
            take_profit=position["take_profit"],
            exit_reason=reason,
            risk_pips=position["risk_pips"],
            pnl_pips=pnl_pips,
            pnl_pct=pnl_pct,
            pnl_currency=pnl_currency,
            duration_bars=duration_bars,
            grade=position["grade"],
            score_at_entry=position["score_at_entry"],
            confidence_at_entry=position["confidence_at_entry"],
            entry_hour=position.get("entry_hour", -1),
            entry_dow=position.get("entry_dow", -1),
            adx_at_entry=position.get("adx_at_entry", 0.0),
            atr_pct_at_entry=position.get("atr_pct_at_entry", 0.0),
            market_regime=position.get("market_regime", ""),
            volatility_regime=position.get("volatility_regime", ""),
            mtf_agreement=position.get("mtf_agreement", False),
            m15_score=position.get("m15_score", 0.0),
            h1_score=position.get("h1_score", 0.0),
            h4_score=position.get("h4_score", 0.0),
            rsi_at_entry=position.get("rsi_at_entry", 50.0),
            ema_distance_pct=position.get("ema_distance_pct", 0.0),
        )


# ══════════════════════════════════════════════════════════════════
#  PARAMETER OPTIMIZER (PARALLEL)
# ══════════════════════════════════════════════════════════════════

def _apply_overrides(config: PlatformConfig, overrides: Dict[str, float]) -> PlatformConfig:
    """Return a deep-copied config with parameter overrides applied."""
    new_config = copy.deepcopy(config)

    if "atr_sl_multiplier" in overrides:
        new_config.signals.atr_sl_multiplier = float(overrides["atr_sl_multiplier"])
    if "default_rr_ratio" in overrides:
        new_config.signals.default_rr_ratio = float(overrides["default_rr_ratio"])
    if "min_trade_grade" in overrides:
        new_config.backtest.min_trade_grade = str(overrides["min_trade_grade"])
    if "min_abs_score" in overrides:
        new_config.backtest.min_abs_score = float(overrides["min_abs_score"])
    if "invert_signal" in overrides:
        new_config.backtest.invert_signal = bool(overrides["invert_signal"])
    if "regime_filter" in overrides:
        new_config.backtest.regime_filter = bool(overrides["regime_filter"])
    if "regime_filter_min_adx" in overrides:
        new_config.backtest.regime_filter_min_adx = float(overrides["regime_filter_min_adx"])
    if "session_filter" in overrides:
        new_config.backtest.session_filter = bool(overrides["session_filter"])
    if "session_start_hour" in overrides:
        new_config.backtest.session_start_hour = int(overrides["session_start_hour"])
    if "session_end_hour" in overrides:
        new_config.backtest.session_end_hour = int(overrides["session_end_hour"])
    if "auto_session_filter" in overrides:
        new_config.backtest.auto_session_filter = bool(overrides["auto_session_filter"])
    if "auto_session_lookback_trades" in overrides:
        new_config.backtest.auto_session_lookback_trades = int(overrides["auto_session_lookback_trades"])
    if "auto_session_threshold_pct" in overrides:
        new_config.backtest.auto_session_threshold_pct = float(overrides["auto_session_threshold_pct"])
    if "risk_per_trade_pct" in overrides:
        new_config.backtest.risk_per_trade_pct = float(overrides["risk_per_trade_pct"])
    if "disable_indicator" in overrides:
        v = overrides["disable_indicator"]
        new_config.technical.disabled_indicators = [str(v)] if v else []

    return new_config


def _run_single_combo(args: Tuple[PlatformConfig, Dict[str, float]]) -> BacktestResult:
    """Worker function — must be top-level for multiprocessing.Pool."""
    base_config, overrides = args
    config = _apply_overrides(base_config, overrides)
    backtester = WalkForwardBacktester(config)
    # Workers reuse the disk parquet cache, so no Twelve Data fetch needed
    result = backtester.run(progress=False)
    result.params = overrides
    return result


class Optimizer:
    """Embarrassingly parallel grid search over BacktestConfig.param_grid."""

    def __init__(self, config: PlatformConfig):
        self.config = config

    def run_grid(
        self,
        param_grid: Optional[Dict[str, List[float]]] = None,
        prewarm_cache: bool = True,
    ) -> List[BacktestResult]:
        if param_grid is None:
            param_grid = self.config.backtest.param_grid

        # Prewarm the disk cache once before fanning out — avoids each worker
        # re-fetching from Twelve Data.
        if prewarm_cache:
            logger.info("Prewarming disk cache for backtest data...")
            warmer = WalkForwardBacktester(self.config)
            warmer.load_data()

        keys = list(param_grid.keys())
        value_lists = [param_grid[k] for k in keys]
        combos = [
            dict(zip(keys, vals))
            for vals in itertools.product(*value_lists)
        ]

        n_workers = min(
            self.config.backtest.parallel_workers,
            len(combos),
            os.cpu_count() or 1,
        )
        logger.info(
            f"Optimizer: {len(combos)} combos × {n_workers} workers"
        )

        tasks = [(self.config, c) for c in combos]

        if n_workers <= 1:
            results = [_run_single_combo(t) for t in tasks]
        else:
            with mp.Pool(processes=n_workers) as pool:
                results = pool.map(_run_single_combo, tasks)

        # Sort by total return descending
        results.sort(
            key=lambda r: r.metrics.get("total_return_pct", -float("inf")),
            reverse=True,
        )
        return results