diff --git a/src/fxhnt/application/em_asia_marginal_eval.py b/src/fxhnt/application/em_asia_marginal_eval.py
new file mode 100644
index 0000000..5961378
--- /dev/null
+++ b/src/fxhnt/application/em_asia_marginal_eval.py
@@ -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()]
diff --git a/tests/application/test_em_asia_marginal_eval.py b/tests/application/test_em_asia_marginal_eval.py
new file mode 100644
index 0000000..67548d8
--- /dev/null
+++ b/tests/application/test_em_asia_marginal_eval.py
@@ -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)