"""Backtest harness — measures the estimator's asking-median accuracy vs real ДКП sold prices.

Forgejo issue #648. **STRICTLY READ-ONLY**: this script issues only SELECT
queries against prod. It never INSERTs, UPDATEs, or runs DDL.

WHAT IT MEASURES
----------------
The estimator predicts a квартира's value from the median of *active asking*
prices of nearby analog listings (Tukey 1.5×IQR filter → median ppm²). Ground
truth is `deals` — registered rosreestr ДКП sales (source='rosreestr') that
carry geom + price_per_m2 + rooms. For a held-out sample of ДКП deals we:

  1. Sample deals (id, lon, lat, rooms, sold_ppm2, deal_date).
  2. For each deal, fetch nearby *active* listings of the same rooms within
     `--radius` metres (PostGIS ST_DWithin on geography).
  3. Predict via the estimator's OWN pure functions for fidelity:
     `_filter_outliers` (Tukey IQR) → `_percentile(sorted, 0.5)` (median).
     This mirrors exactly what `estimate_quality` does to its analog pool.
     Deals with <3 surviving candidates are skipped (no prediction).
  4. Per deal: signed_error_pct = 100*(pred - sold)/sold; abs = |signed|.
  5. Aggregate overall + per-rooms: n_matched, n_no_analogs, median_bias_pct
     (systematic over/under), MAPE (median |error|), p25/p75 of signed error.
     Plus a city-wide deal_median_ppm2 vs ask_median_ppm2 headline spread.

CAVEATS (read these before trusting the numbers)
-----------------------------------------------
  (a) TIME MISMATCH — this compares **CURRENT** active listings against
      **PAST** sold deals. It is NOT a point-in-time backtest: a deal closed
      in 2025-06 is being judged against listings active today. As the market
      moves, asking prices drift away from the historical sold price, which
      inflates the apparent bias. A faithful point-in-time backtest needs
      `listing_source_snapshots` (#570) — once that table accumulates enough
      history we can query the asking median *as of each deal's date*.
  (b) PARTIAL ESTIMATOR — this exercises only the asking-median + IQR CORE.
      It does NOT include the full estimator's tiering (same-house / cohort /
      class-coef), DaData enrichment, Avito-IMV, Cian/Yandex valuation, or the
      repair-state coefficient. Real estimate accuracy may differ.
  (c) ДКП ≠ TRUE MARKET — a registered ДКП price is what the parties declared
      to rosreestr; it can diverge from the genuine transaction price (tax
      optimisation, related-party sales, etc.).

PERFORMANCE
-----------
One spatial listings subquery runs per sampled deal, so runtime scales with
`--sample`. The default (300) finishes quickly; large samples (thousands) are
slow because of the per-deal PostGIS ST_DWithin scan. Bump `--sample` only
when you need tighter per-rooms confidence intervals.

USAGE
-----
    DATABASE_URL=postgresql+psycopg://... \
        python -m scripts.backtest_estimator --sample 300 --since 2025-06-01

    # machine-readable:
    python -m scripts.backtest_estimator --json
"""

from __future__ import annotations

import argparse
import json
import logging
import statistics
from dataclasses import dataclass
from pathlib import Path
from typing import Any

from sqlalchemy import text
from sqlalchemy.orm import Session


def _import_estimator() -> tuple[Any, Any]:
    """Lazy import of the estimator's pure funcs (_filter_outliers, _percentile).

    Deferred so `--help` / the pure-metric unit tests don't pull
    app.core.config.Settings (which fail-fasts when DATABASE_URL is unset).
    Supports both `python -m scripts.backtest_estimator` and stand-alone runs.
    """
    try:
        from app.services.estimator import (  # type: ignore[import-not-found]
            _filter_outliers,
            _percentile,
        )
    except ImportError:  # pragma: no cover — fallback for adhoc invocation
        import sys

        sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
        from app.services.estimator import _filter_outliers, _percentile
    return _filter_outliers, _percentile


def _session() -> Session:
    """Lazy SessionLocal factory — see _import_estimator for why it's deferred."""
    try:
        from app.core.db import SessionLocal  # type: ignore[import-not-found]
    except ImportError:  # pragma: no cover — fallback for adhoc invocation
        import sys

        sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
        from app.core.db import SessionLocal
    return SessionLocal()


logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s %(message)s",
)
logger = logging.getLogger("backtest_estimator")

# Price-per-m² sanity band — shared by the deal sample and the listings
# subquery. Mirrors the estimator's working range for EKB вторичка and drops
# obvious data-entry garbage / commercial outliers.
PPM2_MIN = 30_000
PPM2_MAX = 600_000

# Minimum surviving candidates required to emit a prediction. Below this the
# median is too noisy to be meaningful — count the deal as "no analogs".
MIN_CANDIDATES = 3

# Room buckets for the per-rooms breakdown. 0 = студия; the top bucket is "4+".
ROOM_BUCKETS: tuple[int, ...] = (0, 1, 2, 3, 4)


# --------------------------------------------------------------------------- #
# Data carriers
# --------------------------------------------------------------------------- #


@dataclass
class DealSample:
    """One held-out ДКП deal to backtest against."""

    id: int
    lon: float
    lat: float
    rooms: int
    sold_ppm2: float
    deal_date: Any  # datetime.date | None — carried through for reporting only


# --------------------------------------------------------------------------- #
# Pure metric helpers — NO DB. These are unit-tested in
# tests/test_backtest_estimator.py without a live database.
# --------------------------------------------------------------------------- #


def _rooms_label(rooms: int) -> str:
    """Human label for a room bucket: 0 → 'студия', 4 → '4+', else '<n>к'."""
    if rooms <= 0:
        return "студия"
    if rooms >= ROOM_BUCKETS[-1]:
        return f"{ROOM_BUCKETS[-1]}+"
    return f"{rooms}к"


def _bucketize_rooms(rooms: int) -> int:
    """Clamp a raw room count into a ROOM_BUCKETS key (4+ collapse to 4)."""
    if rooms <= 0:
        return 0
    return min(rooms, ROOM_BUCKETS[-1])


def _errors_summary(signed_errors: list[float]) -> dict[str, Any]:
    """Bias / MAPE / spread for a list of signed error percentages.

    - median_bias_pct  = median(signed_errors)  — systematic over/under-predict
    - mape_pct         = median(|signed_errors|) — typical magnitude (robust;
                          we use the MEDIAN absolute error, matching the brief's
                          MAPE-as-median-abs-error definition rather than mean)
    - p25 / p75        = quartiles of the SIGNED error (skew of the bias)

    Empty input → all-None (caller decides how to render "no data").
    """
    n = len(signed_errors)
    if n == 0:
        return {
            "n": 0,
            "median_bias_pct": None,
            "mape_pct": None,
            "p25_pct": None,
            "p75_pct": None,
        }
    abs_errors = [abs(e) for e in signed_errors]
    _, _percentile = _import_estimator()  # reuse estimator's interpolation percentile
    return {
        "n": n,
        "median_bias_pct": round(statistics.median(signed_errors), 2),
        "mape_pct": round(statistics.median(abs_errors), 2),
        "p25_pct": round(_percentile(sorted(signed_errors), 0.25), 2),
        "p75_pct": round(_percentile(sorted(signed_errors), 0.75), 2),
    }


def _compute_metrics(
    rows: list[tuple[float, float, int]],
    *,
    n_no_analogs: int = 0,
    per_rooms_no_analogs: dict[int, int] | None = None,
) -> dict[str, Any]:
    """Aggregate predicted-vs-sold rows into overall + per-rooms metrics.

    Each input row is (pred_ppm2, sold_ppm2, rooms) for a deal that DID get a
    prediction. `n_no_analogs` / `per_rooms_no_analogs` carry the skipped-deal
    counts so the report can show match coverage; they don't affect the error
    stats (those are computed only over matched deals).

    Returns a dict::

        {
          "overall": {n, n_no_analogs, median_bias_pct, mape_pct, p25_pct, p75_pct},
          "per_rooms": {
              0: {label, n, n_no_analogs, median_bias_pct, mape_pct, p25_pct, p75_pct},
              ...
          },
        }

    Pure: no DB, no I/O. signed_error_pct = 100*(pred - sold)/sold per row.
    Rows with sold_ppm2 <= 0 are dropped (cannot divide) — defensive; the SQL
    sample already excludes them.
    """
    per_rooms_no_analogs = per_rooms_no_analogs or {}

    overall_signed: list[float] = []
    by_bucket_signed: dict[int, list[float]] = {b: [] for b in ROOM_BUCKETS}

    for pred_ppm2, sold_ppm2, rooms in rows:
        if sold_ppm2 <= 0:
            continue
        signed = 100.0 * (pred_ppm2 - sold_ppm2) / sold_ppm2
        overall_signed.append(signed)
        by_bucket_signed[_bucketize_rooms(rooms)].append(signed)

    overall = _errors_summary(overall_signed)
    overall["n_no_analogs"] = n_no_analogs

    per_rooms: dict[int, dict[str, Any]] = {}
    for bucket in ROOM_BUCKETS:
        summary = _errors_summary(by_bucket_signed[bucket])
        summary["label"] = _rooms_label(bucket)
        summary["n_no_analogs"] = per_rooms_no_analogs.get(bucket, 0)
        per_rooms[bucket] = summary

    return {"overall": overall, "per_rooms": per_rooms}


