Light1YT
6f018f4cdb
Extend the read-only harness with an in-memory per-rooms asking→sold ratio (median sold / median pred-ask, global fallback for buckets < MIN_BUCKET=20) and a CORRECTED metrics block re-scored via the existing _compute_metrics. --holdout-split fits on even-id / evaluates on odd-id deals (deterministic) for an honest out-of-sample number. Prod A/B (sample 200, holdout): overall bias +20.4%→-4.0%, MAPE 25.4%→20.2% — a per-rooms ratio removes the systematic asking→sold bias. Still SELECT-only, estimator untouched. +14 tests (33 total).
848 lines
31 KiB
Python
848 lines
31 KiB
Python
"""Backtest harness — measures the estimator's asking-median 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.
|
||
|
||
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:
|
||
|
||
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.
|
||
|
||
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.
|
||
|
||
!! 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.
|
||
|
||
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.
|
||
(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.
|
||
|
||
USAGE
|
||
-----
|
||
DATABASE_URL=postgresql+psycopg://... \
|
||
python -m scripts.backtest_estimator --sample 300 --since 2025-06-01
|
||
|
||
# machine-readable:
|
||
python -m scripts.backtest_estimator --json
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import logging
|
||
import statistics
|
||
from dataclasses import dataclass
|
||
from pathlib import Path
|
||
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 _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
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Data carriers
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
@dataclass
|
||
class DealSample:
|
||
"""One held-out ДКП deal to backtest against."""
|
||
|
||
id: int
|
||
lon: float
|
||
lat: float
|
||
rooms: int
|
||
sold_ppm2: float
|
||
deal_date: Any # datetime.date | None — carried through for reporting only
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# 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
|
||
|
||
|
||
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]) -> list[str]:
|
||
"""Render one OVERALL + per-rooms metrics table (shared by both blocks)."""
|
||
header = (
|
||
f" {'bucket':<8} {'n':>5} {'no_analog':>10} "
|
||
f"{'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(",", " ")
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# 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
|
||
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 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
|
||
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"],
|
||
)
|
||
)
|
||
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 _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"] = {
|
||
"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
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# 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's asking-median accuracy "
|
||
"against rosreestr ДКП sold prices (issue #648)."
|
||
),
|
||
)
|
||
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.",
|
||
)
|
||
return p.parse_args(argv)
|
||
|
||
|
||
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",
|
||
args.sample,
|
||
args.since,
|
||
args.radius,
|
||
args.rooms_tolerance,
|
||
args.holdout_split,
|
||
)
|
||
|
||
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,
|
||
)
|
||
finally:
|
||
db.close()
|
||
|
||
if args.json:
|
||
print(json.dumps(metrics, ensure_ascii=False, indent=2, default=str))
|
||
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)
|