"""Анализ лучших планировок конкурентов по velocity (Issue #113 Phase 2.1).

Источники:
  cad_parcels_geom / cad_quarters_geom — центроид участка
  domrf_kn_objects                     — ЖК в радиусе (latitude/longitude → geography)
  objective_corpus_room_month          — ежемесячные сделки по (project_name, room_bucket)
  objective_complex_mapping            — domrf_obj_id ↔ objective_complex_name
  domrf_kn_flats                       — supply count по (room_bucket, area_bin)

Алгоритм:
  Step 1: центроид участка (cad_parcels_geom → cad_quarters_geom fallback).
  Step 2: obj_id конкурентов в радиусе (domrf_kn_objects + фильтры).
  Step 3: inline SQL из objective_corpus_room_month с честным WHERE report_month фильтром.
  Step 4: velocity_per_month = deals_window / months_in_window (честный time_window).
  Step 5: supply side из domrf_kn_flats — один батч-запрос.
  Step 6: per-row signature + sold_pct.
  Step 7: фильтр min_velocity + sort + rank.
  Step 8: build recommendation_for_tz (unit-mix, price, rationale).
  Step 9: data_quality (coverage + confidence).

Fix SF-01: раньше mv_layout_velocity (24 мес) делился на divisor (4/12) — данные
не менялись при смене time_window. Теперь inline SQL с реальным фильтром report_month.
"""

from __future__ import annotations

import datetime as dt
import logging
from typing import Any

from sqlalchemy import text
from sqlalchemy.orm import Session

from app.schemas.parcel import (
    BestLayoutsRequest,
    BestLayoutsResponse,
    LayoutDataQuality,
    LayoutTzMixRow,
    LayoutTzRecommendation,
    TopLayoutRow,
)
from app.services.site_finder.layout_signature import area_bin, layout_signature

logger = logging.getLogger(__name__)

# Confidence thresholds (per coverage % of objects with MV velocity data)
# Tune via PR if business feedback требует.
LAYOUT_CONFIDENCE_HIGH_PCT = 50.0
LAYOUT_CONFIDENCE_MEDIUM_PCT = 20.0

# Fix SF-09: cap доминирующего bucket чтобы рекомендация не зеркалила перекос рынка.
# Избыток перераспределяется пропорционально остальным bucket'ам.
MAX_BUCKET_SHARE_PCT = 35

# Параметры time_window: (PostgreSQL interval string, months divisor для velocity_per_month).
# Используются в _INLINE_VELOCITY_SQL — реальный фильтр по report_month.
# Fix SF-01: убраны _VELOCITY_DIVISORS, которые делили MV (24 мес) без изменения данных.
_TIME_WINDOW_PARAMS: dict[str, tuple[str, float]] = {
    "last_month": ("1 month", 1.0),
    "last_quarter": ("3 months", 3.0),
    "last_year": ("12 months", 12.0),
}

# ── SQL: центроид участка ─────────────────────────────────────────────────────

_PARCEL_CENTROID_SQL = text("""
    SELECT ST_X(pt) AS center_lon,
           ST_Y(pt) AS center_lat
    FROM (
        SELECT ST_Centroid(geom) AS pt
          FROM cad_parcels_geom
         WHERE cad_num = :cad_num AND geom IS NOT NULL
        UNION ALL
        SELECT ST_Centroid(geom) AS pt
          FROM cad_quarters_geom
         WHERE cad_number = :quarter AND geom IS NOT NULL
    ) sub
    LIMIT 1
""")

# ── SQL: obj_id конкурентов в радиусе ─────────────────────────────────────────
# Геометрия domrf_kn_objects вычисляется on-the-fly из (latitude, longitude)
# как ST_SetSRID(ST_MakePoint(longitude, latitude), 4326)::geography
# (consistency с competitors.py).
# obj_class_filter: NULL = все классы.
# filter_competitor_obj_ids: NULL = не фильтровать по списку.

_COMPETITORS_IN_RADIUS_SQL = text("""
    SELECT DISTINCT ON (obj_id) obj_id
    FROM domrf_kn_objects
    WHERE latitude IS NOT NULL AND longitude IS NOT NULL
      AND ST_DWithin(
          ST_SetSRID(ST_MakePoint(longitude, latitude), 4326)::geography,
          ST_SetSRID(
              ST_MakePoint(CAST(:center_lon AS float), CAST(:center_lat AS float)),
              4326
          )::geography,
          CAST(:radius_m AS float)
      )
      AND (
          CAST(:obj_class_filter AS text) IS NULL
          OR obj_class = CAST(:obj_class_filter AS text)
      )
    ORDER BY obj_id, snapshot_date DESC NULLS LAST
""")

