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

"""Анализ активных конкурентов ЖК в радиусе от участка.

Issue #112 — Demand: активные конкуренты, продажи ЖК в радиусе 1км за квартал.

Источники:
  domrf_kn_objects            — ЖК с lat/lon, flat_count, obj_class, site_status
  objective_complex_mapping   — domrf_obj_id → objective_complex_name
  objective_corpus_room_month — monthly deals_total_count per project_name
  cad_parcels_geom            — centroid участка (fallback: cad_quarters_geom)
  domrf_kn_flats              — avg price_per_m2 по проданным квартирам

Внимание: velocity coverage ~2.5% — большинство ЖК не имеют маппинга в
objective_complex_mapping. LEFT JOIN гарантирует velocity=0 (не ошибку) для
немаппированных объектов.
"""

from __future__ import annotations

import logging
from typing import Any

from sqlalchemy import text
from sqlalchemy.orm import Session

from app.schemas.parcel import (
    Competitor,
    CompetitorParking,
    CompetitorsParkingResponse,
    CompetitorsRequest,
    CompetitorsResponse,
    CompetitorsSummary,
)
from app.services.forecast_request_cache import cached
from app.services.site_finder.premises_lookup import (
    get_building_premises_for_match,
    resolve_cad_for_domrf,
)

logger = logging.getLogger(__name__)

# #96 wiring: сколько top-конкурентов (по relevance_weight DESC) обогащаем
# parking_ratio в ленивом эндпоинте. Строгий top-N: каждый = geom-match SQL
# (~19ms) + НСПД premises (~0.6-2с). 3 — баланс «достаточно для сравнения дефицита
# паркинга» vs суммарная latency (~2-6с на ленивый вызов, ВНЕ analyze/forecast
# хот-пути). Расширять с осторожностью — линейно множит НСПД-запросы.
_PARKING_TOP_N: int = 3


def _competitors_key(db: Session, cad_num: str, request: CompetitorsRequest) -> tuple[Any, ...]:
    """Ключ кэша §22-форсайта для get_competitors (#1129) — что РЕАЛЬНО влияет на вывод.

    Все поля CompetitorsRequest, изменяющие результат (radius/time_window/класс-фильтр/
    horizon_months/exclude_obj_ids), + cad_num. exclude_obj_ids — список → в tuple для
    hashability (его порядок на вывод не влияет, но в форсайт-пути он всегда пуст —
    сортируем для устойчивости ключа). `db` в ключ не входит (одна сессия на отчёт).
    """
    return (
        cad_num,
        request.radius_km,
        request.time_window,
        request.obj_class_filter,
        request.horizon_months,
        tuple(sorted(request.exclude_obj_ids)),
    )


# Маппинг time_window → число месяцев (float для деления velocity)
_TIME_WINDOW_MONTHS: dict[str, float] = {
    "last_month": 1.0,
    "last_quarter": 3.0,
    "last_year": 12.0,
}

# site_status значения, считающиеся «активными» (#1213). domrf_kn_objects.site_status
# хранит русские значения ровно как scraper берёт из siteStatus дом.рф (см. domrf_kn.py:316
# — без перевода). Прод-аудит: 105_add_sales_started_flag.sql фильтрует по 'Строящиеся'
# (~1322 строки), partial index в 66_indexes_recommend.sql использует те же, фронтовые
# проверки MarketTab/CompetitorTable — тоже русские. Английские 'sales'/'construction'
# не совпадали никогда → у всех Competitor is_active=false и active_count=0 при любых
# данных. is_sold_out_filter сохранён как раньше — на стороне фронта.
_ACTIVE_STATUSES = frozenset({"Строящиеся"})

# #968 (949-A): радиус спатиального матча domrf↔complexes для velocity gap-fill.
# 200м + tolerant-name + nearest-per-domrf → подтверждаем «тот же ЖК» И гео, И
# именем. Чистый name-match даёт ~40 cross-city false positives; чистый spatial
# рискует приписать velocity СОСЕДНЕГО ЖК. Подбор по проду: 148 → ~300-400 (#968).
_VELOCITY_MATCH_RADIUS_M: float = 200.0

# ──────────────────────────────────────────────────────────────────────────────
# #949 PR B (ТЗ §9.1, §16) — relevance_weight: детерминированная (без LLM)
# взвешенная оценка релевантности конкурента 0..1 + explainability-breakdown.
# ──────────────────────────────────────────────────────────────────────────────

# Веса саб-скоров для итогового relevance_weight (named constants — tunable).
# Сумма = 1.0; geo и stage_at_horizon — главные сигналы (близость + «будет ли
# конкурировать к нашему запуску»), class/price — корректировки сопоставимости.
_W_GEO: float = 0.30
_W_CLASS: float = 0.25
_W_PRICE: float = 0.20
_W_STAGE: float = 0.25

# Нейтральный саб-скор при недостатке данных (graceful: не штрафуем и не
# завышаем конкурента, по которому нет сигнала — ТЗ §15 spirit).
_NEUTRAL: float = 0.5

# Детерминированный порядок класса жилья (домрф): эконом < комфорт < комфорт+ <
# бизнес < бизнес+ < премиум/элит. class_similarity считается по дистанции в этом
# порядке. Ключи нормализуются через _normalize_class (lower + синонимы домрф/EN,
# чтобы 'Комфорт'/'comfort'/'комфорт-класс' попадали в один индекс).
_CLASS_ORDER: dict[str, int] = {
    "эконом": 0,
    "стандарт": 0,  # домрф канон для эконома (см. domrf_kn._OBJ_CLASS_PATTERNS)
    "типовой": 0,
    "комфорт": 1,
    "комфорт+": 2,
    "бизнес": 3,
    "бизнес+": 4,
    "премиум": 5,
    "элит": 5,
}

