Fix FIFO verifier masking in corpus scores

scripts/test_verify_corpus_metrics.py

@@ -0,0 +1,150 @@
+#!/usr/bin/env python3
+"""Focused tests for fragmented-FIFO schedule metric eligibility."""
+
+from __future__ import annotations
+
+import unittest
+
+from verify_corpus import (
+    TradePair,
+    has_cross_entry_fifo_allocation,
+    schedule_exit_metrics,
+    schedule_rows_for_gate,
+)
+
+
+def trade(
+    number: int,
+    entry_time: int,
+    entry_price: float,
+    exit_time: int,
+    exit_price: float,
+    *,
+    direction: str = "long",
+    qty: float = 1.0,
+    pnl: float = 0.0,
+) -> TradePair:
+    return TradePair(
+        direction=direction,
+        entry_time=entry_time,
+        entry_price=entry_price,
+        exit_time=exit_time,
+        exit_price=exit_price,
+        qty=qty,
+        pnl=pnl,
+        trade_num=number,
+    )
+
+
+class FragmentedFifoEligibilityTests(unittest.TestCase):
+    def test_true_fifo_requires_split_entry_and_cross_entry_exit(self) -> None:
+        rows = [
+            trade(1, 100, 10.0, 200, 12.0, qty=0.5),
+            trade(2, 100, 10.0, 300, 14.0, qty=0.5),
+            trade(3, 150, 11.0, 300, 14.0, qty=1.0),
+        ]
+        self.assertTrue(has_cross_entry_fifo_allocation(rows))
+
+    def test_margin_fragments_are_not_fifo_allocation(self) -> None:
+        rows = [
+            trade(1, 100, 10.0, 200, 9.0, qty=0.1),
+            trade(2, 100, 10.0, 250, 8.0, qty=0.9),
+            trade(3, 300, 11.0, 400, 12.0, qty=1.0),
+        ]
+        self.assertFalse(has_cross_entry_fifo_allocation(rows))
+
+    def test_shared_exit_without_split_entry_needs_no_rescue(self) -> None:
+        rows = [
+            trade(1, 100, 10.0, 300, 14.0),
+            trade(2, 150, 11.0, 300, 14.0),
+        ]
+        self.assertFalse(has_cross_entry_fifo_allocation(rows))
+
+    def test_same_exit_qty_fragments_do_not_create_fifo_allocation(self) -> None:
+        rows = [
+            trade(1, 100, 10.0, 300, 14.0, qty=0.01),
+            trade(2, 100, 10.0, 300, 14.0, qty=0.99),
+            trade(3, 150, 11.0, 300, 14.0),
+        ]
+        self.assertFalse(has_cross_entry_fifo_allocation(rows))
+
+    def test_unrelated_split_and_shared_exit_do_not_compose(self) -> None:
+        rows = [
+            # One isolated margin-fragmented entry.
+            trade(1, 100, 10.0, 200, 9.0, qty=0.1),
+            trade(2, 100, 10.0, 250, 8.0, qty=0.9),
+            # A separate shared exit whose owners are not fragmented.
+            trade(3, 300, 11.0, 500, 14.0),
+            trade(4, 350, 12.0, 500, 14.0),
+        ]
+        self.assertFalse(has_cross_entry_fifo_allocation(rows))
+
+    def test_schedule_rows_include_matched_counterpart_not_in_gate(self) -> None:
+        tv_scored = trade(3, 100, 10.0, 200, 11.0)
+        eng_scored = trade(3, 90, 10.0, 200, 11.0)
+        tv_raw = [
+            trade(1, 80, 8.0, 180, 9.0),
+            trade(2, 95, 9.5, 195, 10.5),
+            tv_scored,
+            trade(4, 120, 12.0, 220, 13.0),
+        ]
+        eng_raw = [
+            trade(1, 70, 7.0, 170, 8.0),
+            eng_scored,
+            trade(2, 95, 9.5, 195, 10.5),
+            trade(4, 110, 12.0, 220, 13.0),
+        ]
+
+        tv_rows, eng_rows = schedule_rows_for_gate(
+            tv_raw,
+            eng_raw,
+            [tv_scored],
+            [],
+            [(tv_scored, eng_scored)],
+        )
+
+        self.assertEqual([row.entry_time for row in tv_rows], [100])
+        self.assertEqual([row.entry_time for row in eng_rows], [90])
+
+    def test_empty_gate_keeps_complete_raw_schedules(self) -> None:
+        tv_raw = [trade(1, 100, 10.0, 200, 11.0)]
+        eng_raw = [trade(1, 90, 10.0, 200, 11.0)]
+        tv_rows, eng_rows = schedule_rows_for_gate(tv_raw, eng_raw, [], [], [])
+        self.assertIs(tv_rows, tv_raw)
+        self.assertIs(eng_rows, eng_raw)
+
+    def test_schedule_eligibility_ignores_fifo_outside_gate(self) -> None:
+        tv_raw = [trade(1, 100, 10.0, 200, 12.0)]
+        eng_raw = [
+            # A genuine FIFO collision outside the scored window.
+            trade(1, 10, 8.0, 20, 9.0, qty=0.5),
+            trade(2, 10, 8.0, 30, 10.0, qty=0.5),
+            trade(3, 15, 9.0, 30, 10.0),
+            # The only engine row inside the scored window.
+            trade(4, 100, 10.0, 200, 12.0),
+        ]
+        self.assertTrue(has_cross_entry_fifo_allocation(eng_raw))
+
+        tv_rows, eng_rows = schedule_rows_for_gate(
+            tv_raw,
+            eng_raw,
+            [tv_raw[-1]],
+            [eng_raw[-1]],
+            [(tv_raw[-1], eng_raw[-1])],
+        )
+
+        self.assertFalse(has_cross_entry_fifo_allocation(tv_rows))
+        self.assertFalse(has_cross_entry_fifo_allocation(eng_rows))
+
+    def test_schedule_metrics_penalize_engine_surplus_at_shared_exit(self) -> None:
+        tv_raw = [trade(1, 100, 10.0, 200, 12.0, qty=1.0, pnl=2.0)]
+        eng_raw = [trade(1, 100, 10.0, 200, 12.0, qty=2.0, pnl=2.0)]
+
+        exit_deltas, pnl_deltas = schedule_exit_metrics(tv_raw, eng_raw)
+
+        self.assertEqual(exit_deltas, [0.0, 1.0])
+        self.assertEqual(pnl_deltas, [0.0])
+
+
+if __name__ == "__main__":
+    unittest.main()

