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

@@ -40,6 +40,13 @@ class AnalogLot(BaseModel):
     listing_date: date | None
     days_on_market: int | None
     photo_url: str | None = None
+    # ── Per-comp coords для MAP / price↔exposure views (web features) ──
+    # ADDITIVE + OPTIONAL. lat/lon из listings.lat-lon / deals.lat-lon (ST_Y/ST_X(geom)).
+    # Nullable: radius-фильтрованные аналоги имеют 100% coords, но Tier S
+    # (same-building через house_id_fk / address-prefix) может включать
+    # address-only Avito-лоты без geom → None (graceful, frontend пропускает на карте).
+    lat: float | None = None
+    lon: float | None = None
     # ── Новые поля (Слой 5.2 — clickable links) ──
     source: str | None = None  # 'avito' / 'cian' / 'domklik' / 'rosreestr'
     source_url: str | None = None  # ссылка на оригинальное объявление / сделку
@@ -105,6 +112,17 @@ class DkpCorridor(BaseModel):
     period_months: int  # окно поиска сделок
 
 
+class PriceTrendPoint(BaseModel):
+    """Одна точка месячного ₽/м² тренда для целевого дома / района (web TREND chart).
+
+    Источник: houses_price_dynamics (если заполнена) ИЛИ агрегация
+    house_placement_history по месяцам. month — 'YYYY-MM', ppm2 — медиана ₽/м².
+    """
+
+    month: str  # 'YYYY-MM'
+    ppm2: int  # медиана ₽/м² за месяц
+
+
 class AggregatedEstimate(BaseModel):
     estimate_id: UUID
     median_price_rub: int
@@ -124,8 +142,15 @@ class AggregatedEstimate(BaseModel):
     target_lon: float | None = None
     sources_used: list[str] = Field(default_factory=list)  # ['avito', 'cian', 'rosreestr']
     data_freshness_minutes: int | None = None  # сколько минут назад был самый свежий парсинг
+    # абсолютный timestamp самого свежего парсинга аналогов
+    last_scraped_at: datetime | None = None
     est_days_on_market: int | None = None  # прогноз срока продажи (медиана по аналогам)
     cian_valuation: CianValuationSummary | None = None
+    # ── Месячный ₽/м² тренд для целевого дома (web TREND chart) — ADDITIVE + OPTIONAL ──
+    # ~12-24 точки. Источник: houses_price_dynamics (preferred, пока пуста в prod) →
+    # fallback агрегация house_placement_history по месяцам для target_house_id.
+    # None если house_id не разрешён / нет истории (graceful, frontend скрывает chart).
+    price_trend: list[PriceTrendPoint] | None = None
     # ── #651/#652: внешние якоря (Avito IMV) + коридор реальных сделок (ДКП) ──
     # avito_imv  — реальная IMV-оценка дома (house_imv_evaluations). Используется
     #              как anchor для blend'а медианы (см. confidence_explanation).

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

@@ -40,6 +40,7 @@ from app.schemas.trade_in import (
     AvitoImvSummary,
     CianValuationSummary,
     DkpCorridor,
+    PriceTrendPoint,
     TradeInEstimateInput,
 )
 from app.services.dadata import DadataAddressResult
