audit.py

# -*- coding: utf-8 -*-
"""Look-ahead audit — the guardrail that catches strategies peeking at the future.

Look-ahead bias is the most common way a backtest lies: the strategy accidentally uses
information from on/after the decision date (today's close, a centered rolling window, a
scaler fit on the full sample), and the resulting curve is beautiful and meaningless.

The audit is brutal and simple: for a sample of rebalance dates, corrupt every price ON and
AFTER the date with random noise and re-ask the strategy for its weights. A strategy that only
uses the past returns identical weights; any change proves it read the future.
"""
from __future__ import annotations

import numpy as np
import pandas as pd

from .engine import DataView, rebalance_dates


def audit_no_lookahead(prices: pd.DataFrame, strategy, start: str = "2007-01-01",
                       n_dates: int = 5, seed: int = 0, noise: float = 0.10) -> dict:
    """Perturb the future, re-run the strategy, and compare weights date by date.

    Returns {"passed": bool, "checked": [dates], "failures": [{date, before, after}]}.
    """
    rng = np.random.default_rng(seed)
    rebal = rebalance_dates(prices.index, start)
    if len(rebal) < 2:
        raise ValueError("not enough rebalance dates to audit")
    # Sample dates away from the very first rebalance so strategies have warm-up history.
    candidates = rebal[max(1, len(rebal) // 10):]
    picks = sorted(rng.choice(len(candidates), size=min(n_dates, len(candidates)),
                              replace=False))
    checked, failures = [], []
    n_nonempty = 0                          # dates on which the strategy made a real (non-flat) decision
    for i in picks:
        d = candidates[i]
        w_clean = strategy(DataView(prices, d)) or {}
        if w_clean:
            n_nonempty += 1
        corrupted = prices.copy()
        future = corrupted.index >= d
        shock = 1.0 + rng.normal(0.0, noise, size=(int(future.sum()), corrupted.shape[1]))
        corrupted.loc[future] = corrupted.loc[future].to_numpy() * np.abs(shock)
        w_dirty = strategy(DataView(corrupted, d)) or {}
        checked.append(str(pd.Timestamp(d).date()))
        same = (set(w_clean) == set(w_dirty)
                and all(abs(w_clean[k] - w_dirty[k]) < 1e-12 for k in w_clean))
        if not same:
            failures.append({"date": str(pd.Timestamp(d).date()),
                             "weights_clean": w_clean, "weights_corrupted": w_dirty})
    # A pass is only meaningful if the strategy actually made a decision on at least one sampled
    # date — otherwise `{} == {}` on every warm-up/all-cash date is a vacuous "pass" that never
    # exercised the logic. Report it honestly and fail closed.
    vacuous = n_nonempty == 0
    passed = len(failures) == 0 and not vacuous
    if vacuous:
        verdict = ("INCONCLUSIVE — the strategy returned no positions on any sampled date; "
                   "the audit never exercised a real decision (widen the window or n_dates)")
    elif failures:
        verdict = f"LOOK-AHEAD DETECTED on {len(failures)} of {len(checked)} dates"
    else:
        verdict = f"no look-ahead detected ({n_nonempty} of {len(checked)} sampled dates made a real decision)"
    return {
        "passed": passed,
        "checked": checked,
        "n_nonempty_checked": n_nonempty,
        "failures": failures,
        "verdict": verdict,
    }