gendesign/tradein-mvp/backend/tests/test_backtest_regression_gate.py

"""Hermetic estimator regression gate (#1966 PR 3/3).

Replays the committed frozen backtest fixture through the full pricing spine
(``app.services.estimator._price_from_inputs``) with ZERO DB / network, recomputes
the backtest metrics, and asserts they match the committed baseline. Any change to
the spine, the metric code, or a config default that moves a metric beyond float
jitter fails this test → regenerate the baseline deliberately:

    cd tradein-mvp/backend
    uv run python -m scripts.backtest_estimator \
        --from-fixture tests/fixtures/backtest_full_fixture.json.gz \
        --update-baseline tests/fixtures/backtest_baseline.json

and justify the per-segment MAPE / coverage deltas in the PR. This is a RELATIVE
regression gate, not an absolute SLA (live coverage ~55% is data-blocked, see #1966).

The fixture is gzipped frozen prod inputs (opaque, rarely changes); the baseline is
the small diff-visible artifact that surfaces accuracy movement right in the PR diff.
"""

from __future__ import annotations

import json
import math
import os
from pathlib import Path

import pytest

os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

from app.services import estimator
from scripts.backtest_estimator import load_fixture, replay_fixture

_FIXTURES = Path(__file__).parent / "fixtures"
_FIXTURE_PATH = _FIXTURES / "backtest_full_fixture.json.gz"
_BASELINE_PATH = _FIXTURES / "backtest_baseline.json"

# Floats: a small relative+absolute tolerance absorbs cross-platform / Python
# libm last-ulp jitter (the replay is otherwise deterministic). A real regression
# moves a metric by orders of magnitude more than this, so it is still caught.
_REL_TOL = 1e-6
_ABS_TOL = 1e-6


def _assert_match(path: str, expected: object, actual: object) -> None:
    if isinstance(expected, dict):
        assert isinstance(actual, dict), f"{path}: expected dict, got {type(actual).__name__}"
        assert (
            expected.keys() == actual.keys()
        ), f"{path}: key set differs\n  expected={sorted(expected)}\n  actual=  {sorted(actual)}"
        for k in expected:
            _assert_match(f"{path}.{k}", expected[k], actual[k])
    elif isinstance(expected, list):
        assert isinstance(actual, list), f"{path}: expected list, got {type(actual).__name__}"
        assert len(expected) == len(actual), f"{path}: list length {len(actual)} != {len(expected)}"
        for i, (e, a) in enumerate(zip(expected, actual, strict=True)):
            _assert_match(f"{path}[{i}]", e, a)
    elif expected is None or isinstance(expected, bool):
        assert actual is expected or actual == expected, f"{path}: {actual!r} != {expected!r}"
    elif isinstance(expected, int):  # bool already handled above
        assert actual == expected, f"{path}: int {actual!r} != {expected!r}"
    elif isinstance(expected, float):
        assert isinstance(actual, int | float), f"{path}: {type(actual).__name__} not numeric"
        assert math.isclose(actual, expected, rel_tol=_REL_TOL, abs_tol=_ABS_TOL), (
            f"{path}: {actual!r} != baseline {expected!r} (Δ={actual - expected:.3e}). "
            f"The estimator/metrics changed — if intentional, regenerate the baseline "
            f"(--from-fixture --update-baseline) and justify the deltas in the PR."
        )
    else:
        assert actual == expected, f"{path}: {actual!r} != {expected!r}"


def test_fixture_and_baseline_committed() -> None:
    assert _FIXTURE_PATH.exists(), f"frozen fixture missing: {_FIXTURE_PATH}"
    assert _BASELINE_PATH.exists(), f"frozen baseline missing: {_BASELINE_PATH}"


def test_backtest_regression_gate(monkeypatch: pytest.MonkeyPatch) -> None:
    # The frozen fixture records the injected-callback control flow captured with
    # cross-source physical dedup OFF (#2087 H4 was a no-op default at capture time).
    # Dedup is now default ON (#2173), but this gate is a byte-identical REPLAY of a
    # frozen OFF capture — with dedup active _dedup_cross_source would trim listings
    # before quarter_indexes_lookup and the recorded call sequence would diverge. Pin
    # the flag OFF so the replay follows the captured control flow. (Accuracy-neutral:
    # backtest #1966 OFF vs ON is identical; dedup only trims user-visible n_analogs,
    # so the OFF baseline stays the valid spine regression reference.)
    monkeypatch.setattr(estimator.settings, "estimate_dedup_analogs_enabled", False)
    fixture = load_fixture(_FIXTURE_PATH)
    baseline = json.loads(_BASELINE_PATH.read_text(encoding="utf-8"))
    # Round-trip the replay output through JSON before comparing: the committed
    # baseline is JSON (string object keys), while replay_fixture returns native
    # dicts whose per_rooms buckets are int keys (0..4). Round-tripping normalises
    # key types to match — the same transform `--update-baseline` applies on write.
    metrics = json.loads(json.dumps(replay_fixture(fixture), ensure_ascii=False))
    _assert_match("metrics", baseline, metrics)