gendesign/backend/app/services/site_finder/best_layouts.py

"""Анализ лучших планировок конкурентов по 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) в ГРУППАХ-КОМПЛЕКСАХ (#2177).

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

Fix #2177: coverage-знаменатель = число КОМПЛЕКСОВ, а не сырых obj_id.
domrf_kn_objects дробит один ЖК на несколько obj_id (очереди / дубли snapshot-строк /
безымянные корпуса), из-за чего «2 из 12 ЖК» завышало знаменатель. group_radius_objects
схлопывает obj_id в комплексы → честное «2 из 5 комплексов».
"""

from __future__ import annotations

import datetime as dt
import logging
import math
import re
from dataclasses import dataclass, field
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, comm_name, dev_name, latitude, longitude
    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


# ── Группировка obj_id в комплексы (Issue #2177) ──────────────────────────────
# domrf_kn_objects дробит один реальный ЖК на несколько obj_id:
#   • разные очереди («2 очередь жилого комплекса «7 ключей»»);
#   • дубли snapshot-строк одного комплекса (одинаковые comm_name+dev_name, точки
#     в ~200 м — «Симфония» 3346/43063/43064);
#   • повторы ядра имени внутри строки («Квартет Квартет Квартет» у Эталона);
#   • безымянные корпуса (comm_name пустой, гео-кучка в ~10 м — 3 Эфеса 55320-22).
# Сырой счёт obj_id завышает знаменатель coverage (прод: 12 obj_id = 5 комплексов).
# group_radius_objects схлопывает их в комплексы для честного «N из M комплексов».

# Радиус single-linkage гео-кластеризации безымянных obj_id (comm_name пустой) с
# одинаковым dev_name. Кучка в пределах NAMELESS_CLUSTER_M — один комплекс; дальше —
# отдельный (честно неизвестный) проект. 3 Эфеса в ~600 м от «7 ключей» → отдельно.
NAMELESS_CLUSTER_M = 300.0

# Generic-префиксы, которые убираем из начала имени перед сравнением ядра.
# «квартал» НЕ входит: в «Квартал Депо» слово — часть имени (без кавычек), не префикс.
_GENERIC_PREFIXES: tuple[str, ...] = (
    "жилой комплекс",
    "жк",
    "мкр",
    "микрорайон",
    "жилой район",
    "жилой квартал",
    "клубный дом",
    "клубный квартал",
    "дом",
)

# «N очередь [жилого комплекса] …» / «N-я очередь …» / «корпус N …» и подобные
# порядковые обёртки — вырезаем, оставляя ядро имени.
_STAGE_PREFIX_RE = re.compile(
    r"^\s*\d+\s*(?:-?\s*(?:я|й|ая|ой|го))?\s*"
    r"(?:очередь|очереди|этап|этапа|корпус|корпуса|секция|секции|"
    r"дом|блок)\b",
    re.IGNORECASE,
)

# «жилого комплекса» / «жилой комплекс» в родительном/именительном — вырезаем
# как соединительную обёртку внутри «… очередь ЖИЛОГО КОМПЛЕКСА «ядро»».
_JK_WORD_RE = re.compile(
    r"\bжил(?:ой|ого|ом|ому|ым)\s+комплекс(?:а|е|у|ом|ы)?\b",
    re.IGNORECASE,
)

# Ёлочки-кавычки любого начертания.
_QUOTE_INNER_RE = re.compile(r"[«»\"“”„'`]([^«»\"“”„'`]+)[«»\"“”„'`]")


def _extract_quoted_core(name: str) -> str | None:
    """Вернуть содержимое ёлочек, если оно есть: «7 ключей» → '7 ключей'."""
    match = _QUOTE_INNER_RE.search(name)
    if match:
        core = match.group(1).strip()
        if core:
            return core
    return None


def _dedup_repeated_words(name: str) -> str:
    """Схлопнуть подряд идущие повторы того же токена: 'квартет квартет' → 'квартет'.

    Эталоновский «Квартет» приходит как «Квартет Квартет Квартет» (повтор ядра ×3
    в одной строке). Убираем соседние дубли, регистр уже нормализован вызывающим.
    """
    tokens = name.split()
    out: list[str] = []
    for tok in tokens:
        if not out or out[-1] != tok:
            out.append(tok)
    return " ".join(out)


