siloqy/prod/clean_arch/violet/test_violet_shadow_journal.py

"""V3e: shadow-decision journal — DDL-shape parity, sink emission, reject-at-source."""

from __future__ import annotations

import re
import sys
from pathlib import Path
from types import SimpleNamespace

sys.path.insert(0, "/mnt/dolphinng5_predict")

from prod.clean_arch.violet.decision_engine import ShadowDecision
from prod.clean_arch.violet.shadow_journal import DecisionRow, TABLE, VioletDecisionJournal

DDL = Path("/mnt/dolphinng5_predict/prod/clickhouse/violet/22_violet_decisions.sql")


def _ddl_columns() -> set:
    body = DDL.read_text()
    inner = body[body.index("("): body.rindex(")")]
    return set(re.findall(r"`(\w+)`", inner))


def _decision(**over):
    base = dict(
        ts_ns=10**12, scan_number=7, asset="BTCUSDT", side="SHORT", vel_div=-0.2,
        fraction=0.2, conviction_leverage=9.0, notional_fraction=1.8,
        target_exposure=124200.0, ars_score=1.48, bucket_idx=1, actuated=True,
    )
    base.update(over)
    return ShadowDecision(**base)


def test_decisionrow_fields_equal_ddl_columns():
    assert set(DecisionRow.model_fields.keys()) == _ddl_columns()


def test_journal_emits_validated_row_to_sink():
    captured = []
    j = VioletDecisionJournal(sink=lambda t, r: captured.append((t, r)), session_id="sess1")
    ok = j.journal(_decision(), mono_ns=999)
    assert ok and j.rows_emitted == 1
    table, row = captured[0]
    assert table == TABLE
    assert set(row.keys()) == _ddl_columns()
    assert row["asset"] == "BTCUSDT" and row["actuated"] == 1
    assert row["session_id"] == "sess1" and row["mono_ns"] == 999


def test_reject_at_source_on_malformed_decision():
    captured = []
    j = VioletDecisionJournal(sink=lambda t, r: captured.append(r), session_id="s")
    # duck-typed bad decision: non-finite vel_div must be rejected before the sink.
    bad = SimpleNamespace(
        scan_number=1, asset="BTCUSDT", side="SHORT", vel_div=float("nan"),
        fraction=0.2, conviction_leverage=9.0, notional_fraction=1.8,
        target_exposure=1.0, ars_score=1.0, bucket_idx=1, actuated=True,
    )
    ok = j.journal(bad, mono_ns=1)
    assert ok is False
    assert j.rows_rejected == 1 and j.rows_emitted == 0
    assert captured == []          # nothing reached the spool


def test_negative_exposure_rejected():
    j = VioletDecisionJournal(sink=lambda t, r: None, session_id="s")
    bad = SimpleNamespace(
        scan_number=1, asset="BTCUSDT", side="SHORT", vel_div=-0.2,
        fraction=0.2, conviction_leverage=9.0, notional_fraction=1.8,
        target_exposure=-5.0, ars_score=1.0, bucket_idx=1, actuated=True,
    )
    assert j.journal(bad, mono_ns=1) is False
    assert j.rows_rejected == 1


_BREAKDOWN_COLS = (
    "base_leverage", "dc_lev_mult", "regime_size_mult",
    "market_ob_mult", "esof_size_mult",
)


def test_full_factor_breakdown_round_trips_into_row():
    captured = []
    j = VioletDecisionJournal(sink=lambda t, r: captured.append(r), session_id="s")
    dec = _decision(
        base_leverage=4.0, dc_lev_mult=1.5, regime_size_mult=1.2,
        market_ob_mult=1.4, esof_size_mult=0.95,
    )
    assert j.journal(dec, mono_ns=1) and j.rows_emitted == 1
    row = captured[0]
    assert set(row.keys()) == _ddl_columns()        # parity holds with new columns
    assert row["base_leverage"] == 4.0 and row["dc_lev_mult"] == 1.5
    assert row["regime_size_mult"] == 1.2 and row["market_ob_mult"] == 1.4
    assert row["esof_size_mult"] == 0.95


def test_base_only_path_leaves_breakdown_null():
    captured = []
    j = VioletDecisionJournal(sink=lambda t, r: captured.append(r), session_id="s")
    assert j.journal(_decision(), mono_ns=1)        # no breakdown supplied
    row = captured[0]
    assert all(row[c] is None for c in _BREAKDOWN_COLS)


def test_negative_breakdown_multiplier_rejected_at_source():
    j = VioletDecisionJournal(sink=lambda t, r: None, session_id="s")
    # ShadowDecision itself guards ge=0.0, so build a duck-typed decision that
    # smuggles a negative multiplier past the engine to prove the row guard catches it.
    bad = SimpleNamespace(
        scan_number=1, asset="BTCUSDT", side="SHORT", vel_div=-0.2,
        fraction=0.2, conviction_leverage=9.0, notional_fraction=1.8,
        target_exposure=1.0, ars_score=1.0, bucket_idx=1, actuated=True,
        base_leverage=4.0, dc_lev_mult=-0.5, regime_size_mult=1.0,
        market_ob_mult=1.0, esof_size_mult=1.0,
    )
    assert j.journal(bad, mono_ns=1) is False
    assert j.rows_rejected == 1