fix(validation): honor zero-aligned probe overrides

scripts/test_verify_corpus_metrics.py

@@ -3,11 +3,22 @@
 
 from __future__ import annotations
 
+import json
+import sys
+import tempfile
 import unittest
+from contextlib import redirect_stdout
+from io import StringIO
+from pathlib import Path
+from unittest.mock import patch
 
 from verify_corpus import (
     TradePair,
+    _apply_declared_tier_override,
+    _print_verification,
+    analyze_strategy,
     has_cross_entry_fifo_allocation,
+    main,
     schedule_exit_metrics,
     schedule_rows_for_gate,
 )
@@ -146,5 +157,194 @@ def test_schedule_metrics_penalize_engine_surplus_at_shared_exit(self) -> None:
         self.assertEqual(pnl_deltas, [0.0])
 
 
+class DeclaredTierOverrideTests(unittest.TestCase):
+    CSV_HEADER = (
+        "Trade #,Type,Date and time,Price,Position size (qty),Net P&L USD\n"
+    )
+
+    def setUp(self) -> None:
+        self.temp = tempfile.TemporaryDirectory()
+        self.root = Path(self.temp.name)
+
+    def tearDown(self) -> None:
+        self.temp.cleanup()
+
+    def strategy(
+        self,
+        name: str,
+        meta: dict,
+        *,
+        tv_empty: bool = False,
+        engine_empty: bool = False,
+        aligned: bool = False,
+    ) -> Path:
+        strategy = self.root / name
+        strategy.mkdir()
+        (strategy / "strategy.pine").write_text(
+            "//@version=6\nstrategy('zero alignment')\n",
+            encoding="utf-8",
+        )
+        applied_meta = {**meta}
+        if aligned:
+            applied_meta["tv_trades_csv_tz"] = "utc"
+        (strategy / "inputs.json").write_text(
+            json.dumps(applied_meta), encoding="utf-8"
+        )
+        tv = self.CSV_HEADER
+        engine = self.CSV_HEADER
+        if not tv_empty:
+            tv += (
+                "1,Entry long,2025-01-01 00:00,100,1,0\n"
+                "1,Exit long,2025-01-01 01:00,101,1,1\n"
+            )
+        if not engine_empty:
+            engine_day = "2025-01-01" if aligned else "2025-01-03"
+            engine += (
+                f"1,Entry long,{engine_day} 00:00,100,1,0\n"
+                f"1,Exit long,{engine_day} 01:00,101,1,1\n"
+            )
+        (strategy / "tv_trades.csv").write_text(tv, encoding="utf-8")
+        (strategy / "engine_trades.csv").write_text(engine, encoding="utf-8")
+        return strategy
+
+    def test_zero_alignment_honors_declared_tiers_and_precedence(self) -> None:
+        cases = (
+            ("ordinary", {}, "minimal"),
+            ("anomaly", {"expected_tier": "anomaly"}, "anomaly"),
+            ("expected-engine", {"expected_tier": "engine_only"}, "engine_only"),
+            (
+                "override-engine",
+                {"validation_overrides": {"expect_tv_match": False}},
+                "engine_only",
+            ),
+            (
+                "override-precedence",
+                {
+                    "expected_tier": "anomaly",
+                    "validation_overrides": {"expect_tv_match": False},
+                },
+                "engine_only",
+            ),
+        )
+        for name, meta, expected in cases:
+            with self.subTest(name=name):
+                result = analyze_strategy(self.strategy(name, meta))
+                self.assertTrue(result.no_aligned_trades)
+                self.assertEqual(result.label, expected)
+
+                output = StringIO()
+                with redirect_stdout(output):
+                    _print_verification(result)
+                self.assertIn(f"-> {expected}", output.getvalue())
+
+    def test_both_empty_excellent_is_not_masked_by_stale_declarations(self) -> None:
+        strategy = self.strategy(
+            "both-empty",
+            {
+                "expected_tier": "anomaly",
+                "validation_overrides": {"expect_tv_match": False},
+            },
+            tv_empty=True,
+            engine_empty=True,
+        )
+
+        result = analyze_strategy(strategy)
+
+        self.assertTrue(result.no_aligned_trades)
+        self.assertEqual(result.label, "excellent")
+
+    def test_single_sided_empty_results_honor_valid_declarations(self) -> None:
+        anomaly = analyze_strategy(self.strategy(
+            "tv-empty",
+            {"expected_tier": "anomaly"},
+            tv_empty=True,
+        ))
+        engine_only = analyze_strategy(self.strategy(
+            "engine-empty",
+            {"validation_overrides": {"expect_tv_match": False}},
+            engine_empty=True,
+        ))
+
+        self.assertEqual((anomaly.tv_raw_count, anomaly.eng_raw_count), (0, 1))
+        self.assertEqual(anomaly.label, "anomaly")
+        self.assertEqual(
+            (engine_only.tv_raw_count, engine_only.eng_raw_count), (1, 0)
+        )
+        self.assertEqual(engine_only.label, "engine_only")
+
+    def test_matched_excellent_preserves_malformed_override_failure(self) -> None:
+        strategy = self.strategy(
+            "matched-malformed",
+            {"validation_overrides": "oops"},
+            aligned=True,
+        )
+
+        with self.assertRaisesRegex(AttributeError, "has no attribute 'get'"):
+            analyze_strategy(strategy)
+
+    def test_malformed_override_preserves_empty_branch_asymmetry(self) -> None:
+        both_empty = self.strategy(
+            "both-empty-malformed",
+            {"validation_overrides": "oops"},
+            tv_empty=True,
+            engine_empty=True,
+        )
+        self.assertEqual(analyze_strategy(both_empty).label, "excellent")
+
+        single_sided = (
+            ("tv-empty-malformed", {"tv_empty": True}),
+            ("engine-empty-malformed", {"engine_empty": True}),
+        )
+        for name, empty_side in single_sided:
+            with self.subTest(name=name):
+                strategy = self.strategy(
+                    name,
+                    {"validation_overrides": "oops"},
+                    **empty_side,
+                )
+                with self.assertRaisesRegex(
+                    AttributeError, "has no attribute 'get'"
+                ):
+                    analyze_strategy(strategy)
+
+    def test_override_helper_preserves_normal_path_precedence(self) -> None:
+        both = {
+            "expected_tier": "anomaly",
+            "validation_overrides": {"expect_tv_match": False},
+        }
+        self.assertEqual(
+            _apply_declared_tier_override("weak", both), "engine_only"
+        )
+        self.assertEqual(
+            _apply_declared_tier_override("weak", {"expected_tier": "anomaly"}),
+            "anomaly",
+        )
+        self.assertEqual(
+            _apply_declared_tier_override("excellent", both), "excellent"
+        )
+
+    def test_single_strategy_cli_exit_matches_declared_zero_alignment(self) -> None:
+        cases = (
+            ("cli-minimal", {}, 1, "minimal"),
+            ("cli-anomaly", {"expected_tier": "anomaly"}, 0, "anomaly"),
+            (
+                "cli-engine-only",
+                {"validation_overrides": {"expect_tv_match": False}},
+                0,
+                "engine_only",
+            ),
+        )
+        for name, meta, expected_rc, expected_label in cases:
+            strategy = self.strategy(name, meta)
+            output = StringIO()
+            with self.subTest(name=name), patch.object(
+                sys, "argv", ["verify_corpus.py", str(strategy)]
+            ), redirect_stdout(output):
+                rc = main()
+
+            self.assertEqual(rc, expected_rc)
+            self.assertIn(f"-> {expected_label}", output.getvalue())
+
+
 if __name__ == "__main__":
     unittest.main()

