Merge: bybit testnet pure logic (slippage, execution_gap, plan_orders)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

src/fxhnt/application/bybit_testnet_execution.py

@@ -0,0 +1,115 @@
+"""Order planning for the Bybit testnet real-paper execution leg (design §4.2) — pure, no I/O.
+
+⚠️ NO OVERLAY RE-APPLY. The design (review #1) assumed `bybit_4edge` applied a risk overlay that had to be
+re-run on the testnet book's own equity/drawdown. The Task-1 gate confirmed it does NOT: the book is built
+naive equal-weight with the live overlay deliberately NOT applied (see the `bybit_paper_book` module docstring
+and `latest_raw_sleeve_weights`, which returns the RAW pre-overlay per-symbol target). So the raw sleeve weights
+ARE the effective weights — there is nothing to re-apply, and `effective_weights` is intentionally absent.
+
+`plan_orders` therefore diffs the raw target weights straight against the live testnet positions:
+  * target_qty[s] = (weight·equity)/marks[s], snapped to the instrument `qty_step` (toward zero, never over);
+  * a target below the tradable minimum (`|target·mark| < max(minNotional, min_order_usd)` or
+    `|target| < minOrderQty`) collapses to 0 — we don't hold a sub-minimum dust position;
+  * delta = target_qty − current_position; an order below the tradable minimum is dropped (the exchange would
+    reject it anyway);
+  * `reduce_only=True` on reductions/closes (magnitude shrinks toward/through zero without flipping sign);
+  * the gross book is capped: Σ|weight·equity| ≤ equity·`max_gross` (=1.0, no leverage) — over-gross weights are
+    scaled down (clip + log), matching the overlay's gross-cap invariant even though the overlay itself is off.
+
+The `OrderIntent` / `InstrumentLimit` frozen DTOs are the stable contract consumed by the ccxt adapter (Task 5)
+and the Dagster reconcile asset (Task 8).
+"""
+from __future__ import annotations
+
+import logging
+import math
+from dataclasses import dataclass
+
+_LOG = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True, slots=True)
+class OrderIntent:
+    """A planned testnet order. `qty` is the signed delta (>0 buy, <0 sell), already snapped to `qty_step`."""
+
+    symbol: str
+    qty: float
+    reduce_only: bool
+
+
+@dataclass(frozen=True, slots=True)
+class InstrumentLimit:
+    """Exchange filters for one symbol (parsed from ccxt `load_markets`)."""
+
+    min_order_qty: float
+    qty_step: float
+    min_notional: float
+
+
+def _snap(qty: float, step: float) -> float:
+    """Snap `qty` toward zero to a multiple of `step` (never overshoots the target / the gross cap)."""
+    if step <= 0.0:
+        return qty
+    n = math.floor(abs(qty) / step + 1e-9)
+    return math.copysign(round(n * step, 12), qty)
+
+
+def _tradable(qty: float, mark: float, limit: InstrumentLimit, min_order_usd: float) -> bool:
+    """A qty is tradable if its notional clears max(minNotional, min_order_usd) and its size clears
+    minOrderQty."""
+    if qty == 0.0:
+        return False
+    if abs(qty) < limit.min_order_qty:
+        return False
+    return abs(qty * mark) >= max(limit.min_notional, min_order_usd)
+
+
+def plan_orders(
+    target_weights: dict[str, float],
+    current_positions: dict[str, float],
+    equity: float,
+    marks: dict[str, float],
+    limits: dict[str, InstrumentLimit],
+    *,
+    min_order_usd: float = 10.0,
+    max_gross: float = 1.0,
+) -> list[OrderIntent]:
+    """Diff the (already-effective) raw target weights against the testnet positions into a list of
+    `OrderIntent` deltas. See the module docstring for the rules. Pure; no I/O."""
+    weights = _cap_gross(target_weights, equity, max_gross)
+
+    intents: list[OrderIntent] = []
+    for symbol in sorted(weights):
+        mark = marks.get(symbol)
+        limit = limits.get(symbol)
+        if mark is None or mark <= 0.0 or limit is None:
+            continue
+        target_qty = _snap(weights[symbol] * equity / mark, limit.qty_step)
+        if not _tradable(target_qty, mark, limit, min_order_usd):
+            target_qty = 0.0  # below the minimum → don't hold a dust position
+        current = current_positions.get(symbol, 0.0)
+        delta = _snap(target_qty - current, limit.qty_step)
+        if not _tradable(delta, mark, limit, min_order_usd):
+            continue  # nothing to do / sub-minimum residual the exchange would reject
+        intents.append(OrderIntent(symbol, delta, _is_reduce_only(target_qty, current)))
+    return intents
+
+
+def _cap_gross(target_weights: dict[str, float], equity: float, max_gross: float) -> dict[str, float]:
+    """Enforce Σ|weight·equity| ≤ equity·max_gross by scaling all weights down if needed (clip + log)."""
+    gross = sum(abs(w) for w in target_weights.values())
+    if gross <= max_gross or gross == 0.0:
+        return dict(target_weights)
+    scale = max_gross / gross
+    _LOG.warning("bybit_testnet gross %.4f > max_gross %.4f → scaling weights by %.4f", gross, max_gross, scale)
+    return {s: w * scale for s, w in target_weights.items()}
+
+
+def _is_reduce_only(target_qty: float, current: float) -> bool:
+    """True when the order shrinks the position toward/through zero without flipping sign (a reduce or close)."""
+    if current == 0.0:
+        return False
+    if target_qty == 0.0:
+        return True
+    same_sign = math.copysign(1.0, target_qty) == math.copysign(1.0, current)
+    return same_sign and abs(target_qty) < abs(current)

