"""BLUE-parity restoration tests (2026-06-10). The DITAv2 rewrite (Sprint 0) dropped two things the original PINK (full-engine launch_dolphin_bingx) had from the start: R1 IRP asset selection over the scan universe — PINK traded only the snapshot anchor (BTCUSDT) since the rewrite. R2 Cubic-convex dynamic leverage — the stub formula's confidence (|vdiv/threshold|) is ≥ 1.0 on every possible ENTER, so leverage was pinned at max_leverage (3.0) flat, and exchange leverage at the cap. These tests pin the restored behavior: - blue_parity.PinkAssetPicker / PinkAlphaSizer (wrappers over BLUE's exact kernels) - DecisionEngine sizer injection (and legacy path preserved verbatim) - IntentEngine honoring decision sizing - dual-leverage conviction map at the venue boundary - PinkDirectRuntime._effective_snapshot retargeting rules """ from __future__ import annotations import logging from collections import deque from datetime import datetime, timezone from types import SimpleNamespace import pytest from prod.clean_arch.dita.contracts import DecisionAction, DecisionConfig, DecisionContext, IntentContext from prod.clean_arch.dita.decision import DecisionEngine from prod.clean_arch.dita.intent import IntentEngine from prod.clean_arch.dita_v2.blue_parity import PinkAlphaSizer, PinkAssetPicker from prod.clean_arch.ports.data_feed import MarketSnapshot from prod.clean_arch.runtime.pink_direct import PinkDirectRuntime from prod.bingx.leverage import map_internal_conviction_to_exchange_leverage LOGGER = logging.getLogger("test_blue_parity") def make_snapshot(symbol="BTCUSDT", price=50000.0, vdiv=-0.03, irp=0.60, scan_number=100, payload=None): return MarketSnapshot( timestamp=datetime.now(timezone.utc), symbol=symbol, price=price, eigenvalues=[1.0], velocity_divergence=vdiv, irp_alignment=irp, scan_number=scan_number, scan_payload=payload if payload is not None else {"vol_ok": True}, ) def make_config(max_leverage=8.0): return DecisionConfig( vel_div_threshold=-0.02, vel_div_extreme=-0.05, fixed_tp_pct=0.0020, max_hold_bars=250, capital_fraction=0.20, max_leverage=max_leverage, allow_short=True, allow_long=False, ) def feed_picker(picker, series: dict, start_scan=1): """series: asset → list of prices; all lists same length.""" n = len(next(iter(series.values()))) for i in range(n): payload = { "assets": list(series), "asset_prices": [series[a][i] for a in series], } picker.observe(payload, scan_number=start_scan + i) def trending_series(lookback, down=0.997, up=1.003): n = lookback + 2 out = {"DOWNUSDT": [], "UPUSDT": []} d, u = 100.0, 100.0 for _ in range(n): d *= down u *= up out["DOWNUSDT"].append(d) out["UPUSDT"].append(u) return out # ── R2: sizer ───────────────────────────────────────────────────────────────── class TestPinkAlphaSizer: def _sizer(self, **kw): defaults = dict(min_leverage=0.5, max_leverage=8.0, leverage_convexity=3.0, vel_div_threshold=-0.02, vel_div_extreme=-0.05, use_dynamic_leverage=True, use_alpha_layers=False) defaults.update(kw) return PinkAlphaSizer(**defaults) def test_cubic_curve_checkpoints(self): s = self._sizer() # At threshold: strength 0 → min leverage assert s.calculate_size(capital=1e5, vel_div=-0.02)["leverage"] == pytest.approx(0.5) # Midpoint: strength 0.5 → 0.5 + 0.125 × 7.5 = 1.4375 assert s.calculate_size(capital=1e5, vel_div=-0.035)["leverage"] == pytest.approx(1.4375) # At/beyond extreme: strength 1 → max leverage assert s.calculate_size(capital=1e5, vel_div=-0.05)["leverage"] == pytest.approx(8.0) assert s.calculate_size(capital=1e5, vel_div=-0.30)["leverage"] == pytest.approx(8.0) def test_leverage_is_not_flat(self): """The regression: every entry used to come out at max_leverage.""" s = self._sizer() levs = {round(s.calculate_size(capital=1e5, vel_div=vd)["leverage"], 4) for vd in (-0.021, -0.03, -0.04, -0.05)} assert len(levs) > 1 def test_vd_trend_needs_ten_scans_and_dedupes(self): s = self._sizer() for i in range(9): s.observe(-0.02 - i * 0.001, scan_number=i + 1) assert s.vd_trend == 0.0 s.observe(-0.05, scan_number=9) # stale scan number → ignored assert s.vd_trend == 0.0 s.observe(-0.031, scan_number=10) assert s.vd_trend == pytest.approx(-0.031 - (-0.02)) def test_trade_feedback_roundtrip(self): s = self._sizer(use_alpha_layers=True) s.calculate_size(capital=1e5, vel_div=-0.06) # extreme bucket s.note_entry() s.record_close(150.0) # win stats = s._sizer.get_stats() assert sum(stats.get("bucket_wins", [0])) >= 1 or stats # recorded without raising def test_record_close_without_entry_is_noop(self): s = self._sizer() s.record_close(100.0) # must not raise # ── R1: picker ──────────────────────────────────────────────────────────────── class TestPinkAssetPicker: def test_warm_after_lookback_scans(self): p = PinkAssetPicker() series = trending_series(p.lookback) feed_picker(p, {k: v[: p.lookback] for k, v in series.items()}) assert not p.warm feed_picker(p, {k: v[p.lookback:] for k, v in series.items()}, start_scan=p.lookback + 1) assert p.warm def test_observe_dedupes_scan_number(self): p = PinkAssetPicker() payload = {"assets": ["AUSDT"], "asset_prices": [10.0]} assert p.observe(payload, scan_number=5) assert not p.observe(payload, scan_number=5) assert not p.observe(payload, scan_number=4) assert p.scans_observed == 1 def test_picks_downtrend_for_short_regime(self): p = PinkAssetPicker() feed_picker(p, trending_series(p.lookback)) choice = p.pick(direction=-1) assert choice is not None asset, px, ars = choice assert asset == "DOWNUSDT" assert px == pytest.approx(p.price_of("DOWNUSDT")) def test_no_candidate_returns_none(self): """All-uptrend universe in a SHORT regime → inverse rankings only → direction gate leaves nothing (BLUE: no fallback asset).""" p = PinkAssetPicker() n = p.lookback + 2 up1, up2 = [], [] a, b = 100.0, 50.0 for _ in range(n): a *= 1.004 b *= 1.003 up1.append(a) up2.append(b) feed_picker(p, {"AUSDT": up1, "BUSDT": up2}) assert p.pick(direction=-1) is None def test_price_of_unknown_asset(self): p = PinkAssetPicker() assert p.price_of("NOPEUSDT") is None # ── R2: decision/intent integration ────────────────────────────────────────── class TestDecisionSizerInjection: def test_sizer_drives_decision_leverage(self): sizer = PinkAlphaSizer(min_leverage=0.5, max_leverage=8.0, leverage_convexity=3.0, vel_div_threshold=-0.02, vel_div_extreme=-0.05, use_alpha_layers=False) eng = DecisionEngine(make_config(), sizer=sizer) ctx = DecisionContext(capital=100_000.0, open_positions=0, trade_seq=0) d = eng.decide(make_snapshot(vdiv=-0.035), ctx, None) assert d.action == DecisionAction.ENTER assert d.leverage == pytest.approx(1.4375) assert d.metadata.get("sizing") == "alpha_bet_sizer_cubic_v1" # target_size = capital × fraction × leverage / price assert d.target_size == pytest.approx(100_000 * 0.20 * 1.4375 / 50000.0) def test_legacy_path_unchanged_without_sizer(self): """Pin the legacy stub exactly: leverage saturates at max_leverage.""" eng = DecisionEngine(make_config(max_leverage=3.0)) ctx = DecisionContext(capital=100_000.0, open_positions=0, trade_seq=0) for vdiv in (-0.021, -0.035, -0.10): d = eng.decide(make_snapshot(vdiv=vdiv), ctx, None) assert d.action == DecisionAction.ENTER assert d.leverage == pytest.approx(3.0) def test_intent_honors_decision_sizing(self): sizer = PinkAlphaSizer(min_leverage=0.5, max_leverage=8.0, leverage_convexity=3.0, vel_div_threshold=-0.02, vel_div_extreme=-0.05, use_alpha_layers=False) cfg = make_config() eng = DecisionEngine(cfg, sizer=sizer) ieng = IntentEngine(cfg) ctx = DecisionContext(capital=100_000.0, open_positions=0, trade_seq=0) d = eng.decide(make_snapshot(vdiv=-0.035), ctx, None) plan = ieng.plan(d, IntentContext(capital=100_000.0, open_positions=0, trade_seq=0)) assert plan.intent.leverage == pytest.approx(d.leverage) assert plan.intent.target_size == pytest.approx(d.target_size) def test_intent_legacy_recompute_identical(self): """Honoring decision sizing must be a no-op for legacy decisions.""" cfg = make_config(max_leverage=3.0) eng = DecisionEngine(cfg) ieng = IntentEngine(cfg) ctx = DecisionContext(capital=100_000.0, open_positions=0, trade_seq=0) d = eng.decide(make_snapshot(vdiv=-0.03), ctx, None) plan = ieng.plan(d, IntentContext(capital=100_000.0, open_positions=0, trade_seq=0)) conf = max(0.05, min(1.0, d.confidence)) legacy_lev = min(cfg.max_leverage, max(1.0, 1.0 + conf * (cfg.max_leverage - 1.0))) assert plan.intent.leverage == pytest.approx(legacy_lev) # ── dual-leverage venue boundary ───────────────────────────────────────────── class TestConvictionToExchangeLeverage: def test_endpoints_and_midrange(self): m = lambda c: map_internal_conviction_to_exchange_leverage(c, exchange_max=3) assert m(0.5) == 1 assert m(9.0) == 3 assert m(4.75) == 2 # exact midpoint of [0.5, 9.0] → 2.0 assert m(0.1) == 1 # clamped below conviction floor assert m(50.0) == 3 # clamped above conviction ceiling def test_monotonic(self): vals = [map_internal_conviction_to_exchange_leverage(c, exchange_max=3) for c in (0.5, 2.0, 4.0, 6.0, 8.0, 9.0)] assert vals == sorted(vals) assert set(vals) == {1, 2, 3} # ── runtime retargeting ────────────────────────────────────────────────────── class FakeSlot: def __init__(self, asset="", size=0.0, free=True): self.asset = asset self.size = size self._free = free def is_free(self): return self._free class FakeKernel: max_slots = 1 def __init__(self, slot=None): self.slot0 = slot or FakeSlot() def slot(self, _i): return self.slot0 def make_runtime(picker=None, sizer=None, slot=None): return PinkDirectRuntime( data_feed=SimpleNamespace(), kernel=FakeKernel(slot), decision_engine=SimpleNamespace(config=make_config()), intent_engine=SimpleNamespace(), persistence=None, logger=LOGGER, asset_picker=picker, alpha_sizer=sizer, ) def universe_payload(prices: dict, scan_number: int): return { "assets": list(prices), "asset_prices": list(prices.values()), "scan_number": scan_number, "vel_div": -0.03, } class TestEffectiveSnapshot: def test_no_picker_passthrough(self): rt = make_runtime() snap = make_snapshot() out, block = rt._effective_snapshot(snap) assert out is snap and block == "" def test_cold_picker_blocks_entries_only(self): rt = make_runtime(picker=PinkAssetPicker()) snap = make_snapshot(payload=universe_payload({"BTCUSDT": 50000.0}, 1)) out, block = rt._effective_snapshot(snap) assert out.symbol == "BTCUSDT" assert "warming" in block and not block.startswith("all:") def test_flat_warm_picker_retargets_entry(self): p = PinkAssetPicker() feed_picker(p, trending_series(p.lookback)) rt = make_runtime(picker=p) snap = make_snapshot(payload=universe_payload( {"DOWNUSDT": p.price_of("DOWNUSDT"), "UPUSDT": p.price_of("UPUSDT")}, p.scans_observed + 1)) out, block = rt._effective_snapshot(snap) assert block == "" assert out.symbol == "DOWNUSDT" assert out.price == pytest.approx(p.price_of("DOWNUSDT")) # Regime signal untouched assert out.velocity_divergence == snap.velocity_divergence def test_open_slot_follows_slot_asset(self): p = PinkAssetPicker() feed_picker(p, trending_series(p.lookback)) slot = FakeSlot(asset="UPUSDT", size=2.0, free=False) rt = make_runtime(picker=p, slot=slot) snap = make_snapshot(payload=universe_payload( {"DOWNUSDT": 90.0, "UPUSDT": 110.0}, p.scans_observed + 1)) out, block = rt._effective_snapshot(snap) assert block == "" assert out.symbol == "UPUSDT" assert out.price == pytest.approx(110.0) def test_open_slot_unpriced_asset_blocks_all(self): p = PinkAssetPicker() slot = FakeSlot(asset="STRAYUSDT", size=1.0, free=False) rt = make_runtime(picker=p, slot=slot) snap = make_snapshot(payload=universe_payload({"BTCUSDT": 50000.0}, 1)) out, block = rt._effective_snapshot(snap) assert block.startswith("all:") assert out.symbol == "BTCUSDT" # unchanged; step() must HOLD def test_no_candidate_blocks_entry(self): p = PinkAssetPicker() n = p.lookback + 2 up = [100.0 * (1.004 ** i) for i in range(1, n + 1)] feed_picker(p, {"AUSDT": up}) rt = make_runtime(picker=p) snap = make_snapshot(payload=universe_payload({"AUSDT": up[-1]}, n + 1)) out, block = rt._effective_snapshot(snap) assert "no IRP candidate" in block def test_sizer_observe_fed_per_scan(self): sizer = PinkAlphaSizer(vel_div_threshold=-0.02, vel_div_extreme=-0.05, use_alpha_layers=False) rt = make_runtime(sizer=sizer) for i in range(12): snap = make_snapshot( scan_number=i + 1, payload={"scan_number": i + 1, "vel_div": -0.02 - i * 0.001}, ) rt._effective_snapshot(snap) assert len(sizer._vd_history) == 10 assert sizer.vd_trend != 0.0 if __name__ == "__main__": import sys sys.exit(pytest.main([__file__, "-v"]))