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

"""Daily recompute of the asking→sold correction ratios (#648 Stage 4).

ПРОБЛЕМА: asking_to_sold_ratios (migration 080) засеяна один раз derivation-CTE
(медиана SOLD ДКП за 12 мес vs медиана ASKING активных listings, per-rooms + global -1
fallback). Оценщик (estimator.py, Stage 3) читает её per-estimate (cache 300s) и домножает
asking-median прогноз на sold/asking. По мере ночного импорта новых ДКП-сделок
(rosreestr_dkp_import) ratio устаревает — нужен периодический пересчёт по тому же derivation.

TRUE-MIRROR REFRESH (флаг database-expert): 080-seed использовал ON CONFLICT DO UPDATE,
который при повторном прогоне ОБНОВЛЯЕТ существующие строки, но НЕ удаляет per-rooms строки
бакетов, упавших ниже порога 30/30 на более позднем прогоне (StaLE rows). Поэтому refresh
сначала DELETE FROM asking_to_sold_ratios WHERE district = '' (все строки #648), затем
заново гоняет ту же 080-derivation INSERT...SELECT. DELETE+INSERT в ОДНОЙ транзакции
(атомарно — таблица никогда не пуста mid-refresh). После commit таблица == свежий re-seed.

Задача синхронная (DB-only, никаких внешних HTTP-вызовов) — запускается kit-scheduler'ом
через product_handlers._job_asking_to_sold_ratio (run_in_executor), по образцу
snapshot_listing_sources / import_rosreestr_dkp.

Окно расписания 06:00-07:00 UTC — ПОСЛЕ rosreestr_dkp_import (04:00-06:00 UTC), чтобы
refresh потреблял свежие ДКП-сделки того же дня.

SQL derivation ниже повторяет seed в data/sql/080_asking_to_sold_ratios.sql (deal_side /
ask_side / per_bucket + deal_global / ask_global / global_row: трейлинг-12мес окно, ppm²-полоса
[_PPM2_MIN, settings.asking_ratio_ppm2_max] (default [30000,1200000]), порог n_deals>=30 AND
n_listings>=30 для per_rooms, global -1 строка всегда). ON CONFLICT убран — DELETE идёт первым,
конфликтов нет (повторный прогон в одной tx невозможен, refresh = re-seed по семантике).

#2656 — ВТОРОЕ ПРЕДНАМЕРЕННОЕ РАСХОЖДЕНИЕ с 080 (первое — #2620 ниже): ask_side/ask_global
несут фильтр свежести `scraped_at > NOW() - LISTINGS_FRESH_DAYS дней` — тот же, что эстиматор
применяет к ЧИСЛИТЕЛЮ (_COMMON_WHERE). Без него знаменатель считался по бессрочной популяции
объявлений, а числитель — по 14-дневной, т.е. коэффициент калибровался на одном рынке, а
применялся к другому. Замер на проде (2026-08, #2656): глобальный коэффициент −0.49%, бакеты
44-62 +2.43% и 62-85 +4.58%; NULL-сегмент в знаменателе схлопывается с 674 строк до 20.

#2620 — ПЕРВОЕ ПРЕДНАМЕРЕННОЕ РАСХОЖДЕНИЕ с 080: deal_side бакетится по
LEAST(GREATEST(rooms,0),4), а ask_side — по _AREA_ROOMS_BUCKET_SQL (площадь, та же формула,
что deals.rooms получает при импорте). Причина — deals.rooms НЕ настоящая комнатность
(Росреестр её не отдаёт), это синтетика из площади; сравнивать её с РЕАЛЬНЫМИ комнатами
listings значило сравнивать разные классификации. Замер на проде (2026-08, #2620) показал
миграцию 23-55% объявлений между бакетами при таком сравнении — не только в бакете «4+»
(который к тому же обрезан обрезкой ELSE 4, тогда как listings.rooms доходит до 10) — и
это и была причина ratio>1 в бакете 4+ (см. _AREA_ROOMS_BUCKET_SQL ниже).
"""

from __future__ import annotations

import logging

from sqlalchemy import text
from sqlalchemy.orm import Session

from app.core.config import LISTINGS_FRESH_DAYS, settings
from app.services import regions as regions_mod
from app.services import scrape_runs as runs_mod

# Нижняя граница ppm² — отсекает нежилые/технические сделки; не меняется.
_PPM2_MIN: int = 30_000

# #C2 — исторически asking-сторона (listings) была покрыта скрейпом ТОЛЬКО по ЕКБ, а
# миграция 177 залила ДКП-сделки по всей обл.66 (368 городов) → sold-медиана смешивала
# дешёвую область с ЕКБ-asking и обваливала ratio (0.877→0.62, «выкупная» −29% системно).
# Скоупили SOLD-сторону (deal_side/deal_global) на ЕКБ, чтобы sold и asking считались по
# ОДНОМУ рынку.
#
# #2583 H2 (аудит, 2026-08): oblast-развёртки заработали 12 июля — областные объявления
# попали в знаменатель (ask_side/ask_global) без городского скоупа, а sold-сторона
# осталась скоуплена на ЕКБ → асимметрия вернулась с другой стороны (дешёвая область
# занижает ask-медиану → ratio завышен на 2.5-5.3% по всем бакетам, выкупные цены
# системно переплачены). Теперь ask_side/ask_global ТОЖЕ скоупятся этим паттерном
# (предикат `city IS NULL OR city ILIKE :asking_city` — см. комментарий на месте в CTE
# ниже) — симметрично deal-стороне.
#
# #3512 (per-region ratio, migration 304): эта городская квота — костыль ИМЕННО региона
# 66 (asking-скрейп исторически покрывал только сам ЕКБ, см. #C2 выше), и трогать её
# нельзя — числа региона 66 обязаны остаться byte-for-byte прежними. Для ЛЮБОГО другого
# региона (deals.region_code / listings.region_code — обе таблицы несут колонку) городской
# квоты не было и не нужно: там достаточно `region_code = :region_code` симметрично на
# обеих сторонах (см. _REDERIVE_SQL_REGION ниже) — колонка `district` остаётся
# зарезервированной под #647 (гео-районы ВНУТРИ региона), per-region разрез теперь
# несёт `region_code`, а не `district`.
_ASKING_CITY_PATTERN: str = "%Екатеринбург%"
# Верхняя граница берётся из settings.asking_ratio_ppm2_max (default 1_200_000).
# QA-note: точное значение сверить с `SELECT max(price_per_m2) FROM deals
# WHERE source='rosreestr'` на проде — ceiling должен быть > max(ppm²) premium-сделок.

# #3512: регионы, для которых считается ОТДЕЛЬНАЯ (не-ЕКБ) деривация ниже —
# весь реестр покрытия (app.services.regions — ЕДИНСТВЕННЫЙ источник правды про регионы)
# минус 66 (у него своя историческая деривация выше). Появление нового региона в реестре
# автоматически включает его в пересчёт ratio, без правки этого файла.
_OTHER_REGION_CODES: tuple[int, ...] = tuple(
    sorted(code for code in regions_mod.REGIONS if code != 66)
)