# class_similarity по дистанции шагов в _CLASS_ORDER: 0 шагов (тот же класс) → 1.0,
# 1 шаг (соседний, комфорт↔комфорт+) → 0.6, 2 → 0.3, ≥3 → 0.1.
_CLASS_SIM_BY_STEPS: dict[int, float] = {0: 1.0, 1: 0.6, 2: 0.3}
_CLASS_SIM_FAR: float = 0.1

# Синонимы для нормализации obj_class к ключам _CLASS_ORDER. obj_class в данных
# приходит и кириллицей ('Комфорт'), и латиницей (фильтр API economy/comfort/
# business), и с суффиксами ('комфорт-класс'). Сводим к каноническому ключу.
_CLASS_SYNONYMS: dict[str, str] = {
    "economy": "эконом",
    "econom": "эконом",
    "standard": "стандарт",
    "comfort": "комфорт",
    "comfortplus": "комфорт+",
    "business": "бизнес",
    "businessplus": "бизнес+",
    "premium": "премиум",
    "elite": "элит",
    "elit": "элит",
}


def _normalize_class(obj_class: str | None) -> str | None:
    """Привести obj_class к каноническому ключу _CLASS_ORDER.

    Терпимо к регистру, латинице (economy/comfort/business из API-фильтра),
    суффиксам ('комфорт-класс', '«Бизнес»') и плюс-формам ('комфорт +',
    'комфорт плюс' → 'комфорт+'). Возвращает None если класс не распознан.
    """
    if not obj_class:
        return None
    s = obj_class.strip().lower()
    # Унифицируем плюс-формы до проверки прямого вхождения.
    s = s.replace(" плюс", "+").replace("плюс", "+").replace(" +", "+")
    if s in _CLASS_ORDER:
        return s
    if s in _CLASS_SYNONYMS:
        return _CLASS_SYNONYMS[s]
    # Подстрочный матч по канон-ключам (длинные ключи первыми: 'комфорт+'
    # раньше 'комфорт', иначе 'комфорт-класс плюс' схлопнется в 'комфорт').
    for key in sorted(_CLASS_ORDER, key=len, reverse=True):
        if key in s:
            return key
    for syn, key in _CLASS_SYNONYMS.items():
        if syn in s:
            return key
    return None


def _geo_proximity(distance_m: float | None, radius_m: float) -> float:
    """geo_proximity ∈ [0,1] — линейный decay расстояния.

    1.0 на участке (distance 0), → 0.0 на краю радиуса. За радиусом / без
    данных → нейтрально не нужно: расстояние всегда известно (PostGIS), но
    защищаемся от None/нулевого радиуса. clamp в [0,1].
    """
    if distance_m is None or radius_m <= 0:
        return _NEUTRAL
    return max(0.0, min(1.0, 1.0 - distance_m / radius_m))


def _class_similarity(competitor_class: str | None, reference_class: str | None) -> float:
    """class_similarity ∈ [0,1] по дистанции в _CLASS_ORDER.

    Тот же класс → 1.0, соседний (1 шаг) → 0.6, 2 шага → 0.3, дальше → 0.1.
    Если класс конкурента или эталон неизвестен/нераспознан → нейтрально 0.5
    (нет сигнала о сопоставимости — не штрафуем).

    reference_class: эталон сравнения. На этой стадии целевой класс участка ещё
    не выбран (нет target), поэтому caller передаёт ЛОКАЛЬНЫЙ ДОМИНИРУЮЩИЙ класс
    (самый частый среди конкурентов в радиусе) — так скор остаётся осмысленным
    «насколько конкурент типичен для этой локации».
    """
    c = _normalize_class(competitor_class)
    r = _normalize_class(reference_class)
    if c is None or r is None:
        return _NEUTRAL
    steps = abs(_CLASS_ORDER[c] - _CLASS_ORDER[r])
    return _CLASS_SIM_BY_STEPS.get(steps, _CLASS_SIM_FAR)


def _price_similarity(competitor_ppm2: float | None, median_ppm2: float | None) -> float:
    """price_similarity ∈ [0,1] — близость цены конкурента к локальной медиане.

    1 - min(1, |c_price - median| / median): на медиане → 1.0, отклонение в
    100%+ → 0.0. Нет цены конкурента или невалидная медиана → нейтрально 0.5.
    """
    if competitor_ppm2 is None or not median_ppm2 or median_ppm2 <= 0:
        return _NEUTRAL
    rel_diff = abs(competitor_ppm2 - median_ppm2) / median_ppm2
    return max(0.0, 1.0 - min(1.0, rel_diff))


def _stage_at_horizon(
    available: float | None,
    velocity_per_month: float | None,
    horizon_months: int,
    flats_total: float | None,
) -> float:
    """stage_at_horizon ∈ [0,1] — horizon-aware ключевой фактор (ТЗ §9.1).

    Проецируем распродажу конкурента на горизонт нашего запуска:
        projected_remaining = max(0, available - velocity * horizon_months)
    Скор = доля проектируемого остатка от изначального объёма
    (projected_remaining / flats_total). Конкурент, у которого к запуску ещё
    значимый остаток → высокий скор (он БУДЕТ конкурировать); проектируемо
    распроданный → низкий.

    Graceful: если velocity или available/total неизвестны (тонкие данные) →
    нейтрально 0.5 — не переоцениваем и не недооцениваем (ТЗ §15). clamp [0,1].
    """
    if available is None or velocity_per_month is None or flats_total is None or flats_total <= 0:
        return _NEUTRAL
    if velocity_per_month <= 0:
        # Нет продаж: остаток не уменьшается — конкурент останется на рынке
        # (макс. релевантность по этому фактору).
        return max(0.0, min(1.0, available / flats_total))
    projected_remaining = max(0.0, available - velocity_per_month * float(horizon_months))
    return max(0.0, min(1.0, projected_remaining / flats_total))


