feat(backtest): per-rooms asking→sold correction block (#648 S1) (#650)

tradein-mvp/backend/scripts/README.md

@@ -225,3 +225,49 @@ final listing_sources coverage:
   cian       5158 / 5158 (100.0%)
   ...
 ```
+
+---
+
+## Estimator backtest (issue #648)
+
+### `backtest_estimator.py` — asking→sold accuracy harness
+
+**STRICTLY READ-ONLY** (SELECT-only; no INSERT/UPDATE/DDL/commit). Measures the
+estimator's asking-median + Tukey-IQR core against rosreestr ДКП **sold** prices.
+For a sample of ДКП deals it predicts the asking median from nearby active
+listings (reusing the estimator's own `_filter_outliers` / `_percentile`), then
+reports per-deal signed/abs error % aggregated overall + per-rooms (студия / 1к /
+2к / 3к / 4+), plus a city-wide deal-vs-asking headline spread.
+
+```bash
+DATABASE_URL=postgresql+psycopg://... \
+    python -m scripts.backtest_estimator --sample 300 --since 2025-06-01
+
+# machine-readable:
+python -m scripts.backtest_estimator --json
+```
+
+**Stage 1 correction block.** On top of the raw `[ASKING]` metrics the harness
+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 buckets with `< MIN_BUCKET = 20`
+matched deals), then re-scores `pred_sold = pred_ask * ratio[bucket]` through the
+same metric math. This DEMONSTRATES that a per-rooms factor removes the
+systematic +29.6% asking→sold bias — it changes nothing in prod.
+
+> **Honesty:** by default the ratio is IN-SAMPLE (derived AND evaluated on the
+> same deals), so the corrected bias is 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 A/B'd on held-out data.
+
+```bash
+# honest out-of-sample corrected number (even-id fit / odd-id eval):
+python -m scripts.backtest_estimator --sample 600 --holdout-split
+```
+
+Caveats (also printed): CURRENT listings vs PAST deals (not point-in-time —
+needs `listing_source_snapshots` #570); asking-median + IQR core only; ДКП =
+registered price. Pure metric/ratio helpers are unit-tested in
+`tests/test_backtest_estimator.py` (no DB).

tradein-mvp/backend/scripts/backtest_estimator.py

@@ -22,6 +22,24 @@ carry geom + price_per_m2 + rooms. For a held-out sample of ДКП deals we:
      (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
@@ -120,6 +138,12 @@ 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
@@ -243,6 +267,113 @@ def _compute_metrics(
     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] = []
@@ -260,22 +391,56 @@ def _render_table(metrics: dict[str, Any], headline: dict[str, Any]) -> str:
     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}"
+    # 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.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.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);")
@@ -284,6 +449,41 @@ def _render_table(metrics: dict[str, Any], headline: dict[str, Any]) -> str:
     return "\n".join(lines)
 
 
+def _render_metrics_block(title: str, metrics: dict[str, Any]) -> list[str]:
+    """Render one OVERALL + per-rooms metrics table (shared by both blocks)."""
+    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 (
@@ -422,6 +622,50 @@ def _predict_for_deal(
     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,
     *,
@@ -429,17 +673,27 @@ def run_backtest(
     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` + a
-    city-wide headline spread. No writes.
+    (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}
 
@@ -457,6 +711,7 @@ def run_backtest(
             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:
@@ -474,6 +729,10 @@ def run_backtest(
         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
@@ -493,6 +752,7 @@ def run_backtest(
         "rooms_tolerance": rooms_tolerance,
         "n_matched": len(matched_rows),
         "n_no_analogs": n_no_analogs,
+        "holdout_split": holdout_split,
     }
     return metrics
 
@@ -534,6 +794,14 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
         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",
@@ -546,11 +814,13 @@ 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",
+        "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()
@@ -561,6 +831,7 @@ def main(argv: list[str] | None = None) -> int:
             since=args.since,
             radius=args.radius,
             rooms_tolerance=args.rooms_tolerance,
+            holdout_split=args.holdout_split,
         )
     finally:
         db.close()

tradein-mvp/backend/tests/test_backtest_estimator.py

@@ -5,6 +5,8 @@ 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
+  - _derive_room_ratios — per-rooms asking→sold ratio, global fallback, guards
+  - _apply_ratios + corrected metrics — ratios that cancel a known bias → ~0
 
 No DB / network / mocks: these operate on plain lists/tuples.
 
@@ -200,6 +202,173 @@ def test_compute_metrics_carries_no_analog_counts() -> None:
     assert m["per_rooms"][0]["n_no_analogs"] == 0
 
 
+# --------------------------------------------------------------------------- #
+# _derive_room_ratios — per-rooms asking→sold ratio, global fallback, guards.
+# --------------------------------------------------------------------------- #
+
+
+def _rows_for_bucket(
+    bucket: int, *, n: int, ask: float, sold: float
+) -> list[tuple[float, float, int]]:
+    """n identical (ask, sold, bucket) rows — keeps per-bucket median == ask/sold."""
+    return [(ask, sold, bucket) for _ in range(n)]
+
+
+def test_derive_ratios_per_bucket_exact() -> None:
+    # Each bucket ≥ MIN_BUCKET deals so every bucket gets its OWN ratio.
+    # bucket 1: sold/ask = 80k/100k = 0.80 ; bucket 2: 150k/200k = 0.75.
+    rows = (
+        _rows_for_bucket(1, n=bt.MIN_BUCKET, ask=100_000.0, sold=80_000.0)
+        + _rows_for_bucket(2, n=bt.MIN_BUCKET, ask=200_000.0, sold=150_000.0)
+    )
+    ratios, meta = bt._derive_room_ratios(rows)
+    assert ratios[1] == pytest.approx(0.80)
+    assert ratios[2] == pytest.approx(0.75)
+    assert meta["fallback_buckets"] == []  # both buckets were big enough
+    assert meta["bucket_n"][1] == bt.MIN_BUCKET
+    assert meta["bucket_n"][2] == bt.MIN_BUCKET
+
+
+def test_derive_ratios_median_not_mean() -> None:
+    # A bucket whose ask/sold pairs vary: ratio must use the MEDIAN of each
+    # series, not a per-row mean. asks median = 100k, solds median = 90k → 0.9.
+    rows = [
+        (80_000.0, 60_000.0, 1),
+        (100_000.0, 90_000.0, 1),  # median row
+        (300_000.0, 200_000.0, 1),
+        *_rows_for_bucket(1, n=bt.MIN_BUCKET, ask=100_000.0, sold=90_000.0),
+    ]
+    ratios, _ = bt._derive_room_ratios(rows)
+    # median ask and median sold are both pinned to 100k/90k by the padding.
+    assert ratios[1] == pytest.approx(0.90)
+
+
+def test_derive_ratios_thin_bucket_uses_global_fallback() -> None:
+    # bucket 1 has plenty (own ratio 0.80); bucket 2 has only 1 deal (< MIN) →
+    # must inherit the GLOBAL ratio, and be flagged as a fallback bucket.
+    rows = [
+        *_rows_for_bucket(1, n=bt.MIN_BUCKET, ask=100_000.0, sold=80_000.0),
+        (200_000.0, 120_000.0, 2),  # lone bucket-2 deal
+    ]
+    ratios, meta = bt._derive_room_ratios(rows)
+    assert ratios[1] == pytest.approx(0.80)
+    assert 2 in meta["fallback_buckets"]
+    # global = median(all sold) / median(all ask). With MIN_BUCKET copies of
+    # (100k/80k) plus one (200k/120k), both medians stay at the dense point.
+    assert ratios[2] == pytest.approx(meta["global_ratio"])
+    assert meta["global_ratio"] is not None
+
+
+def test_derive_ratios_respects_custom_min_bucket() -> None:
+    # With min_bucket=2, a 1-deal bucket falls back; a 2-deal bucket keeps own.
+    rows = [
+        (100_000.0, 50_000.0, 1),  # lone bucket-1 deal → fallback
+        (100_000.0, 90_000.0, 2),
+        (100_000.0, 90_000.0, 2),  # 2 deals → own ratio 0.9
+    ]
+    ratios, meta = bt._derive_room_ratios(rows, min_bucket=2)
+    assert 1 in meta["fallback_buckets"]
+    assert 2 not in meta["fallback_buckets"]
+    assert ratios[2] == pytest.approx(0.90)
+
+
+def test_derive_ratios_empty_returns_empty_and_safe_meta() -> None:
+    ratios, meta = bt._derive_room_ratios([])
+    assert ratios == {}
+    assert meta["global_ratio"] is None
+    assert meta["fallback_buckets"] == []
+    assert all(n == 0 for n in meta["bucket_n"].values())
+
+
+def test_derive_ratios_skips_nonpositive_and_guards_div_by_zero() -> None:
+    # pred<=0 or sold<=0 rows are dropped; a bucket left with only bad rows
+    # gets neither its own ratio NOR a (here non-existent) global one → omitted,
+    # and the function does not raise ZeroDivisionError.
+    rows = [
+        (0.0, 100_000.0, 1),  # pred 0 → dropped (would div-by-zero)
+        (100_000.0, 0.0, 1),  # sold 0 → dropped
+        (-5.0, 100_000.0, 2),  # pred <0 → dropped
+    ]
+    ratios, meta = bt._derive_room_ratios(rows)
+    assert ratios == {}  # nothing valid survived
+    assert meta["global_ratio"] is None  # no valid pred → no global ratio
+
+
+def test_derive_ratios_bucket_zero_global_pred_no_ratio() -> None:
+    # If a bucket's own pred median is 0 (all-zero preds) it can't form a ratio;
+    # with no global fallback either, it must be omitted, not crash.
+    rows = [(0.0, 100_000.0, 1) for _ in range(bt.MIN_BUCKET)]
+    ratios, meta = bt._derive_room_ratios(rows)
+    assert 1 not in ratios
+    assert meta["global_ratio"] is None
+
+
+# --------------------------------------------------------------------------- #
+# _apply_ratios + corrected metrics — ratios that cancel a known bias → ~0.
+# --------------------------------------------------------------------------- #
+
+
+def test_apply_ratios_multiplies_pred_by_bucket_ratio() -> None:
+    rows = [(100_000.0, 90_000.0, 1), (200_000.0, 150_000.0, 2)]
+    out = bt._apply_ratios(rows, {1: 0.9, 2: 0.75})
+    assert out[0] == (pytest.approx(90_000.0), 90_000.0, 1)
+    assert out[1] == (pytest.approx(150_000.0), 150_000.0, 2)
+
+
+def test_apply_ratios_missing_bucket_leaves_pred_unchanged() -> None:
+    # bucket 3 absent from the ratios map → identity multiplier (×1.0).
+    rows = [(123_456.0, 100_000.0, 3)]
+    out = bt._apply_ratios(rows, {1: 0.9})
+    assert out[0][0] == pytest.approx(123_456.0)
+
+
+def test_corrected_metrics_cancel_plus_30_pct_bias_to_zero() -> None:
+    # Construct a uniform +30% asking bias (pred = 1.30 * sold) across buckets
+    # with enough deals that each bucket forms its OWN ratio. Deriving the ratio
+    # in-sample and re-applying it MUST collapse the corrected bias to ~0.
+    rows: list[tuple[float, float, int]] = []
+    for bucket, sold in ((0, 80_000.0), (1, 100_000.0), (2, 150_000.0)):
+        rows += _rows_for_bucket(
+            bucket, n=bt.MIN_BUCKET, ask=1.30 * sold, sold=sold
+        )
+
+    # sanity: the ASKING block really is +30%.
+    asking = bt._compute_metrics(rows)
+    assert asking["overall"]["median_bias_pct"] == pytest.approx(30.0)
+
+    ratios, meta = bt._derive_room_ratios(rows)
+    # every per-bucket ratio == 1/1.30 ≈ 0.7692.
+    for bucket in (0, 1, 2):
+        assert ratios[bucket] == pytest.approx(1.0 / 1.30, rel=1e-6)
+    assert meta["fallback_buckets"] == []
+
+    corrected = bt._compute_metrics(bt._apply_ratios(rows, ratios))
+    assert corrected["overall"]["median_bias_pct"] == pytest.approx(0.0, abs=1e-6)
+    assert corrected["overall"]["mape_pct"] == pytest.approx(0.0, abs=1e-6)
+    for bucket in (0, 1, 2):
+        assert corrected["per_rooms"][bucket]["median_bias_pct"] == pytest.approx(
+            0.0, abs=1e-6
+        )
+
+
+def test_corrected_metrics_global_fallback_cancels_uniform_bias() -> None:
+    # Even when buckets are too thin for their OWN ratio, the GLOBAL fallback
+    # (uniform +30% here) still cancels the systematic bias to ~0.
+    rows = [
+        (130_000.0, 100_000.0, 1),  # +30
+        (260_000.0, 200_000.0, 2),  # +30
+        (104_000.0, 80_000.0, 0),  # +30
+    ]
+    ratios, meta = bt._derive_room_ratios(rows)  # all buckets < MIN_BUCKET
+    assert set(meta["fallback_buckets"]) == {0, 1, 2}
+    # meta["global_ratio"] is rounded to 4dp for the report; the APPLIED ratios
+    # in `ratios` keep full precision, so the corrected bias still cancels to 0.
+    assert meta["global_ratio"] == pytest.approx(0.7692, abs=5e-5)
+    assert ratios[1] == pytest.approx(1.0 / 1.30, rel=1e-9)  # full precision applied
+    corrected = bt._compute_metrics(bt._apply_ratios(rows, ratios))
+    assert corrected["overall"]["median_bias_pct"] == pytest.approx(0.0, abs=1e-6)
+
+
 # --------------------------------------------------------------------------- #
 # Rendering smoke tests — table + empty render must not crash.
 # --------------------------------------------------------------------------- #
@@ -226,6 +395,55 @@ def test_render_table_handles_empty_sample() -> None:
     assert "n/a" in out  # None metrics render as n/a, no crash
 
 
+def test_render_table_includes_corrected_block_and_in_sample_warning() -> None:
+    # Build a metrics dict the way run_backtest does, with a corrected block.
+    rows = [(130_000.0, 100_000.0, 1)] * bt.MIN_BUCKET
+    m = bt._compute_metrics(rows)
+    ratios, meta = bt._derive_room_ratios(rows)
+    meta["holdout_split"] = False
+    m["ratios"] = ratios
+    m["ratios_meta"] = meta
+    m["corrected"] = bt._compute_metrics(bt._apply_ratios(rows, ratios))
+    headline = {
+        "deal_median_ppm2": 100_000.0,
+        "ask_median_ppm2": 130_000.0,
+        "spread_pct": 30.0,
+    }
+    out = bt._render_table(m, headline)
+    assert "ASKING" in out
+    assert "CORRECTED" in out
+    assert "ratio=" in out  # derived ratio line rendered
+    assert "IN-SAMPLE" in out  # honesty caveat shown when not holdout
+
+
+def test_render_table_corrected_block_holdout_message() -> None:
+    rows = [(130_000.0, 100_000.0, 1)] * bt.MIN_BUCKET
+    m = bt._compute_metrics(rows)
+    ratios, meta = bt._derive_room_ratios(rows)
+    meta["holdout_split"] = True
+    m["ratios"] = ratios
+    m["ratios_meta"] = meta
+    m["corrected"] = bt._compute_metrics(bt._apply_ratios(rows, ratios))
+    headline = {"deal_median_ppm2": None, "ask_median_ppm2": None, "spread_pct": None}
+    out = bt._render_table(m, headline)
+    assert "OUT-OF-SAMPLE" in out
+    assert "IN-SAMPLE" not in out  # holdout path swaps the caveat
+
+
+def test_render_table_no_corrected_block_when_absent() -> None:
+    # Backward-compat: a metrics dict without "corrected" still renders (ASKING
+    # only) and does not raise.
+    m = bt._compute_metrics([(122_000.0, 100_000.0, 1)])
+    headline = {
+        "deal_median_ppm2": 100_000.0,
+        "ask_median_ppm2": 122_000.0,
+        "spread_pct": 22.0,
+    }
+    out = bt._render_table(m, headline)
+    assert "ASKING" in out
+    assert "CORRECTED" not in out
+
+
 def test_fmt_helpers_handle_none_and_nan_safely() -> None:
     assert bt._fmt_pct(None) == "  n/a"
     assert bt._fmt_ppm2(None) == "n/a"
@@ -246,15 +464,17 @@ def test_argparse_defaults() -> None:
     assert ns.radius == 1000
     assert ns.rooms_tolerance == 0
     assert ns.json is False
+    assert ns.holdout_split is False
 
 
 def test_argparse_overrides() -> None:
     ns = bt._parse_args(
         ["--sample", "50", "--since", "2024-01-01", "--radius", "2000",
-         "--rooms-tolerance", "1", "--json"]
+         "--rooms-tolerance", "1", "--holdout-split", "--json"]
     )
     assert ns.sample == 50
     assert ns.since == "2024-01-01"
     assert ns.radius == 2000
     assert ns.rooms_tolerance == 1
+    assert ns.holdout_split is True
     assert ns.json is True