# #2620 — синтетический "бакет комнат по площади", ИСТОЧНИК ИСТИНЫ:
# tradein-mvp/deploy/import-rosreestr.sh (Росреестр не отдаёт комнатность — deals.rooms
# синтезируется из area_m2 при импорте ровно этим CASE). Три представления ОДНОЙ формулы —
# держи границы (30/44/62/85) в синхроне при правке: shell (import-rosreestr.sh) → SQL
# (эта константа, ask_side ниже) → Python (area_bucket() ниже, estimator.py rekey #2620-2).
_AREA_ROOMS_BUCKET_SQL = (
    "CASE WHEN area_m2 < 30 THEN 0 WHEN area_m2 < 44 THEN 1 "
    "WHEN area_m2 < 62 THEN 2 WHEN area_m2 < 85 THEN 3 ELSE 4 END"
)


# ── #time-adjust: приведение SOLD-стороны к сегодняшнему дню (rollback за settings) ──
# ПРОБЛЕМА: sold-медиана (deal_side/deal_global/deal_geo) — сырые цены сделок за
# трейлинг-12 мес (на проде фактическое окно сентябрь 2025 - апрель 2026, медиана
# ~декабрь 2025), ask-медиана — объявления за LISTINGS_FRESH_DAYS (сегодня). На растущем
# рынке это занижает ratio (числитель отстаёт от знаменателя во времени). Фикс — тот же
# приём, что estimator._sber_time_factor применяет к ДКП-коридору (#794): каждая сделка
# домножается на factor = idx[последний доступный месяц серии] / idx[месяц сделки]
# (с clamp SBER_TIME_FACTOR_MIN/MAX). Серия и её имя — ТА ЖЕ карта sber_region_series_name
# (app.services.estimator), НЕ вторая карта — импортируется ЛЕНИВО внутри
# recompute_asking_to_sold_ratios(), т.к. estimator.py импортирует area_bucket ИЗ этого
# модуля на верхнем уровне (см. выше) — top-level импорт в обратную сторону дал бы цикл.
#
# Двойной учёт (проверено, см. PR-описание/vault): estimator._fetch_dkp_corridor тоже
# зовёт _sber_time_factor, но на ЖИВОЙ per-request ДКП-коридор (отдельный SQL, отдельная
# цель — condition-guard corridor-clamp поверх ASKING-медианы, estimator.py:3989-4002),
# а не на ratio из asking_to_sold_ratios. Эта таблица и коридор — независимые артефакты
# по одному сырому источнику (deals); один и тот же множитель здесь и там НЕ перемножается
# на одно и то же число дважды.
#
# Дашборд выбирается В SQL по приоритету SBER_COEFF_DASHBOARDS (bind-массив
# :sber_dashboards, array_position — первый непустой побеждает, тот же порядок, что
# estimator._load_sber_index_series перебирает по одному дашборду). Приведение
# управляется флагом settings.asking_ratio_time_adjust_enabled (:time_adjust_enabled) —
# False даёт factor=1.0 для КАЖДОЙ сделки (байт-в-байт прежнее поведение, откат без
# деплоя). Если серии для города вообще нет (sber_bounds пуст) — тоже factor=1.0; это
# НЕ ошибка (сделка не выбрасывается), но recompute_asking_to_sold_ratios логирует
# факт счётчиком, а не молча (см. _sber_series_missing ниже). Если для конкретного
# месяца сделки нет точки — берётся ближайший БОЛЕЕ РАННИЙ месяц серии (или самый
# ранний, если сделка старше начала серии) — то же правило, что estimator._sber_time_factor.
_SBER_FACTOR_CTES = """
    sber_series_raw AS (
        SELECT period_month, index_value_rub_m2, dashboard,
               array_position(CAST(:sber_dashboards AS text[]), dashboard) AS dash_priority
        FROM sber_price_index
        WHERE city = CAST(:sber_city AS text)
          AND (segment IS NULL OR segment ILIKE '%вторичн%')
          AND dashboard = ANY(CAST(:sber_dashboards AS text[]))
    ),
    sber_best_dashboard AS (
        SELECT dashboard FROM sber_series_raw
        WHERE dash_priority IS NOT NULL
        ORDER BY dash_priority
        LIMIT 1
    ),
    sber_series AS (
        SELECT r.period_month, r.index_value_rub_m2
        FROM sber_series_raw r
        JOIN sber_best_dashboard b ON r.dashboard = b.dashboard
    ),
    sber_bounds AS (
        SELECT MAX(period_month) AS latest_month, MIN(period_month) AS earliest_month
        FROM sber_series
    ),
    sber_latest_value AS (
        SELECT s.index_value_rub_m2 AS latest_value
        FROM sber_series s JOIN sber_bounds b ON s.period_month = b.latest_month
    ),
    sber_earliest_value AS (
        SELECT s.index_value_rub_m2 AS earliest_value
        FROM sber_series s JOIN sber_bounds b ON s.period_month = b.earliest_month
    )"""

# LEFT JOIN'ы к sber_* CTE выше — вставляются в `FROM deals` ДО `WHERE` (JOIN не может
# идти после WHERE). ON TRUE — CTE содержат максимум одну строку (не декартово произведение).
# snb — LATERAL «ближайший месяц серии <= месяца сделки», коррелирован по bare deal_date
# (единственная таблица в запросе с этой колонкой — алиас не нужен).
_SBER_FACTOR_JOINS = """
          LEFT JOIN sber_bounds sb ON TRUE
          LEFT JOIN sber_latest_value slv ON TRUE
          LEFT JOIN sber_earliest_value sev ON TRUE
          LEFT JOIN LATERAL (
              SELECT s.index_value_rub_m2 AS base_value
              FROM sber_series s
              WHERE s.period_month <= date_trunc('month', deal_date)::date
              ORDER BY s.period_month DESC
              LIMIT 1
          ) snb ON TRUE
"""

# Сам фактор — 1.0 при выключенном флаге, при пустой серии, при сделке новее последнего
# месяца серии (без экстраполяции вперёд — симметрично estimator._sber_time_factor).
# GREATEST/LEAST — те же клампы SBER_TIME_FACTOR_MIN/MAX, что estimator применяет к
# ДКП-коридору (bind :factor_min/:factor_max — значения ОТТУДА, не второй набор констант).
_SBER_FACTOR_EXPR = """CASE
            WHEN NOT CAST(:time_adjust_enabled AS boolean) THEN 1.0
            WHEN sb.latest_month IS NULL THEN 1.0
            WHEN date_trunc('month', deal_date)::date >= sb.latest_month THEN 1.0
            ELSE GREATEST(
                CAST(:factor_min AS double precision),
                LEAST(
                    CAST(:factor_max AS double precision),
                    slv.latest_value / NULLIF(COALESCE(snb.base_value, sev.earliest_value), 0)
                )
            )
        END"""


def area_bucket(area_m2: float) -> int:
    """Python-двойник _AREA_ROOMS_BUCKET_SQL (границы ИДЕНТИЧНЫ, #2620).

    Используется estimator.py при ПРИМЕНЕНИИ ratio (не только при расчёте здесь) —
    ratio_resolver должен ключевать по ТОМУ ЖЕ area-бакету, что и ask_side при
    деривации, иначе mismatch просто переезжает из расчёта в применение (прод-замер
    ревьюера #2620: 310/1038 = 29.9% исторических запросов легли бы в другой бакет
    при rooms-ключе vs area-ключе).
    """
    if area_m2 < 30:
        return 0
    if area_m2 < 44:
        return 1
    if area_m2 < 62:
        return 2
    if area_m2 < 85:
        return 3
    return 4


logger = logging.getLogger(__name__)