def _relevance_weight(geo: float, class_sim: float, price: float, stage: float) -> float:
    """Взвешенное среднее 4 саб-скоров → relevance_weight ∈ [0,1].

    Веса — named constants (_W_GEO/_W_CLASS/_W_PRICE/_W_STAGE), сумма 1.0.
    Все входы предполагаются уже clamped в [0,1]; результат тоже clamp для
    устойчивости к будущей подстройке весов.
    """
    weighted = geo * _W_GEO + class_sim * _W_CLASS + price * _W_PRICE + stage * _W_STAGE
    return max(0.0, min(1.0, weighted))


def _dominant_class(obj_classes: list[str | None]) -> str | None:
    """Локальный доминирующий класс — самый частый нормализованный obj_class.

    Используется как эталон class_similarity, когда целевой класс участка ещё
    не задан. None если ни один класс не распознан. При ничьей берётся класс с
    наименьшим индексом в _CLASS_ORDER (детерминированно, без рандома).
    """
    counts: dict[str, int] = {}
    for raw in obj_classes:
        norm = _normalize_class(raw)
        if norm is not None:
            counts[norm] = counts.get(norm, 0) + 1
    if not counts:
        return None
    # Сортируем: больше частота → раньше; при равной частоте меньший order-индекс.
    return min(counts, key=lambda k: (-counts[k], _CLASS_ORDER[k]))


def _median(values: list[float]) -> float | None:
    """Детерминированная медиана непустого списка. Пусто → None."""
    if not values:
        return None
    ordered = sorted(values)
    n = len(ordered)
    mid = n // 2
    if n % 2 == 1:
        return ordered[mid]
    return (ordered[mid - 1] + ordered[mid]) / 2.0


def _row_get(row: Any, key: str) -> Any:
    """Безопасно прочитать ключ из row-mapping → None если ключа нет.

    Используем ТОЛЬКО __getitem__: и SQLAlchemy RowMapping, и тестовые
    MagicMock-строки реализуют его корректно и кидают KeyError на отсутствующий
    ключ. (Не используем .get(): на MagicMock это авто-атрибут-заглушка,
    возвращающая MagicMock вместо реального значения.) Нужно для backward-safe
    чтения новой колонки flats_sold — старые моки её просто не отдают."""
    try:
        return row[key]
    except (KeyError, TypeError, IndexError):
        return None


# SQL для получения центроида участка
_PARCEL_CENTROID_SQL = text("""
    SELECT ST_X(pt) AS lon, ST_Y(pt) AS 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
""")