def _render_table(metrics: dict[str, Any], headline: dict[str, Any]) -> str:
    """Render the aggregated metrics as a plain-text stdout report."""
    lines: list[str] = []
    lines.append("=" * 78)
    lines.append("BACKTEST: estimator asking-median vs rosreestr ДКП sold prices")
    lines.append("=" * 78)

    # Headline city-wide spread (asking median vs deal median, ppm²).
    dm = headline.get("deal_median_ppm2")
    am = headline.get("ask_median_ppm2")
    spread = headline.get("spread_pct")
    lines.append("")
    lines.append("CITY-WIDE HEADLINE (sample medians, ₽/m²):")
    lines.append(f"  deal_median_ppm2 (SOLD):    {_fmt_ppm2(dm)}")
    lines.append(f"  ask_median_ppm2  (ASKING):  {_fmt_ppm2(am)}")
    lines.append(f"  spread (ask vs deal):       {_fmt_pct(spread)}")

    # Column layout shared by overall + per-rooms rows.
    header = (
        f"  {'bucket':<8} {'n':>5} {'no_analog':>10} "
        f"{'bias%':>8} {'MAPE%':>8} {'p25%':>8} {'p75%':>8}"
    )
    lines.append("")
    lines.append("PER-DEAL ERROR (signed = 100*(pred-sold)/sold; +ve = over-predict):")
    lines.append(header)
    lines.append("  " + "-" * (len(header) - 2))

    overall = metrics["overall"]
    lines.append(_fmt_row("OVERALL", overall))

    for bucket in ROOM_BUCKETS:
        row = metrics["per_rooms"][bucket]
        lines.append(_fmt_row(row["label"], row))

    lines.append("")
    lines.append("Caveats: CURRENT listings vs PAST deals (not point-in-time);")
    lines.append("measures asking-median+IQR core only; ДКП = registered price.")
    lines.append("=" * 78)
    return "\n".join(lines)


def _fmt_row(label: str, m: dict[str, Any]) -> str:
    """Format one metrics row for the table."""
    return (
        f"  {label:<8} {m.get('n', 0):>5} {m.get('n_no_analogs', 0):>10} "
        f"{_fmt_pct(m.get('median_bias_pct')):>8} {_fmt_pct(m.get('mape_pct')):>8} "
        f"{_fmt_pct(m.get('p25_pct')):>8} {_fmt_pct(m.get('p75_pct')):>8}"
    )


def _fmt_pct(v: float | None) -> str:
    return "  n/a" if v is None else f"{v:+.1f}"


def _fmt_ppm2(v: float | None) -> str:
    return "n/a" if v is None else f"{round(v):,}".replace(",", " ")


# --------------------------------------------------------------------------- #
# DB layer — READ-ONLY SELECTs only.
# --------------------------------------------------------------------------- #