# ── True-mirror cleanup: drop all #648 EKB rows before re-derivation ──────────
# district = '' — все строки #648 (district зарезервирован под #647, пока всегда '').
# #3512: явный region_code=66 — эта DELETE трогает ТОЛЬКО ЕКБ-строки; остальные регионы
# чистит своя _DELETE_SQL_REGION в цикле ниже (иначе один бланкет-DELETE стирал бы
# только что вставленные строки другого региона на повторном прогоне того же transaction).
# Удаляем ПЕРЕД re-derive, чтобы бакеты, упавшие ниже порога 30/30, не оставались
# stale (ON CONFLICT DO UPDATE такие строки бы не тронул). В одной транзакции с INSERT.
_DELETE_SQL = text(
    """
    DELETE FROM asking_to_sold_ratios
    WHERE region_code = 66 AND district = ''
    """
)

# #3512: та же true-mirror очистка, но per-region (используется в цикле для
# _OTHER_REGION_CODES) — CAST(:region_code AS int), никогда :region_code::int (psycopg v3).
_DELETE_SQL_REGION = text(
    """
    DELETE FROM asking_to_sold_ratios
    WHERE region_code = CAST(:region_code AS int) AND district = ''
    """
)

# ── Derivation + re-seed (БАЙТ-В-БАЙТ из 080, ON CONFLICT убран — DELETE идёт первым) ──
# deal_side / ask_side / per_bucket + deal_global / ask_global / global_row:
#   sold_median = percentile_cont(0.5) по deals.price_per_m2 (source='rosreestr',
#     ppm² ∈ [_PPM2_MIN, settings.asking_ratio_ppm2_max], deal_date >= CURRENT_DATE − 12 months),
#     бакет LEAST(GREATEST(rooms,0),4) (rooms уже синтетика-из-площади при импорте, см. #2620
#     комментарий у _AREA_ROOMS_BUCKET_SQL выше).
#   ask_median  = percentile_cont(0.5) по listings.price_per_m2
#     (is_active + свежесть scraped_at ≤ LISTINGS_FRESH_DAYS (#2656), та же ppm²-полоса
#     [_PPM2_MIN, asking_ratio_ppm2_max], тот же город что
#     SOLD-сторона — city IS NULL OR city ILIKE :asking_city, #2583 H2). Бакет —
#     _AREA_ROOMS_BUCKET_SQL (площадь, #2620), НЕ listings.rooms — см. комментарий там.
#   per_rooms строки — только при n_deals>=30 AND n_listings>=30 AND ask>0 AND sold>0.
#   global -1 строка (basis='global_fallback') — всегда (если ask>0 AND sold>0). window_months=12.
# Порог/окно — литералы; ppm²-полоса передаётся bind-параметрами :ppm2_min/:ppm2_max
# (безопасно от SQL-инъекций; CAST не нужен — psycopg v3 передаёт int напрямую).
_REDERIVE_SQL = text(
    f"""
    WITH{_SBER_FACTOR_CTES},
    -- SOLD медианы по бакетам комнат за трейлинг-12мес (ДКП Росреестра). #time-adjust:
    -- price_per_m2 домножен на sber-фактор приведения к последнему месяцу серии.
    deal_side AS (
        SELECT
            LEAST(GREATEST(rooms, 0), 4)                                  AS rooms_bucket,
            percentile_cont(0.5) WITHIN GROUP (
                ORDER BY price_per_m2 * ({_SBER_FACTOR_EXPR})
            )                                                              AS sold_median,
            COUNT(*)                                                      AS n_deals
        FROM deals
        {_SBER_FACTOR_JOINS}
        WHERE source = 'rosreestr'
          AND rooms IS NOT NULL
          AND city ILIKE :asking_city  -- #C2 SOLD-сторона на ЕКБ (match asking-рынок)
          AND price_per_m2 BETWEEN :ppm2_min AND :ppm2_max
          AND deal_date >= CURRENT_DATE - INTERVAL '12 months'
        GROUP BY LEAST(GREATEST(rooms, 0), 4)
    ),
    -- ASKING медианы по ТОМУ ЖЕ area-бакету, что deal_side (#2620) — НЕ по listings.rooms.
    -- deals.rooms — синтетика из площади (Росреестр её не отдаёт), listings.rooms — реальная
    -- комнатность; сравнение area-бакета с area-бакетом (не area-бакета с real-rooms-бакетом)
    -- убирает миграцию объявлений между бакетами (23-55% строк на проде, 2026-08, #2620) —
    -- включая инверсию ratio>1 в бакете «4+» (deals.rooms обрезан ELSE 4, а listings.rooms
    -- нет: 110/782 пяти- и более комнатных объявлений раньше схлопывались в бакет 4).
    ask_side AS (
        SELECT
            {_AREA_ROOMS_BUCKET_SQL}                                      AS rooms_bucket,
            percentile_cont(0.5) WITHIN GROUP (ORDER BY price_per_m2)     AS ask_median,
            COUNT(*)                                                      AS n_listings
        FROM listings
        WHERE is_active
          -- #2656: окно свежести — то же самое, что эстиматор применяет к ЧИСЛИТЕЛЮ
          -- (_COMMON_WHERE, LISTINGS_FRESH_DAYS). Без него числитель оценки считался
          -- по 14-дневной популяции, а знаменатель коэффициента — по бессрочной:
          -- калибровка и применение по разным рынкам. `is_active` для этого не годится
          -- — он означает разное у разных источников (TTL деактивации 30д, NULL-сегмент
          -- не деактивируется никогда: 97.4% таких строк протухшие и при этом дорогие).
          AND scraped_at > NOW() - (:fresh_days || ' days')::interval
          AND rooms IS NOT NULL
          -- #2620 hardening: area_m2 IS NULL falls into the CASE ELSE branch (bucket 4)
          -- of _AREA_ROOMS_BUCKET_SQL — a latent "everything unmeasured looks like a big
          -- flat" trap. Excluded explicitly instead of relying on ELSE-as-junk-drawer.
          AND area_m2 IS NOT NULL
          AND price_per_m2 BETWEEN :ppm2_min AND :ppm2_max
          -- novostroyki guard (#1186): NULL = legacy вторичка до м.011
          AND (listing_segment IS NULL OR listing_segment = 'vtorichka')
          -- #2583 H2: скоупим ASKING-сторону на тот же город, что и SOLD-сторона
          -- (симметрично deal_side выше) — иначе дешёвые oblast-объявления (развёртки
          -- с 12 июля) занижают ask-медиану и завышают ratio. city IS NULL считается
          -- "своим" (не отбрасывается) НАМЕРЕННО: listings.city заполнена пока только у
          -- Авито (Циан/Домклик/Яндекс — NULL, #2598/#2606), симметричный
          -- `city ILIKE :asking_city` без IS NULL выбросил бы ~70% выборки. По мере
          -- роста покрытия колонки этот предикат сам ужесточается без правок кода; когда
          -- покрытие станет полным — заменить на строго симметричный `city ILIKE :asking_city`.
          AND (city IS NULL OR city ILIKE :asking_city)
        GROUP BY {_AREA_ROOMS_BUCKET_SQL}
    ),
    -- Per-rooms строки: только бакеты с обеими сторонами, прошедшие порог 30/30 и ask>0.
    -- Тонкие бакеты (n<30) сюда НЕ попадают → estimator делает fallback на -1.
    per_bucket AS (
        SELECT
            d.rooms_bucket,
            ''::text                                  AS district,
            (d.sold_median / a.ask_median)::numeric   AS ratio,
            round(d.sold_median)::bigint              AS sold_median,
            round(a.ask_median)::bigint               AS ask_median,
            d.n_deals::int                            AS n_deals,
            a.n_listings::int                         AS n_listings,
            12                                        AS window_months,
            'per_rooms'::text                         AS basis
        FROM deal_side d
        JOIN ask_side a USING (rooms_bucket)
        WHERE d.n_deals >= 30
          AND a.n_listings >= 30
          AND a.ask_median IS NOT NULL
          AND a.ask_median > 0
          AND d.sold_median IS NOT NULL          -- divide-safety (порог n_deals>=30 гарантирует)
          AND d.sold_median > 0
    ),
    -- SOLD медиана по ВСЕМ комнатам (без бакет-фильтра) за трейлинг-12мес — для global row.
    deal_global AS (
        SELECT
            percentile_cont(0.5) WITHIN GROUP (
                ORDER BY price_per_m2 * ({_SBER_FACTOR_EXPR})
            )                                                          AS sold_median,
            COUNT(*)                                                  AS n_deals
        FROM deals
        {_SBER_FACTOR_JOINS}
        WHERE source = 'rosreestr'
          AND rooms IS NOT NULL
          AND city ILIKE :asking_city  -- #C2 SOLD-сторона на ЕКБ (match asking-рынок)
          AND price_per_m2 BETWEEN :ppm2_min AND :ppm2_max
          AND deal_date >= CURRENT_DATE - INTERVAL '12 months'
    ),
    -- ASKING медиана по ВСЕМ активным listings (без бакет-фильтра) — для global row.
    ask_global AS (
        SELECT
            percentile_cont(0.5) WITHIN GROUP (ORDER BY price_per_m2) AS ask_median,
            COUNT(*)                                                  AS n_listings
        FROM listings
        WHERE is_active
          -- #2656: то же окно свежести, что и в ask_side выше (см. комментарий там)
          -- — global-строка должна считаться по той же популяции, что per-bucket.
          AND scraped_at > NOW() - (:fresh_days || ' days')::interval
          AND rooms IS NOT NULL
          -- #2620 hardening: same area_m2 IS NOT NULL as ask_side — keeps the global-row
          -- population consistent with the per-bucket rows it's a fallback for.
          AND area_m2 IS NOT NULL
          AND price_per_m2 BETWEEN :ppm2_min AND :ppm2_max
          -- novostroyki guard (#1186): NULL = legacy вторичка до м.011
          AND (listing_segment IS NULL OR listing_segment = 'vtorichka')
          -- #2583 H2: тот же городской скоуп, что и ask_side выше (см. комментарий там
          -- про причину city IS NULL == "свой" и #2598/#2606).
          AND (city IS NULL OR city ILIKE :asking_city)
    ),
    -- Global fallback строка rooms_bucket=-1 (пишется всегда, если ask>0).
    global_row AS (
        SELECT
            -1                                        AS rooms_bucket,
            ''::text                                  AS district,
            (d.sold_median / a.ask_median)::numeric   AS ratio,
            round(d.sold_median)::bigint              AS sold_median,
            round(a.ask_median)::bigint               AS ask_median,
            d.n_deals::int                            AS n_deals,
            a.n_listings::int                         AS n_listings,
            12                                        AS window_months,
            'global_fallback'::text                   AS basis
        FROM deal_global d
        CROSS JOIN ask_global a
        WHERE a.ask_median IS NOT NULL
          AND a.ask_median > 0
          AND d.sold_median IS NOT NULL          -- защита от пустого окна сделок (иначе ratio=NULL)
          AND d.sold_median > 0
    )
    INSERT INTO asking_to_sold_ratios (
        rooms_bucket, district, ratio, sold_median, ask_median,
        n_deals, n_listings, window_months, basis, region_code
    )
    SELECT rooms_bucket, district, ratio, sold_median, ask_median,
           n_deals, n_listings, window_months, basis, 66 FROM global_row
    UNION ALL
    SELECT rooms_bucket, district, ratio, sold_median, ask_median,
           n_deals, n_listings, window_months, basis, 66 FROM per_bucket
    """
)

