hjnormey/sentiment-engine
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

exec_unified: add placer.py (SmartPlacer pre_submit seam) + mutation-litmus tests

Browse Source 
- MarketSnapshot: best_bid, best_ask, spread_bps, tick, step (Decimal, frozen, validated)
- PlacementPlan: limit_price (quantized to tick), post_only=True
- pre_submit: returns None for TAKER, spread gate failure, or quantize-cross; otherwise PlacementPlan
- Spread gate uses live spread_bps (fixes dead _spread_allows_maker)
- Quantize to tick BEFORE returning (ROUND_HALF_EVEN per §4-15)
- Cross-after-quantize guard rejects if tick quantization crosses book
- 33 mutation-litmus tests: spread-gate flip, TAKER gate, cross-after-quantize all go RED
- Router.PROTECT wants_placement=True (at touch)
- Pure stdlib+Decimal, zero I/O/venue
This commit is contained in:
Codex
2026-07-14 23:14:07 +02:00
parent adba9f3fe8
commit d2e6d78abd
3 changed files with 695 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

566
prod/exec_unified/test_placer.py Normal file
View File

@@ -0,0 +1,566 @@
"""Mutation-litmus tests for SmartPlacer (spec §14, §4-12, §4-15, §4-16).
Run with: /home/dolphin/siloqy_env/bin/python3 -m pytest prod/exec_unified/test_placer.py -q
"""
from __future__ import annotations
import pytest
from decimal import Decimal
from prod.exec_unified.contract import (
ExecutionRequest,
UrgencyClass,
Side,
ExecutionAdvice,
ProtectiveSpec,
)
from prod.exec_unified.router import decide
from prod.exec_unified.placer import (
MarketSnapshot,
PlacementPlan,
pre_submit,
MAX_SPREAD_BPS_FOR_MAKER,
_quantize_to_tick,
)
class TestMarketSnapshot:
"""MarketSnapshot validation — frozen, validated, illegal states unrepresentable."""
def test_valid_snapshot(self):
snap = MarketSnapshot(
best_bid=Decimal("100.0"),
best_ask=Decimal("100.1"),
spread_bps=Decimal("1.0"),
tick=Decimal("0.1"),
step=Decimal("0.0001"),
)
assert snap.best_bid == Decimal("100.0")
assert snap.best_ask == Decimal("100.1")
assert snap.spread_bps == Decimal("1.0")
assert snap.tick == Decimal("0.1")
assert snap.step == Decimal("0.0001")
def test_snapshot_rejects_invalid_best_bid(self):
with pytest.raises(ValueError, match="best_bid must be > 0"):
MarketSnapshot(
best_bid=Decimal("0"),
best_ask=Decimal("100.1"),
spread_bps=Decimal("1.0"),
tick=Decimal("0.1"),
step=Decimal("0.0001"),
)
def test_snapshot_rejects_invalid_best_ask(self):
with pytest.raises(ValueError, match="best_ask must be > 0"):
MarketSnapshot(
best_bid=Decimal("100.0"),
best_ask=Decimal("-1"),
spread_bps=Decimal("1.0"),
tick=Decimal("0.1"),
step=Decimal("0.0001"),
)
def test_snapshot_rejects_crossed_book(self):
with pytest.raises(ValueError, match="best_ask.*must be > best_bid"):
MarketSnapshot(
best_bid=Decimal("100.1"),
best_ask=Decimal("100.0"),
spread_bps=Decimal("1.0"),
tick=Decimal("0.1"),
step=Decimal("0.0001"),
)
def test_snapshot_rejects_negative_spread_bps(self):
with pytest.raises(ValueError, match="spread_bps must be >= 0"):
MarketSnapshot(
best_bid=Decimal("100.0"),
best_ask=Decimal("100.1"),
spread_bps=Decimal("-1"),
tick=Decimal("0.1"),
step=Decimal("0.0001"),
)
def test_snapshot_rejects_zero_tick(self):
with pytest.raises(ValueError, match="tick must be > 0"):
MarketSnapshot(
best_bid=Decimal("100.0"),
best_ask=Decimal("100.1"),
spread_bps=Decimal("1.0"),
tick=Decimal("0"),
step=Decimal("0.0001"),
)
def test_snapshot_rejects_zero_step(self):
with pytest.raises(ValueError, match="step must be > 0"):
MarketSnapshot(
best_bid=Decimal("100.0"),
best_ask=Decimal("100.1"),
spread_bps=Decimal("1.0"),
tick=Decimal("0.1"),
step=Decimal("0"),
)
def test_snapshot_mid_price(self):
snap = MarketSnapshot(
best_bid=Decimal("100.0"),
best_ask=Decimal("100.2"),
spread_bps=Decimal("2.0"),
tick=Decimal("0.1"),
step=Decimal("0.0001"),
)
assert snap.mid_price == Decimal("100.1")
def test_snapshot_spread(self):
snap = MarketSnapshot(
best_bid=Decimal("100.0"),
best_ask=Decimal("100.2"),
spread_bps=Decimal("2.0"),
tick=Decimal("0.1"),
step=Decimal("0.0001"),
)
assert snap.spread == Decimal("0.2")
class TestPlacementPlan:
"""PlacementPlan validation."""
def test_valid_plan(self):
plan = PlacementPlan(limit_price=Decimal("100.0"), post_only=True)
assert plan.limit_price == Decimal("100.0")
assert plan.post_only is True
def test_plan_rejects_zero_price(self):
with pytest.raises(ValueError, match="limit_price must be > 0"):
PlacementPlan(limit_price=Decimal("0"), post_only=True)
def test_plan_rejects_negative_price(self):
with pytest.raises(ValueError, match="limit_price must be > 0"):
PlacementPlan(limit_price=Decimal("-1"), post_only=True)
class TestQuantizeToTick:
"""_quantize_to_tick uses banker's rounding (ROUND_HALF_EVEN)."""
def test_quantize_exact_tick(self):
assert _quantize_to_tick(Decimal("100.0"), Decimal("0.1")) == Decimal("100.0")
def test_quantize_half_even_down(self):
# 100.05 with tick 0.1 -> 100.0 (half-even rounds to even)
assert _quantize_to_tick(Decimal("100.05"), Decimal("0.1")) == Decimal("100.0")
def test_quantize_half_even_up(self):
# 100.15 with tick 0.1 -> 100.2 (half-even rounds to even)
assert _quantize_to_tick(Decimal("100.15"), Decimal("0.1")) == Decimal("100.2")
def test_quantize_below_tick(self):
assert _quantize_to_tick(Decimal("100.04"), Decimal("0.1")) == Decimal("100.0")
def test_quantize_above_tick(self):
assert _quantize_to_tick(Decimal("100.06"), Decimal("0.1")) == Decimal("100.1")
def test_quantize_rejects_zero_tick(self):
with pytest.raises(ValueError, match="tick must be > 0"):
_quantize_to_tick(Decimal("100.0"), Decimal("0"))
class TestPreSubmit:
"""pre_submit — the SmartPlacer seam.
Mutation-litmus tests: each gate (TAKER, spread, quantize) must go RED
when the gate logic is flipped.
"""
def _make_buy_request(self, urgency: UrgencyClass, **kw) -> ExecutionRequest:
base = dict(
request_id="test-1",
asset="BTCUSDT",
side=Side.BUY,
size=Decimal("0.001"),
urgency=urgency,
)
base.update(kw)
return ExecutionRequest(**base)
def _make_sell_request(self, urgency: UrgencyClass, **kw) -> ExecutionRequest:
base = dict(
request_id="test-1",
asset="BTCUSDT",
side=Side.SELL,
size=Decimal("0.001"),
urgency=urgency,
)
base.update(kw)
return ExecutionRequest(**base)
def _make_snapshot(
self, spread_bps: Decimal = Decimal("1.0"), tick: Decimal = Decimal("0.1")
) -> MarketSnapshot:
return MarketSnapshot(
best_bid=Decimal("64500.0"),
best_ask=Decimal("64501.0"),
spread_bps=spread_bps,
tick=tick,
step=Decimal("0.0001"),
)
# ---- TAKER gate ----
def test_pre_submit_returns_none_for_taker_catastrophic(self):
"""CATASTROPHIC is TAKER — no placement (spec §6, Router.decide)."""
req = self._make_buy_request(UrgencyClass.CATASTROPHIC)
decision = decide(req)
snap = self._make_snapshot()
assert decision.method.name == "TAKER"
assert pre_submit(req, decision, snap) is None
def test_pre_submit_returns_none_for_any_taker_decision(self):
"""Any TAKER decision returns None — not just CATASTROPHIC."""
from prod.exec_unified.router import RoutingDecision, ExecutionMethod, TtlDiscipline
req = self._make_buy_request(UrgencyClass.ACQUIRE)
# Force a TAKER decision (simulating a mutation of Router)
decision = RoutingDecision(
method=ExecutionMethod.TAKER,
max_reprices=0,
ttl=TtlDiscipline.IMMEDIATE,
cross_on_expiry=True,
wants_placement=True,
ignore_advice=True,
)
snap = self._make_snapshot()
assert pre_submit(req, decision, snap) is None
# ---- Spread gate ----
def test_pre_submit_returns_none_when_spread_exceeds_max(self):
"""Spread gate: live spread_bps > MAX_SPREAD_BPS_FOR_MAKER -> None (§4-12)."""
req = self._make_buy_request(UrgencyClass.ACQUIRE)
decision = decide(req)
# Spread of 10 bps > MAX (8 bps provisional)
snap = self._make_snapshot(spread_bps=Decimal("10.0"))
assert decision.method.name == "MAKER"
assert decision.wants_placement is True
assert pre_submit(req, decision, snap) is None
def test_pre_submit_succeeds_when_spread_within_max(self):
"""Spread gate passes when spread_bps <= MAX."""
req = self._make_buy_request(UrgencyClass.ACQUIRE)
decision = decide(req)
# Spread of 1 bps < MAX (8 bps)
snap = self._make_snapshot(spread_bps=Decimal("1.0"))
assert decision.method.name == "MAKER"
assert decision.wants_placement is True
plan = pre_submit(req, decision, snap)
assert plan is not None
assert isinstance(plan, PlacementPlan)
def test_spread_gate_uses_live_spread_bps_not_vibe(self):
"""The spread gate MUST be fed live spread_bps — no dead code path (§4-12)."""
req = self._make_buy_request(UrgencyClass.ACQUIRE)
decision = decide(req)
# Even with MAKER decision and wants_placement=True,
# a bad spread must reject
snap_bad = self._make_snapshot(spread_bps=Decimal("100.0")) # absurdly wide
assert pre_submit(req, decision, snap_bad) is None
# Good spread passes
snap_good = self._make_snapshot(spread_bps=Decimal("0.5"))
assert pre_submit(req, decision, snap_good) is not None
# ---- Quantization ----
def test_pre_submit_quantizes_price_to_tick(self):
"""limit_price quantized to venue tick BEFORE returning (§4-15)."""
# tick = 0.1, best_bid = 64500.05 -> quantized to 64500.0 (half-even)
req = self._make_buy_request(UrgencyClass.ACQUIRE)
decision = decide(req)
snap = MarketSnapshot(
best_bid=Decimal("64500.05"),
best_ask=Decimal("64501.05"),
spread_bps=Decimal("1.0"),
tick=Decimal("0.1"),
step=Decimal("0.0001"),
)
plan = pre_submit(req, decision, snap)
assert plan is not None
# BUY at best_bid (64500.05) quantized to tick 0.1 = 64500.0
assert plan.limit_price == Decimal("64500.0")
def test_pre_submit_sell_quantizes_to_tick(self):
"""SELL at best_ask quantized to tick."""
req = self._make_sell_request(UrgencyClass.ACQUIRE)
decision = decide(req)
snap = MarketSnapshot(
best_bid=Decimal("64500.0"),
best_ask=Decimal("64501.05"),
spread_bps=Decimal("1.0"),
tick=Decimal("0.1"),
step=Decimal("0.0001"),
)
plan = pre_submit(req, decision, snap)
assert plan is not None
# SELL at best_ask (64501.05) quantized to tick 0.1 = 64501.0
assert plan.limit_price == Decimal("64501.0")
def test_pre_submit_buy_touch_quantized(self):
"""BUY at touch (best_bid) quantized correctly."""
req = self._make_buy_request(UrgencyClass.ACQUIRE)
decision = decide(req)
# Exact tick alignment
snap = MarketSnapshot(
best_bid=Decimal("64500.0"),
best_ask=Decimal("64501.0"),
spread_bps=Decimal("1.0"),
tick=Decimal("0.1"),
step=Decimal("0.0001"),
)
plan = pre_submit(req, decision, snap)
assert plan is not None
assert plan.limit_price == Decimal("64500.0")
def test_pre_submit_sell_touch_quantized(self):
"""SELL at touch (best_ask) quantized correctly."""
req = self._make_sell_request(UrgencyClass.ACQUIRE)
decision = decide(req)
snap = MarketSnapshot(
best_bid=Decimal("64500.0"),
best_ask=Decimal("64501.0"),
spread_bps=Decimal("1.0"),
tick=Decimal("0.1"),
step=Decimal("0.0001"),
)
plan = pre_submit(req, decision, snap)
assert plan is not None
assert plan.limit_price == Decimal("64501.0")
# ---- Post-only ----
def test_pre_submit_plan_always_post_only(self):
"""GTX/post-only is the ONLY certified technique (spec §4-16)."""
req = self._make_buy_request(UrgencyClass.ACQUIRE)
decision = decide(req)
snap = self._make_snapshot()
plan = pre_submit(req, decision, snap)
assert plan is not None
assert plan.post_only is True
# ---- Cross-side protection after quantization ----
def test_pre_submit_rejects_buy_crossed_after_quantize(self):
"""Quantization could push BUY price >= best_ask -> reject."""
req = self._make_buy_request(UrgencyClass.ACQUIRE)
decision = decide(req)
# best_bid=64500.95, best_ask=64501.00, tick=0.1
# quantized BUY at best_bid -> 64501.0 which crosses
snap = MarketSnapshot(
best_bid=Decimal("64500.95"),
best_ask=Decimal("64501.00"),
spread_bps=Decimal("0.5"),
tick=Decimal("0.1"),
step=Decimal("0.0001"),
)
plan = pre_submit(req, decision, snap)
assert plan is None # quantized price crosses book
def test_pre_submit_rejects_sell_crossed_after_quantize(self):
"""Quantization could push SELL price <= best_bid -> reject."""
req = self._make_sell_request(UrgencyClass.ACQUIRE)
decision = decide(req)
# best_bid=64500.00, best_ask=64500.05, tick=0.1
# quantized SELL at best_ask -> 64500.0 which crosses
snap = MarketSnapshot(
best_bid=Decimal("64500.00"),
best_ask=Decimal("64500.05"),
spread_bps=Decimal("0.5"),
tick=Decimal("0.1"),
step=Decimal("0.0001"),
)
plan = pre_submit(req, decision, snap)
assert plan is None # quantized price crosses book
# ---- Mutation-litmus: these tests MUST go RED if the gate logic is inverted ----
class TestMutationLitmus:
"""Mutation litmus — each test proves the gate exists and flips to RED.
To verify: invert the condition in placer.py and run — test must fail (RED).
"""
def test_spread_gate_mutation_red(self):
"""If spread gate is inverted (> -> <=), this goes RED."""
req = ExecutionRequest(
request_id="m1",
asset="BTCUSDT",
side=Side.BUY,
size=Decimal("0.001"),
urgency=UrgencyClass.ACQUIRE,
)
decision = decide(req)
# spread = 10 bps > MAX (8 bps) -> should return None
snap = MarketSnapshot(
best_bid=Decimal("100.0"),
best_ask=Decimal("101.0"),
spread_bps=Decimal("10.0"),
tick=Decimal("0.1"),
step=Decimal("0.0001"),
)
# Current logic: returns None for spread > MAX
assert pre_submit(req, decision, snap) is None
def test_taker_gate_mutation_red(self):
"""If TAKER gate is inverted, this goes RED."""
from prod.exec_unified.router import RoutingDecision, ExecutionMethod, TtlDiscipline
req = ExecutionRequest(
request_id="m2",
asset="BTCUSDT",
side=Side.BUY,
size=Decimal("0.001"),
urgency=UrgencyClass.ACQUIRE,
)
# Forced TAKER
decision = RoutingDecision(
method=ExecutionMethod.TAKER,
max_reprices=0,
ttl=TtlDiscipline.IMMEDIATE,
cross_on_expiry=True,
wants_placement=True,
ignore_advice=True,
)
snap = MarketSnapshot(
best_bid=Decimal("100.0"),
best_ask=Decimal("100.1"),
spread_bps=Decimal("1.0"),
tick=Decimal("0.1"),
step=Decimal("0.0001"),
)
# Current logic: returns None for TAKER
assert pre_submit(req, decision, snap) is None
def test_quantize_crosses_mutation_red(self):
"""If cross-after-quantize check is removed, this goes RED."""
req = ExecutionRequest(
request_id="m3",
asset="BTCUSDT",
side=Side.BUY,
size=Decimal("0.001"),
urgency=UrgencyClass.ACQUIRE,
)
decision = decide(req)
# best_bid quantizes to >= best_ask
snap = MarketSnapshot(
best_bid=Decimal("64500.95"),
best_ask=Decimal("64501.00"),
spread_bps=Decimal("0.5"),
tick=Decimal("0.1"),
step=Decimal("0.0001"),
)
# Current logic: returns None when quantized price crosses
assert pre_submit(req, decision, snap) is None
def test_protect_maker_wants_placement(self):
"""PROTECT is MAKER and wants placement."""
req = ExecutionRequest(
request_id="m4",
asset="BTCUSDT",
side=Side.BUY,
size=Decimal("0.001"),
urgency=UrgencyClass.PROTECT,
)
decision = decide(req)
snap = MarketSnapshot(
best_bid=Decimal("100.0"),
best_ask=Decimal("100.1"),
spread_bps=Decimal("1.0"),
tick=Decimal("0.1"),
step=Decimal("0.0001"),
)
assert decision.method.name == "MAKER"
assert decision.wants_placement is True
plan = pre_submit(req, decision, snap)
assert plan is not None
assert plan.limit_price == Decimal("100.0")
def test_harvest_maker_wants_placement(self):
"""HARVEST is MAKER and wants placement."""
req = ExecutionRequest(
request_id="m5",
asset="BTCUSDT",
side=Side.SELL,
size=Decimal("0.001"),
urgency=UrgencyClass.HARVEST,
)
decision = decide(req)
snap = MarketSnapshot(
best_bid=Decimal("100.0"),
best_ask=Decimal("100.1"),
spread_bps=Decimal("1.0"),
tick=Decimal("0.1"),
step=Decimal("0.0001"),
)
assert decision.method.name == "MAKER"
assert decision.wants_placement is True
plan = pre_submit(req, decision, snap)
assert plan is not None
assert plan.limit_price == Decimal("100.1")
def test_rotate_maker_wants_placement(self):
"""ROTATE is MAKER and wants placement."""
req = ExecutionRequest(
request_id="m6",
asset="BTCUSDT",
side=Side.BUY,
size=Decimal("0.001"),
urgency=UrgencyClass.ROTATE,
deadline_ms=60_000,
)
decision = decide(req)
snap = MarketSnapshot(
best_bid=Decimal("100.0"),
best_ask=Decimal("100.1"),
spread_bps=Decimal("1.0"),
tick=Decimal("0.1"),
step=Decimal("0.0001"),
)
assert decision.method.name == "MAKER"
assert decision.wants_placement is True
plan = pre_submit(req, decision, snap)
assert plan is not None
assert plan.limit_price == Decimal("100.0")
def test_acquire_maker_wants_placement(self):
"""ACQUIRE is MAKER and wants placement (abandons, never crosses)."""
req = ExecutionRequest(
request_id="m7",
asset="BTCUSDT",
side=Side.SELL,
size=Decimal("0.001"),
urgency=UrgencyClass.ACQUIRE,
)
decision = decide(req)
snap = MarketSnapshot(
best_bid=Decimal("100.0"),
best_ask=Decimal("100.1"),
spread_bps=Decimal("1.0"),
tick=Decimal("0.1"),
step=Decimal("0.0001"),
)
assert decision.method.name == "MAKER"
assert decision.wants_placement is True
plan = pre_submit(req, decision, snap)
assert plan is not None
assert plan.limit_price == Decimal("100.1")