gendesign/tradein-mvp/backend/app/tasks/landing_showcase_deals.py

"""Пересчёт витрины лэндинга на РЕАЛЬНЫХ сделках (миграция 276).

ЧТО ЭТО. Публичный лэндинг МЕРЫ показывал ленту «МЕРА сказала X — продали за Y»
на выдуманных константах (frontend `marketing-v3.ts`). Здесь считается её
настоящий источник: берём зарегистрированные ДКП-сделки Росреестра по ЕКБ,
прогоняем каждую через ТОТ ЖЕ спайн оценщика, что и боевой расчёт
(`scripts/backtest_estimator._predict_full_spine` → `estimator._price_from_inputs`),
и кладём получившиеся пары «прогноз / факт» в `landing_showcase_deals`.

ПРАВИЛО ОТБОРА — ЯВНО И БЕЗ ПОДГОНКИ
------------------------------------
Отбираем N строк ключом::

    (полнота данных ↓, свежесть квартала ↓, id сделки ↓)

Величина ошибки в ключе НЕ УЧАСТВУЕТ и участвовать не должна. Отбор по малой
ошибке превращает витрину в рекламу: показанные 20 строк перестают быть
выборкой из работы оценщика и становятся её лучшим хвостом, а посетитель
читает их как «вот так МЕРА обычно и попадает». Это тот самый случай, когда
код формально работает, а продукт врёт. Проверяется тестом
`test_landing_showcase_deals.py::test_selection_ignores_error_magnitude`.

ФИЛЬТРА ПО ОШИБКЕ ТОЖЕ НЕТ — И ЭТО ТО ЖЕ САМОЕ ПРАВИЛО. До 2026-08-29 здесь
жил порог `MAX_ABS_ERR_PCT = 40`, выбрасывавший кандидата ПО ВЕЛИЧИНЕ ОШИБКИ
до ранжирования. Запрет выше он обходил ступенькой раньше: отбор по ошибке в
ключе и отбор по ошибке в фильтре — одно и то же действие, и второе даже
злее, потому что не оставляет строку в кандидатах. Обоснование «отклонение
больше 40% — это почти всегда занижение ДКП ради налога» не держится: см.
следующий раздел, грубые занижения вырезаны выше по потоку и по свойству
самой сделки. Отбраковываем только то, чего в данных НЕТ (нет прогноза, нет
квартала, нет площади) — «число некрасивое» причиной не является.

Полнота — сколько из полей, которые видит посетитель (район, этаж, этажность),
у строки заполнено. Свежесть — порядок квартала сделки.

ЧЕСТНОСТЬ ВИТРИНЫ (нарушение любого пункта = витрина врёт)
----------------------------------------------------------
  * АДРЕСА НЕТ. Номер дома есть у 2.7% сделок, поэтому строка — это «район +
    2-к, 54 м², 5 эт.», и никогда не улица с домом.
  * ДНЯ НЕТ. `deals.deal_date` — первое число квартала (10 различных значений
    на всю таблицу), поэтому в витрине только «II квартал 2026».
  * ЗАМЕР НЕ POINT-IN-TIME. Спайн считает прогноз по СЕГОДНЯШНИМ активным
    объявлениям, а сделка — прошлая. Между ними дрейф рынка, который в ошибку
    входит целиком. Это записано в `note` КАЖДОЙ строки, а не только здесь:
    поле note едет на фронт вместе с числами, а докстринг — нет.
  * ЦЕНА ДКП БЫВАЕТ ЗАНИЖЕНА (налоговая оптимизация, сделки между своими), и
    такая строка выглядит как чудовищный промах оценщика. Санитарный диапазон
    ₽/м² применяется ОДИН раз и ВЫШЕ ПО ПОТОКУ — в `_load_sample`, по свойству
    самой сделки, а не по ошибке прогноза: для ЕКБ это глобальные
    `PPM2_MIN = 30 000` / `PPM2_MAX = 600 000` (город намеренно не заведён в
    `deal_city_price_bands`, там же и комментарий об этом). Значит грубые
    занижения из выборки уже вырезаны ДО того, как сюда приходит кандидат, а
    всё, что после этого дало большую ошибку, — работа оценщика, и витрина
    обязана её показать. Своей копии диапазона здесь нет намеренно: прежние
    `MIN_FACT_PPM2 = 30k` дублировал уже применённый фильтр, а
    `MAX_FACT_PPM2 = 1.2M` был недостижим при потолке выборки 600k — из трёх
    отбраковок в проде срабатывала РОВНО ОДНА, та самая, что льстила витрине.
    Неработающая проверка читается как работающая, поэтому её нет.
  * СЧЁТЧИКИ ЕДУТ НА ФРОНТ, А НЕ ТОЛЬКО В ЛОГ. «Мы показываем 20 отличных
    строк» неотличимо от «столько и было», пока рядом не написано, сколько
    сделок рассмотрено и сколько годных строк не поместилось. Поэтому итог
    прогона пишется в `landing_showcase_runs` (миграция 277) и отдаётся
    ручкой `/api/public/mera/showcase` вместе со строками.

ЗАПУСК (прод, read-mostly: один DELETE+INSERT в свою таблицу)::

    docker exec tradein-backend python -m app.tasks.landing_showcase_deals

Планировщиком пока не дёргается — витрина обновляется редко (сделки приезжают
кварталами), а вешать ежедневный джоб ради данных, которые меняются раз в три
месяца, значит платить сотнями пространственных запросов за ничего.
"""