# ── Geography-matched per-region derivation (#3529) ───────────────────────────
# ПРОБЛЕМА (#3512-путь, прод-замер 2026-09 по региону 50): обе стороны фильтровались
# ТОЛЬКО по region_code и соединялись ТОЛЬКО по бакету комнат — т.е. sold-медиана и
# ask-медиана считались по РАЗНЫМ географическим популяциям одного региона.
# Разложение обл.50 по кольцам 10 км от центра Москвы (сделки 12 мес vs активные объявления):
#   20-30 км: 12 189 сделок / 23 292 объявления → 0.808
#   30-40 км:  4 132 / 10 206 → 0.847
#   50-60 км:  1 432 /  5 243 → 0.951
#   70-80 км:    150 /  2 294 → 0.688
# ВНУТРИ колец отношение 0.69-0.95, ближние кольца (76% сделок) — 0.81-0.85, а общий пул
# давал 0.891: объявления смещены к дальней дешёвой периферии СИЛЬНЕЕ, чем сделки. Это
# перекос СОСТАВА выборки, а не свойство рынка: выкупная цена по области системно завышена.
#
# РЕШЕНИЕ: считать коэффициент на СОГЛАСОВАННОЙ географии — обе стороны раскладываются
# по одним и тем же пространственным ячейкам, медианы берутся ВНУТРИ ячейки, и в итог
# идут только ячейки, где есть ОБЕ стороны, с весами по числу СДЕЛОК. Т.е. ask-сторона
# перевзвешивается на географию сделок (индекс Ласпейреса): ratio = Σ(w·sold) / Σ(w·ask),
# w = n_deals ячейки. Строка остаётся внутренне согласованной: ratio == sold_median/ask_median,
# где оба медианных столбца — взвешенные средние ячеечных медиан с ОДНИМИ весами.
#
# ПУТЬ ЕКБ (66) НЕ ТРОГАЕМ — там своя историческая калибровка городской квотой (#C2/#2583),
# числа региона 66 обязаны остаться byte-for-byte прежними (_REDERIVE_SQL выше).

# РАЗМЕР ЯЧЕЙКИ — регулярная сетка 0.1° широты × 0.2° долготы ≈ 11 км × 12-16 км на
# широтах 45-60°N (0.2° долготы × cos(lat): 15.7 км на 45°, 12.5 км на 55.7°, 11.1 км на 60°).
# Почему именно так:
#   • Масштаб взят от замера выше: именно на ~10-км разрешении отношение перестаёт
#     гулять от состава (внутри кольца 0.69-0.95 вместо 0.891 по пулу), при этом ячейка
#     ещё достаточно крупная, чтобы набрать десятки сделок и объявлений.
#   • Сетка, а НЕ кольца от центра: кольцам нужен центр, а у региона 50 своего
#     города-центра нет (его фактический центр — Москва, т.е. ДРУГОЙ регион), и каждый
#     следующий регион реестра потребовал бы своего анкора и своего шага. Сетке анкор не нужен.
#   • Совмещение по НАЗВАНИЮ муниципалитета НЕВОЗМОЖНО: listings.city у региона 50
#     пуста (3 строки из 70 996). geom есть с обеих сторон (объявления 70 996/70 996,
#     сделки 87 562/113 351) — выравниваем ПРОСТРАНСТВЕННО.
#   • FLOOR по градусам — чистая арифметика по ST_X/ST_Y, без репроекций и без стыковки
#     с админграницами, которых в БД нет. Точность границ ячейки здесь не важна — важно,
#     что ОБЕ стороны режутся ОДИНАКОВО.
_CELL_LAT_DEG: float = 0.1
_CELL_LON_DEG: float = 0.2