# ── SQL: inline velocity из objective_corpus_room_month + mapping ─────────────
# Fix SF-01: честный фильтр по report_month вместо деления mv_layout_velocity (24 мес).
# Параметры:
#   :window_interval — PostgreSQL interval string ('1 month', '3 months', '12 months')
#   :competitor_obj_ids — list[int] obj_id конкурентов в радиусе
# CAST(:window_interval AS interval) — psycopg v3 / SQLAlchemy 2.0 safe (не ::interval).

#
# Fix #1391: fan-out по objective_complex_mapping завышал SUM-метрики. UNIQUE-ключ
# mapping — (objective_complex_name, objective_group), поэтому один project_name может
# матчиться на N mapping-строк (разные группы / domrf_obj_id). Прямой
# JOIN crm × cm дублировал каждую crm-строку N раз → SUM(deals_total_count),
# weighted-avg числители/знаменатели и deals_window раздувались в N раз. DISTINCT-метрики
# (competitor_obj_ids / competitor_count) не страдали, но SUM — да.
# Решение: deals агрегируем per (project_name, room_bucket) в CTE `crm_agg`
# (без mapping-джойна → fan-out невозможен). Список project_name'ов в радиусе берём из
# CTE `mapped_projects` (DISTINCT по objective_complex_name) и джойним один-к-одному —
# каждый project_name вносит свои сделки ровно один раз. competitor_obj_ids /
# competitor_count считаем отдельно в `obj_per_bucket` (через unnest mapping-obj_id),
# чтобы fan-out по obj_id не попал в SUM-агрегаты, и приджойниваем по room_bucket.

_INLINE_VELOCITY_SQL = text("""
    WITH crm_agg AS (
        SELECT
            crm.project_name,
            CASE
                WHEN crm.room_bucket = 'студия' THEN 'studio'
                ELSE crm.room_bucket
            END                                              AS room_bucket,
            SUM(crm.deals_total_count)                       AS deals_window,
            SUM(crm.deals_total_avg_area_m2 * crm.deals_total_count)
                                                             AS area_weighted_sum,
            SUM(crm.deals_total_avg_price_thousand_rub_per_m2 * crm.deals_total_count)
                                                             AS price_weighted_sum,
            MIN(crm.report_month)                            AS window_start,
            MAX(crm.report_month)                            AS window_end
        FROM objective_corpus_room_month crm
        WHERE crm.report_month >= (NOW() - CAST(:window_interval AS interval))::date
          AND crm.room_bucket IS NOT NULL
        GROUP BY crm.project_name,
            CASE
                WHEN crm.room_bucket = 'студия' THEN 'studio'
                ELSE crm.room_bucket
            END
    ),
    mapped_projects AS (
        SELECT DISTINCT cm.objective_complex_name            AS project_name
        FROM objective_complex_mapping cm
        WHERE cm.domrf_obj_id = ANY(:competitor_obj_ids)
    ),
    deals_per_bucket AS (
        SELECT
            a.room_bucket,
            SUM(a.deals_window)                              AS deals_window,
            COALESCE(
                SUM(a.area_weighted_sum)
                / NULLIF(SUM(a.deals_window), 0),
                0
            )::numeric(10, 2)                                AS avg_area_m2,
            COALESCE(
                SUM(a.price_weighted_sum)
                / NULLIF(SUM(a.deals_window), 0),
                0
            )::numeric(12, 2) * 1000.0                       AS avg_price_per_m2_rub,
            MIN(a.window_start)                              AS window_start,
            MAX(a.window_end)                                AS window_end
        FROM crm_agg a
        JOIN mapped_projects mp ON mp.project_name = a.project_name
        GROUP BY a.room_bucket
    ),
    obj_per_bucket AS (
        SELECT
            a.room_bucket,
            array_agg(DISTINCT cm.domrf_obj_id)              AS competitor_obj_ids,
            COUNT(DISTINCT cm.domrf_obj_id)                  AS competitor_count
        FROM crm_agg a
        JOIN objective_complex_mapping cm
            ON cm.objective_complex_name = a.project_name
        WHERE cm.domrf_obj_id = ANY(:competitor_obj_ids)
        GROUP BY a.room_bucket
    )
    SELECT
        d.room_bucket                                        AS room_bucket,
        d.deals_window                                       AS deals_window,
        d.avg_area_m2                                        AS avg_area_m2,
        d.avg_price_per_m2_rub                               AS avg_price_per_m2_rub,
        o.competitor_obj_ids                                 AS competitor_obj_ids,
        o.competitor_count                                   AS competitor_count,
        d.window_start                                       AS window_start,
        d.window_end                                         AS window_end
    FROM deals_per_bucket d
    JOIN obj_per_bucket o ON o.room_bucket = d.room_bucket
""")

