From 2591968286176c55a29372eb999c363e9bee739f Mon Sep 17 00:00:00 2001 From: bot-backend Date: Wed, 16 Sep 2026 19:05:47 +0300 Subject: [PATCH] =?UTF-8?q?feat(backtest):=20=D0=B7=D0=B0=D0=BC=D0=B5?= =?UTF-8?q?=D1=80=20=D0=BA=D0=B0=D0=BB=D0=B8=D0=B1=D1=80=D0=BE=D0=B2=D0=BA?= =?UTF-8?q?=D0=B8=20=D0=B4=D0=BE=D0=B2=D0=B5=D1=80=D0=B8=D1=82=D0=B5=D0=BB?= =?UTF-8?q?=D1=8C=D0=BD=D0=BE=D0=B3=D0=BE=20=D0=B8=D0=BD=D1=82=D0=B5=D1=80?= =?UTF-8?q?=D0=B2=D0=B0=D0=BB=D0=B0=20=D0=BF=D0=BE=20=D1=80=D0=B5=D0=B3?= =?UTF-8?q?=D0=B8=D0=BE=D0=BD=D0=B0=D0=BC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Вилка цены строится множителями estimate_pi_low_mult/high_mult, которые калибровали один раз на Екатеринбурге (#1966, n=2366) как эмпирические p10/p90 отношения sold/expected_sold. Проверить, что они всё ещё дают заявленные 80% попаданий, было нечем: харнесс считал только p25/p75 знаковой ошибки, то есть 50%-ный интервал. --pi-report печатает p10/p50/p90 отношения r = sold_total / expected_sold_price (рублёвые итоги — множитель в эстиматоре применяется именно к итоговой цене), фактическое покрытие текущей вилки и покрытие предложенной [p10, p90], плюс ширину обеих. Разрез по комнатным корзинам. --pi-report-json кладёт те же числа машиночитаемо, чтобы свести регионы. Множители читаются из живых settings, не хардкодятся. Без флага вывод харнесса байт-в-байт прежний — замороженный регресс-гейт не трогается. Проверено на фикстуре: propos% = 79.7-80.1 по всем корзинам, то есть величина r согласована с определением множителей. --- .../backend/scripts/backtest_estimator.py | 256 +++++++++++++++++- 1 file changed, 252 insertions(+), 4 deletions(-) diff --git a/tradein-mvp/backend/scripts/backtest_estimator.py b/tradein-mvp/backend/scripts/backtest_estimator.py index a6dd8f84..74f0210b 100644 --- a/tradein-mvp/backend/scripts/backtest_estimator.py +++ b/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"]) -- 2.45.3