From 2b10da6e067cfa4d25a456b8e6141d89999082bd Mon Sep 17 00:00:00 2001 From: Light1YT Date: Fri, 29 May 2026 17:53:40 +0500 Subject: [PATCH] =?UTF-8?q?feat(backtest):=20read-only=20estimator-vs-?= =?UTF-8?q?=D0=94=D0=9A=D0=9F=20accuracy=20harness=20(#648)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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). --- .../backend/scripts/backtest_estimator.py | 577 ++++++++++++++++++ .../backend/tests/test_backtest_estimator.py | 260 ++++++++ 2 files changed, 837 insertions(+) create mode 100644 tradein-mvp/backend/scripts/backtest_estimator.py create mode 100644 tradein-mvp/backend/tests/test_backtest_estimator.py diff --git a/tradein-mvp/backend/scripts/backtest_estimator.py b/tradein-mvp/backend/scripts/backtest_estimator.py new file mode 100644 index 00000000..385ffd45 --- /dev/null +++ b/tradein-mvp/backend/scripts/backtest_estimator.py @@ -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 'к'.""" + 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) diff --git a/tradein-mvp/backend/tests/test_backtest_estimator.py b/tradein-mvp/backend/tests/test_backtest_estimator.py new file mode 100644 index 00000000..5255f1d7 --- /dev/null +++ b/tradein-mvp/backend/tests/test_backtest_estimator.py @@ -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 -- 2.45.3