gitadmin/fxhnt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

research(em-asia): Stage-1 LOO marginal-Sharpe kill-switch — China/India KILLED

Browse Source 
Backtest-only gate (no forward wiring) for the China/India diversifier question,
decided from ~7y of real Yahoo data. Reuses the SSOT marginal_sharpe primitive +
multistrat.book_series verbatim; cost model = max(vol-norm re-sizing turnover,
monthly-rebalance floor) so a quiet co-crasher can't survive at ~0 modeled cost.

Result (10bp UCITS, 2019-08..2026-07): CNYA.L marginal Sharpe -0.099, NDIA.L
-0.048 — both co-crash the book (+0.23 corr) and worsen with cost. KILL both:
the accessible UCITS wrapper carries EM beta, not the mainland inefficiency
(direct A-shares/NSE closed to the EU-retail entity). Same co-crash death as VRP
and unlock. Do NOT wire the em_asia forward sleeve.

Gate is generic — reusable for any single-instrument diversifier candidate.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Jeroen Grusewski
2026-07-19 22:56:30 +00:00
parent 4e7adf42f2
commit a00001e826
2 changed files with 390 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

234
src/fxhnt/application/em_asia_marginal_eval.py Normal file
View File

@@ -0,0 +1,234 @@
"""Stage-1 KILL-SWITCH for the China/India (EM-Asia) diversifier hunt — a backtest-only, decided-from-history
LOO marginal-Sharpe gate, run BEFORE any forward tracker is wired. This is the ghost-filter: a 20-day
reconciliation gate on two EM-beta ETFs would PASS trivially (the forward NAV just tracks the backtest NAV)
and prove NOTHING about edge. The real question is decision-theoretic and answerable now from ~8-10y of
history: does adding CNYA.L / NDIA.L to the fund's ETF book RAISE the book's risk-adjusted return after
costs, or is it a co-crashing beta sleeve that fails the marginal gate — the SAME bar that killed VRP and
the token-unlock diversifier (see pearl_fxhnt_diversifier_hunt_2026_07_15).
PASS bar (predetermined, not chosen after seeing the numbers): marginal Sharpe > 0 after cost.
> 0 -> proceed to Stage 2 (wire the `em_asia` registry sleeve + reconciliation forward gate).
<= 0 -> KILL: never build a forward tracker (co-crashing EM beta, not a diversifier).
Windows (per operator, 2026-07-19): each candidate judged on its OWN max window against the book over the
overlapping dates — CNYA.L ~10y, NDIA.L ~8y. Per-name verdicts (not a cross-name ranking), so differing
windows are acceptable.
Two marginal measures are reported per name:
1. `marginal_sharpe` (SSOT weight-blend, domain/diversification/marginal.py) — DIRECTLY comparable to the
prior VRP / unlock verdicts, which used this exact primitive.
2. In-book delta-Sharpe — `book_series(FUND + candidate) - book_series(FUND)` through the FULL adaptive
machinery (vol-norm -> trust -> correlation de-risk), i.e. the candidate's effect where it would
actually live. The truer measure; if the two disagree, that disagreement is itself signal.
Context columns (NOT the gate — they explain WHY it passed/failed, without moving the goalpost): correlation
to the book, drawdown in known crash windows, and cost sensitivity (marginal Sharpe at 5/10/20 bp).
No I/O beyond the Yahoo daily fetch (the same free chart endpoint the fund's `multistrat_nav` uses). Pure
numpy math otherwise.
"""
from __future__ import annotations
from dataclasses import dataclass, field
import numpy as np
from fxhnt.domain.diversification.marginal import _sharpe, marginal_sharpe
from fxhnt.domain.strategies import multistrat
# The two live-verified USD-denominated Accumulating UCITS lines (probed on Yahoo 2026-07-19). Acc (not Dist)
# so adjClose = true total return with no distribution gaps; USD so no GBp/FX mismatch with the USD book.
# CNYA.L — iShares MSCI China A UCITS ETF USD Acc — MSCI China A-shares (mainland Shanghai/Shenzhen), the
# retail-dominated inefficient market the fund cannot reach directly but can proxy via this wrapper.
# NDIA.L — iShares MSCI India UCITS ETF USD Acc.
_CANDIDATES: dict[str, str] = {"CNYA.L": "china_a", "NDIA.L": "india"}
# Per config.py class default for a UCITS line without a tighter probed estimate — thinner EM UCITS liquidity.
# Cost per REBALANCE = 2 * half_spread (buy+sell the traded notional). See config.UcitsListing.half_spread_bps.
_DEFAULT_HALF_SPREAD_BPS: float = 10.0
# Calendar-rebalance floor: even a low-turnover stream pays UCITS spread on a periodic re-balance. The book
# vol-normalizes daily but a real sleeve reconciles weights on a cadence; charging AT LEAST one round-trip per
# this many trading days prevents a co-crashing beta sleeve from "surviving" only because the modeled
# vol-norm turnover happened to be near-zero. Monthly (~21 trading days) matches a standard ETF rebalance.
_REBALANCE_FLOOR_DAYS: int = 21
# Known systemic crash windows the fund already reasons about — a diversifier must not merely co-crash here.
_CRASH_WINDOWS: dict[str, tuple[str, str]] = {
"covid_2020q1": ("2020-02-19", "2020-03-23"),
"bear_2022": ("2022-01-03", "2022-10-14"),
"yen_carry_2024": ("2024-07-31", "2024-08-07"),
}
@dataclass
class CandidateVerdict:
symbol: str
sleeve_id: str
window_start: str
window_end: str
n_days: int
marginal_sharpe_blend: float # SSOT weight-blend, cost-inclusive (comparable to VRP/unlock)
delta_sharpe_inbook: float # book_series(FUND+cand) - book_series(FUND), cost-inclusive
corr_to_book: float
crash_drawdowns: dict[str, float] = field(default_factory=dict)
cost_sensitivity: dict[str, float] = field(default_factory=dict) # half_spread_bps -> marginal_sharpe_blend
@property
def passes(self) -> bool:
"""PASS iff BOTH marginal measures are > 0 after cost. Either <= 0 => KILL (co-crashing EM beta)."""
a, b = self.marginal_sharpe_blend, self.delta_sharpe_inbook
return np.isfinite(a) and np.isfinite(b) and a > 0.0 and b > 0.0
def _aligned_returns(closes: dict[str, dict[str, float]], symbols: list[str]) -> tuple[list[str], np.ndarray]:
"""Intersect dates across `symbols` and return (dates, simple-return matrix R[T, S]) with R[0]=0.
Mirrors multistrat._build's intersection but WITHOUT dropping today (this is a pure historical eval)."""
dates = sorted(set.intersection(*[set(closes[s]) for s in symbols]))
R = np.zeros((len(dates), len(symbols)))
for j, sym in enumerate(symbols):
s = np.array([closes[sym][d] for d in dates])
if len(s) > 1:
R[1:, j] = s[1:] / s[:-1] - 1
return dates, R
def _candidate_net_returns(
closes: dict[str, dict[str, float]],
symbol: str,
*,
half_spread_bps: float,
) -> tuple[list[str], np.ndarray]:
"""The candidate sleeve's per-day NET (cost-inclusive) return series, as a single vol-normalized stream
treated the SAME way `book_series` treats each stream: raw simple returns -> `multistrat._volnorm` (the
`min(5, TARGET_VOL/realized_vol)` multiplier over a 63d window) -> MINUS the turnover cost implied by the
day-over-day change in the vol-normalization scale (the only thing that moves position size for a
single-instrument sleeve). Returns (dates, net_ret[T]); the warmup window (< 63d) is 0 and is dropped by
the caller via alignment against the book. Cost = `_turnover_cost` (vol-norm re-sizing OR a monthly
rebalance floor, whichever is larger) — see that function."""
dates = sorted(closes[symbol])
px = np.array([closes[symbol][d] for d in dates], dtype=float)
raw = np.zeros(len(px))
if len(px) > 1:
raw[1:] = px[1:] / px[:-1] - 1.0
scale = multistrat._volnorm_scale(raw) # per-day vol-normalization multiplier (0 in warmup)
gross = raw * scale # vol-normalized stream return (pre-cost)
cost = _turnover_cost(scale, half_spread_bps=half_spread_bps)
net = gross - cost
return dates, net
def _turnover_cost(scale: np.ndarray, *, half_spread_bps: float) -> np.ndarray:
"""Per-day cost drag (as a return) for a single vol-normalized stream whose position size on day t is
`scale[t]`. Charged as the MAXIMUM of two components, so neither a re-sizing spike nor a quiet hold escapes
cost:
1. Vol-normalization re-sizing turnover — the traded notional on day t is |scale[t] - scale[t-1]|; for a
SINGLE-instrument sleeve `scale` is the only thing that moves the position, so this is the exact
turnover from re-sizing. Each unit of traded notional pays a round-trip `2 * half_spread_bps`.
2. A calendar-rebalance FLOOR — one full round-trip of the average held position every
`_REBALANCE_FLOOR_DAYS` (monthly), amortized per day. Without this, a co-crashing beta sleeve whose
vol-norm turnover happens to be near-zero would "survive" the gate at near-zero cost — the ghost this
whole eval exists to kill. Wider EM UCITS spreads make erring conservative the right call.
Both are in RETURN units (fraction of the position's notional). Result is a per-day array aligned to
`scale`. Convention matches the fund's other turnover-cost models (spread_overlay.py: `turnover * bps/1e4`).
"""
n = len(scale)
if n == 0:
return np.zeros(0)
rt = 2.0 * half_spread_bps / 1e4 # round-trip cost per unit traded notional
resize_turnover = np.abs(np.diff(scale, prepend=scale[0])) # |Δscale| per day (day 0 -> 0)
resize_cost = resize_turnover * rt
# Floor: one round-trip of the held position amortized over the rebalance cadence. Use the day's own scale
# as the held notional so the floor scales with position size (0 in the warmup window where scale==0).
floor_cost = scale * rt / float(_REBALANCE_FLOOR_DAYS)
return np.maximum(resize_cost, floor_cost)
def _book_returns(closes: dict[str, dict[str, float]], instruments: list[str]) -> dict[str, float]:
"""The adaptive multistrat book's per-day return for `instruments`, as {date: ret} — reuses
`book_series` verbatim so 'the book' is genuinely the fund's book (full risk overlay), not a proxy blend.
drop_incomplete_today=False: pure historical eval, no live/today concern."""
return dict(multistrat.book_series(closes, instruments, drop_incomplete_today=False))
def _crash_drawdown(dates: list[str], ret: np.ndarray, window: tuple[str, str]) -> float:
"""Peak-to-trough return of the candidate stream inside [start, end] (fraction; negative = drawdown).
Returns nan if the window has < 2 days of overlap with the series."""
lo, hi = window
idx = [i for i, d in enumerate(dates) if lo <= d <= hi]
if len(idx) < 2:
return float("nan")
seg = ret[idx[0]: idx[-1] + 1]
eq = np.cumprod(1.0 + seg)
peak = np.maximum.accumulate(eq)
return float((eq / peak - 1.0).min())
def evaluate_candidate(
closes: dict[str, dict[str, float]],
symbol: str,
sleeve_id: str,
*,
fund_instruments: list[str] | None = None,
blend_weight: float = 0.2,
half_spread_bps: float = _DEFAULT_HALF_SPREAD_BPS,
) -> CandidateVerdict:
"""Run the Stage-1 LOO marginal-Sharpe gate for one candidate against the fund's ETF book, on the
candidate's own max window (intersected with the book's dates). `closes` must already contain adjClose
series for `fund_instruments` + `symbol`."""
fund = list(fund_instruments if fund_instruments is not None else multistrat.FUND_INSTRUMENTS)
# Book WITH vs WITHOUT the candidate (in-book delta-Sharpe, through the full adaptive machinery).
book_without = _book_returns(closes, fund)
book_with = _book_returns(closes, fund + [symbol])
common = sorted(set(book_without) & set(book_with))
bw = np.array([book_without[d] for d in common])
bwith = np.array([book_with[d] for d in common])
delta_inbook = _sharpe(bwith) - _sharpe(bw)
# SSOT weight-blend marginal Sharpe: book (without) vs the candidate's NET vol-normalized stream, aligned.
cand_dates, cand_net = _candidate_net_returns(closes, symbol, half_spread_bps=half_spread_bps)
cand_map = dict(zip(cand_dates, cand_net, strict=True))
blend_dates = sorted(set(book_without) & set(cand_map))
b_series = np.array([book_without[d] for d in blend_dates])
c_series = np.array([cand_map[d] for d in blend_dates])
m_blend = marginal_sharpe(b_series, c_series, weight=blend_weight)
corr = float(np.corrcoef(b_series, c_series)[0, 1]) if len(b_series) >= 2 else float("nan")
crash = {name: _crash_drawdown(cand_dates, cand_net, win) for name, win in _CRASH_WINDOWS.items()}
sens: dict[str, float] = {}
for hs in (5.0, 10.0, 20.0):
cd, cn = _candidate_net_returns(closes, symbol, half_spread_bps=hs)
cm = dict(zip(cd, cn, strict=True))
bd = sorted(set(book_without) & set(cm))
sens[f"{hs:g}bp"] = marginal_sharpe(
np.array([book_without[d] for d in bd]),
np.array([cm[d] for d in bd]),
weight=blend_weight,
)
return CandidateVerdict(
symbol=symbol,
sleeve_id=sleeve_id,
window_start=blend_dates[0] if blend_dates else "",
window_end=blend_dates[-1] if blend_dates else "",
n_days=len(blend_dates),
marginal_sharpe_blend=m_blend,
delta_sharpe_inbook=delta_inbook,
corr_to_book=corr,
crash_drawdowns=crash,
cost_sensitivity=sens,
)
def run_stage1(fetch_closes) -> list[CandidateVerdict]: # type: ignore[no-untyped-def]
"""Fetch each candidate + the fund instruments via `fetch_closes(symbol) -> {date: adjClose}` and return
the per-name verdicts. `fetch_closes` is injected (real = YahooDailyClient.adj_closes; tests pass a stub),
so this function is pure orchestration with no hard Yahoo dependency."""
syms = list(multistrat.FUND_INSTRUMENTS) + list(_CANDIDATES)
closes = {s: fetch_closes(s) for s in syms}
return [evaluate_candidate(closes, sym, sleeve_id) for sym, sleeve_id in _CANDIDATES.items()]