# ── SQL: supply по (room_bucket, area_bin) за последний снимок ───────────────
# Один батч-запрос вместо N — возвращает map (rb, ab) → count.
# room_bucket и area_bin вычисляются в SQL аналогично layout_signature.py.
# Fix #1229: bucket'ы нормализованы под velocity-вокабуляр objective_corpus_room_month
# = {studio, 1, 2, 3, 4+}. Раньше supply отдельно отдавал euro-1/euro-2 для rooms=2
# (area<35 / area<50) — эти ключи никогда не матчились с velocity (Объектив
# euro-форматов не выделяет, см. _INLINE_VELOCITY_SQL passthrough room_bucket).
# Последствия: 1) supply '2' искусственно занижен (rooms=2 + area<50 уходили в
# euro-*) → sold_pct/is_oversold двушек завышены; 2) euro-* строки supply
# становились dead lookup. SF-08 euro-биннинг откладываем до момента, когда его
# начнёт отдавать velocity-сторона.

# Perf + CORRECTNESS fix (perf/best-layouts-objects-first):
# objects-first MATERIALIZED CTE + устранение fan-out по снимкам объекта.
#
# === Корректность (главное) ===
# domrf_kn_objects — это snapshot-таблица: UNIQUE (obj_id, snapshot_date)
# [uq_kn_objects_id_snap], ~8 снимков на obj_id, координаты могут меняться между
# снимками. Прежний запрос джойнил flats к domrf_kn_objects по obj_id БЕЗ фильтра
# o.snapshot_date → каждая квартира считалась по разу на КАЖДУЮ строку-объект,
# прошедшую ST_DWithin (fan-out ~×8.5 на проде, радиус 1.5км: 82251 → 9612).
# Последствие в выдаче (best_layouts.py): supply_units_in_radius завышен ~8.5×,
# sold_pct_of_supply (= deals/supply) занижен ~8.5×, is_oversold недосрабатывает.
# Фикс: latest_obj дедуплицирует до ОДНОЙ строки на obj_id (последний снимок —
# самые свежие координаты) через DISTINCT ON; ST_DWithin применяется к этим
# каноническим координатам. nearby содержит уникальные obj_id → JOIN с flats не
# фанаутит → units = ровно один счёт на квартиру (cross-check на проде:
# units == count(DISTINCT f.id) == 9612 для 65 объектов в радиусе).
#
# === Перформанс ===
# Селективный предикат — ST_DWithin по маленькой таблице объектов (~1.5k уникальных
# obj_id), поэтому фильтруем объекты ПЕРВЫМИ, затем джойним большую flats
# (последний полный снимок ~376k строк) через idx_kn_flats_obj. Прежний flats-first
# план параллельно seq-сканировал весь снимок flats → ~28k buffers / ~240-360ms
# (cold ~549ms). Этот план: objects-first + меньше строк после дедупа.
#
# MATERIALIZED у nearby — обязателен: без него планировщик инлайнит CTE и
# возвращается к flats-first seq scan (domrf_kn_objects не имеет GIST-индекса).
#
# === PER-OBJECT snapshot fix (#1956, regression от #1944) ===
# domrf_kn_flats — это ПО-ОБЪЕКТНЫЙ time-series: каждый ЖК скрейпится в СВОЙ день,
# UNIQUE-снимок (obj_id, snapshot_date). Прежний джойн фильтровал flats по ОДНОЙ
# глобальной дате `f.snapshot_date = MAX(snapshot_date по всей таблице)`. На этой
# единственной глобал-max дате присутствует обычно лишь 1 объект (скрейпнутый
# последним) → у остальных объектов в радиусе 0 квартир → supply_units_in_radius=0
# для ВСЕХ строк → frontend salesPeriodMonths(0,v)=0 («Срок продажи 0 мес»),
# sold_pct_of_supply=null («% продано —»). Прод-факт (66:41:0205010:287): 9
# объектов в радиусе = 5344 квартир суммарно, но 0 на глобал-max снимке.
# Фикс: flats_latest берёт для КАЖДОГО obj_id его СОБСТВЕННЫЙ последний снимок
# (DISTINCT ON (obj_id) ... ORDER BY obj_id, snapshot_date DESC, id DESC) и
# джойнится к nearby по obj_id → supply = реальные per-object счётчики (~2675).
# Дедуп nearby (objects-first MATERIALIZED, #1944) сохранён — flats_latest даёт
# ровно одну строку на (obj_id) → fan-out по снимкам flats невозможен.
# NB: данные domrf_kn flats частично сломаны (#1945 — per-object снимки низкие/
# несвежие пока loader не починен), поэтому supply всё ещё ниже здорового
# состояния, но движется с 0 к реальным per-object числам. Loader — НЕ здесь.
_SUPPLY_BATCH_SQL = text("""
    WITH latest_obj AS (
        SELECT DISTINCT ON (o.obj_id)
            o.obj_id, o.latitude, o.longitude
        FROM domrf_kn_objects o
        WHERE o.latitude IS NOT NULL AND o.longitude IS NOT NULL
        ORDER BY o.obj_id, o.snapshot_date DESC
    ),
    nearby AS MATERIALIZED (
        SELECT lo.obj_id
        FROM latest_obj lo
        WHERE ST_DWithin(
            ST_SetSRID(ST_MakePoint(lo.longitude, lo.latitude), 4326)::geography,
            ST_SetSRID(
                ST_MakePoint(CAST(:center_lon AS float), CAST(:center_lat AS float)),
                4326
            )::geography,
            CAST(:radius_m AS float)
        )
    ),
    flats_latest AS (
        SELECT DISTINCT ON (f.obj_id)
            f.obj_id, f.snapshot_date
        FROM domrf_kn_flats f
        JOIN nearby n ON n.obj_id = f.obj_id
        ORDER BY f.obj_id, f.snapshot_date DESC, f.id DESC
    )
    SELECT
        CASE
            WHEN f.is_studio = TRUE OR f.flat_type = 'Квартира-студия' THEN 'studio'
            WHEN f.rooms = 0 THEN 'studio'
            WHEN f.rooms IN (1, 2, 3) THEN f.rooms::text
            WHEN f.rooms >= 4 THEN '4+'
            ELSE '1'
        END AS rb,
        CASE
            WHEN f.total_area < 25  THEN '<25'
            WHEN f.total_area < 40  THEN '25-40'
            WHEN f.total_area < 60  THEN '40-60'
            WHEN f.total_area < 80  THEN '60-80'
            WHEN f.total_area < 100 THEN '80-100'
            ELSE '100+'
        END AS ab,
        COUNT(*) AS units
    FROM flats_latest fl
    JOIN domrf_kn_flats f
        ON f.obj_id = fl.obj_id
       AND f.snapshot_date = fl.snapshot_date
    GROUP BY rb, ab
""")