from __future__ import annotations

import argparse
import logging
from dataclasses import dataclass
from datetime import date
from typing import Any

from sqlalchemy import text
from sqlalchemy.orm import Session

logger = logging.getLogger(__name__)

# ── Правило отбраковки: одна формулировка, она же едет на фронт ──────────────
#
# Порогов на величину ошибки здесь НЕТ (разбор — в докстринге модуля). Санитарный
# диапазон ₽/м² применён выше по потоку, в `_load_sample`; дублировать его тут
# значило бы завести проверку, которая в проде не срабатывает никогда.
REJECTION_RULE = (
    "Строка не попадает на витрину, только если данных нет: оценщик не дал "
    "ожидаемой цены продажи (мало аналогов), неизвестен квартал сделки или "
    "площадь. Величина отклонения на отбор и отбраковку не влияет — иначе "
    "витрина показывала бы лучший хвост, а не работу оценщика. Санитарный "
    "диапазон цены сделки (30 000–600 000 ₽/м² для Екатеринбурга) применён "
    "к выборке до расчёта, по цене самой сделки."
)

NOTE = (
    "Прогноз посчитан по активным объявлениям на дату пересчёта, сделка — прошлая: "
    "это не point-in-time проверка, дрейф рынка за период входит в отклонение целиком. "
    "Факт — цена ДКП, заявленная в Росреестр: она бывает занижена сторонами, и тогда "
    "строка выглядит как промах оценщика, хотя врёт документ."
)

_ROMAN = {1: "I", 2: "II", 3: "III", 4: "IV"}


def quarter_label(d: date | None) -> str | None:
    """`date(2026, 4, 1)` → ``'II квартал 2026'``. Нет даты — нет ярлыка."""
    if d is None:
        return None
    return f"{_ROMAN[(d.month - 1) // 3 + 1]} квартал {d.year}"


@dataclass(frozen=True)
class ShowcaseRow:
    """Одна строка витрины — ровно то, что уедет в таблицу и на фронт."""

    deal_id: int
    district: str | None
    rooms: int
    area_m2: float
    floor: int | None
    total_floors: int | None
    deal_date: date | None
    deal_quarter: str
    predicted_rub: int
    fact_rub: int
    err_pct: float
    n_analogs: int


def completeness(row: ShowcaseRow) -> int:
    """Сколько ВИДИМЫХ посетителю полей заполнено (0..3).

    Считаем район/этаж/этажность: комнаты и площадь есть у всех кандидатов по
    построению выборки, поэтому в оценке полноты они бесполезны.
    """
    return sum(x is not None for x in (row.district, row.floor, row.total_floors))


def _sort_key(row: ShowcaseRow) -> tuple[int, date, int]:
    """Ключ отбора. Ошибки здесь нет — см. «ПРАВИЛО ОТБОРА» в докстринге модуля."""
    return (
        -completeness(row),
        -(row.deal_date or date.min).toordinal(),
        -row.deal_id,
    )


def select_rows(rows: list[ShowcaseRow], limit: int) -> list[ShowcaseRow]:
    """Отобрать `limit` строк по полноте и свежести (НЕ по величине ошибки)."""
    return sorted(rows, key=_sort_key)[:limit]