scripts/verify_corpus.py

@@ -712,6 +712,26 @@ def load_strategy_metadata(strategy_dir: Path) -> dict:
         return json.load(f)
 
 
+def _apply_declared_tier_override(label: str, meta: dict) -> str:
+    """Apply documented-divergence metadata without masking a real TV match.
+
+    ``expected_tier`` owns explicit anomaly/engine-only declarations.  The
+    canonical ``validation_overrides.expect_tv_match=false`` spelling then
+    resolves to engine-only, matching the historical precedence when both are
+    present.  Metadata is evaluated before the excellent guards, preserving
+    the normal path's fail-closed behavior for malformed override objects.
+    """
+    expected_tier = str(meta.get("expected_tier", "")).strip().lower()
+    if expected_tier in {"anomaly", "engine_only"} and label != "excellent":
+        label = expected_tier
+
+    val_overrides = meta.get("validation_overrides") or {}
+    if (not bool(val_overrides.get("expect_tv_match", True))
+            and label != "excellent"):
+        label = "engine_only"
+    return label
+
+
 def tv_timezone_offset(meta: dict) -> int:
     tz_name = str(meta.get("tv_trades_csv_tz", "")).lower()
     return TV_TZ_BY_NAME.get(tz_name, TV_CSV_TZ_OFFSET_HOURS)
