feat(backtest): read-only estimator-vs-ДКП accuracy harness (#648)
Add scripts/backtest_estimator.py — measures the asking-median+IQR estimator core against rosreestr ДКП sold prices. Reuses estimator._filter_outliers / _percentile for fidelity; SELECT-only, no writes. Reports overall + per-rooms bias / MAPE / p25-p75 and a city-wide ask-vs-deal spread; --json + argparse. Prod run (sample 60): estimator over-predicts sold prices by median +29.6% (MAPE 30.5%) — a confirmed systematic asking→sold bias (#648). Pure metric helpers factored out and unit-tested (19 tests, no live DB).
This commit is contained in:
parent
e925a0ba6a
commit
2b10da6e06
2 changed files with 837 additions and 0 deletions
577
tradein-mvp/backend/scripts/backtest_estimator.py
Normal file
577
tradein-mvp/backend/scripts/backtest_estimator.py
Normal file
|
|
@ -0,0 +1,577 @@
|
|||
"""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.
|
||||
|
||||
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)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# 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 _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)}")
|
||||
|
||||
# Column layout shared by overall + per-rooms rows.
|
||||
header = (
|
||||
f" {'bucket':<8} {'n':>5} {'no_analog':>10} "
|
||||
f"{'bias%':>8} {'MAPE%':>8} {'p25%':>8} {'p75%':>8}"
|
||||
)
|
||||
lines.append("")
|
||||
lines.append("PER-DEAL ERROR (signed = 100*(pred-sold)/sold; +ve = over-predict):")
|
||||
lines.append(header)
|
||||
lines.append(" " + "-" * (len(header) - 2))
|
||||
|
||||
overall = metrics["overall"]
|
||||
lines.append(_fmt_row("OVERALL", overall))
|
||||
|
||||
for bucket in ROOM_BUCKETS:
|
||||
row = metrics["per_rooms"][bucket]
|
||||
lines.append(_fmt_row(row["label"], row))
|
||||
|
||||
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 _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 run_backtest(
|
||||
db: Session,
|
||||
*,
|
||||
sample: int,
|
||||
since: str,
|
||||
radius: int,
|
||||
rooms_tolerance: int,
|
||||
) -> 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` + a
|
||||
city-wide headline spread. No writes.
|
||||
"""
|
||||
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]] = []
|
||||
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))
|
||||
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,
|
||||
)
|
||||
|
||||
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,
|
||||
}
|
||||
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(
|
||||
"--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",
|
||||
args.sample,
|
||||
args.since,
|
||||
args.radius,
|
||||
args.rooms_tolerance,
|
||||
)
|
||||
|
||||
db = _session()
|
||||
try:
|
||||
metrics = run_backtest(
|
||||
db,
|
||||
sample=args.sample,
|
||||
since=args.since,
|
||||
radius=args.radius,
|
||||
rooms_tolerance=args.rooms_tolerance,
|
||||
)
|
||||
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)
|
||||
260
tradein-mvp/backend/tests/test_backtest_estimator.py
Normal file
260
tradein-mvp/backend/tests/test_backtest_estimator.py
Normal file
|
|
@ -0,0 +1,260 @@
|
|||
"""Unit tests for the read-only backtest harness (issue #648).
|
||||
|
||||
Covers the PURE aggregation / metric helpers, factored out of the DB code so
|
||||
they're testable without a live database:
|
||||
- _compute_metrics — signed/abs error %, median bias, MAPE, per-rooms split
|
||||
- _errors_summary — bias / MAPE / p25 / p75 of a signed-error list
|
||||
- _bucketize_rooms / _rooms_label — 4+ collapse, студия labelling
|
||||
|
||||
No DB / network / mocks: these operate on plain lists/tuples.
|
||||
|
||||
NOTE: importing scripts.backtest_estimator pulls app.services.estimator →
|
||||
app.core.config.Settings, which REQUIRES DATABASE_URL. Set a dummy value
|
||||
BEFORE importing app modules (same pattern as tests/test_estimator_pure_units.py
|
||||
and tests/test_audit_address_mismatch.py).
|
||||
"""
|
||||
|
||||
import os
|
||||
|
||||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
||||
|
||||
import math
|
||||
|
||||
import pytest
|
||||
|
||||
from scripts import backtest_estimator as bt
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# _bucketize_rooms / _rooms_label
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_bucketize_studio_and_negative_clamp_to_zero() -> None:
|
||||
assert bt._bucketize_rooms(0) == 0
|
||||
assert bt._bucketize_rooms(-3) == 0
|
||||
|
||||
|
||||
def test_bucketize_four_plus_collapses() -> None:
|
||||
assert bt._bucketize_rooms(4) == 4
|
||||
assert bt._bucketize_rooms(5) == 4
|
||||
assert bt._bucketize_rooms(9) == 4
|
||||
|
||||
|
||||
def test_bucketize_passthrough_for_one_to_three() -> None:
|
||||
assert bt._bucketize_rooms(1) == 1
|
||||
assert bt._bucketize_rooms(2) == 2
|
||||
assert bt._bucketize_rooms(3) == 3
|
||||
|
||||
|
||||
def test_rooms_label() -> None:
|
||||
assert bt._rooms_label(0) == "студия"
|
||||
assert bt._rooms_label(1) == "1к"
|
||||
assert bt._rooms_label(3) == "3к"
|
||||
assert bt._rooms_label(4) == "4+"
|
||||
assert bt._rooms_label(7) == "4+"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# _errors_summary
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_errors_summary_empty_returns_all_none() -> None:
|
||||
s = bt._errors_summary([])
|
||||
assert s["n"] == 0
|
||||
assert s["median_bias_pct"] is None
|
||||
assert s["mape_pct"] is None
|
||||
assert s["p25_pct"] is None
|
||||
assert s["p75_pct"] is None
|
||||
|
||||
|
||||
def test_errors_summary_uses_median_abs_for_mape_not_mean() -> None:
|
||||
# signed errors with an asymmetric outlier: median |err| (=10) differs
|
||||
# sharply from the MEAN |err| (=40). The brief defines MAPE as the MEDIAN
|
||||
# absolute error, so we assert the robust median is used.
|
||||
signed = [10.0, 10.0, 10.0, 130.0]
|
||||
s = bt._errors_summary(signed)
|
||||
assert s["mape_pct"] == 10.0 # median(|10,10,10,130|) = 10, not mean 40
|
||||
assert s["median_bias_pct"] == 10.0 # median([10,10,10,130]) = 10
|
||||
|
||||
|
||||
def test_errors_summary_signed_bias_can_be_negative() -> None:
|
||||
# Under-prediction → negative bias.
|
||||
s = bt._errors_summary([-20.0, -10.0, -30.0])
|
||||
assert s["median_bias_pct"] == -20.0
|
||||
assert s["mape_pct"] == 20.0 # median of |[-20,-10,-30]| = median[10,20,30]
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# _compute_metrics — signed/abs error %, bias, MAPE, per-rooms
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_compute_metrics_empty_overall_is_none_per_rooms_all_present() -> None:
|
||||
m = bt._compute_metrics([])
|
||||
assert m["overall"]["n"] == 0
|
||||
assert m["overall"]["median_bias_pct"] is None
|
||||
assert m["overall"]["mape_pct"] is None
|
||||
assert m["overall"]["n_no_analogs"] == 0
|
||||
# Every room bucket must still appear (with n=0) so the report renders.
|
||||
assert set(m["per_rooms"].keys()) == set(bt.ROOM_BUCKETS)
|
||||
for bucket in bt.ROOM_BUCKETS:
|
||||
assert m["per_rooms"][bucket]["n"] == 0
|
||||
assert m["per_rooms"][bucket]["median_bias_pct"] is None
|
||||
assert m["per_rooms"][bucket]["label"] == bt._rooms_label(bucket)
|
||||
|
||||
|
||||
def test_compute_metrics_known_plus_22_pct_overprediction() -> None:
|
||||
# The headline finding: asking median over-predicts SOLD by ~+22%.
|
||||
# pred = 1.22 * sold for every row → signed error must be exactly +22%,
|
||||
# MAPE +22%, p25 == p75 == +22% (no spread).
|
||||
rows = [
|
||||
(122_000.0, 100_000.0, 1),
|
||||
(244_000.0, 200_000.0, 2),
|
||||
(366_000.0, 300_000.0, 3),
|
||||
]
|
||||
m = bt._compute_metrics(rows)
|
||||
assert m["overall"]["n"] == 3
|
||||
assert m["overall"]["median_bias_pct"] == pytest.approx(22.0)
|
||||
assert m["overall"]["mape_pct"] == pytest.approx(22.0)
|
||||
assert m["overall"]["p25_pct"] == pytest.approx(22.0)
|
||||
assert m["overall"]["p75_pct"] == pytest.approx(22.0)
|
||||
|
||||
|
||||
def test_compute_metrics_signed_error_formula() -> None:
|
||||
# Single row, hand-computed: 100*(150k-120k)/120k = +25.0%.
|
||||
m = bt._compute_metrics([(150_000.0, 120_000.0, 2)])
|
||||
assert m["overall"]["median_bias_pct"] == pytest.approx(25.0)
|
||||
assert m["overall"]["mape_pct"] == pytest.approx(25.0)
|
||||
|
||||
|
||||
def test_compute_metrics_abs_error_distinct_from_signed() -> None:
|
||||
# Mixed over/under: signed bias near 0 but MAPE (median |err|) is positive.
|
||||
# rows: +50%, -50%, +50%, -50% → median signed in {-50,+50} band,
|
||||
# median |err| = 50.
|
||||
rows = [
|
||||
(150_000.0, 100_000.0, 1), # +50
|
||||
(50_000.0, 100_000.0, 1), # -50
|
||||
(150_000.0, 100_000.0, 1), # +50
|
||||
(50_000.0, 100_000.0, 1), # -50
|
||||
]
|
||||
m = bt._compute_metrics(rows)
|
||||
assert m["overall"]["mape_pct"] == pytest.approx(50.0)
|
||||
# signed median of [-50,-50,50,50] = 0.0 (mean of two middles)
|
||||
assert m["overall"]["median_bias_pct"] == pytest.approx(0.0)
|
||||
|
||||
|
||||
def test_compute_metrics_per_rooms_split_and_four_plus_collapse() -> None:
|
||||
rows = [
|
||||
(110_000.0, 100_000.0, 0), # студия: +10
|
||||
(130_000.0, 100_000.0, 0), # студия: +30 → median bucket 0 = +20
|
||||
(90_000.0, 100_000.0, 2), # 2к: -10
|
||||
(200_000.0, 100_000.0, 5), # 4+ (5 collapses): +100
|
||||
(300_000.0, 100_000.0, 4), # 4+ : +200 → median bucket 4 = +150
|
||||
]
|
||||
m = bt._compute_metrics(rows)
|
||||
|
||||
assert m["per_rooms"][0]["n"] == 2
|
||||
assert m["per_rooms"][0]["median_bias_pct"] == pytest.approx(20.0)
|
||||
assert m["per_rooms"][0]["label"] == "студия"
|
||||
|
||||
assert m["per_rooms"][2]["n"] == 1
|
||||
assert m["per_rooms"][2]["median_bias_pct"] == pytest.approx(-10.0)
|
||||
|
||||
# rooms=5 and rooms=4 both land in bucket 4.
|
||||
assert m["per_rooms"][4]["n"] == 2
|
||||
assert m["per_rooms"][4]["median_bias_pct"] == pytest.approx(150.0)
|
||||
assert m["per_rooms"][4]["label"] == "4+"
|
||||
|
||||
# buckets 1 and 3 had no rows.
|
||||
assert m["per_rooms"][1]["n"] == 0
|
||||
assert m["per_rooms"][3]["n"] == 0
|
||||
|
||||
# overall n counts every matched row.
|
||||
assert m["overall"]["n"] == 5
|
||||
|
||||
|
||||
def test_compute_metrics_drops_nonpositive_sold() -> None:
|
||||
# sold_ppm2 <= 0 cannot be divided → row dropped, not counted, no crash.
|
||||
rows = [
|
||||
(120_000.0, 0.0, 1), # dropped
|
||||
(120_000.0, -5.0, 2), # dropped
|
||||
(122_000.0, 100_000.0, 1), # kept → +22
|
||||
]
|
||||
m = bt._compute_metrics(rows)
|
||||
assert m["overall"]["n"] == 1
|
||||
assert m["overall"]["median_bias_pct"] == pytest.approx(22.0)
|
||||
|
||||
|
||||
def test_compute_metrics_carries_no_analog_counts() -> None:
|
||||
rows = [(122_000.0, 100_000.0, 1)]
|
||||
m = bt._compute_metrics(
|
||||
rows,
|
||||
n_no_analogs=7,
|
||||
per_rooms_no_analogs={1: 4, 2: 3},
|
||||
)
|
||||
assert m["overall"]["n_no_analogs"] == 7
|
||||
assert m["per_rooms"][1]["n_no_analogs"] == 4
|
||||
assert m["per_rooms"][2]["n_no_analogs"] == 3
|
||||
# bucket with no skipped deals defaults to 0.
|
||||
assert m["per_rooms"][0]["n_no_analogs"] == 0
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Rendering smoke tests — table + empty render must not crash.
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_render_table_runs_on_real_metrics() -> None:
|
||||
m = bt._compute_metrics([(122_000.0, 100_000.0, 1)], n_no_analogs=2)
|
||||
headline = {
|
||||
"deal_median_ppm2": 100_000.0,
|
||||
"ask_median_ppm2": 122_000.0,
|
||||
"spread_pct": 22.0,
|
||||
}
|
||||
out = bt._render_table(m, headline)
|
||||
assert "BACKTEST" in out
|
||||
assert "OVERALL" in out
|
||||
assert "+22.0" in out # bias rendered with sign
|
||||
assert "100 000" in out # ppm2 formatted with space thousands separator
|
||||
|
||||
|
||||
def test_render_table_handles_empty_sample() -> None:
|
||||
m = bt._compute_metrics([])
|
||||
headline = {"deal_median_ppm2": None, "ask_median_ppm2": None, "spread_pct": None}
|
||||
out = bt._render_table(m, headline)
|
||||
assert "n/a" in out # None metrics render as n/a, no crash
|
||||
|
||||
|
||||
def test_fmt_helpers_handle_none_and_nan_safely() -> None:
|
||||
assert bt._fmt_pct(None) == " n/a"
|
||||
assert bt._fmt_ppm2(None) == "n/a"
|
||||
# sanity: finite values format
|
||||
assert "+22" in bt._fmt_pct(22.0)
|
||||
assert not math.isnan(22.0)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# argparse — defaults match the brief.
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_argparse_defaults() -> None:
|
||||
ns = bt._parse_args([])
|
||||
assert ns.sample == 300
|
||||
assert ns.since == "2025-06-01"
|
||||
assert ns.radius == 1000
|
||||
assert ns.rooms_tolerance == 0
|
||||
assert ns.json is False
|
||||
|
||||
|
||||
def test_argparse_overrides() -> None:
|
||||
ns = bt._parse_args(
|
||||
["--sample", "50", "--since", "2024-01-01", "--radius", "2000",
|
||||
"--rooms-tolerance", "1", "--json"]
|
||||
)
|
||||
assert ns.sample == 50
|
||||
assert ns.since == "2024-01-01"
|
||||
assert ns.radius == 2000
|
||||
assert ns.rooms_tolerance == 1
|
||||
assert ns.json is True
|
||||
Loading…
Add table
Reference in a new issue