diff --git a/nautilus_dolphin/tests/test_green_only_features.py b/nautilus_dolphin/tests/test_green_only_features.py new file mode 100644 index 0000000..efac07e --- /dev/null +++ b/nautilus_dolphin/tests/test_green_only_features.py @@ -0,0 +1,170 @@ +"""GREEN-only feature tests (exp/green-s6-esof-aem-shadow-2026-04-21). + +Split complement to `test_green_blue_parity.py`. That file guards invariants +that must hold across GREEN's divergence (signal, DC, hibernate, bucket SL, etc.). +This file covers the sprint-specific additions: + + * S6 bucket-ban at AlphaAssetSelector (selector skips banned buckets, + rankings slide up so the slot goes to the next-ranked asset) + * S6 + EsoF + int-leverage at the orchestrator single-site (line 565 region) + * AEM per-bucket MAE_MULT table (B3 disables MAE stop, B4 cuts fast, B6 wide) + * NEUTRAL → UNKNOWN label rename (advisor emits UNKNOWN; alias still resolves) + * Toggles-OFF identity: with every GREEN-only kwarg left at the BLUE default + (None / False), the orchestrator's per-entry math equals the pre-sprint form. + +All tests import pure Python modules — no NautilusKernel, no CH/HZ dependency. +""" +from __future__ import annotations + +import math +from typing import Dict, List, Optional + +import pytest + + +# ── Toggles-OFF identity (BLUE invariance) ─────────────────────────────────── + +def test_toggles_off_notional_matches_blue_pre_sprint_formula(): + """With every GREEN toggle left at the BLUE default (None/False), the + single-site notional reduces algebraically to `capital * fraction * leverage` + — the exact form BLUE used before the sprint landed. Guards against silent + regressions that would show up as BLUE PnL drift.""" + from nautilus_dolphin.nautilus.esf_alpha_orchestrator import NDAlphaEngine + + eng = NDAlphaEngine(initial_capital=25000.0) # all GREEN kwargs default + assert eng.s6_size_table is None + assert eng.esof_sizing_table is None + assert eng.asset_bucket_data is None + assert eng.use_int_leverage is False + # EsoF runtime state initialised to 1.0 (no-op) + assert eng._esof_size_mult == 1.0 + assert eng._current_esof_label is None + + +def test_ndposition_new_fields_default_to_neutral_values(): + """New GREEN-only observability fields on NDPosition must default so that + a BLUE entry produces a record indistinguishable from the pre-sprint shape + in every field that already existed.""" + from nautilus_dolphin.nautilus.esf_alpha_orchestrator import NDPosition + + pos = NDPosition( + trade_id="t", asset="BTCUSDT", direction=-1, entry_price=100.0, + entry_bar=0, notional=1000.0, leverage=2.0, fraction=0.1, + entry_vel_div=-0.03, bucket_idx=0, + ) + assert pos.asset_bucket_id is None + assert pos.s6_mult == 1.0 + assert pos.esof_mult == 1.0 + assert pos.esof_label is None + assert pos.leverage_raw is None + + +# ── S6 selector ban ────────────────────────────────────────────────────────── + +def test_selector_bucket_ban_skips_banned_bucket_not_slot(): + """Core caveat from the plan: selector-ban must NOT just zero-size the slot; + it must skip the asset so the next ranking takes the slot. Ban ≠ 0× sizer.""" + from nautilus_dolphin.nautilus.alpha_asset_selector import AlphaAssetSelector + + sel = AlphaAssetSelector( + asset_bucket_ban_set={4}, + asset_bucket_assignments={"AAA": 4, "BBB": 0, "CCC": 2}, + ) + # We inspect the ban wiring directly — the kernel is tested in the parity file. + assert sel.asset_bucket_ban_set == {4} + assert sel.asset_bucket_assignments["AAA"] == 4 + + # Simulate the in-loop guard: banned buckets should be filtered. + candidates = ["AAA", "BBB", "CCC"] + survivors = [ + a for a in candidates + if sel.asset_bucket_assignments.get(a) not in (sel.asset_bucket_ban_set or set()) + ] + assert survivors == ["BBB", "CCC"] + assert "AAA" not in survivors + + +def test_selector_default_ban_set_is_none_for_blue_invariance(): + from nautilus_dolphin.nautilus.alpha_asset_selector import AlphaAssetSelector + + sel = AlphaAssetSelector() + assert sel.asset_bucket_ban_set is None + assert sel.asset_bucket_assignments == {} + + +# ── AEM per-bucket MAE_MULT ────────────────────────────────────────────────── + +def test_mae_mult_b3_disables_stop(): + """B3 (natural winners) is configured with None → MAE_STOP is skipped even + at huge MAE. Giveback/time still apply.""" + from adaptive_exit.adaptive_exit_engine import MAE_MULT_BY_BUCKET + + assert MAE_MULT_BY_BUCKET[3] is None + + +def test_mae_mult_b4_is_strict(): + """B4 (gross-negative alpha) gets a tight 2.0× stop — cut fast.""" + from adaptive_exit.adaptive_exit_engine import MAE_MULT_BY_BUCKET + + assert MAE_MULT_BY_BUCKET[4] == 2.0 + + +def test_mae_mult_b6_is_wide(): + """B6 (extreme vol) gets a wide 6.0× band — noise tolerance.""" + from adaptive_exit.adaptive_exit_engine import MAE_MULT_BY_BUCKET + + assert MAE_MULT_BY_BUCKET[6] == 6.0 + + +def test_mae_mult_default_fallback_matches_legacy_constant(): + """Unknown bucket_id falls back to MAE_MULT_TIER1 (the pre-sprint constant).""" + from adaptive_exit.adaptive_exit_engine import MAE_MULT_BY_BUCKET, MAE_MULT_TIER1 + + unknown_bucket = 99 + assert MAE_MULT_BY_BUCKET.get(unknown_bucket, MAE_MULT_TIER1) == MAE_MULT_TIER1 + + +# ── EsoF label rename (NEUTRAL → UNKNOWN) ──────────────────────────────────── + +def test_esof_advisor_emits_unknown_not_neutral_in_midband(): + """The advisor's mid-band score must produce UNKNOWN, not NEUTRAL. Guards + against accidental revert of the rename (the semantic shift — 'signals in + conflict' — is load-bearing for the regime gate).""" + import Observability.esof_advisor as esof_advisor # noqa: N813 + + # Sanity-scan source for the rename; the computed advisor call requires + # deep Hz/CH plumbing we don't want to mock here. + src = open(esof_advisor.__file__).read() + assert 'advisory_label = "UNKNOWN"' in src + assert 'advisory_label = "NEUTRAL"' not in src + + +def test_esof_gate_tables_accept_both_unknown_and_neutral_keys(): + """S6_MULT and IRP_PARAMS must carry both UNKNOWN and NEUTRAL keys so that + historical CH replays (old label) resolve the same as new advisor output + (new label).""" + from Observability.esof_gate import S6_MULT, IRP_PARAMS + + assert "UNKNOWN" in S6_MULT and "NEUTRAL" in S6_MULT + assert S6_MULT["UNKNOWN"] == S6_MULT["NEUTRAL"] + assert "UNKNOWN" in IRP_PARAMS and "NEUTRAL" in IRP_PARAMS + assert IRP_PARAMS["UNKNOWN"] == IRP_PARAMS["NEUTRAL"] + + +# ── Int-leverage gate ──────────────────────────────────────────────────────── + +def test_int_leverage_gate_default_is_1x_min_clamped(): + """The in-source rule is `leverage_int = 1` plus min/abs_max clamp. Guards + against anyone accidentally switching to Option 1 or 2 without running the + winrate analysis first.""" + import pathlib + + src = pathlib.Path( + "nautilus_dolphin/nautilus_dolphin/nautilus/esf_alpha_orchestrator.py" + ).read_text() + # Fixed default must be explicit and annotated. + assert "leverage_int = 1" in src + assert "FIXED PENDING ANALYSIS" in src + # Both candidate rules must remain documented in-source for the flip PR. + assert "round-half-up" in src + assert "banker's round" in src