# ── Вспомогательные функции ───────────────────────────────────────────────────


def _quarter_from_cad(cad_num: str) -> str:
    """Извлечь кадастровый квартал: '66:41:0303161:123' → '66:41:0303161'."""
    parts = cad_num.split(":")
    if len(parts) >= 3:
        return ":".join(parts[:3])
    return cad_num


def _normalize_pct(buckets: dict[str, float]) -> dict[str, int]:
    """Нормировать доли до целых процентов с суммой ровно 100.

    Алгоритм largest-remainder (Hamilton method):
    1. Floor каждого значения.
    2. Остаток 100 − sum_floors распределить в top-bucket по дробной части.
    """
    if not buckets:
        return {}

    total = sum(buckets.values())
    if total <= 0:
        n = len(buckets)
        base = 100 // n
        result = {k: base for k in buckets}
        # распределить остаток
        remainder = 100 - base * n
        for k in list(buckets.keys())[:remainder]:
            result[k] += 1
        return result

    raw = {k: v / total * 100.0 for k, v in buckets.items()}
    floors = {k: int(v) for k, v in raw.items()}
    remainder = 100 - sum(floors.values())
    # sort by fractional part desc
    fracs = sorted(buckets.keys(), key=lambda k: -(raw[k] - floors[k]))
    for k in fracs[:remainder]:
        floors[k] += 1
    return floors


