1940 lines
76 KiB
Python
1940 lines
76 KiB
Python
"""Backtest harness — measures the estimator's accuracy vs real ДКП sold prices.
|
||
|
||
Forgejo issues #648 (asking-core) + #1966 (full spine). **STRICTLY READ-ONLY**:
|
||
this script issues only SELECT queries against prod. It never INSERTs, UPDATEs,
|
||
or runs DDL.
|
||
|
||
TWO ENGINES (``--engine``)
|
||
--------------------------
|
||
* ``full`` (DEFAULT, #1966) — runs the **full deterministic pricing spine**
|
||
via ``estimator._price_from_inputs``: the same analog tier ladder
|
||
(Tier 0 cohort → Tier A 1 km → wide 2 km → widearea ±25 %), same-building
|
||
anchor comps, ДКП corridor clamp/floor, house Avito-IMV anchor, quarter
|
||
price index, and the asking→sold ratio that yields ``expected_sold``. The
|
||
PRODUCT-relevant prediction is ``expected_sold_per_m2`` (we also keep the
|
||
asking headline ``median_ppm2`` so both are reported).
|
||
* ``asking-core`` (#648) — the legacy asking-median + Tukey-IQR CORE only,
|
||
plus the in-/out-of-sample per-rooms asking→sold correction block. Kept for
|
||
comparison against the full spine.
|
||
|
||
WHAT IT MEASURES (full spine)
|
||
-----------------------------
|
||
Ground truth is `deals` — registered rosreestr ДКП sales (source='rosreestr')
|
||
that carry geom + price_per_m2 + rooms + area_m2 + floor/total_floors/year/type.
|
||
For a held-out sample of ДКП deals we, per deal:
|
||
|
||
1. Replicate ``estimate_quality``'s analog tier ladder with ``_fetch_analogs``
|
||
(same constants: DEFAULT_RADIUS_M, FALLBACK_RADIUS_M, MIN_ANALOGS_TIER_0,
|
||
the <5/<3 widen thresholds) and pre-fetch the spine inputs
|
||
(``_fetch_anchor_comps``, ``_fetch_dkp_corridor``,
|
||
``_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) 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)
|
||
--------------------
|
||
* EXPECTED_SOLD signed error: median bias % + MAPE (median |%|), OVERALL +
|
||
per-rooms + per-price-segment (эконом/комфорт/бизнес/элит/премиум).
|
||
* RANGE COVERAGE: fraction of deals whose actual sold TOTAL (sold_ppm2 ×
|
||
area) falls inside the predicted expected_sold RUB range — OVERALL +
|
||
per-confidence bucket.
|
||
* CONFIDENCE CALIBRATION: n / coverage / MAPE per confidence bucket
|
||
(high should be tighter & more accurate — surfaces R2 risk).
|
||
* SHARPNESS: median relative range width (high-low)/point — guards against
|
||
gaming coverage with very wide ranges.
|
||
* City-wide deal_median_ppm2 vs ask_median_ppm2 headline spread.
|
||
|
||
CAVEATS (read these before trusting the numbers)
|
||
-----------------------------------------------
|
||
(a) TIME MISMATCH — this compares **CURRENT** active listings against
|
||
**PAST** sold deals. It is NOT a point-in-time backtest: a deal closed
|
||
in 2025-06 is judged against listings active today, so asking prices
|
||
drift away from the historical sold price as the market moves. Treat the
|
||
output as a REGRESSION BASELINE (relative change between runs / engines),
|
||
NOT absolute truth. A faithful point-in-time backtest needs
|
||
`listing_source_snapshots` (#570).
|
||
(b) NETWORK LAYERS EXCLUDED — the spine runs offline, so the on-demand
|
||
network valuation sources (Avito-IMV on-demand eval, Yandex valuation,
|
||
Cian valuation) are NOT exercised (``imv_eval=None``,
|
||
``yandex/cian_val_present=False``). The populated `house_imv_evaluations`
|
||
anchor IS injected, but it keys on house_id which the harness does not
|
||
resolve (target_house_id=None) → in practice no IMV anchor fires. Real
|
||
estimate accuracy with the network layers may differ.
|
||
(c) ДКП ≠ TRUE MARKET — a registered ДКП price is what the parties declared
|
||
to rosreestr; it can diverge from the genuine transaction price (tax
|
||
optimisation, related-party sales, etc.).
|
||
|
||
PERFORMANCE
|
||
-----------
|
||
The full spine issues several spatial SELECTs per sampled deal (the tier ladder
|
||
+ anchor/corridor/imv pre-fetches), so runtime scales with `--sample`. The
|
||
default (300) is fine; large samples (thousands) are slow.
|
||
|
||
USAGE
|
||
-----
|
||
DATABASE_URL=postgresql+psycopg://... \
|
||
python -m scripts.backtest_estimator --sample 300 --since 2025-06-01
|
||
|
||
# legacy asking-median core + correction block:
|
||
python -m scripts.backtest_estimator --engine asking-core
|
||
|
||
# machine-readable:
|
||
python -m scripts.backtest_estimator --json
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import dataclasses
|
||
import gzip
|
||
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
|
||
|
||
from sqlalchemy import text
|
||
from sqlalchemy.orm import Session
|
||
|
||
|
||
def _import_estimator() -> tuple[Any, Any]:
|
||
"""Lazy import of the estimator's pure funcs (_filter_outliers, _percentile).
|
||
|
||
Deferred so `--help` / the pure-metric unit tests don't pull
|
||
app.core.config.Settings (which fail-fasts when DATABASE_URL is unset).
|
||
Supports both `python -m scripts.backtest_estimator` and stand-alone runs.
|
||
"""
|
||
try:
|
||
from app.services.estimator import ( # type: ignore[import-not-found]
|
||
_filter_outliers,
|
||
_percentile,
|
||
)
|
||
except ImportError: # pragma: no cover — fallback for adhoc invocation
|
||
import sys
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||
from app.services.estimator import _filter_outliers, _percentile
|
||
return _filter_outliers, _percentile
|
||
|
||
|
||
def _import_estimator_full() -> SimpleNamespace:
|
||
"""Lazy import of the FULL pricing spine + the helpers it needs (#1966).
|
||
|
||
Returns a namespace bundling the estimator module (``m``), its ``settings``
|
||
singleton, and the ``GeocodeResult`` dataclass. Deferred for the same reason
|
||
as _import_estimator — importing app.services.estimator pulls
|
||
app.core.config.Settings, which fail-fasts when DATABASE_URL is unset, so we
|
||
keep it out of `--help` / the pure-metric unit tests.
|
||
|
||
Everything the full-spine prediction path calls lives on ``ns.m``:
|
||
_price_from_inputs, _fetch_analogs, _fetch_anchor_comps, _fetch_dkp_corridor,
|
||
_fetch_house_imv_anchor, _get_asking_sold_ratio, _lookup_quarter_index(es),
|
||
_target_cohort_range, and the DEFAULT_RADIUS_M / FALLBACK_RADIUS_M /
|
||
MIN_ANALOGS_TIER_0 constants.
|
||
"""
|
||
try:
|
||
from app.services import estimator as est_mod # type: ignore[import-not-found]
|
||
except ImportError: # pragma: no cover — fallback for adhoc invocation
|
||
import sys
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||
from app.services import estimator as est_mod
|
||
# settings + GeocodeResult are re-exported as module attributes of estimator
|
||
# (imported there at module level), so we read them off the same object.
|
||
return SimpleNamespace(
|
||
m=est_mod,
|
||
settings=est_mod.settings,
|
||
GeocodeResult=est_mod.GeocodeResult,
|
||
)
|
||
|
||
|
||
def _session() -> Session:
|
||
"""Lazy SessionLocal factory — see _import_estimator for why it's deferred."""
|
||
try:
|
||
from app.core.db import SessionLocal # type: ignore[import-not-found]
|
||
except ImportError: # pragma: no cover — fallback for adhoc invocation
|
||
import sys
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||
from app.core.db import SessionLocal
|
||
return SessionLocal()
|
||
|
||
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format="%(asctime)s %(levelname)s %(name)s %(message)s",
|
||
)
|
||
logger = logging.getLogger("backtest_estimator")
|
||
|
||
# Price-per-m² sanity band — shared by the deal sample and the listings
|
||
# subquery. Mirrors the estimator's working range for EKB вторичка and drops
|
||
# obvious data-entry garbage / commercial outliers.
|
||
PPM2_MIN = 30_000
|
||
PPM2_MAX = 600_000
|
||
|
||
# Minimum surviving candidates required to emit a prediction. Below this the
|
||
# median is too noisy to be meaningful — count the deal as "no analogs".
|
||
MIN_CANDIDATES = 3
|
||
|
||
# Room buckets for the per-rooms breakdown. 0 = студия; the top bucket is "4+".
|
||
ROOM_BUCKETS: tuple[int, ...] = (0, 1, 2, 3, 4)
|
||
|
||
# Minimum matched deals a rooms-bucket must hold before we trust its OWN
|
||
# asking→sold ratio. Thin buckets fall back to the global (all-buckets) ratio,
|
||
# whose larger n keeps the correction stable instead of overfitting a handful
|
||
# of deals. See _derive_room_ratios.
|
||
MIN_BUCKET = 20
|
||
|
||
# Confidence buckets the spine emits (estimator._compute_confidence /
|
||
# anchor / coarse-geo gates only ever return one of these three). Used by the
|
||
# range-coverage + calibration breakdowns. Any unexpected value → "other".
|
||
CONFIDENCE_BUCKETS: tuple[str, ...] = ("high", "medium", "low")
|
||
|
||
# Price-segment bands by ₽/m² (EKB вторичка). The estimator has NO reusable
|
||
# price-tier constant — `listing_segment` is categorical (vtorichka/novostroyki)
|
||
# and DEAL_MIN/MAX_PPM2 are sanity bounds, not class bands — so these are
|
||
# defined here. TUNABLE: rough EKB market tiers (эконом < комфорт < бизнес <
|
||
# элит < премиум). Each entry is (label, upper_bound_exclusive); a value lands
|
||
# in the first band whose upper bound it is below; the last band catches the
|
||
# tail (+inf). Verified against config notes: p99.9 deals ≈ 500k, премиум ~680k.
|
||
PRICE_SEGMENTS_PPM2: tuple[tuple[str, float], ...] = (
|
||
("эконом", 120_000.0),
|
||
("комфорт", 160_000.0),
|
||
("бизнес", 220_000.0),
|
||
("элит", 300_000.0),
|
||
("премиум", float("inf")),
|
||
)
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Data carriers
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
@dataclass
|
||
class DealSample:
|
||
"""One held-out ДКП deal to backtest against.
|
||
|
||
The full spine (#1966) needs the extra unit/house fields (area, address,
|
||
floor, total_floors, year_built, house_type) to replicate ``_fetch_analogs``
|
||
+ the anchor/corridor pre-fetches; the asking-core engine ignores them.
|
||
"""
|
||
|
||
id: int
|
||
lon: float
|
||
lat: float
|
||
rooms: int
|
||
sold_ppm2: float
|
||
deal_date: Any # datetime.date | None — carried through for reporting only
|
||
area_m2: float = 0.0
|
||
address: str | None = None
|
||
floor: int | None = None
|
||
total_floors: int | None = None
|
||
year_built: int | None = None
|
||
house_type: str | None = None
|
||
|
||
|
||
@dataclass
|
||
class Prediction:
|
||
"""One full-spine prediction for a deal that DID get priced (#1966).
|
||
|
||
Carries both the asking headline (``median_ppm2``) and the product-relevant
|
||
``expected_sold_*`` outputs. Range fields are RUB TOTALS (the spine emits
|
||
expected_sold_range_low/high as totals, not ₽/m²). ``expected_sold_*`` may be
|
||
None when the spine produced a headline but no asking→sold ratio resolved.
|
||
"""
|
||
|
||
deal_id: int
|
||
rooms: int
|
||
area_m2: float
|
||
sold_ppm2: float
|
||
median_ppm2: float # asking headline ₽/m²
|
||
confidence: str
|
||
anchor_tier: str | None
|
||
expected_sold_ppm2: float | None
|
||
expected_sold_price: float | None # RUB total point
|
||
range_low: float | None # expected_sold_range_low — RUB total
|
||
range_high: float | None # expected_sold_range_high — RUB total
|
||
|
||
@property
|
||
def sold_total(self) -> float:
|
||
"""Realised sold price as a RUB total (sold ₽/m² × area)."""
|
||
return self.sold_ppm2 * self.area_m2
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Pure metric helpers — NO DB. These are unit-tested in
|
||
# tests/test_backtest_estimator.py without a live database.
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
def _rooms_label(rooms: int) -> str:
|
||
"""Human label for a room bucket: 0 → 'студия', 4 → '4+', else '<n>к'."""
|
||
if rooms <= 0:
|
||
return "студия"
|
||
if rooms >= ROOM_BUCKETS[-1]:
|
||
return f"{ROOM_BUCKETS[-1]}+"
|
||
return f"{rooms}к"
|
||
|
||
|
||
def _bucketize_rooms(rooms: int) -> int:
|
||
"""Clamp a raw room count into a ROOM_BUCKETS key (4+ collapse to 4)."""
|
||
if rooms <= 0:
|
||
return 0
|
||
return min(rooms, ROOM_BUCKETS[-1])
|
||
|
||
|
||
def _errors_summary(signed_errors: list[float]) -> dict[str, Any]:
|
||
"""Bias / MAPE / spread for a list of signed error percentages.
|
||
|
||
- median_bias_pct = median(signed_errors) — systematic over/under-predict
|
||
- mape_pct = median(|signed_errors|) — typical magnitude (robust;
|
||
we use the MEDIAN absolute error, matching the brief's
|
||
MAPE-as-median-abs-error definition rather than mean)
|
||
- p25 / p75 = quartiles of the SIGNED error (skew of the bias)
|
||
|
||
Empty input → all-None (caller decides how to render "no data").
|
||
"""
|
||
n = len(signed_errors)
|
||
if n == 0:
|
||
return {
|
||
"n": 0,
|
||
"median_bias_pct": None,
|
||
"mape_pct": None,
|
||
"p25_pct": None,
|
||
"p75_pct": None,
|
||
}
|
||
abs_errors = [abs(e) for e in signed_errors]
|
||
_, _percentile = _import_estimator() # reuse estimator's interpolation percentile
|
||
return {
|
||
"n": n,
|
||
"median_bias_pct": round(statistics.median(signed_errors), 2),
|
||
"mape_pct": round(statistics.median(abs_errors), 2),
|
||
"p25_pct": round(_percentile(sorted(signed_errors), 0.25), 2),
|
||
"p75_pct": round(_percentile(sorted(signed_errors), 0.75), 2),
|
||
}
|
||
|
||
|
||
def _compute_metrics(
|
||
rows: list[tuple[float, float, int]],
|
||
*,
|
||
n_no_analogs: int = 0,
|
||
per_rooms_no_analogs: dict[int, int] | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Aggregate predicted-vs-sold rows into overall + per-rooms metrics.
|
||
|
||
Each input row is (pred_ppm2, sold_ppm2, rooms) for a deal that DID get a
|
||
prediction. `n_no_analogs` / `per_rooms_no_analogs` carry the skipped-deal
|
||
counts so the report can show match coverage; they don't affect the error
|
||
stats (those are computed only over matched deals).
|
||
|
||
Returns a dict::
|
||
|
||
{
|
||
"overall": {n, n_no_analogs, median_bias_pct, mape_pct, p25_pct, p75_pct},
|
||
"per_rooms": {
|
||
0: {label, n, n_no_analogs, median_bias_pct, mape_pct, p25_pct, p75_pct},
|
||
...
|
||
},
|
||
}
|
||
|
||
Pure: no DB, no I/O. signed_error_pct = 100*(pred - sold)/sold per row.
|
||
Rows with sold_ppm2 <= 0 are dropped (cannot divide) — defensive; the SQL
|
||
sample already excludes them.
|
||
"""
|
||
per_rooms_no_analogs = per_rooms_no_analogs or {}
|
||
|
||
overall_signed: list[float] = []
|
||
by_bucket_signed: dict[int, list[float]] = {b: [] for b in ROOM_BUCKETS}
|
||
|
||
for pred_ppm2, sold_ppm2, rooms in rows:
|
||
if sold_ppm2 <= 0:
|
||
continue
|
||
signed = 100.0 * (pred_ppm2 - sold_ppm2) / sold_ppm2
|
||
overall_signed.append(signed)
|
||
by_bucket_signed[_bucketize_rooms(rooms)].append(signed)
|
||
|
||
overall = _errors_summary(overall_signed)
|
||
overall["n_no_analogs"] = n_no_analogs
|
||
|
||
per_rooms: dict[int, dict[str, Any]] = {}
|
||
for bucket in ROOM_BUCKETS:
|
||
summary = _errors_summary(by_bucket_signed[bucket])
|
||
summary["label"] = _rooms_label(bucket)
|
||
summary["n_no_analogs"] = per_rooms_no_analogs.get(bucket, 0)
|
||
per_rooms[bucket] = summary
|
||
|
||
return {"overall": overall, "per_rooms": per_rooms}
|
||
|
||
|
||
def _derive_room_ratios(
|
||
rows: list[tuple[float, float, int]],
|
||
*,
|
||
min_bucket: int = MIN_BUCKET,
|
||
) -> tuple[dict[int, float], dict[str, Any]]:
|
||
"""Derive a per-rooms asking→sold correction ratio from matched deals.
|
||
|
||
For each row ``(pred_ask_ppm2, sold_ppm2, rooms)`` we want a multiplier that
|
||
maps the estimator's asking-median prediction onto the realised SOLD price::
|
||
|
||
ratio[bucket] = median(sold_ppm2) / median(pred_ask_ppm2)
|
||
|
||
computed over the deals in that ROOM_BUCKETS bucket. A bucket holding fewer
|
||
than ``min_bucket`` matched deals (or whose own median pred_ask is ≤ 0)
|
||
falls back to the GLOBAL ratio — median(all sold) / median(all pred_ask) —
|
||
so a handful of noisy deals can't overfit a per-rooms factor.
|
||
|
||
Returns ``(ratios, meta)`` where:
|
||
|
||
- ``ratios``: ``{bucket: float}`` for every bucket that has data (either
|
||
its own ratio or the global fallback). Buckets with no matched deals at
|
||
all are omitted (nothing to correct).
|
||
- ``meta``: ``{"global_ratio": float | None, "fallback_buckets": [int],
|
||
"bucket_n": {bucket: int}}`` — diagnostics for the report.
|
||
|
||
!!! IN-SAMPLE WARNING
|
||
These ratios are DERIVED from the same rows they will later be evaluated
|
||
on (unless --holdout-split feeds disjoint rows). Used that way they make
|
||
the corrected bias near-zero BY CONSTRUCTION — that proves the MECHANISM
|
||
(a per-rooms multiplier removes the systematic asking→sold gap), NOT
|
||
out-of-sample accuracy. The production ratio (Stage 2) is fit over a
|
||
separate window and A/B'd on held-out deals.
|
||
|
||
Pure: no DB, no I/O. Guards div-by-zero and empty input (→ ({}, meta)).
|
||
"""
|
||
_, _percentile = _import_estimator() # reuse estimator's interpolation percentile
|
||
|
||
def _median(values: list[float]) -> float | None:
|
||
if not values:
|
||
return None
|
||
return _percentile(sorted(values), 0.5)
|
||
|
||
# Collect valid (pred>0, sold>0) ppm² pairs per bucket and globally.
|
||
by_bucket_pred: dict[int, list[float]] = {b: [] for b in ROOM_BUCKETS}
|
||
by_bucket_sold: dict[int, list[float]] = {b: [] for b in ROOM_BUCKETS}
|
||
all_pred: list[float] = []
|
||
all_sold: list[float] = []
|
||
|
||
for pred_ppm2, sold_ppm2, rooms in rows:
|
||
if pred_ppm2 <= 0 or sold_ppm2 <= 0:
|
||
continue
|
||
bucket = _bucketize_rooms(rooms)
|
||
by_bucket_pred[bucket].append(pred_ppm2)
|
||
by_bucket_sold[bucket].append(sold_ppm2)
|
||
all_pred.append(pred_ppm2)
|
||
all_sold.append(sold_ppm2)
|
||
|
||
bucket_n = {b: len(by_bucket_pred[b]) for b in ROOM_BUCKETS}
|
||
|
||
# Global fallback ratio — None when there's no data or pred median is 0.
|
||
global_pred_med = _median(all_pred)
|
||
global_sold_med = _median(all_sold)
|
||
global_ratio: float | None = None
|
||
if global_pred_med and global_pred_med > 0 and global_sold_med is not None:
|
||
global_ratio = global_sold_med / global_pred_med
|
||
|
||
ratios: dict[int, float] = {}
|
||
fallback_buckets: list[int] = []
|
||
for bucket in ROOM_BUCKETS:
|
||
n = bucket_n[bucket]
|
||
if n == 0:
|
||
continue # no deals in this bucket → nothing to correct
|
||
pred_med = _median(by_bucket_pred[bucket])
|
||
sold_med = _median(by_bucket_sold[bucket])
|
||
if n >= min_bucket and pred_med and pred_med > 0 and sold_med is not None:
|
||
ratios[bucket] = sold_med / pred_med
|
||
elif global_ratio is not None:
|
||
ratios[bucket] = global_ratio
|
||
fallback_buckets.append(bucket)
|
||
# else: no own ratio AND no global fallback → leave bucket uncorrected.
|
||
|
||
meta = {
|
||
"global_ratio": (round(global_ratio, 4) if global_ratio is not None else None),
|
||
"fallback_buckets": fallback_buckets,
|
||
"bucket_n": bucket_n,
|
||
"min_bucket": min_bucket,
|
||
}
|
||
return ratios, meta
|
||
|
||
|
||
def _apply_ratios(
|
||
rows: list[tuple[float, float, int]],
|
||
ratios: dict[int, float],
|
||
) -> list[tuple[float, float, int]]:
|
||
"""Apply per-rooms ratios → corrected rows for re-scoring via _compute_metrics.
|
||
|
||
``pred_sold = pred_ask * ratio[bucket]``. A bucket with no ratio (e.g. no
|
||
global fallback was available) leaves its rows unchanged so they still count
|
||
in the corrected block rather than vanishing. Pure: no DB.
|
||
"""
|
||
out: list[tuple[float, float, int]] = []
|
||
for pred_ask, sold, rooms in rows:
|
||
ratio = ratios.get(_bucketize_rooms(rooms), 1.0)
|
||
out.append((pred_ask * ratio, sold, rooms))
|
||
return out
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Full-spine metric helpers (#1966) — PURE, no DB. Unit-tested.
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
def _bucketize_confidence(confidence: str) -> str:
|
||
"""Map a confidence string to a calibration bucket (unknown → 'other')."""
|
||
return confidence if confidence in CONFIDENCE_BUCKETS else "other"
|
||
|
||
|
||
def _segment_label(ppm2: float) -> str:
|
||
"""Price-segment label for a ₽/m² value (see PRICE_SEGMENTS_PPM2 bands)."""
|
||
for label, upper in PRICE_SEGMENTS_PPM2:
|
||
if ppm2 < upper:
|
||
return label
|
||
return PRICE_SEGMENTS_PPM2[-1][0] # +inf tail — unreachable, defensive
|
||
|
||
|
||
def _segment_metrics(rows: list[tuple[float, float]]) -> dict[str, dict[str, Any]]:
|
||
"""Per-price-segment signed-error summary, bucketed by the SOLD ₽/m².
|
||
|
||
Each input row is ``(pred_ppm2, sold_ppm2)``. We bucket by the segment of the
|
||
SOLD price (ground truth), compute signed_error_pct = 100*(pred-sold)/sold,
|
||
and run `_errors_summary` per band. Rows with sold<=0 are dropped (can't
|
||
divide). Every band in PRICE_SEGMENTS_PPM2 is present (n=0 when empty) so the
|
||
report renders a stable table. Pure: no DB.
|
||
"""
|
||
by_seg: dict[str, list[float]] = {label: [] for label, _ in PRICE_SEGMENTS_PPM2}
|
||
for pred, sold in rows:
|
||
if sold <= 0:
|
||
continue
|
||
by_seg[_segment_label(sold)].append(100.0 * (pred - sold) / sold)
|
||
out: dict[str, dict[str, Any]] = {}
|
||
for label, _ in PRICE_SEGMENTS_PPM2:
|
||
out[label] = _errors_summary(by_seg[label])
|
||
return out
|
||
|
||
|
||
def _range_coverage(rows: list[tuple[float, float, float]]) -> dict[str, Any]:
|
||
"""Coverage = fraction of (actual, low, high) rows with low<=actual<=high.
|
||
|
||
Boundary INCLUSIVE (a deal sitting exactly on the range edge counts as
|
||
covered). Empty input → coverage_pct None. Pure: no DB.
|
||
"""
|
||
n = len(rows)
|
||
if n == 0:
|
||
return {"n": 0, "n_covered": 0, "coverage_pct": None}
|
||
covered = sum(1 for actual, lo, hi in rows if lo <= actual <= hi)
|
||
return {"n": n, "n_covered": covered, "coverage_pct": round(100.0 * covered / n, 2)}
|
||
|
||
|
||
def _sharpness(rows: list[tuple[float, float, float]]) -> dict[str, Any]:
|
||
"""Median relative range width = median((high-low)/point) over (point,lo,hi).
|
||
|
||
Guards against gaming coverage with arbitrarily wide ranges: a model can
|
||
always cover 100% by predicting an infinite range, so we report how WIDE the
|
||
ranges are relative to the point estimate. Rows with point<=0 are dropped.
|
||
Empty → median_rel_width None. Pure: no DB.
|
||
"""
|
||
widths = [(hi - lo) / point for point, lo, hi in rows if point > 0]
|
||
if not widths:
|
||
return {"n": 0, "median_rel_width": None}
|
||
_, _percentile = _import_estimator() # reuse estimator's interpolation percentile
|
||
return {"n": len(widths), "median_rel_width": round(_percentile(sorted(widths), 0.5), 4)}
|
||
|
||
|
||
def _calibration_metrics(
|
||
rows: list[tuple[str, float | None, bool | None]],
|
||
*,
|
||
bucket_order: list[str] | None = None,
|
||
) -> dict[str, dict[str, Any]]:
|
||
"""Per-confidence n / coverage% / MAPE% — surfaces the R2 risk.
|
||
|
||
Each input row is ``(confidence, signed_error_pct | None, covered | None)``:
|
||
- ``n`` = predictions in this confidence bucket (always counted).
|
||
- ``coverage_pct``= 100 * mean(covered) over rows where covered is not None.
|
||
- ``mape_pct`` = median(|signed_error_pct|) over rows where signed is not
|
||
None (the brief's median-abs-error MAPE definition).
|
||
|
||
A well-calibrated estimator shows HIGH confidence = higher coverage AND lower
|
||
MAPE than LOW. The canonical high/medium/low buckets are always present (n=0
|
||
when empty); an "other" bucket appears only if an off-enum confidence shows
|
||
up. Pure: no DB.
|
||
"""
|
||
order = list(bucket_order) if bucket_order is not None else list(CONFIDENCE_BUCKETS)
|
||
coll: dict[str, dict[str, Any]] = {}
|
||
for confidence, signed, covered in rows:
|
||
b = _bucketize_confidence(confidence)
|
||
slot = coll.setdefault(b, {"n": 0, "abs_errors": [], "covered": []})
|
||
slot["n"] += 1
|
||
if signed is not None:
|
||
slot["abs_errors"].append(abs(signed))
|
||
if covered is not None:
|
||
slot["covered"].append(covered)
|
||
if b not in order:
|
||
order.append(b)
|
||
|
||
out: dict[str, dict[str, Any]] = {}
|
||
for b in order:
|
||
slot = coll.get(b, {"n": 0, "abs_errors": [], "covered": []})
|
||
cov = slot["covered"]
|
||
abs_errors = slot["abs_errors"]
|
||
n_covered = sum(1 for c in cov if c)
|
||
out[b] = {
|
||
"n": slot["n"],
|
||
"n_covered": n_covered,
|
||
"coverage_pct": (round(100.0 * n_covered / len(cov), 2) if cov else None),
|
||
"mape_pct": (round(statistics.median(abs_errors), 2) if abs_errors else None),
|
||
}
|
||
return out
|
||
|
||
|
||
def _expected_sold_metrics(
|
||
rows: list[tuple[float, float, int]],
|
||
*,
|
||
n_no_prediction: int = 0,
|
||
per_rooms_no_prediction: dict[int, int] | None = None,
|
||
) -> dict[str, Any]:
|
||
"""expected_sold error block: overall + per-rooms (via _compute_metrics) + per-segment.
|
||
|
||
Each input row is ``(expected_sold_ppm2, sold_ppm2, rooms)``. Reuses the
|
||
existing `_compute_metrics` for the overall/per-rooms split (so the no-data
|
||
counts ride along as ``n_no_analogs`` — here meaning "no prediction") and
|
||
attaches a per-price-segment breakdown keyed by the SOLD ₽/m². Pure: no DB.
|
||
"""
|
||
m = _compute_metrics(
|
||
rows,
|
||
n_no_analogs=n_no_prediction,
|
||
per_rooms_no_analogs=per_rooms_no_prediction,
|
||
)
|
||
m["per_segment"] = _segment_metrics([(pred, sold) for pred, sold, _ in rows])
|
||
return m
|
||
|
||
|
||
def _compute_full_metrics(
|
||
predictions: list[Prediction],
|
||
*,
|
||
n_no_prediction: int = 0,
|
||
per_rooms_no_prediction: dict[int, int] | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Aggregate full-spine Prediction records into the complete metrics dict.
|
||
|
||
Blocks (all over deals that DID get an expected_sold, except where noted):
|
||
- ``expected_sold`` : signed bias / MAPE overall + per-rooms + per-segment.
|
||
- ``range_coverage`` : sold-total ∈ expected_sold range, overall +
|
||
per-confidence (counts ALL priced deals; a deal with
|
||
no expected_sold range contributes to n but not to
|
||
coverage).
|
||
- ``calibration`` : per-confidence n / coverage% / MAPE%.
|
||
- ``sharpness`` : median relative range width (high-low)/point.
|
||
|
||
Pure: no DB. Safe on an empty list (every block renders with None/0).
|
||
"""
|
||
# Confidence bucket order: canonical three, then any extras encountered.
|
||
conf_order = list(CONFIDENCE_BUCKETS)
|
||
for p in predictions:
|
||
b = _bucketize_confidence(p.confidence)
|
||
if b not in conf_order:
|
||
conf_order.append(b)
|
||
|
||
es_rows: list[tuple[float, float, int]] = [] # (expected_sold_ppm2, sold_ppm2, rooms)
|
||
cov_rows: list[tuple[float, float, float]] = [] # (sold_total, range_low, range_high)
|
||
cov_by_conf: dict[str, list[tuple[float, float, float]]] = {b: [] for b in conf_order}
|
||
sharp_rows: list[tuple[float, float, float]] = [] # (point, range_low, range_high)
|
||
calib_rows: list[tuple[str, float | None, bool | None]] = []
|
||
|
||
for p in predictions:
|
||
signed: float | None = None
|
||
if p.expected_sold_ppm2 is not None and p.sold_ppm2 > 0:
|
||
signed = 100.0 * (p.expected_sold_ppm2 - p.sold_ppm2) / p.sold_ppm2
|
||
es_rows.append((p.expected_sold_ppm2, p.sold_ppm2, p.rooms))
|
||
|
||
covered: bool | None = None
|
||
if p.range_low is not None and p.range_high is not None:
|
||
covered = p.range_low <= p.sold_total <= p.range_high
|
||
cov_rows.append((p.sold_total, p.range_low, p.range_high))
|
||
cov_by_conf[_bucketize_confidence(p.confidence)].append(
|
||
(p.sold_total, p.range_low, p.range_high)
|
||
)
|
||
if p.expected_sold_price is not None:
|
||
sharp_rows.append((p.expected_sold_price, p.range_low, p.range_high))
|
||
|
||
calib_rows.append((p.confidence, signed, covered))
|
||
|
||
return {
|
||
"expected_sold": _expected_sold_metrics(
|
||
es_rows,
|
||
n_no_prediction=n_no_prediction,
|
||
per_rooms_no_prediction=per_rooms_no_prediction,
|
||
),
|
||
"range_coverage": {
|
||
"overall": _range_coverage(cov_rows),
|
||
"per_confidence": {b: _range_coverage(cov_by_conf[b]) for b in conf_order},
|
||
},
|
||
"calibration": _calibration_metrics(calib_rows, bucket_order=conf_order),
|
||
"sharpness": _sharpness(sharp_rows),
|
||
"confidence_order": conf_order,
|
||
}
|
||
|
||
|
||
def _render_table(metrics: dict[str, Any], headline: dict[str, Any]) -> str:
|
||
"""Render the aggregated metrics as a plain-text stdout report."""
|
||
lines: list[str] = []
|
||
lines.append("=" * 78)
|
||
lines.append("BACKTEST: estimator asking-median vs rosreestr ДКП sold prices")
|
||
lines.append("=" * 78)
|
||
|
||
# Headline city-wide spread (asking median vs deal median, ppm²).
|
||
dm = headline.get("deal_median_ppm2")
|
||
am = headline.get("ask_median_ppm2")
|
||
spread = headline.get("spread_pct")
|
||
lines.append("")
|
||
lines.append("CITY-WIDE HEADLINE (sample medians, ₽/m²):")
|
||
lines.append(f" deal_median_ppm2 (SOLD): {_fmt_ppm2(dm)}")
|
||
lines.append(f" ask_median_ppm2 (ASKING): {_fmt_ppm2(am)}")
|
||
lines.append(f" spread (ask vs deal): {_fmt_pct(spread)}")
|
||
|
||
# ASKING block — the estimator's raw asking-median prediction vs SOLD.
|
||
lines.append("")
|
||
lines.append("PER-DEAL ERROR (signed = 100*(pred-sold)/sold; +ve = over-predict):")
|
||
lines.extend(_render_metrics_block("[ASKING] estimator asking-median (uncorrected)", metrics))
|
||
|
||
# CORRECTED block — asking-median × per-rooms asking→sold ratio (#648 S1).
|
||
corrected = metrics.get("corrected")
|
||
ratios_meta = metrics.get("ratios_meta") or {}
|
||
if corrected is not None:
|
||
lines.append("")
|
||
lines.extend(
|
||
_render_metrics_block(
|
||
"[CORRECTED] asking-median × per-rooms asking→sold ratio",
|
||
corrected,
|
||
)
|
||
)
|
||
lines.append("")
|
||
lines.extend(_render_ratios(metrics.get("ratios") or {}, ratios_meta))
|
||
lines.append("")
|
||
if ratios_meta.get("holdout_split"):
|
||
lines.append(
|
||
"Ratios fit on EVEN-id deals, evaluated on ODD-id deals "
|
||
"(--holdout-split) → this is an OUT-OF-SAMPLE corrected number."
|
||
)
|
||
else:
|
||
lines.append(
|
||
"!! IN-SAMPLE: ratios were derived AND evaluated on the same deals, so the"
|
||
)
|
||
lines.append(
|
||
" corrected bias is near-zero BY CONSTRUCTION. This proves the MECHANISM"
|
||
)
|
||
lines.append(
|
||
" (a per-rooms ratio removes the systematic asking→sold gap), NOT out-of-"
|
||
)
|
||
lines.append(
|
||
" sample accuracy. Re-run with --holdout-split for an honest number; the"
|
||
)
|
||
lines.append(" real A/B (Stage 2) fits the ratio on a separate window.")
|
||
|
||
lines.append("")
|
||
lines.append("Caveats: CURRENT listings vs PAST deals (not point-in-time);")
|
||
lines.append("measures asking-median+IQR core only; ДКП = registered price.")
|
||
lines.append("=" * 78)
|
||
return "\n".join(lines)
|
||
|
||
|
||
def _render_metrics_block(
|
||
title: str, metrics: dict[str, Any], *, no_col: str = "no_analog"
|
||
) -> list[str]:
|
||
"""Render one OVERALL + per-rooms metrics table (shared by all blocks).
|
||
|
||
``no_col`` labels the skipped-deal column header — "no_analog" for the
|
||
asking-core block, "no_pred" for the full-spine expected_sold block (where it
|
||
counts deals the spine could not price).
|
||
"""
|
||
header = (
|
||
f" {'bucket':<8} {'n':>5} {no_col:>10} {'bias%':>8} {'MAPE%':>8} {'p25%':>8} {'p75%':>8}"
|
||
)
|
||
out: list[str] = [title, header, " " + "-" * (len(header) - 2)]
|
||
out.append(_fmt_row("OVERALL", metrics["overall"]))
|
||
for bucket in ROOM_BUCKETS:
|
||
row = metrics["per_rooms"][bucket]
|
||
out.append(_fmt_row(row["label"], row))
|
||
return out
|
||
|
||
|
||
def _render_ratios(ratios: dict[int, float], meta: dict[str, Any]) -> list[str]:
|
||
"""Render the derived per-rooms asking→sold ratios + fallback flags."""
|
||
out: list[str] = ["DERIVED per-rooms asking→sold ratios (sold_med / ask_med):"]
|
||
fallback = set(meta.get("fallback_buckets") or [])
|
||
bucket_n = meta.get("bucket_n") or {}
|
||
gr = meta.get("global_ratio")
|
||
min_bucket = meta.get("min_bucket", MIN_BUCKET)
|
||
if not ratios:
|
||
out.append(" (none — empty / no-prediction sample)")
|
||
for bucket in ROOM_BUCKETS:
|
||
if bucket not in ratios:
|
||
continue
|
||
flag = f" [global fallback, n<{min_bucket}]" if bucket in fallback else ""
|
||
out.append(
|
||
f" {_rooms_label(bucket):<8} "
|
||
f"ratio={ratios[bucket]:.4f} n={bucket_n.get(bucket, 0)}{flag}"
|
||
)
|
||
out.append(f" global fallback ratio: {gr if gr is not None else 'n/a'}")
|
||
return out
|
||
|
||
|
||
def _fmt_row(label: str, m: dict[str, Any]) -> str:
|
||
"""Format one metrics row for the table."""
|
||
return (
|
||
f" {label:<8} {m.get('n', 0):>5} {m.get('n_no_analogs', 0):>10} "
|
||
f"{_fmt_pct(m.get('median_bias_pct')):>8} {_fmt_pct(m.get('mape_pct')):>8} "
|
||
f"{_fmt_pct(m.get('p25_pct')):>8} {_fmt_pct(m.get('p75_pct')):>8}"
|
||
)
|
||
|
||
|
||
def _fmt_pct(v: float | None) -> str:
|
||
return " n/a" if v is None else f"{v:+.1f}"
|
||
|
||
|
||
def _fmt_ppm2(v: float | None) -> str:
|
||
return "n/a" if v is None else f"{round(v):,}".replace(",", " ")
|
||
|
||
|
||
def _fmt_cov(v: float | None) -> str:
|
||
"""Coverage percent — non-signed (0..100), n/a for None."""
|
||
return "n/a" if v is None else f"{v:.1f}"
|
||
|
||
|
||
def _fmt_ratio(v: float | None) -> str:
|
||
"""Relative-width / ratio formatter (n/a for None)."""
|
||
return "n/a" if v is None else f"{v:.3f}"
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Full-spine renderer (#1966)
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
def _render_full_table(metrics: dict[str, Any]) -> str:
|
||
"""Render the full-spine metrics dict (#1966) as a plain-text stdout report."""
|
||
headline = metrics.get("headline") or {}
|
||
lines: list[str] = []
|
||
lines.append("=" * 78)
|
||
lines.append("BACKTEST [full spine]: _price_from_inputs expected_sold vs ДКП sold")
|
||
lines.append("=" * 78)
|
||
|
||
dm = headline.get("deal_median_ppm2")
|
||
am = headline.get("ask_median_ppm2")
|
||
spread = headline.get("spread_pct")
|
||
lines.append("")
|
||
lines.append("CITY-WIDE HEADLINE (sample medians, ₽/m²):")
|
||
lines.append(f" deal_median_ppm2 (SOLD): {_fmt_ppm2(dm)}")
|
||
lines.append(f" ask_median_ppm2 (ASKING head): {_fmt_ppm2(am)}")
|
||
lines.append(f" spread (asking vs deal): {_fmt_pct(spread)}")
|
||
|
||
es = metrics["expected_sold"]
|
||
lines.append("")
|
||
lines.append("EXPECTED_SOLD ERROR (signed = 100*(expected_sold-sold)/sold; +ve = over):")
|
||
lines.extend(_render_metrics_block("[EXPECTED_SOLD] per-rooms", es, no_col="no_pred"))
|
||
lines.append("")
|
||
lines.extend(_render_segment_block(es["per_segment"]))
|
||
|
||
conf_order = metrics.get("confidence_order") or list(CONFIDENCE_BUCKETS)
|
||
lines.append("")
|
||
lines.extend(_render_coverage_block(metrics["range_coverage"], conf_order))
|
||
lines.append("")
|
||
lines.extend(_render_calibration_block(metrics["calibration"], conf_order))
|
||
|
||
sh = metrics["sharpness"]
|
||
lines.append("")
|
||
lines.append(
|
||
f"SHARPNESS: median relative range width (high-low)/point = "
|
||
f"{_fmt_ratio(sh.get('median_rel_width'))} (n={sh.get('n', 0)})"
|
||
)
|
||
|
||
lines.append("")
|
||
lines.append("Caveats: CURRENT listings vs PAST deals (time-mismatched) →")
|
||
lines.append("regression baseline, not absolute truth; network valuation")
|
||
lines.append("layers (Avito-IMV on-demand / Yandex / Cian) excluded.")
|
||
lines.append("=" * 78)
|
||
return "\n".join(lines)
|
||
|
||
|
||
def _render_segment_block(per_segment: dict[str, Any]) -> list[str]:
|
||
"""Render the per-price-segment expected_sold error table."""
|
||
header = f" {'segment':<10} {'n':>5} {'bias%':>8} {'MAPE%':>8} {'p25%':>8} {'p75%':>8}"
|
||
out: list[str] = [
|
||
"[EXPECTED_SOLD] per price-segment (band by SOLD ₽/m²):",
|
||
header,
|
||
" " + "-" * (len(header) - 2),
|
||
]
|
||
for label, _ in PRICE_SEGMENTS_PPM2:
|
||
m = per_segment[label]
|
||
out.append(
|
||
f" {label:<10} {m.get('n', 0):>5} "
|
||
f"{_fmt_pct(m.get('median_bias_pct')):>8} {_fmt_pct(m.get('mape_pct')):>8} "
|
||
f"{_fmt_pct(m.get('p25_pct')):>8} {_fmt_pct(m.get('p75_pct')):>8}"
|
||
)
|
||
return out
|
||
|
||
|
||
def _render_coverage_block(range_coverage: dict[str, Any], conf_order: list[str]) -> list[str]:
|
||
"""Render range-coverage: overall + per-confidence (sold_total ∈ range)."""
|
||
ov = range_coverage["overall"]
|
||
out: list[str] = ["RANGE COVERAGE (sold total within predicted expected_sold range):"]
|
||
out.append(
|
||
f" OVERALL n={ov['n']:>5} covered={ov['n_covered']:>5} "
|
||
f"coverage={_fmt_cov(ov['coverage_pct'])}%"
|
||
)
|
||
per_conf = range_coverage.get("per_confidence") or {}
|
||
for b in conf_order:
|
||
c = per_conf.get(b, {"n": 0, "n_covered": 0, "coverage_pct": None})
|
||
out.append(
|
||
f" {b:<10} n={c['n']:>5} covered={c['n_covered']:>5} "
|
||
f"coverage={_fmt_cov(c['coverage_pct'])}%"
|
||
)
|
||
return out
|
||
|
||
|
||
def _render_calibration_block(calibration: dict[str, Any], conf_order: list[str]) -> list[str]:
|
||
"""Render confidence-calibration: per bucket n / coverage% / MAPE%."""
|
||
header = f" {'conf':<8} {'n':>5} {'coverage%':>10} {'MAPE%':>8}"
|
||
out: list[str] = [
|
||
"CONFIDENCE CALIBRATION (high should be tighter & more accurate):",
|
||
header,
|
||
" " + "-" * (len(header) - 2),
|
||
]
|
||
for b in conf_order:
|
||
c = calibration.get(b, {"n": 0, "coverage_pct": None, "mape_pct": None})
|
||
out.append(
|
||
f" {b:<8} {c.get('n', 0):>5} "
|
||
f"{_fmt_cov(c.get('coverage_pct')):>10} {_fmt_pct(c.get('mape_pct')):>8}"
|
||
)
|
||
return out
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# DB layer — READ-ONLY SELECTs only.
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
# ДКП deal sample. lon/lat extracted via ST_X/ST_Y so the per-deal listings
|
||
# query can rebuild the point without re-reading geom. Parameterized;
|
||
# psycopg3 CAST(:x AS type), never :x::type.
|
||
_SAMPLE_SQL = text(
|
||
"""
|
||
SELECT
|
||
id,
|
||
ST_X(geom::geometry) AS lon,
|
||
ST_Y(geom::geometry) AS lat,
|
||
rooms,
|
||
price_per_m2 AS sold_ppm2,
|
||
deal_date,
|
||
area_m2,
|
||
address,
|
||
floor,
|
||
total_floors,
|
||
year_built,
|
||
house_type
|
||
FROM deals
|
||
WHERE source = 'rosreestr'
|
||
AND geom IS NOT NULL
|
||
AND price_per_m2 BETWEEN CAST(:ppm2_min AS numeric) AND CAST(:ppm2_max AS numeric)
|
||
AND rooms IS NOT NULL
|
||
AND area_m2 IS NOT NULL
|
||
AND area_m2 > 0
|
||
AND deal_date >= CAST(:since AS date)
|
||
ORDER BY id DESC
|
||
LIMIT CAST(:sample AS integer)
|
||
"""
|
||
)
|
||
|
||
# Per-deal candidate active listings. rooms matched within :rooms_lo..:rooms_hi
|
||
# (exact when tolerance=0). Returns raw price_per_m2 values — _filter_outliers
|
||
# is applied in Python for byte-for-byte fidelity with the estimator.
|
||
_CANDIDATES_SQL = text(
|
||
"""
|
||
SELECT price_per_m2
|
||
FROM listings
|
||
WHERE is_active
|
||
AND rooms BETWEEN CAST(:rooms_lo AS integer) AND CAST(:rooms_hi AS integer)
|
||
AND price_per_m2 BETWEEN CAST(:ppm2_min AS numeric) AND CAST(:ppm2_max AS numeric)
|
||
AND ST_DWithin(
|
||
geom::geography,
|
||
ST_SetSRID(ST_MakePoint(
|
||
CAST(:lon AS double precision),
|
||
CAST(:lat AS double precision)
|
||
), 4326)::geography,
|
||
CAST(:radius AS double precision)
|
||
)
|
||
"""
|
||
)
|
||
|
||
|
||
def _load_sample(db: Session, *, sample: int, since: str) -> list[DealSample]:
|
||
"""Run the held-out ДКП deal sampling SELECT → list[DealSample]."""
|
||
rows = (
|
||
db.execute(
|
||
_SAMPLE_SQL,
|
||
{
|
||
"ppm2_min": PPM2_MIN,
|
||
"ppm2_max": PPM2_MAX,
|
||
"since": since,
|
||
"sample": sample,
|
||
},
|
||
)
|
||
.mappings()
|
||
.all()
|
||
)
|
||
out: list[DealSample] = []
|
||
for r in rows:
|
||
if r["lon"] is None or r["lat"] is None or r["sold_ppm2"] is None:
|
||
continue
|
||
if r["area_m2"] is None or float(r["area_m2"]) <= 0:
|
||
continue
|
||
out.append(
|
||
DealSample(
|
||
id=r["id"],
|
||
lon=float(r["lon"]),
|
||
lat=float(r["lat"]),
|
||
rooms=int(r["rooms"]),
|
||
sold_ppm2=float(r["sold_ppm2"]),
|
||
deal_date=r["deal_date"],
|
||
area_m2=float(r["area_m2"]),
|
||
address=r["address"],
|
||
floor=(int(r["floor"]) if r["floor"] is not None else None),
|
||
total_floors=(int(r["total_floors"]) if r["total_floors"] is not None else None),
|
||
year_built=(int(r["year_built"]) if r["year_built"] is not None else None),
|
||
house_type=r["house_type"],
|
||
)
|
||
)
|
||
return out
|
||
|
||
|
||
def _predict_for_deal(
|
||
db: Session,
|
||
deal: DealSample,
|
||
*,
|
||
radius: int,
|
||
rooms_tolerance: int,
|
||
) -> float | None:
|
||
"""Predict asking ppm² for one deal by reusing the estimator's pure funcs.
|
||
|
||
Fetches candidate active-listing ppm² values, wraps them as
|
||
``{"price_per_m2": p}`` dicts, applies the estimator's `_filter_outliers`
|
||
(Tukey IQR), then `_percentile(sorted, 0.5)`. Returns None when fewer than
|
||
MIN_CANDIDATES survive (caller counts it as a no-analog miss).
|
||
"""
|
||
rows = db.execute(
|
||
_CANDIDATES_SQL,
|
||
{
|
||
"rooms_lo": deal.rooms - rooms_tolerance,
|
||
"rooms_hi": deal.rooms + rooms_tolerance,
|
||
"ppm2_min": PPM2_MIN,
|
||
"ppm2_max": PPM2_MAX,
|
||
"lon": deal.lon,
|
||
"lat": deal.lat,
|
||
"radius": radius,
|
||
},
|
||
).all()
|
||
|
||
# Build the same dict shape the estimator feeds _filter_outliers.
|
||
lots = [{"price_per_m2": float(r[0])} for r in rows if r[0] is not None]
|
||
if len(lots) < MIN_CANDIDATES:
|
||
return None
|
||
|
||
_filter_outliers, _percentile = _import_estimator()
|
||
clean = _filter_outliers(lots)
|
||
prices = sorted(lot["price_per_m2"] for lot in clean if lot["price_per_m2"])
|
||
if len(prices) < MIN_CANDIDATES:
|
||
return None
|
||
return _percentile(prices, 0.5)
|
||
|
||
|
||
def _select_analogs_full(
|
||
db: Session, deal: DealSample, est: SimpleNamespace
|
||
) -> tuple[list[dict[str, Any]], str, bool, bool]:
|
||
"""Replicate estimate_quality's analog tier ladder for one deal (#1966).
|
||
|
||
Mirrors the Tier0(cohort)→A(1 km)→wide(2 km)→widearea(±25 %) sequence
|
||
EXACTLY, including the MIN_ANALOGS_TIER_0 gate and the <5 / <3 widen
|
||
thresholds and the "only adopt the wider tier when it returns MORE lots"
|
||
guards. target_house_id is always None here (the harness does not resolve
|
||
canonical houses), so Tier S(canonical) never fires — same as a fresh
|
||
estimate without a house match.
|
||
|
||
Returns ``(listings, analog_tier, fallback_used, area_widened)``.
|
||
"""
|
||
m = est.m
|
||
cohort_range = m._target_cohort_range(deal.year_built)
|
||
if cohort_range is not None:
|
||
cy_min, cy_max = cohort_range
|
||
listings_tier0, _, analog_tier = m._fetch_analogs(
|
||
db,
|
||
lat=deal.lat,
|
||
lon=deal.lon,
|
||
rooms=deal.rooms,
|
||
area=deal.area_m2,
|
||
radius_m=m.DEFAULT_RADIUS_M,
|
||
full_address=deal.address,
|
||
target_house_id=None,
|
||
year_built=deal.year_built,
|
||
house_type=deal.house_type,
|
||
total_floors=deal.total_floors,
|
||
cohort_year_min=cy_min,
|
||
cohort_year_max=cy_max,
|
||
)
|
||
else:
|
||
listings_tier0 = []
|
||
analog_tier = "W"
|
||
|
||
if len(listings_tier0) >= m.MIN_ANALOGS_TIER_0:
|
||
listings = listings_tier0
|
||
fallback_used = False
|
||
else:
|
||
listings, fallback_used, analog_tier = m._fetch_analogs(
|
||
db,
|
||
lat=deal.lat,
|
||
lon=deal.lon,
|
||
rooms=deal.rooms,
|
||
area=deal.area_m2,
|
||
radius_m=m.DEFAULT_RADIUS_M,
|
||
full_address=deal.address,
|
||
target_house_id=None,
|
||
year_built=deal.year_built,
|
||
house_type=deal.house_type,
|
||
total_floors=deal.total_floors,
|
||
)
|
||
area_widened = False
|
||
|
||
if len(listings) < 5:
|
||
listings_wide, _, analog_tier_wide = m._fetch_analogs(
|
||
db,
|
||
lat=deal.lat,
|
||
lon=deal.lon,
|
||
rooms=deal.rooms,
|
||
area=deal.area_m2,
|
||
radius_m=m.FALLBACK_RADIUS_M,
|
||
full_address=deal.address,
|
||
target_house_id=None,
|
||
year_built=deal.year_built,
|
||
house_type=deal.house_type,
|
||
total_floors=deal.total_floors,
|
||
)
|
||
if len(listings_wide) > len(listings):
|
||
listings = listings_wide
|
||
fallback_used = True
|
||
analog_tier = analog_tier_wide
|
||
|
||
if len(listings) < 3:
|
||
listings_widearea, _, analog_tier_wa = m._fetch_analogs(
|
||
db,
|
||
lat=deal.lat,
|
||
lon=deal.lon,
|
||
rooms=deal.rooms,
|
||
area=deal.area_m2,
|
||
radius_m=m.FALLBACK_RADIUS_M,
|
||
area_tolerance=0.25,
|
||
full_address=deal.address,
|
||
target_house_id=None,
|
||
year_built=deal.year_built,
|
||
house_type=deal.house_type,
|
||
total_floors=deal.total_floors,
|
||
)
|
||
if len(listings_widearea) > len(listings):
|
||
listings = listings_widearea
|
||
fallback_used = True
|
||
area_widened = True
|
||
analog_tier = analog_tier_wa
|
||
|
||
return listings, analog_tier, fallback_used, area_widened
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# 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 _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 load_fixture(path: str) -> dict[str, Any]:
|
||
"""Load a backtest fixture JSON — transparently handles gzip (``.gz``) files.
|
||
|
||
The committed fixture is gzipped (~3 MB raw); a plain ``.json`` path is also
|
||
accepted for ad-hoc runs. Exported for the CI regression-gate test.
|
||
"""
|
||
if str(path).endswith(".gz"):
|
||
with gzip.open(path, "rt", encoding="utf-8") as fh:
|
||
return json.loads(fh.read())
|
||
return json.loads(Path(path).read_text(encoding="utf-8"))
|
||
|
||
|
||
def _make_call_stub(
|
||
calls: list[Any], *, label: str, coerce: Callable[[Any], Any]
|
||
) -> Callable[[Any], Any]:
|
||
"""Build an ORDER-based (FIFO) replay stub from recorded ``[arg, return]`` pairs.
|
||
|
||
The stub returns the recorded RETURNS in invocation order and IGNORES the arg
|
||
value. Rationale: the DB-derived ratio / quarter-index is a FROZEN input — when
|
||
estimator logic later shifts e.g. ``median_ppm2``, the regression gate must
|
||
surface that as a clean metric diff vs the baseline, NOT mask it as a lookup
|
||
miss. Matching on the exact computed float arg is also not bit-stable across
|
||
platforms (libm last-ulp jitter), so a Linux-captured fixture would crash when
|
||
replayed off-Linux. The recorded arg is kept (for debuggability) but never
|
||
matched. Each callable fires ≤1×/deal today (so index 0 in practice); the FIFO
|
||
stays correct if a call site ever loops. Calling the stub MORE times than
|
||
recorded raises RuntimeError — control flow diverged from capture.
|
||
|
||
``coerce`` maps each JSON-plain recorded return back to the live callable's
|
||
return type (tuple / dict) so unpacking at the call site behaves identically.
|
||
"""
|
||
returns = [coerce(ret) for _arg, ret in calls]
|
||
idx = 0
|
||
|
||
def _stub(_arg: Any) -> Any:
|
||
nonlocal idx
|
||
if idx >= len(returns):
|
||
raise RuntimeError(
|
||
f"{label}: replay made call #{idx + 1} but fixture recorded only "
|
||
f"{len(returns)} — control flow diverged from capture"
|
||
)
|
||
ret = returns[idx]
|
||
idx += 1
|
||
return ret
|
||
|
||
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 order-based
|
||
(FIFO) 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
|
||
(anchor comps, ДКП corridor, house IMV anchor) and injects the DB callables,
|
||
then calls ``estimator._price_from_inputs`` with parameters that mirror
|
||
``estimate_quality`` for a sold unit whose repair state is unknown and with
|
||
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). 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
|
||
|
||
listings, analog_tier, fallback_used, area_widened = _select_analogs_full(db, deal, est)
|
||
|
||
# ── 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)
|
||
# #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
|
||
)
|
||
|
||
# Synthetic geo = the deal's own coords + address, confidence "exact" — the
|
||
# faithful analog of estimate_quality's client-coords fast path. Real deal
|
||
# addresses carry a house number, so _geocode_is_coarse → False (no coarse
|
||
# downgrade); if an address lacks a number that deal degrades to "low",
|
||
# exactly as the production gate would.
|
||
geo = est.GeocodeResult(
|
||
lat=deal.lat,
|
||
lon=deal.lon,
|
||
full_address=deal.address or "",
|
||
provider="cache",
|
||
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]:
|
||
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:
|
||
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]:
|
||
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,
|
||
area_m2=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=anchor_comps,
|
||
anchor_tier_fetched=anchor_tier,
|
||
dkp_raw=dkp_raw,
|
||
imv_anchor=imv_anchor,
|
||
imv_eval=None,
|
||
yandex_val_present=False,
|
||
cian_val_present=False,
|
||
ratio_resolver=_ratio_resolver,
|
||
quarter_index_lookup=_qi_lookup,
|
||
quarter_indexes_lookup=_qis_lookup,
|
||
target_house_cadnum=None,
|
||
dadata_coarse=False,
|
||
geo=geo,
|
||
dadata_qc_geo=None,
|
||
)
|
||
|
||
# #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,
|
||
area_m2=deal.area_m2,
|
||
sold_ppm2=deal.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,
|
||
)
|
||
|
||
|
||
def _attach_correction(
|
||
metrics: dict[str, Any],
|
||
matched_rows: list[tuple[float, float, int]],
|
||
matched_ids: list[int],
|
||
*,
|
||
holdout_split: bool,
|
||
) -> None:
|
||
"""Derive the per-rooms ratio, apply it, and attach the CORRECTED block.
|
||
|
||
Mutates ``metrics`` in place, adding ``ratios`` (bucket→float),
|
||
``ratios_meta`` (diagnostics), and ``corrected`` (a full _compute_metrics
|
||
dict for ``pred_sold = pred_ask * ratio``).
|
||
|
||
Two modes:
|
||
- default (in-sample): ratios derived on ALL matched rows, evaluated on
|
||
ALL matched rows → near-zero bias by construction (mechanism proof).
|
||
- ``holdout_split``: ratios derived on EVEN-id deals, evaluated on the
|
||
ODD-id half → an honest out-of-sample number. Split is by deal-id
|
||
parity (deterministic, reproducible — no RNG).
|
||
|
||
Pure aside from reusing the no-DB helpers; safe on an empty sample.
|
||
"""
|
||
if holdout_split:
|
||
paired = list(zip(matched_rows, matched_ids, strict=True))
|
||
fit_rows = [r for r, did in paired if did % 2 == 0]
|
||
eval_rows = [r for r, did in paired if did % 2 == 1]
|
||
else:
|
||
fit_rows = matched_rows
|
||
eval_rows = matched_rows
|
||
|
||
ratios, ratios_meta = _derive_room_ratios(fit_rows)
|
||
ratios_meta["holdout_split"] = holdout_split
|
||
ratios_meta["n_fit"] = len(fit_rows)
|
||
ratios_meta["n_eval"] = len(eval_rows)
|
||
|
||
corrected_rows = _apply_ratios(eval_rows, ratios)
|
||
corrected = _compute_metrics(corrected_rows)
|
||
|
||
# JSON-friendly: stringify int bucket keys, round ratios for readability.
|
||
metrics["ratios"] = ratios
|
||
metrics["ratios_meta"] = ratios_meta
|
||
metrics["corrected"] = corrected
|
||
|
||
|
||
def run_backtest(
|
||
db: Session,
|
||
*,
|
||
sample: int,
|
||
since: str,
|
||
radius: int,
|
||
rooms_tolerance: int,
|
||
holdout_split: bool = False,
|
||
) -> dict[str, Any]:
|
||
"""Drive the full read-only backtest and return a metrics dict.
|
||
|
||
Steps: load sample → predict per deal (reusing estimator funcs) → collect
|
||
(pred, sold, rooms) rows + no-analog counts → `_compute_metrics` (ASKING
|
||
block) → `_derive_room_ratios` + `_apply_ratios` → `_compute_metrics`
|
||
again (CORRECTED block) + a city-wide headline spread. No writes.
|
||
|
||
The CORRECTED block multiplies each asking-median prediction by a per-rooms
|
||
asking→sold ratio derived from the SAME sample (issue #648 Stage 1). With
|
||
``holdout_split=False`` (default) that ratio is fit and evaluated in-sample,
|
||
so its bias is near-zero by construction — it proves the MECHANISM, not
|
||
out-of-sample accuracy (see _derive_room_ratios). Pass ``holdout_split=True``
|
||
to fit on even-id deals and evaluate on the odd-id half for an honest number.
|
||
"""
|
||
deals = _load_sample(db, sample=sample, since=since)
|
||
logger.info("loaded sample: %d ДКП deals (since=%s)", len(deals), since)
|
||
|
||
matched_rows: list[tuple[float, float, int]] = []
|
||
matched_ids: list[int] = []
|
||
n_no_analogs = 0
|
||
per_rooms_no_analogs: dict[int, int] = {b: 0 for b in ROOM_BUCKETS}
|
||
|
||
# For the city-wide headline: median of all sampled SOLD ppm², and median
|
||
# of all per-deal predicted ASKING ppm² (matched deals only).
|
||
sold_ppm2_all: list[float] = [d.sold_ppm2 for d in deals]
|
||
pred_ppm2_all: list[float] = []
|
||
|
||
for i, deal in enumerate(deals, start=1):
|
||
pred = _predict_for_deal(db, deal, radius=radius, rooms_tolerance=rooms_tolerance)
|
||
if pred is None:
|
||
n_no_analogs += 1
|
||
per_rooms_no_analogs[_bucketize_rooms(deal.rooms)] += 1
|
||
else:
|
||
matched_rows.append((pred, deal.sold_ppm2, deal.rooms))
|
||
matched_ids.append(deal.id)
|
||
pred_ppm2_all.append(pred)
|
||
|
||
if i % 50 == 0:
|
||
logger.info(
|
||
"progress %d/%d (matched=%d, no_analogs=%d)",
|
||
i,
|
||
len(deals),
|
||
len(matched_rows),
|
||
n_no_analogs,
|
||
)
|
||
|
||
metrics = _compute_metrics(
|
||
matched_rows,
|
||
n_no_analogs=n_no_analogs,
|
||
per_rooms_no_analogs=per_rooms_no_analogs,
|
||
)
|
||
|
||
_attach_correction(metrics, matched_rows, matched_ids, holdout_split=holdout_split)
|
||
|
||
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,
|
||
}
|
||
metrics["params"] = {
|
||
"engine": "asking-core",
|
||
"sample_requested": sample,
|
||
"sample_loaded": len(deals),
|
||
"since": since,
|
||
"radius_m": radius,
|
||
"rooms_tolerance": rooms_tolerance,
|
||
"n_matched": len(matched_rows),
|
||
"n_no_analogs": n_no_analogs,
|
||
"holdout_split": holdout_split,
|
||
}
|
||
return metrics
|
||
|
||
|
||
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 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)
|
||
logger.info("loaded sample: %d ДКП deals (since=%s) [full spine]", len(deals), since)
|
||
|
||
predictions: list[Prediction] = []
|
||
n_no_prediction = 0
|
||
per_rooms_no_prediction: dict[int, int] = {b: 0 for b in ROOM_BUCKETS}
|
||
|
||
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, 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).
|
||
logger.warning("deal %s spine failed (graceful, skipped): %s", deal.id, exc)
|
||
db.rollback()
|
||
pr = None
|
||
|
||
if pr is None:
|
||
n_no_prediction += 1
|
||
per_rooms_no_prediction[_bucketize_rooms(deal.rooms)] += 1
|
||
else:
|
||
predictions.append(pr)
|
||
pred_ppm2_all.append(pr.median_ppm2)
|
||
|
||
if i % 50 == 0:
|
||
logger.info(
|
||
"progress %d/%d (priced=%d, no_pred=%d)",
|
||
i,
|
||
len(deals),
|
||
len(predictions),
|
||
n_no_prediction,
|
||
)
|
||
|
||
metrics = _compute_full_metrics(
|
||
predictions,
|
||
n_no_prediction=n_no_prediction,
|
||
per_rooms_no_prediction=per_rooms_no_prediction,
|
||
)
|
||
|
||
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,
|
||
}
|
||
metrics["params"] = {
|
||
"engine": "full",
|
||
"sample_requested": sample,
|
||
"sample_loaded": len(deals),
|
||
"since": since,
|
||
"n_matched": len(predictions),
|
||
"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
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
||
"""argparse setup, factored out for testability."""
|
||
p = argparse.ArgumentParser(
|
||
description=(
|
||
"READ-ONLY backtest of the estimator against rosreestr ДКП sold "
|
||
"prices. Default engine 'full' runs the deterministic pricing spine "
|
||
"(_price_from_inputs); 'asking-core' is the legacy median+IQR core "
|
||
"(issues #648 / #1966)."
|
||
),
|
||
)
|
||
p.add_argument(
|
||
"--engine",
|
||
choices=("full", "asking-core"),
|
||
default="full",
|
||
help="Prediction engine: 'full' (default) = the deterministic pricing "
|
||
"spine via _price_from_inputs (expected_sold + coverage/calibration/"
|
||
"segment metrics); 'asking-core' = legacy asking-median+IQR core with "
|
||
"the per-rooms asking→sold correction block.",
|
||
)
|
||
p.add_argument(
|
||
"--sample",
|
||
type=int,
|
||
default=300,
|
||
help="Held-out ДКП deals to test (default 300). Large samples are slow "
|
||
"— one PostGIS subquery runs per deal.",
|
||
)
|
||
p.add_argument(
|
||
"--since",
|
||
default="2025-06-01",
|
||
help="Only deals with deal_date >= this ISO date (default 2025-06-01).",
|
||
)
|
||
p.add_argument(
|
||
"--radius",
|
||
type=int,
|
||
default=1000,
|
||
help="Analog search radius in metres (default 1000).",
|
||
)
|
||
p.add_argument(
|
||
"--rooms-tolerance",
|
||
type=int,
|
||
default=0,
|
||
help="± room count tolerance for analogs (default 0 = exact match).",
|
||
)
|
||
p.add_argument(
|
||
"--holdout-split",
|
||
action="store_true",
|
||
help="Fit the per-rooms asking→sold ratio on EVEN-id deals and evaluate "
|
||
"the CORRECTED block on the ODD-id half (deterministic out-of-sample "
|
||
"split, no RNG). Default off → ratio is fit AND evaluated in-sample, so "
|
||
"the corrected bias is near-zero by construction (mechanism proof only).",
|
||
)
|
||
p.add_argument(
|
||
"--json",
|
||
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 (gzip-transparent — "
|
||
"a .gz path is decompressed on the fly), 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) / replayed deals."""
|
||
args = _parse_args(argv)
|
||
|
||
# ── Hermetic replay path (#1966 PR 3/3) — ZERO DB, no SessionLocal opened. ──
|
||
if args.from_fixture:
|
||
fixture = load_fixture(args.from_fixture)
|
||
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 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, dump_fixture=args.dump_fixture
|
||
)
|
||
else:
|
||
metrics = run_backtest(
|
||
db,
|
||
sample=args.sample,
|
||
since=args.since,
|
||
radius=args.radius,
|
||
rooms_tolerance=args.rooms_tolerance,
|
||
holdout_split=args.holdout_split,
|
||
)
|
||
finally:
|
||
db.close()
|
||
|
||
if args.json:
|
||
print(json.dumps(metrics, ensure_ascii=False, indent=2, default=str))
|
||
elif args.engine == "full":
|
||
print(_render_full_table(metrics))
|
||
else:
|
||
print(_render_table(metrics, metrics["headline"]))
|
||
|
||
return int(metrics["params"]["n_matched"])
|
||
|
||
|
||
if __name__ == "__main__": # pragma: no cover
|
||
raise SystemExit(0 if main() >= 0 else 1)
|