diff --git a/scripts/test_verify_corpus_metrics.py b/scripts/test_verify_corpus_metrics.py index 618e7f0..1df0c75 100644 --- a/scripts/test_verify_corpus_metrics.py +++ b/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() diff --git a/scripts/verify_corpus.py b/scripts/verify_corpus.py index 104544b..db2e1ce 100755 --- a/scripts/verify_corpus.py +++ b/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