# Порог НА ЯЧЕЙКУ (все комнатности вместе) — сколько нужно, чтобы ячейка считалась
# покрытой ОБЕИМИ сторонами. Ниже глобального 30/30 НАМЕРЕННО: ячеечная медиана не
# публикуется сама по себе — она входит во взвешенную сумму, а публикуемый барьер
# остаётся прежним 30/30, но уже на СУММЕ по удержанным ячейкам (HAVING ниже).
_CELL_MIN_DEALS: int = 10
_CELL_MIN_LISTINGS: int = 10

# Порог на пару (ячейка, бакет комнат) — ещё мягче: внутри уже отобранной ячейки
# комнатность дробит выборку ещё на 5 частей. Меньше 5 наблюдений на сторону — медиана
# шум, и при большом весе этот шум попадёт в итоговую строку.
_CELL_BUCKET_MIN_DEALS: int = 5
_CELL_BUCKET_MIN_LISTINGS: int = 5

# ГАРДЫ ДЕГРАДАЦИИ (пункт 6 задачи): если согласованной географии по факту нет —
# лучше НЕ писать строку вообще (эстиматор деградирует явно, без коэффициента),
# чем посчитать неверно и выглядеть уверенно.
_MIN_MATCHED_CELLS: int = 3
# Доля СДЕЛОК (с geom), попавших в пересечение ячеек. Именно сделки — целевая популяция
# (на их географию перевзвешивается ask-сторона); объявления за пределами пересечения
# отбрасываются НАМЕРЕННО (это и есть фикс), поэтому гарда на них нет — только счётчик.
_MIN_DEAL_CELL_COVERAGE: float = 0.5
# Доля строк с geom на КАЖДОЙ стороне: если большая часть стороны без координат,
# выравнивать пространственно нечего — получился бы коэффициент по неслучайному остатку.
_MIN_GEOM_COVERAGE: float = 0.5
# Сигнальный (не блокирующий) порог: выше него пишется WARNING. 0.25 выбран чуть выше
# текущего прод-состояния региона 50 (25 789/113 351 = 22.8% сделок ждут геокодера),
# чтобы лог не шумел на норме, но ухудшение было видно сразу. Доля попадает в счётчики
# ВСЕГДА, независимо от порога — строки без geom не выпадают молча (пункт 3 задачи).
_GEOM_WARN_SHARE: float = 0.25

# ОБЩИЕ ФИЛЬТРЫ сторон — один источник правды для stats- и insert-запросов (иначе счётчики
# и деривация разъехались бы при первой же правке одного из них). Состав гардов тот же,
# что у ЕКБ-деривации (12-мес окно, ppm²-полоса, свежесть #2656, novostroyki #1186,
# area_m2 IS NOT NULL #2620) — меняется ТОЛЬКО гео-согласование.
_DEAL_FROM_REGION = """
        FROM deals
"""

_DEAL_WHERE_REGION = """
        WHERE source = 'rosreestr'
          AND rooms IS NOT NULL
          AND region_code = CAST(:region_code AS int)
          AND price_per_m2 BETWEEN :ppm2_min AND :ppm2_max
          AND deal_date >= CURRENT_DATE - INTERVAL '12 months'
"""

# #time-adjust: FROM и WHERE разведены — deal_geo (ниже) вставляет sber-JOIN'ы МЕЖДУ
# ними (JOIN обязан стоять до WHERE), deal_all (stats, без time-adjust) склеивает как раньше.
_DEAL_FROM_WHERE_REGION = _DEAL_FROM_REGION + _DEAL_WHERE_REGION

_ASK_FROM_WHERE_REGION = """
        FROM listings
        WHERE is_active
          AND scraped_at > NOW() - (:fresh_days || ' days')::interval
          AND rooms IS NOT NULL
          AND area_m2 IS NOT NULL
          AND price_per_m2 BETWEEN :ppm2_min AND :ppm2_max
          AND (listing_segment IS NULL OR listing_segment = 'vtorichka')
          AND region_code = CAST(:region_code AS int)
"""

# Ячеечные CTE — ОБЩИЕ для stats-запроса (счётчики + гард) и для самой деривации,
# чтобы решение «писать / не писать» принималось РОВНО по тем ячейкам, которые потом считаются.
_CELL_CTES_REGION = f"""{_SBER_FACTOR_CTES},
    -- #time-adjust: price_per_m2 домножен на sber-фактор ДО попадания в deal_cell/
    -- deal_cell_bucket — приведение проезжает через весь взвешенный (Ласпейрес) расчёт
    -- региона автоматически, отдельно трогать deal_cell/per_bucket не нужно.
    deal_geo AS (
        SELECT
            FLOOR(ST_Y(geom) / {_CELL_LAT_DEG})                            AS cell_lat,
            FLOOR(ST_X(geom) / {_CELL_LON_DEG})                            AS cell_lon,
            LEAST(GREATEST(rooms, 0), 4)                                   AS rooms_bucket,
            price_per_m2 * ({_SBER_FACTOR_EXPR})                           AS price_per_m2
        {_DEAL_FROM_REGION}
        {_SBER_FACTOR_JOINS}
        {_DEAL_WHERE_REGION}          AND geom IS NOT NULL
    ),
    ask_geo AS (
        SELECT
            FLOOR(ST_Y(geom) / {_CELL_LAT_DEG})                            AS cell_lat,
            FLOOR(ST_X(geom) / {_CELL_LON_DEG})                            AS cell_lon,
            {_AREA_ROOMS_BUCKET_SQL}                                       AS rooms_bucket,
            price_per_m2
        {_ASK_FROM_WHERE_REGION}          AND geom IS NOT NULL
    ),
    deal_cell AS (
        SELECT
            cell_lat, cell_lon,
            percentile_cont(0.5) WITHIN GROUP (ORDER BY price_per_m2)      AS sold_median,
            COUNT(*)                                                       AS n_deals
        FROM deal_geo
        GROUP BY cell_lat, cell_lon
    ),
    ask_cell AS (
        SELECT
            cell_lat, cell_lon,
            percentile_cont(0.5) WITHIN GROUP (ORDER BY price_per_m2)      AS ask_median,
            COUNT(*)                                                       AS n_listings
        FROM ask_geo
        GROUP BY cell_lat, cell_lon
    ),
    -- СОГЛАСОВАННАЯ ГЕОГРАФИЯ: ячейки, где ОБЕ стороны имеют свою массу.
    matched_cell AS (
        SELECT
            d.cell_lat, d.cell_lon,
            d.sold_median, d.n_deals,
            a.ask_median, a.n_listings
        FROM deal_cell d
        JOIN ask_cell a USING (cell_lat, cell_lon)
        WHERE d.n_deals >= {_CELL_MIN_DEALS}
          AND a.n_listings >= {_CELL_MIN_LISTINGS}
          AND d.sold_median IS NOT NULL AND d.sold_median > 0
          AND a.ask_median IS NOT NULL AND a.ask_median > 0
    )"""