def normalize_complex_name(raw: str | None) -> str:
    """Нормализовать comm_name до имени-ядра комплекса.

    Шаги:
      1. lower + trim + схлопнуть пробелы;
      2. если есть ёлочки — берём их содержимое как ядро («7 ключей» из
         «2 очередь жилого комплекса «7 ключей»»);
      3. иначе — режем порядковый префикс («N очередь …»), слова «жилого
         комплекса», generic-префиксы; «квартал» сохраняем (часть имени);
      4. схлопываем соседние повторы токенов («Квартет Квартет» → «квартет»).

    Пустой / None → '' (безымянный, пойдёт в гео-кластеризацию).
    """
    if not raw:
        return ""
    name = re.sub(r"\s+", " ", raw.strip().lower())
    if not name:
        return ""

    quoted = _extract_quoted_core(name)
    if quoted is not None:
        return _dedup_repeated_words(re.sub(r"\s+", " ", quoted).strip())

    name = _STAGE_PREFIX_RE.sub("", name).strip()
    name = _JK_WORD_RE.sub(" ", name).strip()

    # generic-префиксы в начале (после снятия «очереди»); «квартал» намеренно не тут.
    changed = True
    while changed:
        changed = False
        for prefix in _GENERIC_PREFIXES:
            if name == prefix:
                # имя ЦЕЛИКОМ generic (например просто «жк») — оставить как есть,
                # иначе получим пустое ядро и склеим разные объекты.
                continue
            if name.startswith(prefix + " "):
                name = name[len(prefix) :].strip()
                changed = True
                break

    name = re.sub(r"\s+", " ", name).strip()
    return _dedup_repeated_words(name)


