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 эт.», и никогда не улица с домом.
  * ТОЧКА НА КАРТЕ — ЦЕНТРОИД УЛИЦЫ, НЕ ДОМ. В выборке витрины (ЕКБ,
    с 2025-01-01; замер на проде 2026-08-29) 991 различная координата на
    34 017 сделок с координатой — ≈34 сделки в одной точке, при 2.7% известных
    номеров дома. Точка честна на масштабе района и улицы и НЕ честна на
    масштабе дома. Это записано в `note` каждой строки и в COMMENT колонок
    (миграция 280), потому что докстринг на фронт не едет, а зумить карту
    будет тот, кто его не читал.
  * ДНЯ НЕТ. `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 — из трёх
    отбраковок в проде срабатывала РОВНО ОДНА, та самая, что льстила витрине.
    Неработающая проверка читается как работающая, поэтому её нет.
  * КАРТА ПОКАЗЫВАЕТ УЛИЦУ, А НЕ ДОМ. `deals.address` — уровня улицы
    («Екатеринбург, Краснолесья»), поэтому в строку кладётся схема окна вокруг
    ЦЕНТРА улицы (`app/services/street_scheme.py`), и в этой схеме намеренно
    нет ни координат окна, ни констант проекции: точку дома по ней нельзя
    поставить даже случайно. Название сматчилось с OSM у 550 из 654 названий —
    92.3% сделок; остальным `street_scheme` = NULL, и это штатно: фронт
    показывает район. Наличие схемы ВХОДИТ В ПОЛНОТУ (см. `completeness`):
    строка, которой нечем нарисовать карту, полнее строки с картой быть не
    может. Это признак «поле заполнено», как район и этаж, а не величина
    ошибки, — правило отбора выше не нарушено.
  * СЧЁТЧИКИ ЕДУТ НА ФРОНТ, А НЕ ТОЛЬКО В ЛОГ. «Мы показываем 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 json
import logging
from dataclasses import dataclass
from datetime import date
from typing import Any

from sqlalchemy import text
from sqlalchemy.orm import Session

from app.services.street_scheme import StreetIndex, build_street_scheme, load_street_index

logger = logging.getLogger(__name__)

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

NOTE = (
    "Прогноз посчитан по активным объявлениям на дату пересчёта, сделка — прошлая: "
    "это не point-in-time проверка, дрейф рынка за период входит в отклонение целиком. "
    "Факт — цена ДКП из договора (поле price_rub Росреестра, не пересчёт из ₽/м²): "
    "она бывает занижена сторонами, и тогда "
    "строка выглядит как промах оценщика, хотя врёт документ. "
    "Схема на карточке — улица сделки, а не её дом: в адресе Росреестра номер дома "
    "есть у 2.7% строк, поэтому дом не показан и показан быть не может."
    "Точка на карте — центроид улицы, а не дом: в выборке витрины 991 различная "
    "координата на 34 017 сделок (≈34 сделки в одной точке), номер дома известен "
    "у 2.7% сделок. Точка честна на масштабе района и улицы и не честна на масштабе дома."
)

_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
    # Координата сделки — ЦЕНТРОИД УЛИЦЫ (замер и разбор в миграции 280 и в NOTE).
    # None штатно: у части сделок координаты нет, подставлять туда нечего.
    lat: float | None = None
    lon: float | None = None
    # Есть ли чем нарисовать схему улицы (название сделки нашлось в OSM).
    # Саму схему строим только для показанных строк — см. `_schemes_for`.
    has_street: bool = False


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

    Считаем район/этаж/этажность и схему улицы: комнаты и площадь есть у всех
    кандидатов по построению выборки, поэтому в оценке полноты они бесполезны.

    СХЕМА УЛИЦЫ — ТАКОЕ ЖЕ ВИДИМОЕ ПОЛЕ, как район. Строка без улицы рисует на
    фронте полигон РАЙОНА вместо схемы улиц, то есть этого поля у неё просто
    нет, и полнее строки с картой она быть не может. Это признак наличия
    данных, а не величина ошибки: строка с большим отклонением, но со схемой,
    показывается как есть.
    """
    return int(row.has_street) + 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_rub: float | None,
    n_analogs: int,
    lat: float | None = None,
    lon: float | None = None,
    has_street: bool = False,
) -> ShowcaseRow | None:
    """Кандидат → строка витрины, либо None если считать не из чего.

    Причины отказа ИСЧЕРПЫВАЮЩИЕ и все — «данных нет»: спайн не дал ожидаемой
    цены продажи; квартал сделки неизвестен; нет площади или цены сделки
    (делить не на что). Величина отклонения причиной НЕ является ни при каких
    значениях — см. «ФИЛЬТРА ПО ОШИБКЕ ТОЖЕ НЕТ» в докстринге модуля.

    ФАКТ — ЭТО `deals.price_rub`, ЦЕНА ИЗ ДОГОВОРА, А НЕ ПРОИЗВЕДЕНИЕ. Колонка на
    витрине называется «Цена ДКП», и подпись обязана называть ту величину, которая
    показана. До 2026-08-30 здесь считалось `price_per_m2 * area_m2`, а
    `deals.price_rub` лежала рядом и не использовалась: `price_per_m2` в базе
    integer, поэтому произведение промахивалось на единицы рублей (на проде
    4 799 995 вместо 4 800 000, 3 649 995 вместо 3 650 000). Расхождение
    копеечное, но показывалась реконструкция под именем документа.

    Строка без `price_rub` НЕ ПОКАЗЫВАЕТСЯ — это «данных нет», и подставить туда
    реконструкцию значило бы вернуть дефект в одной строке из двадцати, где его
    уже никто не найдёт. Замер на проде 2026-08-30: в выборке витрины (ЕКБ,
    rosreestr, с 2025-01-01, санитарный диапазон) price_rub заполнен у 33 555 из
    33 555 сделок, так что отказ по этой причине — защита, а не рабочий путь.

    ОТСУТСТВИЕ КООРДИНАТЫ ПРИЧИНОЙ ТОЖЕ НЕ ЯВЛЯЕТСЯ. Строка без точки едет на
    витрину с lat=lon=None: карта переживёт сделку без точки, а выбрасывание
    сделки из-за отсутствия координаты — отбор по признаку, не связанному с
    качеством оценки, то есть та же порча витрины, что и отбор по ошибке.
    """
    if predicted_rub is None or predicted_rub <= 0 or area_m2 <= 0:
        return None
    if fact_rub is None or fact_rub <= 0:
        return None
    quarter = quarter_label(deal_date)
    if quarter is None:
        return None

    # Знак ошибки — как в бэктесте: (прогноз − факт) / факт. Плюс = МЕРА
    # назвала дороже, чем ушло по ДКП.
    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,
        lat=lat,
        lon=lon,
        has_street=has_street,
    )


# ── Район: 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,
         lat, lon, street_name, street_scheme)
    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),
         CAST(:lat AS double precision), CAST(:lon AS double precision),
         CAST(:street_name AS text), CAST(:street_scheme AS jsonb))
    """
)


def _schemes_for(
    db: Session,
    index: StreetIndex,
    chosen: list[ShowcaseRow],
    addresses: dict[int, str | None],
) -> dict:
    """Схемы улиц ТОЛЬКО для показанных строк: id сделки → схема.

    Считаем после отбора, а не до: схема — это два пространственных запроса на
    сделку, и на двухстах кандидатах ради двадцати показанных это четыреста
    лишних запросов в чужую базу.

    НА ОТБОР ВЛИЯЕТ НЕ ЭТОТ ШАГ, А ПОЛНОТА (`completeness`), куда наличие улицы
    входит наравне с районом и этажом. До 2026-08-31 здесь было записано
    обратное — «схема на отбор не влияет», — и первой строкой витрины и первым
    раундом игры стояла единственная из двадцати сделка БЕЗ улицы: у неё вместо
    схемы улиц рисовался полигон района. Отсутствие видимого поля не может
    делать строку самой полной. Запрет отбора по ВЕЛИЧИНЕ ОШИБКИ этим не
    затронут: строка с большим отклонением, но со схемой, стоит в витрине как
    есть.
    """
    out = {}
    for row in chosen:
        scheme = build_street_scheme(db, index, addresses.get(row.deal_id))
        if scheme is not None:
            out[row.deal_id] = scheme
    return out


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])
    # Индекс улиц нужен ДО отбора: сматчился ли адрес с OSM — это признак
    # полноты строки. Поиск по индексу идёт в памяти, запрос ровно один на
    # прогон, дорогие пространственные запросы остались в `_schemes_for`.
    street_index = load_street_index(db)

    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_rub=deal.price_rub,
            n_analogs=len(capture[0]["kwargs"]["listings"]) if capture else 0,
            # Порядок ровно такой: lat — широта (~56.8 для ЕКБ), lon — долгота
            # (~60.6). Перепутать местами — это точка в другой стране, и никакой
            # тип этого не поймает: обе величины float.
            lat=deal.lat,
            lon=deal.lon,
            has_street=street_index.lookup(deal.address) is not None,
        )
        if row is None:
            n_incomplete += 1
            continue
        candidates.append(row)

    chosen = select_rows(candidates, limit)
    schemes = _schemes_for(db, street_index, chosen, {d.id: d.address for d in deals})

    db.execute(_DELETE_SQL)
    db.execute(_DELETE_RUNS_SQL)
    for row in chosen:
        scheme = schemes.get(row.deal_id)
        db.execute(
            _INSERT_SQL,
            {
                "street_name": scheme["street"] if scheme else None,
                "street_scheme": json.dumps(scheme, ensure_ascii=False) if scheme else None,
                "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,
                "lat": row.lat,
                "lon": row.lon,
            },
        )
    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),
    }
    # Схем — не счётчик в `landing_showcase_runs` намеренно: у КАЖДОЙ строки
    # витрины street_scheme либо есть, либо null, и это едет на фронт вместе со
    # строкой. Отдельное число повторяло бы то, что посетитель и так видит.
    logger.info("схем улиц собрано: %d из %d показанных", len(schemes), len(chosen))
    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())