# ДКП deal sample. lon/lat extracted via ST_X/ST_Y so the per-deal listings
# query can rebuild the point without re-reading geom. Parameterized;
# psycopg3 CAST(:x AS type), never :x::type.
_SAMPLE_SQL = text(
    """
    SELECT
        id,
        ST_X(geom::geometry) AS lon,
        ST_Y(geom::geometry) AS lat,
        rooms,
        price_per_m2 AS sold_ppm2,
        deal_date
    FROM deals
    WHERE source = 'rosreestr'
      AND geom IS NOT NULL
      AND price_per_m2 BETWEEN CAST(:ppm2_min AS numeric) AND CAST(:ppm2_max AS numeric)
      AND rooms IS NOT NULL
      AND deal_date >= CAST(:since AS date)
    ORDER BY id DESC
    LIMIT CAST(:sample AS integer)
    """
)

# Per-deal candidate active listings. rooms matched within :rooms_lo..:rooms_hi
# (exact when tolerance=0). Returns raw price_per_m2 values — _filter_outliers
# is applied in Python for byte-for-byte fidelity with the estimator.
_CANDIDATES_SQL = text(
    """
    SELECT price_per_m2
    FROM listings
    WHERE is_active
      AND rooms BETWEEN CAST(:rooms_lo AS integer) AND CAST(:rooms_hi AS integer)
      AND price_per_m2 BETWEEN CAST(:ppm2_min AS numeric) AND CAST(:ppm2_max AS numeric)
      AND ST_DWithin(
              geom::geography,
              ST_SetSRID(ST_MakePoint(
                  CAST(:lon AS double precision),
                  CAST(:lat AS double precision)
              ), 4326)::geography,
              CAST(:radius AS double precision)
          )
    """
)


def _load_sample(db: Session, *, sample: int, since: str) -> list[DealSample]:
    """Run the held-out ДКП deal sampling SELECT → list[DealSample]."""
    rows = (
        db.execute(
            _SAMPLE_SQL,
            {
                "ppm2_min": PPM2_MIN,
                "ppm2_max": PPM2_MAX,
                "since": since,
                "sample": sample,
            },
        )
        .mappings()
        .all()
    )
    out: list[DealSample] = []
    for r in rows:
        if r["lon"] is None or r["lat"] is None or r["sold_ppm2"] is None:
            continue
        out.append(
            DealSample(
                id=r["id"],
                lon=float(r["lon"]),
                lat=float(r["lat"]),
                rooms=int(r["rooms"]),
                sold_ppm2=float(r["sold_ppm2"]),
                deal_date=r["deal_date"],
            )
        )
    return out


def _predict_for_deal(
    db: Session,
    deal: DealSample,
    *,
    radius: int,
    rooms_tolerance: int,
) -> float | None:
    """Predict asking ppm² for one deal by reusing the estimator's pure funcs.

    Fetches candidate active-listing ppm² values, wraps them as
    ``{"price_per_m2": p}`` dicts, applies the estimator's `_filter_outliers`
    (Tukey IQR), then `_percentile(sorted, 0.5)`. Returns None when fewer than
    MIN_CANDIDATES survive (caller counts it as a no-analog miss).
    """
    rows = db.execute(
        _CANDIDATES_SQL,
        {
            "rooms_lo": deal.rooms - rooms_tolerance,
            "rooms_hi": deal.rooms + rooms_tolerance,
            "ppm2_min": PPM2_MIN,
            "ppm2_max": PPM2_MAX,
            "lon": deal.lon,
            "lat": deal.lat,
            "radius": radius,
        },
    ).all()

    # Build the same dict shape the estimator feeds _filter_outliers.
    lots = [{"price_per_m2": float(r[0])} for r in rows if r[0] is not None]
    if len(lots) < MIN_CANDIDATES:
        return None

    _filter_outliers, _percentile = _import_estimator()
    clean = _filter_outliers(lots)
    prices = sorted(lot["price_per_m2"] for lot in clean if lot["price_per_m2"])
    if len(prices) < MIN_CANDIDATES:
        return None
    return _percentile(prices, 0.5)


