backend/app/api/v1/parcels.py

@@ -27,6 +27,7 @@ from app.services.site_finder.quarter_dump_lookup import (
     get_quarter_dump_data,
     make_empty_result,
 )
+from app.services.site_finder.velocity import compute_velocity
 
 logger = logging.getLogger(__name__)
 
@@ -1806,6 +1807,15 @@ def analyze_parcel(
     # D4 (#36): aggregate pipeline_24mo
     pipeline_24mo = _aggregate_pipeline(pipeline_rows)
 
+    # D2 (#34): velocity-score — темп продаж конкурентов вокруг участка.
+    velocity_data: dict[str, Any] | None = None
+    try:
+        v_result = compute_velocity(db, parcel_geom_wkt=geom_wkt, radius_km=3.0)
+        if v_result is not None:
+            velocity_data = v_result.as_dict()
+    except Exception as _ve:
+        logger.warning("velocity compute failed for %s: %s", cad_num, _ve)
+
     return {
         "cad_num": cad_num,
         "source": source,
@@ -1835,6 +1845,8 @@ def analyze_parcel(
         "competitors": [dict(c) for c in competitor_rows],
         # D4 (#36): 24-month pipeline competition
         "pipeline_24mo": pipeline_24mo,
+        # D2 (#34): velocity-score из domrf_kn_sale_graph
+        "velocity": velocity_data,
         "noise": {
             "score": round(noise_score, 2),
             "estimated_db": round(noise_db_max, 1),

backend/app/services/site_finder/velocity.py

@@ -0,0 +1,318 @@
+"""Velocity-score — темп продаж конкурентов вокруг участка.
+
+Per #34 D2: утилизация domrf_kn_sale_graph (15876 строк).
+Главный demand-сигнал «продастся ли» — среднемесячный объём продаж
+конкурирующих ЖК в радиусе radius_km от участка, нормированный к
+ЕКБ-медиане по region_cd=66.
+
+Foundation: domrf_kn_objects (lat/lon, comm_name, obj_class, region_cd),
+            domrf_kn_sale_graph (obj_id, report_month, area_sq, realised, type).
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from typing import Any, Literal
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+logger = logging.getLogger(__name__)
+
+# Fallback если в БД нет данных за окно months_window.
+# Эмпирика по ЕКБ: ~4 500 м²/мес на один ЖК (apartments, 2024-2025).
+_EKB_MEDIAN_FALLBACK_SQM_PER_MONTH: float = 4500.0
+
+
+@dataclass(frozen=True)
+class VelocityResult:
+    """Результат расчёта velocity-score для участка."""
+
+    competitors_count: int
+    monthly_velocity_sqm: float  # avg м²/мес по конкурентам в радиусе
+    ekb_median_sqm: float  # benchmark ЕКБ для нормализации
+    velocity_score: float  # 0..1 — отношение к benchmark
+    confidence: Literal["high", "medium", "low"]
+    months_observed: int  # фактически использованных месяцев
+    period_start: str  # YYYY-MM
+    period_end: str  # YYYY-MM
+    sample_competitors: list[dict[str, Any]]  # top-5 для UI
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "competitors_count": self.competitors_count,
+            "monthly_velocity_sqm": round(self.monthly_velocity_sqm, 1),
+            "ekb_median_sqm": round(self.ekb_median_sqm, 1),
+            "velocity_score": round(self.velocity_score, 3),
+            "confidence": self.confidence,
+            "months_observed": self.months_observed,
+            "period": {"start": self.period_start, "end": self.period_end},
+            "sample_competitors": self.sample_competitors,
+        }
+
+
+def compute_velocity(
+    db: Session,
+    parcel_geom_wkt: str,
+    radius_km: float = 3.0,
+    obj_class: str | None = None,
+    months_window: int = 6,
+) -> VelocityResult | None:
+    """Вычислить velocity-score для участка.
+
+    Алгоритм:
+    1. Найти ЖК-конкуренты в радиусе radius_km (через lat/lon ST_DWithin).
+    2. Взять sale_graph за последние months_window месяцев (latest snapshot).
+    3. Посчитать суммарный объём (area_sq > 0, иначе realised * avg_area).
+    4. Нормировать на ЕКБ-медиану → score 0..1.
+
+    Возвращает None если parcel_geom_wkt невалиден или конкурентов нет.
+    """
+    # ── Step 1: конкуренты по lat/lon в радиусе ──────────────────────────────
+    # DISTINCT ON (obj_id) ORDER BY snapshot_date DESC — latest snapshot only.
+    # obj_class в domrf_kn_objects заполнен слабо (много NULL); фильтруем
+    # только если явно передан.
+    class_filter = "AND o.obj_class = :obj_class" if obj_class else ""
+    try:
+        comp_rows = (
+            db.execute(
+                text(
+                    f"""
+                    WITH latest_obj AS (
+                        SELECT DISTINCT ON (obj_id)
+                               obj_id,
+                               comm_name,
+                               dev_name,
+                               obj_class,
+                               latitude,
+                               longitude,
+                               district_name
+                          FROM domrf_kn_objects
+                         WHERE latitude IS NOT NULL
+                           AND longitude IS NOT NULL
+                           AND region_cd = 66
+                           {class_filter}
+                         ORDER BY obj_id, snapshot_date DESC NULLS LAST
+                    )
+                    SELECT
+                        o.obj_id,
+                        o.comm_name,
+                        o.dev_name,
+                        o.obj_class,
+                        o.district_name,
+                        ST_Distance(
+                            ST_SetSRID(ST_MakePoint(o.longitude, o.latitude), 4326)::geography,
+                            ST_Centroid(ST_GeomFromText(:parcel_wkt, 4326))::geography
+                        ) AS distance_m
+                      FROM latest_obj o
+                     WHERE ST_DWithin(
+                               ST_SetSRID(ST_MakePoint(o.longitude, o.latitude), 4326)::geography,
+                               ST_Centroid(ST_GeomFromText(:parcel_wkt, 4326))::geography,
+                               :radius_m
+                           )
+                     ORDER BY distance_m ASC
+                     LIMIT 200
+                    """
+                ),
+                {
+                    "parcel_wkt": parcel_geom_wkt,
+                    "radius_m": radius_km * 1000.0,
+                    "obj_class": obj_class,
+                },
+            )
+            .mappings()
+            .all()
+        )
+    except Exception:
+        logger.exception("velocity: competitor query failed for wkt=%s", parcel_geom_wkt[:80])
+        return None
+
+    if not comp_rows:
+        return None
+
+    obj_ids: list[int] = [int(r["obj_id"]) for r in comp_rows]
+    competitor_meta: dict[int, dict[str, Any]] = {
+        int(r["obj_id"]): {
+            "name": r["comm_name"],
+            "dev_name": r["dev_name"],
+            "obj_class": r["obj_class"],
+            "district_name": r["district_name"],
+            "distance_m": round(float(r["distance_m"]), 0),
+        }
+        for r in comp_rows
+    }
+
+    # ── Step 2: sale_graph за последние N месяцев (latest snapshot per obj) ──
+    # area_sq = м² за месяц (primary). Если NULL — realised * 45 м² heuristic.
+    # type = 'apartments' — только жильё.
+    try:
+        sales_rows = (
+            db.execute(
+                text(
+                    """
+                    WITH latest_sg AS (
+                        SELECT DISTINCT ON (obj_id, report_month)
+                               obj_id,
+                               report_month,
+                               area_sq,
+                               realised
+                          FROM domrf_kn_sale_graph
+                         WHERE obj_id = ANY(:obj_ids)
+                           AND type = 'apartments'
+                           AND report_month >= (CURRENT_DATE - :window_interval::interval)
+                         ORDER BY obj_id, report_month, snapshot_date DESC NULLS LAST
+                    )
+                    SELECT
+                        obj_id,
+                        SUM(
+                            COALESCE(area_sq, realised * 45.0)
+                        )                                                AS total_sqm,
+                        COUNT(DISTINCT report_month)                     AS months_with_data,
+                        MIN(report_month)                                AS period_start,
+                        MAX(report_month)                                AS period_end
+                      FROM latest_sg
+                     WHERE area_sq > 0 OR realised > 0
+                     GROUP BY obj_id
+                    """
+                ),
+                {
+                    "obj_ids": obj_ids,
+                    "window_interval": f"{months_window} months",
+                },
+            )
+            .mappings()
+            .all()
+        )
+    except Exception:
+        logger.exception("velocity: sale_graph query failed for obj_ids=%s", obj_ids[:5])
+        return None
+
+    if not sales_rows:
+        return None
+
+    total_sqm = sum(float(r["total_sqm"] or 0.0) for r in sales_rows)
+    months_observed = max((int(r["months_with_data"] or 0) for r in sales_rows), default=0)
+    period_start_dates = [r["period_start"] for r in sales_rows if r["period_start"]]
+    period_end_dates = [r["period_end"] for r in sales_rows if r["period_end"]]
+    period_start = min(period_start_dates).strftime("%Y-%m") if period_start_dates else ""
+    period_end = max(period_end_dates).strftime("%Y-%m") if period_end_dates else ""
+
+    if months_observed == 0 or total_sqm <= 0:
+        return None
+
+    # Среднемесячный объём в расчёте: суммарный по всем конкурентам / месяцев.
+    # Чем больше конкурентов с данными — тем весомее результат.
+    monthly_velocity = total_sqm / months_observed
+
+    # ── Step 3: ЕКБ-медиана ──────────────────────────────────────────────────
+    ekb_median = (
+        _get_ekb_median(db, months_window=months_window) or _EKB_MEDIAN_FALLBACK_SQM_PER_MONTH
+    )
+
+    # ── Step 4: нормализация → score 0..1 ────────────────────────────────────
+    # Логика: сравниваем суммарный velocity радиуса с «нормой» одного ЖК.
+    # Если в радиусе продаётся N × ekb_median → рынок горячий.
+    # Нормируем: score = min(1.0, total_velocity / (n_competitors × ekb_median × 2))
+    # Cap 2×median = «насыщен». Итоговый score 0..1.
+    n_with_sales = len(sales_rows)
+    denominator = n_with_sales * ekb_median * 2.0 if n_with_sales > 0 else ekb_median * 2.0
+    velocity_score = min(1.0, max(0.0, monthly_velocity / denominator))
+
+    # ── Step 5: confidence ───────────────────────────────────────────────────
+    n_comps = len(comp_rows)
+    if n_comps >= 10 and months_observed >= 5:
+        confidence: Literal["high", "medium", "low"] = "high"
+    elif n_comps >= 5 and months_observed >= 3:
+        confidence = "medium"
+    else:
+        confidence = "low"
+
+    # ── Step 6: top-5 конкурентов по объёму продаж ───────────────────────────
+    sales_by_id: dict[int, float] = {
+        int(r["obj_id"]): float(r["total_sqm"] or 0.0) for r in sales_rows
+    }
+    sample = sorted(
+        [
+            {
+                "obj_id": oid,
+                **competitor_meta[oid],
+                "total_sqm_period": round(sales_by_id.get(oid, 0.0), 0),
+            }
+            for oid in obj_ids
+            if oid in competitor_meta
+        ],
+        key=lambda x: x["total_sqm_period"],
+        reverse=True,
+    )[:5]
+
+    return VelocityResult(
+        competitors_count=n_comps,
+        monthly_velocity_sqm=monthly_velocity,
+        ekb_median_sqm=ekb_median,
+        velocity_score=velocity_score,
+        confidence=confidence,
+        months_observed=months_observed,
+        period_start=period_start,
+        period_end=period_end,
+        sample_competitors=sample,
+    )
+
+
+def _get_ekb_median(db: Session, months_window: int = 6) -> float | None:
+    """ЕКБ-wide медиана monthly velocity (м²/мес) per ЖК — benchmark.
+
+    Учитываются только ЖК с ≥3 месяцами данных за окно (стабильный сигнал).
+    Fallback к _EKB_MEDIAN_FALLBACK_SQM_PER_MONTH если нет данных в БД.
+    """
+    try:
+        row = (
+            db.execute(
+                text(
+                    """
+                    WITH latest_sg AS (
+                        SELECT DISTINCT ON (obj_id, report_month)
+                               obj_id,
+                               area_sq,
+                               realised,
+                               report_month
+                          FROM domrf_kn_sale_graph sg
+                         WHERE sg.type = 'apartments'
+                           AND sg.report_month >= (CURRENT_DATE - :window_interval::interval)
+                           AND EXISTS (
+                               SELECT 1 FROM domrf_kn_objects o
+                                WHERE o.obj_id = sg.obj_id
+                                  AND o.region_cd = 66
+                           )
+                         ORDER BY obj_id, report_month, snapshot_date DESC NULLS LAST
+                    ),
+                    per_obj AS (
+                        SELECT
+                            obj_id,
+                            SUM(COALESCE(area_sq, realised * 45.0)) AS total_sqm,
+                            COUNT(DISTINCT report_month)             AS months_data
+                          FROM latest_sg
+                         WHERE area_sq > 0 OR realised > 0
+                         GROUP BY obj_id
+                        HAVING COUNT(DISTINCT report_month) >= 3
+                    ),
+                    per_obj_velocity AS (
+                        SELECT total_sqm / months_data AS velocity
+                          FROM per_obj
+                    )
+                    SELECT PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY velocity) AS median
+                      FROM per_obj_velocity
+                    """
+                ),
+                {"window_interval": f"{months_window} months"},
+            )
+            .mappings()
+            .first()
+        )
+    except Exception:
+        logger.warning("velocity: ekb_median query failed, using fallback")
+        return None
+
+    if row and row["median"] is not None:
+        return float(row["median"])
+    return None

backend/tests/test_velocity.py

@@ -0,0 +1,234 @@
+"""Tests for velocity-score service (#34 D2).
+
+Mock-based — не требуют живой БД.
+"""
+
+from __future__ import annotations
+
+import datetime
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from app.services.site_finder.velocity import (
+    _EKB_MEDIAN_FALLBACK_SQM_PER_MONTH,
+    VelocityResult,
+    compute_velocity,
+)
+
+# Тестовый WKT — небольшой квадрат в центре ЕКБ.
+_PARCEL_WKT = "POINT(60.605 56.838)"
+
+
+# ── Вспомогательные фабрики mock-строк ────────────────────────────────────────
+
+
+def _comp_row(obj_id: int, distance_m: float = 500.0) -> MagicMock:
+    r = MagicMock()
+    r.__getitem__ = lambda self, k: {
+        "obj_id": obj_id,
+        "comm_name": f"ЖК-{obj_id}",
+        "dev_name": "TestDev",
+        "obj_class": "комфорт",
+        "district_name": "Ленинский",
+        "distance_m": distance_m,
+    }[k]
+    return r
+
+
+def _sales_row(
+    obj_id: int,
+    total_sqm: float,
+    months: int,
+    start: str = "2024-11-01",
+    end: str = "2025-04-01",
+) -> MagicMock:
+    r = MagicMock()
+    start_d = datetime.date.fromisoformat(start)
+    end_d = datetime.date.fromisoformat(end)
+    r.__getitem__ = lambda self, k: {
+        "obj_id": obj_id,
+        "total_sqm": total_sqm,
+        "months_with_data": months,
+        "period_start": start_d,
+        "period_end": end_d,
+    }[k]
+    return r
+
+
+def _make_db(comp_rows: list, sales_rows: list) -> MagicMock:
+    """Сконструировать mock Session с двумя последовательными вызовами execute."""
+    db = MagicMock()
+    execute_results = []
+    for rows in [comp_rows, sales_rows]:
+        result = MagicMock()
+        result.mappings.return_value.all.return_value = rows
+        execute_results.append(result)
+    db.execute.side_effect = execute_results
+    return db
+
+
+# ── Тесты ─────────────────────────────────────────────────────────────────────
+
+
+def test_no_competitors_returns_none():
+    """Нет ЖК в радиусе → None."""
+    db = _make_db(comp_rows=[], sales_rows=[])
+    result = compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT)
+    assert result is None
+
+
+def test_no_sales_data_returns_none():
+    """ЖК есть, но нет данных sale_graph → None."""
+    comp_rows = [_comp_row(1), _comp_row(2)]
+    db = _make_db(comp_rows=comp_rows, sales_rows=[])
+    result = compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT)
+    assert result is None
+
+
+def test_healthy_sales_returns_result():
+    """12 конкурентов с нормальными продажами → score в (0,1), confidence='high'."""
+    n = 12
+    comp_rows = [_comp_row(i, distance_m=300.0 + i * 100) for i in range(1, n + 1)]
+    # Каждый ЖК продаёт 4500 м² за 6 мес → 750 м²/мес. Суммарно: 4500*12 = 54000 за 6 мес.
+    sales_rows = [_sales_row(i, total_sqm=4500.0, months=6) for i in range(1, n + 1)]
+
+    db = _make_db(comp_rows=comp_rows, sales_rows=sales_rows)
+
+    with patch(
+        "app.services.site_finder.velocity._get_ekb_median",
+        return_value=_EKB_MEDIAN_FALLBACK_SQM_PER_MONTH,
+    ):
+        result = compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT)
+
+    assert result is not None
+    assert result.competitors_count == n
+    assert 0.0 < result.velocity_score <= 1.0
+    assert result.confidence == "high"
+    assert result.months_observed == 6
+
+
+def test_few_competitors_low_confidence():
+    """2 конкурента → confidence='low'."""
+    comp_rows = [_comp_row(1), _comp_row(2)]
+    sales_rows = [_sales_row(1, total_sqm=3000.0, months=2)]
+
+    db = _make_db(comp_rows=comp_rows, sales_rows=sales_rows)
+
+    with patch(
+        "app.services.site_finder.velocity._get_ekb_median",
+        return_value=_EKB_MEDIAN_FALLBACK_SQM_PER_MONTH,
+    ):
+        result = compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT)
+
+    assert result is not None
+    assert result.confidence == "low"
+
+
+def test_medium_confidence():
+    """7 конкурентов, 4 месяца → confidence='medium'."""
+    n = 7
+    comp_rows = [_comp_row(i) for i in range(1, n + 1)]
+    sales_rows = [_sales_row(i, total_sqm=4000.0, months=4) for i in range(1, n + 1)]
+
+    db = _make_db(comp_rows=comp_rows, sales_rows=sales_rows)
+
+    with patch(
+        "app.services.site_finder.velocity._get_ekb_median",
+        return_value=_EKB_MEDIAN_FALLBACK_SQM_PER_MONTH,
+    ):
+        result = compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT)
+
+    assert result is not None
+    assert result.confidence == "medium"
+
+
+def test_ekb_median_fallback_used_when_none():
+    """Если _get_ekb_median вернул None — используется fallback-константа."""
+    comp_rows = [_comp_row(1)]
+    sales_rows = [_sales_row(1, total_sqm=9000.0, months=6)]
+
+    db = _make_db(comp_rows=comp_rows, sales_rows=sales_rows)
+
+    with patch("app.services.site_finder.velocity._get_ekb_median", return_value=None):
+        result = compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT)
+
+    assert result is not None
+    assert result.ekb_median_sqm == _EKB_MEDIAN_FALLBACK_SQM_PER_MONTH
+
+
+def test_score_capped_at_1():
+    """Огромный объём → score не превышает 1.0."""
+    comp_rows = [_comp_row(1)]
+    # 1 000 000 м² за месяц — абсурдно много
+    sales_rows = [_sales_row(1, total_sqm=6_000_000.0, months=6)]
+
+    db = _make_db(comp_rows=comp_rows, sales_rows=sales_rows)
+
+    with patch(
+        "app.services.site_finder.velocity._get_ekb_median",
+        return_value=_EKB_MEDIAN_FALLBACK_SQM_PER_MONTH,
+    ):
+        result = compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT)
+
+    assert result is not None
+    assert result.velocity_score == pytest.approx(1.0)
+
+
+def test_score_zero_when_no_sales_sqm():
+    """total_sqm=0 → None (нет данных, не score=0)."""
+    comp_rows = [_comp_row(1)]
+    # total_sqm=0 — нет продаж → должен вернуть None
+    sales_rows = [_sales_row(1, total_sqm=0.0, months=5)]
+
+    db = _make_db(comp_rows=comp_rows, sales_rows=sales_rows)
+
+    with patch(
+        "app.services.site_finder.velocity._get_ekb_median",
+        return_value=_EKB_MEDIAN_FALLBACK_SQM_PER_MONTH,
+    ):
+        result = compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT)
+
+    assert result is None
+
+
+def test_as_dict_structure():
+    """as_dict() содержит все ожидаемые ключи."""
+    vr = VelocityResult(
+        competitors_count=5,
+        monthly_velocity_sqm=3000.0,
+        ekb_median_sqm=4500.0,
+        velocity_score=0.333,
+        confidence="medium",
+        months_observed=4,
+        period_start="2024-11",
+        period_end="2025-02",
+        sample_competitors=[],
+    )
+    d = vr.as_dict()
+    assert "competitors_count" in d
+    assert "velocity_score" in d
+    assert "confidence" in d
+    assert "period" in d
+    assert d["period"]["start"] == "2024-11"
+    assert d["period"]["end"] == "2025-02"
+    assert d["velocity_score"] == pytest.approx(0.333, abs=1e-3)
+
+
+def test_sample_competitors_top5():
+    """sample_competitors содержит не более 5 элементов, отсортированных по убыванию."""
+    n = 8
+    comp_rows = [_comp_row(i) for i in range(1, n + 1)]
+    sales_rows = [_sales_row(i, total_sqm=float(i * 1000), months=5) for i in range(1, n + 1)]
+    db = _make_db(comp_rows=comp_rows, sales_rows=sales_rows)
+
+    with patch(
+        "app.services.site_finder.velocity._get_ekb_median",
+        return_value=_EKB_MEDIAN_FALLBACK_SQM_PER_MONTH,
+    ):
+        result = compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT)
+
+    assert result is not None
+    assert len(result.sample_competitors) <= 5
+    sqms = [c["total_sqm_period"] for c in result.sample_competitors]
+    assert sqms == sorted(sqms, reverse=True)