def _cap_and_redistribute(pct_map: dict[str, int]) -> tuple[dict[str, int], bool]:
    """Fix SF-09 round 2: capacity-aware redistribute, bounded iterations.

    Round 1 bug: surplus распределялся пропорционально текущему `v` free bucket'а,
    что переливало его выше cap — на 2-bucket вход цикл осциллировал бесконечно.

    Round 2 fix: surplus распределяется пропорционально **available capacity**
    `(cap - v)` каждого free bucket'а. Тогда free никогда не вылетит выше cap →
    цикл сходится за ≤ len(pct_map) итераций. Hard guard `for _ in range(N+1)`.

    Если surplus > total_capacity (геометрически невозможно поместить излишек ниже
    cap) — забиваем все free к cap, возвращаем `cap_skipped=True` + warning log.

    Returns:
        (result_map, cap_skipped) — cap_skipped=True если cap не удержан
        (pathological: всё хочет > cap, или surplus > available capacity).
    """
    if not pct_map:
        return pct_map, False

    cap = MAX_BUCKET_SHARE_PCT

    # Быстрый path: нет доминирующих
    if all(v <= cap for v in pct_map.values()):
        return pct_map, False

    work: dict[str, float] = {k: float(v) for k, v in pct_map.items()}

    # Bounded iteration: после k-й итерации число clamped не убывает только если
    # surplus > capacity (тогда — pathological). При корректном capacity-aware
    # redistribute достаточно ≤ len(pct_map) итераций.
    for _ in range(len(pct_map) + 1):
        clamped = [k for k, v in work.items() if v > cap]
        if not clamped:
            break

        free = [k for k, v in work.items() if v < cap]
        if not free:
            # Все bucket'ы либо >cap либо ровно =cap — некуда переливать.
            logger.warning(
                "MAX_BUCKET_SHARE cap: нет free bucket'ов (%d total) — cap_skipped",
                len(pct_map),
            )
            return pct_map, True

        surplus = sum(work[k] - cap for k in clamped)
        capacities = {k: cap - work[k] for k in free}
        total_capacity = sum(capacities.values())

        for k in clamped:
            work[k] = float(cap)

        if surplus > total_capacity + 1e-9:
            # Излишек не помещается ниже cap — pathological.
            # Возвращаем оригинал (sum=100 invariant) + флаг для frontend banner.
            logger.warning(
                "MAX_BUCKET_SHARE cap: surplus %.2f > total_capacity %.2f — cap_skipped",
                surplus,
                total_capacity,
            )
            return pct_map, True

        for k in free:
            work[k] += capacities[k] / total_capacity * surplus
    else:
        # Hard guard: не сошлись за N+1 итераций — bug. Лог + cap_skipped.
        logger.error(
            "MAX_BUCKET_SHARE cap: не сошлись за %d итераций — algorithm bug",
            len(pct_map) + 1,
        )
        return pct_map, True

    return _hamilton_round(work), False


def _hamilton_round(work: dict[str, float]) -> dict[str, int]:
    """Hamilton apportionment: float → integer pct с суммой ровно 100."""
    floors = {k: int(v) for k, v in work.items()}
    remainder = 100 - sum(floors.values())
    fracs = sorted(work.keys(), key=lambda k: -(work[k] - floors[k]))
    for k in fracs[: max(0, remainder)]:
        floors[k] += 1
    return floors


# ── Главная функция ───────────────────────────────────────────────────────────


