7fa06ced65
mechanics_verdict (PASS iff every post-rebalance position tracks its effective target within tolerance + no error/gap rows; else FAIL naming offending symbols) and indicative_slippage (per-tier mean slippage-vs- assumed excluding low_confidence fills, reporting coverage). Frozen DTOs PositionRow / SlippageFill / Verdict. Pure (design §5). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
83 lines
3.0 KiB
Python
83 lines
3.0 KiB
Python
"""Execution-gap verdict + indicative slippage (design §5) — the execution analog of the recon gate.
|
|
|
|
Pure functions over the recorded testnet rows:
|
|
* `mechanics_verdict` — MECHANICS PASS iff every recorded post-rebalance position tracks its effective
|
|
target within tolerance, with no error/gap rows; else FAIL naming the offending symbol(s).
|
|
* `indicative_slippage` — per cost-tier mean of `slippage_vs_assumed`, EXCLUDING `low_confidence` (thin-book)
|
|
fills, plus the coverage (real-liquidity fills / all fills in the tier).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from fxhnt.application.execution_gap import (
|
|
PositionRow,
|
|
SlippageFill,
|
|
indicative_slippage,
|
|
mechanics_verdict,
|
|
)
|
|
|
|
_TOL = 0.01
|
|
|
|
|
|
def test_pass_when_all_positions_track_targets():
|
|
rows = [PositionRow("BTCUSDT", 50.0), PositionRow("ETHUSDT", -10.0)]
|
|
v = mechanics_verdict(rows, {"BTCUSDT": 50.0, "ETHUSDT": -10.0}, _TOL)
|
|
assert v.passed is True
|
|
assert v.offending == ()
|
|
|
|
|
|
def test_pass_within_tolerance():
|
|
rows = [PositionRow("BTCUSDT", 50.005)]
|
|
v = mechanics_verdict(rows, {"BTCUSDT": 50.0}, _TOL)
|
|
assert v.passed is True
|
|
|
|
|
|
def test_fail_names_symbol_off_target():
|
|
rows = [PositionRow("BTCUSDT", 50.0), PositionRow("ETHUSDT", -5.0)]
|
|
v = mechanics_verdict(rows, {"BTCUSDT": 50.0, "ETHUSDT": -10.0}, _TOL)
|
|
assert v.passed is False
|
|
assert v.offending == ("ETHUSDT",)
|
|
|
|
|
|
def test_fail_on_error_row():
|
|
rows = [PositionRow("BTCUSDT", 50.0), PositionRow("DOGEUSDT", 0.0, error=True)]
|
|
v = mechanics_verdict(rows, {"BTCUSDT": 50.0, "DOGEUSDT": 100.0}, _TOL)
|
|
assert v.passed is False
|
|
assert "DOGEUSDT" in v.offending
|
|
|
|
|
|
def test_fail_on_missing_row_for_target():
|
|
rows = [PositionRow("BTCUSDT", 50.0)]
|
|
v = mechanics_verdict(rows, {"BTCUSDT": 50.0, "ETHUSDT": -10.0}, _TOL)
|
|
assert v.passed is False
|
|
assert v.offending == ("ETHUSDT",)
|
|
|
|
|
|
def test_indicative_slippage_aggregates_per_tier():
|
|
fills = [
|
|
SlippageFill(tier="taker", slippage_delta_bps=2.0),
|
|
SlippageFill(tier="taker", slippage_delta_bps=4.0),
|
|
SlippageFill(tier="unlock", slippage_delta_bps=10.0),
|
|
]
|
|
out = indicative_slippage(fills)
|
|
assert out["taker"]["mean_delta_bps"] == 3.0
|
|
assert out["taker"]["coverage"] == 1.0
|
|
assert out["unlock"]["mean_delta_bps"] == 10.0
|
|
|
|
|
|
def test_indicative_slippage_excludes_low_confidence_and_reports_coverage():
|
|
fills = [
|
|
SlippageFill(tier="unlock", slippage_delta_bps=8.0),
|
|
SlippageFill(tier="unlock", slippage_delta_bps=999.0, low_confidence=True),
|
|
]
|
|
out = indicative_slippage(fills)
|
|
# low-confidence excluded from the mean; coverage = 1 real / 2 total
|
|
assert out["unlock"]["mean_delta_bps"] == 8.0
|
|
assert out["unlock"]["coverage"] == 0.5
|
|
|
|
|
|
def test_indicative_slippage_all_low_confidence_tier_has_no_mean():
|
|
fills = [SlippageFill(tier="unlock", slippage_delta_bps=5.0, low_confidence=True)]
|
|
out = indicative_slippage(fills)
|
|
assert out["unlock"]["mean_delta_bps"] is None
|
|
assert out["unlock"]["coverage"] == 0.0
|