src/fxhnt/application/execution_gap.py

@@ -0,0 +1,83 @@
+"""Execution-gap verdict + indicative slippage (design §5) — the execution analog of the recon gate.
+
+Pure functions over the recorded testnet rows (no I/O): the reconcile/record assets persist post-rebalance
+positions and per-fill slippage to `bybit_testnet_nav` / `bybit_testnet_fills`; the read-models load them into
+the small frozen DTOs below and these functions compute the verdict the cockpit (Phase 2) will surface.
+
+  * `mechanics_verdict` — MECHANICS PASS iff, over the window, every recorded post-rebalance position tracks its
+    effective target within `tolerance` (each symbol within ~1 qtyStep + dust) AND there are no reconcile-error
+    / execution-gap rows. Otherwise FAIL, naming the offending symbol(s). A target with no recorded row is an
+    execution gap and counts as offending.
+  * `indicative_slippage` — aggregate `slippage_vs_assumed` (the `slippage_delta_bps`) per A8 cost tier,
+    EXCLUDING `low_confidence` (thin-book / one-sided-mid) fills, and report `coverage` = real-liquidity fills /
+    all fills in that tier. Explicitly indicative: magnitude is pending mainnet micro-live (spec §3, §11).
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True, slots=True)
+class PositionRow:
+    """A recorded post-rebalance position for one symbol on one run. `error=True` flags a reconcile error /
+    execution gap for the symbol (a placement failure, a missing fill) — a visible gap, not a silent pass."""
+
+    symbol: str
+    position: float          # recorded post-rebalance signed base qty
+    error: bool = False
+
+
+@dataclass(frozen=True, slots=True)
+class SlippageFill:
+    """A recorded fill's slippage-vs-assumed for tier aggregation. `low_confidence=True` marks a fill whose mid
+    came from an empty/one-sided thin testnet book (`mid` was `None`) — recorded but excluded from the verdict."""
+
+    tier: str
+    slippage_delta_bps: float
+    low_confidence: bool = False
+
+
+@dataclass(frozen=True, slots=True)
+class Verdict:
+    passed: bool
+    offending: tuple[str, ...]
+
+
+def mechanics_verdict(
+    rows: list[PositionRow], target_positions: dict[str, float], tolerance: float
+) -> Verdict:
+    """PASS iff every effective target is tracked by a recorded post-rebalance position within `tolerance` and
+    no row is an error/gap row; else FAIL with the sorted offending symbols. A target lacking a recorded row is
+    an execution gap (offending)."""
+    by_symbol = {r.symbol: r for r in rows}
+    offending: set[str] = set()
+    for row in rows:
+        if row.error:
+            offending.add(row.symbol)
+    for symbol, target in target_positions.items():
+        row = by_symbol.get(symbol)
+        if row is None:                                  # intended a position, recorded nothing → gap
+            offending.add(symbol)
+            continue
+        if abs(row.position - target) > tolerance:
+            offending.add(symbol)
+    ordered = tuple(sorted(offending))
+    return Verdict(passed=not ordered, offending=ordered)
+
+
+def indicative_slippage(fills: list[SlippageFill]) -> dict[str, dict[str, float | None]]:
+    """`{tier: {"mean_delta_bps", "coverage"}}` — per-tier mean `slippage_delta_bps` over real-liquidity fills
+    (low-confidence excluded), with `coverage` = real fills / all fills in the tier. A tier with only
+    low-confidence fills has `mean_delta_bps=None` and `coverage=0.0`."""
+    totals: dict[str, list[float]] = {}
+    counts: dict[str, int] = {}
+    for fill in fills:
+        counts[fill.tier] = counts.get(fill.tier, 0) + 1
+        if not fill.low_confidence:
+            totals.setdefault(fill.tier, []).append(fill.slippage_delta_bps)
+    out: dict[str, dict[str, float | None]] = {}
+    for tier, n_total in counts.items():
+        reals = totals.get(tier, [])
+        mean = sum(reals) / len(reals) if reals else None
+        out[tier] = {"mean_delta_bps": mean, "coverage": len(reals) / n_total}
+    return out

src/fxhnt/application/slippage.py

@@ -0,0 +1,42 @@
+"""Pure decomposed-slippage measurement model (design §3) — the testnet real-paper leg's first-class output.
+
+Three pure functions over floats (no I/O, no network) so the bps math is unit-tested in isolation. The Bybit
+execution adapter supplies `fill_price`, `mid_at_order` (the live order-book mid AT order time) and `book_mark`
+(the book's mark/close the targets were sized off); these split the total fill-vs-mark move into:
+
+  * `execution_slippage_bps` — fill vs the live mid: what the A8 cost model claims to capture (the 5/8/6/35bp
+    taker+spread tiers). >0 = paid worse than mid.
+  * `timing_drift_bps` — the live mid vs the book mark: the close→execution price move, a SEPARATE effect.
+    Conflating the two (v1's bug) made the number meaningless; the decomposition is the whole point.
+
+`slippage_vs_assumed_bps` is the validation delta vs the per-symbol assumed cost tier. Sign convention:
+`+1` for BUY, `-1` for SELL, so a positive bp is always an adverse cost regardless of side. Guards: a
+non-positive `mid_at_order`/`book_mark` (empty/one-sided thin testnet book) returns `None` — never a
+divide-by-zero or a garbage bp; the caller flags such a fill `low_confidence` and excludes it from the verdict.
+"""
+from __future__ import annotations
+
+
+def _sign(side: str) -> int:
+    return 1 if side.upper() == "BUY" else -1
+
+
+def execution_slippage_bps(fill: float, mid_at_order: float, side: str) -> float | None:
+    """Fill vs the live mid at order time, in bps (>0 = paid worse than mid). `None` if `mid_at_order <= 0`."""
+    if mid_at_order <= 0.0:
+        return None
+    return _sign(side) * (fill - mid_at_order) / mid_at_order * 1e4
+
+
+def timing_drift_bps(mid_at_order: float, book_mark: float, side: str) -> float | None:
+    """The live mid vs the book mark, in bps (>0 = market moved adversely since the mark). `None` if
+    `book_mark <= 0`."""
+    if book_mark <= 0.0:
+        return None
+    return _sign(side) * (mid_at_order - book_mark) / book_mark * 1e4
+
+
+def slippage_vs_assumed_bps(execution_slippage: float, assumed_tier_bps: float) -> float:
+    """The validation delta: measured execution slippage minus the assumed A8 cost tier (>0 = worse than
+    assumed)."""
+    return execution_slippage - assumed_tier_bps

tests/unit/test_bybit_testnet_execution.py

@@ -0,0 +1,73 @@
+"""Order planning for the Bybit testnet real-paper leg (design §4.2).
+
+NO overlay re-apply: `bybit_4edge` is built naive equal-weight with NO risk overlay (see
+`bybit_paper_book` module docstring + `latest_raw_sleeve_weights`), so the raw sleeve weights ARE the
+effective weights — there is nothing to re-apply. `plan_orders` therefore diffs the raw target weights
+straight against the testnet positions: target_qty = weight·equity/mark snapped to qtyStep, dust-filtered on
+minNotional/minOrderQty, delta vs current, reduce_only on reductions/closes, gross capped at max_gross=1.0.
+"""
+from __future__ import annotations
+
+from fxhnt.application.bybit_testnet_execution import (
+    InstrumentLimit,
+    OrderIntent,
+    plan_orders,
+)
+
+
+def _lim(min_order_qty: float = 0.001, qty_step: float = 0.001,
+         min_notional: float = 10.0) -> dict[str, InstrumentLimit]:
+    return {
+        "BTCUSDT": InstrumentLimit(min_order_qty, qty_step, min_notional),
+        "ETHUSDT": InstrumentLimit(min_order_qty, qty_step, min_notional),
+        "X": InstrumentLimit(min_order_qty, qty_step, min_notional),
+        "A": InstrumentLimit(min_order_qty, qty_step, min_notional),
+        "B": InstrumentLimit(min_order_qty, 0.5, min_notional),
+    }
+
+
+def test_buy_when_target_above_current():
+    intents = plan_orders({"BTCUSDT": 0.5}, {"BTCUSDT": 0.0}, equity=10_000,
+                          marks={"BTCUSDT": 100.0}, limits=_lim())
+    assert intents == [OrderIntent("BTCUSDT", qty=50.0, reduce_only=False)]  # 0.5*10000/100
+
+
+def test_reduce_only_on_close():
+    intents = plan_orders({"BTCUSDT": 0.0}, {"BTCUSDT": 50.0}, 10_000,
+                          {"BTCUSDT": 100.0}, _lim())
+    assert intents[0].reduce_only is True
+    assert intents[0].qty == -50.0
+
+
+def test_partial_reduce_is_reduce_only():
+    intents = plan_orders({"BTCUSDT": 0.2}, {"BTCUSDT": 50.0}, 10_000,
+                          {"BTCUSDT": 100.0}, _lim())
+    assert intents[0].qty == -30.0  # 20 target - 50 current
+    assert intents[0].reduce_only is True
+
+
+def test_min_notional_filters_dust():
+    intents = plan_orders({"X": 0.0001}, {"X": 0.0}, 10_000, {"X": 100.0},
+                          _lim(min_notional=50))
+    assert intents == []  # 0.0001*10000=1 USD < 50 minNotional
+
+
+def test_qty_step_snap():
+    # 0.507*10000/100 = 50.7, snapped to a 0.5 step → 50.5
+    intents = plan_orders({"B": 0.507}, {"B": 0.0}, 10_000, {"B": 100.0}, _lim())
+    assert intents == [OrderIntent("B", qty=50.5, reduce_only=False)]
+
+
+def test_gross_cap_clips_oversized_weights():
+    # Σ|weight·equity| = 1.6·equity > equity·1.0 → scale by 0.625 → 0.5 each → qty 50 each
+    intents = plan_orders({"A": 0.8, "B": 0.8}, {"A": 0.0, "B": 0.0}, 10_000,
+                          {"A": 100.0, "B": 100.0}, _lim())
+    by_symbol = {i.symbol: i for i in intents}
+    assert by_symbol["A"].qty == 50.0
+    assert by_symbol["B"].qty == 50.0
+
+
+def test_no_order_when_already_on_target():
+    intents = plan_orders({"BTCUSDT": 0.5}, {"BTCUSDT": 50.0}, 10_000,
+                          {"BTCUSDT": 100.0}, _lim())
+    assert intents == []

tests/unit/test_execution_gap.py

@@ -0,0 +1,82 @@
+"""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

tests/unit/test_slippage.py

@@ -0,0 +1,50 @@
+"""Pure decomposed-slippage model (spec §3). All math over floats — no I/O — so it is unit-tested directly.
+
+The adapter supplies `fill_price`, `mid_at_order`, `book_mark`; the bps math is exercised here in isolation:
+sign on both sides, a known fill/mid → exact bps, the decomposition sums to the fill-vs-mark total, the
+assumed-tier delta, and the `mid<=0`/`book_mark<=0` None-guards (never a divide-by-zero or a garbage bp).
+"""
+from __future__ import annotations
+
+from fxhnt.application.slippage import (
+    execution_slippage_bps,
+    slippage_vs_assumed_bps,
+    timing_drift_bps,
+)
+
+
+def test_buy_pays_above_mid_is_positive():
+    # bought at 101 vs mid 100 → +100 bps cost
+    assert execution_slippage_bps(101.0, 100.0, "BUY") == 100.0
+
+
+def test_sell_below_mid_is_positive_cost():
+    # sold at 99 vs mid 100 → +100 bps cost
+    assert execution_slippage_bps(99.0, 100.0, "SELL") == 100.0
+
+
+def test_timing_drift_sign():
+    # BUY, mid moved 100→101 from the book mark → +100 bps adverse drift
+    assert timing_drift_bps(101.0, 100.0, "BUY") == 100.0
+
+
+def test_decomposition_sums_to_total():
+    fill, mid, mark = 102.0, 101.0, 100.0
+    exec_bps = execution_slippage_bps(fill, mid, "BUY")
+    timing_bps = timing_drift_bps(mid, mark, "BUY")
+    total = exec_bps + timing_bps
+    # The decomposition (spec §3) DEFINES total_fill_vs_mark_bps = execution + timing. The two terms use
+    # different denominators (mid for execution, mark for timing), so the sum reconstructs the single-
+    # denominator fill-vs-mark move ((102-100)/100*1e4 = 200) only up to that ~1bp compounding — it is NOT
+    # exactly 200 (the plan's `< 1e-6` assertion was off by this compounding term).
+    fill_vs_mark = (fill - mark) / mark * 1e4
+    assert abs(total - fill_vs_mark) < 1.0
+
+
+def test_assumed_delta():
+    assert slippage_vs_assumed_bps(40.0, 35.0) == 5.0  # 5bp worse than the 35bp tier
+
+
+def test_guards_none_on_bad_inputs():
+    assert execution_slippage_bps(101.0, 0.0, "BUY") is None
+    assert timing_drift_bps(101.0, -1.0, "SELL") is None