def get_best_layouts(
    db: Session,
    cad_num: str,
    request: BestLayoutsRequest,
) -> BestLayoutsResponse:
    """Top layouts (rooms × area_bin) конкурентов с рейтингом по velocity.

    Raises:
        ValueError: если центроид участка не найден (caller → HTTP 404).
    """
    quarter = _quarter_from_cad(cad_num)
    radius_m = request.radius_km * 1000.0

    # time_window → (interval_str, months divisor)
    window_interval, months_in_window = _TIME_WINDOW_PARAMS.get(
        request.time_window, ("3 months", 3.0)
    )

    # ── Step 1: центроид участка ─────────────────────────────────────────────
    try:
        coord_row = (
            db.execute(
                _PARCEL_CENTROID_SQL,
                {"cad_num": cad_num, "quarter": quarter},
            )
            .mappings()
            .first()
        )
    except Exception:
        logger.exception("best_layouts: centroid query failed for cad_num=%s", cad_num)
        raise

    if not coord_row:
        raise ValueError(f"Геометрия для {cad_num} не найдена")

    center_lon = float(coord_row["center_lon"])
    center_lat = float(coord_row["center_lat"])

    # ── Step 2: obj_id конкурентов в радиусе ────────────────────────────────
    try:
        id_rows = (
            db.execute(
                _COMPETITORS_IN_RADIUS_SQL,
                {
                    "center_lon": center_lon,
                    "center_lat": center_lat,
                    "radius_m": radius_m,
                    "obj_class_filter": request.obj_class_filter,
                },
            )
            .mappings()
            .all()
        )
    except Exception:
        logger.exception("best_layouts: competitors-in-radius query failed for cad_num=%s", cad_num)
        raise

    all_obj_ids: list[int] = [int(r["obj_id"]) for r in id_rows]
    objects_total_in_radius = len(all_obj_ids)

    # Применить exclude / filter из request
    exclude_set = set(request.exclude_competitor_obj_ids)
    if exclude_set:
        all_obj_ids = [oid for oid in all_obj_ids if oid not in exclude_set]

    if request.filter_competitor_obj_ids is not None:
        filter_set = set(request.filter_competitor_obj_ids)
        all_obj_ids = [oid for oid in all_obj_ids if oid in filter_set]

    if not all_obj_ids:
        return _empty_response(
            radius_km=request.radius_km,
            time_window=request.time_window,
            objects_total_in_radius=objects_total_in_radius,
        )

    # ── Step 3: inline velocity из objective_corpus_room_month ──────────────
    # Fix SF-01: честный фильтр report_month >= NOW() - window_interval.
    # Разные time_window → разные deals_window, разный mix.
    try:
        vel_rows = (
            db.execute(
                _INLINE_VELOCITY_SQL,
                {
                    "window_interval": window_interval,
                    "competitor_obj_ids": all_obj_ids,
                },
            )
            .mappings()
            .all()
        )
    except Exception:
        logger.exception(
            "best_layouts: inline velocity query failed for cad_num=%s obj_count=%d",
            cad_num,
            len(all_obj_ids),
        )
        raise

    if not vel_rows:
        return _empty_response(
            radius_km=request.radius_km,
            time_window=request.time_window,
            objects_total_in_radius=objects_total_in_radius,
        )

    # ── Step 5: supply side (батч-запрос) ────────────────────────────────────
    # Fix #1956: supply берёт ПЕР-ОБЪЕКТНЫЙ последний снимок из domrf_kn_flats
    # (каждый ЖК скрейпится в свой день). Прежний единый глобальный
    # MAX(snapshot_date) давал 0 квартир почти у всех объектов → «Срок продажи
    # 0 мес» / «% продано —». Per-object latest snapshot встроен в _SUPPLY_BATCH_SQL
    # (flats_latest CTE) → отдельный pre-compute MAX больше не нужен.
    try:
        supply_rows = (
            db.execute(
                _SUPPLY_BATCH_SQL,
                {
                    "center_lon": center_lon,
                    "center_lat": center_lat,
                    "radius_m": radius_m,
                },
            )
            .mappings()
            .all()
        )
    except Exception:
        logger.warning("best_layouts: supply query failed, supply=0 fallback")
        supply_rows = []

    supply_map: dict[tuple[str, str], int] = {
        (str(r["rb"]), str(r["ab"])): int(r["units"]) for r in supply_rows
    }

    # ── Step 4 + 6: velocity из реального окна и enrichment per row ─────────
    # Fix SF-01: velocity_per_month = deals_window / months_in_window.
    # deals_window уже отфильтрован по report_month — разные time_window дают разные данные.

    enriched: list[dict[str, Any]] = []
    window_start: dt.date | None = None
    window_end: dt.date | None = None

    # Собираем obj_ids с данными в objective_corpus_room_month (для data_quality)
    obj_ids_with_data: set[int] = set()

    for r in vel_rows:
        room_bucket = str(r["room_bucket"])
        deals_window = float(r["deals_window"]) if r["deals_window"] is not None else 0.0
        avg_area = float(r["avg_area_m2"]) if r["avg_area_m2"] is not None else 0.0
        price_rub = (
            float(r["avg_price_per_m2_rub"]) if r["avg_price_per_m2_rub"] is not None else None
        )
        competitor_obj_ids: list[int] = (
            [int(oid) for oid in r["competitor_obj_ids"]] if r["competitor_obj_ids"] else []
        )
        competitor_count = int(r["competitor_count"])

        obj_ids_with_data.update(competitor_obj_ids)

        # Step 4: честный velocity = сделки за окно / длина окна в месяцах
        velocity_per_month = round(deals_window / months_in_window, 2)

        # Step 6: area_bin по avg_area (layout_signature.area_bin)
        ab = area_bin(avg_area) if avg_area > 0 else "<25"
        sig = layout_signature(room_bucket, ab)  # type: ignore[arg-type]

        supply_count = supply_map.get((room_bucket, ab), 0)
        sold_pct: float | None = None
        is_oversold = False
        if supply_count > 0:
            sold_pct_raw = deals_window / supply_count * 100.0
            is_oversold = sold_pct_raw > 100.0
            # Clamp at 100%: сделки за 24 мес / текущий snapshot supply несопоставимы.
            # Значения >100% артефакт окна, не реальная «распроданность».
            sold_pct = round(min(sold_pct_raw, 100.0), 1)

        # data window
        if r["window_start"] is not None:
            ws = r["window_start"]
            if isinstance(ws, str):
                ws = dt.date.fromisoformat(ws)
            elif isinstance(ws, dt.datetime):
                ws = ws.date()
            window_start = ws if window_start is None else min(window_start, ws)

        if r["window_end"] is not None:
            we = r["window_end"]
            if isinstance(we, str):
                we = dt.date.fromisoformat(we)
            elif isinstance(we, dt.datetime):
                we = we.date()
            window_end = we if window_end is None else max(window_end, we)

        enriched.append(
            {
                "room_bucket": room_bucket,
                "area_bin": ab,
                "signature": sig,
                "competitor_obj_ids": competitor_obj_ids,
                "competitor_count": competitor_count,
                "sum_deals": deals_window,
                "velocity_per_month": velocity_per_month,
                "avg_price_per_m2_rub": price_rub,
                "avg_area_m2": avg_area,
                "supply_units_in_radius": supply_count,
                "sold_pct_of_supply": sold_pct,
                "is_oversold": is_oversold,
            }
        )

    # ── Step 7: фильтр min_velocity + sort + rank ────────────────────────────
    filtered = [
        row for row in enriched if row["velocity_per_month"] >= request.min_velocity_per_month
    ]
    filtered.sort(key=lambda r: r["velocity_per_month"], reverse=True)

    top_layouts: list[TopLayoutRow] = []
    for rank_idx, row in enumerate(filtered, start=1):
        top_layouts.append(
            TopLayoutRow(
                rank=rank_idx,
                room_bucket=row["room_bucket"],
                area_bin=row["area_bin"],
                signature=row["signature"],
                competitor_obj_ids=row["competitor_obj_ids"],
                competitor_count=row["competitor_count"],
                total_sold_in_window=int(row["sum_deals"]),
                velocity_per_month=row["velocity_per_month"],
                avg_price_per_m2_rub=row["avg_price_per_m2_rub"],
                avg_area_m2=round(row["avg_area_m2"], 1),
                supply_units_in_radius=row["supply_units_in_radius"],
                sold_pct_of_supply=row["sold_pct_of_supply"],
                is_oversold=row["is_oversold"],
            )
        )

    # ── Step 8: build recommendation_for_tz ─────────────────────────────────
    # Используем filtered (только > min_velocity) для recommendation.
    # Если после фильтрации всё пустое — используем enriched (все данные без фильтра).
    rec_source = filtered if filtered else enriched

    today = dt.date.today()
    ws_date = window_start if window_start is not None else today
    we_date = window_end if window_end is not None else today

    recommendation = _build_recommendation(
        rows=rec_source,
        radius_km=request.radius_km,
        time_window=request.time_window,
        target_total_flats=request.target_total_flats,
        window_start=ws_date,
        window_end=we_date,
        all_enriched=enriched,
    )

    # ── Step 9: data_quality ─────────────────────────────────────────────────
    # Denominator = post-filter set (effective consideration set после exclude/filter).
    objects_total_after_filter = len(all_obj_ids)
    objects_with_data = len(obj_ids_with_data & set(all_obj_ids))
    coverage_pct = (
        round(objects_with_data / objects_total_after_filter * 100.0, 1)
        if objects_total_after_filter > 0
        else 0.0
    )
    if coverage_pct >= LAYOUT_CONFIDENCE_HIGH_PCT:
        confidence: str = "high"
    elif coverage_pct >= LAYOUT_CONFIDENCE_MEDIUM_PCT:
        confidence = "medium"
    else:
        confidence = "low"

    data_quality = LayoutDataQuality(
        objects_with_velocity_data=objects_with_data,
        objects_total_in_radius=objects_total_after_filter,
        velocity_coverage_pct=coverage_pct,
        confidence=confidence,  # type: ignore[arg-type]
    )

    return BestLayoutsResponse(
        top_layouts=top_layouts,
        recommendation_for_tz=recommendation,
        data_quality=data_quality,
    )


