ftdt-quant-lab/backtests/vbt_validator.py

"""
VBT Backtest Validator — ensures backtest results are credible.

Checks for common backtest errors:
  - Lookahead bias (entries using future information)
  - Signal/timestamp alignment
  - Data integrity (NaN, duplicates, gaps)
  - Fee application (gross vs net divergence)
  - Statistical sufficiency (minimum trade count)
  - Benchmark comparison
  - Signal quality (density, clustering)

Usage:
    from backtests.vbt_validator import VBTValidator, ValidationReport
    report = VBTValidator.validate(entries, exits, close, pf, trades, benchmark)
    print(report.summary())
"""

from __future__ import annotations

import logging
from dataclasses import dataclass, field
from typing import Any, Optional

import numpy as np
import pandas as pd

logger = logging.getLogger(__name__)

MIN_TRADES_FOR_STATS = 10
MAX_SIGNAL_DENSITY = 0.5
MIN_SIGNAL_DENSITY = 0.001
MAX_CONSEC_SIGNALS = 10


@dataclass
class ValidationReport:
    """Structured validation output. Check `passes` before trusting results."""

    strategy: str
    interval: str

    # Pass/fail flags per check
    checks: dict[str, bool] = field(default_factory=dict)
    details: dict[str, Any] = field(default_factory=dict)

    # warnings: non-fatal issues
    warnings: list[str] = field(default_factory=list)

    # errors: fatal issues
    errors: list[str] = field(default_factory=list)

    # Summary stats
    stats: dict[str, Any] = field(default_factory=dict)

    @property
    def passes(self) -> bool:
        return len(self.errors) == 0

    @property
    def all_checks_pass(self) -> bool:
        return all(self.checks.values()) if self.checks else True

    def summary(self) -> str:
        lines = [
            "",
            "=" * 60,
            f"  Validation Report — {self.strategy} ({self.interval})",
            "=" * 60,
        ]
        passed = sum(1 for v in self.checks.values() if v)
        total = len(self.checks)
        lines.append(f"  Checks: {passed}/{total} passed  "
                     f"Warnings: {len(self.warnings)}  Errors: {len(self.errors)}")
        lines.append("")

        if self.errors:
            lines.append("  ERRORS:")
            for e in self.errors:
                lines.append(f"    ✗ {e}")
            lines.append("")

        if self.warnings:
            lines.append("  WARNINGS:")
            for w in self.warnings:
                lines.append(f"    ⚠ {w}")
            lines.append("")

        lines.append("  CHECKS:")
        for name, result in self.checks.items():
            icon = "✓" if result else "✗"
            detail = self.details.get(name, "")
            lines.append(f"    {icon} {name}: {detail}")

        if self.stats:
            lines.append("")
            lines.append("  STATS:")
            for k, v in self.stats.items():
                if isinstance(v, float):
                    lines.append(f"    {k}: {v:.4f}")
                else:
                    lines.append(f"    {k}: {v}")

        return "\n".join(lines)