def build_row(
    *,
    deal_id: int,
    district: str | None,
    rooms: int,
    area_m2: float,
    floor: int | None,
    total_floors: int | None,
    deal_date: date | None,
    predicted_rub: float | None,
    fact_ppm2: float,
    n_analogs: int,
) -> ShowcaseRow | None:
    """Кандидат → строка витрины, либо None если считать не из чего.

    Причины отказа ИСЧЕРПЫВАЮЩИЕ и все — «данных нет»: спайн не дал ожидаемой
    цены продажи; квартал сделки неизвестен; нет площади или цены сделки
    (делить не на что). Величина отклонения причиной НЕ является ни при каких
    значениях — см. «ФИЛЬТРА ПО ОШИБКЕ ТОЖЕ НЕТ» в докстринге модуля.
    """
    if predicted_rub is None or predicted_rub <= 0 or area_m2 <= 0 or fact_ppm2 <= 0:
        return None
    quarter = quarter_label(deal_date)
    if quarter is None:
        return None

    fact_rub = fact_ppm2 * area_m2
    # Знак ошибки — как в бэктесте: (прогноз − факт) / факт. Плюс = МЕРА
    # назвала дороже, чем ушло по ДКП.
    err_pct = 100.0 * (predicted_rub - fact_rub) / fact_rub

    return ShowcaseRow(
        deal_id=deal_id,
        district=district,
        rooms=rooms,
        area_m2=round(area_m2, 2),
        floor=floor,
        total_floors=total_floors,
        deal_date=deal_date,
        deal_quarter=quarter,
        predicted_rub=round(predicted_rub),
        fact_rub=round(fact_rub),
        err_pct=round(err_pct, 2),
        n_analogs=n_analogs,
    )


# ── Район: FDW-вьюха чужой базы, поэтому best-effort ─────────────────────────
_DISTRICT_SQL = text(
    """
    SELECT d.id AS deal_id, g.district_name
    FROM deals d
    JOIN gendesign_ekb_districts_geom g
      ON ST_Contains(g.geom, d.geom::geometry)
    WHERE d.id = ANY(CAST(:ids AS bigint[]))
    """
)


def _fetch_districts(db: Session, deal_ids: list[int]) -> dict[int, str]:
    """id сделки → район. Недоступна вьюха — пустой словарь, а не выдуманный район.

    `gendesign_ekb_districts_geom` — foreign table в базу gendesign, и её гранты
    на той стороне уже терялись (DROP MV CASCADE снимает GRANT). Оборачиваем в
    SAVEPOINT ИМЕННО ЗДЕСЬ, на месте глушения: провалившийся SELECT переводит
    транзакцию в aborted, и следующий запрос упал бы уже не по своей вине.
    """
    if not deal_ids:
        return {}
    try:
        with db.begin_nested():
            rows = db.execute(_DISTRICT_SQL, {"ids": deal_ids}).mappings().all()
    except Exception as exc:
        logger.warning("район не резолвится (витрина будет без района): %s", exc)
        return {}
    return {int(r["deal_id"]): r["district_name"] for r in rows if r["district_name"]}


_DELETE_SQL = text("DELETE FROM landing_showcase_deals")
_DELETE_RUNS_SQL = text("DELETE FROM landing_showcase_runs")
# Тот же `now()`, что у DEFAULT в строках витрины: в Postgres now() — время
# НАЧАЛА транзакции, а батч и его итог пишутся одной транзакцией. Ручка по
# этому computed_at и связывает счётчики со строками.
_INSERT_RUN_SQL = text(
    """
    INSERT INTO landing_showcase_runs
        (considered, priced, no_prediction, incomplete, eligible, written,
         with_district, rejection_rule)
    VALUES
        (CAST(:considered AS integer), CAST(:priced AS integer),
         CAST(:no_prediction AS integer), CAST(:incomplete AS integer),
         CAST(:eligible AS integer), CAST(:written AS integer),
         CAST(:with_district AS integer), CAST(:rejection_rule AS text))
    """
)
_INSERT_SQL = text(
    """
    INSERT INTO landing_showcase_deals
        (district, rooms, area_m2, floor, total_floors, deal_quarter,
         predicted_rub, fact_rub, err_pct, n_analogs, note)
    VALUES
        (CAST(:district AS text), CAST(:rooms AS integer), CAST(:area_m2 AS numeric),
         CAST(:floor AS integer), CAST(:total_floors AS integer),
         CAST(:deal_quarter AS text), CAST(:predicted_rub AS bigint),
         CAST(:fact_rub AS bigint), CAST(:err_pct AS numeric),
         CAST(:n_analogs AS integer), CAST(:note AS text))
    """
)