def _build_recommendation(
    rows: list[dict[str, Any]],
    radius_km: float,
    time_window: str,
    target_total_flats: int | None,
    window_start: dt.date,
    window_end: dt.date,
    all_enriched: list[dict[str, Any]],
) -> LayoutTzRecommendation:
    """Собрать LayoutTzRecommendation из enriched rows."""
    if not rows:
        return LayoutTzRecommendation(
            rationale_text=(
                f"В радиусе {radius_km}км: нет layout-паттернов с достаточной velocity."
            ),
            mix=[],
            weighted_avg_price_per_m2_rub=None,
            based_on_obj_count=0,
            based_on_total_deals=0,
            data_window_start=window_start,
            data_window_end=window_end,
        )

    # Группировка по room_bucket (строки уже могут быть per-bucket из MV GROUP BY)
    rb_deals: dict[str, float] = {}
    rb_area_weighted: dict[str, float] = {}
    rb_price_weighted: dict[str, float] = {}
    rb_price_total_deals: dict[str, float] = {}
    all_competitor_ids: set[int] = set()

    for row in rows:
        rb = row["room_bucket"]
        sd = float(row["sum_deals"])
        rb_deals[rb] = rb_deals.get(rb, 0.0) + sd
        rb_area_weighted[rb] = rb_area_weighted.get(rb, 0.0) + row["avg_area_m2"] * sd
        all_competitor_ids.update(row["competitor_obj_ids"])
        if row["avg_price_per_m2_rub"] is not None:
            rb_price_weighted[rb] = rb_price_weighted.get(rb, 0.0) + (
                row["avg_price_per_m2_rub"] * sd
            )
            rb_price_total_deals[rb] = rb_price_total_deals.get(rb, 0.0) + sd

    total_deals = sum(rb_deals.values())
    pct_map = _normalize_pct(rb_deals)
    pct_map, cap_skipped = _cap_and_redistribute(pct_map)

    mix: list[LayoutTzMixRow] = []
    for rb, pct in sorted(pct_map.items(), key=lambda x: -x[1]):
        avg_area = (
            round(rb_area_weighted[rb] / rb_deals[rb], 1) if rb_deals.get(rb, 0) > 0 else None
        )
        abs_units: int | None = None
        if target_total_flats is not None:
            abs_units = round(pct / 100.0 * target_total_flats)
        mix.append(
            LayoutTzMixRow(
                room_bucket=rb,
                pct=pct,
                abs_units=abs_units,
                avg_target_area_m2=avg_area,
            )
        )

    # Weighted avg price across all room_buckets
    total_price_deals = sum(rb_price_total_deals.values())
    weighted_price: float | None = None
    if total_price_deals > 0:
        weighted_price = round(sum(rb_price_weighted.values()) / total_price_deals, 0)

    # Rationale
    competitor_count = len(all_competitor_ids)
    tw_label = {"last_month": "1 мес", "last_quarter": "квартал", "last_year": "год"}.get(
        time_window, time_window
    )
    rationale_text = (
        f"В радиусе {radius_km}км за {tw_label}: "
        f"{len(rows)} активных layout-паттернов, "
        f"total {int(total_deals)} продаж в {competitor_count} ЖК"
    )

    # based_on_obj_count из all_enriched (уникальные obj_id с данными MV)
    all_mv_obj_ids: set[int] = set()
    for row in all_enriched:
        all_mv_obj_ids.update(row["competitor_obj_ids"])

    return LayoutTzRecommendation(
        rationale_text=rationale_text,
        mix=mix,
        weighted_avg_price_per_m2_rub=weighted_price,
        based_on_obj_count=len(all_mv_obj_ids),
        based_on_total_deals=int(total_deals),
        data_window_start=window_start,
        data_window_end=window_end,
        cap_skipped=cap_skipped,
    )


def _empty_response(
    radius_km: float,
    time_window: str,
    objects_total_in_radius: int,
) -> BestLayoutsResponse:
    """Ответ когда нет конкурентов или нет MV данных."""
    today = dt.date.today()
    tw_label = {"last_month": "1 мес", "last_quarter": "квартал", "last_year": "год"}.get(
        time_window, time_window
    )
    return BestLayoutsResponse(
        top_layouts=[],
        recommendation_for_tz=LayoutTzRecommendation(
            rationale_text=(
                f"В радиусе {radius_km}км за {tw_label}: "
                f"конкуренты не найдены или нет данных velocity."
            ),
            mix=[],
            weighted_avg_price_per_m2_rub=None,
            based_on_obj_count=0,
            based_on_total_deals=0,
            data_window_start=today,
            data_window_end=today,
        ),
        data_quality=LayoutDataQuality(
            objects_with_velocity_data=0,
            objects_total_in_radius=objects_total_in_radius,
            velocity_coverage_pct=0.0,
            confidence="low",
        ),
    )