ParallelLLC/algorithmic_trading

1"""Monte-Carlo permutation test for trading rules.2 3The question this answers is not "did the strategy make money" but "would a4rule of this shape have made this much money on a market with no exploitable5structure?". We destroy the serial dependence in the price path while keeping6its distribution of moves intact, re-run the *same* strategy on each shuffled7market, and see where the real result lands in that null distribution.8 9A strategy whose Sharpe sits comfortably inside the null is not a strategy —10it is a lottery ticket that happened to win.11"""12 13from __future__ import annotations14 15from dataclasses import dataclass16from typing import Callable, Optional17 18import numpy as np19import pandas as pd20 21from ..engine import bars_to_returns, run_backtest22from ..types import CostModel23 24__all__ = ["permutation_test", "PermutationResult", "permute_bars"]25 26 27@dataclass28class PermutationResult:29    observed: float30    null: np.ndarray31    p_value: float32    method: str33    n_permutations: int34 35    @property36    def null_mean(self) -> float:37        return float(np.mean(self.null)) if self.null.size else 0.038 39    @property40    def percentile(self) -> float:41        """Where the observed Sharpe sits in the null distribution, 0-100."""42        if not self.null.size:43            return 50.044        return float((self.null < self.observed).mean() * 100.0)45 46 47def _decompose(df: pd.DataFrame) -> tuple[np.ndarray, float]:48    """Split bars into scale-free log moves that can be reshuffled safely."""49    open_ = df["open"].to_numpy(dtype=float)50    high = df["high"].to_numpy(dtype=float)51    low = df["low"].to_numpy(dtype=float)52    close = df["close"].to_numpy(dtype=float)53    volume = df["volume"].to_numpy(dtype=float)54 55    gap = np.log(open_[1:] / close[:-1])56    hi = np.log(np.maximum(high[1:], open_[1:]) / open_[1:])57    lo = np.log(np.minimum(low[1:], open_[1:]) / open_[1:])58    body = np.log(close[1:] / open_[1:])59    return np.column_stack([gap, hi, lo, body, volume[1:]]), float(close[0])60 61 62def _rebuild(parts: np.ndarray, anchor: float, index: pd.Index, first_row: pd.Series) -> pd.DataFrame:63    gap, hi, lo, body, volume = (parts[:, i] for i in range(5))64    n = parts.shape[0] + 165 66    close = np.empty(n)67    open_ = np.empty(n)68    high = np.empty(n)69    low = np.empty(n)70    vol = np.empty(n)71 72    close[0] = anchor73    open_[0] = float(first_row["open"])74    high[0] = float(first_row["high"])75    low[0] = float(first_row["low"])76    vol[0] = float(first_row["volume"])77 78    # Cumulative product form: close[i] = close[0] * exp(cumsum(gap + body)).79    close[1:] = anchor * np.exp(np.cumsum(gap + body))80    open_[1:] = close[:-1] * np.exp(gap)81    high[1:] = open_[1:] * np.exp(hi)82    low[1:] = open_[1:] * np.exp(lo)83    vol[1:] = volume84 85    return pd.DataFrame(86        {"open": open_, "high": high, "low": low, "close": close, "volume": vol}, index=index87    )88 89 90def permute_bars(91    df: pd.DataFrame,92    rng: np.random.Generator,93    method: str = "permute",94    block: int = 20,95) -> pd.DataFrame:96    """Return a shuffled market with the same index and bar anatomy.97 98    ``permute`` reshuffles individual bars, destroying all serial structure.99    ``block`` resamples contiguous blocks with replacement, which preserves100    short-horizon autocorrelation and volatility clustering — a harder null101    that trend strategies deserve to be tested against.102    """103    parts, anchor = _decompose(df)104    m = parts.shape[0]105    if m < 2:106        return df.copy()107 108    if method == "block":109        size = max(2, min(int(block), m))110        starts = rng.integers(0, m, size=int(np.ceil(m / size)))111        order = np.concatenate([(np.arange(s, s + size) % m) for s in starts])[:m]112    else:113        order = rng.permutation(m)114 115    return _rebuild(parts[order], anchor, df.index, df.iloc[0])116 117 118def permutation_test(119    df: pd.DataFrame,120    signal_fn: Callable[[pd.DataFrame], pd.Series],121    n_permutations: int = 300,122    method: str = "permute",123    block: int = 20,124    costs: Optional[CostModel] = None,125    lag: int = 1,126    max_leverage: float = 1.0,127    allow_short: bool = True,128    seed: int = 0,129    observed: Optional[float] = None,130    progress: Optional[Callable[[float, str], None]] = None,131) -> PermutationResult:132    """Run ``signal_fn`` against ``n_permutations`` shuffled markets.133 134    ``signal_fn`` must be the strategy's target-exposure generator; it is135    re-evaluated on every synthetic market, which is the whole point — a rule136    that only works because of the specific path it was tuned on will fall137    apart here.138    """139    costs = costs or CostModel()140    rng = np.random.default_rng(seed)141 142    def sharpe_on(frame: pd.DataFrame) -> float:143        target = signal_fn(frame)144        result = run_backtest(145            frame,146            target,147            costs=costs,148            lag=lag,149            max_leverage=max_leverage,150            allow_short=allow_short,151        )152        return result.sharpe153 154    if observed is None:155        observed = sharpe_on(df)156 157    null = np.empty(n_permutations, dtype=float)158    for i in range(n_permutations):159        null[i] = sharpe_on(permute_bars(df, rng, method, block))160        if progress is not None and (i % 25 == 0 or i == n_permutations - 1):161            progress((i + 1) / n_permutations, f"Permutation {i + 1}/{n_permutations}")162 163    # +1 in both places: the observed result is itself one draw from the null164    # under H0, which keeps the test from ever reporting an impossible p = 0.165    p_value = float((1 + np.sum(null >= observed)) / (n_permutations + 1))166 167    return PermutationResult(168        observed=float(observed),169        null=null,170        p_value=p_value,171        method=method,172        n_permutations=n_permutations,173    )174 175 176def bootstrap_return_paths(returns: pd.Series, n: int = 500, seed: int = 0) -> np.ndarray:177    """Bootstrap terminal-wealth outcomes from a realised return stream.178 179    Useful for the "how wide is the cone of outcomes?" chart — the same edge180    can produce wildly different equity curves.181    """182    arr = np.asarray(returns.dropna(), dtype=float)183    if arr.size == 0:184        return np.zeros((n, 1))185    rng = np.random.default_rng(seed)186    draws = rng.choice(arr, size=(n, arr.size), replace=True)187    return np.cumprod(1.0 + draws, axis=1)188