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,
    CompetitorsRequest,
    CompetitorsResponse,
    CompetitorsSummary,
)
from app.services.forecast_request_cache import cached

logger = logging.getLogger(__name__)


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,
            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)
      AND (
          CAST(:obj_class_filter AS text) IS NULL
          OR d.obj_class = CAST(:obj_class_filter AS text)
      )
    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 даёт корректную среднюю цену ЖК.
#
# #949 PR B: добавлена колонка flats_sold (COUNT квартир со status LIKE '%прод%')
# — нужна для stage_at_horizon (available = flat_count - flats_sold). status
# заполнен слабо (~0.2%), поэтому flats_sold чаще будет 0/мало → stage_at_horizon
# падает в нейтраль 0.5 (graceful, см. helper). Колонка добавлена в ТОТ ЖЕ запрос
# (без 4-го db.execute), чтобы не ломать существующие callers/тесты.
# #1210: domrf_kn_flats версионируется (UNIQUE(id, snapshot_date) — м.50,
# scraper UPSERT per snapshot). Без фильтра snapshot_date AVG усреднял
# ИСТОРИЮ цен (stale на растущем рынке), а COUNT '%прод%' множил sold ×N
# снапшотов (raw_sold/flat_count кратно завышен → попадал в гард-нейтраль
# 0.5 или искажал stage_at_horizon). Берём ТОЛЬКО 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,
        COUNT(*) FILTER (
            WHERE LOWER(COALESCE(f.status, '')) LIKE '%прод%'
               OR LOWER(COALESCE(f.status, '')) = 'sold'
        ) AS flats_sold
    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
""")


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,
                    "obj_class_filter": request.obj_class_filter,
                    "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]

    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. Средняя цена м² + sold-count (graceful — таблица может быть пуста) ─
    # flats_sold нужен для stage_at_horizon (available = flat_count - flats_sold).
    # Читаем через _row_get: существующие тесты мокируют только avg_price_per_m2,
    # отсутствующий ключ flats_sold → None (не KeyError) → нейтральный stage.
    avg_price_map: dict[int, float] = {}
    sold_count_map: dict[int, int] = {}
    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)
            sold = _row_get(r, "flats_sold")
            if sold is not None:
                sold_count_map[oid] = int(sold)
    except Exception:
        logger.warning("competitors: avg_price query failed, continuing without prices")

    # ── 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 из domrf_kn_flats по status LIKE '%прод%' (заполнен слабо
    # ~0.2%, поэтому чаще 0/None). При 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),
                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)