def run_backtest(
    db: Session,
    *,
    sample: int,
    since: str,
    radius: int,
    rooms_tolerance: int,
) -> dict[str, Any]:
    """Drive the full read-only backtest and return a metrics dict.

    Steps: load sample → predict per deal (reusing estimator funcs) → collect
    (pred, sold, rooms) rows + no-analog counts → `_compute_metrics` + a
    city-wide headline spread. No writes.
    """
    deals = _load_sample(db, sample=sample, since=since)
    logger.info("loaded sample: %d ДКП deals (since=%s)", len(deals), since)

    matched_rows: list[tuple[float, float, int]] = []
    n_no_analogs = 0
    per_rooms_no_analogs: dict[int, int] = {b: 0 for b in ROOM_BUCKETS}

    # For the city-wide headline: median of all sampled SOLD ppm², and median
    # of all per-deal predicted ASKING ppm² (matched deals only).
    sold_ppm2_all: list[float] = [d.sold_ppm2 for d in deals]
    pred_ppm2_all: list[float] = []

    for i, deal in enumerate(deals, start=1):
        pred = _predict_for_deal(
            db, deal, radius=radius, rooms_tolerance=rooms_tolerance
        )
        if pred is None:
            n_no_analogs += 1
            per_rooms_no_analogs[_bucketize_rooms(deal.rooms)] += 1
        else:
            matched_rows.append((pred, deal.sold_ppm2, deal.rooms))
            pred_ppm2_all.append(pred)

        if i % 50 == 0:
            logger.info(
                "progress %d/%d (matched=%d, no_analogs=%d)",
                i,
                len(deals),
                len(matched_rows),
                n_no_analogs,
            )

    metrics = _compute_metrics(
        matched_rows,
        n_no_analogs=n_no_analogs,
        per_rooms_no_analogs=per_rooms_no_analogs,
    )

    deal_median = statistics.median(sold_ppm2_all) if sold_ppm2_all else None
    ask_median = statistics.median(pred_ppm2_all) if pred_ppm2_all else None
    spread_pct: float | None = None
    if deal_median and ask_median and deal_median > 0:
        spread_pct = round(100.0 * (ask_median - deal_median) / deal_median, 2)

    metrics["headline"] = {
        "deal_median_ppm2": deal_median,
        "ask_median_ppm2": ask_median,
        "spread_pct": spread_pct,
    }
    metrics["params"] = {
        "sample_requested": sample,
        "sample_loaded": len(deals),
        "since": since,
        "radius_m": radius,
        "rooms_tolerance": rooms_tolerance,
        "n_matched": len(matched_rows),
        "n_no_analogs": n_no_analogs,
    }
    return metrics


# --------------------------------------------------------------------------- #
# Entry point
# --------------------------------------------------------------------------- #


def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    """argparse setup, factored out for testability."""
    p = argparse.ArgumentParser(
        description=(
            "READ-ONLY backtest of the estimator's asking-median accuracy "
            "against rosreestr ДКП sold prices (issue #648)."
        ),
    )
    p.add_argument(
        "--sample",
        type=int,
        default=300,
        help="Held-out ДКП deals to test (default 300). Large samples are slow "
        "— one PostGIS subquery runs per deal.",
    )
    p.add_argument(
        "--since",
        default="2025-06-01",
        help="Only deals with deal_date >= this ISO date (default 2025-06-01).",
    )
    p.add_argument(
        "--radius",
        type=int,
        default=1000,
        help="Analog search radius in metres (default 1000).",
    )
    p.add_argument(
        "--rooms-tolerance",
        type=int,
        default=0,
        help="± room count tolerance for analogs (default 0 = exact match).",
    )
    p.add_argument(
        "--json",
        action="store_true",
        help="Emit machine-readable JSON instead of the text table.",
    )
    return p.parse_args(argv)


def main(argv: list[str] | None = None) -> int:
    """CLI entry point. Returns the count of matched (predicted) deals."""
    args = _parse_args(argv)
    logger.info(
        "backtest start: sample=%d since=%s radius=%dm rooms_tolerance=%d",
        args.sample,
        args.since,
        args.radius,
        args.rooms_tolerance,
    )

    db = _session()
    try:
        metrics = run_backtest(
            db,
            sample=args.sample,
            since=args.since,
            radius=args.radius,
            rooms_tolerance=args.rooms_tolerance,
        )
    finally:
        db.close()

    if args.json:
        print(json.dumps(metrics, ensure_ascii=False, indent=2, default=str))
    else:
        print(_render_table(metrics, metrics["headline"]))

    return int(metrics["params"]["n_matched"])


if __name__ == "__main__":  # pragma: no cover
    raise SystemExit(0 if main() >= 0 else 1)