# Статистика СОСТАВА выборки — считается ДО деривации и решает, писать ли регион вообще.
# Строки БЕЗ geom тоже считаются (n_all vs n_geo) — они выпадают из деривации, и это
# должно быть видно в счётчиках, а не тихо (пункт 3 задачи).
_GEO_STATS_SQL_REGION = text(
    f"""
    WITH{_CELL_CTES_REGION},
    deal_all AS (
        SELECT
            COUNT(*)                                        AS n_all,
            COUNT(*) FILTER (WHERE geom IS NOT NULL)        AS n_geo
        {_DEAL_FROM_WHERE_REGION}    ),
    ask_all AS (
        SELECT
            COUNT(*)                                        AS n_all,
            COUNT(*) FILTER (WHERE geom IS NOT NULL)        AS n_geo
        {_ASK_FROM_WHERE_REGION}    )
    SELECT
        da.n_all                                            AS deals_total,
        da.n_geo                                            AS deals_geo,
        aa.n_all                                            AS listings_total,
        aa.n_geo                                            AS listings_geo,
        (SELECT COUNT(*) FROM deal_cell)                    AS cells_deal,
        (SELECT COUNT(*) FROM ask_cell)                     AS cells_ask,
        (SELECT COUNT(*) FROM deal_cell d
             JOIN ask_cell a USING (cell_lat, cell_lon))    AS cells_both_sides,
        (SELECT COUNT(*) FROM matched_cell)                 AS cells_matched,
        (SELECT COALESCE(SUM(n_deals), 0) FROM matched_cell)    AS deals_in_cells,
        (SELECT COALESCE(SUM(n_listings), 0) FROM matched_cell) AS listings_in_cells
    FROM deal_all da CROSS JOIN ask_all aa
"""
)

# Деривация на согласованной географии. Отличий от ЕКБ-пути (_REDERIVE_SQL) ровно два:
#   1. городская квота ЕКБ → симметричный region_code на обеих сторонах (#3512);
#   2. медианы считаются ВНУТРИ ячейки и агрегируются с весами по числу сделок (#3529).
# Окно 12 мес, ppm²-полоса, area-бакет ask-стороны, порог 30/30 на публикуемую строку — прежние.
_REDERIVE_SQL_REGION = text(
    f"""
    WITH{_CELL_CTES_REGION},
    -- Внутри УЖЕ отобранных ячеек — разрез по бакету комнат (обе стороны — тот же набор
    -- ячеек, т.е. гео-ключ есть И в фильтре, И в соединении — в отличие от старого
    -- `JOIN ... USING (rooms_bucket)`, где географии в соединении не было вообще).
    deal_cell_bucket AS (
        SELECT
            g.cell_lat, g.cell_lon, g.rooms_bucket,
            percentile_cont(0.5) WITHIN GROUP (ORDER BY g.price_per_m2)    AS sold_median,
            COUNT(*)                                                       AS n_deals
        FROM deal_geo g
        JOIN matched_cell m USING (cell_lat, cell_lon)
        GROUP BY g.cell_lat, g.cell_lon, g.rooms_bucket
    ),
    ask_cell_bucket AS (
        SELECT
            g.cell_lat, g.cell_lon, g.rooms_bucket,
            percentile_cont(0.5) WITHIN GROUP (ORDER BY g.price_per_m2)    AS ask_median,
            COUNT(*)                                                       AS n_listings
        FROM ask_geo g
        JOIN matched_cell m USING (cell_lat, cell_lon)
        GROUP BY g.cell_lat, g.cell_lon, g.rooms_bucket
    ),
    bucket_cell AS (
        SELECT
            d.rooms_bucket,
            d.sold_median, d.n_deals,
            a.ask_median, a.n_listings
        FROM deal_cell_bucket d
        JOIN ask_cell_bucket a USING (cell_lat, cell_lon, rooms_bucket)
        WHERE d.n_deals >= {_CELL_BUCKET_MIN_DEALS}
          AND a.n_listings >= {_CELL_BUCKET_MIN_LISTINGS}
          AND d.sold_median IS NOT NULL AND d.sold_median > 0
          AND a.ask_median IS NOT NULL AND a.ask_median > 0
    ),
    -- Взвешивание по числу СДЕЛОК: ask-сторона приводится к географии сделок.
    -- ratio == sold_median/ask_median построчно (оба — взвешенные средние с ОДНИМИ весами),
    -- так что публикуемые столбцы остаются взаимно согласованными.
    per_bucket AS (
        SELECT
            rooms_bucket,
            ''::text                                                       AS district,
            CAST(SUM(sold_median * n_deals) / SUM(ask_median * n_deals)
                 AS numeric)                                               AS ratio,
            round(SUM(sold_median * n_deals) / SUM(n_deals))::bigint       AS sold_median,
            round(SUM(ask_median * n_deals) / SUM(n_deals))::bigint        AS ask_median,
            SUM(n_deals)::int                                              AS n_deals,
            SUM(n_listings)::int                                           AS n_listings,
            12                                                             AS window_months,
            'per_rooms'::text                                              AS basis
        FROM bucket_cell
        GROUP BY rooms_bucket
        -- ТОТ ЖЕ публикуемый барьер 30/30, что и раньше — теперь на сумме по ячейкам.
        HAVING SUM(n_deals) >= 30
           AND SUM(n_listings) >= 30
           AND SUM(ask_median * n_deals) > 0
    ),
    -- Global -1 fallback — те же ячейки, но без разреза по комнатности.
    global_row AS (
        SELECT
            -1                                                             AS rooms_bucket,
            ''::text                                                       AS district,
            CAST(SUM(sold_median * n_deals) / SUM(ask_median * n_deals)
                 AS numeric)                                               AS ratio,
            round(SUM(sold_median * n_deals) / SUM(n_deals))::bigint       AS sold_median,
            round(SUM(ask_median * n_deals) / SUM(n_deals))::bigint        AS ask_median,
            SUM(n_deals)::int                                              AS n_deals,
            SUM(n_listings)::int                                           AS n_listings,
            12                                                             AS window_months,
            'global_fallback'::text                                        AS basis
        FROM matched_cell
        HAVING SUM(n_deals) > 0
           AND SUM(ask_median * n_deals) > 0
    )
    INSERT INTO asking_to_sold_ratios (
        rooms_bucket, district, ratio, sold_median, ask_median,
        n_deals, n_listings, window_months, basis, region_code
    )
    SELECT rooms_bucket, district, ratio, sold_median, ask_median,
           n_deals, n_listings, window_months, basis, CAST(:region_code AS int) FROM global_row
    UNION ALL
    SELECT rooms_bucket, district, ratio, sold_median, ask_median,
           n_deals, n_listings, window_months, basis, CAST(:region_code AS int) FROM per_bucket
"""
)

# КОЛОНКА district (#647-слот) ОСТАЁТСЯ ПУСТОЙ и здесь. Идентификатор ячейки в неё не
# ложится: ячейка — ПРОМЕЖУТОЧНАЯ единица расчёта, а не единица публикации. На выходе
# по-прежнему одна строка на (регион, бакет) — агрегат по всем ячейкам; записать в
# district «какую-то одну» ячейку было бы враньём, а писать строку НА ЯЧЕЙКУ нельзя:
# потребитель (estimator._get_asking_sold_ratio) читает строго `district = ''` и
# ключевать оценку по гео-ячейке пока не умеет — это отдельная задача #647.


