engine.py

"""Backtest engine for an ICT-style intraday strategy.

This module is pure and testable: it does not depend on Streamlit, nor does it
perform network I/O inside the backtest logic. Data downloading is isolated in
``download_data`` so that the engine can be tested with synthetic datasets.

Quick glossary (audience: maintainer / reviewer):
  * RTH   : Regular Trading Hours, 09:30-16:00 New York time (ET).
  * PDH/PDL : Previous Day High/Low, the previous day's RTH high/low.
  * FVG   : Fair Value Gap, the 3-bar imbalance described in ``detect_fvgs``.
  * Sweep : a liquidity grab beyond PDH (buyside) or PDL (sellside).
  * R     : risk unit = |stop - entry|. All outcomes are expressed in R.

The mechanical rules are documented in the project prompt and faithfully
replicated here. Wherever a choice is not unambiguous it is flagged with a
"why" comment.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import time
from typing import Literal, Optional

import numpy as np
import pandas as pd

# Reference timezone: all intraday logic works in ET.
ET_TZ = "America/New_York"

# RTH (Regular Trading Hours) session boundaries.
RTH_START = time(9, 30)
RTH_END = time(16, 0)

Direction = Literal["long", "short"]
TargetMethod = Literal["rr_fisso", "liquidita", "trailing"]
ExitReason = Literal["target", "stop", "time-stop"]


@dataclass(frozen=True)
class BacktestParams:
    """Initial backtest parameters.

    The defaults replicate those required by the prompt. All percentage
    parameters are expressed as fractions (0.0005 = 0.05%).
    """

    overnight: bool = False
    target_method: TargetMethod = "rr_fisso"
    rr: float = 2.0
    trailing_pct: float = 0.005  # used only if target_method == "trailing"
    max_hold_days: int = 5  # active only if overnight=True
    swing_win: int = 2
    eq_tol: float = 0.0010  # equal highs/lows tolerance (0.10%) — reserved
    last_entry: time = time(14, 30)
    max_trades_day: int = 1
    buffer: float = 0.0005  # 0.05% beyond the swept extreme for the stop
    stop_mult: float = 1.0  # stop-width multiplier (1.0 = base)
    fvg_lookback: int = 6
    cost: float = 0.0002  # 0.02% per side

    def __post_init__(self) -> None:
        # Explicit boundary validation: inconsistent parameters would make the
        # results meaningless, so it is better to fail fast.
        if self.rr <= 0:
            raise ValueError("rr must be > 0")
        if self.max_hold_days < 1:
            raise ValueError("max_hold_days must be >= 1")
        if self.swing_win < 1:
            raise ValueError("swing_win must be >= 1")
        if self.max_trades_day < 1:
            raise ValueError("max_trades_day must be >= 1")
        if self.buffer < 0 or self.cost < 0 or self.trailing_pct <= 0:
            raise ValueError("buffer/cost must be >= 0 and trailing_pct > 0")
        if self.stop_mult <= 0:
            raise ValueError("stop_mult must be > 0")
        if self.target_method not in ("rr_fisso", "liquidita", "trailing"):
            raise ValueError(f"invalid target_method: {self.target_method}")


@dataclass(frozen=True)
class FVG:
    """3-bar Fair Value Gap.

    ``lo``/``hi`` delimit the imbalance zone. ``formed_idx`` is the (positional,
    within the sub-session) index of bar i+1 that completes the pattern: from
    that index onward the FVG is "visible" and can act as a trigger.
    """

    direction: Literal["bull", "bear"]
    lo: float
    hi: float
    formed_idx: int


@dataclass
class Trade:
    """Outcome of a single trade, expressed in multiples of R net of costs."""

    date: pd.Timestamp
    direction: Direction
    entry: float
    stop: float
    target: float
    r_multiple: float
    exit_reason: ExitReason


def detect_fvgs(sub: pd.DataFrame) -> list[FVG]:
    """Detect 3-bar Fair Value Gaps within a sub-session.

    For each triplet (i-1, i, i+1):
      * Bullish FVG if ``low[i+1] > high[i-1]`` -> zone [high[i-1], low[i+1]]
      * Bearish FVG if ``high[i+1] < low[i-1]`` -> zone [high[i+1], low[i-1]]

    Args:
        sub: bars of a single RTH day with columns open/high/low/close.

    Returns:
        List of :class:`FVG` ordered by ``formed_idx``. ``formed_idx`` is the
        positional index of bar i+1 (the one that completes the pattern).
    """
    fvgs: list[FVG] = []
    high = sub["high"].to_numpy()
    low = sub["low"].to_numpy()
    n = len(sub)
    # i runs over the middle bar; we need i-1 and i+1, hence 1 <= i <= n-2.
    for i in range(1, n - 1):
        if low[i + 1] > high[i - 1]:
            fvgs.append(FVG("bull", lo=high[i - 1], hi=low[i + 1], formed_idx=i + 1))
        elif high[i + 1] < low[i - 1]:
            fvgs.append(FVG("bear", lo=high[i + 1], hi=low[i - 1], formed_idx=i + 1))
    return fvgs


def find_swings(
    sub: pd.DataFrame, swing_win: int
) -> tuple[list[tuple[int, float]], list[tuple[int, float]]]:
    """Identify swing highs/lows over a symmetric ±``swing_win`` bar window.

    A swing high is a local maximum: ``high[i]`` is the maximum within the
    window [i-swing_win, i+swing_win]. Mirror logic for swing lows.

    Returns:
        (swing_highs, swing_lows) as lists of tuples (positional_index, price).
        They serve as the liquidity pool for the "liquidita" target.
    """
    high = sub["high"].to_numpy()
    low = sub["low"].to_numpy()
    n = len(sub)
    swing_highs: list[tuple[int, float]] = []
    swing_lows: list[tuple[int, float]] = []
    for i in range(swing_win, n - swing_win):
        window_h = high[i - swing_win : i + swing_win + 1]
        window_l = low[i - swing_win : i + swing_win + 1]
        # >= / <= on the center: on a plateau the swing is registered anyway.
        if high[i] == window_h.max():
            swing_highs.append((i, float(high[i])))
        if low[i] == window_l.min():
            swing_lows.append((i, float(low[i])))
    return swing_highs, swing_lows


def _compute_target(
    direction: Direction,
    method: TargetMethod,
    entry: float,
    risk: float,
    rr: float,
    swing_highs: list[tuple[int, float]],
    swing_lows: list[tuple[int, float]],
    pdh: float,
    pdl: float,
) -> float:
    """Compute the target price according to the chosen method.

    For ``trailing`` the nominal "target" is placed very far away: the actual
    exit is handled by the trailing stop in :func:`_simulate`, so the fixed
    target must never be the constraint that closes the trade.
    """
    if method == "rr_fisso":
        return entry - rr * risk if direction == "short" else entry + rr * risk

    if method == "liquidita":
        if direction == "short":
            # Nearest pool below the entry: the swing low just below it.
            candidates = [p for _, p in swing_lows if p < entry]
            return max(candidates) if candidates else pdl  # fallback to PDL
        candidates = [p for _, p in swing_highs if p > entry]
        return min(candidates) if candidates else pdh  # fallback to PDH

    # trailing: "unreachable" target in the direction of the trade.
    return -np.inf if direction == "short" else np.inf


def _simulate(
    direction: Direction,
    entry: float,
    stop: float,
    target: float,
    risk: float,
    entry_idx: int,
    day_bars: pd.DataFrame,
    future_bars: pd.DataFrame,
    params: BacktestParams,
) -> tuple[float, ExitReason]:
    """Simulate a trade's evolution bar by bar and return (R, reason).

    The outcome is in multiples of R net of round-trip costs
    (``cost_R = 2 * cost * entry / risk``).

    Rules replicated from the prompt:
      * If both stop and target are touched within the same bar, the STOP
        counts (conservative assumption, no intrabar look-ahead).
      * overnight=NO: we iterate only until end of day (16:00 time-stop).
      * overnight=YES: we continue until target/stop for at most
        ``max_hold_days`` trading days.
      * If nothing is hit, exit at the close of the last available bar.

    Args:
        entry_idx: positional index of the entry bar within ``day_bars``.
        day_bars: RTH bars of the entry day.
        future_bars: RTH bars of the following days (used only if overnight).
    """
    cost_r = 2.0 * params.cost * entry / risk

    # We build the sequence of bars to iterate over after entry.
    # We start from the bar after the entry: execution happens at the close of
    # the trigger bar, so that bar cannot close the trade.
    after_today = day_bars.iloc[entry_idx + 1 :]
    if params.overnight:
        # Limit to the allowed holding trading days (entry day included ->
        # max_hold_days-1 full subsequent days remain).
        if params.max_hold_days > 1 and len(future_bars) > 0:
            future_days = future_bars.index.normalize().unique()
            allowed = future_days[: params.max_hold_days - 1]
            extra = future_bars[future_bars.index.normalize().isin(allowed)]
        else:
            extra = future_bars.iloc[:0]
        walk = pd.concat([after_today, extra])
    else:
        walk = after_today

    last_close = entry  # if there are no bars, we exit flat at the entry
    # State for the trailing stop; we update the stop following the price.
    trail_stop = stop
    for _, bar in walk.iterrows():
        hi = float(bar["high"])
        lo = float(bar["low"])
        last_close = float(bar["close"])

        if params.target_method == "trailing":
            # Update the trailing stop based on the bar's favorable extreme.
            if direction == "short":
                trail_stop = min(trail_stop, hi * (1 + params.trailing_pct))
                if hi >= trail_stop:
                    r = (entry - trail_stop) / risk - cost_r
                    return r, "stop"
            else:
                trail_stop = max(trail_stop, lo * (1 - params.trailing_pct))
                if lo <= trail_stop:
                    r = (trail_stop - entry) / risk - cost_r
                    return r, "stop"
            continue

        if direction == "short":
            hit_stop = hi >= stop
            hit_target = lo <= target
            # Conservative: stop takes precedence if both are touched in the same bar.
            if hit_stop:
                return -1.0 - cost_r, "stop"
            if hit_target:
                # (entry - target)/risk equals rr for rr_fisso, and is generic for
                # the "liquidita" method where the target is a liquidity pool.
                return (entry - target) / risk - cost_r, "target"
        else:
            hit_stop = lo <= stop
            hit_target = hi >= target
            if hit_stop:
                return -1.0 - cost_r, "stop"
            if hit_target:
                return (target - entry) / risk - cost_r, "target"

    # No level hit: market exit (time-stop / end of window).
    if direction == "short":
        r = (entry - last_close) / risk - cost_r
    else:
        r = (last_close - entry) / risk - cost_r
    return r, "time-stop"


def run_backtest(df: pd.DataFrame, params: BacktestParams) -> list[Trade]:
    """Run the backtest on a single instrument.

    Args:
        df: RTH DataFrame ordered by timestamp, tz-aware index in ET, with
            columns ``open/high/low/close``. Each row is an intraday bar.
        params: strategy parameters.

    Returns:
        List of :class:`Trade` in chronological order.
    """
    if df.empty:
        return []

    df = df.sort_index()
    # Unique trading days (ET dates), sorted.
    df_days = df.index.normalize()
    days = list(pd.Index(df_days).unique())

    trades: list[Trade] = []

    for di in range(1, len(days)):
        prev_day = df[df_days == days[di - 1]]
        sub = df[df_days == days[di]]
        if len(prev_day) == 0 or len(sub) < 4:
            # We need >= 4 bars to have at least one FVG and one trigger bar.
            continue

        pdh = float(prev_day["high"].max())
        pdl = float(prev_day["low"].min())

        # Bars of the following days: used only for overnight holding.
        future_bars = df[df_days > days[di]]

        fvgs = detect_fvgs(sub)
        swing_highs, swing_lows = find_swings(sub, params.swing_win)

        sub_high = sub["high"].to_numpy()
        sub_low = sub["low"].to_numpy()
        sub_close = sub["close"].to_numpy()
        sub_times = sub.index

        taken = 0
        swept_up = swept_dn = False
        swept_high = -np.inf
        swept_low = np.inf

        # k runs over the bars from the third (index 2) to the second-to-last.
        for k in range(2, len(sub) - 1):
            if taken >= params.max_trades_day:
                break
            if sub_times[k].time() > params.last_entry:
                break

            if sub_high[k] > pdh:
                swept_up = True
                swept_high = max(swept_high, sub_high[k])
            if sub_low[k] < pdl:
                swept_dn = True
                swept_low = min(swept_low, sub_low[k])

            entered = False

            # --- SHORT: after a buyside sweep, reversal on a bullish FVG ---
            if swept_up:
                for fvg in fvgs:
                    if fvg.direction != "bull":
                        continue
                    if not (k - params.fvg_lookback <= fvg.formed_idx <= k):
                        continue
                    if sub_close[k] < fvg.lo:
                        entry = float(sub_close[k])
                        # Base stop beyond the swept extreme; stop_mult widens
                        # (or narrows) its size relative to the entry.
                        base_stop = swept_high * (1 + params.buffer)
                        stop = entry + params.stop_mult * (base_stop - entry)
                        risk = stop - entry
                        if risk <= 0:
                            continue
                        target = _compute_target(
                            "short", params.target_method, entry, risk, params.rr,
                            swing_highs, swing_lows, pdh, pdl,
                        )
                        r, reason = _simulate(
                            "short", entry, stop, target, risk, k,
                            sub, future_bars, params,
                        )
                        trades.append(Trade(days[di], "short", entry, stop, target, r, reason))
                        taken += 1
                        entered = True
                        break

            if entered or taken >= params.max_trades_day:
                continue

            # --- LONG: after a sellside sweep, reversal on a bearish FVG ---
            if swept_dn:
                for fvg in fvgs:
                    if fvg.direction != "bear":
                        continue
                    if not (k - params.fvg_lookback <= fvg.formed_idx <= k):
                        continue
                    if sub_close[k] > fvg.hi:
                        entry = float(sub_close[k])
                        base_stop = swept_low * (1 - params.buffer)
                        stop = entry - params.stop_mult * (entry - base_stop)
                        risk = entry - stop
                        if risk <= 0:
                            continue
                        target = _compute_target(
                            "long", params.target_method, entry, risk, params.rr,
                            swing_highs, swing_lows, pdh, pdl,
                        )
                        r, reason = _simulate(
                            "long", entry, stop, target, risk, k,
                            sub, future_bars, params,
                        )
                        trades.append(Trade(days[di], "long", entry, stop, target, r, reason))
                        taken += 1
                        break

    return trades


def trades_to_dataframe(trades: list[Trade]) -> pd.DataFrame:
    """Convert the list of trades into a DataFrame ready for UI/CSV."""
    if not trades:
        return pd.DataFrame(
            columns=["data", "direzione", "entry", "stop", "target", "R", "motivo_uscita"]
        )
    rows = [
        {
            "data": t.date.date() if hasattr(t.date, "date") else t.date,
            "direzione": t.direction,
            "entry": round(t.entry, 4),
            "stop": round(t.stop, 4),
            "target": (None if not np.isfinite(t.target) else round(t.target, 4)),
            "R": round(t.r_multiple, 4),
            "motivo_uscita": t.exit_reason,
        }
        for t in trades
    ]
    return pd.DataFrame(rows)


def compute_stats(trades: list[Trade]) -> dict:
    """Compute the summary metrics of the backtest.

    Returns:
        dict with: n_trade, win_rate, aspettativa_R, R_totale, profit_factor,
        max_drawdown_R, breakdown (count per exit reason) and equity
        (cumulative R series for the curve).
    """
    n = len(trades)
    breakdown = {"target": 0, "stop": 0, "time-stop": 0}
    for t in trades:
        breakdown[t.exit_reason] += 1

    if n == 0:
        return {
            "n_trade": 0,
            "win_rate": 0.0,
            "aspettativa_R": 0.0,
            "R_totale": 0.0,
            "profit_factor": 0.0,
            "max_drawdown_R": 0.0,
            "breakdown": breakdown,
            "equity": [],
        }

    r_values = np.array([t.r_multiple for t in trades], dtype=float)
    wins = r_values[r_values > 0]
    losses = r_values[r_values < 0]

    gross_profit = float(wins.sum())
    gross_loss = float(-losses.sum())
    profit_factor = gross_profit / gross_loss if gross_loss > 0 else float("inf")

    equity = np.cumsum(r_values)
    running_max = np.maximum.accumulate(equity)
    drawdowns = equity - running_max
    max_dd = float(drawdowns.min()) if len(drawdowns) else 0.0

    return {
        "n_trade": n,
        "win_rate": float(len(wins) / n),
        "aspettativa_R": float(r_values.mean()),
        "R_totale": float(r_values.sum()),
        "profit_factor": profit_factor,
        "max_drawdown_R": max_dd,
        "breakdown": breakdown,
        "equity": equity.tolist(),
    }


def _trend_metrics(returns: pd.Series, starting_budget: float) -> dict:
    """Performance metrics from a series of net daily returns."""
    returns = returns.dropna()
    if returns.empty:
        return {"return_pct": 0.0, "cagr": 0.0, "vol": 0.0, "sharpe": 0.0,
                "max_drawdown_pct": 0.0, "budget_finale": starting_budget}
    total = float((1 + returns).prod() - 1)
    years = len(returns) / 252.0
    cagr = (1 + total) ** (1 / years) - 1 if years > 0 and total > -1 else 0.0
    vol = float(returns.std() * np.sqrt(252))
    sharpe = float(returns.mean() * 252 / vol) if vol > 0 else 0.0
    eq = (1 + returns).cumprod()
    max_dd = float((eq / eq.cummax() - 1).min())
    return {
        "return_pct": total, "cagr": cagr, "vol": vol, "sharpe": sharpe,
        "max_drawdown_pct": max_dd, "budget_finale": starting_budget * (1 + total),
    }


def backtest_trend(
    data: dict, sma_window: int = 200, allow_short: bool = False,
    cost: float = 0.0002, starting_budget: float = 10000.0,
) -> dict | None:
    """Backtest of a trend-following strategy (time-series momentum).

    Rule: we are LONG when the closing price is above the ``sma_window``-day
    moving average; otherwise flat (or short, if ``allow_short``). A given day's
    signal is applied to the next day's return (``shift(1)``), so there is
    **no look-ahead**. With multiple instruments an equal-weighted portfolio is
    built. It is a strategy on **daily** bars.

    Args:
        data: mapping ticker -> daily DataFrame (column ``close``).
        sma_window: moving-average window (e.g. 200 days).
        allow_short: if True, go short below the average, otherwise stay flat.
        cost: cost per position change (fraction, e.g. 0.0002).
        starting_budget: the fund's initial capital.

    Returns:
        dict with curves (dates, strategy equity, buy&hold equity), strategy
        and buy&hold metrics, average exposure and number of switches. ``None``
        if no instrument has sufficient data.
    """
    rets: dict = {}
    bh: dict = {}
    expo: list[float] = []
    switches = 0
    for tk, df in data.items():
        close = df["close"].astype(float)
        if len(close) < sma_window + 2:
            continue
        sma = close.rolling(sma_window).mean()
        sig = pd.Series(0.0, index=close.index)
        sig[close > sma] = 1.0
        if allow_short:
            sig[close < sma] = -1.0
        daily = close.pct_change()
        pos = sig.shift(1).fillna(0.0)  # act the day AFTER the signal
        switch = pos.diff().abs().fillna(0.0)
        rets[tk] = pos * daily - switch * cost
        bh[tk] = daily
        expo.append(float((pos != 0).mean()))
        switches += int((switch > 0).sum())

    if not rets:
        return None

    # Equal-weighted portfolio: average of the returns over the available instruments.
    port = pd.DataFrame(rets).mean(axis=1).dropna()
    port_bh = pd.DataFrame(bh).mean(axis=1).reindex(port.index)
    eq = starting_budget * (1 + port).cumprod()
    eq_bh = starting_budget * (1 + port_bh).cumprod()
    return {
        "dates": [d.date() if hasattr(d, "date") else d for d in port.index],
        "equity": eq.tolist(),
        "equity_bh": eq_bh.tolist(),
        "metrics": _trend_metrics(port, starting_budget),
        "metrics_bh": _trend_metrics(port_bh, starting_budget),
        "exposure": float(np.mean(expo)) if expo else 0.0,
        "switches": switches,
    }


def split_by_date(df: pd.DataFrame, frac: float) -> tuple[pd.DataFrame, pd.DataFrame]:
    """Split a DataFrame into two parts by date: first ``frac``, the rest after.

    Used for in-sample / out-of-sample validation: you optimize on the first
    part (the past) and verify on the second (a never-seen future), to expose
    over-fitted combinations.

    Args:
        df: data for a single instrument (datetime index).
        frac: initial share (0..1) allocated to the in-sample. E.g. 0.7 = 70%.

    Returns:
        (in_sample, out_of_sample). If there are too few days, the out-of-sample
        may turn out empty.
    """
    if df.empty:
        return df, df
    days = pd.Index(df.index.normalize()).unique()
    if len(days) < 2:
        return df, df.iloc[:0]
    cut_idx = min(max(int(len(days) * frac), 1), len(days) - 1)
    cut_day = days[cut_idx]
    mask = df.index.normalize() < cut_day
    return df[mask], df[~mask]


def build_param_grid(base: dict, grid: dict) -> list[BacktestParams]:
    """Build all combinations of :class:`BacktestParams` from the grid.

    Args:
        base: fixed parameters (not varied).
        grid: mapping parameter_name -> list of values to try.

    Returns:
        List of ``BacktestParams``, one per combination (Cartesian product).
    """
    import itertools

    keys = list(grid.keys())
    combos: list[BacktestParams] = []
    for values in itertools.product(*[grid[k] for k in keys]):
        kw = dict(base)
        kw.update(dict(zip(keys, values)))
        combos.append(BacktestParams(**kw))
    return combos


def _pool_stats(data: dict, params: BacktestParams) -> dict:
    """Run the backtest on all instruments and aggregate the trades into a single
    statistic (R is already normalized, so the tickers are comparable)."""
    pool: list[Trade] = []
    for df in data.values():
        pool.extend(run_backtest(df, params))
    return compute_stats(pool)


def optimize_grid(
    data: dict, combos: list[BacktestParams], split_frac: float | None = 0.7
) -> list[dict]:
    """Evaluate each parameter combination over the instrument universe.

    Trades from all instruments are aggregated to obtain a large sample. If
    ``split_frac`` is set, each series is split into in-sample (optimization)
    and out-of-sample (verification), and both statistics are reported: a
    combination is credible only if it holds up **out of sample as well**.

    Args:
        data: mapping ticker -> DataFrame (already downloaded and normalized).
        combos: parameter combinations to evaluate.
        split_frac: in-sample share (0..1) or ``None`` to use the whole period.

    Returns:
        List of dicts, one per combination, with key ``params`` and the
        statistics (``all`` if no split, otherwise ``is`` and ``oos``).
    """
    results: list[dict] = []
    for params in combos:
        if split_frac is None:
            results.append({"params": params, "all": _pool_stats(data, params)})
            continue
        is_pool: list[Trade] = []
        oos_pool: list[Trade] = []
        for df in data.values():
            ins, oos = split_by_date(df, split_frac)
            is_pool.extend(run_backtest(ins, params))
            oos_pool.extend(run_backtest(oos, params))
        results.append(
            {"params": params,
             "is": compute_stats(is_pool),
             "oos": compute_stats(oos_pool)}
        )
    return results


def compute_money_curve(
    trades: list[Trade], starting_budget: float, max_size_pct: float
) -> dict:
    """Translate the R outcomes into a monetary evolution of the capital.

    Position sizing: on each trade we commit, as notional, a share equal to
    ``max_size_pct`` of the *current* budget — so the capital compounds over
    time. Given the notional, the number of shares is ``notional / entry`` and
    the monetary P&L is::

        rischio_per_azione = |stop - entry|
        azioni             = notional / entry
        pnl                = R * azioni * rischio_per_azione

    where ``R`` is already net of costs. ``rischio_per_azione`` derives from the
    trade's real prices, so the P&L reflects the actual stop width.

    Args:
        trades: chronological list of trades.
        starting_budget: initial capital (> 0), in the currency of the prices.
        max_size_pct: maximum size per trade as a fraction of the budget
            (0.10 = 10%). Must be > 0.

    Returns:
        dict with:
          * ``rows``: list of dicts for each trade (data, direzione, R, notional,
            azioni, pnl, budget_dopo).
          * ``budget_curve``: progressive budget (including the initial value).
          * ``budget_finale``, ``pnl_totale``, ``return_pct``, ``volume_totale``,
            ``max_drawdown_pct``.

    Raises:
        ValueError: if ``starting_budget`` <= 0 or ``max_size_pct`` <= 0.
    """
    if starting_budget <= 0:
        raise ValueError("starting_budget must be > 0")
    if max_size_pct <= 0:
        raise ValueError("max_size_pct must be > 0")

    budget = starting_budget
    curve = [starting_budget]
    rows: list[dict] = []
    volume_totale = 0.0

    for t in trades:
        # Committed notional: a share of the current budget. Never negative as
        # long as the capital stays positive; with budget <=0 the trade no
        # longer operates.
        notional = max(budget, 0.0) * max_size_pct
        risk_per_share = abs(t.stop - t.entry)
        shares = notional / t.entry if t.entry > 0 else 0.0
        pnl = t.r_multiple * shares * risk_per_share
        budget += pnl
        volume_totale += notional
        curve.append(budget)
        rows.append(
            {
                "data": t.date.date() if hasattr(t.date, "date") else t.date,
                "direzione": t.direction,
                "R": round(t.r_multiple, 4),
                "notional": round(notional, 2),
                "azioni": round(shares, 4),
                "pnl": round(pnl, 2),
                "budget_dopo": round(budget, 2),
            }
        )

    curve_arr = np.array(curve, dtype=float)
    running_max = np.maximum.accumulate(curve_arr)
    # Percentage drawdown from the capital's previous peak.
    dd_pct = float(((curve_arr - running_max) / running_max).min()) if len(curve_arr) else 0.0

    return {
        "rows": rows,
        "budget_curve": curve,
        "budget_finale": budget,
        "pnl_totale": budget - starting_budget,
        "return_pct": (budget - starting_budget) / starting_budget,
        "volume_totale": volume_totale,
        "max_drawdown_pct": dd_pct,
    }


def money_rows_to_dataframe(money: dict) -> pd.DataFrame:
    """Convert ``money['rows']`` into a DataFrame ready for UI/CSV."""
    cols = ["data", "direzione", "R", "notional", "azioni", "pnl", "budget_dopo"]
    if not money["rows"]:
        return pd.DataFrame(columns=cols)
    return pd.DataFrame(money["rows"])[cols]


def filter_rth(df: pd.DataFrame) -> pd.DataFrame:
    """Filter the bars to the RTH session (09:30-16:00 ET) and convert to ET.

    Accepts a DataFrame with a tz-aware index (any tz) and brings it back to ET.
    Bars with a timestamp exactly at 16:00 are excluded: they represent the
    opening of the interval that closes the session.
    """
    if df.empty:
        return df
    idx = df.index
    if idx.tz is None:
        # yfinance intraday data is tz-aware; if it is missing, we assume UTC.
        df = df.tz_localize("UTC")
    df = df.tz_convert(ET_TZ)
    t = df.index.time
    mask = (t >= RTH_START) & (t < RTH_END)
    return df[mask]


# Maximum span (days) of ONE intraday request to yfinance: Yahoo rejects or
# ignores wider windows, returning an empty DataFrame. For this reason requests
# must be split into chunks within these limits.
_INTERVAL_CHUNK_DAYS = {
    "1m": 7,
    "2m": 60, "5m": 60, "15m": 60, "30m": 60, "90m": 60,
    "60m": 730, "1h": 730,
}


def _yf_download(ticker: str, interval: str, start, end) -> pd.DataFrame:
    """Thin wrapper over yfinance.download (isolated for testability)."""
    import yfinance as yf

    return yf.download(
        ticker, interval=interval, start=start, end=end,
        auto_adjust=False, progress=False,
    )


def download_chunked(ticker: str, interval: str, start, end, fetch=_yf_download) -> pd.DataFrame:
    """Download the data by splitting the period into chunks within yfinance limits.

    Yahoo limits each intraday request to a maximum window (see
    ``_INTERVAL_CHUNK_DAYS``): a single request that is too wide comes back
    empty. This function iterates over allowed windows and concatenates whatever
    is available, **skipping** chunks that are empty or that raise transient
    errors. This way the system always recovers the existing data, even only
    partially, instead of failing wholesale.

    Args:
        ticker: symbol.
        interval: yfinance interval.
        start, end: period bounds (dates/strings/Timestamps).
        fetch: download function (default yfinance); injectable in tests.

    Returns:
        Raw concatenated DataFrame (columns as from yfinance), sorted and
        without duplicate timestamps. Empty if no chunk returned data.
    """
    chunk_days = _INTERVAL_CHUNK_DAYS.get(interval)
    frames: list[pd.DataFrame] = []

    if chunk_days is None:
        # Daily or interval with no known limit: a single request.
        try:
            raw = fetch(ticker, interval, start, end)
            if raw is not None and not raw.empty:
                frames.append(raw)
        except Exception:
            pass
    else:
        start_ts = pd.Timestamp(start).normalize()
        seg_end = pd.Timestamp(end).normalize()
        # We iterate BACKWARDS from end: Yahoo requires the start of each
        # intraday request to be within the maximum window (e.g. the last 60
        # days), so the most recent chunk must start close to end. A 2-day
        # margin avoids hitting the limit exactly and being rejected.
        step = pd.Timedelta(days=max(chunk_days - 2, 1))
        while seg_end > start_ts:
            seg_start = max(seg_end - step, start_ts)
            try:
                raw = fetch(ticker, interval, seg_start, seg_end)
                if raw is not None and not raw.empty:
                    frames.append(raw)
            except Exception:
                # Chunk failed (network/limit): we continue with the others.
                pass
            seg_end = seg_start

    if not frames:
        return pd.DataFrame()

    out = pd.concat(frames)
    # Adjacent chunks may overlap on a timestamp: dedup.
    out = out[~out.index.duplicated(keep="first")].sort_index()
    return out


# --------------------------------------------------------------------------- #
# Data providers (free) with automatic fallback
# --------------------------------------------------------------------------- #
# Each provider is a function (ticker, interval, start, end) -> raw OHLC-style
# DataFrame (columns Open/High/Low/Close, datetime index). It returns an empty
# frame if it has no data for that request (so the fallback kicks in). The
# providers are deliberately simple and API-key-free, to stay "free".


def _provider_yahoo(ticker: str, interval: str, start, end) -> pd.DataFrame:
    """Yahoo provider (yfinance), with chunked download for intraday."""
    return download_chunked(ticker, interval, start, end)


def _provider_stooq(ticker: str, interval: str, start, end) -> pd.DataFrame:
    """Stooq provider via public CSV. Supports daily data (free, reliable).

    For intraday, Stooq does not offer usable history without an account, so
    here it returns empty and leaves the fallback to other providers. US
    stock/ETF symbols on Stooq use the ``.us`` suffix (e.g. ``spy.us``).
    """
    if interval != "1d":
        return pd.DataFrame()

    import io
    import urllib.request

    sym = ticker.lower()
    if "." not in sym and "=" not in sym and "^" not in sym:
        sym = f"{sym}.us"
    d1 = pd.Timestamp(start).strftime("%Y%m%d")
    d2 = pd.Timestamp(end).strftime("%Y%m%d")

    # We try both Stooq hosts: sometimes one is reachable and the other is
    # behind an anti-bot gate (typical from datacenter IPs). If the body is not
    # a CSV ("Date,Open,..."), we consider it failed and the provider fallback
    # kicks in.
    body = ""
    for host in ("stooq.com", "stooq.pl"):
        url = f"https://{host}/q/d/l/?s={sym}&d1={d1}&d2={d2}&i=d"
        try:
            req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
            with urllib.request.urlopen(req, timeout=20) as resp:  # noqa: S310
                candidate = resp.read().decode("utf-8", errors="replace")
        except Exception:
            continue
        if candidate.lstrip().startswith("Date"):
            body = candidate
            break
    if not body:
        return pd.DataFrame()
    df = pd.read_csv(io.StringIO(body))
    if df.empty or "Date" not in df.columns:
        return pd.DataFrame()
    df["Date"] = pd.to_datetime(df["Date"])
    df = df.set_index("Date").sort_index()
    cols = [c for c in ("Open", "High", "Low", "Close") if c in df.columns]
    if len(cols) < 4:
        return pd.DataFrame()
    return df[["Open", "High", "Low", "Close"]]


# Registry of available providers. The keys are the names shown in the UI.
PROVIDERS = {
    "Yahoo (yfinance)": _provider_yahoo,
    "Stooq": _provider_stooq,
}


def _normalizza_ohlc(raw: pd.DataFrame, interval: str) -> pd.DataFrame:
    """Normalize a raw frame to open/high/low/close, ET index, RTH."""
    if isinstance(raw.columns, pd.MultiIndex):
        # yfinance may return MultiIndex columns even with a single ticker.
        raw.columns = raw.columns.get_level_values(0)
    raw = raw.rename(columns=str.lower)
    needed = ["open", "high", "low", "close"]
    if not all(c in raw.columns for c in needed):
        return pd.DataFrame()
    df = raw[needed].copy().dropna()
    if df.empty:
        return df

    if interval == "1d":
        # For daily data there is no intraday session to filter.
        if df.index.tz is None:
            df = df.tz_localize(ET_TZ)
        else:
            df = df.tz_convert(ET_TZ)
        return df
    return filter_rth(df)


def download_data(
    ticker: str, interval: str, start, end,
    providers: list[str] | None = None, registry: dict | None = None,
) -> pd.DataFrame:
    """Download the data by trying providers in priority order, with fallback.

    Tries each provider in the given order: it stops at the first one that
    returns usable data; if a provider fails (error) or has no data, it moves
    straight to the next. Isolated from the engine for testability.

    Args:
        ticker: symbol (e.g. "SPY").
        interval: bar interval ("60m", "30m", "15m", "1d").
        start, end: period bounds (dates or strings).
        providers: provider names in priority order. Default: all those in
            ``registry`` in definition order.
        registry: mapping name -> provider function. Default :data:`PROVIDERS`
            (injectable in tests to avoid the network).

    Returns:
        open/high/low/close DataFrame, ET index, filtered to RTH for intraday.
        ``df.attrs['provider']`` reports the provider that served the data.

    Raises:
        ValueError: if no provider returns usable data.
    """
    reg = registry if registry is not None else PROVIDERS
    ordine = providers or list(reg.keys())

    falliti: list[str] = []
    for nome in ordine:
        fn = reg.get(nome)
        if fn is None:
            continue
        try:
            raw = fn(ticker, interval, start, end)
        except Exception:
            falliti.append(nome)
            continue
        if raw is None or raw.empty:
            falliti.append(nome)
            continue
        df = _normalizza_ohlc(raw, interval)
        if df.empty:
            falliti.append(nome)
            continue
        df.attrs["provider"] = nome
        return df

    raise ValueError(
        f"No data for {ticker} ({interval}) from any provider "
        f"({', '.join(ordine)}). Likely a non-existent ticker or a period "
        "outside the available window (intraday: 1m ~7d, 15-30m ~60d, hourly "
        "~730d). For long histories use 1d; Stooq covers daily data only."
    )