scripts/verify_corpus.py

@@ -591,14 +591,73 @@ def consolidate_fragments(pairs: list[TradePair]) -> list[TradePair]:
     return out
 
 
-def has_fragmented_fifo_groups(pairs: list[TradePair]) -> bool:
-    seen: set[tuple[int, float, str]] = set()
-    for t in pairs:
-        key = (t.entry_time, t.entry_price, t.direction)
-        if key in seen:
-            return True
-        seen.add(key)
-    return False
+def has_cross_entry_fifo_allocation(pairs: list[TradePair]) -> bool:
+    """Return whether raw rows expose a genuinely ambiguous FIFO schedule.
+
+    A repeated entry key alone is not enough.  Margin calls and other partial
+    exits split one ordinary (pyramiding=0) position into several rows too,
+    but there is no cross-entry allocation ambiguity in that shape.  The
+    schedule-level rescue is justified only when BOTH facts are present:
+
+    1. one entry fill drains through multiple distinct exit events; and
+    2. one exact exit event closes quantity owned by multiple distinct entry
+       fills (the defining FIFO-grid allocation collision).
+
+    Exit identity deliberately matches :func:`schedule_exit_metrics` so the
+    eligibility predicate and the metric it enables cannot disagree about
+    which broker event is shared.
+    """
+    entry_exits: dict[
+        tuple[int, float, str], set[tuple[int, float, str]]
+    ] = {}
+    exit_owners: dict[
+        tuple[int, float, str], set[tuple[int, float, str]]
+    ] = {}
+    for trade in pairs:
+        entry_key = (trade.entry_time, trade.entry_price, trade.direction)
+        exit_key = (trade.exit_time, round(trade.exit_price, 6), trade.direction)
+        entry_exits.setdefault(entry_key, set()).add(exit_key)
+        exit_owners.setdefault(exit_key, set()).add(entry_key)
+
+    multi_exit_entries = {
+        entry_key for entry_key, exits in entry_exits.items() if len(exits) > 1
+    }
+    return any(
+        len(owners) > 1 and bool(owners & multi_exit_entries)
+        for owners in exit_owners.values()
+    )
+
+
+def schedule_rows_for_gate(
+    tv_raw: list[TradePair],
+    eng_raw: list[TradePair],
+    tv_gate: list[TradePair],
+    eng_gate: list[TradePair],
+    gating_matched: list[tuple[TradePair, TradePair]],
+) -> tuple[list[TradePair], list[TradePair]]:
+    """Select the scored raw-entry groups for schedule comparison.
+
+    Canonical alignment permits a TV/engine entry pair to differ by up to the
+    match window.  Bounding both raw schedules only by independently gated
+    entry times can then drop one half of an already scored boundary pair.
+    Select each side's gate entries plus its half of every scored match.  Exact
+    entry keys preserve every raw fragment without re-importing unrelated
+    out-of-gate rows that happen to lie between two boundary timestamps.
+    """
+    def entry_key(trade: TradePair) -> tuple[int, float, str]:
+        return (trade.entry_time, trade.entry_price, trade.direction)
+
+    tv_keys = {entry_key(trade) for trade in tv_gate}
+    eng_keys = {entry_key(trade) for trade in eng_gate}
+    for tv_trade, eng_trade in gating_matched:
+        tv_keys.add(entry_key(tv_trade))
+        eng_keys.add(entry_key(eng_trade))
+    if not tv_keys and not eng_keys:
+        return tv_raw, eng_raw
+    return (
+        [trade for trade in tv_raw if entry_key(trade) in tv_keys],
+        [trade for trade in eng_raw if entry_key(trade) in eng_keys],
+    )
 
 
 def schedule_exit_metrics(tv_raw: list[TradePair], eng_raw: list[TradePair]) -> tuple[list[float], list[float]]:
