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`.

ПРАВИЛО ОТБОРА — ЯВНО И БЕЗ ПОДГОНКИ
------------------------------------
Витрина показывает ПОЛОСУ РАСХОЖДЕНИЯ, а не всю сверку. С 2026-09-12 решением
владельца продукта на витрину попадают только сделки, у которых расхождение
прогноза с ценой ДКП лежит в пределах `BAND_MIN_ERR_PCT`..`BAND_MAX_ERR_PCT`
(−5 %..+20 % включительно). Оставшиеся `limit` строк ранжируются ключом::

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

ЭТО ОТБОР ПОКАЗАТЕЛЬНЫХ СТРОК, И НАЗЫВАТЬ ЕГО НАДО ТАК. До 2026-09-12 здесь
не было ни фильтра, ни слагаемого ошибки в ключе, и подпись витрины это прямо
утверждала. Теперь утверждать это нельзя: строки с промахом крупнее полосы в
данных есть (на проде 30.08.2026 из показанных двадцати вне полосы было
двенадцать — от −27,9 % до +75,7 %), и они не показываются. Поэтому полоса
названа в `REJECTION_RULE`, которое едет на фронт вместе со счётчиками
прогона, и в подписи под таблицей рядом с медианой расхождения ПО ВСЕЙ
СВЕРКЕ: два числа рядом не дают прочитать двадцать отобранных строк как
«вот так МЕРА обычно и попадает».

ЧТО ЭТО НЕ ОТМЕНЯЕТ. Внутри полосы отбор по величине ошибки по-прежнему
запрещён — иначе витрина показывала бы лучший хвост уже самой полосы
(`test_selection_ignores_error_magnitude`). Счётчики прогона считаются ДО
полосы: `eligible` — сколько строк прогон вообще собрал, `written` — сколько
из них прошло полосу и поместилось в `limit`. Разница между ними видна
посетителю, и она честная ровно потому, что рядом сказано, чем именно
отобраны показанные. Если в полосу попало меньше `limit` строк — показываем
сколько есть; добирать соседями по ошибке нельзя, это вернуло бы отбор по
величине ошибки в обход полосы.

Отбраковка по «данных нет» (нет прогноза, нет квартала, нет площади) осталась
прежней и живёт в `build_row`: строка вне полосы ОСТАЁТСЯ кандидатом и
попадает в счётчик `eligible`, её снимает отбор, а не отбраковка.

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

ЧЕСТНОСТЬ ВИТРИНЫ (нарушение любого пункта = витрина врёт)
----------------------------------------------------------
  * АДРЕСА НЕТ. Номер дома есть у 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`, там же и комментарий об этом). Значит грубые
    занижения из выборки уже вырезаны ДО того, как сюда приходит кандидат, а
    всё, что после этого дало большую ошибку, — работа оценщика. С 2026-09-12
    такая строка на витрину не выходит (полоса), но остаётся в `eligible` и
    в подписи названа отобранной, а не несуществующей. Своей копии диапазона
    здесь нет намеренно: прежние
    `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__)

# ── Полоса расхождения: что показываем и что об этом сказано ─────────────────
#
# Границы ВКЛЮЧИТЕЛЬНЫЕ. Полоса несимметрична намеренно: решение владельца от
# 2026-09-12 — показывать сделки, где МЕРА не занизила больше чем на 5 % и не
# завысила больше чем на 20 %.
BAND_MIN_ERR_PCT = -5.0
BAND_MAX_ERR_PCT = 20.0

# Подпись полосы ВЫВОДИТСЯ из границ, а не вписывается рядом: «−5 %…+20 %» в
# тексте и `>= -5.0` в коде — две независимые величины, и разъедутся они
# ровно тогда, когда порог однажды подвинут.
BAND_LABEL = f"от {BAND_MIN_ERR_PCT:+.0f} % до {BAND_MAX_ERR_PCT:+.0f} % включительно"


def in_band(err_pct: float) -> bool:
    """Попадает ли расхождение в показываемую полосу (границы включительно)."""
    return BAND_MIN_ERR_PCT <= err_pct <= BAND_MAX_ERR_PCT


# ── Правило отбора и отбраковки: одна формулировка, она же едет на фронт ──────
#
# Санитарный диапазон ₽/м² применён выше по потоку, в `_load_sample`;
# дублировать его тут значило бы завести проверку, которая в проде не
# срабатывает никогда.
#
# ТЕКСТ ОБЯЗАН НАЗЫВАТЬ ПОЛОСУ. Пока фильтра не было, здесь стояло «величина
# отклонения на отбор и отбраковку не влияет — иначе витрина показывала бы
# лучший хвост, а не работу расчёта». С фильтром эта фраза стала ложью ровно
# про то, чего опасалась, поэтому она снята, а не смягчена.
REJECTION_RULE = (
    f"На витрине — ОТОБРАННАЯ полоса расхождения, а не вся сверка: показаны "
    f"только сделки, у которых расхождение прогноза с ценой ДКП лежит {BAND_LABEL}. "
    "Промахи крупнее полосы в данных есть, и здесь их не видно — судить по этим "
    "строкам о точности расчёта нельзя, для этого есть медиана расхождения по "
    "всей сверке. Внутри полосы порядок задают полнота данных и свежесть "
    "квартала: величина отклонения на него не влияет, лучший хвост самой полосы "
    "витрина тоже не показывает. Кроме полосы строку снимает только отсутствие "
    "данных: расчёт МЕРЫ не дал ожидаемой цены продажи (мало аналогов), "
    "неизвестен квартал сделки или площадь. Санитарный диапазон цены сделки "
    "(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]:
    """Строки полосы −5 %..+20 %, до `limit` штук, по полноте и свежести.

    ДВА ДЕЙСТВИЯ, И ОНИ РАЗНЫЕ. Сначала ФИЛЬТР по величине расхождения
    (`in_band`) — это и есть «витрина показывает отобранную полосу, а не всю
    сверку», названное так же в `REJECTION_RULE` и в подписи под таблицей.
    Потом РАНЖИРОВАНИЕ уцелевших по полноте данных и свежести квартала —
    внутри полосы величина ошибки на порядок не влияет, иначе показывался бы
    лучший хвост уже самой полосы.

    Фильтр стоит ЗДЕСЬ, а не в `build_row`, намеренно: строка вне полосы
    обязана остаться кандидатом и попасть в счётчик `eligible`. Отбраковав её
    раньше, мы получили бы «показано 20 из 20 годных» — счётчик, из которого
    отбор не виден вообще.

    В полосе меньше `limit` строк — возвращаем сколько есть. Добирать
    ближайшими по ошибке нельзя: это тот же отбор по величине ошибки, просто
    с другой стороны.
    """
    return sorted((r for r in rows if in_band(r.err_pct)), 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 если считать не из чего.

    Причины отказа ИСЧЕРПЫВАЮЩИЕ и все — «данных нет»: спайн не дал ожидаемой
    цены продажи; квартал сделки неизвестен; нет площади или цены сделки
    (делить не на что). Величина отклонения причиной отказа НЕ является ни при
    каких значениях: строка с любым промахом становится кандидатом и попадает
    в счётчик `eligible`. Полоса, по которой из кандидатов отбираются
    показанные, применяется позже и в другом месте — `select_rows`; здесь её
    нет намеренно, иначе отбор перестал бы быть виден в счётчиках.

    ФАКТ — ЭТО `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 у скольких показанных удалось определить район

    `eligible` минус `written` — это НЕ «столько не поместилось»: с 2026-09-12
    в разницу входят и строки вне полосы −5 %..+20 %. Поэтому подпись под
    таблицей называет `eligible` собранными строками, а чем отобраны
    показанные — говорит `REJECTION_RULE`, который едет тем же ответом.
    """
    # Импорт внутри функции: `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)
    # Сколько собранных строк вообще попало в полосу — в лог, а не в счётчики:
    # колонки под него в `landing_showcase_runs` нет, а без него по `written`
    # не отличить «полоса оставила мало» от «упёрлись в limit».
    logger.info(
        "в полосе %s: %d из %d собранных, показано %d",
        BAND_LABEL,
        sum(1 for r in candidates if in_band(r.err_pct)),
        len(candidates),
        len(chosen),
    )
    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())