"""The Unified Execution Layer contract — "an asset, a size, and a prayer". This is the WHOLE input surface (spec §2.1). Nothing about the caller's world may leak in: not leverage math, not postures, not capital, not why-now. Every brain the factory produces (BLUE, UV/BLUE-PRIME, VIOLET, VIBRASS children, MALKHUT) executes through the same body by speaking only this vocabulary. Pure stdlib + Decimal. Zero I/O, zero venue knowledge. V-TYPES discipline: frozen, validated at construction, illegal states unrepresentable. Provenance: prod/docs/SPEC_UNIFIED_EXEC_LAYER_20260714.md §2, §6. Home-to-be: prod/clean_arch/dita_v2/exec/ (vendored). Lives standalone until the vendor flow relocates it — do not import from the vendored dita_v2 tree yet. """ from __future__ import annotations from dataclasses import dataclass, field from decimal import Decimal from enum import Enum class Side(Enum): BUY = "BUY" SELL = "SELL" class UrgencyClass(Enum): """The caller's declaration of *why* — the only "why" this layer ever learns (§6). Ordered most-urgent → least. The ordinal is meaningful: dispatch priority and the "never delay a more-urgent order" invariant read it (spec §9 priority ladder). """ CATASTROPHIC = 0 # SL / kill / liquidation — taker MARKET now, no cleverness ever PROTECT = 1 # TP_FLOOR give-back, ADVSL retract — maker@touch, 1 reprice, then cross HARVEST = 2 # FIXED_TP — maker at target ± offset, chase ≤N, give up on regression ROTATE = 3 # MAX_HOLD / admin — patient maker, TTL from deadline_ms, cross at TTL ACQUIRE = 4 # ENTER — patient maker inside spread, abandon (never chase) @property def is_exit(self) -> bool: """Exit-side urgencies decrease a position; entries (ACQUIRE) increase it. Exits may only ever tighten toward urgency, never reprice away (spec §4-2). """ return self in (UrgencyClass.CATASTROPHIC, UrgencyClass.PROTECT, UrgencyClass.HARVEST, UrgencyClass.ROTATE) class SGrade(Enum): """Size axis — orthogonal to the T-tier smartness ladder (spec §15.2). Computed at intake from Appendix E's power-law depth model (per-asset, stress-adjusted), or supplied by the caller. Below T14 the layer executes S0/S1 directly and REFUSES S2+ whole (the pain-fence) so nobody meets the large-size pain class by accident. """ S0 = 0 # size << top-of-book depth (walk < 1 tick) — any tier executes directly S1 = 1 # walks visible levels (expected walk >= spread) — T2+ prices the walk S2 = 2 # exceeds book capacity in the urgency window — must be SLICED; refused below T14 S3 = 3 # size IS the market (footprint moves MM behaviour) — MALKHUT/VIBRASS territory @property def needs_slicing(self) -> bool: return self in (SGrade.S2, SGrade.S3) @dataclass(frozen=True) class ProtectiveSpec: """Attach-a-dead-man's-stop geometry (spec §10). Mark-based, rides the entry order. Either an explicit ``stop_price`` OR a ``stop_mult`` multiple of the caller's software SL distance — never both, never neither. The venue-attached STOP_MARKET fires only when the client is dead or blind (HZ silent-death lineage): parity- invisible, because it only acts once the software layer has already failed. """ stop_price: Decimal | None = None stop_mult: Decimal | None = None # × software-SL distance (e.g. 2.0) working_type: str = "MARK_PRICE" # venue geometry truth = mark, never last reduce_only: bool = True def __post_init__(self) -> None: has_px = self.stop_price is not None has_mult = self.stop_mult is not None if has_px == has_mult: raise ValueError("ProtectiveSpec needs exactly one of stop_price / stop_mult") if has_px and self.stop_price <= 0: raise ValueError(f"stop_price must be positive, got {self.stop_price}") if has_mult and self.stop_mult <= 0: raise ValueError(f"stop_mult must be positive, got {self.stop_mult}") if self.working_type != "MARK_PRICE": # Geometry truth is the mark (spec §9 three-price-truths). Guardrail, not law. raise ValueError("protective working_type must be MARK_PRICE") @dataclass(frozen=True) class ExecutionAdvice: """MALKHUT/other advisor hints — an advisor, never an authority (spec §5, §11-L11). The layer may take these hints or ignore them; CATASTROPHIC ignores them *by law*. Advice can never change WHAT was asked (side/size/asset/reduce_only are sacred). """ source: str = "unknown" # who advised (provenance, non-authoritative) prefer_post_only: bool | None = None # hint: rest as maker if possible qty_fraction: Decimal | None = None # hint: work this fraction per slice (slicer above) note: str = "" def __post_init__(self) -> None: if self.qty_fraction is not None and not (Decimal(0) < self.qty_fraction <= Decimal(1)): raise ValueError(f"qty_fraction must be in (0,1], got {self.qty_fraction}") @dataclass(frozen=True) class ExecutionRequest: """The whole surface. Nothing else may leak in (spec §2.1). ``guideline_price`` is a decision-time *reference*, NOT a limit — it resolves the ADVSL stale-price concern: scan-clock exits pass the price that triggered them; the drive loop executes against the live book, never against this frozen number. """ request_id: str # caller-minted; idempotency key end-to-end asset: str # canonical undashed ("BTCUSDT"); dialect dashes it side: Side size: Decimal # base quantity; sizing is the caller's solved problem urgency: UrgencyClass # assigned by CALLER, NEVER inferred here reduce_only: bool = False # position-decreasing intent (exits set True) guideline_price: Decimal | None = None # reference, not a limit protective: ProtectiveSpec | None = None advice: ExecutionAdvice | None = None deadline_ms: int | None = None # caller's patience budget (ROTATE/ACQUIRE) s_grade: SGrade = SGrade.S0 # size regime (§15.2); default S0 = fits at touch parent_request_id: str | None = None # slicer family linkage (§15.2 T*.s), telemetry only def __post_init__(self) -> None: if not self.request_id: raise ValueError("request_id is mandatory (idempotency key)") if self.asset != self.asset.upper() or "-" in self.asset: raise ValueError(f"asset must be canonical undashed upper, got {self.asset!r}") if self.size <= 0: raise ValueError(f"size must be positive, got {self.size}") if self.guideline_price is not None and self.guideline_price <= 0: raise ValueError(f"guideline_price must be positive, got {self.guideline_price}") if self.deadline_ms is not None and self.deadline_ms <= 0: raise ValueError(f"deadline_ms must be positive, got {self.deadline_ms}") # Contract invariant: an entry (increases position) can't be reduce_only. if self.reduce_only and self.urgency is UrgencyClass.ACQUIRE: raise ValueError("ACQUIRE (entry) cannot be reduce_only") def client_order_id_core(self, attempt: int = 0) -> str: """Venue-clientOrderId core, UNIQUE PER ATTEMPT (audit H4 / FIX ClOrdID law). A retry after an INDETERMINATE submit MUST carry a *fresh* venue id — exchanges reject a duplicate clientOrderId (idempotency-key rule) — while staying linkable to the parent request (FIX ``OrigClOrdID``). The attempt counter provides exactly that: attempt 0 is the original, 1+ are retries; the parent is recoverable by stripping the trailing ``-``. The dialect layer (§11) prepends the venue prefix (``u-``/ ``m-``) and enforces the venue's charset + length cap (BingX caps clientOrderId length) — that layer owns venue legality, this owns per-attempt uniqueness. """ if attempt < 0: raise ValueError(f"attempt cannot be negative, got {attempt}") return f"{self.request_id}-{attempt}"