def _pct(part: float, whole: float) -> int:
    """Доля part/whole в ЦЕЛЫХ процентах (счётчики scrape_runs — dict[str, int])."""
    if whole <= 0:
        return 0
    return round(100.0 * part / whole)


def geo_region_verdict(region_code: int, stats: dict[str, int] | None) -> tuple[bool, str]:
    """Писать ли строки региона по согласованной географии (пункт 6 — явная деградация).

    Возвращает (ok, reason). ok=False — регион НЕ получает НИ ОДНОЙ строки (старые всё
    равно удалены), и эстиматор честно остаётся без коэффициента вместо неверного.
    Чистая функция от строки stats — тестируется без базы.
    """
    if not stats:
        return False, "geo-stats не вернулись"
    deals_total = int(stats.get("deals_total") or 0)
    deals_geo = int(stats.get("deals_geo") or 0)
    listings_total = int(stats.get("listings_total") or 0)
    listings_geo = int(stats.get("listings_geo") or 0)
    cells_matched = int(stats.get("cells_matched") or 0)
    deals_in_cells = int(stats.get("deals_in_cells") or 0)

    if deals_total == 0 or listings_total == 0:
        return False, f"нет данных: deals={deals_total} listings={listings_total}"
    if deals_geo / deals_total < _MIN_GEOM_COVERAGE:
        return False, (
            f"сделки без geom: {_pct(deals_total - deals_geo, deals_total)}% "
            f"(порог покрытия {_MIN_GEOM_COVERAGE:.0%})"
        )
    if listings_geo / listings_total < _MIN_GEOM_COVERAGE:
        return False, (
            f"объявления без geom: {_pct(listings_total - listings_geo, listings_total)}% "
            f"(порог покрытия {_MIN_GEOM_COVERAGE:.0%})"
        )
    if cells_matched < _MIN_MATCHED_CELLS:
        return False, (
            f"ячеек с обеими сторонами {cells_matched} < {_MIN_MATCHED_CELLS} "
            f"(географии сделок и объявлений практически не пересекаются)"
        )
    if deals_geo > 0 and deals_in_cells / deals_geo < _MIN_DEAL_CELL_COVERAGE:
        return False, (
            f"в пересечение ячеек попало {_pct(deals_in_cells, deals_geo)}% сделок "
            f"(порог {_MIN_DEAL_CELL_COVERAGE:.0%})"
        )
    _ = region_code
    return True, "ok"


def _geo_region_counters(region_code: int, stats: dict[str, int] | None) -> dict[str, int]:
    """Порегионные счётчики состава выборки (пункты 3 и 5 задачи), всё — int."""
    s = stats or {}
    deals_total = int(s.get("deals_total") or 0)
    deals_geo = int(s.get("deals_geo") or 0)
    listings_total = int(s.get("listings_total") or 0)
    listings_geo = int(s.get("listings_geo") or 0)
    deals_in_cells = int(s.get("deals_in_cells") or 0)
    listings_in_cells = int(s.get("listings_in_cells") or 0)
    cells_matched = int(s.get("cells_matched") or 0)
    cells_both = int(s.get("cells_both_sides") or 0)
    p = f"geo_r{region_code}_"
    return {
        p + "cells_deal": int(s.get("cells_deal") or 0),
        p + "cells_ask": int(s.get("cells_ask") or 0),
        p + "cells_matched": cells_matched,
        # Ячейки, где есть обе стороны, но одна из них тоньше порога ячейки.
        p + "cells_dropped": max(cells_both - cells_matched, 0),
        # Сколько массы осталось ЗА пределами пересечения (от строк с geom).
        p + "deals_outside_pct": _pct(deals_geo - deals_in_cells, deals_geo),
        p + "listings_outside_pct": _pct(listings_geo - listings_in_cells, listings_geo),
        # Строки без координат — не выпадают молча (пункт 3).
        p + "deals_no_geom_pct": _pct(deals_total - deals_geo, deals_total),
        p + "listings_no_geom_pct": _pct(listings_total - listings_geo, listings_total),
    }


# ── Post-insert counters ──────────────────────────────────────────────────────
# Считываем итог из таблицы (всё ещё в той же транзакции — до commit): сколько строк
# записано всего, сколько per_rooms, был ли использован global -1 fallback.
# #3512: считается ОДИН раз в самом конце, ПОСЛЕ всех регионов (ЕКБ + цикл по
# _OTHER_REGION_CODES) — district='' покрывает все регионы разом, счётчики суммарные.
_COUNTERS_SQL = text(
    """
    SELECT
        COUNT(*)                                              AS rows_written,
        COUNT(*) FILTER (WHERE basis = 'per_rooms')           AS per_rooms_rows,
        COUNT(*) FILTER (WHERE rooms_bucket = -1)             AS used_global_fallback
    FROM asking_to_sold_ratios
    WHERE district = ''
    """
)


