5abe0cc6e0
intent_translator.py: the injected to_kernel_intent KernelExecPort needs, wiring the pure engine to the live DITAv2 kernel. Side/size/cancel mappings each verified against vendored source (bingx_venue:627, rust_backend:448/897); EXIT inversion mutation-verified RED. Tested against REAL dita_v2 (importorskip). Lazy dita_v2 import keeps the rest of the package clean. 143 tests green. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
120 lines
5.3 KiB
Python
120 lines
5.3 KiB
Python
"""IntentSpec → real dita_v2 KernelIntent translation (spec §7). Runs against the ACTUAL
|
||
vendored kernel contracts (not fakes) so the field mapping is verified where it matters.
|
||
|
||
The side truth table is the load-bearing test: a single wrong cell = a real wrong-direction
|
||
order. Mutation: drop the `not is_buy` inversion in _position_side -> the EXIT rows go RED.
|
||
|
||
dita_v2 lives under prod/clean_arch; add it to the path if the repo root isn't the anchor.
|
||
If it truly can't be imported (a box without the kernel), skip loudly rather than fake it.
|
||
|
||
Run: /home/dolphin/siloqy_env/bin/python3 -m pytest prod/exec_unified/test_intent_translator.py -q
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import sys
|
||
from datetime import datetime
|
||
from decimal import Decimal
|
||
|
||
import pytest
|
||
|
||
# Anchor dita_v2: try the repo's clean_arch dir (works both at /mnt and /root/dev copies).
|
||
for _cand in ("/mnt/dolphinng5_predict/prod/clean_arch",
|
||
os.path.join(os.path.dirname(__file__), "..", "clean_arch")):
|
||
if os.path.isdir(_cand) and _cand not in sys.path:
|
||
sys.path.insert(0, _cand)
|
||
|
||
dita = pytest.importorskip("dita_v2.contracts", reason="vendored dita_v2 not importable on this box")
|
||
TradeSide = dita.TradeSide
|
||
KernelCommandType = dita.KernelCommandType
|
||
|
||
from prod.exec_unified.contract import Side
|
||
from prod.exec_unified.intent_translator import make_kernel_intent_translator
|
||
from prod.exec_unified.kernel_port import IntentSpec
|
||
|
||
FIXED_TS = datetime(2026, 7, 15, 12, 0, 0)
|
||
|
||
|
||
def _xlate():
|
||
return make_kernel_intent_translator(lambda: FIXED_TS)
|
||
|
||
|
||
def _spec(**kw):
|
||
base = dict(request_id="req-0", base_request_id="pos-1", asset="BTCUSDT", side=Side.BUY,
|
||
action_kind="ENTER", order_type="LIMIT", limit_price=Decimal("100.5"),
|
||
size=Decimal("0.001"), tif="PostOnly", reduce_only=False, reason="test")
|
||
base.update(kw)
|
||
return IntentSpec(**base)
|
||
|
||
|
||
# ── THE side truth table (verified vs bingx_venue._legacy_intent) ─────────────
|
||
|
||
@pytest.mark.parametrize("action,order_side,expect_pos", [
|
||
("ENTER", Side.BUY, TradeSide.LONG), # open long = buy
|
||
("ENTER", Side.SELL, TradeSide.SHORT), # open short = sell
|
||
("EXIT", Side.SELL, TradeSide.LONG), # close long = sell (INVERTED)
|
||
("EXIT", Side.BUY, TradeSide.SHORT), # close short = buy (INVERTED)
|
||
])
|
||
def test_position_side_truth_table(action, order_side, expect_pos):
|
||
reduce = action != "ENTER"
|
||
ki = _xlate()(_spec(action_kind=action, side=order_side, reduce_only=reduce,
|
||
order_type="MARKET" if reduce else "LIMIT",
|
||
limit_price=Decimal("0") if reduce else Decimal("100.5")))
|
||
assert ki.side is expect_pos
|
||
# Mutation: `is_long = is_buy` (drop the inversion) -> the two EXIT rows fail.
|
||
|
||
|
||
# ── size / leverage: quantity is verbatim, never levered ─────────────────────
|
||
|
||
def test_target_size_is_verbatim_and_leverage_is_unit():
|
||
ki = _xlate()(_spec(size=Decimal("0.0731")))
|
||
assert ki.target_size == 0.0731 # the exact order quantity — no ×leverage
|
||
assert ki.leverage == 1.0 # unit: exec_unified is leverage-agnostic (§2.3)
|
||
|
||
|
||
def test_custom_leverage_never_scales_size():
|
||
# even with a non-unit leverage the quantity is untouched (leverage is accounting-only).
|
||
ki = make_kernel_intent_translator(lambda: FIXED_TS, leverage=5.0)(_spec(size=Decimal("2")))
|
||
assert ki.target_size == 2.0
|
||
assert ki.leverage == 5.0
|
||
|
||
|
||
# ── action + order_type + price passthrough ──────────────────────────────────
|
||
|
||
def test_action_mapping():
|
||
assert _xlate()(_spec(action_kind="ENTER")).action is KernelCommandType.ENTER
|
||
assert _xlate()(_spec(action_kind="EXIT", reduce_only=True)).action is KernelCommandType.EXIT
|
||
assert _xlate()(_spec(action_kind="CANCEL")).action is KernelCommandType.CANCEL
|
||
|
||
|
||
def test_unknown_action_raises():
|
||
with pytest.raises(ValueError):
|
||
_xlate()(_spec(action_kind="FROBNICATE"))
|
||
|
||
|
||
def test_maker_limit_carries_price_and_tif_metadata():
|
||
ki = _xlate()(_spec(order_type="LIMIT", limit_price=Decimal("64547.5"), tif="PostOnly"))
|
||
assert ki.order_type == "LIMIT"
|
||
assert ki.limit_price == 64547.5 and ki.reference_price == 64547.5
|
||
assert ki.metadata["tif"] == "PostOnly"
|
||
assert ki.metadata["reduce_only"] is False
|
||
assert ki.metadata["exec_unified"] is True
|
||
|
||
|
||
def test_market_carries_zero_price():
|
||
ki = _xlate()(_spec(action_kind="EXIT", side=Side.SELL, reduce_only=True,
|
||
order_type="MARKET", limit_price=Decimal("0")))
|
||
assert ki.order_type == "MARKET"
|
||
assert ki.limit_price == 0.0
|
||
|
||
|
||
# ── identity + clock ─────────────────────────────────────────────────────────
|
||
|
||
def test_ids_and_slot_and_injected_clock():
|
||
ki = _xlate()(_spec(request_id="req-3", base_request_id="pos-9"))
|
||
assert ki.intent_id == "req-3" # per-attempt id
|
||
assert ki.trade_id == "pos-9" # position identity
|
||
assert ki.slot_id == 0 # single-slot flatten
|
||
assert ki.timestamp == FIXED_TS # injected, deterministic — no ambient now()
|
||
assert ki.asset == "BTCUSDT" # undashed at kernel layer (dialect dashes at venue)
|