tradein-mvp/backend/scripts/backtest_estimator.py

@@ -766,11 +766,106 @@ def _expected_sold_metrics(
     return m
 
 
+def _pi_calibration_metrics(
+    rows: list[tuple[float, int]],
+    *,
+    low_mult: float,
+    high_mult: float,
+) -> dict[str, Any]:
+    """PI (prediction interval) calibration — overall + per-rooms (``--pi-report``).
+
+    Each input row is ``(r, rooms)`` where ``r = sold_total / expected_sold_price``
+    — BOTH RUB TOTALS, not ₽/m². This matches exactly how the interval is
+    APPLIED in prod: ``estimator.py`` builds ``expected_sold_range_low/high`` as
+    ``expected_sold_price * estimate_pi_low_mult/high_mult`` (a total-price
+    multiplier). config.py's own comment on those constants — "empirical p10/p90
+    of sold/expected_sold (#1966, n=2366)" — confirms p10/p90 of THIS SAME ratio
+    is exactly what was captured to produce 0.649 / 1.392 originally; per-m² would
+    silently drop the area term and give a different (wrong) distribution.
+
+    For each slice (overall, per room bucket) returns:
+      - ``n``                     : observations with a priced expected_sold_price.
+      - ``p10`` / ``p50`` / ``p90``: percentiles of r — p10/p90 are the CANDIDATE
+                                     new low_mult/high_mult for this slice.
+      - ``achieved_coverage_pct`` : % of r inside the CURRENT [low_mult, high_mult]
+                                     — this is the number to compare against the
+                                     80% target (e.g. Москва: 70.5%).
+      - ``proposed_coverage_pct`` : % of r inside the PROPOSED [p10, p90] — a
+                                     sanity check, ≈80.0 BY CONSTRUCTION (p10/p90
+                                     are defined as the 10th/90th percentile of the
+                                     same sample). If this drifts far from 80, `r`
+                                     was computed on the wrong quantity — mismatch
+                                     with how the mult is applied, not a real find.
+      - ``width_current`` / ``width_proposed``: interval width relative to the
+                                     slice's mean r — ``(high_mult - low_mult) /
+                                     mean(r)`` and ``(p90 - p10) / mean(r)`` — so a
+                                     wider region-specific band shows up as a
+                                     visible precision cost, not just "coverage
+                                     fixed for free".
+
+    Pure: no DB (percentile via the reused estimator ``_percentile``, same as
+    every other block in this file). Rows with expected_sold_price<=0 must
+    already be filtered out by the caller (mirrors ``_range_coverage`` etc.).
+    """
+    _, _percentile = _import_estimator()  # reuse estimator's interpolation percentile
+
+    def _slice(ratios: list[float]) -> dict[str, Any]:
+        n = len(ratios)
+        if n == 0:
+            return {
+                "n": 0,
+                "p10": None,
+                "p50": None,
+                "p90": None,
+                "achieved_coverage_pct": None,
+                "proposed_coverage_pct": None,
+                "width_current": None,
+                "width_proposed": None,
+            }
+        s = sorted(ratios)
+        p10 = _percentile(s, 0.10)
+        p50 = _percentile(s, 0.50)
+        p90 = _percentile(s, 0.90)
+        mean_r = statistics.mean(ratios)
+        achieved = sum(1 for r in ratios if low_mult <= r <= high_mult) / n
+        proposed = sum(1 for r in ratios if p10 <= r <= p90) / n
+        return {
+            "n": n,
+            "p10": round(p10, 4),
+            "p50": round(p50, 4),
+            "p90": round(p90, 4),
+            "achieved_coverage_pct": round(100.0 * achieved, 2),
+            "proposed_coverage_pct": round(100.0 * proposed, 2),
+            "width_current": round((high_mult - low_mult) / mean_r, 4) if mean_r else None,
+            "width_proposed": round((p90 - p10) / mean_r, 4) if mean_r else None,
+        }
+
+    by_bucket: dict[int, list[float]] = {b: [] for b in ROOM_BUCKETS}
+    overall: list[float] = []
+    for r, rooms in rows:
+        overall.append(r)
+        by_bucket[_bucketize_rooms(rooms)].append(r)
+
+    per_rooms: dict[int, dict[str, Any]] = {}
+    for bucket in ROOM_BUCKETS:
+        summary = _slice(by_bucket[bucket])
+        summary["label"] = _rooms_label(bucket)
+        per_rooms[bucket] = summary
+
+    return {
+        "low_mult": low_mult,
+        "high_mult": high_mult,
+        "overall": _slice(overall),
+        "per_rooms": per_rooms,
+    }
+
+
 def _compute_full_metrics(
     predictions: list[Prediction],
     *,
     n_no_prediction: int = 0,
     per_rooms_no_prediction: dict[int, int] | None = None,
+    pi_report: bool = False,
 ) -> dict[str, Any]:
     """Aggregate full-spine Prediction records into the complete metrics dict.
 
@@ -782,6 +877,13 @@ def _compute_full_metrics(
                              coverage).
       - ``calibration``    : per-confidence n / coverage% / MAPE%.
       - ``sharpness``      : median relative range width (high-low)/point.
+      - ``pi_calibration`` : ONLY when ``pi_report=True`` (``--pi-report``) — PI
+                             calibration (p10/p50/p90 of sold_total/expected_
+                             sold_price, achieved vs proposed coverage, width)
+                             overall + per-rooms, to re-derive estimate_pi_low_
+                             mult/high_mult PER REGION instead of the Екатеринбург
+                             constants. Default False → key absent, prior dict
+                             shape/JSON output UNCHANGED (regression-gate safe).
 
     Pure: no DB. Safe on an empty list (every block renders with None/0).
     """
@@ -798,6 +900,9 @@ def _compute_full_metrics(
     sharp_rows: list[tuple[float, float, float]] = []  # (point, range_low, range_high)
     calib_rows: list[tuple[str, float | None, bool | None]] = []
     es_area_rows: list[tuple[float, float, float]] = []  # #3251 (pred, sold, area)
+    pi_rows: list[
+        tuple[float, int]
+    ] = []  # --pi-report only: (sold_total/expected_sold_price, rooms)
 
     for p in predictions:
         signed: float | None = None
@@ -806,6 +911,9 @@ def _compute_full_metrics(
             es_rows.append((p.expected_sold_ppm2, p.sold_ppm2, p.rooms))
             es_area_rows.append((p.expected_sold_ppm2, p.sold_ppm2, p.area_m2))
 
+        if pi_report and p.expected_sold_price is not None and p.expected_sold_price > 0:
+            pi_rows.append((p.sold_total / p.expected_sold_price, p.rooms))
+
         covered: bool | None = None
         if p.range_low is not None and p.range_high is not None:
             covered = p.range_low <= p.sold_total <= p.range_high
@@ -818,7 +926,7 @@ def _compute_full_metrics(
 
         calib_rows.append((p.confidence, signed, covered))
 
-    return {
+    out: dict[str, Any] = {
         "expected_sold": _expected_sold_metrics(
             es_rows,
             n_no_prediction=n_no_prediction,
@@ -833,6 +941,14 @@ def _compute_full_metrics(
         "sharpness": _sharpness(sharp_rows),
         "confidence_order": conf_order,
     }
+    if pi_report:
+        _settings = _import_estimator_full().settings
+        out["pi_calibration"] = _pi_calibration_metrics(
+            pi_rows,
+            low_mult=_settings.estimate_pi_low_mult,
+            high_mult=_settings.estimate_pi_high_mult,
+        )
+    return out
 
 
 def _render_table(metrics: dict[str, Any], headline: dict[str, Any]) -> str:
@@ -1106,6 +1222,53 @@ def _render_calibrate_segments_block(per_segment: dict[str, Any]) -> list[str]:
     return out
 
 
+def _render_pi_calibration_block(pi_calibration: dict[str, Any]) -> list[str]:
+    """Render the PI calibration block (``--pi-report``): overall + per-rooms.
+
+    ``r = sold_total / expected_sold_price``. p10/p90 are candidate replacement
+    low/high multipliers FOR THIS SLICE (region/rooms). achieved_coverage_pct is
+    what the CURRENT constants score here — the number to compare against 80%.
+    proposed_coverage_pct must land ≈80.00 by construction; a big deviation means
+    ``r`` was built from the wrong quantity, not a real calibration finding.
+    """
+    low_mult = pi_calibration["low_mult"]
+    high_mult = pi_calibration["high_mult"]
+    out: list[str] = [
+        "PI CALIBRATION (r = sold_total / expected_sold_price; current mult = "
+        f"[{low_mult:.3f}, {high_mult:.3f}]):",
+    ]
+    header = (
+        f"  {'':<8} {'n':>6} {'p10':>7} {'p50':>7} {'p90':>7} "
+        f"{'achiev%':>8} {'propos%':>8} {'w_cur':>7} {'w_prop':>7}"
+    )
+    out.append(header)
+    out.append("  " + "-" * (len(header) - 2))
+
+    def _row(label: str, s: dict[str, Any]) -> str:
+        n = int(s.get("n", 0) or 0)
+        if n == 0:
+            dash = "—"
+            return (
+                f"  {label:<8} {n:>6} {dash:>7} {dash:>7} {dash:>7} "
+                f"{dash:>8} {dash:>8} {dash:>7} {dash:>7}"
+            )
+        return (
+            f"  {label:<8} {n:>6} {s['p10']:>7.3f} {s['p50']:>7.3f} {s['p90']:>7.3f} "
+            f"{s['achieved_coverage_pct']:>8.2f} {s['proposed_coverage_pct']:>8.2f} "
+            f"{s['width_current']:>7.3f} {s['width_proposed']:>7.3f}"
+        )
+
+    out.append(_row("OVERALL", pi_calibration["overall"]))
+    for bucket in ROOM_BUCKETS:
+        row = pi_calibration["per_rooms"][bucket]
+        out.append(_row(row["label"], row))
+    out.append(
+        "  NB: proposed% should be ≈80.00 by construction (sanity check). "
+        "achieved% is the number to compare against the 80% target."
+    )
+    return out
+
+
 def _render_coverage_block(range_coverage: dict[str, Any], conf_order: list[str]) -> list[str]:
     """Render range-coverage: overall + per-confidence (sold_total ∈ range)."""
     ov = range_coverage["overall"]
@@ -1798,7 +1961,7 @@ def _make_call_stub(
     return _stub
 
 
-def replay_fixture(fixture: dict[str, Any]) -> dict[str, Any]:
+def replay_fixture(fixture: dict[str, Any], *, pi_report: bool = False) -> dict[str, Any]:
     """Replay a frozen backtest fixture through the full spine — hermetic, ZERO DB.
 
     For every captured deal record: rebuild the ``GeocodeResult``, build order-based
@@ -1907,7 +2070,7 @@ def replay_fixture(fixture: dict[str, Any]) -> dict[str, Any]:
         m.settings.estimate_dedup_analogs_enabled = _dedup_saved
 
     # The fixture stores ONLY priced deals, so n_no_prediction is 0 here.
-    metrics = _compute_full_metrics(predictions, n_no_prediction=0)
+    metrics = _compute_full_metrics(predictions, n_no_prediction=0, pi_report=pi_report)
 
     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
@@ -2325,6 +2488,7 @@ def run_backtest_full(
     scattered: bool = False,
     seed: str = "mera",
     region_code: int = DEFAULT_BACKTEST_REGION_CODE,
+    pi_report: bool = False,
 ) -> dict[str, Any]:
     """Drive the FULL-spine read-only backtest and return a metrics dict (#1966).
 
@@ -2363,6 +2527,12 @@ def run_backtest_full(
     (byte-identical to the pre-oblast-D behaviour). Independently of this flag,
     ``_predict_full_spine`` ALWAYS resolves + passes a per-deal target city to
     ``_fetch_dkp_corridor`` (oblast C2 parity fix) — see its docstring.
+
+    ``pi_report`` (default False, ``--pi-report``) attaches a ``pi_calibration``
+    block (see ``_pi_calibration_metrics``) — p10/p50/p90 of sold_total/expected_
+    sold_price overall + per-rooms, plus achieved/proposed PI coverage, so
+    estimate_pi_low_mult/high_mult (calibrated on Екатеринбург only, #1966) can be
+    re-derived PER REGION. Default False → byte-identical prior output.
     """
     est = _import_estimator_full()
     deals = _load_sample(
@@ -2434,6 +2604,7 @@ def run_backtest_full(
         predictions,
         n_no_prediction=n_no_prediction,
         per_rooms_no_prediction=per_rooms_no_prediction,
+        pi_report=pi_report,
     )
 
     deal_median = statistics.median(sold_ppm2_all) if sold_ppm2_all else None
@@ -2634,6 +2805,27 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
         "proposes estimate_segment_multipliers, it does NOT apply them or touch "
         "the baseline. Run with --resolve-house-id for prod-parity biases.",
     )
+    p.add_argument(
+        "--pi-report",
+        action="store_true",
+        help="FULL engine only: after the run, print a PI (prediction interval) "
+        "calibration block — p10/p50/p90 of sold_total/expected_sold_price overall "
+        "+ per-rooms, achieved coverage of the CURRENT estimate_pi_low_mult/"
+        "high_mult (calibrated on Екатеринбург only, #1966), and the coverage the "
+        "sample's own [p10, p90] would give. Use --region to re-derive multipliers "
+        "for a non-EKB region. PRINT-ONLY, applies nothing. Also adds a "
+        "'pi_calibration' key to --json output. Default OFF → byte-identical to "
+        "the prior behaviour (both text and --json).",
+    )
+    p.add_argument(
+        "--pi-report-json",
+        metavar="PATH",
+        default=None,
+        help="With --pi-report: also write JUST the pi_calibration block (plus "
+        "region/city/since/sample run params) to PATH as JSON, so multiple "
+        "per-region runs can be compared without parsing the text table. "
+        "Independent of --json (which dumps the FULL metrics dict to stdout).",
+    )
     # #1966 PR 3/3 — fixture capture + hermetic replay. --dump-fixture (DB run,
     # full engine) and --from-fixture (NO DB) are mutually exclusive modes.
     fixture_mode = p.add_mutually_exclusive_group()
@@ -2663,14 +2855,50 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
     return p.parse_args(argv)
 
 
+def _emit_pi_report(
+    metrics: dict[str, Any],
+    *,
+    json_path: str | None,
+    region_code: int,
+    city: str | None,
+    since: str,
+    sample: int,
+) -> None:
+    """``--pi-report`` side effects — text block + optional JSON — shared by the
+    live-DB and ``--from-fixture`` paths. No-op if ``pi_calibration`` is absent
+    (i.e. ``--pi-report`` wasn't actually threaded into the metrics computation).
+    """
+    pi = metrics.get("pi_calibration")
+    if pi is None:
+        return
+    print("\n" + "\n".join(_render_pi_calibration_block(pi)))
+    if json_path:
+        payload = {
+            "region_code": region_code,
+            "city": city,
+            "since": since,
+            "sample": sample,
+            "pi_calibration": pi,
+        }
+        Path(json_path).write_text(
+            json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
+            encoding="utf-8",
+        )
+        logger.info("wrote pi-report json: %s", json_path)
+        print(f"wrote pi-report json: {json_path}")
+
+
 def main(argv: list[str] | None = None) -> int:
     """CLI entry point. Returns the count of matched (predicted) / replayed deals."""
     args = _parse_args(argv)
 
+    if args.pi_report_json and not args.pi_report:
+        raise SystemExit("--pi-report-json requires --pi-report")
+
     # ── Hermetic replay path (#1966 PR 3/3) — ZERO DB, no SessionLocal opened. ──
     if args.from_fixture:
         fixture = load_fixture(args.from_fixture)
-        metrics = replay_fixture(fixture)
+        metrics = replay_fixture(fixture, pi_report=args.pi_report)
         print(json.dumps(metrics, ensure_ascii=False, indent=2, sort_keys=True))
         if args.update_baseline:
             out = Path(args.update_baseline)
@@ -2679,6 +2907,14 @@ def main(argv: list[str] | None = None) -> int:
                 encoding="utf-8",
             )
             logger.info("wrote baseline: %s", out)
+        _emit_pi_report(
+            metrics,
+            json_path=args.pi_report_json,
+            region_code=args.region,
+            city=args.city,
+            since=args.since,
+            sample=len(fixture.get("deals") or []),
+        )
         return len(fixture.get("deals") or [])
 
     if args.update_baseline:
@@ -2689,6 +2925,8 @@ def main(argv: list[str] | None = None) -> int:
         raise SystemExit("--resolve-house-id is only supported with --engine full")
     if args.calibrate_segments and args.engine != "full":
         raise SystemExit("--calibrate-segments is only supported with --engine full")
+    if args.pi_report and args.engine != "full":
+        raise SystemExit("--pi-report is only supported with --engine full")
 
     logger.info(
         "backtest start: engine=%s region=%d sample=%d since=%s radius=%dm "
@@ -2718,6 +2956,7 @@ def main(argv: list[str] | None = None) -> int:
                 scattered=(args.spread == "scattered"),
                 seed=args.seed,
                 region_code=args.region,
+                pi_report=args.pi_report,
             )
         else:
             metrics = run_backtest(
@@ -2749,6 +2988,15 @@ def main(argv: list[str] | None = None) -> int:
         per_segment = metrics["expected_sold"]["per_segment"]
         print("\n" + "\n".join(_render_calibrate_segments_block(per_segment)))
 
+    _emit_pi_report(
+        metrics,
+        json_path=args.pi_report_json,
+        region_code=args.region,
+        city=args.city,
+        since=args.since,
+        sample=args.sample,
+    )
+
     return int(metrics["params"]["n_matched"])