# Основной запрос конкурентов в радиусе.
# Velocity через objective_corpus_room_month (актуальные данные, обновляется еженедельно).
# domrf_kn_sale_graph устарел (данные до 2026-01) — не используется.
# Coverage velocity ~2.5%: большинство obj_id нет в objective_complex_mapping →
# LEFT JOIN → velocity=0 (не ошибка).
_COMPETITORS_SQL = text("""
    WITH latest_obj AS (
        SELECT DISTINCT ON (obj_id)
            obj_id,
            comm_name,
            dev_name,
            -- #38: реальный obj_class в приоритете, иначе obj_class_fallback
            -- (yandex_match / price_inference). Класс-фильтр и dominant_class
            -- в Python-слое работают по этому эффективному значению.
            COALESCE(obj_class, obj_class_fallback) AS obj_class,
            latitude,
            longitude,
            flat_count,
            site_status,
            snapshot_date
        FROM domrf_kn_objects
        WHERE latitude IS NOT NULL
          AND longitude IS NOT NULL
        ORDER BY obj_id, snapshot_date DESC NULLS LAST
    ),
    nearest_cx AS (
        -- #968 (949-A) velocity gap-fill: для domrf-объектов, КОТОРЫХ НЕТ
        -- в явном objective_complex_mapping, ищем «тот же ЖК» в complexes по
        -- ГЕОГРАФИИ (≤ :velocity_match_radius_m м) И ТОЛЕРАНТНОМУ имени.
        -- DISTINCT ON (obj_id) ORDER BY distance → ровно ОДИН (ближайший)
        -- complex на конкурента: имя+гео подтверждают «тот же ЖК» (precision
        -- важна — velocity питает forecast). Спатиальный join дёшев, т.к.
        -- complexes мал (~1.5k строк с geom) — НЕ потому что latest_obj
        -- пре-фильтрован (радиус участка применяется ниже, в финальном SELECT).
        SELECT DISTINCT ON (o.obj_id)
               o.obj_id,
               c.id AS complex_id
          FROM latest_obj o
          JOIN complexes c
            ON c.latitude IS NOT NULL
           AND c.longitude IS NOT NULL
           AND c.canonical_name IS NOT NULL
           -- #968: только complexes, у которых ЕСТЬ objective velocity-данные →
           -- «ближайший» выбирается среди data-bearing. Иначе ближайший complex
           -- без objective_lots съедал бы матч (prod: +37 наивно → +181 так).
           AND EXISTS (
                   SELECT 1 FROM objective_lots ol
                    WHERE ol.complex_id = c.id AND ol.project_name IS NOT NULL
               )
           AND ST_DWithin(
                   ST_SetSRID(ST_MakePoint(o.longitude, o.latitude), 4326)::geography,
                   ST_SetSRID(ST_MakePoint(c.longitude, c.latitude), 4326)::geography,
                   CAST(:velocity_match_radius_m AS float)
               )
           AND (
                   lower(btrim(o.comm_name)) = lower(btrim(c.canonical_name))
                OR lower(btrim(c.canonical_name)) LIKE '%' || lower(btrim(o.comm_name)) || '%'
                OR lower(btrim(o.comm_name)) LIKE '%' || lower(btrim(c.canonical_name)) || '%'
               )
         WHERE o.comm_name IS NOT NULL
           AND btrim(o.comm_name) <> ''  -- иначе LIKE '%%' матчит всё → spatial-only
           AND o.obj_id NOT IN (SELECT domrf_obj_id FROM objective_complex_mapping)
         ORDER BY o.obj_id,
                  ST_Distance(
                      ST_SetSRID(ST_MakePoint(o.longitude, o.latitude), 4326)::geography,
                      ST_SetSRID(ST_MakePoint(c.longitude, c.latitude), 4326)::geography
                  ) ASC
    ),
    mapped AS (
        -- PRIMARY (unchanged): 148 явных маппингов — velocity для них byte-identical.
        SELECT cm.domrf_obj_id AS obj_id,
               cm.objective_complex_name
          FROM objective_complex_mapping cm
        UNION
        -- FALLBACK (gap-fill only): все project_name ОДНОГО ближайшего complex'а
        -- (у комплекса может быть несколько корпус-project_name → velocity
        -- легитимно суммируется по ним, но НЕ по нескольким комплексам).
        SELECT nc.obj_id,
               ol.project_name AS objective_complex_name
          FROM nearest_cx nc
          JOIN objective_lots ol
            ON ol.complex_id = nc.complex_id
           AND ol.project_name IS NOT NULL
    ),
    velocity AS (
        SELECT
            m.obj_id,
            SUM(COALESCE(crm.deals_total_count, 0))
                / CAST(:time_window_months AS float) AS velocity_per_month
          FROM objective_corpus_room_month crm
          JOIN mapped m ON m.objective_complex_name = crm.project_name
         WHERE crm.report_month >= (NOW() - CAST(:window_interval AS interval))
         GROUP BY m.obj_id
    ),
    distances AS (
        SELECT
            o.obj_id,
            o.comm_name,
            o.dev_name,
            o.obj_class,
            o.latitude,
            o.longitude,
            o.flat_count,
            o.site_status,
            ST_Distance(
                ST_SetSRID(ST_MakePoint(o.longitude, o.latitude), 4326)::geography,
                ST_SetSRID(
                    ST_MakePoint(CAST(:center_lon AS float), CAST(:center_lat AS float)),
                    4326
                )::geography
            ) AS distance_m
        FROM latest_obj o
    )
    SELECT
        d.obj_id,
        d.comm_name,
        d.dev_name,
        d.obj_class,
        d.latitude,
        d.longitude,
        d.flat_count,
        d.site_status,
        d.distance_m,
        COALESCE(v.velocity_per_month, 0.0) AS velocity_per_month
    FROM distances d
    LEFT JOIN velocity v ON v.obj_id = d.obj_id
    WHERE d.distance_m <= CAST(:radius_m AS float)
    ORDER BY d.distance_m ASC
""")

# Средняя цена м² по квартирам с известной ценой для набора obj_id.
# Фильтр status='sold' убран: поле status в domrf_kn_flats заполнено в ~0.2% строк
# (99.8% NULL) — фильтр давал 0 строк и avg_price_per_m2 всегда None (Issue #112/227).
# AVG по всем квартирам с price_per_m2 IS NOT NULL даёт корректную среднюю цену ЖК.
#
# #1210: domrf_kn_flats версионируется (UNIQUE(id, snapshot_date) — м.50,
# scraper UPSERT per snapshot). Без фильтра snapshot_date AVG усреднял
# ИСТОРИЮ цен (stale на растущем рынке). Берём ТОЛЬКО latest snapshot — зеркало
# паттерна best_layouts._SUPPLY_BATCH_SQL и _COMPETITORS_SQL DISTINCT ON.
_AVG_PRICE_SQL = text("""
    SELECT
        f.obj_id,
        AVG(f.price_per_m2) FILTER (WHERE f.price_per_m2 IS NOT NULL)
            AS avg_price_per_m2
    FROM domrf_kn_flats f
    WHERE f.obj_id = ANY(:obj_ids)
      AND f.snapshot_date = (
          SELECT MAX(snapshot_date) FROM domrf_kn_flats
      )
    GROUP BY f.obj_id
""")

