fix(tradein): миграция платежей 277 → 279 (столкновение номеров) + lock_timeout

Два параллельных агента взяли ОДИН номер: 277_landing_showcase_runs.sql в ветке
витрины и 277_payments_live_checkout_uidx.sql здесь. Обе ветки по отдельности
зелёные, но на main второй файл встал бы конфликтом — ровно ловушка из шапки
tests/test_migration_numbering.py: номер сверяется с origin/main, а не с чужими
открытыми ветками.

Заняты сейчас: 275 (метрики), 276+277 (витрина), 278 (публичный токен) → этот 279.
Ссылки на номер обновлены в payments.py и test_payments_router.py, включая путь,
по которому тест читает предикат частичного UNIQUE.

Плюс CREATE UNIQUE INDEX на существующей таблице payments обёрнут в
SET LOCAL lock_timeout = '5s' — гейт #2752.

data/sql/194_grant_ekb_districts_fdw.sql

@@ -0,0 +1,26 @@
+-- 194_grant_ekb_districts_fdw.sql
+-- Витрина сделок МЕРЫ не могла показать район: FDW-чтение падало с
+--   permission denied for view ekb_districts_geom
+-- (foreign table tradein.gendesign_ekb_districts_geom → сервер gendesign_remote,
+-- роль tradein_fdw_reader).
+--
+-- Причина не в потере гранта, а в том, что его не выдавали НИКОГДА.
+-- 74_dedupe_unified_views.sql делает `DROP TABLE ekb_districts_geom;
+-- CREATE OR REPLACE VIEW ekb_districts_geom ...` — объект сменил тип с таблицы
+-- на вьюху и был создан заново, а строки GRANT рядом не появилось. Все четыре
+-- соседних объекта, которые tradein читает через тот же сервер
+-- (rosreestr_deals, mv_quarter_price_index, v_tradein_cad_buildings,
+-- v_tradein_osm_poi_ekb), грант имеют — этот остался единственным без него.
+--
+-- Тот же класс, что #2583 (188_regrant_quarter_price_index_fdw.sql): грант живёт
+-- отдельно от объекта и молча исчезает при пересоздании. Поэтому GRANT дописан
+-- ТАКЖЕ в конец 74-й — чтобы повторное применение той миграции его не теряло.
+--
+-- Признак починки: на tradein-стороне
+--   SELECT count(*) FROM gendesign_ekb_districts_geom;  -- было ERROR, стало 8
+
+BEGIN;
+
+GRANT SELECT ON public.ekb_districts_geom TO tradein_fdw_reader;
+
+COMMIT;

data/sql/74_dedupe_unified_views.sql

@@ -254,3 +254,19 @@ SELECT 'nspd', log_id, run_id, ts, level, stage, cad_number, message
 COMMENT ON VIEW v_scrape_log_unified IS
     'Per-step log двух скраперов. entity_id = obj_id для kn / cad_number для nspd. '
     'objective scraper пока не пишет log — добавим если понадобится debug.';
+
+-- ── ГРАНТ ЖИВЁТ ЗДЕСЬ, А НЕ ОТДЕЛЬНО (29.08.2026) ──────────────────────────
+-- Выше `DROP TABLE ekb_districts_geom` + `CREATE OR REPLACE VIEW` меняет тип
+-- объекта, то есть создаёт его заново — а вместе с объектом исчезают и его
+-- гранты. Из-за этого tradein-сторона (foreign table
+-- gendesign_ekb_districts_geom, роль tradein_fdw_reader) читала вьюху с
+-- `permission denied` и витрина сделок МЕРЫ теряла район.
+-- Тот же урок, что в 188_regrant_quarter_price_index_fdw.sql: грант, лежащий в
+-- отдельной миграции, переживает ровно до следующего пересоздания объекта.
+-- Держим его рядом с CREATE, чтобы объект и его права ехали вместе.
+--
+-- ЧЕСТНО О ДЕЙСТВИИ: на ДЕЙСТВУЮЩЕМ проде эта строка не выполнится никогда —
+-- деплой применяет только файлы, которых ещё нет в _schema_migrations, а 74-я
+-- давно записана. Работу делает 194_grant_ekb_districts_fdw.sql. Здесь строка
+-- нужна для чистой базы и для случая, когда файл прогоняют руками.
+GRANT SELECT ON public.ekb_districts_geom TO tradein_fdw_reader;

tradein-mvp/backend/app/api/public/mera.py