@@ -629,6 +688,8 @@ def build(rows: list[TradePair]) -> dict[tuple[int, float, str], float]:
             exit_deltas.append(0.0)
         if tv_qty - matched_qty > 1e-9:
             exit_deltas.append(1.0)
+        if eng_qty - matched_qty > 1e-9:
+            exit_deltas.append(1.0)
     for key, eng_qty in eng_sched.items():
         if key not in tv_sched and eng_qty > 1e-9:
             exit_deltas.append(1.0)
@@ -851,15 +912,14 @@ def _split_interior(pairs: list[tuple[TradePair, TradePair]]) -> list[tuple[Trad
     # position-level close events while entry-grouped consolidation smears exit
     # prices across different FIFO lot boundaries. Under a strict guard, score
     # exit/PnL on the raw exit schedule instead of the consolidated deal blend.
-    fragmented_fifo = has_fragmented_fifo_groups(tv_raw_all) or has_fragmented_fifo_groups(eng_raw_all)
+    tv_sched_rows, eng_sched_rows = schedule_rows_for_gate(
+        tv_raw_all, eng_raw_all, tv_gate, eng_gate, gating_matched
+    )
+    fragmented_fifo = (
+        has_cross_entry_fifo_allocation(tv_sched_rows)
+        or has_cross_entry_fifo_allocation(eng_sched_rows)
+    )
     if fragmented_fifo and entry_p90 < 1e-12 and count_delta < STRONG_COUNT_DELTA:
-        if tv_gate:
-            lo = min(t.entry_time for t in tv_gate)
-            hi = max(t.entry_time for t in tv_gate)
-            tv_sched_rows = [t for t in tv_raw_all if lo <= t.entry_time <= hi]
-            eng_sched_rows = [e for e in eng_raw_all if lo <= e.entry_time <= hi]
-        else:
-            tv_sched_rows, eng_sched_rows = tv_raw_all, eng_raw_all
         sched_exit_deltas, sched_pnl_deltas = schedule_exit_metrics(tv_sched_rows, eng_sched_rows)
         if sched_exit_deltas:
             sched_exit_p90 = percentile(sched_exit_deltas, 0.90)

tests/CMakeLists.txt

@@ -123,6 +123,12 @@ add_test(
             ${PROJECT_SOURCE_DIR}/scripts/fingerprint_self_test.py
 )
 
+add_test(
+    NAME test_verify_corpus_metrics
+    COMMAND ${Python3_EXECUTABLE}
+            ${PROJECT_SOURCE_DIR}/scripts/test_verify_corpus_metrics.py
+)
+
 # When PINEFORGE_ENABLE_COVERAGE is ON we also instrument the test
 # binaries — header-only code (Series, na, color, log, math::pine_random)
 # is otherwise reported as 0% because it's only inlined into test TUs.