# Кол-во проданных квартир (flats_sold) для набора obj_id — нужно для
# stage_at_horizon (available = flat_count - flats_sold).
#
# #1926 (аудит #1871): РАНЬШЕ flats_sold считался из domrf_kn_flats.status
# (LIKE '%прод%'/'sold'). status заполнен в ~0.2% строк (99.8% NULL) → массовый
# UNDERCOUNT (в репортнутом кейсе ~1600 ед. недосчёта): почти все конкуренты
# получали flats_sold≈0 → stage_at_horizon вырождался в нейтраль/«ничего не
# продано», и блок конкурентов РАСХОДИЛСЯ с MarketMetrics, который считает sold
# по objective_lots.
#
# FIX: считаем sold ИЗ ТОГО ЖЕ ИСТОЧНИКА И ПО ТОМУ ЖЕ ПРЕДИКАТУ, что
# MarketMetrics (_STOCK_SQL): лот продан, если
#   is_sold IS TRUE OR contract_date IS NOT NULL OR LOWER(status)='продан'
# (Объектив заполняет эти поля неконсистентно, поэтому берём ИХ ИЛИ; см.
# market_metrics._STOCK_SQL:285-289). Так flats_sold побайтово зеркалит n_sold.
#
# Мост domrf-obj → лоты — тот же, что в _OBJECTIVE_PRICE_FALLBACK_SQL
# (primary_price CTE): domrf_kn_objects.obj_id == objective_complex_mapping
# .domrf_obj_id → objective_complex_mapping.objective_complex_name ==
# objective_lots.project_name.
#
# ВАЖНО (code-review #1927): objective_complex_mapping UNIQUE — это
# (objective_complex_name, objective_group), а НЕ domrf_obj_id (68_schema:321).
# Значит ОДИН obj_id может иметь НЕСКОЛЬКО mapping-строк (то же имя, разный
# objective_group) → наивный COUNT(*) по cm JOIN ol задвоил бы лоты (GROUP BY
# схлопывает выходные строки, но не дубли ВНУТРИ агрегата). Поэтому
# COUNT(DISTINCT ol.objective_lot_id) — дедуп на уровне лота, fan-out-safe.
# (_OBJECTIVE_PRICE_FALLBACK не прецедент: percentile/median dup-нечувствительны.)
#
# #1964: objective_lots — current-state UPSERT, но НЕ «ровно одна строка на лот».
# UNIQUE = objective_lot_id, а Объектив за пере-листинги присваивает ОДНОМУ
# физлоту (project,corpus,section,floor,lot) НЕСКОЛЬКО objective_lot_id → таблица
# раздута ~2.91× → flats_sold завышался. Читаем из v_objective_lots_latest
# (physflat-дедуп view, по строке на физлот). В view objective_lot_id уникален
# (603k строк = 603k distinct lot_id), поэтому COUNT(DISTINCT ol.objective_lot_id)
# == COUNT(*) по физлотам — НО оставляем DISTINCT: маппинг 1:N по domrf_obj_id
# (см. выше) при будущем fan-out задвоил бы строки view, COUNT(*) был бы не
# fan-out-safe. DISTINCT даёт И physflat-дедуп (через view), И fan-out-защиту.
#
# premise_kind='квартира' — как в _STOCK_SQL: без него проданные паркинги/
# кладовые/нежилое считались бы как проданные квартиры (over-count).
#
# Дефолт premise_kind зеркалит market_metrics.compute_market_metrics
# (premise_kind="квартира") — оба считают продажи жилых лотов.
_SOLD_PREMISE_KIND = "квартира"
_SOLD_COUNT_SQL = text("""
    SELECT
        cm.domrf_obj_id AS obj_id,
        COUNT(DISTINCT ol.objective_lot_id) FILTER (
            WHERE ol.is_sold IS TRUE
               OR ol.contract_date IS NOT NULL
               OR LOWER(COALESCE(ol.status, '')) = 'продан'
        ) AS flats_sold
    FROM objective_complex_mapping cm
    JOIN v_objective_lots_latest ol
        ON ol.project_name = cm.objective_complex_name
    WHERE cm.domrf_obj_id = ANY(:obj_ids)
      AND ol.premise_kind = :premise_kind
    GROUP BY cm.domrf_obj_id
""")

# OBJ-3 (#307): additive fallback цены м² из objective_lots, когда domrf-цены нет.
# Замер прода 2026-06: на ПОСЛЕДНЕМ snapshot domrf_kn_flats.price_per_m2 = 0 строк
# (скрапер перестал отдавать per-flat цены), поэтому _AVG_PRICE_SQL выше даёт
# avg_price_per_m2=None почти всем конкурентам. objective_lots.price_per_m2_rub
# покрывает ~80% (919795/1123771 строк) и мостится к domrf-obj_id через
# objective_complex_mapping (то же звено, что _INLINE_VELOCITY_SQL в best_layouts).
#
# Мост: domrf_kn_objects.obj_id == objective_complex_mapping.domrf_obj_id
#       → objective_complex_mapping.objective_complex_name == objective_lots.project_name.
# Маппинг 1:1 (прод: 263 строки, 263 distinct obj_id, 0 obj_id с >1 именем), поэтому
# на obj_id приходится РОВНО один project_name; корпуса проекта агрегируются в один
# проект естественно. Берём МЕДИАНУ price_per_m2_rub по всем лотам проекта
# (percentile_cont 0.5) — устойчива к хвостовым выбросам пентхаусов/студий, в отличие
# от AVG. Прод-санити: медиана медиан 136k руб/м² (диапазон 80k–250k для ЕКБ ✓).
#
# Additive-контракт: возвращаем ТОЛЬКО obj_id, у которых objective-цена есть; в Python
# заполняем пробелы (domrf-цена приоритетна, objective — fallback). Существующие
# непустые domrf-выводы НЕ меняются. price_source делает источник прозрачным.
#
# #1615: velocity обогащается из ДВУХ источников (см. _COMPETITORS_SQL mapped CTE) —
# явного objective_complex_mapping И спатиально-именного nearest_cx gap-fill. Ценовой
# fallback должен покрывать ОБА, иначе конкурент с velocity>0 из spatial-матча получает
# avg_price=None и price_similarity падает в нейтраль. Зеркалим тот же мост obj→lots:
#   PRIMARY:  objective_complex_mapping.objective_complex_name == objective_lots.project_name
#   GAP-FILL: nearest_cx (≤ :velocity_match_radius_m м + tolerant-name) → complex_id →
#             objective_lots по complex_id (тот же DISTINCT ON ближайший complex, что и
#             velocity gap-fill — обязан совпадать, чтобы цена и velocity были про ОДИН ЖК).
# obj_id мапится в РОВНО один источник (mapping 1:1; gap-fill — только для obj_id ВНЕ
# mapping, см. NOT IN ниже), поэтому пересечения нет и UNION ALL безопасен.
_OBJECTIVE_PRICE_FALLBACK_SQL = text("""
    WITH primary_price AS (
        SELECT
            cm.domrf_obj_id AS obj_id,
            ol.price_per_m2_rub AS price_per_m2_rub
        FROM objective_complex_mapping cm
        JOIN objective_lots ol
            ON ol.project_name = cm.objective_complex_name
        WHERE cm.domrf_obj_id = ANY(:obj_ids)
          AND ol.price_per_m2_rub IS NOT NULL
    ),
    nearest_cx AS (
        SELECT DISTINCT ON (o.obj_id)
               o.obj_id,
               c.id AS complex_id
          FROM domrf_kn_objects o
          JOIN complexes c
            ON c.latitude IS NOT NULL
           AND c.longitude IS NOT NULL
           AND c.canonical_name IS NOT NULL
           AND EXISTS (
                   SELECT 1 FROM objective_lots ol
                    WHERE ol.complex_id = c.id AND ol.project_name IS NOT NULL
               )
           AND ST_DWithin(
                   ST_SetSRID(ST_MakePoint(o.longitude, o.latitude), 4326)::geography,
                   ST_SetSRID(ST_MakePoint(c.longitude, c.latitude), 4326)::geography,
                   CAST(:velocity_match_radius_m AS float)
               )
           AND (
                   lower(btrim(o.comm_name)) = lower(btrim(c.canonical_name))
                OR lower(btrim(c.canonical_name)) LIKE '%' || lower(btrim(o.comm_name)) || '%'
                OR lower(btrim(o.comm_name)) LIKE '%' || lower(btrim(c.canonical_name)) || '%'
               )
         WHERE o.obj_id = ANY(:obj_ids)
           AND o.latitude IS NOT NULL
           AND o.longitude IS NOT NULL
           AND o.comm_name IS NOT NULL
           AND btrim(o.comm_name) <> ''
           AND o.obj_id NOT IN (SELECT domrf_obj_id FROM objective_complex_mapping)
         ORDER BY o.obj_id,
                  ST_Distance(
                      ST_SetSRID(ST_MakePoint(o.longitude, o.latitude), 4326)::geography,
                      ST_SetSRID(ST_MakePoint(c.longitude, c.latitude), 4326)::geography
                  ) ASC
    ),
    gapfill_price AS (
        SELECT
            nc.obj_id AS obj_id,
            ol.price_per_m2_rub AS price_per_m2_rub
        FROM nearest_cx nc
        JOIN objective_lots ol
            ON ol.complex_id = nc.complex_id
           AND ol.price_per_m2_rub IS NOT NULL
    )
    SELECT
        p.obj_id,
        percentile_cont(0.5) WITHIN GROUP (ORDER BY p.price_per_m2_rub)
            AS median_price_per_m2
    FROM (
        SELECT obj_id, price_per_m2_rub FROM primary_price
        UNION ALL
        SELECT obj_id, price_per_m2_rub FROM gapfill_price
    ) p
    GROUP BY p.obj_id
""")


