diff --git a/tradein-mvp/backend/scripts/backtest_estimator.py b/tradein-mvp/backend/scripts/backtest_estimator.py index 0508c85a..2d7a348e 100644 --- a/tradein-mvp/backend/scripts/backtest_estimator.py +++ b/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 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, diff --git a/tradein-mvp/backend/tests/test_backtest_fixture_roundtrip.py b/tradein-mvp/backend/tests/test_backtest_fixture_roundtrip.py new file mode 100644 index 00000000..6b35d1d4 --- /dev/null +++ b/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")