DOLPHIN/nautilus_dolphin/tests/test_green_only_features.py

"""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