def _quarter_from_cad(cad_num: str) -> str:
    """Извлечь кадастровый квартал из номера участка/здания.

    66:41:0303161:123 → 66:41:0303161
    Если формат нестандартный — возвращаем cad_num как есть (fallback).
    """
    parts = cad_num.split(":")
    if len(parts) >= 3:
        return ":".join(parts[:3])
    return cad_num


@cached(_competitors_key, label="get_competitors")
def get_competitors(
    db: Session,
    cad_num: str,
    request: CompetitorsRequest,
) -> CompetitorsResponse:
    """Получить список конкурентов ЖК в радиусе от участка.

    Шаги:
    1. Найти центроид участка (cad_parcels_geom → cad_quarters_geom fallback).
    2. Выбрать ЖК из domrf_kn_objects в радиусе с velocity из objective_corpus_room_month.
    3. Применить exclude_obj_ids фильтр в Python (избегаем array cast).
    4. Подтянуть avg_price_per_m2 из domrf_kn_flats.
    5. Собрать CompetitorsResponse.

    Raises:
        ValueError: если центроид участка не найден (caller должен вернуть 404).
    """
    quarter = _quarter_from_cad(cad_num)

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

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

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

    # ── 2. Конкуренты в радиусе ──────────────────────────────────────────────
    time_window_months = _TIME_WINDOW_MONTHS[request.time_window]
    window_interval = f"{int(time_window_months)} months"

    try:
        rows = (
            db.execute(
                _COMPETITORS_SQL,
                {
                    "center_lat": center_lat,
                    "center_lon": center_lon,
                    "radius_m": request.radius_km * 1000.0,
                    "time_window_months": time_window_months,
                    "window_interval": window_interval,
                    "velocity_match_radius_m": _VELOCITY_MATCH_RADIUS_M,
                },
            )
            .mappings()
            .all()
        )
    except Exception:
        logger.exception(
            "competitors: main query failed for cad_num=%s radius_km=%.2f",
            cad_num,
            request.radius_km,
        )
        raise

    # ── 3. Применить exclude_obj_ids ─────────────────────────────────────────
    exclude_set = set(request.exclude_obj_ids)
    if exclude_set:
        rows = [r for r in rows if int(r["obj_id"]) not in exclude_set]

    # ── 3b. Фильтр класса объекта (#1340) ────────────────────────────────────
    # obj_class_filter из API приходит англ. (economy/comfort/business), а в БД
    # domrf_kn_objects.obj_class хранит РУССКИЙ канон (Стандарт/Комфорт/Бизнес/…
    # см. domrf_kn._OBJ_CLASS_PATTERNS). Прямое SQL-равенство 'comfort'='Комфорт'
    # всегда false → пустой список. Нормализуем обе стороны через _normalize_class
    # и сравниваем по индексу _CLASS_ORDER, поэтому economy (индекс 0) корректно
    # матчит и 'Стандарт', и 'Типовой' (оба индекс 0).
    if request.obj_class_filter is not None:
        target_norm = _normalize_class(request.obj_class_filter)
        target_order = _CLASS_ORDER.get(target_norm) if target_norm else None
        if target_order is not None:
            rows = [
                r
                for r in rows
                if (norm := _normalize_class(r["obj_class"])) is not None
                and _CLASS_ORDER[norm] == target_order
            ]

    if not rows:
        return CompetitorsResponse(
            competitors=[],
            summary=CompetitorsSummary(
                total_competitors=0,
                active_count=0,
                weighted_avg_velocity=0.0,
                radius_km=request.radius_km,
                time_window=request.time_window,
            ),
        )

    obj_ids: list[int] = [int(r["obj_id"]) for r in rows]

    # ── 4. Средняя цена м² (graceful — таблица может быть пуста) ──────────────
    avg_price_map: dict[int, float] = {}
    price_source_map: dict[int, str] = {}
    try:
        price_rows = db.execute(_AVG_PRICE_SQL, {"obj_ids": obj_ids}).mappings().all()
        for r in price_rows:
            oid = int(r["obj_id"])
            price = _row_get(r, "avg_price_per_m2")
            if price is not None:
                avg_price_map[oid] = float(price)
                price_source_map[oid] = "domrf"
    except Exception:
        logger.warning("competitors: avg_price query failed, continuing without prices")

    # ── 4·sold. flats_sold из objective_lots (#1926) ─────────────────────────
    # flats_sold нужен для stage_at_horizon (available = flat_count - flats_sold).
    # Источник И предикат выровнены с MarketMetrics._STOCK_SQL (is_sold/
    # contract_date/status='продан', premise_kind='квартира'), см. _SOLD_COUNT_SQL
    # — ушли от domrf_kn_flats.status (0.2%, недосчёт). Отдельный db.execute (не
    # ломает мок-тесты _AVG_PRICE_SQL); _row_get graceful: отсутствующий ключ
    # flats_sold → None → нейтральный stage.
    sold_count_map: dict[int, int] = {}
    try:
        sold_rows = (
            db.execute(_SOLD_COUNT_SQL, {"obj_ids": obj_ids, "premise_kind": _SOLD_PREMISE_KIND})
            .mappings()
            .all()
        )
        for r in sold_rows:
            oid = int(r["obj_id"])
            sold = _row_get(r, "flats_sold")
            if sold is not None:
                sold_count_map[oid] = int(sold)
    except Exception:
        logger.warning("competitors: sold-count query failed, continuing without flats_sold")

    # ── 4a. OBJ-3 (#307) additive fallback: где domrf-цены НЕТ → objective_lots ──
    # Заполняем ТОЛЬКО пробелы (obj_id без domrf-цены). Существующие domrf-выводы
    # не трогаем (precedence: domrf > objective). Прод 2026-06: domrf на последнем
    # snapshot = 0 цен → почти все obj_id попадают в этот fallback (~80% покрытие).
    missing_price_ids = [oid for oid in obj_ids if oid not in avg_price_map]
    if missing_price_ids:
        try:
            obj_price_rows = (
                db.execute(
                    _OBJECTIVE_PRICE_FALLBACK_SQL,
                    {
                        "obj_ids": missing_price_ids,
                        "velocity_match_radius_m": _VELOCITY_MATCH_RADIUS_M,
                    },
                )
                .mappings()
                .all()
            )
            for r in obj_price_rows:
                oid = int(r["obj_id"])
                price = _row_get(r, "median_price_per_m2")
                if price is not None and oid not in avg_price_map:
                    avg_price_map[oid] = float(price)
                    price_source_map[oid] = "objective"
        except Exception:
            logger.warning(
                "competitors: objective price fallback failed, continuing with domrf only"
            )

    # ── 4b. Локальные эталоны для relevance (медиана цены, доминирующий класс) ─
    radius_m = request.radius_km * 1000.0
    local_median_ppm2 = _median(list(avg_price_map.values()))
    dominant_class = _dominant_class([r["obj_class"] for r in rows])

    # ── 5. Сборка результата + relevance_weight (ТЗ §9.1) ────────────────────
    # flats_sold: COUNT проданных лотов из objective_lots.contract_date (#1926,
    # выровнено с MarketMetrics; раньше domrf_kn_flats.status ~0.2% → недосчёт).
    # При known flats_sold считаем sold_pct и available для stage_at_horizon;
    # иначе оба поля None и stage → нейтраль 0.5.
    competitors: list[Competitor] = []
    for r in rows:
        obj_id = int(r["obj_id"])
        flats_total = int(r["flat_count"]) if r["flat_count"] is not None else None

        site_status = r["site_status"]
        is_active = site_status in _ACTIVE_STATUSES if site_status else False

        velocity = round(float(r["velocity_per_month"]), 2)

        # flats_sold / sold_pct / available — только если sold-count реально есть
        # И он осмыслен (не превышает total). Иначе None (graceful, не выдумываем).
        flats_sold: int | None = None
        sold_pct: float | None = None
        available: float | None = None
        raw_sold = sold_count_map.get(obj_id)
        if raw_sold is not None and flats_total and flats_total > 0 and raw_sold <= flats_total:
            flats_sold = raw_sold
            sold_pct = round(raw_sold / flats_total * 100.0, 1)
            available = float(flats_total - raw_sold)

        # ── relevance саб-скоры (каждый 0..1) + взвешенный weight ─────────────
        geo = _geo_proximity(float(r["distance_m"]), radius_m)
        class_sim = _class_similarity(r["obj_class"], dominant_class)
        price = _price_similarity(avg_price_map.get(obj_id), local_median_ppm2)
        stage = _stage_at_horizon(
            available=available,
            velocity_per_month=velocity,
            horizon_months=request.horizon_months,
            flats_total=float(flats_total) if flats_total else None,
        )
        relevance_weight = _relevance_weight(geo, class_sim, price, stage)
        relevance_breakdown = {
            "geo_proximity": round(geo, 3),
            "class_similarity": round(class_sim, 3),
            "price_similarity": round(price, 3),
            "stage_at_horizon": round(stage, 3),
        }

        competitors.append(
            Competitor(
                obj_id=obj_id,
                comm_name=r["comm_name"],
                dev_name=r["dev_name"],
                obj_class=r["obj_class"],
                distance_m=round(float(r["distance_m"]), 1),
                lat=float(r["latitude"]),
                lng=float(r["longitude"]),
                stage=site_status,
                flats_total=flats_total,
                flats_sold=flats_sold,
                sold_pct=sold_pct,
                velocity_per_month=velocity,
                avg_price_per_m2=avg_price_map.get(obj_id),
                price_source=price_source_map.get(obj_id),
                is_active=is_active,
                relevance_weight=round(relevance_weight, 3),
                relevance_breakdown=relevance_breakdown,
            )
        )

    # ── 5b. Сортировка по relevance_weight DESC (distance остаётся в полях) ────
    # tie-break по distance ASC (ближе — раньше), затем obj_id для стабильности.
    competitors.sort(key=lambda c: (-(c.relevance_weight or 0.0), c.distance_m, c.obj_id))

    # ── 6. Summary ───────────────────────────────────────────────────────────
    active_count = sum(1 for c in competitors if c.is_active)
    n = len(competitors)
    # #949 audit fix: size-weight velocity by flats_total (count-weighted AVG,
    # sql.md principle) — ЖК на 500 квартир должен весить больше, чем на 20, а не
    # наравне (раньше было наивное среднее вопреки имени weighted_*). Конкуренты с
    # неизвестным flats_total исключаются из весов; если размеры неизвестны У ВСЕХ —
    # graceful fallback на простое среднее (den>0 guard = NULLIF-эквивалент).
    weight_num = sum(c.velocity_per_month * c.flats_total for c in competitors if c.flats_total)
    weight_den = sum(c.flats_total for c in competitors if c.flats_total)
    if weight_den > 0:
        weighted_avg_velocity = round(weight_num / weight_den, 2)
    elif n > 0:
        weighted_avg_velocity = round(sum(c.velocity_per_month for c in competitors) / n, 2)
    else:
        weighted_avg_velocity = 0.0

    summary = CompetitorsSummary(
        total_competitors=n,
        active_count=active_count,
        weighted_avg_velocity=weighted_avg_velocity,
        radius_km=request.radius_km,
        time_window=request.time_window,
    )

    return CompetitorsResponse(competitors=competitors, summary=summary)