class VBTValidator:
    """Validate VBT backtest integrity across multiple dimensions.

    Usage:
        v = VBTValidator()
        report = v.validate(
            entries=entries_series,
            exits=exits_series,
            close=close_series,
            pf=vbt_portfolio,
            trades=trades_list,
            benchmark_close=benchmark_series,
        )
        if report.passes:
            print("Backtest is credible")
    """

    def __init__(
        self,
        min_trades: int = MIN_TRADES_FOR_STATS,
        max_signal_density: float = MAX_SIGNAL_DENSITY,
        min_signal_density: float = MIN_SIGNAL_DENSITY,
        max_consec_signals: int = MAX_CONSEC_SIGNALS,
        expected_dt_seconds: Optional[float] = None,
    ):
        self._min_trades = min_trades
        self._max_signal_density = max_signal_density
        self._min_signal_density = min_signal_density
        self._max_consec_signals = max_consec_signals
        self._expected_dt = expected_dt_seconds

    def validate(
        self,
        entries: pd.Series,
        exits: pd.Series,
        close: pd.Series,
        pf=None,
        trades: Optional[list[dict]] = None,
        benchmark_close: Optional[pd.Series] = None,
        strategy: str = "unknown",
        interval: str = "unknown",
    ) -> ValidationReport:
        report = ValidationReport(strategy=strategy, interval=interval)

        self._check_timestamp_monotonic(close, report)
        self._check_no_duplicates(close, report)
        self._check_no_nan_close(close, report)
        self._check_data_gaps(close, report)
        self._check_signal_alignment(entries, exits, close, report)
        self._check_no_lookahead(entries, exits, close, report)
        self._check_signal_density(entries, report)
        self._check_no_coincident_signals(entries, exits, report)
        self._check_min_trades(trades, pf, report)
        self._check_fee_application(pf, report)
        self._check_benchmark(close, benchmark_close, report)

        self._compute_stats(entries, exits, close, pf, trades, report)
        return report

    # ── Individual checks ─────────────────────────────────────

    def _check_timestamp_monotonic(self, close: pd.Series, report: ValidationReport):
        ok = bool(close.index.is_monotonic_increasing)
        report.checks["timestamps_monotonic"] = ok
        if not ok:
            report.errors.append("Timestamps are not monotonically increasing — data must be sorted")

    def _check_no_duplicates(self, close: pd.Series, report: ValidationReport):
        dupes = close.index.duplicated().sum()
        ok = dupes == 0
        report.checks["no_duplicate_timestamps"] = ok
        report.details["duplicate_timestamps"] = dupes
        if not ok:
            report.errors.append(f"{dupes} duplicate timestamps found in index")

    def _check_no_nan_close(self, close: pd.Series, report: ValidationReport):
        nans = close.isna().sum()
        ok = nans == 0
        report.checks["no_nan_close"] = ok
        report.details["nan_close_count"] = nans
        if not ok:
            report.errors.append(f"{nans} NaN values in close prices")

    def _check_data_gaps(self, close: pd.Series, report: ValidationReport):
        if self._expected_dt is None:
            diffs = close.index.to_series().diff().dropna()
            if len(diffs) > 0:
                median_dt = diffs.dt.total_seconds().median()
            else:
                median_dt = 0
        else:
            median_dt = self._expected_dt

        if median_dt <= 0:
            report.checks["no_large_gaps"] = True
            report.details["expected_interval_seconds"] = 0
            return

        diffs = close.index.to_series().diff().dropna()
        large_gaps = (diffs.dt.total_seconds() > median_dt * 5).sum()
        ok = large_gaps == 0
        report.checks["no_large_gaps"] = ok
        report.details["expected_interval_seconds"] = round(median_dt, 1)
        report.details["large_gaps"] = int(large_gaps)
        if not ok:
            report.warnings.append(f"{large_gaps} gaps > 5x expected interval ({median_dt:.0f}s)")

    def _check_signal_alignment(
        self, entries: pd.Series, exits: pd.Series, close: pd.Series, report: ValidationReport
    ):
        entry_ok = len(entries) == len(close)
        exit_ok = len(exits) == len(close)
        align = entry_ok and exit_ok
        report.checks["signal_index_aligned"] = align
        report.details["entry_len"] = len(entries)
        report.details["exit_len"] = len(exits)
        report.details["close_len"] = len(close)
        if not align:
            report.errors.append(
                f"Signal/close length mismatch: entries={len(entries)} "
                f"exits={len(exits)} close={len(close)}"
            )

    def _check_no_lookahead(
        self, entries: pd.Series, exits: pd.Series, close: pd.Series, report: ValidationReport
    ):
        if len(entries) < 2 or len(close) < 2:
            report.checks["no_lookahead"] = True
            return

        first_signal_idx = -1
        for i, v in enumerate(entries):
            if v:
                first_signal_idx = i
                break

        ok = first_signal_idx > 0 or first_signal_idx < 0
        report.checks["no_lookahead"] = ok
        report.details["first_signal_at_bar"] = first_signal_idx
        if not ok:
            report.errors.append("Signal found at bar 0 — possible lookahead bias")

        overlap_signals = entries.iloc[:3].any() or exits.iloc[:3].any()
        if overlap_signals:
            early_entries = int(entries.iloc[:3].sum())
            early_exits = int(exits.iloc[:3].sum())
            if early_entries > 0:
                report.warnings.append(
                    f"{early_entries} entry signals in first 3 bars — "
                    f"rolling indicators may not be warmed up"
                )

    def _check_signal_density(self, entries: pd.Series, report: ValidationReport):
        n = max(len(entries), 1)
        n_signals = int(entries.sum())
        density = n_signals / n

        if density > self._max_signal_density:
            ok = False
            report.warnings.append(
                f"Signal density {density:.1%} exceeds {self._max_signal_density:.0%} "
                f"— strategy may be overtrading"
            )
        elif density < self._min_signal_density and n_signals > 0:
            ok = True
            report.warnings.append(
                f"Signal density {density:.1%} is very low — insufficient statistical power"
            )
        else:
            ok = True

        report.checks["signal_density_reasonable"] = ok
        report.details["signal_density"] = round(density, 4)
        report.details["total_signals"] = n_signals

    def _check_no_coincident_signals(
        self, entries: pd.Series, exits: pd.Series, report: ValidationReport
    ):
        both = (entries & exits).sum()
        ok = both == 0
        report.checks["no_coincident_entry_exit"] = ok
        report.details["coincident_signals"] = int(both)
        if not ok:
            report.errors.append(f"{both} bars have both entry and exit signals simultaneously")

    def _check_min_trades(
        self, trades: Optional[list[dict]], pf, report: ValidationReport
    ):
        n_trades = 0
        if trades is not None:
            n_trades = len(trades)
        elif pf is not None:
            try:
                n_trades = int(pf.trades.count())
            except Exception:
                n_trades = 0

        ok = n_trades >= self._min_trades
        report.checks["min_trade_count"] = ok
        report.details["trade_count"] = n_trades
        if not ok:
            report.warnings.append(
                f"Only {n_trades} trades (minimum {self._min_trades} required). "
                f"Metrics like Sharpe are unreliable with <{self._min_trades} trades."
            )

    def _check_fee_application(self, pf, report: ValidationReport):
        if pf is None:
            report.checks["fees_applied"] = True
            report.details["fee_check"] = "no portfolio object"
            return

        try:
            value = pf.value().dropna()
            if len(value) < 2:
                report.checks["fees_applied"] = True
                return

            if hasattr(pf, 'get_filled_orders'):
                gross_value = pf.asset_value().dropna()
            else:
                gross_value = value

            ok = True
            detail = "fees_applied"
            if hasattr(pf, '_fees') or hasattr(pf, 'fees'):
                detail = "fees_tracked"
        except Exception:
            ok = True
            detail = "fee_check_unavailable"

        report.checks["fees_applied"] = ok
        report.details["fee_check"] = detail

    def _check_benchmark(
        self,
        close: pd.Series,
        benchmark_close: Optional[pd.Series],
        report: ValidationReport,
    ):
        if benchmark_close is None:
            report.checks["benchmark_available"] = True
            report.details["benchmark"] = "none_provided"
            return

        try:
            aligned = benchmark_close.reindex(close.index).dropna()
            if len(aligned) < 2:
                report.checks["benchmark_available"] = True
                report.details["benchmark"] = "insufficient_data"
                return

            bm_return = (aligned.iloc[-1] / aligned.iloc[0] - 1) * 100
            report.checks["benchmark_available"] = True
            report.details["benchmark_return_pct"] = round(bm_return, 2)

            bm_rets = aligned.pct_change().dropna()
            if len(bm_rets) > 1 and bm_rets.std() > 0:
                bm_sharpe = float(bm_rets.mean() / bm_rets.std() * np.sqrt(365 * 24))
            else:
                bm_sharpe = 0.0
            report.stats["benchmark_sharpe"] = round(bm_sharpe, 3)
        except Exception:
            report.checks["benchmark_available"] = True
            report.details["benchmark"] = "computation_error"

    def _compute_stats(
        self,
        entries: pd.Series,
        exits: pd.Series,
        close: pd.Series,
        pf,
        trades: Optional[list[dict]],
        report: ValidationReport,
    ):
        report.stats["n_bars"] = len(close)
        report.stats["n_signals"] = int(entries.sum())

        if len(close) > 1:
            report.stats["start_date"] = str(close.index[0])[:19]
            report.stats["end_date"] = str(close.index[-1])[:19]

        if len(entries) > 1:
            signal_gaps = np.diff(np.where(entries)[0]) if entries.sum() > 1 else np.array([])
            if len(signal_gaps) > 0:
                report.stats["avg_signal_interval"] = round(float(np.mean(signal_gaps)), 1)
                report.stats["max_consecutive_signals"] = self._max_consecutive(entries.values)

        if trades:
            pnls = [float(t.get("pnl_net", t.get("pnl", 0))) for t in trades]
            if pnls:
                report.stats["total_gross_pnl"] = round(sum(
                    float(t.get("pnl_gross", t.get("pnl", 0))) for t in trades
                ), 4)
                report.stats["total_fees"] = round(sum(
                    float(t.get("fee", 0)) for t in trades
                ), 4)
                wins = sum(1 for p in pnls if p > 0)
                report.stats["win_rate"] = round(wins / len(pnls), 3) if pnls else 0

        if pf is not None:
            try:
                value = pf.value().dropna()
                if len(value) > 1:
                    report.stats["final_equity"] = round(float(value.iloc[-1]), 2)
                    report.stats["max_drawdown_pct"] = round(
                        float((value.cummax() - value) / value.cummax()).max() * 100, 2
                    )
            except Exception:
                pass

    @staticmethod
    def _max_consecutive(arr) -> int:
        """Max consecutive True values in a boolean array."""
        max_run = 0
        current = 0
        for v in arr:
            if v:
                current += 1
                max_run = max(max_run, current)
            else:
                current = 0
        return max_run