156
tests/application/test_em_asia_marginal_eval.py Normal file
View File

@@ -0,0 +1,156 @@
"""Tests for the Stage-1 EM-Asia (China/India) LOO marginal-Sharpe kill-switch."""
from __future__ import annotations
import datetime as dt
import numpy as np
import pytest
from fxhnt.application import em_asia_marginal_eval as ev
from fxhnt.domain.strategies import multistrat
def _dates(n: int, start: str = "2016-01-04") -> list[str]:
d0 = dt.date.fromisoformat(start)
out, d = [], d0
while len(out) < n:
if d.weekday() < 5: # weekdays only (equity calendar)
out.append(d.isoformat())
d += dt.timedelta(days=1)
return out
def _closes_from_rets(dates: list[str], rets: np.ndarray, p0: float = 100.0) -> dict[str, float]:
px = p0 * np.cumprod(1.0 + rets)
return dict(zip(dates, px, strict=True))
# ---------------- _turnover_cost ----------------
def test_turnover_cost_zero_scale_is_zero():
cost = ev._turnover_cost(np.zeros(50), half_spread_bps=10.0)
assert np.allclose(cost, 0.0)
def test_turnover_cost_charges_resizing():
# A single re-size step of +1.0 in scale pays a full round-trip (2*hs) on that unit of notional.
scale = np.array([0.0, 0.0, 1.0, 1.0])
cost = ev._turnover_cost(scale, half_spread_bps=10.0)
rt = 2.0 * 10.0 / 1e4
# day 2: |1.0-0.0|*rt = rt (resize) vs floor 1.0*rt/21 -> resize dominates
assert cost[2] == pytest.approx(rt)
def test_turnover_cost_floor_applies_on_quiet_hold():
# A constant nonzero scale => zero re-size turnover, so the calendar floor must still charge cost.
scale = np.full(30, 2.0)
cost = ev._turnover_cost(scale, half_spread_bps=10.0)
rt = 2.0 * 10.0 / 1e4
expected_floor = 2.0 * rt / ev._REBALANCE_FLOOR_DAYS
# day 0 has prepend==self so resize=0; every day the floor applies (scale constant => resize 0 after day 0)
assert cost[5] == pytest.approx(expected_floor)
assert cost[5] > 0.0 # the ghost-killer: a quiet hold is NOT free
def test_turnover_cost_takes_max_of_components():
scale = np.array([1.0, 5.0]) # big re-size on day 1
cost = ev._turnover_cost(scale, half_spread_bps=10.0)
rt = 2.0 * 10.0 / 1e4
resize = abs(5.0 - 1.0) * rt
floor = 5.0 * rt / ev._REBALANCE_FLOOR_DAYS
assert cost[1] == pytest.approx(max(resize, floor)) == pytest.approx(resize)
def test_turnover_cost_scales_with_half_spread():
scale = np.full(30, 2.0)
c10 = ev._turnover_cost(scale, half_spread_bps=10.0)
c20 = ev._turnover_cost(scale, half_spread_bps=20.0)
assert np.allclose(c20, 2.0 * c10)
# ---------------- CandidateVerdict.passes ----------------
def test_passes_requires_both_positive():
base = dict(symbol="X.L", sleeve_id="x", window_start="a", window_end="b", n_days=100, corr_to_book=0.1)
assert ev.CandidateVerdict(**base, marginal_sharpe_blend=0.1, delta_sharpe_inbook=0.1).passes
assert not ev.CandidateVerdict(**base, marginal_sharpe_blend=0.1, delta_sharpe_inbook=-0.1).passes
assert not ev.CandidateVerdict(**base, marginal_sharpe_blend=-0.1, delta_sharpe_inbook=0.1).passes
assert not ev.CandidateVerdict(**base, marginal_sharpe_blend=0.0, delta_sharpe_inbook=0.1).passes
def test_passes_false_on_nan():
base = dict(symbol="X.L", sleeve_id="x", window_start="a", window_end="b", n_days=100, corr_to_book=0.1)
assert not ev.CandidateVerdict(**base, marginal_sharpe_blend=float("nan"), delta_sharpe_inbook=0.1).passes
# ---------------- evaluate_candidate end-to-end (synthetic) ----------------
def _fund_closes(dates: list[str], rng: np.random.Generator) -> dict[str, dict[str, float]]:
"""Independent low-vol streams for the fund instruments so the book is a real adaptive book."""
closes: dict[str, dict[str, float]] = {}
for sym in multistrat.FUND_INSTRUMENTS:
rets = rng.normal(0.0004, 0.008, len(dates))
closes[sym] = _closes_from_rets(dates, rets)
return closes
def test_evaluate_candidate_kills_cocrashing_beta():
# A candidate that is just amplified book beta (perfectly correlated, higher vol, zero own alpha) must
# KILL: it adds risk without risk-adjusted return. This is the VRP/unlock failure mode.
rng = np.random.default_rng(7)
dates = _dates(1500)
closes = _fund_closes(dates, rng)
# book proxy = SPY stream; candidate = 1.5x that stream + noise (co-crashing beta, no diversification)
spy = np.array([closes["SPY"][d] for d in dates])
spy_ret = np.zeros(len(dates))
spy_ret[1:] = spy[1:] / spy[:-1] - 1
cand_ret = 1.5 * spy_ret + rng.normal(0, 0.001, len(dates))
closes["BETA.L"] = _closes_from_rets(dates, cand_ret)
v = ev.evaluate_candidate(closes, "BETA.L", "beta_test")
assert v.n_days > 200
# corr is measured against the ADAPTIVE book return (vol-normed, trust-weighted), so raw 1.5x-SPY beta
# shows as moderate positive co-movement, not ~1.0 — that dilution is the point of the adaptive overlay.
assert v.corr_to_book > 0.15 # genuinely co-moving (diluted through the overlay)
assert v.marginal_sharpe_blend < 0.0 # amplified beta adds vol without risk-adjusted return
assert not v.passes # KILL — no marginal risk-adjusted return
def test_evaluate_candidate_passes_true_diversifier():
# A candidate with independent positive-Sharpe returns (a genuine uncorrelated edge) should PASS both
# measures — proves the gate isn't rigged to always kill.
rng = np.random.default_rng(11)
dates = _dates(1500)
closes = _fund_closes(dates, rng)
# strong independent Sharpe, uncorrelated to the book
cand_ret = rng.normal(0.0009, 0.006, len(dates))
closes["DIV.L"] = _closes_from_rets(dates, cand_ret)
v = ev.evaluate_candidate(closes, "DIV.L", "div_test")
assert v.marginal_sharpe_blend > 0.0
assert v.delta_sharpe_inbook > 0.0
assert v.passes
def test_evaluate_candidate_reports_context_columns():
rng = np.random.default_rng(3)
dates = _dates(1500, start="2019-01-02") # covers covid_2020q1 + bear_2022 windows
closes = _fund_closes(dates, rng)
closes["CTX.L"] = _closes_from_rets(dates, rng.normal(0.0003, 0.007, len(dates)))
v = ev.evaluate_candidate(closes, "CTX.L", "ctx")
assert set(v.cost_sensitivity) == {"5bp", "10bp", "20bp"}
# more cost never HELPS marginal Sharpe (monotone non-increasing in spread)
assert v.cost_sensitivity["5bp"] >= v.cost_sensitivity["20bp"] - 1e-9
assert "covid_2020q1" in v.crash_drawdowns
def test_run_stage1_orchestration_uses_injected_fetch():
rng = np.random.default_rng(5)
dates = _dates(1200)
base = _fund_closes(dates, rng)
base["CNYA.L"] = _closes_from_rets(dates, rng.normal(0.0003, 0.012, len(dates)))
base["NDIA.L"] = _closes_from_rets(dates, rng.normal(0.0004, 0.011, len(dates)))
verdicts = ev.run_stage1(lambda s: base[s])
assert {v.sleeve_id for v in verdicts} == {"china_a", "india"}
assert all(v.n_days > 100 for v in verdicts)