feat(tradein): GET /street-deals — ДКП-сделки Росреестра по улице target адреса (#555)

tradein-mvp/backend/app/api/v1/trade_in.py

@@ -6,7 +6,7 @@
 from __future__ import annotations
 
 import logging
-from datetime import UTC, datetime
+from datetime import UTC, date, datetime, timedelta
 from typing import Annotated, Any
 from uuid import UUID
 
@@ -29,6 +29,7 @@ from app.schemas.trade_in import (
     RecentSoldEntry,
     SellTimeBucket,
     SellTimeSensitivityResponse,
+    StreetDealsResponse,
     TradeInEstimateInput,
 )
 from app.services.exporters.trade_in_pdf import generate_trade_in_pdf
@@ -1063,3 +1064,123 @@ def get_estimate_imv_benchmark(
         our_median_price=our_median,
         diff_pct=diff_pct,
     )
+
+
+# ── Street-level deals (rosreestr open dataset) ───────────────────────────────
+
+
+@router.get("/street-deals", response_model=StreetDealsResponse)
+def get_street_deals(
+    address: str,
+    area_m2: float,
+    rooms: int,
+    db: Annotated[Session, Depends(get_db)] = None,  # type: ignore[assignment]
+    period_months: int = 12,
+    area_tolerance: float = 0.15,
+) -> StreetDealsResponse:
+    """ДКП-сделки Росреестра по улице целевого адреса.
+
+    Open dataset Росреестра агрегирует адреса до улицы (без номера дома).
+    Поэтому это per-street view, не per-house. Фильтр по rooms + area
+    сужает выборку до квартир-аналогов.
+
+    После PR-A (#549) таблица deals содержит только ДКП (ДДУ-первичка отфильтрована
+    в import-rosreestr.sh).
+    """
+    from app.services.estimator import _deal_to_analog, _percentile, extract_street_name
+
+    now = datetime.now(tz=UTC)
+    period_from: date = (now - timedelta(days=period_months * 30)).date()
+    period_to: date = now.date()
+
+    def _empty() -> StreetDealsResponse:
+        return StreetDealsResponse(
+            street=None,
+            period_from=period_from,
+            period_to=period_to,
+            count=0,
+            median_price_rub=0,
+            median_price_per_m2=0,
+            range_low_rub=0,
+            range_high_rub=0,
+            deals=[],
+        )
+
+    street_name = extract_street_name(address)
+    if not street_name:
+        logger.warning("street-deals: could not extract street from address=%r", address)
+        return _empty()
+
+    area_min = area_m2 * (1.0 - area_tolerance)
+    area_max = area_m2 * (1.0 + area_tolerance)
+
+    rows = db.execute(
+        text(
+            """
+            SELECT address, area_m2, rooms, floor, total_floors,
+                   price_rub, price_per_m2, deal_date, source
+            FROM deals
+            WHERE source = 'rosreestr'
+              AND address ILIKE :street_pattern
+              AND rooms = CAST(:rooms AS integer)
+              AND area_m2 BETWEEN :area_min AND :area_max
+              AND deal_date > NOW() - (CAST(:period_months AS integer) || ' months')::interval
+              AND price_rub > 0
+            ORDER BY deal_date DESC
+            """
+        ),
+        {
+            "street_pattern": "%" + street_name + "%",
+            "rooms": rooms,
+            "area_min": area_min,
+            "area_max": area_max,
+            "period_months": period_months,
+        },
+    ).mappings().all()
+
+    if not rows:
+        logger.info(
+            "street-deals: no rows found street=%r rooms=%d area=%.1f±%.0f%%",
+            street_name, rooms, area_m2, area_tolerance * 100,
+        )
+        return StreetDealsResponse(
+            street=street_name,
+            period_from=period_from,
+            period_to=period_to,
+            count=0,
+            median_price_rub=0,
+            median_price_per_m2=0,
+            range_low_rub=0,
+            range_high_rub=0,
+            deals=[],
+        )
+
+    count = len(rows)
+    prices_rub = sorted(float(r["price_rub"]) for r in rows)
+    prices_ppm2 = sorted(float(r["price_per_m2"]) for r in rows if r["price_per_m2"])
+
+    median_ppm2 = _percentile(prices_ppm2, 0.5) if prices_ppm2 else 0.0
+    median_price_rub = (
+        int(median_ppm2 * area_m2) if median_ppm2 else int(_percentile(prices_rub, 0.5))
+    )
+    range_low_rub = int(prices_rub[0])
+    range_high_rub = int(prices_rub[-1])
+
+    top10 = [_deal_to_analog(dict(r)) for r in rows[:10]]
+
+    logger.info(
+        "street-deals: street=%r rooms=%d area=%.1f count=%d median_ppm2=%.0f",
+        street_name, rooms, area_m2, count, median_ppm2,
+    )
+
+    return StreetDealsResponse(
+        street=street_name,
+        period_from=period_from,
+        period_to=period_to,
+        count=count,
+        median_price_rub=median_price_rub,
+        median_price_per_m2=int(median_ppm2),
+        range_low_rub=range_low_rub,
+        range_high_rub=range_high_rub,
+        deals=top10,
+    )

tradein-mvp/backend/app/schemas/trade_in.py

@@ -286,3 +286,25 @@ class SellTimeSensitivityResponse(BaseModel):
     radius_m: int
     target_median_price_per_m2: int | None  # benchmark — медиана ₽/м² за последние 2 года
     buckets: list[SellTimeBucket]
+
+
+# ── Street-level deals (rosreestr open dataset) ──────────────────────────────
+
+
+class StreetDealsResponse(BaseModel):
+    """Ответ GET /api/v1/trade-in/street-deals.
+
+    Open dataset Росреестра агрегирует адреса до уровня улицы (без номера дома).
+    Поэтому это per-street view, а не per-house.
+    """
+
+    # извлечённая улица, напр. «Космонавтов» / None если не определилась
+    street: str | None
+    period_from: date
+    period_to: date
+    count: int                      # число всех matching сделок, не только топ-10
+    median_price_rub: int           # 0 если count == 0
+    median_price_per_m2: int
+    range_low_rub: int
+    range_high_rub: int
+    deals: list[AnalogLot]          # последние 10 по deal_date DESC

tradein-mvp/backend/app/services/estimator.py

@@ -1112,6 +1112,51 @@ def _extract_short_addr(full_address: str | None) -> str | None:
     return s.strip(" ,.") or None
 
 
+# Strips trailing house-number and any subsequent parts from a short address.
+_TRAILING_HOUSE_RE = re.compile(
+    r",\s*\d+.*$",
+    flags=re.UNICODE,
+)
+
+# Strips leading street-type prefix (ул., проспект, пер., etc.) from a street token.
+_STREET_PREFIX_RE = re.compile(
+    r"^(?:ул\.|улица|пр\.|пр-т|проспект|пер\.|переулок|"
+    r"б-р|бульвар|ш\.|шоссе|наб\.|набережная|проезд|тракт|пл\.|площадь|"
+    r"мкр\.?|микрорайон)\s+",
+    flags=re.IGNORECASE | re.UNICODE,
+)
+
+
+def extract_street_name(full_address: str | None) -> str | None:
+    """Извлекает только название улицы (без типа и номера дома) из полного адреса.
+
+    Используется для запросов в open dataset Росреестра, где адреса агрегированы
+    до уровня улицы (без номера дома).
+
+    Примеры:
+      "Свердловская обл., г. Екатеринбург, ул. Космонавтов, 12"  → "Космонавтов"
+      "Екатеринбург, проспект Ленина, 50"                         → "Ленина"
+      "Екатеринбург, ул. 8 Марта, 18"                             → "8 Марта"
+      "г. Екатеринбург, пер. Красный, 4"                          → "Красный"
+
+    Алгоритм:
+      1. _extract_short_addr → «ул. Космонавтов, 12» (или fallback строка).
+      2. Отрезаем trailing «, 12...» (house number и всё после).
+      3. Отрезаем leading street-type prefix («ул.», «проспект» и т.д.).
+      4. None если ничего не осталось.
+    """
+    short = _extract_short_addr(full_address)
+    if not short:
+        return None
+
+    # Strip trailing house number: ", 12" / ", 12А" / ", 48/2"
+    without_house = _TRAILING_HOUSE_RE.sub("", short).strip(" ,.")
+
+    # Strip leading type prefix
+    street_name = _STREET_PREFIX_RE.sub("", without_house).strip()
+    return street_name or None
+
+
 def _stratify_candidates(candidates: list[dict[str, Any]]) -> list[dict[str, Any]]:
     """Стратифицированная выборка Approach B — гарантирует MIN_ANALOGS_PER_SOURCE слотов.
 

tradein-mvp/backend/tests/test_street_deals_endpoint.py

@@ -0,0 +1,209 @@
+"""Tests for GET /api/v1/trade-in/street-deals endpoint.
+
+Covers:
+  - extract_street_name parsing (various address formats)
+  - empty result when no DB rows match
+  - correct aggregation (median/count) with fixture deals
+"""
+
+import os
+import sys
+from unittest.mock import MagicMock
+
+# psycopg v3 driver (psycopg2 not installed)
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+# WeasyPrint requires GTK — not present in CI/Windows. Stub before any app import.
+_wp_mock = MagicMock()
+sys.modules.setdefault("weasyprint", _wp_mock)
+sys.modules.setdefault("weasyprint.CSS", _wp_mock)
+sys.modules.setdefault("weasyprint.HTML", _wp_mock)
+
+from datetime import date  # noqa: E402
+
+import pytest  # noqa: E402
+from fastapi import FastAPI  # noqa: E402
+from fastapi.testclient import TestClient  # noqa: E402
+
+# ── Unit: extract_street_name ─────────────────────────────────────────────────
+
+
+def test_extract_street_name_basic() -> None:
+    """Various address formats should yield the bare street name."""
+    from app.services.estimator import extract_street_name
+
+    cases = [
+        (
+            "Свердловская обл., г. Екатеринбург, ул. Космонавтов, 12",
+            "Космонавтов",
+        ),
+        (
+            "г. Екатеринбург, проспект Ленина, 50",
+            "Ленина",
+        ),
+        (
+            "Екатеринбург, ул. 8 Марта, 18",
+            "8 Марта",
+        ),
+        (
+            "Россия, Екатеринбург, пер. Красный, 4",
+            "Красный",
+        ),
+        (
+            "Екатеринбург, ул. Малышева, 1",
+            "Малышева",
+        ),
+    ]
+    for addr, expected in cases:
+        result = extract_street_name(addr)
+        assert result == expected, f"addr={addr!r}: expected {expected!r}, got {result!r}"
+
+
+def test_extract_street_name_none_and_empty() -> None:
+    from app.services.estimator import extract_street_name
+
+    assert extract_street_name(None) is None
+    assert extract_street_name("") is None
+
+
+def test_extract_street_name_no_street_keyword() -> None:
+    """Address without a recognisable street keyword — must not crash."""
+    from app.services.estimator import extract_street_name
+
+    result = extract_street_name("Industrial zone X")
+    assert result is None or isinstance(result, str)
+
+
+# ── Helpers ───────────────────────────────────────────────────────────────────
+
+
+@pytest.fixture()
+def trade_in_app() -> FastAPI:
+    """Minimal FastAPI app mounting only the trade-in router with DB overridden."""
+    from app.api.v1 import trade_in as trade_in_module
+    from app.core.db import get_db
+
+    application = FastAPI()
+    application.include_router(trade_in_module.router, prefix="/api/v1/trade-in")
+
+    def _override_db():
+        yield MagicMock()
+
+    application.dependency_overrides[get_db] = _override_db
+    return application
+
+
+def _make_db_mock(rows) -> MagicMock:
+    """DB session mock returning *rows* from .mappings().all()."""
+    db = MagicMock()
+    mapping_result = MagicMock()
+    mapping_result.all.return_value = rows
+    execute_result = MagicMock()
+    execute_result.mappings.return_value = mapping_result
+    db.execute.return_value = execute_result
+    return db
+
+
+def _make_deal_row(
+    *,
+    address: str = "Екатеринбург, Космонавтов",
+    area_m2: float = 50.0,
+    rooms: int = 2,
+    floor: int | None = 5,
+    total_floors: int | None = 9,
+    price_rub: int = 5_000_000,
+    price_per_m2: int = 100_000,
+    deal_date: date | None = None,
+    source: str = "rosreestr",
+) -> dict:
+    return {
+        "address": address,
+        "area_m2": area_m2,
+        "rooms": rooms,
+        "floor": floor,
+        "total_floors": total_floors,
+        "price_rub": price_rub,
+        "price_per_m2": price_per_m2,
+        "deal_date": deal_date or date(2025, 6, 1),
+        "source": source,
+    }
+
+
+# ── Test: empty result when DB returns no rows ────────────────────────────────
+
+
+def test_street_deals_returns_empty_when_no_match(trade_in_app: FastAPI) -> None:
+    """Endpoint returns count=0 / empty deals list when DB has no matching rows."""
+    db_mock = _make_db_mock([])
+
+    from app.core.db import get_db
+
+    def _override():
+        yield db_mock
+
+    trade_in_app.dependency_overrides[get_db] = _override
+
+    client = TestClient(trade_in_app)
+    resp = client.get(
+        "/api/v1/trade-in/street-deals",
+        params={
+            "address": "г. Екатеринбург, ул. Космонавтов, 50",
+            "area_m2": 50.0,
+            "rooms": 2,
+        },
+    )
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["count"] == 0
+    assert data["deals"] == []
+    assert data["street"] == "Космонавтов"
+    assert data["median_price_rub"] == 0
+
+
+# ── Test: aggregation with fixture deals ─────────────────────────────────────
+
+
+def test_street_deals_aggregates_correctly(trade_in_app: FastAPI) -> None:
+    """With 5 fake deals the endpoint computes correct count + median."""
+    # price_per_m2 values sorted: [80k, 90k, 100k, 110k, 120k] → median = 100k
+    # median_price_rub = 100k * 50m² = 5_000_000
+    fixture_rows = [
+        _make_deal_row(price_rub=6_000_000, price_per_m2=120_000, deal_date=date(2025, 10, 1)),
+        _make_deal_row(price_rub=5_500_000, price_per_m2=110_000, deal_date=date(2025, 9, 1)),
+        _make_deal_row(price_rub=5_000_000, price_per_m2=100_000, deal_date=date(2025, 8, 1)),
+        _make_deal_row(price_rub=4_500_000, price_per_m2=90_000, deal_date=date(2025, 7, 1)),
+        _make_deal_row(price_rub=4_000_000, price_per_m2=80_000, deal_date=date(2025, 6, 1)),
+    ]
+
+    db_mock = _make_db_mock(fixture_rows)
+
+    from app.core.db import get_db
+
+    def _override():
+        yield db_mock
+
+    trade_in_app.dependency_overrides[get_db] = _override
+
+    client = TestClient(trade_in_app)
+    resp = client.get(
+        "/api/v1/trade-in/street-deals",
+        params={
+            "address": "г. Екатеринбург, ул. Космонавтов, 50",
+            "area_m2": 50.0,
+            "rooms": 2,
+            "period_months": 12,
+        },
+    )
+    assert resp.status_code == 200
+    data = resp.json()
+
+    assert data["street"] == "Космонавтов"
+    assert data["count"] == 5
+    assert data["median_price_per_m2"] == 100_000
+    assert data["median_price_rub"] == 5_000_000    # 100k * 50m²
+    assert data["range_low_rub"] == 4_000_000
+    assert data["range_high_rub"] == 6_000_000
+    # All 5 deals returned (top-10 threshold)
+    assert len(data["deals"]) == 5
+    # DB rows already ordered by deal_date DESC — first is most recent (2025-10-01)
+    assert data["deals"][0]["price_rub"] == 6_000_000