feat(tradein): GET /street-deals — ДКП-сделки Росреестра по улице target адреса #555
4 changed files with 398 additions and 1 deletions
|
|
@ -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,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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 слотов.
|
||||
|
||||
|
|
|
|||
209
tradein-mvp/backend/tests/test_street_deals_endpoint.py
Normal file
209
tradein-mvp/backend/tests/test_street_deals_endpoint.py
Normal file
|
|
@ -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
|
||||
Loading…
Add table
Reference in a new issue