@@ -1,4 +1,4 @@
-"""Публичный API МЕРЫ (B2C, meraocenka.ru) — анонимный, ровно две ручки.
+"""Публичный API МЕРЫ (B2C, meraocenka.ru) — анонимный.
 
 ЗАЧЕМ ОТДЕЛЬНЫЙ ПРЕФИКС, А НЕ ОТКРЫТИЕ КУСКА /api/v1/*
 -------------------------------------------------------
@@ -26,20 +26,29 @@ API, нужно было выбрать одно из двух:
 АНОНИМНОСТЬ
 -----------
 `rbac_guard` (app/core/rbac.py) требует `X-Authenticated-User` для любого
-non-public пути. Обе ручки перечислены в `_PUBLIC_PATHS` ТОЧНЫМИ строками —
+non-public пути. Все ручки перечислены в `_PUBLIC_PATHS` ТОЧНЫМИ строками —
 не префиксом: множество там — frozenset с проверкой `path in ...`, и
-добавление префиксной ветки ради двух путей расширило бы механизм, которым
-пользуется весь бэкенд, ради одной фичи.
+добавление префиксной ветки расширило бы механизм, которым пользуется весь
+бэкенд, ради одной фичи. Поэтому и чтение по токену — POST с постоянным
+путём `/estimate/read`, а не `GET /estimate/{token}`: переменный сегмент
+пути потребовал бы ровно такой префиксной ветки (плюс сам токен уехал бы в
+access-лог Caddy, см. `PublicEstimateTokenInput`).
 
-ЧТО ЭТИ РУЧКИ НЕ ДЕЛАЮТ
------------------------
-Ни одна из них не пишет в БД строк с адресом пользователя: `/coverage` —
-чистое чтение (один SELECT), `/suggest` — прокси автокомплита. Это не
-случайность, а условие, при котором публичная форма может работать ДО того,
-как появится контур согласия 152-ФЗ (issue #2895: сегодня адрес физлица
-попадает в `trade_in_estimates` раньше любого согласия, а пути удаления
-данных в бэкенде нет). Платный расчёт, который писать будет, открывается
-отдельно и только вместе с этим контуром.
+ЧТО ЭТИ РУЧКИ ДЕЛАЮТ С ДАННЫМИ
+------------------------------
+`/suggest` и `/coverage` не пишут в БД ничего: первый — прокси автокомплита,
+второй — один SELECT. Это по-прежнему так и меняться не должно.
+
+`/estimate` — единственная, которая ПИШЕТ строку с адресом физлица (issue
+#2895), и поэтому устроена иначе: она закрыта флагом
+`settings.public_estimate_enabled` (дефолт false → 404) и требует явного
+согласия 152-ФЗ в payload'е строго `True` — на уровне схемы, то есть 422
+прилетает до входа в хендлер и до любого обращения к БД. Пути удаления
+данных в бэкенде всё ещё нет; включать флаг на проде — вместе с ним и с
+правкой п.5.5 политики обработки ПДн.
+
+`/estimate/read` возвращает по капабилити-токену ТОЛЬКО бесплатную часть
+(`PublicEstimateResult`) — цену и прогнозы продаёт платный контур.
 
 БЮДЖЕТЫ
 -------
@@ -58,19 +67,28 @@ non-public пути. Обе ручки перечислены в `_PUBLIC_PATHS`
 from __future__ import annotations
 
 import asyncio
+import hashlib
 import logging
-from typing import Annotated
+import secrets
+from datetime import datetime
+from typing import Annotated, Literal
 
-from fastapi import APIRouter, Depends, HTTPException, Request
+from fastapi import APIRouter, Depends, HTTPException, Request, Response
 from pydantic import BaseModel, Field
+from sqlalchemy import text
 from sqlalchemy.orm import Session
 
 from app.api.v1.geocode import SuggestResponse, suggest_addresses
-from app.api.v1.trade_in import coverage_probe
+from app.api.v1.trade_in import coverage_probe, estimate
+from app.core.config import settings
 from app.core.db import get_db
 from app.core.public_request import install_address_log_redaction, public_request_scope
 from app.core.ratelimit import SlidingWindowLimiter, _client_ip
-from app.schemas.trade_in import CoverageProbeInput, CoverageProbeResponse
+from app.schemas.trade_in import (
+    CoverageProbeInput,
+    CoverageProbeResponse,
+    TradeInEstimateInput,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -91,10 +109,26 @@ router = APIRouter()
 # нажал ещё раз».
 _SUGGEST_LIMIT = 20
 _COVERAGE_LIMIT = 15
+# Витрина — один SELECT по своей же маленькой таблице, внешних вызовов нет,
+# поэтому бюджет шире соседних: он здесь против перебора-в-цикле, а не против
+# денежных трат. Лэндинг дёргает ручку один раз на загрузку страницы.
+_SHOWCASE_LIMIT = 60
+# Расчёт — самый дорогой шаг публичного контура (геокодер + десяток SQL по
+# листингам + запись строки). Бюджет намеренно ниже пробы покрытия: живой
+# человек нажимает «рассчитать» единицы раз, а анонимная месячная квота
+# (settings.anon_estimate_quota_limit) обходится сменой IP — минутное окно
+# делает такой обход дорогим по времени.
+_ESTIMATE_LIMIT = 5
+# Чтение по токену дешевле расчёта (один SELECT по уникальному индексу), но
+# ослаблять его до бесконечности нельзя: перебор токенов — это перебор.
+_ESTIMATE_READ_LIMIT = 30
 _WINDOW_S = 60.0
 
 _suggest_limiter = SlidingWindowLimiter(limit=_SUGGEST_LIMIT, window_s=_WINDOW_S)
 _coverage_limiter = SlidingWindowLimiter(limit=_COVERAGE_LIMIT, window_s=_WINDOW_S)
+_showcase_limiter = SlidingWindowLimiter(limit=_SHOWCASE_LIMIT, window_s=_WINDOW_S)
+_estimate_limiter = SlidingWindowLimiter(limit=_ESTIMATE_LIMIT, window_s=_WINDOW_S)
+_estimate_read_limiter = SlidingWindowLimiter(limit=_ESTIMATE_READ_LIMIT, window_s=_WINDOW_S)
 
 # ── Общий суточный потолок публичных подсказок ──────────────────────────────
 #
@@ -286,3 +320,432 @@ def public_coverage(
     """
     _enforce(_coverage_limiter, request, "coverage")
     return coverage_probe(payload=payload, db=db)
+
+
+class LandingStat(BaseModel):
+    """Одна витринная величина лэндинга.
+
+    `sample_n` и `note` едут наружу вместе со значением намеренно: цифра без
+    размера выборки и без описания измеренного — это ровно тот литерал, который
+    лежал во фронте до появления landing_stats. Пусть фронт решает, показывать
+    ли их мелким шрифтом, но получить число БЕЗ них он не может.
+    """
+
+    value: float | str | None
+    sample_n: int | None
+    note: str | None
+    computed_at: datetime
+
+
+_STATS_LIMIT = 30
+_stats_limiter = SlidingWindowLimiter(limit=_STATS_LIMIT, window_s=_WINDOW_S)
+
+# Все строки витрины — их единицы, LIMIT не нужен, но потолок пусть будет:
+# таблица наполняется только ночной задачей, и если она когда-нибудь начнёт
+# писать метрику на город, ручка не должна молча вырасти в мегабайты.
+_STATS_SQL = text("""
+    SELECT metric, value_num, value_text, sample_n, note, computed_at
+    FROM landing_stats
+    ORDER BY metric
+    LIMIT 200
+""")
+
+
+@router.get("/stats", response_model=dict[str, LandingStat])
+def public_stats(
+    request: Request,
+    db: Annotated[Session, Depends(get_db)],
+) -> dict[str, LandingStat]:
+    """Витринные метрики лэндинга — готовый ночной срез (issue: числа по проду).
+
+    GET, в отличие от соседей: здесь в запросе нет ни адреса, ни чего-либо
+    относящегося к посетителю, поэтому довод «URI попадает в access-лог» не
+    работает, а кэшируемость GET'а для страницы, которую открывают все, полезна.
+
+    Читает готовые строки, НЕ считает на лету: агрегаты по offer_price_history с
+    подзапросами на листинг — секунды, а анонимная ручка, которая стоит секунду
+    CPU, это рычаг DoS. Считает их app/tasks/landing_stats.py раз в сутки.
+
+    Пустая таблица — валидные `{}` и 200. Это штатное состояние сразу после
+    накатки миграции (задача ещё не отработала) и оно же — состояние «данных для
+    метрики нет»: задача не пишет строку, когда мерить нечего. Фронт обязан это
+    пережить и не рисовать блок, а не получить 500 и сломанную страницу.
+
+    `value` — числовое value_num, если оно есть; иначе value_text (для метрик,
+    у которых значение не число). Оба NULL — отдаём null, а не выдуманный ноль.
+    """
+    _enforce(_stats_limiter, request, "stats")
+
+    rows = db.execute(_STATS_SQL).fetchall()
+    return {
+        row.metric: LandingStat(
+            value=(float(row.value_num) if row.value_num is not None else row.value_text),
+            sample_n=row.sample_n,
+            note=row.note,
+            computed_at=row.computed_at,
+        )
+        for row in rows
+    }
+
+
+class ShowcaseDeal(BaseModel):
+    """Одна строка витрины «МЕРА сказала X — продали за Y».
+
+    `district` / `floor` / `total_floors` НУЛЛАБЕЛЬНЫ намеренно: этих величин в
+    ДКП-данных может не быть, и фронт обязан пережить null, а не получить
+    правдоподобную подстановку. Улицы и дома в модели нет вовсе — номер дома
+    есть у 2.7% сделок (разбор в миграции 276).
+    """
+
+    district: str | None
+    rooms: int
+    area_m2: float
+    floor: int | None
+    total_floors: int | None
+    deal_quarter: str
+    predicted_rub: int
+    fact_rub: int
+    err_pct: float
+    n_analogs: int
+    note: str
+
+
+class ShowcaseStats(BaseModel):
+    """Итог прогона, который дал показанные строки. Подпись под витриной.
+
+    Без этих чисел «20 отличных строк» неотличимо от «столько и было»:
+    посетитель не может отличить выборку из работы оценщика от её лучшего
+    хвоста. `eligible` минус `written` — сколько годных строк не поместилось
+    в витрину; `rejection_rule` — по какому правилу отсеяно остальное,
+    записанное ТЕМ прогоном, который эти строки посчитал.
+    """
+
+    considered: int
+    priced: int
+    no_prediction: int
+    incomplete: int
+    eligible: int
+    written: int
+    with_district: int
+    rejection_rule: str
+
+
+class ShowcaseResponse(BaseModel):
+    """Витрина целиком: когда считали, что показываем и из чего это отобрано.
+
+    `stats` = None только до первого пересчёта — тогда и `deals` пуст.
+    """
+
+    computed_at: str | None
+    deals: list[ShowcaseDeal]
+    stats: ShowcaseStats | None = None
+
+
+# Последний прогон — единственная точка отсчёта: и `computed_at`, и счётчики, и
+# набор строк берутся ИЗ НЕГО. Брать строки по своему max(computed_at) значило
+# бы, что прогон, не давший ни одной строки, показывает вчерашние строки под
+# сегодняшними счётчиками.
+_SHOWCASE_RUN_SQL = text(
+    """
+    SELECT computed_at, considered, priced, no_prediction, incomplete,
+           eligible, written, with_district, rejection_rule
+    FROM landing_showcase_runs
+    ORDER BY computed_at DESC, id DESC
+    LIMIT 1
+    """
+)
+
+_SHOWCASE_SQL = text(
+    """
+    SELECT district, rooms, area_m2, floor, total_floors, deal_quarter,
+           predicted_rub, fact_rub, err_pct, n_analogs, note
+    FROM landing_showcase_deals
+    WHERE computed_at = CAST(:computed_at AS timestamptz)
+    ORDER BY id
+    """
+)
+
+
+@router.get("/showcase", response_model=ShowcaseResponse)
+def public_showcase(
+    request: Request,
+    db: Annotated[Session, Depends(get_db)],
+) -> ShowcaseResponse:
+    """Витрина лэндинга: реальные ДКП-сделки против прогноза МЕРЫ.
+
+    Читает готовый батч из `landing_showcase_deals` (пересчёт —
+    `app/tasks/landing_showcase_deals.py`), а не считает прогноз на лету:
+    один прогноз — это несколько пространственных SELECT'ов, двадцать штук на
+    анонимный GET были бы рычагом для DoS.
+
+    Пустой список — штатный ответ, а не ошибка: до первого пересчёта показывать
+    нечего, и это ровно то, что фронт должен увидеть вместо выдуманных строк.
+
+    Вместе со строками едет `stats` — сколько сделок рассмотрено, сколько
+    годных строк не поместилось и по какому правилу отсеяно остальное. Числа
+    считает пересчёт; без них витрина не имеет права подписаться честно.
+    """
+    _enforce(_showcase_limiter, request, "showcase")
+    run = db.execute(_SHOWCASE_RUN_SQL).mappings().first()
+    if run is None:
+        return ShowcaseResponse(computed_at=None, deals=[], stats=None)
+    rows = db.execute(_SHOWCASE_SQL, {"computed_at": run["computed_at"]}).mappings().all()
+    return ShowcaseResponse(
+        computed_at=run["computed_at"].isoformat(),
+        stats=ShowcaseStats(
+            considered=int(run["considered"]),
+            priced=int(run["priced"]),
+            no_prediction=int(run["no_prediction"]),
+            incomplete=int(run["incomplete"]),
+            eligible=int(run["eligible"]),
+            written=int(run["written"]),
+            with_district=int(run["with_district"]),
+            rejection_rule=run["rejection_rule"],
+        ),
+        deals=[
+            ShowcaseDeal(
+                district=r["district"],
+                rooms=int(r["rooms"]),
+                area_m2=float(r["area_m2"]),
+                floor=(int(r["floor"]) if r["floor"] is not None else None),
+                total_floors=(int(r["total_floors"]) if r["total_floors"] is not None else None),
+                deal_quarter=r["deal_quarter"],
+                predicted_rub=int(r["predicted_rub"]),
+                fact_rub=int(r["fact_rub"]),
+                err_pct=float(r["err_pct"]),
+                n_analogs=int(r["n_analogs"]),
+                note=r["note"],
+            )
+            for r in rows
+        ],
+    )
+
+
+# ── Анонимный расчёт и повторное чтение его бесплатной части ────────────────
+#
+# ФЛАГ. Обе ручки ниже мертвы, пока `settings.public_estimate_enabled` не
+# включён в .env.runtime: это первая пара публичных ручек, которая ПИШЕТ в
+# `trade_in_estimates` адрес физлица, а такое включение — продуктовое решение
+# владельца (нужна правка п.5.5 политики обработки ПДн, она сегодня анонимный
+# расчёт не описывает), а не следствие мержа кода. Тот же приём, что у
+# `payments_enabled`. Отвечаем 404, а не 403: выключенная ручка не должна
+# подтверждать, что она существует.
+#
+# ОБЩИЙ СУТОЧНЫЙ ПОТОЛОК. Per-IP окно ограничивает одного клиента, но не сумму:
+# 5/мин с адреса — это 7200 расчётов в сутки, каждый из которых дёргает
+# геокодер и пишет строку. Потолок ниже — про защиту закрытого контура и
+# квоты геокодера (тот же довод, что у `_DAILY_SUGGEST_BUDGET`), исчерпание —
+# сигнал абуза, не штатный режим.
+_DAILY_ESTIMATE_BUDGET = 300
+_daily_estimate_limiter = SlidingWindowLimiter(limit=_DAILY_ESTIMATE_BUDGET, window_s=86_400.0)
+_ESTIMATE_GLOBAL_KEY = "public-estimate"
+
+# Срок жизни капабилити-ссылки. Ссылка — единственный способ анонима вернуться
+# к своему расчёту (аккаунта у него нет), поэтому недели мало не будет для
+# «оплатил, закрыл вкладку, вернулся вечером», а бесконечной она быть не может:
+# это ссылка на данные о конкретной квартире конкретного человека.
+_TOKEN_TTL = "7 days"
+
+
+def _require_public_estimate_enabled() -> None:
+    """404, пока публичный расчёт не включён владельцем явно."""
+    if not settings.public_estimate_enabled:
+        raise HTTPException(status_code=404, detail="Not Found")
+
+
+def _token_hash(token: str) -> str:
+    """sha256(hex) — в БД лежит только это, сам токен не хранится нигде.
+
+    Без соли намеренно: вход — `secrets.token_urlsafe(32)` (256 бит), перебор
+    по словарю невозможен, а соль сделала бы невозможным поиск по равенству.
+    """
+    return hashlib.sha256(token.encode()).hexdigest()
+
+
+class PublicEstimateInput(TradeInEstimateInput):
+    """Вход анонимного расчёта: тот же payload, что у закрытого контура, но
+    согласие 152-ФЗ — ОБЯЗАТЕЛЬНОЕ и строго True.
+
+    В базовой схеме `consent: bool | None = None`: сделать его обязательным там
+    нельзя — B2B-пилоты согласия в UI не дают, у них договор, и их фронт поля
+    не шлёт. Здесь же анонимный посетитель — единственный источник согласия,
+    поэтому `Literal[True]`: без него Pydantic отвечает 422 ДО входа в
+    хендлер, то есть до первого обращения к БД. Это не дубль гейта в
+    `estimate_quality` (тот ловит любых вызывающих), а его сдвиг на самую
+    раннюю возможную границу — «согласие фиксируется до первого INSERT»
+    перестаёт зависеть от порядка строк внутри эстиматора.
+    """
+
+    consent: Literal[True]
+
+
+class PublicEstimateTokenInput(BaseModel):
+    """Токен едет ТЕЛОМ, а ручка чтения — POST, а не GET /{token}.
+
+    Причина ровно та же, по которой POST'ом сделан `/suggest`: access-лог Caddy
+    на публичном домене пишет URI целиком, поэтому капабилити-ссылка в пути
+    легла бы в файл рядом с IP посетителя — и любой, у кого есть доступ к
+    логам (или их бэкапу), открыл бы чужой расчёт. Токен — это пароль; пароли
+    в URL не кладут.
+    """
+
+    token: str = Field(min_length=16, max_length=128)
+
+
+class PublicEstimateResult(BaseModel):
+    """БЕСПЛАТНАЯ часть расчёта — ровно то, что можно показать до оплаты.
+
+    Здесь СОЗНАТЕЛЬНО нет ни одного поля из `AggregatedEstimate` с ценой,
+    прогнозом или списком аналогов (`median_price_rub`, `range_*`,
+    `expected_sold_*`, `analogs`, `actual_deals`, `market_percentile`,
+    `est_days_on_market`, `cian_valuation`, `avito_imv`, `dkp_corridor`…).
+    Модель отдельная, а не `AggregatedEstimate` с `exclude`: список исключений
+    надо помнить и дополнять при каждом новом поле эстиматора, а отдельная
+    модель молчит по умолчанию — новое платное поле не утечёт само.
+
+    `coverage` — тот же ответ, что у бесплатной пробы `/coverage` (в нём цен
+    нет по построению, см. `CoverageProbeResponse`): число похожих квартир,
+    возраст объявлений, вердикт покрытия. null — координаты дома не
+    разрезолвились, честнее отдать «неизвестно», чем правдоподобное число.
+    """
+
+    token: str
+    token_expires_at: datetime
+    n_analogs: int
+    coverage: CoverageProbeResponse | None = None
+
+
+def _coverage_for(
+    db: Session, lat: float | None, lon: float | None, rooms: int, area_m2: float
+) -> CoverageProbeResponse | None:
+    """Вердикт покрытия для уже посчитанной оценки — через ту же `coverage_probe`.
+
+    Своей копии порогов здесь нет намеренно: разъедься она с бесплатной пробой,
+    один и тот же адрес получил бы «есть данные» на одном экране и «мало» на
+    соседнем.
+    """
+    if lat is None or lon is None:
+        return None
+    return coverage_probe(
+        payload=CoverageProbeInput(lat=lat, lon=lon, rooms=rooms, area_m2=area_m2),
+        db=db,
+    )
+
+
+@router.post("/estimate", response_model=PublicEstimateResult)
+async def public_estimate(
+    request: Request,
+    response: Response,
+    payload: PublicEstimateInput,
+    db: Annotated[Session, Depends(get_db)],
+) -> PublicEstimateResult:
+    """Анонимный расчёт: считает полную оценку, отдаёт только бесплатную часть.
+
+    Делегирует в `app.api.v1.trade_in.estimate` — ту же функцию, что обслуживает
+    закрытый контур, а не копию её тела. Оттуда же бесплатно достаётся всё
+    анти-абузное хозяйство: анонимная квота на связку (подписанная cookie + IP)
+    с лимитом `settings.anon_estimate_quota_limit`, семафор одновременности,
+    503 вместо непрозрачного 502 при сбое и consent-гейт (`require_consent`
+    включается ровно потому, что мы зовём её без `X-Authenticated-User`).
+
+    Полный расчёт при этом СОХРАНЯЕТСЯ в `trade_in_estimates` целиком — платный
+    контур (сосед, `/api/v1/trade-in/r/{token}`) открывает его после оплаты по
+    своему токену. Наружу здесь уезжает только `PublicEstimateResult`.
+    """
+    _require_public_estimate_enabled()
+    _enforce(_estimate_limiter, request, "estimate")
+
+    daily_retry = _daily_estimate_limiter.retry_after(_ESTIMATE_GLOBAL_KEY)
+    if daily_retry is not None:
+        logger.error(
+            "публичные расчёты исчерпали суточный бюджет (%d) — закрытый контур "
+            "защищён, но форма на лэндинге сейчас не считает",
+            _DAILY_ESTIMATE_BUDGET,
+        )
+        raise HTTPException(
+            status_code=429,
+            detail="Расчёт временно недоступен. Попробуйте позже.",
+            headers={"Retry-After": str(int(daily_retry) + 1)},
+        )
+    _daily_estimate_limiter.record(_ESTIMATE_GLOBAL_KEY)
+
+    # Пометка публичного запроса — как у `/suggest`: внутри цепочки геокодер
+    # печатает введённый адрес, а публичная форма обещает обратное.
+    with public_request_scope():
+        result = await estimate(
+            payload=payload,
+            request=request,
+            response=response,
+            db=db,
+            x_authenticated_user=None,
+        )
+
+    # Токен выдаём ПОСЛЕ успешного расчёта: ссылка на несуществующий результат
+    # не нужна никому, а строка в БД уже есть — estimate() её закоммитил.
+    token = secrets.token_urlsafe(32)
+    row = db.execute(
+        text(
+            """
+            UPDATE trade_in_estimates
+               SET public_token_hash = :token_hash,
+                   public_token_expires_at = NOW() + CAST(:ttl AS interval)
+             WHERE id = CAST(:id AS uuid)
+         RETURNING public_token_expires_at
+            """
+        ),
+        {"token_hash": _token_hash(token), "ttl": _TOKEN_TTL, "id": str(result.estimate_id)},
+    ).fetchone()
+    db.commit()
+    if row is None:  # pragma: no cover — оценка только что записана этой же транзакцией
+        raise HTTPException(status_code=503, detail="estimate temporarily unavailable")
+
+    return PublicEstimateResult(
+        token=token,
+        token_expires_at=row.public_token_expires_at,
+        n_analogs=result.n_analogs,
+        coverage=_coverage_for(
+            db, result.target_lat, result.target_lon, payload.rooms, payload.area_m2
+        ),
+    )
+
+
+@router.post("/estimate/read", response_model=PublicEstimateResult)
+def public_estimate_read(
+    request: Request,
+    payload: PublicEstimateTokenInput,
+    db: Annotated[Session, Depends(get_db)],
+) -> PublicEstimateResult:
+    """Повторное чтение бесплатной части по капабилити-токену.
+
+    Существует потому, что у анонима нет аккаунта: без этой ручки результат
+    жил бы ровно в теле POST-ответа и не переживал бы перезагрузку страницы
+    (`_assert_estimate_access` в закрытом контуре отдаёт 404 на оценку с
+    `created_by IS NULL` всем, кроме админа — и это правильно, менять его
+    ради анонима значило бы ослабить IDOR-гейт для всех).
+
+    Просрочка и «нет такого токена» отвечают ОДИНАКОВО (404): различать их
+    значит подтверждать существование расчёта тому, кто угадал токен.
+    """
+    _require_public_estimate_enabled()
+    _enforce(_estimate_read_limiter, request, "estimate-read")
+
+    row = db.execute(
+        text(
+            """
+            SELECT n_analogs, lat, lon, rooms, area_m2, public_token_expires_at
+              FROM trade_in_estimates
+             WHERE public_token_hash = :token_hash
+               AND public_token_expires_at > NOW()
+            """
+        ),
+        {"token_hash": _token_hash(payload.token)},
+    ).fetchone()
+    if row is None:
+        raise HTTPException(status_code=404, detail="Расчёт не найден или ссылка устарела.")
+
+    return PublicEstimateResult(
+        token=payload.token,
+        token_expires_at=row.public_token_expires_at,
+        n_analogs=row.n_analogs,
+        coverage=_coverage_for(db, row.lat, row.lon, row.rooms, row.area_m2),
+    )

tradein-mvp/backend/app/core/config.py

@@ -1226,5 +1226,13 @@ class Settings(BaseSettings):
     # обязаны отказывать сразу, ничего не вызывая у T-Bank.
     payments_enabled: bool = Field(default=False, validation_alias="PAYMENTS_ENABLED")
 
+    # Kill-switch анонимного расчёта на публичном домене (POST/GET
+    # /api/public/mera/estimate*). false — обе ручки отвечают 404, как будто их
+    # нет: включение публичного расчёта открывает запись адреса физлица в
+    # trade_in_estimates, и решение это продуктовое (нужна правка п.5.5 политики
+    # обработки ПДн), а не «смержили код». Тот же приём и та же причина, что у
+    # payments_enabled выше. ENV: PUBLIC_ESTIMATE_ENABLED.
+    public_estimate_enabled: bool = Field(default=False, validation_alias="PUBLIC_ESTIMATE_ENABLED")
+
 
 settings = Settings()

tradein-mvp/backend/app/core/rbac.py

@@ -114,6 +114,21 @@ _PUBLIC_PATHS = frozenset(
         # держится на структуре пакета app/api/public/, а не на матчере.
         "/api/public/mera/suggest",
         "/api/public/mera/coverage",
+        # Витринные числа лэндинга (landing_stats, миграция 275): агрегаты по
+        # проду без единой персональной строки — их и показывают анонимному
+        # посетителю, ради чего метрики и считаются.
+        "/api/public/mera/stats",
+        # Витрина реальных ДКП-сделок против прогноза (миграция 276): читает
+        # СВОЮ таблицу-витрину, где по построению нет ни адреса, ни владельца —
+        # район + характеристики квартиры + пара «прогноз/факт».
+        "/api/public/mera/showcase",
+        # Анонимный расчёт и повторное чтение его бесплатной части. Оба —
+        # POST с точным путём: у чтения токен едет ТЕЛОМ, а не в URI, иначе
+        # капабилити-ссылка легла бы в access-лог Caddy рядом с IP посетителя
+        # (тот же довод, что у /suggest — см. app/api/public/mera.py).
+        # Обе ручки дополнительно закрыты флагом settings.public_estimate_enabled.
+        "/api/public/mera/estimate",
+        "/api/public/mera/estimate/read",
         # Вебхук Т-Банка (app/api/v1/payments.py::notify): сервер-к-серверу,
         # X-Authenticated-User/сессии у банка нет и быть не может. Путь не
         # секрет — аутентификацией здесь работает подпись `Token` тела,

tradein-mvp/backend/app/services/product_handlers.py

@@ -308,6 +308,16 @@ async def _job_deals_freshness_monitor(
     await loop.run_in_executor(None, check_deals_freshness, db, run_id, params)
 
 
+# ── landing_stats_refresh — sync DB-only пересчёт витрины в executor ─────────
+async def _job_landing_stats(
+    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
+) -> None:
+    from app.tasks.landing_stats import refresh_landing_stats
+
+    loop = asyncio.get_event_loop()
+    await loop.run_in_executor(None, refresh_landing_stats, db, run_id, params)
+
+
 # ── sber_freshness_monitor — sync DB-only freshness check в executor ──────────
 async def _job_sber_freshness_monitor(
     db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
@@ -657,6 +667,7 @@ def build_product_handlers(ctx: SchedulerContext) -> dict[str, Handler]:
         "rosreestr_quarter_poll": Handler(_job_rosreestr_quarter_poll, "rosreestr_quarter_poll"),
         "deals_freshness_monitor": Handler(_job_deals_freshness_monitor, "deals_freshness_monitor"),
         "sber_freshness_monitor": Handler(_job_sber_freshness_monitor, "sber_freshness_monitor"),
+        "landing_stats_refresh": Handler(_job_landing_stats, "landing_stats_refresh"),
         "newbuilding_enrich": Handler(_job_newbuilding_enrich, "newbuilding_enrich"),
         "yandex_newbuilding_sweep": Handler(
             _job_yandex_newbuilding_sweep, "yandex_newbuilding_sweep"

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

@@ -0,0 +1,407 @@
+"""Пересчёт витрины лэндинга на РЕАЛЬНЫХ сделках (миграция 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())

tradein-mvp/backend/app/tasks/landing_stats.py

@@ -0,0 +1,397 @@
+"""Пересчёт витринных метрик публичного лэндинга МЕРЫ (таблица landing_stats).
+
+ЗАЧЕМ
+-----
+Числа на лэндинге (frontend/src/app/mera-public/marketing-v3.ts) были литералами
+— то есть придуманными. Публичная страница, которая продаёт «расчёт по данным»,
+не может показывать цифры, которых в данных нет: это ровно та подмена, против
+которой продукт и позиционируется. Здесь каждая витринная величина считается
+запросом к проду, и вместе с ней пишется размер выборки.
+
+ГЛАВНОЕ ПРАВИЛО: НЕТ ВХОДА — НЕТ СТРОКИ
+---------------------------------------
+Ни одна метрика не пишется с подставленным значением. Если выборка пуста
+(нет оценок, нет истории цен, нет сделок) — строка в landing_stats просто не
+появляется, ручка её не отдаёт, фронт не рисует блок. Ноль здесь читался бы как
+измеренный ноль («ни одно объявление не снижало цену»), а это враньё другого
+рода, чем отсутствие данных. Правило действует и на ВТОРОМ прогоне: пропавшая
+метрика удаляется из таблицы (см. refresh_landing_stats), иначе она осталась бы
+на витрине со старым computed_at и читалась бы как измеренная сегодня.
+
+ЧЕГО ЗДЕСЬ НЕТ И НЕ БУДЕТ
+-------------------------
+«Точность прогноза» и «срок продажи» — величин с такими именами в базе нет.
+Точность считает бэктест (своя задача, свои допущения), а срок продажи требует
+пары «объявление снято → сделка», которой у нас нет: снятие объявления не
+означает продажу. `listing_age_median_days` НЕ является сроком продажи и назван
+экспозицией активного объявления — см. note метрики.
+
+ПОЧЕМУ ТОЛЬКО DOMKLIK В ЦЕНОВЫХ МЕТРИКАХ
+----------------------------------------
+`offer_price_history` наполняется триггером, и наполняется по-разному:
+у avito/yandex стартовая цена в историю НЕ пишется (первая строка появляется
+только при изменении, то есть «снизил» и «не снижал» неразличимы), а yandex
+вдобавок сеет синтетическую пару со сдвигом в сутки. Считать долю снижений по
+такой смеси — значит получить число, у которого нет смысла. Domklik пишет старт,
+поэтому только он.
+
+Задача синхронная (только SELECT'ы + UPSERT), запускается kit-scheduler'ом через
+product_handlers._job_landing_stats в run_in_executor — по образцу
+deals_freshness_monitor / listing_source_snapshot.
+"""
+
+from __future__ import annotations
+
+import logging
+from decimal import Decimal
+from typing import Any
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from app.services import scrape_runs as runs_mod
+
+logger = logging.getLogger(__name__)
+
+__all__ = ["EKB", "collect_landing_metrics", "refresh_landing_stats"]
+
+# Город витрины. Лэндинг сегодня продаёт Екатеринбург, и метрики обязаны быть
+# про него же: медиана по всей области смешала бы рынки с разной динамикой.
+EKB = "Екатеринбург"
+
+# Порог наблюдения для ценовых метрик. За две недели объявление успевает получить
+# первую правку цены; более короткие живут слишком мало, чтобы «не снижал» было
+# наблюдением, а не «не успел».
+_PRICE_SPAN_DAYS = 14
+# Отсечка аномалий: изменение больше 30% за наблюдение — это, как правило, смена
+# объекта под тем же id (перевыставили другую квартиру) или опечатка в цене,
+# а не торг. Медиану такие хвосты не двигают, но долю снижений — двигают.
+_PRICE_MAX_ABS_PCT = 30
+
+# ── Оценки ──────────────────────────────────────────────────────────────────
+# Период считаем по фактическим краям created_at, а не «с даты запуска»: витрина
+# обещает «за N дней работы», и N должен быть измеренным.
+_ESTIMATES_SQL = text("""
+    SELECT count(*)                                              AS total,
+           EXTRACT(EPOCH FROM (max(created_at) - min(created_at)))
+               / 86400.0                                         AS period_days
+    FROM trade_in_estimates
+""")
+
+# n_analogs > 0: оценка без аналогов — это отказ расчёта, а не «ноль аналогов»;
+# включив её, мы бы занизили медиану наблюдениями, где измерять было нечего.
+_ANALOGS_SQL = text("""
+    SELECT count(*)                                            AS n,
+           percentile_cont(0.5) WITHIN GROUP (ORDER BY n_analogs) AS median
+    FROM trade_in_estimates
+    WHERE n_analogs > 0
+""")
+
+# Возраст АКТИВНОГО объявления = экспозиция на сегодня, а не срок продажи:
+# знаменатель — те, кто ещё висит, поэтому величина по построению занижена
+# относительно «сколько в итоге продавалось». Это ограничение уезжает в note.
+_LISTING_AGE_SQL = text("""
+    SELECT count(*)                                     AS n,
+           percentile_cont(0.5) WITHIN GROUP (
+               ORDER BY (CURRENT_DATE - listing_date)
+           )                                            AS median
+    FROM listings
+    WHERE is_active
+      AND city = CAST(:city AS text)
+      AND listing_date IS NOT NULL
+      AND listing_date <= CURRENT_DATE
+""")
+
+# ── Динамика цены объявлений ────────────────────────────────────────────────
+#
+# Знаменатель — объявления, которые МОЖНО было наблюдать: от первой записи в
+# истории до последнего показа прошло >= 14 дней. Сюда попадают и те, у кого
+# запись одна (domklik пишет старт → одна запись означает «цену не менял»); без
+# них доля снижений считалась бы только по менявшим и давала 85% вместо 48%.
+#
+# Скорость снижения нормируем на 30 дней по интервалу МЕЖДУ КРАЙНИМИ ПРАВКАМИ,
+# а не по всему наблюдению: цена не менялась после последней правки, и растягивая
+# знаменатель на «висит до сих пор», мы измеряли бы терпение продавца, а не торг.
+_PRICE_MOVES_SQL = text("""
+    WITH hist AS (
+        SELECT listing_id,
+               min(change_time) AS first_change,
+               max(change_time) AS last_change,
+               count(*)         AS n_rows
+        FROM offer_price_history
+        WHERE source = 'domklik'
+        GROUP BY listing_id
+    ),
+    observed AS (
+        SELECT h.listing_id,
+               h.n_rows,
+               EXTRACT(EPOCH FROM (h.last_change - h.first_change)) / 86400.0 AS change_days
+        FROM hist h
+        JOIN listings l ON l.id = h.listing_id
+        WHERE GREATEST(h.last_change, COALESCE(l.last_seen_at, h.last_change)) - h.first_change
+              >= make_interval(days => CAST(:span_days AS integer))
+    ),
+    priced AS (
+        SELECT o.listing_id,
+               o.n_rows,
+               o.change_days,
+               (SELECT p.price_rub FROM offer_price_history p
+                 WHERE p.listing_id = o.listing_id AND p.source = 'domklik'
+                 ORDER BY p.change_time ASC, p.id ASC LIMIT 1)  AS price_first,
+               (SELECT p.price_rub FROM offer_price_history p
+                 WHERE p.listing_id = o.listing_id AND p.source = 'domklik'
+                 ORDER BY p.change_time DESC, p.id DESC LIMIT 1) AS price_last
+        FROM observed o
+    ),
+    moved AS (
+        SELECT listing_id,
+               change_days,
+               CASE WHEN n_rows >= 2
+                    THEN (price_last - price_first) / price_first * 100.0
+                    ELSE 0
+               END AS pct
+        FROM priced
+        WHERE price_first IS NOT NULL AND price_first > 0
+    )
+    SELECT count(*)                                       AS n,
+           count(*) FILTER (WHERE pct < 0)                AS n_cut,
+           percentile_cont(0.5) WITHIN GROUP (
+               ORDER BY pct * 30.0 / NULLIF(change_days, 0)
+           ) FILTER (WHERE pct < 0)                       AS median_pct_per_month
+    FROM moved
+    WHERE abs(pct) <= CAST(:max_abs_pct AS numeric)
+""")
+
+# 12 месяцев от сегодня. deal_date у Росреестра — лейбл начала квартала, поэтому
+# окно накрывает 4-5 кварталов и число «за год» тут приблизительно по построению;
+# это сказано в note, а не спрятано.
+_DEALS_SQL = text("""
+    SELECT count(*) AS n
+    FROM deals
+    WHERE city = CAST(:city AS text)
+      AND deal_date >= (CURRENT_DATE - INTERVAL '12 months')
+""")
+
+_UPSERT_SQL = text("""
+    INSERT INTO landing_stats (metric, value_num, value_text, sample_n, note, computed_at)
+    VALUES (
+        CAST(:metric AS text),
+        CAST(:value_num AS numeric),
+        CAST(:value_text AS text),
+        CAST(:sample_n AS integer),
+        CAST(:note AS text),
+        now()
+    )
+    ON CONFLICT (metric) DO UPDATE SET
+        value_num   = EXCLUDED.value_num,
+        value_text  = EXCLUDED.value_text,
+        sample_n    = EXCLUDED.sample_n,
+        note        = EXCLUDED.note,
+        computed_at = EXCLUDED.computed_at
+""")
+
+# Строки метрик, которых в СЕГОДНЯШНЕМ наборе нет, удаляются. Метрика исчезает
+# из набора ровно тогда, когда у неё пропал вход (см. «нет входа — нет строки»),
+# и оставленная строка продолжала бы отдаваться ручкой как обычная — со старым
+# computed_at, который витрина не обязана читать. Удалённая метрика — блок,
+# которого на странице нет; протухшая — блок с враньём.
+_PRUNE_SQL = text("""
+    DELETE FROM landing_stats
+    WHERE metric <> ALL(CAST(:kept AS text[]))
+""")
+
+
+def _num(value: Any) -> float | None:
+    """Привести значение агрегата к float; None остаётся None.
+
+    percentile_cont возвращает Decimal/float в зависимости от типа входа — в
+    numeric-колонку и в JSON поедет одинаково только после явного приведения.
+    """
+    if value is None:
+        return None
+    if isinstance(value, Decimal):
+        return float(value)
+    return float(value)
+
+
+def collect_landing_metrics(db: Session) -> list[dict[str, Any]]:
+    """Посчитать метрики витрины. Метрика без данных в список НЕ попадает.
+
+    Отделено от записи, чтобы тест мог проверить сами ЗНАЧЕНИЯ на подготовленной
+    базе, не разбирая по дороге счётчики прогона.
+    """
+    metrics: list[dict[str, Any]] = []
+
+    row = db.execute(_ESTIMATES_SQL).first()
+    total = int(row.total) if row is not None and row.total else 0
+    if total > 0:
+        metrics.append(
+            {
+                "metric": "estimates_total",
+                "value_num": float(total),
+                "value_text": None,
+                "sample_n": total,
+                "note": "Расчётов сделано в системе (все города, весь срок работы)",
+            }
+        )
+        period = _num(row.period_days)
+        # Один-единственный расчёт даёт период 0 дней — это не измерение, а
+        # артефакт единственной точки; такую строку не пишем.
+        if period is not None and total > 1:
+            metrics.append(
+                {
+                    "metric": "estimates_period_days",
+                    "value_num": round(period, 1),
+                    "value_text": None,
+                    "sample_n": total,
+                    "note": "Дней между первым и последним расчётом",
+                }
+            )
+
+    row = db.execute(_ANALOGS_SQL).first()
+    if row is not None and row.n and _num(row.median) is not None:
+        metrics.append(
+            {
+                "metric": "analogs_median",
+                "value_num": round(_num(row.median) or 0.0, 1),
+                "value_text": None,
+                "sample_n": int(row.n),
+                "note": "Медиана числа аналогов на расчёт (только расчёты, где аналоги нашлись)",
+            }
+        )
+
+    row = db.execute(_LISTING_AGE_SQL, {"city": EKB}).first()
+    if row is not None and row.n and _num(row.median) is not None:
+        metrics.append(
+            {
+                "metric": "listing_age_median_days",
+                "value_num": round(_num(row.median) or 0.0, 1),
+                "value_text": None,
+                "sample_n": int(row.n),
+                "note": (
+                    "Медианная ЭКСПОЗИЦИЯ активного объявления в Екатеринбурге "
+                    "(сколько дней висит на сегодня). Это НЕ срок продажи: "
+                    "считается по тем, кто ещё продаётся, и снятие объявления "
+                    "не означает сделку"
+                ),
+            }
+        )
+
+    row = db.execute(
+        _PRICE_MOVES_SQL,
+        {"span_days": _PRICE_SPAN_DAYS, "max_abs_pct": _PRICE_MAX_ABS_PCT},
+    ).first()
+    if row is not None and row.n:
+        n = int(row.n)
+        base_note = (
+            f"Только Домклик (единственный источник, где триггер пишет стартовую цену), "
+            f"наблюдение от {_PRICE_SPAN_DAYS} дней, изменения свыше "
+            f"{_PRICE_MAX_ABS_PCT}% отброшены как смена объекта"
+        )
+        metrics.append(
+            {
+                "metric": "price_cut_share_pct",
+                "value_num": round(int(row.n_cut) * 100.0 / n, 1),
+                "value_text": None,
+                "sample_n": n,
+                "note": f"Доля объявлений, снижавших цену. {base_note}",
+            }
+        )
+        median_move = _num(row.median_pct_per_month)
+        if median_move is not None:
+            metrics.append(
+                {
+                    "metric": "price_cut_median_pct_per_month",
+                    "value_num": round(median_move, 2),
+                    "value_text": None,
+                    # Выборка ЗДЕСЬ — только снижавшие: медиана считается по ним,
+                    # и подставить сюда общий n значило бы приписать величине
+                    # выборку, по которой её не считали.
+                    "sample_n": int(row.n_cut),
+                    "note": (
+                        f"Медианное изменение цены за 30 дней среди снижавших "
+                        f"(отрицательное). {base_note}"
+                    ),
+                }
+            )
+
+    row = db.execute(_DEALS_SQL, {"city": EKB}).first()
+    if row is not None and row.n:
+        metrics.append(
+            {
+                "metric": "deals_total_12m",
+                "value_num": float(row.n),
+                "value_text": None,
+                "sample_n": int(row.n),
+                "note": (
+                    "Сделок Росреестра по Екатеринбургу за последние 12 месяцев. "
+                    "Дата сделки — лейбл начала квартала, поэтому окно накрывает "
+                    "целые кварталы, а не ровно год"
+                ),
+            }
+        )
+
+    return metrics
+
+
+def refresh_landing_stats(
+    db: Session,
+    run_id: int,
+    params: dict[str, Any] | None = None,
+) -> dict[str, int]:
+    """Пересчитать landing_stats и финализировать прогон.
+
+    Sync (вызывается scheduler-триггером в executor, как check_deals_freshness).
+    `params` не используется — принимается ради единой сигнатуры обработчиков.
+
+    Метрика, у которой пропал вход, СНИМАЕТСЯ с витрины, а не доживает со старым
+    computed_at: строки, которых нет в сегодняшнем наборе, удаляются в той же
+    транзакции. Иначе «нет входа — нет строки» действует только на первом
+    прогоне, а дальше отсутствие данных выглядит как данные — ручка отдаёт такую
+    строку неотличимо от свежей, и отличить её можно только сравнив computed_at с
+    соседями, чего фронт не делает.
+
+    Пустой результат — НЕ ошибка прогона: на свежей базе метрик может не быть ни
+    одной, и падать в failed из-за этого значит завести шумный алерт там, где
+    система работает штатно. Но и чистка в этом случае НЕ выполняется: разом
+    отвалившиеся все входы — это признак поломки самого прогона (пустая/недоступная
+    база), а не пяти одновременных «данных больше нет», и стирать по такому
+    признаку всю витрину нельзя. Чистка ходит только с непустым набором, где
+    пропажу конкретной метрики видно на фоне посчитавшихся соседей.
+    """
+    del params
+    counters: dict[str, int] = {"metrics_written": 0, "metrics_removed": 0}
+    try:
+        runs_mod.update_heartbeat(db, run_id, counters)
+
+        metrics = collect_landing_metrics(db)
+        for row in metrics:
+            db.execute(_UPSERT_SQL, row)
+        if metrics:
+            removed = db.execute(_PRUNE_SQL, {"kept": [m["metric"] for m in metrics]})
+            counters["metrics_removed"] = int(removed.rowcount or 0)
+        db.commit()
+
+        counters["metrics_written"] = len(metrics)
+        if not metrics:
+            logger.warning(
+                "landing_stats run_id=%d: ни одной метрики не посчиталось — "
+                "витрина покажет прошлый срез (или пусто, если его не было)",
+                run_id,
+            )
+        runs_mod.mark_done(db, run_id, counters)
+        logger.info(
+            "refresh_landing_stats run_id=%d done: %d метрик (%s)",
+            run_id,
+            len(metrics),
+            ", ".join(m["metric"] for m in metrics) or "—",
+        )
+        return counters
+    except Exception as exc:
+        logger.exception("refresh_landing_stats run_id=%d failed", run_id)
+        try:
+            db.rollback()
+        except Exception:
+            pass
+        runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters)
+        raise

tradein-mvp/backend/data/sql/275_landing_stats.sql

@@ -0,0 +1,90 @@
+-- 275_landing_stats.sql
+-- Витринные метрики публичного лэндинга МЕРЫ — считаются по проду, не пишутся руками.
+--
+-- ЗАЧЕМ ТАБЛИЦА, А НЕ ЗАПРОС ИЗ РУЧКИ
+-- -----------------------------------
+-- Числа на лэндинге сегодня лежат литералами во фронте
+-- (frontend/src/app/mera-public/marketing-v3.ts) — то есть выдуманы и не имеют
+-- срока годности: когда база меняется, страница врёт молча. Но и считать их в
+-- момент запроса нельзя: медиана по offer_price_history с подзапросами на
+-- листинг — это секунды на анонимной ручке без авторизации, то есть готовый
+-- рычаг для DoS. Поэтому срез считает ночная задача
+-- (app/tasks/landing_stats.py), а ручка отдаёт готовые строки.
+--
+-- ОДНА СТРОКА НА МЕТРИКУ, ИСТОРИИ НЕТ
+-- -----------------------------------
+-- PK (metric) + UPSERT: лэндингу нужно «сколько сейчас», а не тренд. Заводить
+-- историю впрок значит выбрать схему под запрос, которого никто не задавал;
+-- когда понадобится динамика — она приедет отдельной таблицей со своим PK,
+-- и это будет дешевле, чем сейчас угадывать её ключ.
+--
+-- value_num И value_text РАЗДЕЛЬНО
+-- -------------------------------
+-- Числовые метрики фронт форматирует сам (округление, склонение, разделители
+-- разрядов), поэтому числу нельзя приезжать строкой. value_text оставлен для
+-- метрик, у которых значение — не число (например период «май–август 2026»);
+-- сегодня такие не пишутся, но колонка дешевле, чем миграция под первую же.
+--
+-- sample_n ОБЯЗАТЕЛЕН ПО СМЫСЛУ, NULL ПО СХЕМЕ
+-- -------------------------------------------
+-- Требование продукта: у каждой витринной цифры видно, по скольким наблюдениям
+-- она получена — иначе «медиана» неотличима от «медиана по двум объявлениям».
+-- Гарантирует это задача (она НЕ пишет строку, если входа нет), а не NOT NULL:
+-- жёсткое ограничение на колонке заставило бы будущую метрику без выборки
+-- подставлять фиктивный ноль, то есть врать ради схемы.
+--
+-- Идемпотентно: IF NOT EXISTS — безопасно переприменять.
+
+BEGIN;
+
+CREATE TABLE IF NOT EXISTS landing_stats (
+    metric       text        PRIMARY KEY,
+    value_num    numeric,
+    value_text   text,
+    sample_n     integer,
+    note         text,
+    computed_at  timestamptz NOT NULL DEFAULT now()
+);
+
+COMMENT ON TABLE landing_stats IS
+    'Витринные метрики лэндинга МЕРЫ; пересчёт — app/tasks/landing_stats.py (раз в сутки)';
+COMMENT ON COLUMN landing_stats.sample_n IS
+    'Размер выборки, по которой получено значение — показывается рядом с цифрой';
+COMMENT ON COLUMN landing_stats.note IS
+    'Что именно измерено, человеческим языком — защита от подмены смысла на витрине';
+
+-- ── Регистрация в планировщике ──────────────────────────────────────────────
+--
+-- Периодические задачи МЕРЫ живут не в crontab, а строками scrape_schedules:
+-- kit-scheduler (app/scheduler_main.py) выбирает source по окну и резолвит
+-- обработчик через app/services/product_handlers.py. Поэтому «регистрация»
+-- задачи — это ровно две вещи: Handler в реестре и вот эта строка.
+--
+-- Окно 05:00–06:00 UTC: после ночных лоадеров листингов и после
+-- asking_to_sold_ratio_refresh (06:00–07:00 UTC мы бы догоняли), но до
+-- рабочего дня по Екатеринбургу (UTC+5) — витрина к утру уже пересчитана.
+-- Задача читающая (несколько агрегирующих SELECT, внешних вызовов нет), так
+-- что enabled=true сразу: цена ошибки — минуты CPU ночью.
+--
+-- next_run_at на завтра: не выстреливает прямо в момент деплоя (образец —
+-- 162_seed_deals_freshness_monitor.sql).
+INSERT INTO scrape_schedules (
+    source,
+    enabled,
+    window_start_hour,
+    window_end_hour,
+    next_run_at,
+    default_params
+)
+VALUES
+(
+    'landing_stats_refresh',
+    true,
+    5,
+    6,
+    ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 5)) AT TIME ZONE 'UTC',
+    '{}'::jsonb
+)
+ON CONFLICT (source) DO NOTHING;
+
+COMMIT;

tradein-mvp/backend/data/sql/276_landing_showcase_deals.sql

@@ -0,0 +1,50 @@
+-- 276: витрина лэндинга на РЕАЛЬНЫХ сделках (issue B2C-showcase).
+--
+-- ЗАЧЕМ ТАБЛИЦА, А НЕ ВЫЧИСЛЕНИЕ В РУЧКЕ. Прогноз считается полным спайном
+-- оценщика: несколько пространственных SELECT'ов на КАЖДУЮ сделку. Двадцать
+-- сделок — это сотни запросов; на публичной ручке без авторизации это готовый
+-- рычаг для DoS. Поэтому пересчёт — офлайн-задача (app/tasks/landing_showcase_deals.py),
+-- ручка читает готовые строки.
+--
+-- ЧЕГО ЗДЕСЬ НАМЕРЕННО НЕТ — АДРЕСА. В `deals` номер дома есть у 2.7% строк
+-- (620 различных адресов на 24 644 сделки), то есть «улица + дом» на витрине
+-- была бы додумана. Показываем район + характеристики квартиры; улицы нет
+-- даже колонкой, чтобы её нельзя было «на минутку» вывести.
+--
+-- deal_quarter — ТЕКСТ КВАРТАЛА, не дата: `deals.deal_date` принимает всего 10
+-- различных значений на всю таблицу (первое число квартала), то есть дня
+-- сделки в данных нет. Хранить date здесь значило бы отдать фронту точность,
+-- которой не существует.
+--
+-- district и floor — NULLABLE. Район резолвится через FDW-вьюху чужой базы
+-- (gendesign_ekb_districts_geom), и её гранты уже терялись (см. C3); floor в
+-- части ДКП-строк пуст. Правило проекта: нет величины — пишем NULL, а не
+-- правдоподобное значение. Отбор в задаче ранжирует такие строки ниже, но не
+-- запрещает их: пустая витрина хуже витрины без района.
+BEGIN;
+-- Конвенция проекта (#2752): блокирующий DDL идёт под lock_timeout, иначе он
+-- встанет в очередь за чужой сессией и утащит за собой запросы приложения.
+SET LOCAL lock_timeout = '5s';
+
+CREATE TABLE IF NOT EXISTS landing_showcase_deals (
+    id            bigserial PRIMARY KEY,
+    computed_at   timestamptz NOT NULL DEFAULT now(),
+    district      text,
+    rooms         integer     NOT NULL,
+    area_m2       numeric(8, 2) NOT NULL,
+    floor         integer,
+    total_floors  integer,
+    deal_quarter  text        NOT NULL,
+    predicted_rub bigint      NOT NULL,
+    fact_rub      bigint      NOT NULL,
+    err_pct       numeric(6, 2) NOT NULL,
+    n_analogs     integer     NOT NULL,
+    note          text        NOT NULL
+);
+
+-- Ручка всегда читает ПОСЛЕДНИЙ пересчёт (max computed_at) — старые батчи
+-- остаются для сверки «что показывали неделю назад».
+CREATE INDEX IF NOT EXISTS idx_landing_showcase_deals_computed_at
+    ON landing_showcase_deals (computed_at DESC);
+
+COMMIT;

tradein-mvp/backend/data/sql/277_landing_showcase_runs.sql

@@ -0,0 +1,44 @@
+-- 277: итог пересчёта витрины лэндинга — счётчики рядом со строками.
+--
+-- ЗАЧЕМ ОТДЕЛЬНАЯ ТАБЛИЦА, А НЕ КОЛОНКИ В landing_showcase_deals. Счётчики —
+-- факт ПРОГОНА, а не строки: на 20 строк они были бы продублированы 20 раз, а
+-- в самом важном случае — когда показывать оказалось нечего — исчезли бы
+-- вместе со строками. «Рассмотрено 200, показывать нечего» обязано доезжать до
+-- фронта ровно так же, как «рассмотрено 200, показано 20».
+--
+-- ЗАЧЕМ ВООБЩЕ. Витрина показывает 20 сделок «МЕРА сказала X — продали за Y».
+-- Без чисел рядом эти 20 строк читаются как «столько и было»: посетитель не
+-- может отличить выборку из работы оценщика от её лучшего хвоста. Поэтому
+-- ручка /api/public/mera/showcase отдаёт вместе со строками, сколько сделок
+-- рассмотрено, сколько годных строк не поместилось и по какому правилу
+-- отбраковано остальное.
+--
+-- rejection_rule — ТЕКСТ, А НЕ КОД ПРАВИЛА. Правило живёт в
+-- app/tasks/landing_showcase_deals.REJECTION_RULE и пишется сюда тем прогоном,
+-- который эти числа и посчитал: подпись под витриной обязана описывать ТОТ
+-- отбор, что дал эти строки, а не тот, что задеплоен сегодня.
+--
+-- computed_at совпадает со строками батча: задача пишет строки и этот итог
+-- ОДНОЙ транзакцией, а now() в Postgres — время начала транзакции.
+BEGIN;
+-- Конвенция проекта (#2752): блокирующий DDL идёт под lock_timeout, иначе он
+-- встанет в очередь за чужой сессией и утащит за собой запросы приложения.
+SET LOCAL lock_timeout = '5s';
+
+CREATE TABLE IF NOT EXISTS landing_showcase_runs (
+    id             bigserial PRIMARY KEY,
+    computed_at    timestamptz NOT NULL DEFAULT now(),
+    considered     integer NOT NULL,
+    priced         integer NOT NULL,
+    no_prediction  integer NOT NULL,
+    incomplete     integer NOT NULL,
+    eligible       integer NOT NULL,
+    written        integer NOT NULL,
+    with_district  integer NOT NULL,
+    rejection_rule text    NOT NULL
+);
+
+CREATE INDEX IF NOT EXISTS idx_landing_showcase_runs_computed_at
+    ON landing_showcase_runs (computed_at DESC);
+
+COMMIT;

tradein-mvp/backend/data/sql/278_trade_in_estimates_public_token.sql

@@ -0,0 +1,45 @@
+-- 278_trade_in_estimates_public_token.sql
+-- Purpose: capability-ссылка на БЕСПЛАТНУЮ часть анонимного расчёта
+--          (GET /api/public/mera/estimate/{token}).
+--
+-- ЗАЧЕМ КОЛОНКА, А НЕ ОТДЕЛЬНАЯ ТАБЛИЦА. Токен — атрибут ровно одной оценки и
+-- живёт/умирает вместе с ней; таблица 1:1 добавила бы join и вторую точку,
+-- где строку можно забыть удалить.
+--
+-- ХРАНИМ ХЭШ, А НЕ ТОКЕН. Дамп/бэкап/случайный SELECT в поддержке не должны
+-- давать доступ к чужому расчёту. sha256 без соли осознанно: вход —
+-- secrets.token_urlsafe(32), 256 бит энтропии, словарь по нему невозможен,
+-- а соль сломала бы поиск по равенству (пришлось бы сканировать таблицу).
+--
+-- ПОЧЕМУ ОТДЕЛЬНЫЙ СРОК, А НЕ expires_at/retain_until. expires_at — про то,
+-- сколько живёт сам расчёт, retain_until двигает контур оплаты. Ссылка на
+-- бесплатную часть — третье, независимое обещание («ссылка работает N дней»),
+-- и склеивать его с чужими сроками значит менять его молча при каждой правке
+-- соседей.
+--
+-- Dependencies: trade_in_estimates (миграция 001+).
+-- Deploy order: применять после 277.
+BEGIN;
+-- Конвенция проекта (#2752): ALTER TABLE на ЖИВОЙ таблице берёт ACCESS
+-- EXCLUSIVE, и без lock_timeout встанет в очередь за запросами приложения,
+-- утащив их за собой. trade_in_estimates — таблица боевая (1123 строки на
+-- 29.08), поэтому это не формальность.
+SET LOCAL lock_timeout = '5s';
+
+ALTER TABLE trade_in_estimates
+    ADD COLUMN IF NOT EXISTS public_token_hash text,
+    ADD COLUMN IF NOT EXISTS public_token_expires_at timestamptz;
+
+-- UNIQUE, а не просто индекс: коллизия хэшей означала бы, что по одной ссылке
+-- отдаются два разных расчёта. Partial — токен есть у меньшинства строк
+-- (B2B-пилоты его не получают вовсе), индексировать NULL'ы незачем.
+CREATE UNIQUE INDEX IF NOT EXISTS idx_trade_in_estimates_public_token_hash
+    ON trade_in_estimates (public_token_hash)
+    WHERE public_token_hash IS NOT NULL;
+
+COMMENT ON COLUMN trade_in_estimates.public_token_hash IS
+    'sha256(hex) капабилити-токена бесплатной части (app/api/public/mera.py). Сам токен не хранится.';
+COMMENT ON COLUMN trade_in_estimates.public_token_expires_at IS
+    'До какого момента работает GET /api/public/mera/estimate/{token}. Не связан с expires_at/retain_until.';
+
+COMMIT;

tradein-mvp/backend/tests/test_landing_showcase_deals.py

@@ -0,0 +1,171 @@
+"""Витрина лэндинга на реальных сделках — отбор и отбраковка (миграция 276).
+
+Главное, что здесь защищается, — НЕ формат строки, а свойство отбора: витрина
+показывает выборку из работы оценщика, а не её лучший хвост. Отбор по малой
+ошибке дал бы формально работающий код и врущий продукт, и заметить это на
+глаз в проде нельзя — числа будут красивые. Поэтому проверка двусторонняя:
+самая точная строка, у которой не хватает данных, обязана проиграть менее
+точной, но полной.
+"""
+
+from __future__ import annotations
+
+import os
+from datetime import date
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from app.tasks.landing_showcase_deals import (
+    ShowcaseRow,
+    build_row,
+    quarter_label,
+    select_rows,
+)
+
+
+def _row(
+    deal_id: int,
+    *,
+    district: str | None = "Кировский",
+    floor: int | None = 5,
+    total_floors: int | None = 9,
+    deal_date: date = date(2026, 1, 1),
+    err_pct: float = 10.0,
+) -> ShowcaseRow:
+    return ShowcaseRow(
+        deal_id=deal_id,
+        district=district,
+        rooms=2,
+        area_m2=54.0,
+        floor=floor,
+        total_floors=total_floors,
+        deal_date=deal_date,
+        deal_quarter="I квартал 2026",
+        predicted_rub=6_000_000,
+        fact_rub=5_500_000,
+        err_pct=err_pct,
+        n_analogs=40,
+    )
+
+
+def test_selection_ignores_error_magnitude() -> None:
+    """Точнейшая строка с дырами в данных НЕ должна оказаться впереди полной.
+
+    Ломать так: добавить в `_sort_key` слагаемое `abs(row.err_pct)` — тест
+    покраснеет с id 1 на первом месте вместо id 2.
+    """
+    almost_perfect_but_thin = _row(1, district=None, floor=None, err_pct=0.1)
+    complete_but_worse = _row(2, err_pct=27.0)
+
+    chosen = select_rows([almost_perfect_but_thin, complete_but_worse], limit=1)
+
+    assert [r.deal_id for r in chosen] == [2], (
+        "отбор поехал за величиной ошибки — витрина перестала быть выборкой "
+        "и стала рекламой лучшего хвоста"
+    )
+
+
+def test_selection_prefers_fresher_quarter_at_equal_completeness() -> None:
+    older = _row(1, deal_date=date(2025, 1, 1), err_pct=1.0)
+    fresher = _row(2, deal_date=date(2026, 4, 1), err_pct=35.0)
+
+    assert [r.deal_id for r in select_rows([older, fresher], limit=1)] == [2]
+
+
+def test_selection_is_deterministic_on_full_ties() -> None:
+    """Полные совпадения ключа разводятся id — иначе витрина «мерцает»."""
+    rows = [_row(7), _row(9), _row(8)]
+    assert [r.deal_id for r in select_rows(rows, limit=3)] == [9, 8, 7]
+
+
+# ── Отбраковка: только «данных нет», никогда «число некрасивое» ──────────────
+
+
+def _build(**over: object) -> ShowcaseRow | None:
+    kwargs: dict[str, object] = {
+        "deal_id": 1,
+        "district": "Кировский",
+        "rooms": 2,
+        "area_m2": 50.0,
+        "floor": 5,
+        "total_floors": 9,
+        "deal_date": date(2026, 4, 1),
+        "predicted_rub": 5_000_000.0,
+        "fact_ppm2": 100_000.0,  # → факт 5 000 000 ₽, ошибка 0%
+        "n_analogs": 30,
+    }
+    kwargs.update(over)
+    return build_row(**kwargs)  # type: ignore[arg-type]
+
+
+def test_plain_row_survives_and_carries_signed_error() -> None:
+    row = _build(predicted_rub=5_500_000.0)
+    assert row is not None
+    assert row.fact_rub == 5_000_000
+    assert row.err_pct == 10.0, "знак и база ошибки: (прогноз − факт) / факт"
+    assert row.deal_quarter == "II квартал 2026"
+
+
+def test_no_error_magnitude_is_ever_rejected() -> None:
+    """Промах оценщика ЛЮБОГО размера остаётся на витрине.
+
+    Это второй половина запрета «не отбирать по ошибке»: фильтр по величине
+    ошибки — тот же отбор, просто ступенькой раньше, и он тем злее, что не
+    оставляет строку даже в кандидатах.
+
+    Ломать так: вернуть в `build_row` любой порог вида
+    `if abs(err_pct) > X: return None` — тест покраснеет на первом же
+    отклонении больше X с этим отклонением в сообщении.
+    """
+    fact_rub = 5_000_000.0  # 100 000 ₽/м² × 50 м²
+    for err_pct in (-95.0, -60.0, -41.0, -5.0, 0.0, 5.0, 41.0, 150.0, 900.0):
+        row = _build(predicted_rub=fact_rub * (1 + err_pct / 100))
+        assert row is not None, (
+            f"строка с отклонением {err_pct:+.0f}% выброшена: витрина снова "
+            "показывает лучший хвост, а не работу оценщика"
+        )
+        assert row.err_pct == round(err_pct, 2)
+
+
+def test_underdeclared_dkp_is_shown_not_hidden() -> None:
+    """Занижение ДКП ради налога выглядит как промах — и всё равно показывается.
+
+    Прятать такие строки нельзя: «отклонение больше 40% — это почти всегда
+    дефект ДКП» было догадкой, а санитарный диапазон ₽/м² уже применён к
+    выборке выше по потоку (`_load_sample`, для ЕКБ 30k..600k). Всё, что
+    прошло его и дало большую ошибку, — работа оценщика. Честность за счёт
+    строки в `note`, а не за счёт отсева.
+    """
+    # Факт 2 000 000 ₽ против прогноза 5 000 000 — отклонение +150%.
+    row = _build(fact_ppm2=40_000.0)
+    assert row is not None
+    assert row.err_pct == 150.0
+
+
+def test_ppm2_band_is_not_duplicated_here() -> None:
+    """Своей копии ₽/м²-диапазона в `build_row` нет — она была мёртвой.
+
+    `MIN_FACT_PPM2 = 30k` дублировал уже применённый фильтр выборки, а
+    `MAX_FACT_PPM2 = 1.2M` был недостижим при её потолке 600k: из трёх
+    отбраковок срабатывала ровно одна — по ошибке. Ломать так: вернуть любую
+    из границ — покраснеет соответствующая половина.
+    """
+    assert _build(fact_ppm2=20_000.0, predicted_rub=1_000_000.0) is not None
+    assert _build(fact_ppm2=2_000_000.0, predicted_rub=100_000_000.0) is not None
+
+
+def test_missing_fact_price_is_rejected() -> None:
+    """Нулевая цена сделки — это «данных нет», а не «число некрасивое»: делить не на что."""
+    assert _build(fact_ppm2=0.0) is None
+    assert _build(area_m2=0.0) is None
+
+
+def test_no_expected_sold_price_is_not_invented() -> None:
+    """Спайн не дал ожидаемой цены продажи — строки нет. Подставлять нечего."""
+    assert _build(predicted_rub=None) is None
+
+
+def test_unknown_quarter_is_not_invented() -> None:
+    assert _build(deal_date=None) is None
+    assert quarter_label(None) is None
+    assert quarter_label(date(2026, 7, 1)) == "III квартал 2026"

tradein-mvp/backend/tests/test_landing_stats.py

@@ -0,0 +1,470 @@
+"""Витринные метрики лэндинга — задача пересчёта + публичная ручка.
+
+ЧТО ЗДЕСЬ ПРОВЕРЯЕТСЯ И ПОЧЕМУ ИМЕННО ЭТО
+
+  1. ЗНАЧЕНИЯ. Арифметика витрины (доля снижавших, нормировка на 30 дней,
+     округления, какой sample_n к какой метрике) живёт в Python, и она проверена
+     по ЧИСЛАМ: подставляем агрегаты и сверяем ровно то, что уедет на страницу.
+     Тест обязан краснеть, если share посчитать от не того знаменателя или
+     приписать медиане общий n вместо числа снижавших.
+
+  2. НЕТ ВХОДА — НЕТ СТРОКИ. Отдельная проверка на каждую пустую выборку:
+     подстановка правдоподобного нуля — главный способ соврать на витрине, и
+     запрещена она поведением задачи, а не комментарием.
+
+  3. ГРАНИЦЫ ВЫБОРКИ В SQL. Условия «только domklik», «наблюдение >= 14 дней»,
+     «|изменение| <= 30%» на mock-сессии не проявляются: их исполняет Postgres.
+     Поэтому они запинены статически по тексту запроса — иначе их молчаливое
+     исчезновение (а с ним и мусор от yandex-синтетики) прошло бы незамеченным.
+
+  4. РУЧКА. Публичность (rbac), форма ответа, и главное — пустая таблица даёт
+     200 и {}, а не 500: это штатное состояние сразу после накатки миграции.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+import sys
+from datetime import UTC, datetime
+from decimal import Decimal
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any
+from unittest.mock import MagicMock
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+_wp_mock = MagicMock()
+sys.modules.setdefault("weasyprint", _wp_mock)
+sys.modules.setdefault("weasyprint.CSS", _wp_mock)
+sys.modules.setdefault("weasyprint.HTML", _wp_mock)
+
+import pytest  # noqa: E402
+from fastapi import FastAPI  # noqa: E402
+from fastapi.testclient import TestClient  # noqa: E402
+
+from app.api.public import mera as public_mera  # noqa: E402
+from app.core.db import get_db  # noqa: E402
+from app.core.rbac import _PUBLIC_PATHS, rbac_guard  # noqa: E402
+from app.tasks import landing_stats as ls  # noqa: E402
+
+_SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql"
+_MIGRATION_275 = _SQL_DIR / "275_landing_stats.sql"
+
+PREFIX = "/api/public/mera"
+
+
+# ── Мок-сессия: отдаёт заранее заданную строку на каждый из запросов задачи ───
+#
+# Раскладываем ответы по ПОРЯДКУ вызовов, а не по тексту SQL: порядок — часть
+# контракта collect_landing_metrics (он же порядок метрик на витрине), и его
+# перестановка должна быть заметна.
+class _FakeSession:
+    def __init__(self, rows: list[Any], *, prune_rowcount: int = 0) -> None:
+        self._rows = list(rows)
+        self.upserts: list[dict[str, Any]] = []
+        # Чистка протухших метрик: пишем сюда параметры каждого DELETE, чтобы
+        # тест видел И факт вызова, И список оставляемых метрик.
+        self.prunes: list[dict[str, Any] | None] = []
+        self._prune_rowcount = prune_rowcount
+        self.committed = 0
+
+    # Считываем ТОЛЬКО запросы самой витрины: по этой же сессии ходит
+    # runs_mod (heartbeat/mark_done пишут в scrape_runs), и если раздавать
+    # заготовленные строки по любому execute, первый же heartbeat съест
+    # агрегат оценок — тест краснел бы не по своей причине.
+    _METRIC_SQL_MARKERS = (
+        "FROM trade_in_estimates",
+        "FROM listings",
+        "offer_price_history",
+        "FROM deals",
+    )
+
+    def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> Any:
+        sql = str(stmt)
+        if "INSERT INTO landing_stats" in sql:
+            assert params is not None
+            self.upserts.append(params)
+            return MagicMock()
+        if "DELETE FROM landing_stats" in sql:
+            self.prunes.append(params)
+            return SimpleNamespace(rowcount=self._prune_rowcount)
+        if not any(marker in sql for marker in self._METRIC_SQL_MARKERS):
+            return MagicMock()
+        assert self._rows, f"неожиданный лишний SELECT: {sql[:80]}"
+        row = self._rows.pop(0)
+        return SimpleNamespace(first=lambda: row)
+
+    def commit(self) -> None:
+        self.committed += 1
+
+    def rollback(self) -> None:  # pragma: no cover — путь ошибки здесь не гоняется
+        pass
+
+
+def _rows(**overrides: Any) -> list[Any]:
+    """Пять агрегатов в порядке вызова. Значения — прод-срез на 29.08.2026."""
+    base: dict[str, Any] = {
+        "estimates": SimpleNamespace(total=1123, period_days=94.0),
+        "analogs": SimpleNamespace(n=975, median=Decimal("12")),
+        "listing_age": SimpleNamespace(n=25943, median=Decimal("26")),
+        "price": SimpleNamespace(n=6276, n_cut=3018, median_pct_per_month=Decimal("-2.174")),
+        "deals": SimpleNamespace(n=18657),
+    }
+    base.update(overrides)
+    return [base["estimates"], base["analogs"], base["listing_age"], base["price"], base["deals"]]
+
+
+def _by_metric(metrics: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
+    return {m["metric"]: m for m in metrics}
+
+
+# ── 1. Значения ──────────────────────────────────────────────────────────────
+
+
+def test_all_metrics_computed_from_aggregates() -> None:
+    """Каждая витринная цифра — ровно то, что следует из выборки."""
+    got = _by_metric(ls.collect_landing_metrics(_FakeSession(_rows())))
+
+    assert got["estimates_total"]["value_num"] == 1123.0
+    assert got["estimates_total"]["sample_n"] == 1123
+    assert got["estimates_period_days"]["value_num"] == 94.0
+    assert got["analogs_median"]["value_num"] == 12.0
+    assert got["analogs_median"]["sample_n"] == 975
+    assert got["listing_age_median_days"]["value_num"] == 26.0
+    assert got["deals_total_12m"]["value_num"] == 18657.0
+
+    # 3018/6276 = 48.087...% → 48.1 после округления до десятых.
+    assert got["price_cut_share_pct"]["value_num"] == 48.1
+    assert got["price_cut_share_pct"]["sample_n"] == 6276
+    assert got["price_cut_median_pct_per_month"]["value_num"] == -2.17
+    # Медиана считается ТОЛЬКО по снижавшим — и выборка у неё их, а не общая.
+    assert got["price_cut_median_pct_per_month"]["sample_n"] == 3018
+
+
+def test_share_uses_full_observed_denominator_not_only_cutters() -> None:
+    """Знаменатель доли — все наблюдавшиеся, а не только снижавшие.
+
+    Если считать от снижавших, доля всегда 100% — ровно тот дефект, который на
+    проде давал 85% вместо 48% (в выборку попадали только менявшие цену).
+    """
+    rows = _rows(price=SimpleNamespace(n=200, n_cut=50, median_pct_per_month=Decimal("-3")))
+    got = _by_metric(ls.collect_landing_metrics(_FakeSession(rows)))
+    assert got["price_cut_share_pct"]["value_num"] == 25.0
+
+
+def test_every_metric_carries_sample_n() -> None:
+    """Цифра без размера выборки неотличима от литерала, ради замены которого
+    вся эта таблица и заведена."""
+    for m in ls.collect_landing_metrics(_FakeSession(_rows())):
+        assert isinstance(m["sample_n"], int) and m["sample_n"] > 0, m["metric"]
+        assert m["note"], m["metric"]
+
+
+def test_listing_age_note_says_exposure_not_time_to_sell() -> None:
+    """Величина по построению — экспозиция ЕЩЁ ВИСЯЩЕГО объявления. Названная
+    «сроком продажи», она врёт (и врёт в выгодную сторону)."""
+    got = _by_metric(ls.collect_landing_metrics(_FakeSession(_rows())))
+    note = got["listing_age_median_days"]["note"]
+    assert "ЭКСПОЗИЦИЯ" in note
+    assert "НЕ срок продажи" in note
+
+
+def test_forecast_accuracy_and_time_to_sell_are_never_produced() -> None:
+    """Этих величин в данных нет; их считает бэктест со своими допущениями."""
+    names = {m["metric"] for m in ls.collect_landing_metrics(_FakeSession(_rows()))}
+    assert not {n for n in names if "accuracy" in n or "time_to_sell" in n or "days_to_sell" in n}
+
+
+# ── 2. Нет входа — нет строки ────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize(
+    ("kwargs", "absent"),
+    [
+        ({"estimates": SimpleNamespace(total=0, period_days=None)}, "estimates_total"),
+        ({"analogs": SimpleNamespace(n=0, median=None)}, "analogs_median"),
+        ({"listing_age": SimpleNamespace(n=0, median=None)}, "listing_age_median_days"),
+        (
+            {"price": SimpleNamespace(n=0, n_cut=0, median_pct_per_month=None)},
+            "price_cut_share_pct",
+        ),
+        ({"deals": SimpleNamespace(n=0)}, "deals_total_12m"),
+    ],
+)
+def test_empty_input_writes_no_row_instead_of_zero(kwargs: dict[str, Any], absent: str) -> None:
+    """Ноль читается как измеренный ноль («никто не снижал цену») — а измерения
+    не было. Строки просто нет, фронт не рисует блок."""
+    names = {m["metric"] for m in ls.collect_landing_metrics(_FakeSession(_rows(**kwargs)))}
+    assert absent not in names
+
+
+def test_single_estimate_gives_no_period_metric() -> None:
+    """Период между первым и последним расчётом при одном расчёте — 0 дней,
+    что является артефактом единственной точки, а не сроком работы."""
+    rows = _rows(estimates=SimpleNamespace(total=1, period_days=0.0))
+    names = {m["metric"] for m in ls.collect_landing_metrics(_FakeSession(rows))}
+    assert "estimates_total" in names
+    assert "estimates_period_days" not in names
+
+
+def test_no_cutters_leaves_share_but_drops_median() -> None:
+    """Никто не снижал — доля 0% ИЗМЕРЕНА (наблюдения были), а медианы снижения
+    не существует: писать её нулём значило бы выдумать «снижают на 0%»."""
+    rows = _rows(price=SimpleNamespace(n=120, n_cut=0, median_pct_per_month=None))
+    got = _by_metric(ls.collect_landing_metrics(_FakeSession(rows)))
+    assert got["price_cut_share_pct"]["value_num"] == 0.0
+    assert "price_cut_median_pct_per_month" not in got
+
+
+def test_refresh_upserts_every_metric_and_commits() -> None:
+    db = _FakeSession(_rows())
+    counters = ls.refresh_landing_stats(db, run_id=1)  # type: ignore[arg-type]
+    assert counters["metrics_written"] == len(db.upserts) == 7
+    # >=1, а не ==1: runs_mod коммитит свои heartbeat/mark_done по той же сессии.
+    assert db.committed >= 1
+    assert {u["metric"] for u in db.upserts} == {
+        "estimates_total",
+        "estimates_period_days",
+        "analogs_median",
+        "listing_age_median_days",
+        "price_cut_share_pct",
+        "price_cut_median_pct_per_month",
+        "deals_total_12m",
+    }
+
+
+def test_metric_that_stopped_computing_is_deleted_not_left_stale() -> None:
+    """Пропал вход у метрики — строка УДАЛЯЕТСЯ, а не доживает со старым
+    computed_at: иначе ручка отдаёт её неотличимо от посчитанной сегодня.
+
+    Здесь сделок нет (`deals.n = 0`), значит `deals_total_12m` в наборе не
+    появляется — и именно её обязан вынести DELETE, оставив ровно посчитанные.
+    """
+    rows = _rows(deals=SimpleNamespace(n=0))
+    db = _FakeSession(rows, prune_rowcount=1)
+    counters = ls.refresh_landing_stats(db, run_id=1)  # type: ignore[arg-type]
+
+    assert len(db.prunes) == 1, "чистка протухших метрик не выполнена"
+    kept = set(db.prunes[0]["kept"])  # type: ignore[index]
+    assert kept == {u["metric"] for u in db.upserts}
+    assert "deals_total_12m" not in kept, "метрика без входа осталась бы на витрине"
+    assert counters["metrics_removed"] == 1
+
+
+def test_totally_empty_run_keeps_the_showcase_instead_of_wiping_it() -> None:
+    """Разом пропали ВСЕ входы — это похоже на поломку прогона (пустая или
+    недоступная база), а не на пять одновременных «данных больше нет». По такому
+    признаку витрина не стирается: DELETE не выполняется вовсе."""
+    empty = _rows(
+        estimates=SimpleNamespace(total=0, period_days=None),
+        analogs=SimpleNamespace(n=0, median=None),
+        listing_age=SimpleNamespace(n=0, median=None),
+        price=SimpleNamespace(n=0, n_cut=0, median_pct_per_month=None),
+        deals=SimpleNamespace(n=0),
+    )
+    db = _FakeSession(empty)
+    counters = ls.refresh_landing_stats(db, run_id=1)  # type: ignore[arg-type]
+
+    assert db.upserts == []
+    assert db.prunes == [], "пустой прогон стёр бы всю витрину"
+    assert counters["metrics_written"] == 0
+    assert counters["metrics_removed"] == 0
+
+
+# ── 3. Границы выборки, которые исполняет Postgres ───────────────────────────
+
+
+def test_price_sql_takes_domklik_only() -> None:
+    """avito/yandex сюда попасть не могут: у первого нет стартовой цены в
+    истории, второй сеет синтетическую пару со сдвигом в сутки."""
+    sql = str(ls._PRICE_MOVES_SQL)
+    assert "source = 'domklik'" in sql
+    assert "avito" not in sql and "yandex" not in sql
+
+
+def test_price_sql_keeps_single_row_listings_in_denominator() -> None:
+    """Знаменатель доли снижений включает объявления с ОДНОЙ записью истории.
+
+    Это тот самый дефект, из-за которого на проде получалось бы 84.8% вместо
+    48.1%: у domklik триггер пишет стартовую цену, поэтому одна запись означает
+    «цену не менял» — наблюдение, а не отсутствие данных. Выкинув такие строки,
+    считаешь долю снижавших ТОЛЬКО среди менявших цену, то есть почти единицу.
+
+    Гейт текстовый, а не прогон на живой базе: DATABASE_URL в CI —
+    заглушка (deploy-tradein.yml: `test:` job), Postgres в тестовой джобе нет,
+    и живой тест по образцу test_purge_expired_trade_in_data.py тут молча
+    скипался бы — то есть не гейтил бы ничего. Пин проверяет две половины
+    дефекта: (1) однострочные попадают в `moved` через ветку CASE со значением
+    0 («не снижал»), а не отбрасываются; (2) нигде в запросе нет фильтра по
+    числу записей, который бы их отсёк.
+    """
+    sql = str(ls._PRICE_MOVES_SQL)
+
+    case = re.search(r"CASE\b(?P<body>.*?)\bEND\b", sql, re.S | re.I)
+    assert case is not None, "исчезла ветка для однострочных — они больше не «не снижал»"
+    body = case.group("body")
+    assert "n_rows" in body, "ветка перестала различать однострочные записи истории"
+    assert re.search(r"\b(THEN|ELSE)\s+0\b", body), (
+        "однострочным объявлениям больше не приписывается изменение 0% — "
+        "они либо выпали из выборки, либо получили выдуманное значение"
+    )
+
+    rest = sql.replace(case.group(0), "")
+    leftover = re.search(r"n_rows\s*(>=|>|<|<>|=|!=)", rest)
+    assert leftover is None, (
+        f"появился фильтр по числу записей истории вне ветки CASE ({leftover.group(0)!r}) — "
+        "он выкидывает не менявших цену из знаменателя, доля вырастет с ~48% до ~85%"
+    )
+    assert not re.search(r"\bHAVING\b", rest, re.I), (
+        "HAVING в агрегате истории отсекает однострочные ещё до знаменателя"
+    )
+
+
+def test_price_sql_keeps_span_and_outlier_gates() -> None:
+    sql = str(ls._PRICE_MOVES_SQL)
+    assert "span_days" in sql, "исчез порог наблюдения — короткоживущие дадут ложное «не снижал»"
+    assert "max_abs_pct" in sql, "исчезла отсечка аномалий — перевыставленные объекты как торг"
+    assert ls._PRICE_SPAN_DAYS == 14
+    assert ls._PRICE_MAX_ABS_PCT == 30
+
+
+def test_city_scoped_metrics_are_parameterised_by_ekb() -> None:
+    for sql in (str(ls._LISTING_AGE_SQL), str(ls._DEALS_SQL)):
+        assert "CAST(:city AS text)" in sql
+    assert ls.EKB == "Екатеринбург"
+
+
+def test_analogs_median_excludes_estimates_without_analogs() -> None:
+    """n_analogs=0 — это отказ расчёта, а не «ноль аналогов»; в медиане он
+    занизил бы величину наблюдением, где мерить было нечего."""
+    assert "n_analogs > 0" in str(ls._ANALOGS_SQL)
+
+
+# ── Миграция ─────────────────────────────────────────────────────────────────
+
+
+def test_migration_275_is_idempotent_and_registers_the_job() -> None:
+    sql = _MIGRATION_275.read_text("utf-8")
+    assert "CREATE TABLE IF NOT EXISTS landing_stats" in sql
+    assert "metric       text        PRIMARY KEY" in sql
+    assert "ON CONFLICT (source) DO NOTHING" in sql
+    assert "'landing_stats_refresh'" in sql
+
+
+def test_migration_275_has_no_psycopg_cast_trap() -> None:
+    """`:x::type` psycopg v3 разбирает как именованный параметр — в проекте
+    разрешён только CAST(:x AS type)."""
+    assert not re.search(r":\w+::", _MIGRATION_275.read_text("utf-8"))
+
+
+def test_task_is_registered_in_the_scheduler_registry() -> None:
+    """Без Handler'а строка расписания резолвится в никуда и джоба не бежит."""
+    from app.services.product_handlers import build_product_handlers
+
+    handlers = build_product_handlers(MagicMock())
+    assert "landing_stats_refresh" in handlers
+
+
+# ── 4. Публичная ручка ───────────────────────────────────────────────────────
+
+
+_STAT_ROWS = [
+    SimpleNamespace(
+        metric="estimates_total",
+        value_num=Decimal("1123"),
+        value_text=None,
+        sample_n=1123,
+        note="Расчётов сделано",
+        computed_at=datetime(2026, 8, 29, 5, 0, tzinfo=UTC),
+    ),
+    SimpleNamespace(
+        metric="price_cut_share_pct",
+        value_num=Decimal("48.1"),
+        value_text=None,
+        sample_n=6276,
+        note="Только Домклик",
+        computed_at=datetime(2026, 8, 29, 5, 0, tzinfo=UTC),
+    ),
+]
+
+
+def _client(rows: list[Any]) -> TestClient:
+    """Приложение с РЕАЛЬНЫМ rbac_guard — тем же, что вешает app/main.py."""
+    app = FastAPI()
+    app.middleware("http")(rbac_guard)
+    app.include_router(public_mera.router, prefix=PREFIX)
+
+    db = MagicMock()
+    db.execute.return_value.fetchall.return_value = rows
+
+    def _override_db():
+        yield db
+
+    app.dependency_overrides[get_db] = _override_db
+    return TestClient(app)
+
+
+@pytest.fixture(autouse=True)
+def _reset_stats_limiter():
+    public_mera._stats_limiter._hits.clear()
+    yield
+    public_mera._stats_limiter._hits.clear()
+
+
+def test_stats_path_is_public_in_rbac() -> None:
+    assert f"{PREFIX}/stats" in _PUBLIC_PATHS, (
+        "без строки в rbac._PUBLIC_PATHS анониму прилетит 401 и лэндинг останется без чисел"
+    )
+
+
+def test_anonymous_gets_stats_keyed_by_metric() -> None:
+    resp = _client(_STAT_ROWS).get(f"{PREFIX}/stats")
+    assert resp.status_code == 200
+    body = resp.json()
+    assert set(body) == {"estimates_total", "price_cut_share_pct"}
+    assert body["estimates_total"]["value"] == 1123.0
+    assert body["price_cut_share_pct"]["value"] == 48.1
+    assert body["price_cut_share_pct"]["sample_n"] == 6276
+    assert body["price_cut_share_pct"]["note"] == "Только Домклик"
+    assert body["estimates_total"]["computed_at"].startswith("2026-08-29T05:00")
+
+
+def test_empty_table_is_a_valid_answer_not_an_error() -> None:
+    """Состояние сразу после накатки миграции: задача ещё не отрабатывала.
+    500 здесь сломал бы страницу целиком ради отсутствующего блока."""
+    resp = _client([]).get(f"{PREFIX}/stats")
+    assert resp.status_code == 200
+    assert resp.json() == {}
+
+
+def test_metric_without_numeric_value_falls_back_to_text_then_null() -> None:
+    rows = [
+        SimpleNamespace(
+            metric="period_label",
+            value_num=None,
+            value_text="май–август 2026",
+            sample_n=1123,
+            note=None,
+            computed_at=datetime(2026, 8, 29, tzinfo=UTC),
+        ),
+        SimpleNamespace(
+            metric="nothing_measured",
+            value_num=None,
+            value_text=None,
+            sample_n=None,
+            note=None,
+            computed_at=datetime(2026, 8, 29, tzinfo=UTC),
+        ),
+    ]
+    body = _client(rows).get(f"{PREFIX}/stats").json()
+    assert body["period_label"]["value"] == "май–август 2026"
+    assert body["nothing_measured"]["value"] is None
+
+
+def test_stats_rate_limited_per_ip() -> None:
+    client = _client(_STAT_ROWS)
+    codes = [client.get(f"{PREFIX}/stats").status_code for _ in range(public_mera._STATS_LIMIT + 1)]
+    assert codes[-1] == 429
+    assert set(codes[:-1]) == {200}

tradein-mvp/backend/tests/test_public_mera_api.py

@@ -26,6 +26,7 @@ from __future__ import annotations
 
 import os
 import sys
+from datetime import UTC, datetime
 from unittest.mock import AsyncMock, MagicMock, patch
 
 # Settings требует DATABASE_URL на конструирование — stub до любого app-импорта
@@ -75,9 +76,13 @@ def _reset_limiters():
     """
     public_mera._suggest_limiter._hits.clear()
     public_mera._coverage_limiter._hits.clear()
+    public_mera._stats_limiter._hits.clear()
+    public_mera._showcase_limiter._hits.clear()
     yield
     public_mera._suggest_limiter._hits.clear()
     public_mera._coverage_limiter._hits.clear()
+    public_mera._stats_limiter._hits.clear()
+    public_mera._showcase_limiter._hits.clear()
 
 
 @pytest.fixture()
@@ -105,9 +110,16 @@ def client() -> TestClient:
 # ── 1-2. Периметр и его связка с rbac ────────────────────────────────────────
 
 
-def test_public_router_exposes_exactly_two_routes() -> None:
+def test_public_router_exposes_exactly_the_declared_routes() -> None:
     paths = {r.path for r in public_mera.router.routes}
-    assert paths == {"/suggest", "/coverage"}, (
+    assert paths == {
+        "/suggest",
+        "/coverage",
+        "/stats",
+        "/showcase",
+        "/estimate",
+        "/estimate/read",
+    }, (
         "изменился набор публичных (анонимных) ручек МЕРЫ. Это не рефакторинг: "
         "всё под /api/public/ проксируется на meraocenka.ru целиком и доступно "
         "без идентичности. Обнови тест ОСОЗНАННО вместе с rbac._PUBLIC_PATHS."
@@ -155,6 +167,107 @@ def test_anonymous_gets_suggest(client: TestClient) -> None:
     assert resp.json() == {"items": []}
 
 
+_SHOWCASE_ROW = {
+    "district": None,
+    "rooms": 2,
+    "area_m2": 54.0,
+    "floor": 5,
+    "total_floors": None,
+    "deal_quarter": "II квартал 2026",
+    "predicted_rub": 6_100_000,
+    "fact_rub": 5_900_000,
+    "err_pct": 3.39,
+    "n_analogs": 41,
+    "note": "не point-in-time",
+}
+_SHOWCASE_RUN = {
+    "computed_at": datetime(2026, 8, 29, 10, 0, tzinfo=UTC),
+    "considered": 200,
+    "priced": 173,
+    "no_prediction": 27,
+    "incomplete": 4,
+    "eligible": 169,
+    "written": 20,
+    "with_district": 18,
+    "rejection_rule": "данных нет: нет прогноза / квартала / площади",
+}
+
+
+def _showcase_db(run: dict | None = _SHOWCASE_RUN, rows: list | None = None) -> MagicMock:
+    db = MagicMock()
+    chain = db.execute.return_value.mappings.return_value
+    chain.first.return_value = run
+    chain.all.return_value = [_SHOWCASE_ROW] if rows is None else rows
+    return db
+
+
+def test_anonymous_gets_showcase(client: TestClient) -> None:
+    """Витрина открыта анониму и отдаёт то, что лежит в таблице.
+
+    Пустое поле района проходит НАСКВОЗЬ как null: витрина не имеет права
+    подставить правдоподобный район там, где его не удалось определить.
+    """
+    client.app.dependency_overrides[get_db] = lambda: _showcase_db()
+
+    resp = client.get(f"{PREFIX}/showcase")
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert body["computed_at"].startswith("2026-08-29T10:00")
+    assert body["deals"][0]["district"] is None
+    assert body["deals"][0]["fact_rub"] == 5_900_000
+    # Адреса в контракте ручки нет вовсе — в `deals` дом известен у 2.7% строк.
+    assert "address" not in body["deals"][0]
+
+
+def test_showcase_carries_counters_so_20_rows_cannot_read_as_all_there_was(
+    client: TestClient,
+) -> None:
+    """Счётчики прогона доезжают до фронта, а не остаются в логе бэкенда.
+
+    Без них «20 отличных строк» неотличимо от «столько и было»: посетитель не
+    может отличить выборку из работы оценщика от её лучшего хвоста. Здесь
+    показано 20 из 169 годных — и оба числа обязаны быть в ответе, вместе с
+    правилом, по которому отсеяно остальное.
+
+    Ломать так: убрать `stats` из `ShowcaseResponse` (или перестать его
+    заполнять) — тест покраснеет на отсутствующем ключе, а не на форме.
+    """
+    client.app.dependency_overrides[get_db] = lambda: _showcase_db()
+
+    body = client.get(f"{PREFIX}/showcase").json()
+    stats = body["stats"]
+    assert stats is not None, "витрина отдаёт строки без счётчиков — подписаться нечем"
+    assert stats["considered"] == 200
+    assert stats["eligible"] == 169
+    assert stats["written"] == 20
+    assert stats["eligible"] - stats["written"] == 149, "не видно, сколько годных не влезло"
+    assert stats["no_prediction"] == 27
+    assert stats["rejection_rule"], "правило отбраковки не подписано"
+
+
+def test_showcase_without_a_run_shows_nothing_and_says_so(client: TestClient) -> None:
+    """Пересчёта не было — ни строк, ни счётчиков, и это штатный ответ.
+
+    Обратная сторона предыдущего теста: `stats` не выдумывается там, где
+    прогона не было. Заодно это гейт на связку «строки берутся ИЗ прогона»:
+    строки в таблице есть, но прогона нет — показывать их не из чего.
+    """
+    client.app.dependency_overrides[get_db] = lambda: _showcase_db(run=None)
+
+    body = client.get(f"{PREFIX}/showcase").json()
+    assert body == {"computed_at": None, "deals": [], "stats": None}
+
+
+def test_showcase_rate_limited_per_ip(client: TestClient) -> None:
+    db = _showcase_db(rows=[])
+    client.app.dependency_overrides[get_db] = lambda: db
+    codes = [
+        client.get(f"{PREFIX}/showcase").status_code for _ in range(public_mera._SHOWCASE_LIMIT + 1)
+    ]
+    assert codes[: public_mera._SHOWCASE_LIMIT] == [200] * public_mera._SHOWCASE_LIMIT
+    assert codes[-1] == 429, f"бюджет витрины не сработал: {codes}"
+
+
 def test_suggest_is_post_so_address_never_lands_in_access_log() -> None:
     """Адрес едет ТЕЛОМ, а не в query.
 

tradein-mvp/backend/tests/test_public_mera_estimate.py

@@ -0,0 +1,384 @@
+"""Анонимный расчёт /api/public/mera/estimate* — что здесь запинено и почему.
+
+1. СОГЛАСИЕ ДО ЗАПИСИ. Отказ без согласия обязан случиться ДО того, как
+   адрес физлица дойдёт до БД. Поэтому проверяется не только код 422, но и
+   то, что расчёт вообще не запускался и в сессию не ушло ни одного запроса:
+   «422, но строка записана» выглядит в логах как успех приватности и им не
+   является.
+
+2. ФЛАГ. Выключенный `public_estimate_enabled` обязан давать 404 — иначе
+   мерж этого кода сам по себе открывает наружу запись ПДн.
+
+3. ТОКЕН. В БД уходит ХЭШ, а не токен; выборка отфильтрована по сроку
+   жизни; «нет такого» и «протух» неотличимы снаружи.
+
+4. БЕСПЛАТНАЯ ЧАСТЬ. Публичный ответ не содержит ни одного платного поля.
+   Проверяется набором ключей на равенство: любое добавленное поле роняет
+   тест, в том числе случайно добавленное платное.
+
+5. ДЕЛЕГАЦИЯ. Весь анти-абузный рассказ ручки (анонимная квота на связку
+   cookie+IP, семафор одновременности, 503 вместо 502, consent-гейт) держится
+   ровно на ОДНОМ факте: в `app.api.v1.trade_in.estimate` уходит
+   `x_authenticated_user=None`. `AsyncMock` съедает любую сигнатуру, поэтому
+   «расчёт вызвали» тут ничего не значит — проверяются фактические kwargs
+   вызова, причём запрос идёт С заголовком `X-Authenticated-User`: подмена
+   `None` на чтение заголовка обязана красить тест, иначе публичная ручка
+   молча начнёт считать чужим пользователем и мимо анонимной квоты.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import os
+import sys
+from datetime import UTC, datetime
+from unittest.mock import AsyncMock, MagicMock, patch
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+_wp_mock = MagicMock()
+sys.modules.setdefault("weasyprint", _wp_mock)
+sys.modules.setdefault("weasyprint.CSS", _wp_mock)
+sys.modules.setdefault("weasyprint.HTML", _wp_mock)
+
+import pytest  # noqa: E402
+from fastapi import FastAPI  # noqa: E402
+from fastapi.testclient import TestClient  # noqa: E402
+
+from app.api.public import mera as public_mera  # noqa: E402
+from app.core.db import get_db  # noqa: E402
+from app.schemas.trade_in import AggregatedEstimate, CoverageProbeResponse  # noqa: E402
+
+PREFIX = "/api/public/mera"
+
+_BODY = {
+    "address": "Малышева 51",
+    "area_m2": 54.0,
+    "rooms": 2,
+    "city_hint": "Екатеринбург",
+    "consent": True,
+}
+
+_FAKE_COVERAGE = CoverageProbeResponse(
+    status="ok",
+    n_listings=34,
+    median_listing_age_days=44,
+    n_with_age=6,
+    radius_m=1000,
+    city="Екатеринбург",
+    threshold=10,
+)
+
+_TOKEN_EXPIRES = datetime(2026, 9, 5, 12, 0, tzinfo=UTC)
+_FAKE_ESTIMATE_ID = "11111111-1111-1111-1111-111111111111"
+
+
+def _fake_estimate_result() -> MagicMock:
+    """Результат закрытого контура. MagicMock, а не собранный AggregatedEstimate:
+    хендлеру нужны четыре атрибута, а перечисление всех полей платной модели
+    здесь означало бы, что тест придётся править при каждой правке эстиматора."""
+    result = MagicMock()
+    result.estimate_id = _FAKE_ESTIMATE_ID
+    result.n_analogs = 27
+    result.target_lat = 56.838
+    result.target_lon = 60.597
+    return result
+
+
+@pytest.fixture(autouse=True)
+def _reset_limiters():
+    public_mera._estimate_limiter._hits.clear()
+    public_mera._estimate_read_limiter._hits.clear()
+    public_mera._daily_estimate_limiter._hits.clear()
+    yield
+    public_mera._estimate_limiter._hits.clear()
+    public_mera._estimate_read_limiter._hits.clear()
+    public_mera._daily_estimate_limiter._hits.clear()
+
+
+@pytest.fixture()
+def db() -> MagicMock:
+    session = MagicMock()
+    session.execute.return_value.fetchone.return_value = MagicMock(
+        public_token_expires_at=_TOKEN_EXPIRES,
+        n_analogs=27,
+        lat=56.838,
+        lon=60.597,
+        rooms=2,
+        area_m2=54.0,
+    )
+    return session
+
+
+@pytest.fixture()
+def client(db: MagicMock) -> TestClient:
+    """Без rbac-мидлвари: узость auth-исключения проверяет соседний
+    tests/test_public_mera_api.py, здесь предмет — сами ручки."""
+    app = FastAPI()
+    app.include_router(public_mera.router, prefix=PREFIX)
+    app.dependency_overrides[get_db] = lambda: db
+    return TestClient(app)
+
+
+@pytest.fixture()
+def flag_on():
+    with patch.object(public_mera.settings, "public_estimate_enabled", True):
+        yield
+
+
+# ── 1. Согласие фиксируется до первого обращения к БД ────────────────────────
+
+
+def test_estimate_without_consent_is_422_and_writes_nothing(
+    client: TestClient, db: MagicMock, flag_on: None
+) -> None:
+    body = {k: v for k, v in _BODY.items() if k != "consent"}
+    with patch.object(public_mera, "estimate", AsyncMock()) as estimate_mock:
+        resp = client.post(f"{PREFIX}/estimate", json=body)
+
+    assert resp.status_code == 422, resp.text
+    assert not estimate_mock.called, (
+        "расчёт запустился без согласия — адрес физлица дошёл бы до "
+        "trade_in_estimates раньше, чем человек что-либо разрешил"
+    )
+    assert db.execute.call_args_list == [], (
+        "в БД ушёл запрос при отсутствии согласия: проверка согласия сдвинулась "
+        "ПОСЛЕ работы с данными"
+    )
+
+
+def test_estimate_with_consent_false_is_also_422(client: TestClient, flag_on: None) -> None:
+    """`consent: false` — это осознанный отказ, а не «поле не прислали»."""
+    with patch.object(public_mera, "estimate", AsyncMock()) as estimate_mock:
+        resp = client.post(f"{PREFIX}/estimate", json={**_BODY, "consent": False})
+    assert resp.status_code == 422, resp.text
+    assert not estimate_mock.called
+
+
+# ── 2. Флаг ──────────────────────────────────────────────────────────────────
+
+
+def test_estimate_is_404_while_flag_is_off(client: TestClient) -> None:
+    with patch.object(public_mera, "estimate", AsyncMock()) as estimate_mock:
+        resp = client.post(f"{PREFIX}/estimate", json=_BODY)
+    assert resp.status_code == 404, resp.text
+    assert not estimate_mock.called
+
+
+def test_estimate_read_is_404_while_flag_is_off(client: TestClient, db: MagicMock) -> None:
+    resp = client.post(f"{PREFIX}/estimate/read", json={"token": "x" * 32})
+    assert resp.status_code == 404, resp.text
+    assert db.execute.call_args_list == []
+
+
+def test_flag_default_is_off() -> None:
+    """Дефолт — выключено: мерж кода не должен открывать запись ПДн наружу."""
+    from app.core.config import Settings
+
+    assert Settings.model_fields["public_estimate_enabled"].default is False
+
+
+# ── 3. Токен ─────────────────────────────────────────────────────────────────
+
+
+def test_estimate_stores_token_hash_not_token(
+    client: TestClient, db: MagicMock, flag_on: None
+) -> None:
+    with (
+        patch.object(public_mera, "estimate", AsyncMock(return_value=_fake_estimate_result())),
+        patch.object(public_mera, "coverage_probe", return_value=_FAKE_COVERAGE),
+    ):
+        resp = client.post(f"{PREFIX}/estimate", json=_BODY)
+
+    assert resp.status_code == 200, resp.text
+    token = resp.json()["token"]
+    assert len(token) >= 32, "короткий токен перебираем"
+
+    params = db.execute.call_args_list[0].args[1]
+    assert token not in str(params), "сам токен уехал в БД — дамп базы открывает чужие расчёты"
+    assert params["token_hash"] == hashlib.sha256(token.encode()).hexdigest()
+    assert params["id"] == _FAKE_ESTIMATE_ID, (
+        "токен вешается не на ту строку: UPDATE обязан идти по id ТОЛЬКО ЧТО "
+        "посчитанной оценки (result.estimate_id), иначе ссылка либо ведёт в чужой "
+        "расчёт, либо не ведёт никуда"
+    )
+
+
+def test_read_filters_by_expiry_and_hash(client: TestClient, db: MagicMock, flag_on: None) -> None:
+    """ЧТО проверено: в тексте отправленного SQL присутствует предикат срока жизни,
+    а в параметры уехал ХЭШ токена, не сам токен.
+
+    ЧЕГО НЕ проверено: что протухший токен действительно не читается. Сессия здесь —
+    MagicMock, запрос не исполняется: SQL не разбирается, `fetchone()` отдаёт строку
+    из фикстуры независимо от WHERE. Поэтому зелёными останутся, например, предикат,
+    перенесённый туда, где он ничего не отсекает, сравнение NOW() с полем не того
+    типа и любая ошибка в самом сравнении — текст-то совпадает.
+
+    Поведенческой версии нет намеренно: она требует живого Postgres с миграцией 278
+    (образец self-skip-теста — tests/test_purge_expired_trade_in_data.py::_live_session),
+    а писать её вслепую, ни разу не прогнав, значит завести ещё один тест, зелёный
+    по построению. Появится доступный Postgres — этот тест заменяется на вставку
+    двух строк (свежий токен и просроченный) с проверкой 200 против 404.
+    """
+    token = "t" * 40
+    with patch.object(public_mera, "coverage_probe", return_value=_FAKE_COVERAGE):
+        resp = client.post(f"{PREFIX}/estimate/read", json={"token": token})
+    assert resp.status_code == 200, resp.text
+
+    sql = str(db.execute.call_args_list[0].args[0])
+    assert "public_token_expires_at > NOW()" in sql, (
+        "из выборки пропал срок жизни ссылки — протухший токен снова читает расчёт"
+    )
+    params = db.execute.call_args_list[0].args[1]
+    assert params["token_hash"] == hashlib.sha256(token.encode()).hexdigest()
+
+
+def test_unknown_or_expired_token_is_404(client: TestClient, db: MagicMock, flag_on: None) -> None:
+    """Чужой и протухший токен неотличимы: 404 в обоих случаях.
+
+    Мок сессии отдаёт None ровно так же, как реальный SELECT с предикатом
+    `public_token_expires_at > NOW()` — то есть и для несуществующего токена,
+    и для просроченного.
+    """
+    db.execute.return_value.fetchone.return_value = None
+    resp = client.post(f"{PREFIX}/estimate/read", json={"token": "z" * 40})
+    assert resp.status_code == 404, resp.text
+    assert "не найден" in resp.json()["detail"].lower()
+
+
+# ── 4. В публичном ответе нет платного ───────────────────────────────────────
+
+# Поля AggregatedEstimate, которые продукт продаёт. Список не для красоты: он
+# сверяется с реальной моделью ниже, поэтому переименование поля в эстиматоре
+# не оставит здесь мёртвую строку-обманку.
+_PAID_FIELDS = {
+    "median_price_rub",
+    "range_low_rub",
+    "range_high_rub",
+    "median_price_per_m2",
+    "market_percentile",
+    "analogs",
+    "actual_deals",
+    "expected_sold_price_rub",
+    "est_days_on_market",
+    "price_trend",
+}
+
+_FREE_FIELDS = {"token", "token_expires_at", "n_analogs", "coverage"}
+
+
+def test_paid_field_names_still_exist_in_estimator_model() -> None:
+    """Контроль самого контроля: если поле переименовали, тест ниже сравнивал бы
+    публичный ответ с несуществующими именами и был бы зелёным по построению."""
+    missing = _PAID_FIELDS - set(AggregatedEstimate.model_fields)
+    assert not missing, f"эти поля исчезли из AggregatedEstimate, обнови список: {missing}"
+
+
+def test_public_result_model_exposes_only_free_fields() -> None:
+    assert set(public_mera.PublicEstimateResult.model_fields) == _FREE_FIELDS, (
+        "изменился состав публичного ответа. Любое поле отсюда видит любой аноним "
+        "без оплаты — правь ОСОЗНАННО вместе с этим тестом"
+    )
+
+
+def test_public_estimate_response_leaks_no_paid_field(client: TestClient, flag_on: None) -> None:
+    with (
+        patch.object(public_mera, "estimate", AsyncMock(return_value=_fake_estimate_result())),
+        patch.object(public_mera, "coverage_probe", return_value=_FAKE_COVERAGE),
+    ):
+        resp = client.post(f"{PREFIX}/estimate", json=_BODY)
+    body = resp.json()
+    assert set(body) == _FREE_FIELDS, f"публичный ответ отдаёт лишнее: {set(body) - _FREE_FIELDS}"
+    assert not _PAID_FIELDS & set(body)
+    # Вложенный coverage тоже без цен — по построению CoverageProbeResponse,
+    # но проверяем, а не верим: он мог обрасти полем «медиана ₽/м²».
+    assert not _PAID_FIELDS & set(body["coverage"])
+    assert body["n_analogs"] == 27
+    assert body["coverage"]["median_listing_age_days"] == 44
+
+
+def test_read_returns_same_free_shape(client: TestClient, flag_on: None) -> None:
+    """POST и повторное чтение обязаны отдавать ОДНУ форму: разойдись они —
+    фронт после перезагрузки страницы показал бы не то же самое."""
+    with patch.object(public_mera, "coverage_probe", return_value=_FAKE_COVERAGE):
+        resp = client.post(f"{PREFIX}/estimate/read", json={"token": "t" * 40})
+    assert resp.status_code == 200, resp.text
+    assert set(resp.json()) == _FREE_FIELDS
+    assert resp.json()["n_analogs"] == 27
+
+
+# ── 5. Делегация в закрытый контур ───────────────────────────────────────────
+
+
+def test_estimate_delegates_anonymously_despite_auth_header(
+    client: TestClient, db: MagicMock, flag_on: None
+) -> None:
+    """Единственный тест, который смотрит НА АРГУМЕНТЫ делегации, а не на факт вызова.
+
+    Запрос идёт с `X-Authenticated-User: admin` — то есть ровно тем заголовком,
+    который в закрытом контуре означает «это авторизованный пользователь». Ручка
+    обязана всё равно позвать эстиматор анонимом: `x_authenticated_user=None`
+    включает там consent-гейт и анонимную квоту на связку cookie+IP. Пробрось
+    сюда заголовок — и публичная форма начнёт считать от чужого имени в обход
+    квоты, а все остальные тесты этого файла останутся зелёными: `AsyncMock`
+    принимает любую сигнатуру.
+    """
+    with (
+        patch.object(
+            public_mera, "estimate", AsyncMock(return_value=_fake_estimate_result())
+        ) as estimate_mock,
+        patch.object(public_mera, "coverage_probe", return_value=_FAKE_COVERAGE),
+    ):
+        resp = client.post(
+            f"{PREFIX}/estimate", json=_BODY, headers={"X-Authenticated-User": "admin"}
+        )
+
+    assert resp.status_code == 200, resp.text
+    call = estimate_mock.await_args
+    assert call is not None, "делегации не было вовсе"
+    assert call.args == (), (
+        "аргументы поехали позиционно — сверка по именам ниже перестала что-либо "
+        "проверять; зови эстиматор с ключевыми словами"
+    )
+    assert "x_authenticated_user" in call.kwargs, (
+        "аргумент исчез из вызова: у эстиматора он по умолчанию None, но тогда "
+        "гарантия анонимности держится на чужом дефолте, а не на этой ручке"
+    )
+    assert call.kwargs["x_authenticated_user"] is None, (
+        "публичная ручка передала пользователя в закрытый контур: consent-гейт и "
+        "анонимная квота выключаются, аноним считает от чужого имени "
+        f"(пришло: {call.kwargs['x_authenticated_user']!r})"
+    )
+    # Остальное едет тем же вызовом: сессия — та, что отдана зависимостью (иначе
+    # UPDATE токена ниже пишет в другую транзакцию), тело — то, что прислал аноним.
+    assert call.kwargs["db"] is db
+    assert call.kwargs["payload"].address == _BODY["address"]
+    assert call.kwargs["payload"].area_m2 == _BODY["area_m2"]
+    assert call.kwargs["payload"].rooms == _BODY["rooms"]
+
+
+# ── 6. Бюджеты ───────────────────────────────────────────────────────────────
+
+
+def test_estimate_rate_limited_per_ip(client: TestClient, flag_on: None) -> None:
+    with (
+        patch.object(public_mera, "estimate", AsyncMock(return_value=_fake_estimate_result())),
+        patch.object(public_mera, "coverage_probe", return_value=_FAKE_COVERAGE),
+    ):
+        codes = [
+            client.post(f"{PREFIX}/estimate", json=_BODY).status_code
+            for _ in range(public_mera._ESTIMATE_LIMIT + 1)
+        ]
+    assert codes[:-1] == [200] * public_mera._ESTIMATE_LIMIT
+    assert codes[-1] == 429, f"бюджет не сработал: {codes}"
+
+
+def test_daily_budget_stops_estimates_for_everyone(client: TestClient, flag_on: None) -> None:
+    """Общий потолок — про сумму по всем IP, а не про одного клиента."""
+    for _ in range(public_mera._DAILY_ESTIMATE_BUDGET):
+        public_mera._daily_estimate_limiter.record(public_mera._ESTIMATE_GLOBAL_KEY)
+    with patch.object(public_mera, "estimate", AsyncMock()) as estimate_mock:
+        resp = client.post(f"{PREFIX}/estimate", json=_BODY)
+    assert resp.status_code == 429, resp.text
+    assert int(resp.headers["Retry-After"]) > 0
+    assert not estimate_mock.called

tradein-mvp/backend/tests/test_scraper_kit_scheduler_parity.py

@@ -68,6 +68,7 @@ _PRODUCT_SOURCES: set[str] = {
     "sber_index_pull",
     "rosreestr_quarter_poll",
     "deals_freshness_monitor",
+    "landing_stats_refresh",
     "newbuilding_enrich",
     "yandex_newbuilding_sweep",
     "geoportal_coords_backfill",