@@ -2053,6 +2054,14 @@ async def estimate_quality(
     if cian_val is not None and cian_val.sale_price_rub:
         sources_used = sorted(set(sources_used) | {"cian_valuation"})
     freshness_min = _compute_freshness_minutes(listings_clean)
+    last_scraped_at = _compute_last_scraped_at(listings_clean)
+    # Месячный ₽/м² тренд целевого дома (web TREND chart) — best-effort, None если нет данных.
+    price_trend_raw = _fetch_price_trend(db, target_house_id=target_house_id)
+    price_trend = (
+        [PriceTrendPoint(month=p["month"], ppm2=p["ppm2"]) for p in price_trend_raw]
+        if price_trend_raw
+        else None
+    )
 
     return AggregatedEstimate(
         estimate_id=estimate_id,
@@ -2072,6 +2081,8 @@ async def estimate_quality(
         target_lon=geo.lon,
         sources_used=sources_used,
         data_freshness_minutes=freshness_min,
+        last_scraped_at=last_scraped_at,
+        price_trend=price_trend,
         est_days_on_market=_estimate_days_on_market(listings_clean, deals),
         cian_valuation=(
             CianValuationSummary(
@@ -2169,6 +2180,145 @@ def _compute_freshness_minutes(lots: list[dict[str, Any]]) -> int | None:
     return int((now - max(scraped_dt)).total_seconds() / 60)
 
 
+def _compute_last_scraped_at(lots: list[dict[str, Any]]) -> datetime | None:
+    """Абсолютный timestamp самого свежего парсинга среди аналогов (для UI).
+
+    Дополняет _compute_freshness_minutes (относительные минуты): отдаёт точную
+    дату/время, чтобы фронт мог отрендерить «обновлено DD.MM HH:MM». None если
+    ни у одного лота нет scraped_at/listing_date с tzinfo (graceful)."""
+    if not lots:
+        return None
+    scraped = [lot.get("scraped_at") or lot.get("listing_date") for lot in lots]
+    scraped_dt: list[datetime] = []
+    for s in scraped:
+        if s is None:
+            continue
+        if hasattr(s, "tzinfo"):
+            scraped_dt.append(s if s.tzinfo else s.replace(tzinfo=UTC))
+    return max(scraped_dt) if scraped_dt else None
+
+
+def _fetch_price_trend(
+    db: Session,
+    *,
+    target_house_id: int | None,
+    months: int = 24,
+    min_points: int = 3,
+) -> list[dict[str, Any]] | None:
+    """Месячный ₽/м² тренд для целевого дома (web TREND chart) — best-effort.
+
+    Предпочитает `houses_price_dynamics` (house_id, month_date, price_per_sqm) —
+    готовая помесячная серия. В prod эта таблица пока ПУСТА, поэтому fallback —
+    агрегация `house_placement_history` по месяцам (медиана COALESCE(last_price,
+    start_price)/area_m2, дата = COALESCE(last_price_date, start_price_date)).
+
+    Возвращает список [{month: 'YYYY-MM', ppm2: int}, ...] (≤ `months` точек,
+    ASC по месяцу) или None если house_id не задан / точек < `min_points` /
+    любая ошибка (graceful — фронт скрывает chart, без регрессий).
+    """
+    if target_house_id is None:
+        return None
+
+    # ── Source 1 (preferred): houses_price_dynamics ──────────────────────────
+    try:
+        rows = (
+            db.execute(
+                text(
+                    """
+                    SELECT to_char(month_date, 'YYYY-MM') AS month,
+                           round(
+                               percentile_cont(0.5) WITHIN GROUP (ORDER BY price_per_sqm)
+                           )::int AS ppm2
+                    FROM houses_price_dynamics
+                    WHERE house_id = CAST(:hid AS bigint)
+                      AND price_per_sqm > 0
+                      AND month_date > (CURRENT_DATE
+                          - (CAST(:months AS integer) || ' months')::interval)
+                    GROUP BY month_date
+                    ORDER BY month_date ASC
+                    """
+                ),
+                {"hid": target_house_id, "months": months},
+            )
+            .mappings()
+            .all()
+        )
+    except Exception as exc:  # pragma: no cover — defensive
+        logger.warning("price_trend houses_price_dynamics lookup failed (graceful): %s", exc)
+        try:
+            db.rollback()
+        except Exception:
+            pass
+        rows = []
+
+    trend = [{"month": r["month"], "ppm2": int(r["ppm2"])} for r in rows if r["ppm2"]]
+    if len(trend) >= min_points:
+        logger.info(
+            "price_trend house_id=%s source=houses_price_dynamics → %d points",
+            target_house_id,
+            len(trend),
+        )
+        return trend
+
+    # ── Source 2 (fallback): aggregate house_placement_history by month ──────
+    try:
+        rows = (
+            db.execute(
+                text(
+                    """
+                    SELECT to_char(
+                               date_trunc('month',
+                                          COALESCE(last_price_date, start_price_date)),
+                               'YYYY-MM'
+                           ) AS month,
+                           round(
+                               percentile_cont(0.5) WITHIN GROUP (
+                                   ORDER BY COALESCE(last_price, start_price)
+                                            / NULLIF(area_m2, 0)
+                               )
+                           )::int AS ppm2
+                    FROM house_placement_history
+                    WHERE house_id = CAST(:hid AS bigint)
+                      AND area_m2 > 0
+                      AND COALESCE(last_price, start_price) > 0
+                      AND COALESCE(last_price_date, start_price_date) IS NOT NULL
+                      AND COALESCE(last_price_date, start_price_date) > (CURRENT_DATE
+                          - (CAST(:months AS integer) || ' months')::interval)
+                    GROUP BY 1
+                    ORDER BY 1 ASC
+                    """
+                ),
+                {"hid": target_house_id, "months": months},
+            )
+            .mappings()
+            .all()
+        )
+    except Exception as exc:  # pragma: no cover — defensive
+        logger.warning("price_trend house_placement_history lookup failed (graceful): %s", exc)
+        try:
+            db.rollback()
+        except Exception:
+            pass
+        rows = []
+
+    trend = [{"month": r["month"], "ppm2": int(r["ppm2"])} for r in rows if r["ppm2"]]
+    if len(trend) >= min_points:
+        logger.info(
+            "price_trend house_id=%s source=house_placement_history → %d points",
+            target_house_id,
+            len(trend),
+        )
+        return trend
+
+    logger.info(
+        "price_trend house_id=%s → only %d points (<%d) → None",
+        target_house_id,
+        len(trend),
+        min_points,
+    )
+    return None
+
+
 # ── Internals ────────────────────────────────────────────────────────────────
 
 # Compiled regexes for _extract_short_addr — module-level for performance.
@@ -2912,6 +3062,8 @@ def _listing_to_analog(row: dict[str, Any]) -> AnalogLot:
         source=row.get("source"),
         source_url=row.get("source_url"),
         distance_m=int(row["distance_m"]) if row.get("distance_m") is not None else None,
+        lat=float(row["lat"]) if row.get("lat") is not None else None,
+        lon=float(row["lon"]) if row.get("lon") is not None else None,
     )
 
 
@@ -2940,6 +3092,8 @@ def _deal_to_analog(row: dict[str, Any]) -> AnalogLot:
         source=row.get("source"),
         source_url=None,  # rosreestr сделки без публичной ссылки
         distance_m=int(row["distance_m"]) if row.get("distance_m") is not None else None,
+        lat=float(row["lat"]) if row.get("lat") is not None else None,
+        lon=float(row["lon"]) if row.get("lon") is not None else None,
         tier=tier,
     )
 

tradein-mvp/backend/tests/test_estimator_enrich.py

@@ -0,0 +1,210 @@
+"""Unit tests for the web-features enrichment of the estimate response.
+
+Covers the ADDITIVE + OPTIONAL fields surfaced for the new web views (map /
+comp distribution / price↔exposure / ppm² trend):
+
+  - estimator._listing_to_analog / _deal_to_analog — now carry lat/lon
+  - estimator._fetch_price_trend — monthly ppm² series shape (mock db)
+  - estimator._compute_last_scraped_at — absolute freshness timestamp
+
+No real DB / network: the trend test injects a fake Session whose execute()
+returns canned mapping rows.
+
+NOTE: importing app.services.estimator pulls app.core.config.Settings, which
+requires DATABASE_URL — set BEFORE importing app modules (same pattern as the
+sibling estimator unit tests).
+"""
+
+import os
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from datetime import UTC, datetime
+
+import pytest
+
+from app.schemas.trade_in import AnalogLot, PriceTrendPoint
+from app.services import estimator
+
+# --------------------------------------------------------------------------- #
+# Per-comp coords on analogs / deals (map + exposure views)
+# --------------------------------------------------------------------------- #
+
+
+def test_listing_to_analog_carries_lat_lon() -> None:
+    row = {
+        "address": "ул. Тестовая, 1",
+        "area_m2": 50.0,
+        "rooms": 2,
+        "floor": 3,
+        "total_floors": 9,
+        "price_rub": 10_000_000,
+        "price_per_m2": 200_000,
+        "listing_date": None,
+        "days_on_market": 14,
+        "photo_urls": None,
+        "source": "cian",
+        "source_url": "https://example.test/1",
+        "distance_m": 120.0,
+        "lat": 56.8389,
+        "lon": 60.6057,
+    }
+    lot = estimator._listing_to_analog(row)
+    assert isinstance(lot, AnalogLot)
+    assert lot.lat == pytest.approx(56.8389)
+    assert lot.lon == pytest.approx(60.6057)
+    # exposure + comp-distribution keys still present
+    assert lot.days_on_market == 14
+    assert lot.price_per_m2 == 200_000
+    assert lot.area_m2 == 50.0
+    assert lot.rooms == 2
+
+
+def test_deal_to_analog_carries_lat_lon() -> None:
+    row = {
+        "address": "ул. Тестовая, 2",
+        "area_m2": 60.0,
+        "rooms": 2,
+        "floor": 5,
+        "total_floors": 10,
+        "price_rub": 11_000_000,
+        "price_per_m2": 183_333,
+        "deal_date": None,
+        "days_on_market": None,
+        "kadastr_num": "66:41:0204016:10",
+        "source": "rosreestr",
+        "distance_m": 80.0,
+        "lat": 56.84,
+        "lon": 60.61,
+    }
+    lot = estimator._deal_to_analog(row)
+    assert lot.lat == pytest.approx(56.84)
+    assert lot.lon == pytest.approx(60.61)
+    assert lot.tier == "T0_per_house"  # kadastr with участок → per-house tier
+
+
+def test_analog_lat_lon_optional_none_when_missing() -> None:
+    # Tier S address-only Avito lots can lack geom → lat/lon absent → None (graceful).
+    row = {
+        "address": "ул. Без Координат, 3",
+        "area_m2": 40.0,
+        "rooms": 1,
+        "floor": None,
+        "total_floors": None,
+        "price_rub": 8_000_000,
+        "price_per_m2": 200_000,
+        "listing_date": None,
+        "days_on_market": None,
+        "photo_urls": None,
+        "source": "avito",
+        "source_url": None,
+        "distance_m": None,
+    }
+    lot = estimator._listing_to_analog(row)
+    assert lot.lat is None
+    assert lot.lon is None
+
+
+# --------------------------------------------------------------------------- #
+# _compute_last_scraped_at
+# --------------------------------------------------------------------------- #
+
+
+def test_last_scraped_at_returns_max_timestamp() -> None:
+    older = datetime(2026, 5, 1, 10, 0, tzinfo=UTC)
+    newer = datetime(2026, 5, 28, 9, 30, tzinfo=UTC)
+    lots = [{"scraped_at": older}, {"scraped_at": newer}]
+    assert estimator._compute_last_scraped_at(lots) == newer
+
+
+def test_last_scraped_at_empty_returns_none() -> None:
+    assert estimator._compute_last_scraped_at([]) is None
+
+
+# --------------------------------------------------------------------------- #
+# _fetch_price_trend (mock db)
+# --------------------------------------------------------------------------- #
+
+
+class _FakeResult:
+    def __init__(self, rows: list[dict]) -> None:
+        self._rows = rows
+
+    def mappings(self) -> "_FakeResult":
+        return self
+
+    def all(self) -> list[dict]:
+        return self._rows
+
+
+class _FakeSession:
+    """Minimal Session stub: returns canned rows per execute() call in order."""
+
+    def __init__(self, *results: list[dict]) -> None:
+        self._results = list(results)
+        self.calls = 0
+
+    def execute(self, *_args, **_kwargs) -> _FakeResult:
+        rows = self._results[self.calls] if self.calls < len(self._results) else []
+        self.calls += 1
+        return _FakeResult(rows)
+
+    def rollback(self) -> None:  # pragma: no cover — not exercised on happy path
+        pass
+
+
+def test_fetch_price_trend_none_when_no_house_id() -> None:
+    db = _FakeSession()
+    assert estimator._fetch_price_trend(db, target_house_id=None) is None
+    assert db.calls == 0  # short-circuits before any query
+
+
+def test_fetch_price_trend_prefers_houses_price_dynamics() -> None:
+    hpd_rows = [
+        {"month": "2025-01", "ppm2": 200_000},
+        {"month": "2025-02", "ppm2": 205_000},
+        {"month": "2025-03", "ppm2": 210_000},
+    ]
+    db = _FakeSession(hpd_rows)  # first query hits → no fallback needed
+    trend = estimator._fetch_price_trend(db, target_house_id=123)
+    assert trend is not None
+    assert db.calls == 1  # did NOT touch the fallback source
+    assert trend == [
+        {"month": "2025-01", "ppm2": 200_000},
+        {"month": "2025-02", "ppm2": 205_000},
+        {"month": "2025-03", "ppm2": 210_000},
+    ]
+    # Shape contract: each point validates as PriceTrendPoint(month:str, ppm2:int).
+    points = [PriceTrendPoint(**p) for p in trend]
+    assert all(isinstance(p.month, str) and isinstance(p.ppm2, int) for p in points)
+
+
+def test_fetch_price_trend_falls_back_to_placement_history() -> None:
+    fallback_rows = [
+        {"month": "2024-06", "ppm2": 190_000},
+        {"month": "2024-07", "ppm2": 195_000},
+        {"month": "2024-08", "ppm2": 198_000},
+    ]
+    # First query (houses_price_dynamics) empty → second (placement_history) hits.
+    db = _FakeSession([], fallback_rows)
+    trend = estimator._fetch_price_trend(db, target_house_id=235340)
+    assert trend == fallback_rows
+    assert db.calls == 2
+
+
+def test_fetch_price_trend_none_when_below_min_points() -> None:
+    # Both sources return < min_points (3) → None (graceful, frontend hides chart).
+    db = _FakeSession([{"month": "2025-01", "ppm2": 200_000}], [])
+    assert estimator._fetch_price_trend(db, target_house_id=999) is None
+
+
+def test_price_trend_point_shape() -> None:
+    p = PriceTrendPoint(month="2026-05", ppm2=251_429)
+    assert p.month == "2026-05"
+    assert p.ppm2 == 251_429
+    dumped = p.model_dump()
+    assert set(dumped) == {"month", "ppm2"}
+
+
+if __name__ == "__main__":  # pragma: no cover
+    raise SystemExit(pytest.main([__file__, "-q"]))