def recompute_asking_to_sold_ratios(db: Session, run_id: int) -> dict[str, int]:
    """Пересчитать asking_to_sold_ratios per region (#3512, TRUE-MIRROR refresh #648 Stage 4).

    Sync (вызывается scheduler-триггером в executor, как snapshot_listing_sources).
    В ОДНОЙ транзакции (атомарно — таблица никогда не пуста mid-refresh для уже
    посчитанных регионов):
      1. Регион 66 (ЕКБ): DELETE (region_code=66, district='') → 080-derivation
         INSERT...SELECT (byte-for-byte прежняя логика — городская квота ЕКБ, #C2/#2583).
      2. Каждый прочий регион реестра app.services.regions (минус 66): тот же
         DELETE/INSERT цикл, но derivation скоупится по `region_code` симметрично
         на sold- и asking-стороне (без городской квоты — она не применима вне
         ЕКБ, см. комментарий у generic-derivation SQL ниже).
    Регион без своих ДКП-сделок ИЛИ без активных listings просто не получает строк
    (global_row/per_bucket WHERE-гарды отфильтровывают NULL-медианы) — это НЕ
    ошибка, а честный признак «данных пока недостаточно», ловится потребителем
    (estimator._get_asking_sold_ratio) через отсутствие строки → явная деградация.
    Затем ОДИН общий counters-запрос по всей таблице, commit, mark_done.

    Финализирует scrape_runs (mark_done / mark_failed) и пишет counters.

    LIMITATION (#2620, честно задокументировано — не гард, а факт данных): sold-сторона
    (deals) НЕ имеет маркера новостройка/вторичка — Росреестр таким свойством ДКП не
    делится, а listing_segment (гард #1186) существует только у listings. ask_side/ask_global
    отфильтрованы на вторичку, deal_side/deal_global — нет. Замер на проде (2026-08, #2620):
    доля сделок с year_built >= 2020 (грубый прокси новостройки) — 44.1% в бакете «4+» против
    23.1% в бакетах 1-3 — заметный перекос, но year_built НЕ идентифицирует первичку/вторичку
    (продажа квартиры 2021 года постройки в 2026м — легитимная вторичка), поэтому фильтр по
    году НЕ добавлен (создал бы новую, столь же спекулятивную асимметрию). Area-бакет-фикс
    ниже (см. _AREA_ROOMS_BUCKET_SQL) сам по себе убрал инверсию ratio>1 в бакете «4+»
    (0.8315 на замере прод-данных 2026-08, было 1.0257) — снятие миграции между бакетами было
    root cause, а не новостройки.

    Returns {"rows_written": N, "per_rooms_rows": M, "used_global_fallback": 0|1}.
    """
    counters: dict[str, int] = {
        "rows_written": 0,
        "per_rooms_rows": 0,
        "used_global_fallback": 0,
        # #3529: сколько регионов посчитано по согласованной географии, а сколько
        # деградировало явно (строк нет → эстиматор без коэффициента).
        "geo_regions_written": 0,
        "geo_regions_skipped": 0,
        # #time-adjust: сколько регионов реально получили sber-приведение SOLD-стороны
        # (серия найдена) vs посчитаны с factor=1.0 (флаг выключен ИЛИ серии нет вовсе).
        "sber_time_adjust_regions_applied": 0,
        "sber_time_adjust_regions_missing_series": 0,
    }
    # #time-adjust: ленивый импорт — estimator.py импортирует area_bucket ИЗ этого модуля
    # на верхнем уровне, top-level импорт в обратную сторону дал бы цикл (см. комментарий
    # у _SBER_FACTOR_CTES выше).
    from app.services.estimator import (
        SBER_COEFF_DASHBOARDS,
        SBER_TIME_FACTOR_MAX,
        SBER_TIME_FACTOR_MIN,
        sber_region_series_name,
    )

    sber_dashboards = list(SBER_COEFF_DASHBOARDS)
    time_adjust_enabled = bool(settings.asking_ratio_time_adjust_enabled)

    def _sber_params(sber_city: str) -> dict[str, object]:
        return {
            "sber_city": sber_city,
            "sber_dashboards": sber_dashboards,
            "time_adjust_enabled": time_adjust_enabled,
            "factor_min": SBER_TIME_FACTOR_MIN,
            "factor_max": SBER_TIME_FACTOR_MAX,
        }

    def _check_sber_series(region_code: int, sber_city: str) -> None:
        """Пункт 3 задачи: если серии для города вообще нет — не молча, счётчик+лог."""
        if not time_adjust_enabled:
            return
        try:
            row = (
                db.execute(
                    text(
                        """
                        SELECT COUNT(*) AS n FROM sber_price_index
                        WHERE city = CAST(:sber_city AS text)
                          AND dashboard = ANY(CAST(:sber_dashboards AS text[]))
                        """
                    ),
                    {"sber_city": sber_city, "sber_dashboards": sber_dashboards},
                )
                .mappings()
                .first()
            )
        except Exception as exc:  # pragma: no cover — defensive, graceful
            logger.warning("sber series presence-check failed (graceful): %s", exc)
            row = None
        n = int(row["n"]) if row and row.get("n") is not None else 0
        if n:
            counters["sber_time_adjust_regions_applied"] += 1
        else:
            counters["sber_time_adjust_regions_missing_series"] += 1
            logger.warning(
                "asking_to_sold_ratio region_code=%d: sber_price_index серии нет "
                "(city=%s) — SOLD-сторона считается БЕЗ time-adjust (factor=1.0)",
                region_code,
                sber_city,
            )

    try:
        # DELETE + re-derive INSERT в одной транзакции (НЕ коммитим между ними —
        # таблица не должна остаться пустой, если INSERT упадёт). Регион 66 —
        # прежняя ЕКБ-деривация байт-в-байт; остальные регионы — цикл ниже (#3512).
        ekb_sber_city = sber_region_series_name(66)
        _check_sber_series(66, ekb_sber_city)
        db.execute(_DELETE_SQL)
        db.execute(
            _REDERIVE_SQL,
            {
                "ppm2_min": _PPM2_MIN,
                "ppm2_max": settings.asking_ratio_ppm2_max,
                "asking_city": _ASKING_CITY_PATTERN,
                "fresh_days": LISTINGS_FRESH_DAYS,
                **_sber_params(ekb_sber_city),
            },
        )
        for region_code in _OTHER_REGION_CODES:
            region_sber_city = sber_region_series_name(region_code)
            _check_sber_series(region_code, region_sber_city)
            params = {
                "region_code": region_code,
                "ppm2_min": _PPM2_MIN,
                "ppm2_max": settings.asking_ratio_ppm2_max,
                "fresh_days": LISTINGS_FRESH_DAYS,
                **_sber_params(region_sber_city),
            }
            # Сначала состав выборки (#3529) — он же решает, писать ли регион вообще.
            stats_row = db.execute(_GEO_STATS_SQL_REGION, params).mappings().first()
            stats = dict(stats_row) if stats_row is not None else None
            counters.update(_geo_region_counters(region_code, stats))
            ok, reason = geo_region_verdict(region_code, stats)

            # DELETE идёт В ЛЮБОМ случае: если согласованной географии больше нет, старый
            # (считанный по пулу) коэффициент тем более не должен оставаться в таблице.
            db.execute(_DELETE_SQL_REGION, {"region_code": region_code})

            no_geom_deals = counters.get(f"geo_r{region_code}_deals_no_geom_pct", 0)
            no_geom_listings = counters.get(f"geo_r{region_code}_listings_no_geom_pct", 0)
            if max(no_geom_deals, no_geom_listings) >= int(_GEOM_WARN_SHARE * 100):
                # Пункт 3: строки без координат не выпадают молча — это сигнал.
                logger.warning(
                    "asking_to_sold_ratio region_code=%d: без geom сделок %d%%, "
                    "объявлений %d%% — гео-согласование считается по остатку",
                    region_code,
                    no_geom_deals,
                    no_geom_listings,
                )

            if not ok:
                counters["geo_regions_skipped"] += 1
                counters[f"geo_r{region_code}_skipped"] = 1
                logger.warning(
                    "asking_to_sold_ratio region_code=%d: СТРОКИ НЕ ПИШУТСЯ — %s. "
                    "Оценка останется без коэффициента (явная деградация)",
                    region_code,
                    reason,
                )
                continue

            counters[f"geo_r{region_code}_skipped"] = 0
            counters["geo_regions_written"] += 1
            db.execute(_REDERIVE_SQL_REGION, params)
            logger.info(
                "asking_to_sold_ratio region_code=%d: ячеек с обеими сторонами %d "
                "(отброшено по порогу %d), вне пересечения: сделок %d%%, объявлений %d%%",
                region_code,
                counters.get(f"geo_r{region_code}_cells_matched", 0),
                counters.get(f"geo_r{region_code}_cells_dropped", 0),
                counters.get(f"geo_r{region_code}_deals_outside_pct", 0),
                counters.get(f"geo_r{region_code}_listings_outside_pct", 0),
            )

        row = db.execute(_COUNTERS_SQL).mappings().first()
        if row is not None:
            counters["rows_written"] = int(row["rows_written"] or 0)
            counters["per_rooms_rows"] = int(row["per_rooms_rows"] or 0)
            counters["used_global_fallback"] = int(row["used_global_fallback"] or 0)

        db.commit()
        runs_mod.mark_done(db, run_id, counters)
        logger.info(
            "recompute_asking_to_sold_ratios run_id=%d done: "
            "rows_written=%d per_rooms_rows=%d used_global_fallback=%d",
            run_id,
            counters["rows_written"],
            counters["per_rooms_rows"],
            counters["used_global_fallback"],
        )
        return counters
    except Exception as exc:
        logger.exception("recompute_asking_to_sold_ratios run_id=%d failed", run_id)
        db.rollback()
        runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters)
        raise