PINK: TUI Hz fix + DC gate + ACB boost + 10 new tests (104/104 green)
TUI Hz fix: - hazelcast_projection.py: write_engine_snapshot now writes all NAUTILUS-era field aliases (trades_executed, current_leverage, open_positions as list, last_scan_number, last_vel_div, vol_ok, open_notional) so gear_rows/capital panel work with no TUI changes. - dolphin_status_pink.py: _normalize_eng_for_tui() safety-net translation added; render() uses it on every Hz read. DC gate (SYSTEM BIBLE §4.2, champion config): - pink_direct.py: _dc_contradicts() — 7-tick lookback, 0.75 bps threshold. Rising price (chg > 0.75 bps) blocks ENTER via dataclasses.replace(HOLD, DC_CONTRADICT). Price history deque initialized in connect(); dc_skip_contradicts=True enforced. ACB boost (SYSTEM BIBLE §10): - hazelcast_feed.py: fix wrong key "latest_acb" → "acb_boost" (DOLPHIN_FEATURES key written by acb_processor_service.py). - pink_direct.py: _last_acb_boost read from scan_payload["acb_boost"] first (scan bridge may embed it), then Hz direct fallback. Applied to intent.leverage via dataclasses.replace() after IntentEngine.plan(), capped at 3x. - _last_scan_number, _last_vel_div, _last_vol_ok tracked from scan_payload. OBF gate: NOT implemented. OBF shards (DOLPHIN_FEATURES_SHARD_*) require new Hz map connections + symbol routing. Gap documented; requires separate decision. Tests: TestDCGate (5) + TestNormalizeEngForTui (5) — 10 new, 104 total, all green. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -1392,3 +1392,90 @@ class TestVolOkGate:
|
||||
assert decision.action.value not in ("VOL_GATE",), (
|
||||
"vol_ok=True must not block on vol_ok gate"
|
||||
)
|
||||
|
||||
|
||||
# ============================================================
|
||||
# DC gate (_dc_contradicts) — SYSTEM BIBLE §4.2
|
||||
# ============================================================
|
||||
|
||||
class TestDCGate:
|
||||
"""Direction Confirmation gate: rising price over 7-tick window blocks SHORT entry."""
|
||||
|
||||
def _rt(self, prices: list):
|
||||
"""Build a minimal PinkDirectRuntime-like object with price history populated."""
|
||||
from collections import deque
|
||||
import types
|
||||
|
||||
obj = types.SimpleNamespace()
|
||||
obj._price_history = deque(prices, maxlen=10)
|
||||
|
||||
# Bind the method to obj
|
||||
from prod.clean_arch.runtime.pink_direct import PinkDirectRuntime
|
||||
obj._dc_contradicts = lambda **kw: PinkDirectRuntime._dc_contradicts(obj, **kw)
|
||||
return obj
|
||||
|
||||
def test_rising_price_contradicts(self):
|
||||
# p[-8] = 100.0, p[-1] = 100.2 → chg = +20 bps → CONTRADICT
|
||||
prices = [100.0] + [100.05] * 6 + [100.2]
|
||||
rt = self._rt(prices)
|
||||
assert rt._dc_contradicts(), "Rising price must be DC CONTRADICT"
|
||||
|
||||
def test_falling_price_confirms(self):
|
||||
# p[-8] = 100.0, p[-1] = 99.9 → chg = -10 bps → CONFIRM (not a block)
|
||||
prices = [100.0] + [99.95] * 6 + [99.9]
|
||||
rt = self._rt(prices)
|
||||
assert not rt._dc_contradicts(), "Falling price must NOT be DC CONTRADICT"
|
||||
|
||||
def test_flat_price_neutral(self):
|
||||
# < 0.75 bps change → NEUTRAL
|
||||
prices = [100.0] * 8
|
||||
rt = self._rt(prices)
|
||||
assert not rt._dc_contradicts(), "Flat price must NOT be DC CONTRADICT"
|
||||
|
||||
def test_insufficient_history_neutral(self):
|
||||
# < 8 prices → not enough data → NEUTRAL (allow entry)
|
||||
prices = [100.0, 100.5] # only 2 entries
|
||||
rt = self._rt(prices)
|
||||
assert not rt._dc_contradicts(), "Insufficient history must NOT block entry"
|
||||
|
||||
def test_exactly_threshold_neutral(self):
|
||||
# Exactly 0.75 bps → NOT a CONTRADICT (> not >=)
|
||||
prices = [100.0] + [100.0] * 6 + [100.0075] # 0.75 bps exactly
|
||||
rt = self._rt(prices)
|
||||
assert not rt._dc_contradicts(), "Exactly at threshold must NOT be CONTRADICT"
|
||||
|
||||
|
||||
# ============================================================
|
||||
# TUI normalization — _normalize_eng_for_tui
|
||||
# ============================================================
|
||||
|
||||
class TestNormalizeEngForTui:
|
||||
"""_normalize_eng_for_tui translates DITAv2 Hz snapshot to NAUTILUS-era field names."""
|
||||
|
||||
def _norm(self, eng: dict) -> dict:
|
||||
from Observability.dolphin_status_pink import _normalize_eng_for_tui
|
||||
return _normalize_eng_for_tui(eng)
|
||||
|
||||
def test_empty_returns_empty(self):
|
||||
assert self._norm({}) == {}
|
||||
|
||||
def test_already_nautilus_format_passthrough(self):
|
||||
eng = {"trades_executed": 5, "capital": 25000.0}
|
||||
out = self._norm(eng)
|
||||
assert out is eng or out["trades_executed"] == 5
|
||||
|
||||
def test_ditav2_format_adds_trades_executed(self):
|
||||
eng = {"trade_seq": 7, "capital": 25000.0, "slot": {}}
|
||||
out = self._norm(eng)
|
||||
assert out["trades_executed"] == 7, "trade_seq must be aliased as trades_executed"
|
||||
|
||||
def test_open_positions_becomes_list(self):
|
||||
eng = {"open_positions": 1, "slot": {"size": 0.5, "entry_price": 50000.0}}
|
||||
out = self._norm(eng)
|
||||
assert isinstance(out["open_positions"], list), "open_positions int must become list"
|
||||
assert len(out["open_positions"]) == 1
|
||||
|
||||
def test_zero_open_positions_empty_list(self):
|
||||
eng = {"open_positions": 0, "slot": {}}
|
||||
out = self._norm(eng)
|
||||
assert out["open_positions"] == [], "zero open_positions must become empty list"
|
||||
|
||||
Reference in New Issue
Block a user