backend/app/api/v1/parcels.py

@@ -15,6 +15,8 @@ from sqlalchemy.orm import Session
 from app.core.config import settings
 from app.core.db import get_db
 from app.schemas.parcel import (
+    BestLayoutsRequest,
+    BestLayoutsResponse,
     CompetitorsRequest,
     CompetitorsResponse,
     ConnectionPointsResponse,
@@ -22,6 +24,7 @@ from app.schemas.parcel import (
     ParcelSearchRequest,
     ParcelSearchResponse,
 )
+from app.services.site_finder.best_layouts import get_best_layouts
 from app.services.site_finder.cadastre_fetch import (
     cad_exists_in_db,
     find_or_enqueue_fetch,
@@ -2106,3 +2109,24 @@ async def get_parcel_competitors(
             status_code=500,
             detail="Ошибка расчёта конкурентов",
         ) from exc
+
+
+@router.post("/{cad_num}/best-layouts", response_model=BestLayoutsResponse)
+async def get_parcel_best_layouts(
+    cad_num: str,
+    body: BestLayoutsRequest,
+    db: Annotated[Session, Depends(get_db)],
+) -> BestLayoutsResponse:
+    """Top layouts (rooms × area_bin) у конкурентов с ranking по velocity.
+
+    Issue #113 Phase 2.1: "Анализ лучших планировок конкурентов → ТЗ на проектирование".
+    Reads from mv_layout_velocity (auto-populated via objective_corpus_room_month
+    × objective_complex_mapping).
+    """
+    try:
+        return get_best_layouts(db=db, cad_num=cad_num, request=body)
+    except ValueError as exc:
+        raise HTTPException(status_code=404, detail=str(exc)) from exc
+    except Exception as exc:
+        logger.error("best_layouts endpoint failed for %s: %s", cad_num, exc)
+        raise HTTPException(status_code=500, detail="Internal server error") from exc

backend/app/services/site_finder/best_layouts.py

@@ -0,0 +1,583 @@
+"""Анализ лучших планировок конкурентов по velocity (Issue #113 Phase 2.1).
+
+Источники:
+  cad_parcels_geom / cad_quarters_geom — центроид участка
+  domrf_kn_objects                     — ЖК в радиусе (latitude/longitude → geography)
+  mv_layout_velocity                   — (obj_id, room_bucket) → агрегат продаж 24 мес
+  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: JOIN mv_layout_velocity GROUP BY room_bucket.
+  Step 4: scale velocity по 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).
+"""
+
+from __future__ import annotations
+
+import datetime as dt
+import logging
+from typing import Any
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from app.schemas.parcel import (
+    BestLayoutsRequest,
+    BestLayoutsResponse,
+    LayoutDataQuality,
+    LayoutTzMixRow,
+    LayoutTzRecommendation,
+    TopLayoutRow,
+)
+from app.services.site_finder.layout_signature import area_bin, layout_signature
+
+logger = logging.getLogger(__name__)
+
+# Confidence thresholds (per coverage % of objects with MV velocity data)
+# Tune via PR if business feedback требует.
+LAYOUT_CONFIDENCE_HIGH_PCT = 50.0
+LAYOUT_CONFIDENCE_MEDIUM_PCT = 20.0
+
+# Делители velocity: 24 мес → масштаб на указанный window
+_VELOCITY_DIVISORS: dict[str, float] = {
+    "last_month": 24.0,
+    "last_quarter": 8.0,
+    "last_year": 2.0,
+}
+
+# ── SQL: центроид участка ─────────────────────────────────────────────────────
+
+_PARCEL_CENTROID_SQL = text("""
+    SELECT ST_X(pt) AS center_lon,
+           ST_Y(pt) AS center_lat
+    FROM (
+        SELECT ST_Centroid(geom) AS pt
+          FROM cad_parcels_geom
+         WHERE cad_num = :cad_num AND geom IS NOT NULL
+        UNION ALL
+        SELECT ST_Centroid(geom) AS pt
+          FROM cad_quarters_geom
+         WHERE cad_number = :quarter AND geom IS NOT NULL
+    ) sub
+    LIMIT 1
+""")
+
+# ── SQL: obj_id конкурентов в радиусе ─────────────────────────────────────────
+# Геометрия domrf_kn_objects вычисляется on-the-fly из (latitude, longitude)
+# как ST_SetSRID(ST_MakePoint(longitude, latitude), 4326)::geography
+# (consistency с competitors.py).
+# obj_class_filter: NULL = все классы.
+# filter_competitor_obj_ids: NULL = не фильтровать по списку.
+
+_COMPETITORS_IN_RADIUS_SQL = text("""
+    SELECT DISTINCT ON (obj_id) obj_id
+    FROM domrf_kn_objects
+    WHERE latitude IS NOT NULL AND longitude IS NOT NULL
+      AND ST_DWithin(
+          ST_SetSRID(ST_MakePoint(longitude, latitude), 4326)::geography,
+          ST_SetSRID(
+              ST_MakePoint(CAST(:center_lon AS float), CAST(:center_lat AS float)),
+              4326
+          )::geography,
+          CAST(:radius_m AS float)
+      )
+      AND (
+          CAST(:obj_class_filter AS text) IS NULL
+          OR obj_class = CAST(:obj_class_filter AS text)
+      )
+    ORDER BY obj_id, snapshot_date DESC NULLS LAST
+""")
+
+# ── SQL: mv_layout_velocity GROUP BY room_bucket ─────────────────────────────
+
+_VELOCITY_BY_ROOM_SQL = text("""
+    SELECT
+        room_bucket,
+        SUM(total_deals_24mo)                               AS sum_deals,
+        AVG(avg_area_m2)                                    AS avg_area_m2,
+        AVG(avg_price_thousand_rub_per_m2) * 1000.0         AS avg_price_per_m2_rub,
+        array_agg(DISTINCT obj_id)                          AS competitor_obj_ids,
+        COUNT(DISTINCT obj_id)                              AS competitor_count,
+        MIN(window_start)                                   AS window_start,
+        MAX(window_end)                                     AS window_end
+    FROM mv_layout_velocity
+    WHERE obj_id = ANY(:obj_ids)
+    GROUP BY room_bucket
+""")
+
+# ── SQL: supply по (room_bucket, area_bin) за последний снимок ───────────────
+# Один батч-запрос вместо N — возвращает map (rb, ab) → count.
+# room_bucket и area_bin вычисляются в SQL аналогично layout_signature.py.
+
+_SUPPLY_BATCH_SQL = text("""
+    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 domrf_kn_flats f
+    JOIN domrf_kn_objects o ON f.obj_id = o.obj_id
+    WHERE o.latitude IS NOT NULL AND o.longitude IS NOT NULL
+      AND ST_DWithin(
+          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,
+          CAST(:radius_m AS float)
+      )
+      AND f.snapshot_date = CAST(:latest_snap AS 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 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
+
+    # ── Step 1: центроид участка ─────────────────────────────────────────────
+    try:
+        coord_row = (
+            db.execute(
+                _PARCEL_CENTROID_SQL,
+                {"cad_num": cad_num, "quarter": quarter},
+            )
+            .mappings()
+            .first()
+        )
+    except Exception:
+        logger.exception("best_layouts: centroid query failed for cad_num=%s", cad_num)
+        raise
+
+    if not coord_row:
+        raise ValueError(f"Геометрия для {cad_num} не найдена")
+
+    center_lon = float(coord_row["center_lon"])
+    center_lat = float(coord_row["center_lat"])
+
+    # ── Step 2: obj_id конкурентов в радиусе ────────────────────────────────
+    try:
+        id_rows = (
+            db.execute(
+                _COMPETITORS_IN_RADIUS_SQL,
+                {
+                    "center_lon": center_lon,
+                    "center_lat": center_lat,
+                    "radius_m": radius_m,
+                    "obj_class_filter": request.obj_class_filter,
+                },
+            )
+            .mappings()
+            .all()
+        )
+    except Exception:
+        logger.exception("best_layouts: competitors-in-radius query failed for cad_num=%s", cad_num)
+        raise
+
+    all_obj_ids: list[int] = [int(r["obj_id"]) for r in id_rows]
+    objects_total_in_radius = len(all_obj_ids)
+
+    # Применить exclude / filter из request
+    exclude_set = set(request.exclude_competitor_obj_ids)
+    if exclude_set:
+        all_obj_ids = [oid for oid in all_obj_ids if oid not in exclude_set]
+
+    if request.filter_competitor_obj_ids is not None:
+        filter_set = set(request.filter_competitor_obj_ids)
+        all_obj_ids = [oid for oid in all_obj_ids if oid in filter_set]
+
+    if not all_obj_ids:
+        return _empty_response(
+            radius_km=request.radius_km,
+            time_window=request.time_window,
+            objects_total_in_radius=objects_total_in_radius,
+        )
+
+    # ── Step 3: mv_layout_velocity GROUP BY room_bucket ─────────────────────
+    try:
+        vel_rows = db.execute(_VELOCITY_BY_ROOM_SQL, {"obj_ids": all_obj_ids}).mappings().all()
+    except Exception:
+        logger.exception(
+            "best_layouts: velocity query failed for cad_num=%s obj_count=%d",
+            cad_num,
+            len(all_obj_ids),
+        )
+        raise
+
+    if not vel_rows:
+        return _empty_response(
+            radius_km=request.radius_km,
+            time_window=request.time_window,
+            objects_total_in_radius=objects_total_in_radius,
+        )
+
+    # ── Step 5: supply side (батч-запрос) ────────────────────────────────────
+    # Pre-compute последний snapshot_date один раз — избегаем subquery на каждый scan.
+    latest_snap: dt.date | None = db.scalar(text("SELECT MAX(snapshot_date) FROM domrf_kn_flats"))
+    if latest_snap is None:
+        logger.warning("best_layouts: domrf_kn_flats пустой (нет snapshot_date), supply=0 fallback")
+        supply_rows = []
+    else:
+        try:
+            supply_rows = (
+                db.execute(
+                    _SUPPLY_BATCH_SQL,
+                    {
+                        "center_lon": center_lon,
+                        "center_lat": center_lat,
+                        "radius_m": radius_m,
+                        "latest_snap": latest_snap,
+                    },
+                )
+                .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: scale velocity и enrichment per row ──────────────────────
+    divisor = _VELOCITY_DIVISORS[request.time_window]
+
+    enriched: list[dict[str, Any]] = []
+    window_start: dt.date | None = None
+    window_end: dt.date | None = None
+
+    # Собираем obj_ids с данными в MV (для data_quality)
+    obj_ids_with_data: set[int] = set()
+
+    for r in vel_rows:
+        room_bucket = str(r["room_bucket"])
+        sum_deals = float(r["sum_deals"]) if r["sum_deals"] 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: scale
+        velocity_per_month = round(sum_deals / divisor, 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
+        if supply_count > 0:
+            sold_pct = round(sum_deals / supply_count * 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": sum_deals,
+                "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,
+            }
+        )
+
+    # ── 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"],
+            )
+        )
+
+    # ── Step 8: build recommendation_for_tz ─────────────────────────────────
+    # Используем filtered (только > min_velocity) для recommendation.
+    # Если после фильтрации всё пустое — используем enriched (все данные без фильтра).
+    rec_source = filtered if filtered else enriched
+
+    today = dt.date.today()
+    ws_date = window_start if window_start is not None else today
+    we_date = window_end if window_end is not None else today
+
+    recommendation = _build_recommendation(
+        rows=rec_source,
+        radius_km=request.radius_km,
+        time_window=request.time_window,
+        target_total_flats=request.target_total_flats,
+        window_start=ws_date,
+        window_end=we_date,
+        all_enriched=enriched,
+    )
+
+    # ── Step 9: data_quality ─────────────────────────────────────────────────
+    # Denominator = post-filter set (effective consideration set после exclude/filter).
+    objects_total_after_filter = len(all_obj_ids)
+    objects_with_data = len(obj_ids_with_data & set(all_obj_ids))
+    coverage_pct = (
+        round(objects_with_data / objects_total_after_filter * 100.0, 1)
+        if objects_total_after_filter > 0
+        else 0.0
+    )
+    if coverage_pct >= LAYOUT_CONFIDENCE_HIGH_PCT:
+        confidence: str = "high"
+    elif coverage_pct >= LAYOUT_CONFIDENCE_MEDIUM_PCT:
+        confidence = "medium"
+    else:
+        confidence = "low"
+
+    data_quality = LayoutDataQuality(
+        objects_with_velocity_data=objects_with_data,
+        objects_total_in_radius=objects_total_after_filter,
+        velocity_coverage_pct=coverage_pct,
+        confidence=confidence,  # type: ignore[arg-type]
+    )
+
+    return BestLayoutsResponse(
+        top_layouts=top_layouts,
+        recommendation_for_tz=recommendation,
+        data_quality=data_quality,
+    )
+
+
+def _build_recommendation(
+    rows: list[dict[str, Any]],
+    radius_km: float,
+    time_window: str,
+    target_total_flats: int | None,
+    window_start: dt.date,
+    window_end: dt.date,
+    all_enriched: list[dict[str, Any]],
+) -> LayoutTzRecommendation:
+    """Собрать LayoutTzRecommendation из enriched rows."""
+    if not rows:
+        return LayoutTzRecommendation(
+            rationale_text=(
+                f"В радиусе {radius_km}км: нет layout-паттернов с достаточной velocity."
+            ),
+            mix=[],
+            weighted_avg_price_per_m2_rub=None,
+            based_on_obj_count=0,
+            based_on_total_deals=0,
+            data_window_start=window_start,
+            data_window_end=window_end,
+        )
+
+    # Группировка по room_bucket (строки уже могут быть per-bucket из MV GROUP BY)
+    rb_deals: dict[str, float] = {}
+    rb_area_weighted: dict[str, float] = {}
+    rb_price_weighted: dict[str, float] = {}
+    rb_price_total_deals: dict[str, float] = {}
+    all_competitor_ids: set[int] = set()
+
+    for row in rows:
+        rb = row["room_bucket"]
+        sd = float(row["sum_deals"])
+        rb_deals[rb] = rb_deals.get(rb, 0.0) + sd
+        rb_area_weighted[rb] = rb_area_weighted.get(rb, 0.0) + row["avg_area_m2"] * sd
+        all_competitor_ids.update(row["competitor_obj_ids"])
+        if row["avg_price_per_m2_rub"] is not None:
+            rb_price_weighted[rb] = rb_price_weighted.get(rb, 0.0) + (
+                row["avg_price_per_m2_rub"] * sd
+            )
+            rb_price_total_deals[rb] = rb_price_total_deals.get(rb, 0.0) + sd
+
+    total_deals = sum(rb_deals.values())
+    pct_map = _normalize_pct(rb_deals)
+
+    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,
+    )
+
+
+def _empty_response(
+    radius_km: float,
+    time_window: str,
+    objects_total_in_radius: int,
+) -> BestLayoutsResponse:
+    """Ответ когда нет конкурентов или нет MV данных."""
+    today = dt.date.today()
+    tw_label = {"last_month": "1 мес", "last_quarter": "квартал", "last_year": "год"}.get(
+        time_window, time_window
+    )
+    return BestLayoutsResponse(
+        top_layouts=[],
+        recommendation_for_tz=LayoutTzRecommendation(
+            rationale_text=(
+                f"В радиусе {radius_km}км за {tw_label}: "
+                f"конкуренты не найдены или нет данных velocity."
+            ),
+            mix=[],
+            weighted_avg_price_per_m2_rub=None,
+            based_on_obj_count=0,
+            based_on_total_deals=0,
+            data_window_start=today,
+            data_window_end=today,
+        ),
+        data_quality=LayoutDataQuality(
+            objects_with_velocity_data=0,
+            objects_total_in_radius=objects_total_in_radius,
+            velocity_coverage_pct=0.0,
+            confidence="low",
+        ),
+    )

backend/tests/api/v1/test_parcel_best_layouts.py

@@ -0,0 +1,379 @@
+"""Тесты для POST /api/v1/parcels/{cad_num}/best-layouts (Issue #113 Phase 2.1).
+
+Mock-based — не требуют живой БД.
+Паттерн mock DB: аналогично test_parcel_competitors.py — dependency_overrides[get_db].
+
+Порядок вызовов в get_best_layouts:
+  db.scalar()  → MAX(snapshot_date)            (только когда vel_rows non-empty)
+  db.execute() calls:
+  1. _PARCEL_CENTROID_SQL       → .mappings().first()
+  2. _COMPETITORS_IN_RADIUS_SQL → .mappings().all()
+  3. _VELOCITY_BY_ROOM_SQL      → .mappings().all()
+  4. _SUPPLY_BATCH_SQL          → .mappings().all()  (пропускается если latest_snap is None)
+"""
+
+from __future__ import annotations
+
+import datetime as dt
+from unittest.mock import MagicMock
+
+import pytest
+from fastapi.testclient import TestClient
+
+from app.main import app
+
+# ── Фабрики mock-строк ────────────────────────────────────────────────────────
+
+CAD_NUM = "66:41:0303161:123"
+_TODAY = dt.date.today()
+
+
+def _coord_row(lon: float = 60.6, lat: float = 56.85) -> MagicMock:
+    """Центроид участка (EPSG:4326 lon/lat)."""
+    r = MagicMock()
+    r.__getitem__ = lambda self, k: {"center_lon": lon, "center_lat": lat}[k]
+    return r
+
+
+def _obj_id_row(obj_id: int) -> MagicMock:
+    """Строка obj_id из _COMPETITORS_IN_RADIUS_SQL."""
+    r = MagicMock()
+    r.__getitem__ = lambda self, k: {"obj_id": obj_id}[k]
+    return r
+
+
+def _vel_row(
+    room_bucket: str = "2",
+    sum_deals: float = 48.0,
+    avg_area: float = 55.0,
+    avg_price_rub: float | None = 120000.0,
+    obj_ids: list[int] | None = None,
+    window_start: dt.date | None = None,
+    window_end: dt.date | None = None,
+) -> MagicMock:
+    """Строка из mv_layout_velocity GROUP BY room_bucket."""
+    oids = obj_ids if obj_ids is not None else [1]
+    ws = window_start or _TODAY - dt.timedelta(days=730)
+    we = window_end or _TODAY
+
+    r = MagicMock()
+    r.__getitem__ = lambda self, k: {
+        "room_bucket": room_bucket,
+        "sum_deals": sum_deals,
+        "avg_area_m2": avg_area,
+        "avg_price_per_m2_rub": avg_price_rub,
+        "competitor_obj_ids": oids,
+        "competitor_count": len(oids),
+        "window_start": ws,
+        "window_end": we,
+    }[k]
+    return r
+
+
+def _supply_row(rb: str, ab: str, units: int) -> MagicMock:
+    """Строка из _SUPPLY_BATCH_SQL."""
+    r = MagicMock()
+    r.__getitem__ = lambda self, k: {"rb": rb, "ab": ab, "units": units}[k]
+    return r
+
+
+# ── Построение mock DB ────────────────────────────────────────────────────────
+
+
+def _make_db(
+    coord: MagicMock | None = None,
+    id_rows: list[MagicMock] | None = None,
+    vel_rows: list[MagicMock] | None = None,
+    supply_rows: list[MagicMock] | None = None,
+    latest_snap: dt.date | None = None,
+) -> MagicMock:
+    """Сконструировать mock Session.
+
+    db.scalar() возвращает latest_snap (MAX snapshot_date) — вызывается перед supply.
+    Порядок db.execute():
+    1. centroid → .mappings().first()
+    2. competitors-in-radius → .mappings().all()
+    3. velocity → .mappings().all()
+    4. supply → .mappings().all()  (только если latest_snap is not None)
+    """
+    db = MagicMock()
+
+    # db.scalar — pre-computed MAX(snapshot_date) для supply query
+    db.scalar.return_value = latest_snap if latest_snap is not None else _TODAY
+
+    results: list[MagicMock] = []
+
+    # 1: centroid
+    r0 = MagicMock()
+    r0.mappings.return_value.first.return_value = coord
+    results.append(r0)
+
+    # 2: competitors-in-radius
+    r1 = MagicMock()
+    r1.mappings.return_value.all.return_value = id_rows or []
+    results.append(r1)
+
+    # 3: velocity (only queried if id_rows non-empty)
+    r2 = MagicMock()
+    r2.mappings.return_value.all.return_value = vel_rows or []
+    results.append(r2)
+
+    # 4: supply
+    r3 = MagicMock()
+    r3.mappings.return_value.all.return_value = supply_rows or []
+    results.append(r3)
+
+    db.execute.side_effect = results
+    return db
+
+
+def _override_db(db: MagicMock):
+    def _get_db_override():
+        yield db
+
+    return _get_db_override
+
+
+def _post(client: TestClient, cad: str = CAD_NUM, **body_kwargs) -> dict:
+    payload = {"radius_km": 1.0, "time_window": "last_quarter", **body_kwargs}
+    resp = client.post(f"/api/v1/parcels/{cad}/best-layouts", json=payload)
+    return resp
+
+
+# ── Тесты ─────────────────────────────────────────────────────────────────────
+
+
+def test_parcel_not_found_404() -> None:
+    """Если центроид не найден → 404."""
+    db = _make_db(coord=None)
+    from app.core.db import get_db
+
+    app.dependency_overrides[get_db] = _override_db(db)
+    try:
+        resp = _post(TestClient(app), cad="99:99:9999999:999")
+        assert resp.status_code == 404, resp.text
+    finally:
+        app.dependency_overrides.clear()
+
+
+def test_empty_competitor_set_returns_low_confidence() -> None:
+    """Нет конкурентов в радиусе → пустые top_layouts + confidence=low."""
+    db = _make_db(coord=_coord_row(), id_rows=[])
+    from app.core.db import get_db
+
+    app.dependency_overrides[get_db] = _override_db(db)
+    try:
+        resp = _post(TestClient(app))
+        assert resp.status_code == 200, resp.text
+        body = resp.json()
+        assert body["top_layouts"] == []
+        assert body["data_quality"]["confidence"] == "low"
+        assert body["data_quality"]["objects_total_in_radius"] == 0
+        rec = body["recommendation_for_tz"]
+        assert rec["based_on_obj_count"] == 0
+        assert rec["based_on_total_deals"] == 0
+        assert rec["mix"] == []
+    finally:
+        app.dependency_overrides.clear()
+
+
+def test_three_obj_ids_ranking_and_pct_sum_100() -> None:
+    """3 obj_id, 3 room_buckets — ranking по velocity, sum pct = 100."""
+    id_rows = [_obj_id_row(1), _obj_id_row(2), _obj_id_row(3)]
+    vel_rows = [
+        _vel_row("studio", sum_deals=8.0, avg_area=26.0, obj_ids=[1]),
+        _vel_row("1", sum_deals=32.0, avg_area=40.0, obj_ids=[2]),
+        _vel_row("2", sum_deals=48.0, avg_area=55.0, obj_ids=[3]),
+    ]
+    supply_rows = [
+        _supply_row("studio", "25-40", 20),
+        _supply_row("1", "40-60", 60),
+        _supply_row("2", "40-60", 80),
+    ]
+    db = _make_db(coord=_coord_row(), id_rows=id_rows, vel_rows=vel_rows, supply_rows=supply_rows)
+    from app.core.db import get_db
+
+    app.dependency_overrides[get_db] = _override_db(db)
+    try:
+        resp = _post(TestClient(app), time_window="last_quarter")
+        assert resp.status_code == 200, resp.text
+        body = resp.json()
+        top = body["top_layouts"]
+        assert len(top) == 3
+        # rank 1 = самая высокая velocity (2-комн: 48/8=6.0 per month)
+        assert top[0]["rank"] == 1
+        assert top[0]["room_bucket"] == "2"
+        # все ранги уникальны
+        assert sorted(t["rank"] for t in top) == [1, 2, 3]
+        # sum pct = 100
+        mix = body["recommendation_for_tz"]["mix"]
+        assert sum(m["pct"] for m in mix) == 100
+    finally:
+        app.dependency_overrides.clear()
+
+
+def test_exclude_competitor_obj_ids_filter() -> None:
+    """exclude_competitor_obj_ids исключает obj_id: при all excluded → пустой ответ."""
+    # Если после исключения obj_id_list пуст → _empty_response → top_layouts=[]
+    id_rows = [_obj_id_row(20)]  # единственный конкурент
+    db = _make_db(coord=_coord_row(), id_rows=id_rows, vel_rows=[])
+    from app.core.db import get_db
+
+    app.dependency_overrides[get_db] = _override_db(db)
+    try:
+        resp = _post(TestClient(app), exclude_competitor_obj_ids=[20])
+        assert resp.status_code == 200, resp.text
+        body = resp.json()
+        # После исключения obj_id=20 список пуст → пустой ответ
+        assert body["top_layouts"] == []
+        assert body["data_quality"]["confidence"] == "low"
+        # objects_total_in_radius = 1 (до исключения)
+        assert body["data_quality"]["objects_total_in_radius"] == 1
+    finally:
+        app.dependency_overrides.clear()
+
+
+def test_min_velocity_per_month_filters_low_rows() -> None:
+    """min_velocity_per_month=5 → строки с velocity<5 не попадают в top_layouts."""
+    id_rows = [_obj_id_row(1), _obj_id_row(2)]
+    # last_quarter divisor=8 → 16/8=2.0 (ниже порога), 80/8=10.0 (выше)
+    vel_rows = [
+        _vel_row("studio", sum_deals=16.0, obj_ids=[1]),
+        _vel_row("1", sum_deals=80.0, obj_ids=[2]),
+    ]
+    db = _make_db(coord=_coord_row(), id_rows=id_rows, vel_rows=vel_rows)
+    from app.core.db import get_db
+
+    app.dependency_overrides[get_db] = _override_db(db)
+    try:
+        resp = _post(TestClient(app), min_velocity_per_month=5.0)
+        assert resp.status_code == 200, resp.text
+        body = resp.json()
+        top = body["top_layouts"]
+        assert len(top) == 1
+        assert top[0]["room_bucket"] == "1"
+        assert top[0]["velocity_per_month"] == pytest.approx(10.0)
+    finally:
+        app.dependency_overrides.clear()
+
+
+def test_time_window_velocity_scaling() -> None:
+    """last_month vs last_year дают разный velocity_per_month для одних deals."""
+    # sum_deals=24 → last_month: 24/24=1.0, last_year: 24/2=12.0
+    id_rows = [_obj_id_row(1)]
+    vel_rows_fixed = [_vel_row("2", sum_deals=24.0, obj_ids=[1])]
+
+    from app.core.db import get_db
+
+    # last_month
+    db_m = _make_db(coord=_coord_row(), id_rows=id_rows, vel_rows=vel_rows_fixed)
+    app.dependency_overrides[get_db] = _override_db(db_m)
+    try:
+        resp_m = _post(TestClient(app), time_window="last_month")
+        assert resp_m.status_code == 200, resp_m.text
+        v_month = resp_m.json()["top_layouts"][0]["velocity_per_month"]
+    finally:
+        app.dependency_overrides.clear()
+
+    # last_year
+    db_y = _make_db(coord=_coord_row(), id_rows=id_rows, vel_rows=vel_rows_fixed)
+    app.dependency_overrides[get_db] = _override_db(db_y)
+    try:
+        resp_y = _post(TestClient(app), time_window="last_year")
+        assert resp_y.status_code == 200, resp_y.text
+        v_year = resp_y.json()["top_layouts"][0]["velocity_per_month"]
+    finally:
+        app.dependency_overrides.clear()
+
+    # last_year velocity должна быть выше (делитель меньше: 2 vs 24)
+    assert v_year > v_month
+    assert v_month == pytest.approx(1.0)
+    assert v_year == pytest.approx(12.0)
+
+
+def test_obj_class_filter_passes_through() -> None:
+    """obj_class_filter передаётся в SQL — endpoint не ломается, возвращает 200."""
+    db = _make_db(
+        coord=_coord_row(),
+        id_rows=[_obj_id_row(5)],
+        vel_rows=[_vel_row("2", obj_ids=[5])],
+        supply_rows=[],
+    )
+    from app.core.db import get_db
+
+    app.dependency_overrides[get_db] = _override_db(db)
+    try:
+        resp = _post(TestClient(app), obj_class_filter="comfort")
+        assert resp.status_code == 200, resp.text
+        body = resp.json()
+        assert len(body["top_layouts"]) > 0
+    finally:
+        app.dependency_overrides.clear()
+
+
+def test_mv_empty_for_competitors_returns_empty_top_layouts() -> None:
+    """Конкуренты есть в радиусе, но MV пустой → top_layouts=[], data_quality.confidence=low."""
+    id_rows = [_obj_id_row(1), _obj_id_row(2)]
+    db = _make_db(coord=_coord_row(), id_rows=id_rows, vel_rows=[])
+    from app.core.db import get_db
+
+    app.dependency_overrides[get_db] = _override_db(db)
+    try:
+        resp = _post(TestClient(app))
+        assert resp.status_code == 200, resp.text
+        body = resp.json()
+        assert body["top_layouts"] == []
+        dq = body["data_quality"]
+        assert dq["objects_total_in_radius"] == 2
+        assert dq["objects_with_velocity_data"] == 0
+        assert dq["confidence"] == "low"
+    finally:
+        app.dependency_overrides.clear()
+
+
+def test_target_total_flats_fills_abs_units() -> None:
+    """target_total_flats=100 → abs_units заполнен в mix, sum примерно = 100."""
+    id_rows = [_obj_id_row(1), _obj_id_row(2)]
+    vel_rows = [
+        _vel_row("1", sum_deals=60.0, obj_ids=[1]),
+        _vel_row("2", sum_deals=40.0, obj_ids=[2]),
+    ]
+    db = _make_db(coord=_coord_row(), id_rows=id_rows, vel_rows=vel_rows)
+    from app.core.db import get_db
+
+    app.dependency_overrides[get_db] = _override_db(db)
+    try:
+        resp = _post(TestClient(app), target_total_flats=100)
+        assert resp.status_code == 200, resp.text
+        mix = resp.json()["recommendation_for_tz"]["mix"]
+        # все abs_units заполнены
+        for m in mix:
+            assert m["abs_units"] is not None
+        # сумма abs_units близка к 100 (round-off ±1)
+        total_abs = sum(m["abs_units"] for m in mix)
+        assert 98 <= total_abs <= 102
+    finally:
+        app.dependency_overrides.clear()
+
+
+def test_filter_competitor_obj_ids_applied() -> None:
+    """filter_competitor_obj_ids=[1] оставляет только obj_id=1."""
+    id_rows = [_obj_id_row(1), _obj_id_row(2), _obj_id_row(3)]
+    # После фильтрации остаётся только obj_id=1, velocity запрос получит [1]
+    vel_rows = [_vel_row("2", sum_deals=24.0, obj_ids=[1])]
+    db = _make_db(coord=_coord_row(), id_rows=id_rows, vel_rows=vel_rows)
+    from app.core.db import get_db
+
+    app.dependency_overrides[get_db] = _override_db(db)
+    try:
+        resp = _post(TestClient(app), filter_competitor_obj_ids=[1])
+        assert resp.status_code == 200, resp.text
+        body = resp.json()
+        top = body["top_layouts"]
+        assert len(top) >= 1
+        # competitor_obj_ids должен содержать только 1
+        for row in top:
+            for oid in row["competitor_obj_ids"]:
+                assert oid == 1
+    finally:
+        app.dependency_overrides.clear()