"""IntentSpec → dita_v2 KernelIntent — the ONE module allowed to import dita_v2 (spec §7). ``kernel_port.KernelExecPort`` takes an INJECTED ``to_kernel_intent`` so the port (and the whole pure engine) never imports the vendored kernel and stays local-testable against a fake. This module supplies the real translator. The dita_v2 import is LAZY (inside the factory) so merely importing this module never drags in dita_v2 — you pay the dependency only when you actually wire a live kernel. THREE mappings, each VERIFIED against the vendored kernel source (not assumed — a wrong branch here fires real wrong-direction / wrong-size orders): * SIDE (bingx_venue.py:627-628 `_legacy_intent`): ``KernelIntent.side`` is the POSITION side (TradeSide.LONG/SHORT); the venue BUY/SELL is derived DOWNSTREAM from (side, action). Our engine speaks ORDER side (Side.BUY/SELL), so the translator inverts on the reducing leg: ENTER BUY→LONG SELL→SHORT (opening: order side == position direction) EXIT BUY→SHORT SELL→LONG (reduceOnly: order side is opposite the position) Truth table is exhaustively + mutation-tested. Getting one cell wrong = catastrophic. * SIZE / LEVERAGE (rust_backend.py:448, mock_venue.py:60/88/168/198): the submitted order quantity is ``float(intent.target_size)`` VERBATIM — there is NO ``target_size × leverage`` in the quantity path (leverage rides separately for margin/accounting). So target_size = spec.size and leverage = 1.0 is correct AND leverage-agnostic (§2.3). The "×leverage poison" in [[ditav2_kernel_audit]] was in the capital/PnL path, not the order-quantity path. * CANCEL side (rust_backend.py:897-909): a CANCEL resolves its target via the slot's active_entry/exit order by slot_id — ``intent.side`` is NOT consulted. So CANCEL.side is informational-only; we set it by the reducing-leg convention and it changes nothing. Pure translation. The datetime source is INJECTED (no ambient clock — §NFR). """ from __future__ import annotations from datetime import datetime from typing import Any, Callable from .contract import Side from .kernel_port import IntentSpec # One-way flatten: single slot, unit leverage (size is the final order quantity). _SLOT_ID = 0 _UNIT_LEVERAGE = 1.0 def make_kernel_intent_translator( clock: Callable[[], datetime], *, slot_id: int = _SLOT_ID, leverage: float = _UNIT_LEVERAGE, ) -> Callable[[IntentSpec], Any]: """Build a ``to_kernel_intent(spec) -> KernelIntent`` bound to an injected datetime clock. dita_v2 is imported HERE, lazily, so the module stays importable without it. Tries the bare package name first, then the in-repo path, so it works whether ``clean_arch`` or the repo root is the import anchor.""" try: from dita_v2.contracts import KernelCommandType, KernelIntent, TradeSide except ImportError: # pragma: no cover - path depends on how the kernel is vendored from prod.clean_arch.dita_v2.contracts import ( # type: ignore KernelCommandType, KernelIntent, TradeSide, ) _ACTION = { "ENTER": KernelCommandType.ENTER, "EXIT": KernelCommandType.EXIT, "CANCEL": KernelCommandType.CANCEL, } def _position_side(order_side: Side, action_kind: str): """Order side (BUY/SELL) + action → POSITION side (LONG/SHORT). Verified truth table. ENTER is the only OPENING action: there the order side IS the position direction. Every other action (EXIT, CANCEL) is on the reducing leg, where the order side is the OPPOSITE of the position it closes. So: is_long = BUY on open, SELL on reduce.""" opening = action_kind == "ENTER" is_buy = order_side is Side.BUY is_long = is_buy if opening else not is_buy return TradeSide.LONG if is_long else TradeSide.SHORT def to_kernel_intent(spec: IntentSpec): action = _ACTION.get(spec.action_kind) if action is None: raise ValueError(f"unknown action_kind {spec.action_kind!r} (want ENTER|EXIT|CANCEL)") return KernelIntent( timestamp=clock(), intent_id=spec.request_id, trade_id=spec.base_request_id, slot_id=slot_id, asset=spec.asset, side=_position_side(spec.side, spec.action_kind), action=action, reference_price=float(spec.limit_price), target_size=float(spec.size), # VERBATIM order quantity — never ×leverage leverage=leverage, reason=spec.reason, order_type=spec.order_type, # "LIMIT" | "MARKET" (matches KernelIntent default) limit_price=float(spec.limit_price), # 0 for MARKET (spec builds it that way) metadata={ "tif": spec.tif, # "PostOnly" | "GTC" — venue TIF rides metadata "reduce_only": spec.reduce_only, "exec_unified": True, # provenance: this intent came from the unified layer }, ) return to_kernel_intent