diff --git a/tradein-mvp/backend/scripts/backtest_estimator.py b/tradein-mvp/backend/scripts/backtest_estimator.py index e9496665..0508c85a 100644 --- a/tradein-mvp/backend/scripts/backtest_estimator.py +++ b/tradein-mvp/backend/scripts/backtest_estimator.py @@ -1,74 +1,85 @@ -"""Backtest harness — measures the estimator's asking-median accuracy vs real ДКП sold prices. +"""Backtest harness — measures the estimator's accuracy vs real ДКП sold prices. -Forgejo issue #648. **STRICTLY READ-ONLY**: this script issues only SELECT -queries against prod. It never INSERTs, UPDATEs, or runs DDL. +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. -WHAT IT MEASURES ----------------- -The estimator predicts a квартира's value from the median of *active asking* -prices of nearby analog listings (Tukey 1.5×IQR filter → median ppm²). Ground -truth is `deals` — registered rosreestr ДКП sales (source='rosreestr') that -carry geom + price_per_m2 + rooms. For a held-out sample of ДКП deals we: +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. - 1. Sample deals (id, lon, lat, rooms, sold_ppm2, deal_date). - 2. For each deal, fetch nearby *active* listings of the same rooms within - `--radius` metres (PostGIS ST_DWithin on geography). - 3. Predict via the estimator's OWN pure functions for fidelity: - `_filter_outliers` (Tukey IQR) → `_percentile(sorted, 0.5)` (median). - This mirrors exactly what `estimate_quality` does to its analog pool. - Deals with <3 surviving candidates are skipped (no prediction). - 4. Per deal: signed_error_pct = 100*(pred - sold)/sold; abs = |signed|. - 5. Aggregate overall + per-rooms: n_matched, n_no_analogs, median_bias_pct - (systematic over/under), MAPE (median |error|), p25/p75 of signed error. - Plus a city-wide deal_median_ppm2 vs ask_median_ppm2 headline spread. +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: -CORRECTION BLOCK (issue #648 Stage 1) -------------------------------------- -On top of the raw ASKING metrics the harness now emits a second CORRECTED -block. From the SAME matched sample it derives a per-rooms asking→sold ratio -``ratio[bucket] = median(sold_ppm2) / median(pred_ask_ppm2)`` (global fallback -for thin buckets), then re-scores ``pred_sold = pred_ask * ratio[bucket]`` -through the same `_compute_metrics`. This DEMONSTRATES that a per-rooms factor -removes the systematic +29.6% asking→sold bias. + 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 / <3 analogs) are skipped. + 3. Score ``expected_sold_per_m2`` vs the realised SOLD ppm². - !! HONESTY — this ratio is IN-SAMPLE by default: it is derived AND evaluated - on the same deals, so the corrected bias lands near zero BY CONSTRUCTION. - That proves the MECHANISM, NOT out-of-sample accuracy. Pass - ``--holdout-split`` to fit on even-id deals and evaluate on the odd-id - half (deterministic, no RNG) for an honest number. The production ratio - (Stage 2) is fit over a SEPARATE window and the real A/B is current-vs- - corrected on held-out data. This script changes NOTHING in prod — it only - proves the correction is worth building. +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 being judged against listings active today. As the market - moves, asking prices drift away from the historical sold price, which - inflates the apparent bias. A faithful point-in-time backtest needs - `listing_source_snapshots` (#570) — once that table accumulates enough - history we can query the asking median *as of each deal's date*. - (b) PARTIAL ESTIMATOR — this exercises only the asking-median + IQR CORE. - It does NOT include the full estimator's tiering (same-house / cohort / - class-coef), DaData enrichment, Avito-IMV, Cian/Yandex valuation, or the - repair-state coefficient. Real estimate accuracy may differ. + 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 ----------- -One spatial listings subquery runs per sampled deal, so runtime scales with -`--sample`. The default (300) finishes quickly; large samples (thousands) are -slow because of the per-deal PostGIS ST_DWithin scan. Bump `--sample` only -when you need tighter per-rooms confidence intervals. +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 """ @@ -81,6 +92,7 @@ import logging import statistics from dataclasses import dataclass from pathlib import Path +from types import SimpleNamespace from typing import Any from sqlalchemy import text @@ -107,6 +119,37 @@ def _import_estimator() -> tuple[Any, Any]: 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: @@ -144,6 +187,26 @@ ROOM_BUCKETS: tuple[int, ...] = (0, 1, 2, 3, 4) # 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 @@ -152,7 +215,12 @@ MIN_BUCKET = 20 @dataclass class DealSample: - """One held-out ДКП deal to backtest against.""" + """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 @@ -160,6 +228,40 @@ class DealSample: 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 # --------------------------------------------------------------------------- # @@ -374,6 +476,206 @@ def _apply_ratios( 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] = [] @@ -393,14 +695,8 @@ def _render_table(metrics: dict[str, Any], headline: dict[str, Any]) -> str: # 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 - ) - ) + 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") @@ -423,24 +719,18 @@ def _render_table(metrics: dict[str, Any], headline: dict[str, Any]) -> str: ) else: lines.append( - "!! IN-SAMPLE: ratios were derived AND evaluated on the same " - "deals, so the" + "!! 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" + " 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-" + " (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." + " 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);") @@ -449,11 +739,17 @@ def _render_table(metrics: dict[str, Any], headline: dict[str, Any]) -> str: return "\n".join(lines) -def _render_metrics_block(title: str, metrics: dict[str, Any]) -> list[str]: - """Render one OVERALL + per-rooms metrics table (shared by both blocks).""" +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_analog':>10} " - f"{'bias%':>8} {'MAPE%':>8} {'p25%':>8} {'p75%':>8}" + 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"])) @@ -501,6 +797,119 @@ 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. # --------------------------------------------------------------------------- # @@ -516,12 +925,20 @@ _SAMPLE_SQL = text( ST_Y(geom::geometry) AS lat, rooms, price_per_m2 AS sold_ppm2, - deal_date + 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) @@ -569,6 +986,8 @@ def _load_sample(db: Session, *, sample: int, since: str) -> 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"], @@ -577,6 +996,12 @@ def _load_sample(db: Session, *, sample: int, since: str) -> list[DealSample]: 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 @@ -622,6 +1047,217 @@ def _predict_for_deal( 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 + + +def _predict_full_spine(db: Session, deal: DealSample, est: SimpleNamespace) -> 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 or tuple[float | None, str | None]: + return m._get_asking_sold_ratio(db, deal.rooms, anchor_ppm2=appm2) + + def _qi_lookup(q: str) -> tuple[float, int] | None: + return m._lookup_quarter_index( + db, + quarter_cad_number=q, + min_n_deals=settings.estimate_quarter_index_min_n_deals, + ) + + def _qis_lookup(qs: list[str]) -> dict[str, float]: + return m._lookup_quarter_indexes( + db, + quarter_cad_numbers=qs, + min_n_deals=settings.estimate_quarter_index_min_n_deals, + ) + + 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, + ) + + # Skip when the spine couldn't price it — mirror the asking-core skip. + if pr.median_price <= 0 or pr.n_analogs < MIN_CANDIDATES: + 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 + 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]], @@ -703,9 +1339,7 @@ def run_backtest( 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 - ) + 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 @@ -729,8 +1363,88 @@ def run_backtest( per_rooms_no_analogs=per_rooms_no_analogs, ) - _attach_correction( - metrics, matched_rows, matched_ids, holdout_split=holdout_split + _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) -> 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. + + ``--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). + """ + 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) + + for i, deal in enumerate(deals, start=1): + try: + pr = _predict_full_spine(db, deal, est) + 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 @@ -745,14 +1459,13 @@ def run_backtest( "spread_pct": spread_pct, } metrics["params"] = { + "engine": "full", "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, + "n_matched": len(predictions), + "n_no_prediction": n_no_prediction, + "price_segments_ppm2": [list(seg) for seg in PRICE_SEGMENTS_PPM2], } return metrics @@ -766,10 +1479,21 @@ 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's asking-median accuracy " - "against rosreestr ДКП sold prices (issue #648)." + "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, @@ -814,8 +1538,9 @@ def main(argv: list[str] | None = None) -> int: """CLI entry point. Returns the count of matched (predicted) deals.""" args = _parse_args(argv) logger.info( - "backtest start: sample=%d since=%s radius=%dm rooms_tolerance=%d " - "holdout_split=%s", + "backtest start: engine=%s sample=%d since=%s radius=%dm " + "rooms_tolerance=%d holdout_split=%s", + args.engine, args.sample, args.since, args.radius, @@ -825,19 +1550,24 @@ def main(argv: list[str] | None = None) -> int: db = _session() try: - metrics = run_backtest( - db, - sample=args.sample, - since=args.since, - radius=args.radius, - rooms_tolerance=args.rooms_tolerance, - holdout_split=args.holdout_split, - ) + if args.engine == "full": + metrics = run_backtest_full(db, sample=args.sample, since=args.since) + 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"]))