siloqy/prod/tests/test_pink_ditav2_kernel_bridge.py

"""Decision → KernelIntent mapping table tests for PINK → DITAv2 bridge."""

from __future__ import annotations

from datetime import datetime, timezone
import unittest
import sys
import os

# Minimal import path — avoid dita_v2.__init__ which pulls in bingx_venue + legacy DITA
sys.path.insert(0, "/mnt/dolphinng5_predict/prod")
sys.path.insert(0, "/mnt/dolphinng5_predict/prod/clean_arch")

os.environ.setdefault("HZ_CLUSTER", "dolphin")
os.environ.setdefault("HZ_HOST", "localhost:5701")
os.environ.setdefault("BINGX_API_KEY", "test")
os.environ.setdefault("BINGX_SECRET_KEY", "test")

from clean_arch.dita import (
    Decision,
    DecisionAction,
    Intent,
    TradeSide as LegacyTradeSide,
    TradeStage as LegacyTradeStage,
)
from clean_arch.dita_v2.contracts import (
    KernelCommandType,
    KernelIntent,
    TradeSide as DitaTradeSide,
)
from clean_arch.runtime.pink_direct import _decision_to_kernel_intent


def _make_test_decision(
    action: DecisionAction = DecisionAction.ENTER,
    side: LegacyTradeSide = LegacyTradeSide.SHORT,
) -> Decision:
    return Decision(
        timestamp=datetime.now(timezone.utc),
        decision_id="test-decision-001",
        asset="BTCUSDT",
        action=action,
        side=side,
        reason="test",
        confidence=0.8,
        velocity_divergence=-0.03,
        irp_alignment=0.5,
        reference_price=65000.0,
        target_size=0.01,
        leverage=2.0,
        bars_held=0,
        stage=LegacyTradeStage.ORDER_REQUESTED,
        metadata={},
    )


def _make_test_intent(
    action: DecisionAction = DecisionAction.ENTER,
    side: LegacyTradeSide = LegacyTradeSide.SHORT,
) -> Intent:
    return Intent(
        timestamp=datetime.now(timezone.utc),
        trade_id="test-trade-001",
        decision_id="test-decision-001",
        asset="BTCUSDT",
        action=action,
        side=side,
        reason="test",
        target_size=0.01,
        leverage=2.0,
        reference_price=65000.0,
        confidence=0.8,
        bars_held=0,
        stage=LegacyTradeStage.INTENT_CREATED,
        exit_leg_ratios=(0.5, 1.0),
        metadata={"entry_velocity_divergence": -0.03},
    )


class TestDecisionToKernelIntent(unittest.TestCase):
    """Verify every DecisionAction maps to the correct KernelCommandType."""

    maxDiff = None

    def test_enter_maps_to_enter(self):
        decision = _make_test_decision(DecisionAction.ENTER)
        intent = _make_test_intent(DecisionAction.ENTER)
        ki = _decision_to_kernel_intent(decision, intent, slot_id=0)
        self.assertEqual(ki.action, KernelCommandType.ENTER)
        self.assertEqual(ki.slot_id, 0)
        self.assertEqual(ki.trade_id, "test-trade-001")
        self.assertEqual(ki.asset, "BTCUSDT")
        self.assertEqual(ki.side, DitaTradeSide.SHORT)
        self.assertEqual(ki.reference_price, 65000.0)
        self.assertEqual(ki.target_size, 0.01)
        self.assertEqual(ki.leverage, 2.0)
        self.assertEqual(ki.exit_leg_ratios, (0.5, 1.0))

    def test_exit_maps_to_exit(self):
        decision = _make_test_decision(DecisionAction.EXIT)
        intent = _make_test_intent(DecisionAction.EXIT)
        ki = _decision_to_kernel_intent(decision, intent, slot_id=0)
        self.assertEqual(ki.action, KernelCommandType.EXIT)

    def test_hold_maps_to_mark_price(self):
        decision = _make_test_decision(DecisionAction.HOLD)
        intent = _make_test_intent(DecisionAction.HOLD)
        ki = _decision_to_kernel_intent(decision, intent, slot_id=0)
        self.assertEqual(ki.action, KernelCommandType.MARK_PRICE)

    def test_side_long_maps_correctly(self):
        decision = _make_test_decision(DecisionAction.ENTER, LegacyTradeSide.LONG)
        intent = _make_test_intent(DecisionAction.ENTER, LegacyTradeSide.LONG)
        ki = _decision_to_kernel_intent(decision, intent, slot_id=0)
        self.assertEqual(ki.side, DitaTradeSide.LONG)

    def test_side_short_maps_correctly(self):
        decision = _make_test_decision(DecisionAction.ENTER, LegacyTradeSide.SHORT)
        intent = _make_test_intent(DecisionAction.ENTER, LegacyTradeSide.SHORT)
        ki = _decision_to_kernel_intent(decision, intent, slot_id=0)
        self.assertEqual(ki.side, DitaTradeSide.SHORT)

    def test_metadata_is_preserved(self):
        decision = _make_test_decision()
        intent = _make_test_intent()
        intent.metadata["exit_ratio"] = 0.5
        ki = _decision_to_kernel_intent(decision, intent, slot_id=0)
        self.assertEqual(ki.metadata.get("exit_ratio"), 0.5)
        self.assertEqual(ki.metadata.get("entry_velocity_divergence"), -0.03)

    def test_slot_id_passthrough(self):
        decision = _make_test_decision()
        intent = _make_test_intent()
        ki = _decision_to_kernel_intent(decision, intent, slot_id=5)
        self.assertEqual(ki.slot_id, 5)


if __name__ == "__main__":
    unittest.main()