def refresh_landing_showcase_deals(
    db: Session,
    *,
    sample: int = 200,
    since: str = "2025-01-01",
    limit: int = 20,
    city: str = "Екатеринбург",
) -> dict[str, int]:
    """Прогнать бэктест по ЕКБ и перезаписать витрину. Возвращает счётчики.

    Счётчики — не отладочный шум: без них «на витрине 20 отличных строк»
    неотличимо от «столько и было». Поэтому они не только пишутся в лог, но и
    сохраняются в `landing_showcase_runs` и уезжают на фронт вместе со
    строками. Значения:

      considered    сколько ДКП-сделок взято в работу
      priced        из них оценщик дал ожидаемую цену продажи
      no_prediction не дал (мало аналогов / спайн упал)
      incomplete    цена есть, но нет квартала/площади — строку не собрать
      eligible      годных строк ВСЕГО (никакого отсева по ошибке нет)
      written       из них показано (обрезано по `limit`)
      with_district у скольких показанных удалось определить район
    """
    # Импорт внутри функции: `scripts.backtest_estimator` тянет оценщик со всеми
    # его зависимостями, а web-процессу это на импорте приложения не нужно.
    from scripts.backtest_estimator import (
        _import_estimator_full,
        _load_sample,
        _predict_full_spine,
    )

    est = _import_estimator_full()
    deals = _load_sample(db, sample=sample, since=since, city=city)
    logger.info("витрина: загружено %d ДКП-сделок (city=%s, since=%s)", len(deals), city, since)

    districts = _fetch_districts(db, [d.id for d in deals])

    candidates: list[ShowcaseRow] = []
    n_priced = 0
    n_incomplete = 0
    for deal in deals:
        capture: list[dict[str, Any]] = []
        try:
            pr = _predict_full_spine(db, deal, est, capture=capture)
        except Exception as exc:
            logger.warning("сделка %s: спайн упал, пропускаем: %s", deal.id, exc)
            db.rollback()
            continue
        if pr is None:
            continue
        n_priced += 1
        row = build_row(
            deal_id=deal.id,
            district=districts.get(deal.id),
            rooms=deal.rooms,
            area_m2=deal.area_m2,
            floor=deal.floor,
            total_floors=deal.total_floors,
            deal_date=deal.deal_date,
            predicted_rub=pr.expected_sold_price,
            fact_ppm2=deal.sold_ppm2,
            n_analogs=len(capture[0]["kwargs"]["listings"]) if capture else 0,
        )
        if row is None:
            n_incomplete += 1
            continue
        candidates.append(row)

    chosen = select_rows(candidates, limit)

    db.execute(_DELETE_SQL)
    db.execute(_DELETE_RUNS_SQL)
    for row in chosen:
        db.execute(
            _INSERT_SQL,
            {
                "district": row.district,
                "rooms": row.rooms,
                "area_m2": row.area_m2,
                "floor": row.floor,
                "total_floors": row.total_floors,
                "deal_quarter": row.deal_quarter,
                "predicted_rub": row.predicted_rub,
                "fact_rub": row.fact_rub,
                "err_pct": row.err_pct,
                "n_analogs": row.n_analogs,
                "note": NOTE,
            },
        )
    counters = {
        "considered": len(deals),
        "priced": n_priced,
        "no_prediction": len(deals) - n_priced,
        "incomplete": n_incomplete,
        "eligible": len(candidates),
        "written": len(chosen),
        "with_district": sum(1 for r in chosen if r.district is not None),
    }
    db.execute(_INSERT_RUN_SQL, {**counters, "rejection_rule": REJECTION_RULE})
    db.commit()

    logger.info(
        "витрина обновлена: рассмотрено=%d оценено=%d без_прогноза=%d неполных=%d "
        "годных=%d записано=%d с_районом=%d",
        counters["considered"],
        counters["priced"],
        counters["no_prediction"],
        counters["incomplete"],
        counters["eligible"],
        counters["written"],
        counters["with_district"],
    )
    return counters


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--sample", type=int, default=200)
    parser.add_argument("--since", default="2025-01-01")
    parser.add_argument("--limit", type=int, default=20)
    parser.add_argument("--city", default="Екатеринбург")
    args = parser.parse_args(argv)

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

    from app.core.db import SessionLocal

    db = SessionLocal()
    try:
        refresh_landing_showcase_deals(
            db, sample=args.sample, since=args.since, limit=args.limit, city=args.city
        )
    finally:
        db.close()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())