181 lines
12 KiB
Python
181 lines
12 KiB
Python
from fxhnt.application.allocation_engine import compute_allocation
|
||
from fxhnt.application.allocation_policy import AllocationPolicy
|
||
|
||
|
||
def _series(n, ret): # n days of a constant-ish return
|
||
return {f"2026-{(6 + d // 30):02d}-{(d % 30) + 1:02d}": ret for d in range(n)}
|
||
|
||
|
||
def test_excludes_short_history_and_empty_returns_empty():
|
||
pol = AllocationPolicy(min_history_days=60, marginal_gate=False)
|
||
assert compute_allocation({"short": _series(10, 0.001)}, pol) == {} # < 60 obs → excluded → empty
|
||
|
||
|
||
def test_inverse_vol_base_and_ex_ante_leverage():
|
||
pol = AllocationPolicy(min_history_days=5, marginal_gate=False, max_leverage=1.5,
|
||
kelly_fraction=0.5, target_vol=0.12, kill_dd=0.99, reenter_dd=0.98)
|
||
# NOTE: a and b use the SAME alternating pattern (not exact complements) — two strategies that are
|
||
# perfect complements make the equal-weight book flat (constant per-day average), i.e. zero full-history
|
||
# vol, which degenerately forces K=0 and defeats the point of this test (a positive, bounded K).
|
||
a = {f"2026-06-{d:02d}": (0.01 if d % 2 else -0.005) for d in range(1, 21)}
|
||
b = {f"2026-06-{d:02d}": (0.01 if d % 3 == 0 else -0.005) for d in range(1, 21)}
|
||
w = compute_allocation({"a": a, "b": b}, pol)
|
||
assert set(w) == {"a", "b"}
|
||
# base is now risk-parity (inverse full-history vol), capped — a and b have DIFFERENT own-vol (this
|
||
# fixture was never equal-vol), so they get DIFFERENT base weights. Hand-computed from each series'
|
||
# own full_history_vol: vol_a ~= 0.14329 (10x +0.01/10x -0.005, alternating), vol_b ~= 0.13132 (6x
|
||
# +0.01 on d%3==0 / 14x -0.005) -> b is lower-vol -> rp_b ~= 0.52178 > the 0.5 cap -> base_b = 0.5
|
||
# (capped); rp_a ~= 0.47822 < 0.5 -> base_a = 0.47822 (uncapped). b ends up with strictly MORE weight
|
||
# than a (lower vol -> more risk-parity weight, only clipped by the concentration cap).
|
||
from fxhnt.application.allocation_engine import full_history_vol
|
||
from fxhnt.application.promotion_corr import deploy_book_returns
|
||
vol_a, vol_b = full_history_vol(a), full_history_vol(b)
|
||
inv_a, inv_b = 1.0 / vol_a, 1.0 / vol_b
|
||
rp_a, rp_b = inv_a / (inv_a + inv_b), inv_b / (inv_a + inv_b)
|
||
base_a, base_b = min(rp_a, pol.max_strategy_weight), min(rp_b, pol.max_strategy_weight)
|
||
assert abs(base_b - 0.5) < 1e-9 and base_a < 0.5 # cap binds on b, not on a
|
||
assert w["b"] > w["a"] # lower-vol b gets more weight
|
||
# K (the ex-ante vol-target leverage) is UNCHANGED by the base-weighting scheme — it derives from the
|
||
# equal-weight book's full-history vol, same formula as before this task.
|
||
book_vol = full_history_vol(deploy_book_returns({"a": a, "b": b}))
|
||
k = min(pol.max_leverage, pol.kelly_fraction * pol.target_vol / book_vol)
|
||
assert 0 < k <= 1.5
|
||
assert abs(w["a"] - k * base_a) < 1e-9
|
||
assert abs(w["b"] - k * base_b) < 1e-9
|
||
|
||
|
||
def test_recent_vol_spike_does_not_change_K_antireactive():
|
||
# THE load-bearing invariant: ex-ante vol uses the FULL history, so appending a recent spike must not
|
||
# change the leverage on the earlier window.
|
||
pol = AllocationPolicy(min_history_days=5, marginal_gate=False, kill_dd=0.99, reenter_dd=0.98)
|
||
base = {f"2026-06-{d:02d}": (0.01 if d % 2 else -0.01) for d in range(1, 61)}
|
||
w_before = compute_allocation({"a": dict(base), "b": dict(base)}, pol)
|
||
spiked = dict(base)
|
||
spiked["2026-06-30"] = 0.5 # a single huge recent day
|
||
w_after = compute_allocation({"a": spiked, "b": dict(base)}, pol)
|
||
# the full-history vol of "a" changes (it must — vol IS a full-history stat), so K adapts SLOWLY over the
|
||
# whole record, NOT reactively to the last window. Assert K moved only via the full-history vol, i.e. the
|
||
# book vol recomputed over ALL 60 days (not a trailing subset):
|
||
from fxhnt.application.allocation_engine import full_history_vol
|
||
from fxhnt.application.promotion_corr import deploy_book_returns
|
||
vol_all = full_history_vol(deploy_book_returns({"a": spiked, "b": dict(base)}))
|
||
k_raw = min(pol.max_leverage, pol.kelly_fraction * pol.target_vol / vol_all)
|
||
# The spike also blows up "a"'s OWN full-history vol relative to "b" (unspiked), so the risk-parity base
|
||
# is no longer 1/2 each: rp_a ~= 0.13378 (uncapped), rp_b ~= 0.86622 -> capped at the 0.5 concentration
|
||
# limit -> base_a ~= 0.13378, base_b = 0.5, base sum ~= 0.63378 (< 1, cap binds and the book holds cash
|
||
# on the "b" side). sum(w_after) is therefore k_raw * base_sum, NOT k_raw alone (that identity only held
|
||
# under the old equal-weight base where base always summed to 1 when the cap didn't bind).
|
||
vol_spiked_a, vol_unspiked_b = full_history_vol(spiked), full_history_vol(dict(base))
|
||
inv_a, inv_b = 1.0 / vol_spiked_a, 1.0 / vol_unspiked_b
|
||
rp_a, rp_b = inv_a / (inv_a + inv_b), inv_b / (inv_a + inv_b)
|
||
base_a, base_b = min(rp_a, pol.max_strategy_weight), min(rp_b, pol.max_strategy_weight)
|
||
assert abs(base_b - 0.5) < 1e-9 and base_a < 0.5 # cap binds on b (the lower-vol, unspiked one)
|
||
k_before, k_after = sum(w_before.values()), sum(w_after.values())
|
||
assert abs(k_after - k_before) > 1e-9 # the spike DOES move full-history vol...
|
||
assert abs(k_after - k_raw * (base_a + base_b)) < 1e-9 # ...but K itself is still the SAME raw formula
|
||
|
||
|
||
def test_killswitch_flattens():
|
||
pol = AllocationPolicy(min_history_days=5, marginal_gate=False, kill_dd=0.20, reenter_dd=0.10)
|
||
crash = {f"2026-06-{d:02d}": -0.04 for d in range(1, 11)} # deep DD → killswitch
|
||
w = compute_allocation({"a": crash, "b": crash}, pol)
|
||
assert all(v == 0.0 for v in w.values()) # flattened
|
||
|
||
|
||
def test_killswitch_uses_injected_live_record_not_backtest():
|
||
# Fix A: the killswitch is LIVE protection. The backtest SIZING series has a deep (~>20%) drawdown that
|
||
# WOULD latch the killswitch, but the injected forward/live record is calm -> it must NOT latch, so the
|
||
# allocation populates. A backtest maxDD must never permanently disable an edge.
|
||
pol = AllocationPolicy(min_history_days=5, marginal_gate=False, kill_dd=0.20, reenter_dd=0.10)
|
||
# alternating losses -> a deep drawdown (trips the killswitch) AND non-zero full-history vol (so K>0 and a
|
||
# benign run is genuinely non-zero, not degenerately flat).
|
||
crash = {f"2026-06-{d:02d}": (-0.08 if d % 2 else -0.02) for d in range(1, 11)}
|
||
benign = {f"2026-06-{d:02d}": (0.001 if d % 2 else -0.001) for d in range(1, 11)} # calm live record
|
||
w = compute_allocation({"a": crash, "b": crash}, pol, killswitch_returns={"a": benign, "b": benign})
|
||
assert w and any(v != 0.0 for v in w.values()) # NOT flattened -> allocation populates
|
||
|
||
|
||
def test_killswitch_none_matches_sizing_series_behavior():
|
||
# killswitch_returns=None (or omitted) -> the killswitch runs on the sizing series (pre-fix behavior): the
|
||
# deep-drawdown backtest still flattens, and omitting the arg is byte-identical to passing None.
|
||
pol = AllocationPolicy(min_history_days=5, marginal_gate=False, kill_dd=0.20, reenter_dd=0.10)
|
||
crash = {f"2026-06-{d:02d}": (-0.08 if d % 2 else -0.02) for d in range(1, 11)}
|
||
w_omit = compute_allocation({"a": crash, "b": crash}, pol)
|
||
w_none = compute_allocation({"a": crash, "b": crash}, pol, killswitch_returns=None)
|
||
assert all(v == 0.0 for v in w_omit.values()) # sizing-series drawdown -> flattened
|
||
assert w_omit == w_none # None is byte-identical to omitting the arg
|
||
|
||
|
||
def test_killswitch_absent_from_live_record_does_not_latch():
|
||
# Survivors absent from the injected live record -> empty ks_book -> killswitch_active({}) is False (no
|
||
# kill), so the allocation still populates rather than spuriously flattening.
|
||
pol = AllocationPolicy(min_history_days=5, marginal_gate=False, kill_dd=0.20, reenter_dd=0.10)
|
||
crash = {f"2026-06-{d:02d}": (-0.08 if d % 2 else -0.02) for d in range(1, 11)}
|
||
w = compute_allocation({"a": crash, "b": crash}, pol, killswitch_returns={})
|
||
assert w and any(v != 0.0 for v in w.values())
|
||
|
||
|
||
def test_per_strategy_cap_binds_and_holds_cash():
|
||
# n=1 with the default 0.5 cap: the sole survivor must be capped at 0.5 (not 1.0) BEFORE leverage —
|
||
# the concentration guardrail fires and the book holds cash.
|
||
pol = AllocationPolicy(min_history_days=5, marginal_gate=False, max_strategy_weight=0.5,
|
||
kill_dd=0.99, reenter_dd=0.98)
|
||
a = {f"2026-06-{d:02d}": (0.01 if d % 2 else -0.005) for d in range(1, 21)}
|
||
w = compute_allocation({"a": a}, pol)
|
||
# base weight is capped at 0.5; target = K * 0.5, so w["a"] / K == 0.5 (cap fired, not 1.0)
|
||
from fxhnt.application.allocation_engine import full_history_vol
|
||
from fxhnt.application.promotion_corr import deploy_book_returns
|
||
vol = full_history_vol(deploy_book_returns({"a": a}))
|
||
k = min(pol.max_leverage, pol.kelly_fraction * pol.target_vol / vol)
|
||
assert abs(w["a"] - k * 0.5) < 1e-9 # capped at 0.5, NOT k*1.0
|
||
|
||
|
||
def test_cap_does_not_bind_when_many_strategies():
|
||
pol = AllocationPolicy(min_history_days=5, marginal_gate=False, max_strategy_weight=0.5,
|
||
kill_dd=0.99, reenter_dd=0.98)
|
||
strat = {f"s{i}": {f"2026-06-{d:02d}": (0.01 if (d + i) % 2 else -0.005) for d in range(1, 21)}
|
||
for i in range(3)}
|
||
w = compute_allocation(strat, pol)
|
||
# 1/3 < 0.5 → no cap → equal 1/3 base each (× K)
|
||
k = sum(w.values())
|
||
for v in w.values():
|
||
assert abs(v - k / 3) < 1e-9
|
||
|
||
|
||
import numpy as np
|
||
|
||
|
||
def _book(strat_returns): # single strategy so the book == its returns; marginal_gate off
|
||
pol = AllocationPolicy(min_history_days=5, marginal_gate=False, max_leverage=10.0,
|
||
kelly_fraction=0.5, target_vol=0.12, kill_dd=0.99, reenter_dd=0.98, cvar_alpha=0.10)
|
||
return compute_allocation({"a": {f"d{i}": v for i, v in enumerate(strat_returns)}}, pol)
|
||
|
||
|
||
def test_cvar_noop_on_gaussian_book():
|
||
from fxhnt.application.allocation_engine import full_history_vol
|
||
rng = np.random.default_rng(0)
|
||
g = list(rng.normal(0.0, 0.01, 400))
|
||
book = {f"d{i}": v for i, v in enumerate(g)}
|
||
w = _book(g)["a"]
|
||
# On a Gaussian book the CVaR floor is a NO-OP (k_cvar >= k_vol), so the weight must equal the std-only
|
||
# vol-target weight EXACTLY. Base weight for the single strategy = min(1/1, max_strategy_weight=0.5) = 0.5.
|
||
# This would FAIL if CVaR bit (a fat tail would de-lever below k_vol) — not a `w > 0` tautology.
|
||
k_vol = min(10.0, 0.5 * 0.12 / full_history_vol(book))
|
||
assert abs(w - k_vol * 0.5) < 1e-9
|
||
|
||
|
||
def test_cvar_bites_fat_left_tail():
|
||
# same std as a Gaussian control, but a fat LEFT tail (rare big losses) -> CVaR must de-lever it.
|
||
rng = np.random.default_rng(1)
|
||
gauss = list(rng.normal(0.0, 0.02, 400))
|
||
fat = list(rng.normal(0.0008, 0.006, 380)) + [-0.15, -0.12, -0.18, -0.10] * 5 # steady small gains + rare crashes
|
||
# match std EXACTLY so k_vol is identical for both books and the ONLY thing that can separate their
|
||
# leverage is the CVaR tail cap. Rescaling is std-invariant on the cv/std ratio, so the fat book keeps
|
||
# its fat left tail (cv/std ~= 2.39 vs the Gaussian's ~= 1.85). Without this the crashes would inflate the
|
||
# fat book's std and k_vol alone would de-lever it — the test would pass even under std-only sizing and
|
||
# prove nothing about CVaR.
|
||
fat = list(np.asarray(fat) * (np.std(gauss) / np.std(fat)))
|
||
w_g = _book(gauss)["a"]
|
||
w_f = _book(fat)["a"]
|
||
assert w_f < w_g # matched std -> identical k_vol; strictly-less leverage comes ONLY from the CVaR floor
|