def get_competitors_parking(
    db: Session,
    cad_num: str,
    request: CompetitorsRequest,
    *,
    top_n: int = _PARKING_TOP_N,
) -> CompetitorsParkingResponse:
    """parking_ratio top-N конкурентов через geom-match + НСПД premises (#96).

    Ленивый путь (НЕ inline в analyze/forecast — см. parcel.py CompetitorParking
    docstring про p95). Для каждого из top-N конкурентов (get_competitors уже
    сортирует по relevance_weight DESC):
      1. resolve_cad_for_domrf(lat, lon) → ближайшее cad_buildings-здание (PostGIS
         KNN ≤ радиуса). Промах → parking отсутствует, НЕ падаем.
      2. get_building_premises_for_match(match) → НСПД tab-group → parking_ratio.
         (objdoc_id из cad_buildings экономит НСПД search-запрос.)

    Каждый конкурент изолирован try/except: сбой одного (геометрия/НСПД/WAF) не
    рушит остальные и не валит эндпоинт (graceful, ТЗ §15). Конкуренты без
    geom-match всё равно попадают в items с None-полями (стабильный контракт +
    фронт видит «искали, не нашли»).

    Raises:
        ValueError: если центроид участка не найден (из get_competitors — caller
            возвращает 404).
    """
    response = get_competitors(db=db, cad_num=cad_num, request=request)
    top = response.competitors[:top_n]

    items: list[CompetitorParking] = []
    matched_count = 0
    for competitor in top:
        item = CompetitorParking(obj_id=competitor.obj_id)
        try:
            match = resolve_cad_for_domrf(db, lat=competitor.lat, lon=competitor.lng)
        except Exception:
            # resolve_cad_for_domrf уже глотает SQL-сбои → None; этот guard на
            # случай неожиданного исключения — один конкурент без паркинга, не 500.
            logger.exception(
                "competitors_parking: geom-match raised for obj_id=%s", competitor.obj_id
            )
            match = None

        if match is not None:
            item = item.model_copy(
                update={
                    "matched_cad_num": match.cad_num,
                    "match_distance_m": round(match.distance_m, 1),
                }
            )
            try:
                listing = get_building_premises_for_match(match)
            except Exception:
                # НСПД-клиент уже graceful (NspdLiteError/NspdBulkError → None);
                # неожиданное — логируем и продолжаем без parking для этого ЖК.
                logger.exception(
                    "competitors_parking: premises lookup raised for obj_id=%s cad_num=%s",
                    competitor.obj_id,
                    match.cad_num,
                )
                listing = None
            if listing is not None:
                item = item.model_copy(
                    update={
                        "parking_ratio": listing.parking_ratio,
                        "parking_count": listing.parking_count,
                        "flats_count": listing.flats_count,
                    }
                )
                if listing.parking_ratio is not None:
                    matched_count += 1

        items.append(item)

    return CompetitorsParkingResponse(items=items, matched_count=matched_count)