tradein-mvp/backend/scripts/backtest_estimator.py

@@ -30,7 +30,8 @@ For a held-out sample of ДКП deals we, per deal:
      ``_fetch_house_imv_anchor``) + inject the DB callables
      (``_get_asking_sold_ratio``, ``_lookup_quarter_index(es)``).
   2. Call ``_price_from_inputs`` for a byte-identical headline + expected_sold.
-     Deals the spine cannot price (median<=0 / <3 analogs) are skipped.
+     Deals the spine cannot price (median<=0) are skipped — prod parity (#1966):
+     there is NO analog-count floor, low-analog deals surface at low confidence.
   3. Score ``expected_sold_per_m2`` vs the realised SOLD ppm².
 
 METRICS (full spine)
@@ -87,10 +88,15 @@ USAGE
 from __future__ import annotations
 
 import argparse
+import dataclasses
 import json
 import logging
+import math
 import statistics
+from collections.abc import Callable
 from dataclasses import dataclass
+from datetime import date
+from decimal import Decimal
 from pathlib import Path
 from types import SimpleNamespace
 from typing import Any
@@ -1146,7 +1152,188 @@ def _select_analogs_full(
     return listings, analog_tier, fallback_used, area_widened
 
 
-def _predict_full_spine(db: Session, deal: DealSample, est: SimpleNamespace) -> Prediction | None:
+# --------------------------------------------------------------------------- #
+# Fixture capture + hermetic replay (#1966 PR 3/3)
+#
+# ``--dump-fixture`` freezes per-deal RESOLVED inputs to ``_price_from_inputs``
+# into a committed JSON file; ``replay_fixture`` replays the spine offline (ZERO
+# DB) so a CI test can assert metrics match a frozen baseline. The 3 DB-backed
+# callables (asking→sold ratio + quarter-index lookups) hide all I/O, so we
+# capture each as an ordered ``[arg -> return]`` call-list and replay it via an
+# exact-match stub. Everything else fed to ``_price_from_inputs`` is pure data.
+# --------------------------------------------------------------------------- #
+
+FIXTURE_SCHEMA_VERSION = 1
+
+
+def _sanitize_json(obj: Any) -> Any:
+    """Recursively coerce a captured value into a finite, JSON-plain structure.
+
+    - non-finite float (inf / nan) → None (``json.dump(allow_nan=False)`` would
+      otherwise raise);
+    - Decimal → float; date / datetime → ISO str; tuple → list; dict / list
+      recursed; bool / int / str / None pass through; any other type → ``str()``.
+
+    Idempotent on already-plain data, so applying it both at capture time (to the
+    recorded call args) and at replay time (to the live args before stub lookup)
+    keeps the lookup keys byte-stable across capture and replay.
+    """
+    if obj is None or isinstance(obj, bool | int | str):
+        return obj
+    if isinstance(obj, float):
+        return obj if math.isfinite(obj) else None
+    if isinstance(obj, Decimal):
+        f = float(obj)
+        return f if math.isfinite(f) else None
+    if isinstance(obj, dict):
+        return {str(k): _sanitize_json(v) for k, v in obj.items()}
+    if isinstance(obj, list | tuple):
+        return [_sanitize_json(x) for x in obj]
+    if isinstance(obj, date):  # date | datetime (datetime is a date subclass)
+        return obj.isoformat()
+    return str(obj)
+
+
+def _to_hashable(obj: Any) -> Any:
+    """Normalise a JSON-plain value into a hashable key (list→tuple, dict→sorted items)."""
+    if isinstance(obj, list | tuple):
+        return tuple(_to_hashable(x) for x in obj)
+    if isinstance(obj, dict):
+        return tuple(sorted((str(k), _to_hashable(v)) for k, v in obj.items()))
+    return obj
+
+
+def _coerce_ratio_return(ret: Any) -> tuple[Any, Any]:
+    """Recorded ``[ratio, basis]`` → ``(ratio, basis)`` tuple (unpacked at the call site)."""
+    return (ret[0], ret[1])
+
+
+def _coerce_qi_return(ret: Any) -> tuple[Any, Any] | None:
+    """Recorded ``[qi, n]`` or ``null`` → ``(qi, n)`` tuple or None."""
+    return None if ret is None else (ret[0], ret[1])
+
+
+def _coerce_qis_return(ret: Any) -> dict[str, float]:
+    """Recorded ``{quarter: qi}`` dict → ``dict[str, float]`` (keys forced to str)."""
+    return {str(k): v for k, v in dict(ret).items()}
+
+
+def _make_call_stub(
+    calls: list[Any], *, label: str, coerce: Callable[[Any], Any]
+) -> Callable[[Any], Any]:
+    """Build an exact-match replay stub from recorded ``[arg, return]`` pairs.
+
+    Each recorded entry is ``[serialized_single_arg, serialized_return]``. The
+    stub looks up the (normalised) positional arg and raises ``KeyError`` on an
+    unseen arg — that must never happen for a faithful fixture (the spine is
+    deterministic, so replay requests exactly the args capture recorded).
+    ``coerce`` maps the JSON-plain recorded return back to the live callable's
+    return type (tuple / dict) so unpacking at the call site behaves identically.
+    """
+    table: dict[Any, Any] = {}
+    for arg, ret in calls:
+        table[_to_hashable(_sanitize_json(arg))] = coerce(ret)
+
+    def _stub(arg: Any) -> Any:
+        key = _to_hashable(_sanitize_json(arg))
+        if key not in table:
+            raise KeyError(
+                f"{label}: unseen arg {arg!r} during replay — fixture is not faithful "
+                f"(recorded {len(table)} call(s))"
+            )
+        return table[key]
+
+    return _stub
+
+
+def replay_fixture(fixture: dict[str, Any]) -> dict[str, Any]:
+    """Replay a frozen backtest fixture through the full spine — hermetic, ZERO DB.
+
+    For every captured deal record: rebuild the ``GeocodeResult``, build exact-match
+    stubs for the 3 DB callables from the recorded call-lists, call
+    ``_price_from_inputs`` with the frozen kwargs, and rebuild the ``Prediction``
+    EXACTLY as ``_predict_full_spine`` does. Runs the harness's own
+    ``_compute_full_metrics`` over the replayed predictions plus a city-wide
+    headline computed from the stored (priced) deals.
+
+    The returned dict keeps ``expected_sold`` / ``range_coverage`` / ``calibration``
+    / ``sharpness`` / ``confidence_order`` / ``headline`` and DROPS the volatile
+    ``params`` block. Touches NO DB / network and does NOT consult
+    ``settings_at_capture`` — it prices against the live committed
+    ``estimator.settings`` defaults (so a settings change is caught as a metric
+    drift, not silently honoured). Deterministic: same fixture → identical dict.
+    """
+    est = _import_estimator_full()
+    m = est.m
+
+    deals = fixture.get("deals") or []
+    predictions: list[Prediction] = []
+    sold_ppm2_all: list[float] = []
+    pred_ppm2_all: list[float] = []
+
+    for rec in deals:
+        kw = dict(rec["kwargs"])
+        sold_ppm2_all.append(float(rec["sold_ppm2"]))
+        kw["geo"] = est.GeocodeResult(**kw["geo"])
+        kw["ratio_resolver"] = _make_call_stub(
+            rec.get("ratio_calls") or [], label="ratio_resolver", coerce=_coerce_ratio_return
+        )
+        kw["quarter_index_lookup"] = _make_call_stub(
+            rec.get("qi_calls") or [], label="quarter_index_lookup", coerce=_coerce_qi_return
+        )
+        kw["quarter_indexes_lookup"] = _make_call_stub(
+            rec.get("qis_calls") or [], label="quarter_indexes_lookup", coerce=_coerce_qis_return
+        )
+
+        pr = m._price_from_inputs(**kw)
+
+        es_ppm2 = float(pr.expected_sold_per_m2) if pr.expected_sold_per_m2 is not None else None
+        es_price = float(pr.expected_sold_price) if pr.expected_sold_price is not None else None
+        r_low = (
+            float(pr.expected_sold_range_low) if pr.expected_sold_range_low is not None else None
+        )
+        r_high = (
+            float(pr.expected_sold_range_high) if pr.expected_sold_range_high is not None else None
+        )
+        prediction = Prediction(
+            deal_id=int(rec["deal_id"]),
+            rooms=rec["rooms"],
+            area_m2=float(rec["area_m2"]),
+            sold_ppm2=float(rec["sold_ppm2"]),
+            median_ppm2=float(pr.median_ppm2),
+            confidence=pr.confidence,
+            anchor_tier=pr.anchor_tier,
+            expected_sold_ppm2=es_ppm2,
+            expected_sold_price=es_price,
+            range_low=r_low,
+            range_high=r_high,
+        )
+        predictions.append(prediction)
+        pred_ppm2_all.append(prediction.median_ppm2)
+
+    # The fixture stores ONLY priced deals, so n_no_prediction is 0 here.
+    metrics = _compute_full_metrics(predictions, n_no_prediction=0)
+
+    deal_median = statistics.median(sold_ppm2_all) if sold_ppm2_all else None
+    ask_median = statistics.median(pred_ppm2_all) if pred_ppm2_all else None
+    spread_pct: float | None = None
+    if deal_median and ask_median and deal_median > 0:
+        spread_pct = round(100.0 * (ask_median - deal_median) / deal_median, 2)
+    metrics["headline"] = {
+        "deal_median_ppm2": deal_median,
+        "ask_median_ppm2": ask_median,
+        "spread_pct": spread_pct,
+    }
+    return metrics
+
+
+def _predict_full_spine(
+    db: Session,
+    deal: DealSample,
+    est: SimpleNamespace,
+    *,
+    capture: list[dict[str, Any]] | None = None,
+) -> Prediction | None:
     """Predict one deal through the FULL deterministic spine (#1966).
 
     Selects analogs via the replicated tier ladder, pre-fetches the spine inputs
@@ -1156,7 +1343,13 @@ def _predict_full_spine(db: Session, deal: DealSample, est: SimpleNamespace) ->
     the network valuation layers excluded (imv_eval=None, yandex/cian absent).
 
     Returns a Prediction, or None when the spine cannot price the deal
-    (median<=0 or <MIN_CANDIDATES analogs) — mirroring the asking-core skip.
+    (median<=0). Prod parity (#1966): there is NO analog-count skip — a positive
+    median surfaces an estimate (at low confidence for thin samples), matching
+    estimate_quality which never hard-drops a priced result.
+
+    When ``capture`` is a list, every deal that yields a prediction also appends a
+    fully JSON-plain replay record (frozen kwargs + the 3 callables' recorded
+    ``[arg -> return]`` call-lists) — see ``--dump-fixture`` / ``replay_fixture``.
     """
     m = est.m
     settings = est.settings
@@ -1165,15 +1358,21 @@ def _predict_full_spine(db: Session, deal: DealSample, est: SimpleNamespace) ->
 
     # ── Pre-fetch the spine inputs (same calls estimate_quality hoists) ───────
     dkp_raw = m._fetch_dkp_corridor(db, address=deal.address, rooms=deal.rooms, area=deal.area_m2)
-    anchor_comps, anchor_tier = m._fetch_anchor_comps(
-        db,
-        address=deal.address,
-        target_house_id=None,
-        lat=deal.lat,
-        lon=deal.lon,
-        rooms=deal.rooms,
-        area=deal.area_m2,
-    )
+    # #1966 prod parity: same-building anchor pre-fetch is GATED exactly like
+    # estimate_quality (estimator.py L2862-2881) — disabled / no-area / no-address
+    # → ([], None) instead of an unconditional fetch.
+    if settings.estimate_same_building_anchor_enabled and deal.area_m2 and deal.address:
+        anchor_comps, anchor_tier = m._fetch_anchor_comps(
+            db,
+            address=deal.address,
+            target_house_id=None,
+            lat=deal.lat,
+            lon=deal.lon,
+            rooms=deal.rooms,
+            area=deal.area_m2,
+        )
+    else:
+        anchor_comps, anchor_tier = [], None
     imv_anchor = m._fetch_house_imv_anchor(
         db, target_house_id=None, rooms=deal.rooms, area=deal.area_m2
     )
@@ -1191,22 +1390,38 @@ def _predict_full_spine(db: Session, deal: DealSample, est: SimpleNamespace) ->
         confidence="exact",
     )
 
+    # In capture mode each closure RECORDS its (single-arg -> return) call so the
+    # offline replay can rebuild a DB-free stub. The real value is still returned.
+    ratio_calls: list[Any] = []
+    qi_calls: list[Any] = []
+    qis_calls: list[Any] = []
+
     def _ratio_resolver(appm2: float | None) -> tuple[float | None, str | None]:
-        return m._get_asking_sold_ratio(db, deal.rooms, anchor_ppm2=appm2)
+        res = m._get_asking_sold_ratio(db, deal.rooms, anchor_ppm2=appm2)
+        if capture is not None:
+            ratio_calls.append([_sanitize_json(appm2), _sanitize_json(list(res))])
+        return res
 
     def _qi_lookup(q: str) -> tuple[float, int] | None:
-        return m._lookup_quarter_index(
+        res = m._lookup_quarter_index(
             db,
             quarter_cad_number=q,
             min_n_deals=settings.estimate_quarter_index_min_n_deals,
         )
+        if capture is not None:
+            ret = _sanitize_json(list(res) if res is not None else None)
+            qi_calls.append([_sanitize_json(q), ret])
+        return res
 
     def _qis_lookup(qs: list[str]) -> dict[str, float]:
-        return m._lookup_quarter_indexes(
+        res = m._lookup_quarter_indexes(
             db,
             quarter_cad_numbers=qs,
             min_n_deals=settings.estimate_quarter_index_min_n_deals,
         )
+        if capture is not None:
+            qis_calls.append([_sanitize_json(list(qs)), _sanitize_json(dict(res))])
+        return res
 
     pr = m._price_from_inputs(
         listings=listings,
@@ -1235,14 +1450,54 @@ def _predict_full_spine(db: Session, deal: DealSample, est: SimpleNamespace) ->
         dadata_qc_geo=None,
     )
 
-    # Skip when the spine couldn't price it — mirror the asking-core skip.
-    if pr.median_price <= 0 or pr.n_analogs < MIN_CANDIDATES:
+    # #1966 prod parity: skip ONLY when the spine could not price the deal
+    # (median<=0). NO analog-count floor — estimate_quality surfaces any positive
+    # estimate at low confidence. MIN_CANDIDATES still gates the asking-core path.
+    if pr.median_price <= 0:
         return None
 
     es_ppm2 = float(pr.expected_sold_per_m2) if pr.expected_sold_per_m2 is not None else None
     es_price = float(pr.expected_sold_price) if pr.expected_sold_price is not None else None
     r_low = float(pr.expected_sold_range_low) if pr.expected_sold_range_low is not None else None
     r_high = float(pr.expected_sold_range_high) if pr.expected_sold_range_high is not None else None
+
+    if capture is not None:
+        capture.append(
+            {
+                "deal_id": deal.id,
+                "sold_ppm2": float(deal.sold_ppm2),
+                "area_m2": float(deal.area_m2),
+                "rooms": deal.rooms,
+                "deal_date": str(deal.deal_date),
+                "kwargs": {
+                    "listings": _sanitize_json(listings),
+                    "area_m2": float(deal.area_m2),
+                    "rooms": deal.rooms,
+                    "repair_state": None,
+                    "floor": deal.floor,
+                    "total_floors": deal.total_floors,
+                    "target_year": deal.year_built,
+                    "analog_tier": analog_tier,
+                    "fallback_used": fallback_used,
+                    "area_widened": area_widened,
+                    "anchor_comps": _sanitize_json(anchor_comps),
+                    "anchor_tier_fetched": anchor_tier,
+                    "dkp_raw": _sanitize_json(dkp_raw),
+                    "imv_anchor": _sanitize_json(imv_anchor),
+                    "imv_eval": None,
+                    "yandex_val_present": False,
+                    "cian_val_present": False,
+                    "target_house_cadnum": None,
+                    "dadata_coarse": False,
+                    "geo": _sanitize_json(dataclasses.asdict(geo)),
+                    "dadata_qc_geo": None,
+                },
+                "ratio_calls": ratio_calls,
+                "qi_calls": qi_calls,
+                "qis_calls": qis_calls,
+            }
+        )
+
     return Prediction(
         deal_id=deal.id,
         rooms=deal.rooms,
@@ -1390,19 +1645,26 @@ def run_backtest(
     return metrics
 
 
-def run_backtest_full(db: Session, *, sample: int, since: str) -> dict[str, Any]:
+def run_backtest_full(
+    db: Session, *, sample: int, since: str, dump_fixture: str | None = None
+) -> dict[str, Any]:
     """Drive the FULL-spine read-only backtest and return a metrics dict (#1966).
 
     Per deal: load sample → ``_predict_full_spine`` (replicate the analog tier
     ladder + pre-fetch spine inputs → ``_price_from_inputs``) → collect Prediction
     records + no-prediction counts → ``_compute_full_metrics`` (expected_sold
     error overall/per-rooms/per-segment + range-coverage + calibration +
-    sharpness) + a city-wide asking-vs-deal headline spread. No writes.
+    sharpness) + a city-wide asking-vs-deal headline spread. No writes to the DB.
 
     ``--radius`` / ``--rooms-tolerance`` / ``--holdout-split`` do NOT apply here —
     the tier ladder uses the estimator's OWN constants (DEFAULT_RADIUS_M /
     FALLBACK_RADIUS_M) and there is no per-rooms correction block (the spine
     already emits expected_sold via the asking→sold ratio).
+
+    When ``dump_fixture`` is a path, every priced deal's RESOLVED spine inputs +
+    the 3 DB callables' recorded call-lists are frozen into a committed JSON file
+    (schema_version=1) so ``replay_fixture`` can re-score offline against a frozen
+    baseline (the CI regression gate). This is the only path that writes a file.
     """
     est = _import_estimator_full()
     deals = _load_sample(db, sample=sample, since=since)
@@ -1414,10 +1676,11 @@ def run_backtest_full(db: Session, *, sample: int, since: str) -> dict[str, Any]
 
     sold_ppm2_all: list[float] = [d.sold_ppm2 for d in deals]
     pred_ppm2_all: list[float] = []  # asking headline median_ppm2 (priced deals)
+    capture: list[dict[str, Any]] | None = [] if dump_fixture else None
 
     for i, deal in enumerate(deals, start=1):
         try:
-            pr = _predict_full_spine(db, deal, est)
+            pr = _predict_full_spine(db, deal, est, capture=capture)
         except Exception as exc:
             # Read-only: a failed SELECT can poison the tx → rollback so the next
             # deal's queries run clean. Skip this deal (counts as no-prediction).
@@ -1467,9 +1730,47 @@ def run_backtest_full(db: Session, *, sample: int, since: str) -> dict[str, Any]
         "n_no_prediction": n_no_prediction,
         "price_segments_ppm2": [list(seg) for seg in PRICE_SEGMENTS_PPM2],
     }
+
+    if dump_fixture is not None and capture is not None:
+        _write_fixture(dump_fixture, capture=capture, since=since, est=est)
+
     return metrics
 
 
+def _write_fixture(
+    path: str, *, capture: list[dict[str, Any]], since: str, est: SimpleNamespace
+) -> None:
+    """Freeze captured per-deal replay records into a committed JSON fixture.
+
+    ``settings_at_capture`` records every ``estimate_*`` Settings field as an
+    informational snapshot (NOT consulted by ``replay_fixture``). A final
+    recursive ``_sanitize_json`` pass guarantees the whole document is finite +
+    JSON-plain before ``json.dump(allow_nan=False)``.
+    """
+    settings_at_capture = {
+        name: _sanitize_json(getattr(est.settings, name))
+        for name in sorted(type(est.settings).model_fields)
+        if name.startswith("estimate_")
+    }
+    fixture = {
+        "schema_version": FIXTURE_SCHEMA_VERSION,
+        "engine": "full",
+        "since": since,
+        "n_deals": len(capture),
+        "settings_at_capture": settings_at_capture,
+        "deals": capture,
+    }
+    fixture = _sanitize_json(fixture)
+    out_path = Path(path)
+    if out_path.parent != Path():
+        out_path.parent.mkdir(parents=True, exist_ok=True)
+    with out_path.open("w", encoding="utf-8") as fh:
+        json.dump(fixture, fh, ensure_ascii=False, indent=2, allow_nan=False)
+        fh.write("\n")
+    logger.info("dumped fixture: %s (%d deals)", out_path, len(capture))
+    print(f"dumped fixture: {out_path} ({len(capture)} deals)")
+
+
 # --------------------------------------------------------------------------- #
 # Entry point
 # --------------------------------------------------------------------------- #
@@ -1531,27 +1832,76 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
         action="store_true",
         help="Emit machine-readable JSON instead of the text table.",
     )
+    # #1966 PR 3/3 — fixture capture + hermetic replay. --dump-fixture (DB run,
+    # full engine) and --from-fixture (NO DB) are mutually exclusive modes.
+    fixture_mode = p.add_mutually_exclusive_group()
+    fixture_mode.add_argument(
+        "--dump-fixture",
+        metavar="PATH",
+        default=None,
+        help="FULL engine only: freeze each priced deal's resolved _price_from_"
+        "inputs inputs + the 3 DB callables' recorded calls into a committed JSON "
+        "fixture at PATH, so replay_fixture can re-score offline (CI gate).",
+    )
+    fixture_mode.add_argument(
+        "--from-fixture",
+        metavar="PATH",
+        default=None,
+        help="Hermetic replay: load the frozen fixture at PATH, re-run the spine "
+        "offline (ZERO DB) via replay_fixture, and print the metrics JSON. No DB "
+        "connection is opened.",
+    )
+    p.add_argument(
+        "--update-baseline",
+        metavar="OUT",
+        default=None,
+        help="With --from-fixture: also write the replayed metrics to OUT "
+        "(sorted-keys JSON + trailing newline) — regenerates the frozen baseline.",
+    )
     return p.parse_args(argv)
 
 
 def main(argv: list[str] | None = None) -> int:
-    """CLI entry point. Returns the count of matched (predicted) deals."""
+    """CLI entry point. Returns the count of matched (predicted) / replayed deals."""
     args = _parse_args(argv)
+
+    # ── Hermetic replay path (#1966 PR 3/3) — ZERO DB, no SessionLocal opened. ──
+    if args.from_fixture:
+        fixture = json.loads(Path(args.from_fixture).read_text(encoding="utf-8"))
+        metrics = replay_fixture(fixture)
+        print(json.dumps(metrics, ensure_ascii=False, indent=2, sort_keys=True))
+        if args.update_baseline:
+            out = Path(args.update_baseline)
+            out.write_text(
+                json.dumps(metrics, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
+                encoding="utf-8",
+            )
+            logger.info("wrote baseline: %s", out)
+        return len(fixture.get("deals") or [])
+
+    if args.update_baseline:
+        raise SystemExit("--update-baseline requires --from-fixture")
+    if args.dump_fixture and args.engine != "full":
+        raise SystemExit("--dump-fixture is only supported with --engine full")
+
     logger.info(
         "backtest start: engine=%s sample=%d since=%s radius=%dm "
-        "rooms_tolerance=%d holdout_split=%s",
+        "rooms_tolerance=%d holdout_split=%s dump_fixture=%s",
         args.engine,
         args.sample,
         args.since,
         args.radius,
         args.rooms_tolerance,
         args.holdout_split,
+        args.dump_fixture,
     )
 
     db = _session()
     try:
         if args.engine == "full":
-            metrics = run_backtest_full(db, sample=args.sample, since=args.since)
+            metrics = run_backtest_full(
+                db, sample=args.sample, since=args.since, dump_fixture=args.dump_fixture
+            )
         else:
             metrics = run_backtest(
                 db,

tradein-mvp/backend/tests/test_backtest_fixture_roundtrip.py

@@ -0,0 +1,235 @@
+"""Synthetic round-trip test for the hermetic fixture replay (#1966 PR 3/3).
+
+Proves the ``--dump-fixture`` / ``replay_fixture`` machinery end-to-end WITHOUT a
+DB: a hand-crafted in-memory fixture (the exact JSON shape ``--dump-fixture``
+writes) is replayed through the full pricing spine via ``bt.replay_fixture`` and
+the resulting metrics dict is asserted for structure + determinism.
+
+The deals are crafted so every call ``_price_from_inputs`` makes to the 3 injected
+callables is recorded up-front, and so each headline ``median_ppm2`` is an exact,
+predictable value (3 listings → the middle ₽/m²; no anchor / quarter-index / ДКП
+mutation), which is what the recorded ``ratio_calls`` key must match.
+
+NOTE: importing scripts.backtest_estimator → app.services.estimator →
+app.core.config.Settings REQUIRES DATABASE_URL. Set a dummy value BEFORE importing
+app modules (same pattern as tests/test_backtest_estimator.py:19-21). The dummy URL
+is never connected to — replay_fixture opens NO session.
+"""
+
+import os
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+import json
+from typing import Any
+
+from scripts import backtest_estimator as bt
+
+# --------------------------------------------------------------------------- #
+# Fixture builders — hand-crafted, fully predictable per-deal records.
+# --------------------------------------------------------------------------- #
+
+
+def _geo(address: str) -> dict[str, Any]:
+    """A synthetic GeocodeResult dict (the asdict shape replay rebuilds)."""
+    return {
+        "lat": 56.84,
+        "lon": 60.60,
+        "full_address": address,
+        "provider": "cache",
+        "confidence": "exact",
+    }
+
+
+def _deal_record(
+    *,
+    deal_id: int,
+    sold_ppm2: float,
+    area_m2: float,
+    rooms: int,
+    listings: list[dict[str, Any]],
+    anchor_comps: list[dict[str, Any]],
+    anchor_tier_fetched: str | None,
+    ratio_calls: list[Any],
+    qi_calls: list[Any],
+    qis_calls: list[Any],
+    address: str,
+) -> dict[str, Any]:
+    """Assemble one fixture deal record in the exact ``--dump-fixture`` schema."""
+    return {
+        "deal_id": deal_id,
+        "sold_ppm2": sold_ppm2,
+        "area_m2": area_m2,
+        "rooms": rooms,
+        "deal_date": "2025-06-15",
+        "kwargs": {
+            "listings": listings,
+            "area_m2": area_m2,
+            "rooms": rooms,
+            "repair_state": None,
+            "floor": 3,
+            "total_floors": 9,
+            "target_year": 2010,
+            "analog_tier": "W",
+            "fallback_used": False,
+            "area_widened": False,
+            "anchor_comps": anchor_comps,
+            "anchor_tier_fetched": anchor_tier_fetched,
+            "dkp_raw": None,
+            "imv_anchor": None,
+            "imv_eval": None,
+            "yandex_val_present": False,
+            "cian_val_present": False,
+            "target_house_cadnum": None,
+            "dadata_coarse": False,
+            "geo": _geo(address),
+            "dadata_qc_geo": None,
+        },
+        "ratio_calls": ratio_calls,
+        "qi_calls": qi_calls,
+        "qis_calls": qis_calls,
+    }
+
+
+def _build_fixture() -> dict[str, Any]:
+    """3 hand-crafted deals spanning эконом / бизнес / элит SOLD segments.
+
+    deal 1: plain — no anchor, no quarter index → single ratio call.
+    deal 2: a listing carries a cadastral number → one recorded quarter-index
+            call (returns null, leaves the median untouched) + a ratio call.
+    deal 3: carries non-empty anchor_comps (2 comps < min_comps=4 → anchor never
+            fires, so the median stays the radius median) + a ratio call.
+    """
+    # ── deal 1 — median of [90k, 100k, 110k] = 100k → ratio_resolver(100000.0). ──
+    deal1 = _deal_record(
+        deal_id=1,
+        sold_ppm2=100_000.0,  # SOLD эконом (< 120k)
+        area_m2=50.0,
+        rooms=1,
+        listings=[
+            {"price_per_m2": 90_000.0, "source": "avito"},
+            {"price_per_m2": 100_000.0, "source": "avito"},
+            {"price_per_m2": 110_000.0, "source": "avito"},
+        ],
+        anchor_comps=[],
+        anchor_tier_fetched=None,
+        ratio_calls=[[100_000.0, [0.95, "per_rooms_all"]]],
+        qi_calls=[],
+        qis_calls=[],
+        address="ул. Тестовая, 1",
+    )
+
+    # ── deal 2 — median 150k; first lot's cadnum → quarter "66:41:0204016". ──
+    deal2 = _deal_record(
+        deal_id=2,
+        sold_ppm2=200_000.0,  # SOLD бизнес (160k..220k)
+        area_m2=60.0,
+        rooms=2,
+        listings=[
+            {
+                "price_per_m2": 140_000.0,
+                "source": "cian",
+                "building_cadastral_number": "66:41:0204016:350",
+            },
+            {"price_per_m2": 150_000.0, "source": "cian"},
+            {"price_per_m2": 160_000.0, "source": "cian"},
+        ],
+        anchor_comps=[],
+        anchor_tier_fetched=None,
+        # quarter-index lookup returns null → spine leaves the median untouched.
+        qi_calls=[["66:41:0204016", None]],
+        qis_calls=[],
+        ratio_calls=[[150_000.0, [0.90, "per_rooms_all"]]],
+        address="ул. Тестовая, 2",
+    )
+
+    # ── deal 3 — median 310k; anchor_comps present but below min_comps → no fire. ──
+    deal3 = _deal_record(
+        deal_id=3,
+        sold_ppm2=290_000.0,  # SOLD элит (220k..300k)
+        area_m2=80.0,
+        rooms=3,
+        listings=[
+            {"price_per_m2": 300_000.0, "source": "yandex"},
+            {"price_per_m2": 310_000.0, "source": "yandex"},
+            {"price_per_m2": 320_000.0, "source": "yandex"},
+        ],
+        anchor_comps=[
+            {"price_per_m2": 305_000.0, "area_m2": 80.0, "rooms": 3, "floor": 5, "total_floors": 9},
+            {"price_per_m2": 308_000.0, "area_m2": 82.0, "rooms": 3, "floor": 6, "total_floors": 9},
+        ],
+        anchor_tier_fetched=None,
+        ratio_calls=[[310_000.0, [0.92, "per_rooms_tier:high"]]],
+        qi_calls=[],
+        qis_calls=[],
+        address="ул. Тестовая, 3",
+    )
+
+    deals = [deal1, deal2, deal3]
+    return {
+        "schema_version": bt.FIXTURE_SCHEMA_VERSION,
+        "engine": "full",
+        "since": "2025-06-01",
+        "n_deals": len(deals),
+        "settings_at_capture": {},
+        "deals": deals,
+    }
+
+
+# --------------------------------------------------------------------------- #
+# Tests
+# --------------------------------------------------------------------------- #
+
+
+def test_replay_fixture_structure_and_keys() -> None:
+    fixture = _build_fixture()
+    metrics = bt.replay_fixture(fixture)
+
+    # Keeps the non-volatile metric blocks ...
+    for key in (
+        "expected_sold",
+        "range_coverage",
+        "calibration",
+        "sharpness",
+        "confidence_order",
+        "headline",
+    ):
+        assert key in metrics, f"missing metric block: {key}"
+    # ... and DROPS the volatile params block.
+    assert "params" not in metrics
+
+    # Every crafted deal resolves a ratio → an expected_sold row, so overall n
+    # equals the number of deals in the fixture.
+    assert metrics["expected_sold"]["overall"]["n"] == len(fixture["deals"])
+
+
+def test_replay_fixture_is_deterministic() -> None:
+    fixture = _build_fixture()
+    first = bt.replay_fixture(fixture)
+    second = bt.replay_fixture(fixture)
+    # Byte-identical across two independent replays (no DB, no RNG, no caches).
+    assert json.dumps(first, ensure_ascii=False, sort_keys=True) == json.dumps(
+        second, ensure_ascii=False, sort_keys=True
+    )
+
+
+def test_replay_fixture_segments_span_multiple_bands() -> None:
+    # The 3 deals sit in distinct SOLD price-segments (эконом / бизнес / элит),
+    # so the per-segment expected_sold breakdown must show ≥ 3 non-empty bands.
+    metrics = bt.replay_fixture(_build_fixture())
+    per_segment = metrics["expected_sold"]["per_segment"]
+    non_empty = [label for label, m in per_segment.items() if m["n"] > 0]
+    assert len(non_empty) >= 3
+
+
+def test_unseen_callable_arg_raises_keyerror() -> None:
+    # A fixture whose recorded ratio call does not match the median the spine
+    # actually computes must surface a clear KeyError (not a silent wrong answer).
+    fixture = _build_fixture()
+    fixture["deals"][0]["ratio_calls"] = [[999_999.0, [0.9, "per_rooms_all"]]]
+    try:
+        bt.replay_fixture(fixture)
+    except KeyError as exc:
+        assert "ratio_resolver" in str(exc)
+    else:
+        raise AssertionError("expected KeyError for an unrecorded ratio_resolver arg")