@@ -824,6 +844,11 @@ def analyze_strategy(strategy_dir: Path) -> VerificationResult:
 
     if not matched:
         label = "excellent" if len(tv_cmp) == 0 and len(eng_cmp) == 0 else "minimal"
+        # The historical both-empty Excellent branch returned before reading
+        # override metadata. Keep that exact behavior while fixing declared
+        # divergence for the non-excellent zero-alignment branch.
+        if label != "excellent":
+            label = _apply_declared_tier_override(label, meta)
         return VerificationResult(
             strategy_dir=strategy_dir,
             rel=rel,
@@ -1021,33 +1046,10 @@ def _is_dropped_open(t):
     else:
         label = "minimal"
 
-    # Per-probe override: if inputs.json declares an `expected_tier` of
-    # "anomaly" or "engine_only", honor it instead of the raw computed tier.
-    # Mirrors canonical pineforge-utils/validator/validate.py semantics:
-    #   - "anomaly"     = engine produces correct output; TV is non-deterministic
-    #                     or wrong on this probe (documented divergence). Reported
-    #                     separately so it doesn't mask as a regression.
-    #   - "engine_only" = probe is engine-only by design (no faithful TV reference);
-    #                     surfaced separately, excluded from headline parity counts.
-    # The override only fires when the computed tier is below `excellent` so a
-    # future engine improvement that genuinely matches TV is NOT silently masked.
-    expected_tier = str(meta.get("expected_tier", "")).strip().lower()
-    if expected_tier in {"anomaly", "engine_only"} and label != "excellent":
-        label = expected_tier
-
-    # Per-probe override (canonical schema): inputs.json may declare
-    # `validation_overrides.expect_tv_match: false` for probes where the
-    # engine is correct but TV intentionally diverges (e.g. documented
-    # TV-side boundary-bar non-determinism). Mirrors canonical
-    # pineforge-utils/validator/validate.py::expect_tv_match=False handling
-    # (search ENGINE_ONLY_LABEL there): relabel as `engine_only` so it
-    # doesn't mask as a low-tier regression. Same `excellent` guard as
-    # above so a future engine fix that genuinely matches TV is not
-    # silently masked.
-    val_overrides = meta.get("validation_overrides") or {}
-    expect_tv_match = bool(val_overrides.get("expect_tv_match", True))
-    if not expect_tv_match and label != "excellent":
-        label = "engine_only"
+    # Documented-divergence metadata is shared with the no-alignment path.
+    # Keeping the precedence in one helper prevents an early-return branch
+    # from silently turning declared engine-only/anomaly probes into Minimal.
+    label = _apply_declared_tier_override(label, meta)
 
     notes = ""
     if label != "excellent":
@@ -1125,7 +1127,7 @@ def _print_verification(result: VerificationResult, *, show_diffs: int = 0) -> N
     if result.no_aligned_trades:
         print(
             f"{result.rel}: TV={result.tv_raw_count} engine={result.eng_raw_count} "
-            "matched=0  (no aligned trades)"
+            f"matched=0  (no aligned trades)\n  -> {result.label}"
         )
         return