test_fuzz_engine.py

# -*- coding: utf-8 -*-
"""Property fuzzer: throw hundreds of randomized markets and strategies at the engine and
assert the invariants that must hold no matter what. Deterministic (seeded), fully offline.

Invariants:
  I1  returns are finite (no NaN/inf), and never <= -100% in a day
  I2  weights stay in [0, 1]; row sums <= 1 (== 1 when a cash symbol absorbs the residual)
  I3  gross exposure stays in [0, 1]
  I4  turnover is non-negative and no rebalance exceeds 2.0 (full sell + full buy)
  I5  costs only ever hurt: the zero-cost run cumulates >= the with-cost run
  I6  any honest strategy (one that only reads view.prices) passes the look-ahead audit
"""
import hashlib

import numpy as np
import pandas as pd
import pytest

from backtest_lab.audit import audit_no_lookahead
from backtest_lab.engine import run

N_SCENARIOS = 40          # scenarios; each runs twice (cost / no-cost) -> plenty of coverage
DAYS = 700


def synth_market(rng):
    """Random market: 2-5 assets, GBM with random drift/vol, occasional crash days and gaps."""
    n_assets = int(rng.integers(2, 6))
    cols = [f"A{i}" for i in range(n_assets)] + ["CASH"]
    data = {}
    for c in cols[:-1]:
        mu = rng.uniform(-0.05, 0.15) / 252
        sig = rng.uniform(0.08, 0.45) / np.sqrt(252)
        r = rng.normal(mu, sig, DAYS)
        crashes = rng.random(DAYS) < 0.002
        r[crashes] -= rng.uniform(0.05, 0.20, crashes.sum())
        data[c] = 50 * np.cumprod(1 + r)
    data["CASH"] = 100 * np.cumprod(1 + np.full(DAYS, 0.02 / 252))
    idx = pd.bdate_range("2018-01-01", periods=DAYS)
    return pd.DataFrame(data, index=idx)


def random_honest_strategy(seed: int, columns):
    """A deterministic-but-arbitrary strategy: weights depend only on the view's PAST data.

    Determinism matters for invariant I5 (two runs must make identical decisions), so the
    randomness is derived from a hash of the visible history, not from global state.
    """
    risk = [c for c in columns if c != "CASH"]

    def strategy(view):
        px = view.prices
        if len(px) < 30:
            return {}
        h = int(hashlib.md5(f"{seed}-{view.date}".encode()).hexdigest()[:8], 16)
        local = np.random.default_rng(h)
        if local.random() < 0.15:
            return {}                                    # sometimes: all cash
        k = int(local.integers(1, len(risk) + 1))
        picks = list(local.choice(risk, size=k, replace=False))
        raw = local.random(k)
        gross = local.uniform(0.2, 1.0)
        w = raw / raw.sum() * gross
        return {s: float(v) for s, v in zip(picks, w)}

    return strategy


@pytest.mark.parametrize("scenario", range(N_SCENARIOS))
def test_engine_invariants_hold_under_fuzz(scenario):
    rng = np.random.default_rng(1000 + scenario)
    prices = synth_market(rng)
    strategy = random_honest_strategy(scenario, prices.columns)
    kw = dict(start=str(prices.index[60].date()), cash="CASH")

    paid = run(prices, strategy, cost_per_side=0.0020, **kw)
    free = run(prices, strategy, cost_per_side=0.0, **kw)

    # I1 — sane returns
    assert np.isfinite(paid.returns).all(), "non-finite return"
    assert (paid.returns > -1.0).all(), "a daily return wiped out more than 100%"
    # I2 — weight bounds
    assert (paid.weights >= -1e-12).all().all()
    assert (paid.weights <= 1.0 + 1e-9).all().all()
    assert np.allclose(paid.weights.sum(axis=1), 1.0), "cash residual broken"
    # I3 — gross bounds
    assert (paid.gross >= -1e-12).all() and (paid.gross <= 1.0 + 1e-9).all()
    # I4 — turnover bounds
    assert (paid.turnover >= -1e-12).all()
    assert float(paid.turnover.max()) <= 2.0 + 1e-9, "single rebalance turned over more than 200%"
    # I5 — costs only hurt
    assert float((1 + free.returns).prod()) >= float((1 + paid.returns).prod()) - 1e-12


@pytest.mark.parametrize("scenario", range(0, N_SCENARIOS, 8))
def test_random_honest_strategies_pass_lookahead_audit(scenario):
    rng = np.random.default_rng(2000 + scenario)
    prices = synth_market(rng)
    strategy = random_honest_strategy(scenario, prices.columns)
    result = audit_no_lookahead(prices, strategy, start=str(prices.index[60].date()),
                                n_dates=3, seed=scenario)
    assert result["passed"], result