def _haversine_m(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
    """Расстояние между двумя точками (метры) по формуле haversine."""
    r = 6_371_000.0
    p1, p2 = math.radians(lat1), math.radians(lat2)
    dphi = math.radians(lat2 - lat1)
    dlam = math.radians(lon2 - lon1)
    a = math.sin(dphi / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dlam / 2) ** 2
    return 2 * r * math.asin(min(1.0, math.sqrt(a)))


@dataclass
class ComplexGroup:
    """Сгруппированный «комплекс» = один реальный ЖК из N сырых obj_id.

    key — стабильный идентификатор группы (для дедупа / логов), НЕ user-facing.
    obj_ids — все obj_id, схлопнутые в этот комплекс.
    """

    key: str
    obj_ids: list[int] = field(default_factory=list)
    dev_name_norm: str = ""
    core_name: str = ""


def group_radius_objects(rows: list[dict[str, Any]]) -> list[ComplexGroup]:
    """Схлопнуть сырые obj_id радиуса в комплексы (Issue #2177).

    Args:
        rows: строки из _COMPETITORS_IN_RADIUS_SQL — dict с ключами
            obj_id, comm_name, dev_name, latitude, longitude.

    Логика:
      • Именованные (comm_name непустой): ключ = (dev_name_норм, имя-ядро).
        Один и тот же комплекс (одинаковое ядро+застройщик) → одна группа,
        независимо от очереди / дублей snapshot-строк / разброса точек.
      • Безымянные (comm_name пустой): single-linkage гео-кластеризация по
        haversine в пределах NAMELESS_CLUSTER_M с ОДИНАКОВЫМ dev_name_норм.
        Каждый кластер = отдельная группа (честно неизвестный проект); НЕ
        приклеиваем к именованным группам (3 Эфеса в 600 м от «7 ключей»
        остаются отдельным комплексом).

    Returns:
        Список ComplexGroup, порядок детерминирован (по key).
    """
    named: dict[tuple[str, str], ComplexGroup] = {}
    nameless: list[dict[str, Any]] = []

    for row in rows:
        obj_id = int(row["obj_id"])
        dev_norm = re.sub(r"\s+", " ", (row.get("dev_name") or "").strip().lower())
        core = normalize_complex_name(row.get("comm_name"))

        if core:
            gkey = (dev_norm, core)
            grp = named.get(gkey)
            if grp is None:
                grp = ComplexGroup(
                    key=f"named:{dev_norm}|{core}",
                    dev_name_norm=dev_norm,
                    core_name=core,
                )
                named[gkey] = grp
            grp.obj_ids.append(obj_id)
        else:
            nameless.append(
                {
                    "obj_id": obj_id,
                    "dev_norm": dev_norm,
                    "lat": row.get("latitude"),
                    "lon": row.get("longitude"),
                }
            )

    groups: list[ComplexGroup] = list(named.values())
    groups.extend(_cluster_nameless(nameless))

    for grp in groups:
        grp.obj_ids.sort()
    groups.sort(key=lambda g: g.key)
    return groups


def _cluster_nameless(items: list[dict[str, Any]]) -> list[ComplexGroup]:
    """Single-linkage гео-кластеризация безымянных obj_id в пределах NAMELESS_CLUSTER_M.

    Кластеризуем ТОЛЬКО точки с одинаковым dev_norm (разные застройщики — заведомо
    разные проекты). Точек мало (единицы), поэтому наивный O(n²) union-по-соседям
    достаточен. obj_id без координат — каждый в свою группу (не с чем сравнивать).
    """
    groups: list[ComplexGroup] = []
    # Партиционируем по застройщику, чтобы не сливать разных девелоперов.
    by_dev: dict[str, list[dict[str, Any]]] = {}
    for it in items:
        by_dev.setdefault(it["dev_norm"], []).append(it)

    for dev_norm, pts in by_dev.items():
        # union-find по соседству ≤ NAMELESS_CLUSTER_M
        parent = list(range(len(pts)))

        def _find(i: int, parent: list[int] = parent) -> int:
            while parent[i] != i:
                parent[i] = parent[parent[i]]
                i = parent[i]
            return i

        for i in range(len(pts)):
            lat_i, lon_i = pts[i]["lat"], pts[i]["lon"]
            if lat_i is None or lon_i is None:
                continue
            for j in range(i + 1, len(pts)):
                lat_j, lon_j = pts[j]["lat"], pts[j]["lon"]
                if lat_j is None or lon_j is None:
                    continue
                if (
                    _haversine_m(float(lat_i), float(lon_i), float(lat_j), float(lon_j))
                    <= NAMELESS_CLUSTER_M
                ):
                    parent[_find(i)] = _find(j)

        clusters: dict[int, list[int]] = {}
        for idx, pt in enumerate(pts):
            root = _find(idx)
            clusters.setdefault(root, []).append(int(pt["obj_id"]))

        for obj_ids in clusters.values():
            anchor = min(obj_ids)
            groups.append(
                ComplexGroup(
                    key=f"nameless:{dev_norm}|{anchor}",
                    obj_ids=obj_ids,
                    dev_name_norm=dev_norm,
                    core_name="",
                )
            )
    return groups


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


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

    # Issue #2177: сохраняем полные строки (comm_name/dev_name/гео) для группировки
    # obj_id в комплексы. Сырой obj_id-счёт (raw_objects_total) — для прозрачности.
    id_row_dicts: list[dict[str, Any]] = [dict(r) for r in id_rows]
    all_obj_ids: list[int] = [int(r["obj_id"]) for r in id_row_dicts]
    raw_objects_total_in_radius = len(all_obj_ids)
    # Число комплексов ДО exclude/filter — знаменатель для empty-ответа (совместимо
    # со старой семантикой: пустой ответ отражал полный радиус, а не post-filter 0).
    groups_total_pre_filter = len(group_radius_objects(id_row_dicts))

    # Применить exclude / filter из request (и к obj_ids, и к строкам для группировки)
    exclude_set = set(request.exclude_competitor_obj_ids)
    filter_set: set[int] | None = (
        set(request.filter_competitor_obj_ids)
        if request.filter_competitor_obj_ids is not None
        else None
    )

    def _keep(oid: int) -> bool:
        if oid in exclude_set:
            return False
        if filter_set is not None and oid not in filter_set:
            return False
        return True

    id_row_dicts = [r for r in id_row_dicts if _keep(int(r["obj_id"]))]
    all_obj_ids = [oid for oid in all_obj_ids if _keep(oid)]

    # Группируем отфильтрованные obj_id в комплексы — знаменатель coverage.
    complex_groups = group_radius_objects(id_row_dicts)

    if not all_obj_ids:
        return _empty_response(
            radius_km=request.radius_km,
            time_window=request.time_window,
            objects_total_in_radius=groups_total_pre_filter,
            raw_objects_total=raw_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=len(complex_groups),
            raw_objects_total=raw_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 ─────────────────────────────────────────────────
    # Issue #2177: знаменатель = число КОМПЛЕКСОВ (сгруппированных obj_id), а не
    # сырых obj_id. Группа считается «с данными», если ХОТЯ БЫ ОДИН её obj_id есть
    # в obj_ids_with_data (velocity-покрытые). Прод-кейс (радиус 1 км): 12 obj_id
    # = 5 комплексов, из них 2 с velocity → «40% (2 из 5)» вместо «17% (2 из 12)».
    groups_total_after_filter = len(complex_groups)
    groups_with_data = sum(
        1 for g in complex_groups if any(oid in obj_ids_with_data for oid in g.obj_ids)
    )
    coverage_pct = (
        round(groups_with_data / groups_total_after_filter * 100.0, 1)
        if groups_total_after_filter > 0
        else 0.0
    )
    # Пороги confidence — от процента (не трогаем).
    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=groups_with_data,
        objects_total_in_radius=groups_total_after_filter,
        raw_objects_total=raw_objects_total_in_radius,
        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,
    raw_objects_total: int = 0,
) -> BestLayoutsResponse:
    """Ответ когда нет конкурентов или нет MV данных.

    objects_total_in_radius — число комплексов (сгруппированных obj_id, #2177),
    raw_objects_total — сырое число obj_id до группировки (прозрачность).
    """
    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,
            raw_objects_total=raw_objects_total,
            velocity_coverage_pct=0.0,
            confidence="low",
        ),
    )