994bb59641
- pytest-xdist added to dev deps; addopts='-n auto' runs the ~1940-test suite across all cores (~6-9min -> ~1.5min); registered a 'slow' marker for a serial '-m "not slow"' loop. - equity_backtest_runner liquidity fixture: trimmed the 3000-bar (8yr) LIVELONG/ILLIQUID histories to 800 bars (still >> min_history=300, multi-year for peak-yearly ranking) — the peak-yearly aggregation over 8yr was the cost: 50s/30s/19s tests -> ~5.8s each, assertions unchanged. - paper_sim perf guards: made them parallel-robust (the wall-clock <1.6s bound flaked under -n auto CPU contention). full-history guard relaxed to <10s (it targets an O(D²) blowup = orders of magnitude, not a constant factor); cheap-reapply made RELATIVE (< full/10, ratio is contention-invariant). Both marked slow. Full suite 1940 passed. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
477 lines
24 KiB
Python
477 lines
24 KiB
Python
"""Unit tests for the pure `simulate_book` engine (paper_sim.py).
|
||
|
||
The load-bearing guarantee: with `controller="killswitch", kelly_fraction=1.0, cost_bps=0` the engine
|
||
reproduces the ESTABLISHED LIVE book math — the per-date overlay path the live backfill/snapshot use
|
||
(`paper_risk_overlay`/`IncrementalRiskOverlay` called on the strictly-prior window, eff_lev=0 on kill,
|
||
applied to that date) — so the sim never diverges from the live/replay book. The other tests cover
|
||
capital scaling, the date window, sleeve selection, the cost_bps turnover model, and hand-checked
|
||
metrics.
|
||
|
||
NOTE on the golden reference: the spec phrases the golden as "equal the existing `paper_book_returns`
|
||
(+killswitch)". In the actual code, `paper_book_returns` is the INTERNAL un-throttled regime signal
|
||
that feeds the killswitch — it is NOT what the live book compounds. The live book
|
||
(paper_backfill.py / paper_snapshot.py) compounds the OVERLAY's `(weights, eff_lev)` applied to each
|
||
date's sleeve returns. Those two differ at the vol-target warmup boundary (the overlay's leverage is
|
||
computed from the full trailing window via paper_risk_overlay, whereas paper_book_returns sizes off
|
||
its own incrementally-built throttled series). The engine is built against — and golden-tested
|
||
against — the live OVERLAY path, since that is the book the cockpit + live wiring must reproduce. See
|
||
the report.
|
||
"""
|
||
import json
|
||
import math
|
||
import pathlib
|
||
import random
|
||
import statistics as st
|
||
import time
|
||
|
||
import pytest
|
||
|
||
from fxhnt.application.paper_book import paper_risk_overlay
|
||
from fxhnt.application.paper_sim import (
|
||
apply_knobs,
|
||
overlay_pass,
|
||
simulate_book,
|
||
simulate_naive_eqwt,
|
||
)
|
||
|
||
_GOLDEN_PATH = pathlib.Path(__file__).parent / "data" / "golden_sim.json"
|
||
|
||
|
||
def _big_fixture() -> dict[str, dict[int, float]]:
|
||
"""A full-history-shaped fixture (~1473 days, 3 staggered sleeves) for the perf benchmark — mirrors
|
||
the production combined-book history that the interactive /paper/sim page recomputes per knob change."""
|
||
rng = random.Random(42)
|
||
return {
|
||
"crypto_tstrend": {1 + i: rng.gauss(0.0012, 0.012) for i in range(1473)},
|
||
"unlock": {30 + i: rng.gauss(-0.001, 0.022) for i in range(1400)},
|
||
"stablecoin_rotation": {10 + i: rng.gauss(0.0004, 0.004) for i in range(1450)},
|
||
}
|
||
|
||
# --- fixtures ------------------------------------------------------------------------------------
|
||
|
||
def _two_edge_fixture() -> dict[str, dict[int, float]]:
|
||
"""A 2-edge-shaped fixture (tstrend + unlock) with different vols + staggered starts + a
|
||
negative-drift sleeve so the killswitch actually fires (exercises the killed path). Deterministic."""
|
||
rng = random.Random(20260623)
|
||
return {
|
||
"crypto_tstrend": {1 + i: rng.gauss(0.0012, 0.012) for i in range(220)},
|
||
# Strong negative drift → the un-throttled book series breaches the −25% kill threshold.
|
||
"unlock": {16 + i: rng.gauss(-0.010, 0.022) for i in range(200)},
|
||
}
|
||
|
||
|
||
def _three_edge_fixture() -> dict[str, dict[int, float]]:
|
||
fx = _two_edge_fixture()
|
||
rng = random.Random(99)
|
||
fx["stablecoin_rotation"] = {6 + i: rng.gauss(0.0004, 0.004) for i in range(210)}
|
||
return fx
|
||
|
||
|
||
def _overlay_book_returns(returns_by_sleeve: dict[str, dict[int, float]]) -> dict[int, float]:
|
||
"""Faithful per-date THROTTLED book return via the LIVE overlay path (mirrors paper_backfill.py):
|
||
for each date d, call the stateless `paper_risk_overlay` on the strictly-prior window (keys < d),
|
||
take eff_lev = 0 if killed else lev, and book `eff_lev · Σ weights·return[d]`. This is exactly the
|
||
gross return the live book compounds → the ground-truth golden the engine must reproduce."""
|
||
all_days = sorted({d for r in returns_by_sleeve.values() for d in r})
|
||
out: dict[int, float] = {}
|
||
for d in all_days:
|
||
before = {s: {k: v for k, v in r.items() if k < d} for s, r in returns_by_sleeve.items()}
|
||
active = [s for s, r in before.items() if r]
|
||
w, lev, killed = paper_risk_overlay(before, active)
|
||
eff = 0.0 if killed else lev
|
||
out[d] = eff * sum(w.get(s, 0.0) * returns_by_sleeve.get(s, {}).get(d, 0.0) for s in w)
|
||
return out
|
||
|
||
|
||
# --- golden vs the established book math ----------------------------------------------------------
|
||
|
||
def test_golden_matches_overlay_book_returns_with_killswitch():
|
||
fx = _two_edge_fixture()
|
||
res = simulate_book(fx, capital=100_000.0, sleeves=["crypto_tstrend", "unlock"],
|
||
controller="killswitch", kelly_fraction=1.0, cost_bps=0.0)
|
||
|
||
ref = _overlay_book_returns(fx) # the live overlay path's per-date throttled book return
|
||
# The sim emits an entry for every union day (the first one is flat: overlay on empty → no weights).
|
||
assert res.dates == sorted(ref)
|
||
|
||
# Reconstruct the sim's per-day net return from its compounding equity curve and compare to ref.
|
||
eq = res.equity
|
||
dates = res.dates
|
||
cap = 100_000.0
|
||
for i, d in enumerate(dates):
|
||
prev = cap if i == 0 else eq[i - 1]
|
||
sim_ret = eq[i] / prev - 1.0
|
||
assert math.isclose(sim_ret, ref[d], rel_tol=0, abs_tol=1e-12), \
|
||
f"day {d}: sim {sim_ret!r} != book {ref[d]!r}"
|
||
|
||
# And the equity compounding itself matches a direct compound of the reference book returns.
|
||
expected_eq = cap
|
||
for d in dates:
|
||
expected_eq *= (1.0 + ref[d])
|
||
assert math.isclose(res.equity[-1], expected_eq, rel_tol=0, abs_tol=1e-6)
|
||
|
||
|
||
def test_killswitch_actually_fires_in_fixture():
|
||
# Guard: the golden test is only meaningful if the kill path is exercised.
|
||
fx = _two_edge_fixture()
|
||
res = simulate_book(fx, capital=100_000.0, sleeves=["crypto_tstrend", "unlock"])
|
||
assert any(res.killed), "fixture never killed — golden test not exercising the killed path"
|
||
assert all((lev == 0.0) for lev, k in zip(res.leverage, res.killed, strict=True) if k), \
|
||
"effective leverage must be 0 on killed days"
|
||
|
||
|
||
# --- capital scaling --------------------------------------------------------------------------------
|
||
|
||
def test_capital_scales_curve_linearly():
|
||
fx = _two_edge_fixture()
|
||
sleeves = ["crypto_tstrend", "unlock"]
|
||
a = simulate_book(fx, capital=10_000.0, sleeves=sleeves)
|
||
b = simulate_book(fx, capital=40_000.0, sleeves=sleeves)
|
||
assert a.dates == b.dates
|
||
for ea, eb in zip(a.equity, b.equity, strict=True):
|
||
assert math.isclose(eb, ea * 4.0, rel_tol=1e-12, abs_tol=1e-6)
|
||
# Shape-invariant stats are identical.
|
||
assert math.isclose(a.metrics["sharpe"], b.metrics["sharpe"], rel_tol=0, abs_tol=1e-12)
|
||
assert math.isclose(a.metrics["max_dd"], b.metrics["max_dd"], rel_tol=0, abs_tol=1e-12)
|
||
assert math.isclose(a.metrics["cagr"], b.metrics["cagr"], rel_tol=0, abs_tol=1e-12)
|
||
|
||
|
||
# --- date window ------------------------------------------------------------------------------------
|
||
|
||
def test_date_window_restricts_curve():
|
||
fx = _two_edge_fixture()
|
||
sleeves = ["crypto_tstrend", "unlock"]
|
||
full = simulate_book(fx, capital=100_000.0, sleeves=sleeves)
|
||
lo, hi = 50, 120
|
||
win = simulate_book(fx, capital=100_000.0, sleeves=sleeves, start=lo, end=hi)
|
||
|
||
assert win.dates == [d for d in full.dates if lo <= d <= hi]
|
||
assert all(lo <= d <= hi for d in win.dates)
|
||
# Equity restarts at `capital` on the first in-window date (it compounds from capital, not from the
|
||
# full-curve level at that date) — but the overlay state is still the full prior history (causal).
|
||
first = win.dates[0]
|
||
first_net = win.equity[0] / 100_000.0 - 1.0
|
||
# The first in-window day applies the THROTTLED overlay book return of that day to `capital`.
|
||
ref = _overlay_book_returns(fx)
|
||
assert math.isclose(first_net, ref[first], rel_tol=0, abs_tol=1e-12)
|
||
|
||
|
||
# --- sleeve selection -------------------------------------------------------------------------------
|
||
|
||
def test_sleeve_selection_distinct_curves():
|
||
fx = _three_edge_fixture()
|
||
two = simulate_book(fx, capital=100_000.0, sleeves=["crypto_tstrend", "unlock"])
|
||
three = simulate_book(fx, capital=100_000.0,
|
||
sleeves=["crypto_tstrend", "unlock", "stablecoin_rotation"])
|
||
assert two.equity[-1] != three.equity[-1], "2-edge and 3-edge curves should differ"
|
||
|
||
|
||
def test_single_sleeve_is_that_sleeve_overlaid_return():
|
||
fx = _three_edge_fixture()
|
||
res = simulate_book(fx, capital=100_000.0, sleeves=["crypto_tstrend"])
|
||
# A single sleeve gets weight 1.0; its net return each day = eff_lev · 1.0 · r[d] (killswitch +
|
||
# vol-target leverage still apply). Reproduce via the same single-sleeve overlay book math.
|
||
ref = _overlay_book_returns({"crypto_tstrend": fx["crypto_tstrend"]})
|
||
cap = 100_000.0
|
||
assert res.dates == sorted(ref)
|
||
for i, d in enumerate(res.dates):
|
||
prev = cap if i == 0 else res.equity[i - 1]
|
||
sim_ret = res.equity[i] / prev - 1.0
|
||
assert math.isclose(sim_ret, ref[d], rel_tol=0, abs_tol=1e-12), f"day {d}"
|
||
# Single active sleeve → its weight is 1.0 on days it is active.
|
||
for w in res.weights:
|
||
if w:
|
||
assert math.isclose(sum(w.values()), 1.0, rel_tol=0, abs_tol=1e-9)
|
||
|
||
|
||
def test_intersection_with_unknown_sleeve():
|
||
fx = _two_edge_fixture()
|
||
a = simulate_book(fx, capital=100_000.0, sleeves=["crypto_tstrend", "unlock"])
|
||
# Adding a sleeve absent from the data must not change the result (intersection).
|
||
b = simulate_book(fx, capital=100_000.0, sleeves=["crypto_tstrend", "unlock", "does_not_exist"])
|
||
assert a.dates == b.dates
|
||
assert a.equity == b.equity
|
||
|
||
|
||
# --- cost_bps ---------------------------------------------------------------------------------------
|
||
|
||
def test_cost_bps_zero_is_identical_to_gross():
|
||
fx = _two_edge_fixture()
|
||
sleeves = ["crypto_tstrend", "unlock"]
|
||
gross = simulate_book(fx, capital=100_000.0, sleeves=sleeves, cost_bps=0.0)
|
||
# Bit-identical: cost_bps=0 must take the no-cost branch.
|
||
again = simulate_book(fx, capital=100_000.0, sleeves=sleeves, cost_bps=0.0)
|
||
assert gross.equity == again.equity
|
||
|
||
|
||
def test_cost_bps_lowers_curve_by_turnover_charge():
|
||
fx = _two_edge_fixture()
|
||
sleeves = ["crypto_tstrend", "unlock"]
|
||
gross = simulate_book(fx, capital=100_000.0, sleeves=sleeves, cost_bps=0.0)
|
||
costed = simulate_book(fx, capital=100_000.0, sleeves=sleeves, cost_bps=10.0)
|
||
# There IS turnover (weights/leverage move day to day), so the costed curve ends lower.
|
||
assert costed.equity[-1] < gross.equity[-1]
|
||
# Every day, the costed net return <= gross net return (cost only subtracts).
|
||
cap = 100_000.0
|
||
for i in range(len(gross.dates)):
|
||
gp = cap if i == 0 else gross.equity[i - 1]
|
||
cp = cap if i == 0 else costed.equity[i - 1]
|
||
g_ret = gross.equity[i] / gp - 1.0
|
||
c_ret = costed.equity[i] / cp - 1.0
|
||
assert c_ret <= g_ret + 1e-15
|
||
|
||
|
||
def test_cost_bps_exact_turnover_on_two_day_fixture():
|
||
"""Hand-checked turnover charge on a tiny single-sleeve fixture where the math is traceable.
|
||
|
||
Single sleeve, within warmup so leverage is 1.0 throughout. Day 1 is flat (overlay over empty prior
|
||
view → no weights; eff_w = {}). Day 2: the sleeve is active with weight 1.0, eff_lev 1.0, so
|
||
eff_w = {sleeve: 1.0}; turnover vs prev_eff ({}) = |1.0 − 0| = 1.0 → cost = 1.0 · 25/1e4 = 0.0025.
|
||
Day 3: eff_w unchanged at {sleeve: 1.0} → turnover 0 → no cost."""
|
||
fx = {"crypto_tstrend": {1: 0.01, 2: 0.02, 3: -0.01}}
|
||
gross = simulate_book(fx, capital=100_000.0, sleeves=["crypto_tstrend"], cost_bps=0.0)
|
||
costed = simulate_book(fx, capital=100_000.0, sleeves=["crypto_tstrend"], cost_bps=25.0)
|
||
assert gross.dates == [1, 2, 3]
|
||
cap = 100_000.0
|
||
g_rets, c_rets = [], []
|
||
for i in range(3):
|
||
gp = cap if i == 0 else gross.equity[i - 1]
|
||
cp = cap if i == 0 else costed.equity[i - 1]
|
||
g_rets.append(gross.equity[i] / gp - 1.0)
|
||
c_rets.append(costed.equity[i] / cp - 1.0)
|
||
# Day 1: flat, no turnover → no cost; identical.
|
||
assert math.isclose(c_rets[0], g_rets[0], abs_tol=1e-15)
|
||
# Day 2: exactly 0.0025 of turnover cost (turnover 1.0 × 25bp).
|
||
assert math.isclose(g_rets[1] - c_rets[1], 0.0025, rel_tol=0, abs_tol=1e-12)
|
||
# Day 3: eff weights unchanged → zero turnover → no incremental cost.
|
||
assert math.isclose(c_rets[2], g_rets[2], rel_tol=0, abs_tol=1e-12)
|
||
|
||
|
||
# --- metrics ----------------------------------------------------------------------------------------
|
||
|
||
def test_metrics_hand_checked_on_deterministic_fixture():
|
||
"""Single sleeve, no killswitch action, leverage stays in warmup (1.0) so the book return == the
|
||
raw sleeve return for each day → metrics are hand-computable from those returns.
|
||
|
||
Direct-application semantics (mirroring the live backfill overlay path): the weights applied to day
|
||
d come from the overlay over data strictly before d. The first union day is FLAT (overlay over an
|
||
empty prior view → no weights). From the second day on, the single sleeve carries weight 1.0."""
|
||
# 5 days of a single sleeve.
|
||
rets = {1: 0.05, 2: -0.02, 3: 0.03, 4: 0.01, 5: 0.04}
|
||
fx = {"crypto_tstrend": rets}
|
||
res = simulate_book(fx, capital=100_000.0, sleeves=["crypto_tstrend"])
|
||
|
||
assert res.dates == [1, 2, 3, 4, 5]
|
||
# day 1 flat (overlay over empty prior view); days 2-5 carry r[d] (weight 1.0, leverage 1.0 warmup).
|
||
net = [0.0, rets[2], rets[3], rets[4], rets[5]]
|
||
cap = 100_000.0
|
||
eq = cap
|
||
peak = cap
|
||
dds = []
|
||
for r in net:
|
||
eq *= (1 + r)
|
||
peak = max(peak, eq)
|
||
dds.append((eq - peak) / peak)
|
||
assert math.isclose(res.equity[-1], eq, rel_tol=0, abs_tol=1e-6)
|
||
assert math.isclose(res.metrics["end_value"], eq, rel_tol=0, abs_tol=1e-6)
|
||
assert math.isclose(res.metrics["total_return"], eq / cap - 1.0, rel_tol=0, abs_tol=1e-12)
|
||
|
||
# max_dd hand-checked.
|
||
assert math.isclose(res.metrics["max_dd"], min(dds), rel_tol=0, abs_tol=1e-12)
|
||
|
||
# CAGR over the window span n_days = dates[-1] - dates[0] = 5 - 1 = 4.
|
||
years = (5 - 1) / 365
|
||
expected_cagr = (eq / cap) ** (1.0 / years) - 1.0
|
||
assert math.isclose(res.metrics["cagr"], expected_cagr, rel_tol=0, abs_tol=1e-9)
|
||
|
||
# Sharpe over the net-return series.
|
||
sd = st.pstdev(net)
|
||
expected_sharpe = st.mean(net) / sd * math.sqrt(365)
|
||
assert math.isclose(res.metrics["sharpe"], expected_sharpe, rel_tol=0, abs_tol=1e-9)
|
||
|
||
# All days fall in 1970 (epoch days 1-5) → worst==best==total compounded return.
|
||
assert math.isclose(res.metrics["worst_year"], eq / cap - 1.0, rel_tol=0, abs_tol=1e-12)
|
||
assert math.isclose(res.metrics["best_year"], eq / cap - 1.0, rel_tol=0, abs_tol=1e-12)
|
||
|
||
|
||
def test_worst_and_best_year_split_across_calendar_years():
|
||
"""Epoch days spanning two calendar years → distinct worst/best year returns.
|
||
|
||
Direct-application semantics: a single sleeve carries weight 1.0 from its SECOND observation on
|
||
(day 1 is flat: overlay over empty prior view). Lay out obs in each year so real returns land in each."""
|
||
from fxhnt.application.paper_sim import _epoch_day_to_year
|
||
# epoch day 1 = 1970-01-02 ... day 366 = 1971-01-02.
|
||
fx = {"crypto_tstrend": {1: -0.05, 2: -0.05, 3: -0.05, # 1970 (day 1 flat; days 2,3 live)
|
||
366: 0.10, 367: 0.10}} # 1971 (days 366,367 live)
|
||
assert _epoch_day_to_year(3) == 1970
|
||
assert _epoch_day_to_year(366) == 1971
|
||
res = simulate_book(fx, capital=100_000.0, sleeves=["crypto_tstrend"])
|
||
# 1970 live returns: day1=0 (flat), day2=−0.05, day3=−0.05 → (1)(0.95)(0.95)−1 = −0.0975.
|
||
# 1971 live returns: day366=+0.10, day367=+0.10 → (1.10)(1.10)−1 = +0.21.
|
||
assert res.metrics["worst_year"] < 0.0 < res.metrics["best_year"]
|
||
assert math.isclose(res.metrics["worst_year"], 0.95 * 0.95 - 1.0, rel_tol=0, abs_tol=1e-9)
|
||
assert math.isclose(res.metrics["best_year"], 1.10 * 1.10 - 1.0, rel_tol=0, abs_tol=1e-9)
|
||
|
||
|
||
def test_empty_inputs_return_seeded_metrics():
|
||
res = simulate_book({}, capital=100_000.0, sleeves=["crypto_tstrend"])
|
||
assert res.dates == []
|
||
assert res.equity == []
|
||
assert res.metrics["end_value"] == 100_000.0
|
||
assert res.metrics["total_return"] == 0.0
|
||
assert res.metrics["max_dd"] == 0.0
|
||
|
||
|
||
# --- bit-identical golden vs the PRE-OPTIMIZATION capture --------------------------------------------
|
||
|
||
def _golden_fixture() -> dict[str, dict[int, float]]:
|
||
"""The exact fixture the golden snapshot in tests/unit/data/golden_sim.json was captured from
|
||
(pre-optimization `simulate_book` output). Must reproduce these values bit-identically (abs_tol 1e-9)."""
|
||
rng = random.Random(20260623)
|
||
return {
|
||
"crypto_tstrend": {1 + i: rng.gauss(0.0012, 0.012) for i in range(220)},
|
||
"unlock": {16 + i: rng.gauss(-0.010, 0.022) for i in range(200)},
|
||
"stablecoin_rotation": {6 + i: rng.gauss(0.0004, 0.004) for i in range(210)},
|
||
}
|
||
|
||
|
||
def test_simulate_book_bit_identical_to_pre_optimization_golden():
|
||
"""The optimized engine reproduces the PRE-OPTIMIZATION `simulate_book` output (full SimResult —
|
||
equity, drawdown, weights, leverage, killed, metrics) bit-identically (abs_tol 1e-9) across every
|
||
controller / kelly / cost / capital combo captured in the frozen golden file."""
|
||
golden = json.loads(_GOLDEN_PATH.read_text())
|
||
fx = _golden_fixture()
|
||
sleeves = ["crypto_tstrend", "unlock", "stablecoin_rotation"]
|
||
assert golden, "golden snapshot is empty"
|
||
for g in golden:
|
||
res = simulate_book(fx, sleeves=sleeves, **g["cfg"])
|
||
assert res.dates == g["dates"], f"date axis changed for {g['cfg']}"
|
||
assert res.killed == g["killed"], f"killed changed for {g['cfg']}"
|
||
for a, b in zip(res.equity, g["equity"], strict=True):
|
||
assert math.isclose(a, b, rel_tol=0, abs_tol=1e-9), f"equity {g['cfg']}"
|
||
for a, b in zip(res.drawdown, g["drawdown"], strict=True):
|
||
assert math.isclose(a, b, rel_tol=0, abs_tol=1e-9), f"drawdown {g['cfg']}"
|
||
for a, b in zip(res.leverage, g["leverage"], strict=True):
|
||
assert math.isclose(a, b, rel_tol=0, abs_tol=1e-9), f"leverage {g['cfg']}"
|
||
for wa, wb in zip(res.weights, g["weights"], strict=True):
|
||
for k in set(wa) | set(wb):
|
||
assert math.isclose(wa.get(k, 0.0), wb.get(k, 0.0), rel_tol=0, abs_tol=1e-9), \
|
||
f"weight[{k}] {g['cfg']}"
|
||
for k, v in g["metrics"].items():
|
||
assert math.isclose(res.metrics[k], v, rel_tol=0, abs_tol=1e-9), f"metric {k} {g['cfg']}"
|
||
|
||
|
||
# --- expensive/cheap split seam ---------------------------------------------------------------------
|
||
|
||
def test_cheap_reapply_equals_full_simulate_book():
|
||
"""`apply_knobs(overlay_pass(...), capital, cost_bps)` is bit-identical to a full `simulate_book` with
|
||
the same capital/cost_bps — the seam Phase-2's page caches (expensive overlay once; cheap knobs per
|
||
drag). capital + cost_bps are the genuinely-cheap knobs (kelly is NOT — it's in the overlay pass)."""
|
||
fx = _three_edge_fixture()
|
||
sleeves = ["crypto_tstrend", "unlock", "stablecoin_rotation"]
|
||
for kelly in (1.0, 0.5):
|
||
op = overlay_pass(fx, sleeves=sleeves, kelly_fraction=kelly)
|
||
for cap in (100_000.0, 37_000.0):
|
||
for cost in (0.0, 10.0, 25.0):
|
||
cheap = apply_knobs(op, capital=cap, cost_bps=cost)
|
||
full = simulate_book(fx, capital=cap, sleeves=sleeves,
|
||
kelly_fraction=kelly, cost_bps=cost)
|
||
assert cheap.dates == full.dates
|
||
assert cheap.killed == full.killed
|
||
assert cheap.leverage == full.leverage
|
||
assert cheap.equity == full.equity, f"k={kelly} cap={cap} cost={cost}"
|
||
assert cheap.drawdown == full.drawdown
|
||
assert cheap.weights == full.weights
|
||
assert cheap.metrics == full.metrics
|
||
|
||
|
||
def test_overlay_pass_independent_of_capital_and_cost():
|
||
"""The expensive overlay pass depends ONLY on (returns, sleeves, controller, kelly, window) — NOT on
|
||
capital/cost_bps. Re-using ONE pass for all capital/cost combos must give identical curves to passes
|
||
that never knew those knobs (the property that makes caching it sound)."""
|
||
fx = _three_edge_fixture()
|
||
sleeves = ["crypto_tstrend", "unlock", "stablecoin_rotation"]
|
||
op = overlay_pass(fx, sleeves=sleeves, controller="killswitch", kelly_fraction=1.0)
|
||
# The pass has no capital/cost notion; the gross (killed-aware) returns are fixed.
|
||
a = apply_knobs(op, capital=10_000.0, cost_bps=0.0)
|
||
b = apply_knobs(op, capital=40_000.0, cost_bps=0.0)
|
||
for ea, eb in zip(a.equity, b.equity, strict=True):
|
||
assert math.isclose(eb, ea * 4.0, rel_tol=1e-12, abs_tol=1e-6)
|
||
|
||
|
||
# --- performance benchmark --------------------------------------------------------------------------
|
||
|
||
@pytest.mark.slow
|
||
def test_full_history_simulate_book_under_bound():
|
||
"""A full ~1473-day run must not regress to the O(D²) recomputation (which would be MINUTES on this
|
||
series). The bound is generous (10s) ON PURPOSE: it must survive `pytest -n auto` CPU contention +
|
||
slower CI hardware while still catching the only regression that matters here — a quadratic blowup,
|
||
which is orders of magnitude, not a constant factor."""
|
||
fx = _big_fixture()
|
||
sleeves = list(fx)
|
||
simulate_book(fx, capital=100_000.0, sleeves=sleeves) # warm imports/JIT-free, fair timing
|
||
runs = [_time(lambda: simulate_book(fx, capital=100_000.0, sleeves=sleeves)) for _ in range(3)]
|
||
best = min(runs)
|
||
assert best < 10.0, f"full-history simulate_book too slow: {best:.3f}s (O(D²) regression?)"
|
||
|
||
|
||
@pytest.mark.slow
|
||
def test_cheap_knob_reapply_is_effectively_instant():
|
||
"""After the expensive overlay pass is computed once, re-applying capital/cost_bps must be FAR cheaper
|
||
than a full simulate — so dragging the capital / cost sliders on /paper/sim doesn't recompute the whole
|
||
overlay. Asserted RELATIVE to a full run (contention-robust: both times inflate together under load, the
|
||
ratio holds), not an absolute ms bound that flakes under `-n auto`."""
|
||
fx = _big_fixture()
|
||
sleeves = list(fx)
|
||
full = min(_time(lambda: simulate_book(fx, capital=100_000.0, sleeves=sleeves)) for _ in range(2))
|
||
op = overlay_pass(fx, sleeves=sleeves)
|
||
apply_knobs(op, capital=100_000.0, cost_bps=10.0) # warm
|
||
best = min(_time(lambda cap=cap: apply_knobs(op, capital=cap, cost_bps=10.0))
|
||
for cap in (100_000.0, 50_000.0, 200_000.0))
|
||
assert best < full / 10.0, f"cheap re-apply not cheap: {best*1000:.1f}ms vs full {full*1000:.1f}ms"
|
||
|
||
|
||
def _time(fn) -> float:
|
||
t0 = time.perf_counter()
|
||
fn()
|
||
return time.perf_counter() - t0
|
||
|
||
|
||
# --- simulate_naive_eqwt: the Bybit book's naive equal-weight curve (no overlay) -----------------
|
||
|
||
def test_naive_eqwt_compounds_static_equal_weight():
|
||
"""Hand-checked: 2 sleeves, eq-wt → daily ret = (ra+rb)/2, equity compounds from capital, lev≡1."""
|
||
rbs = {"a": {0: 0.10, 1: 0.20}, "b": {0: 0.30, 1: -0.10}}
|
||
res = simulate_naive_eqwt(rbs, capital=1000.0, sleeves=["a", "b"], cost_bps=0.0)
|
||
assert res.dates == [0, 1]
|
||
assert math.isclose(res.equity[0], 1000.0 * 1.20) # (0.1+0.3)/2 = 0.20
|
||
assert math.isclose(res.equity[1], 1000.0 * 1.20 * 1.05) # (0.2-0.1)/2 = 0.05
|
||
assert res.leverage == [1.0, 1.0] and res.killed == [False, False]
|
||
|
||
|
||
def test_naive_eqwt_capital_scales_end_value_linearly():
|
||
rbs = {"a": {0: 0.10, 1: 0.20}, "b": {0: 0.30, 1: -0.10}}
|
||
one = simulate_naive_eqwt(rbs, capital=10_000.0, sleeves=["a", "b"])
|
||
two = simulate_naive_eqwt(rbs, capital=20_000.0, sleeves=["a", "b"])
|
||
assert math.isclose(two.metrics["end_value"], 2.0 * one.metrics["end_value"])
|
||
# total_return is capital-invariant (shape only).
|
||
assert math.isclose(two.metrics["total_return"], one.metrics["total_return"])
|
||
|
||
|
||
def test_naive_eqwt_cost_bps_reduces_return():
|
||
rbs = {"a": {0: 0.01, 1: 0.01, 2: 0.01}, "b": {0: 0.01, 1: 0.01, 2: 0.01}}
|
||
cheap = simulate_naive_eqwt(rbs, capital=1000.0, sleeves=["a", "b"], cost_bps=0.0)
|
||
pricey = simulate_naive_eqwt(rbs, capital=1000.0, sleeves=["a", "b"], cost_bps=50.0)
|
||
assert pricey.metrics["end_value"] < cheap.metrics["end_value"]
|
||
|
||
|
||
def test_naive_eqwt_start_window_restricts_dates():
|
||
rbs = {"a": {0: 0.01, 1: 0.02, 2: 0.03, 3: 0.04}}
|
||
res = simulate_naive_eqwt(rbs, capital=1000.0, sleeves=["a"], start=2)
|
||
assert res.dates == [2, 3]
|
||
|
||
|
||
def test_naive_eqwt_empty_is_graceful():
|
||
res = simulate_naive_eqwt({}, capital=1000.0, sleeves=["a"])
|
||
assert res.dates == [] and res.metrics["end_value"] == 1000.0
|