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

@@ -40,10 +40,17 @@ 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      # ссылка на оригинальное объявление / сделку
-    distance_m: int | None = None      # расстояние до целевой квартиры в метрах
+    source: str | None = None  # 'avito' / 'cian' / 'domklik' / 'rosreestr'
+    source_url: str | None = None  # ссылка на оригинальное объявление / сделку
+    distance_m: int | None = None  # расстояние до целевой квартиры в метрах
     # ── Confidence tier (PR M / #564 Phase 3) ──
     # Только для rosreestr-сделок: T0_per_house (kadastr_num exact match),
     # T1_per_street (street-level only). Open dataset Росреестра не имеет
@@ -82,10 +89,10 @@ class AvitoImvSummary(BaseModel):
     ценовой шкале. None если для дома нет свежей IMV-записи.
     """
 
-    recommended_price: int | None = None   # рекомендованная цена Avito, ₽
-    lower_price: int | None = None          # нижняя граница IMV-коридора, ₽
-    higher_price: int | None = None         # верхняя граница IMV-коридора, ₽
-    market_count: int | None = None         # объём рынка, на котором построена оценка
+    recommended_price: int | None = None  # рекомендованная цена Avito, ₽
+    lower_price: int | None = None  # нижняя граница IMV-коридора, ₽
+    higher_price: int | None = None  # верхняя граница IMV-коридора, ₽
+    market_count: int | None = None  # объём рынка, на котором построена оценка
 
 
 class DkpCorridor(BaseModel):
@@ -98,11 +105,22 @@ class DkpCorridor(BaseModel):
     None / count=0 если по улице нет сопоставимых сделок.
     """
 
-    count: int                              # число ДКП-сделок в выборке
-    low_ppm2: int                           # min ₽/м² по сделкам (P10-ish — берём минимум)
-    median_ppm2: int                        # медиана ₽/м²
-    high_ppm2: int                          # max ₽/м²
-    period_months: int                      # окно поиска сделок
+    count: int  # число ДКП-сделок в выборке
+    low_ppm2: int  # min ₽/м² по сделкам (P10-ish — берём минимум)
+    median_ppm2: int  # медиана ₽/м²
+    high_ppm2: int  # max ₽/м²
+    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):
@@ -112,20 +130,27 @@ class AggregatedEstimate(BaseModel):
     range_high_rub: int
     median_price_per_m2: int
     confidence: Literal["low", "medium", "high"]
-    confidence_explanation: str | None = None        # «Найдено 15 аналогов, разброс ±7%»
+    confidence_explanation: str | None = None  # «Найдено 15 аналогов, разброс ±7%»
     n_analogs: int
     period_months: int  # 24
     analogs: list[AnalogLot]  # top 5-10 listings
     actual_deals: list[AnalogLot]  # реальные продажи last 12 mo
     expires_at: datetime
     # ── Дополнительные метаданные ──
-    target_address: str | None = None                # geocoded full address
+    target_address: str | None = None  # geocoded full address
     target_lat: float | None = None
     target_lon: float | None = None
     sources_used: list[str] = Field(default_factory=list)  # ['avito', 'cian', 'rosreestr']
-    data_freshness_minutes: int | None = None        # сколько минут назад был самый свежий парсинг
-    est_days_on_market: int | None = None            # прогноз срока продажи (медиана по аналогам)
+    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).
@@ -143,8 +168,8 @@ class AggregatedEstimate(BaseModel):
     expected_sold_range_low_rub: int | None = None
     expected_sold_range_high_rub: int | None = None
     expected_sold_per_m2: int | None = None
-    asking_to_sold_ratio: float | None = None       # =sold/asking, ~0.72–0.93
-    ratio_basis: str | None = None                   # 'per_rooms' | 'global_fallback'
+    asking_to_sold_ratio: float | None = None  # =sold/asking, ~0.72–0.93
+    ratio_basis: str | None = None  # 'per_rooms' | 'global_fallback'
     # ── DaData enrichment (PR Q1) — on-demand для target адреса ──
     # canonical_address — DaData-нормализованная форма (с улицей в short form).
     # house_cadnum     — кадастровый номер ДОМА (для будущего matching Росреестра).
@@ -189,7 +214,7 @@ class HouseInfoForEstimate(BaseModel):
     """Summary информации о доме целевой квартиры (для GET /estimate/{id}/houses)."""
 
     house_id: int | None = None
-    source: str | None = None           # 'avito' / 'derived' / 'cian_newbuilding' / etc.
+    source: str | None = None  # 'avito' / 'derived' / 'cian_newbuilding' / etc.
     ext_house_id: str | None = None
     address: str | None = None
     short_address: str | None = None
@@ -213,7 +238,7 @@ class HouseInfoForEstimate(BaseModel):
 class IMVBenchmarkResponse(BaseModel):
     """Avito IMV benchmark для UI (GET /estimate/{id}/imv-benchmark)."""
 
-    available: bool                          # есть ли IMV для этого estimate
+    available: bool  # есть ли IMV для этого estimate
     cache_key: str | None = None
     recommended_price: int | None = None
     lower_price: int | None = None
@@ -222,7 +247,7 @@ class IMVBenchmarkResponse(BaseModel):
     fetched_at: datetime | None = None
     # comparison vs our estimate
     our_median_price: int | None = None
-    diff_pct: float | None = None            # (our - imv) / imv * 100
+    diff_pct: float | None = None  # (our - imv) / imv * 100
 
 
 class PlacementHistoryEntry(BaseModel):
@@ -261,7 +286,7 @@ class ScheduleConfig(BaseModel):
     last_run_id: int | None = None
     last_run_at: str | None = None  # ISO
     next_run_at: str | None = None  # ISO
-    updated_at: str | None = None   # ISO
+    updated_at: str | None = None  # ISO
 
 
 class ScheduleConfigUpdate(BaseModel):
@@ -294,12 +319,12 @@ class CianPriceChangeStats(BaseModel):
 
     cian_id: str
     listing_id: int
-    n_changes: int                   # COUNT(*) из offer_price_history
+    n_changes: int  # COUNT(*) из offer_price_history
     last_change_time: datetime | None
     last_diff_percent: float | None  # последняя дельта (-5% если цена снизилась)
-    total_change_pct: float | None   # суммарно (current - first) / first * 100
+    total_change_pct: float | None  # суммарно (current - first) / first * 100
     first_seen_price: int | None
-    current_price: int               # из listings.price_rub
+    current_price: int  # из listings.price_rub
 
 
 class RecentSoldEntry(BaseModel):
@@ -343,8 +368,8 @@ class HouseAnalyticsResponse(BaseModel):
 class SellTimeBucket(BaseModel):
     """Один бакет срока продажи для данного ценового диапазона."""
 
-    price_premium_label: str    # 'cheap' | 'median' | 'plus5' | 'plus10'
-    price_premium_pct: float    # -5.0, 0.0, 5.0, 10.0 для UI
+    price_premium_label: str  # 'cheap' | 'median' | 'plus5' | 'plus10'
+    price_premium_pct: float  # -5.0, 0.0, 5.0, 10.0 для UI
     median_exposure_days: int | None
     p25_days: int | None
     p75_days: int | None
@@ -374,12 +399,12 @@ class StreetDealsResponse(BaseModel):
     street: str | None
     period_from: date
     period_to: date
-    count: int                      # число всех matching сделок, не только топ-10
-    median_price_rub: int           # 0 если count == 0
+    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
+    deals: list[AnalogLot]  # последние 10 по deal_date DESC
 
 
 # ── Sales vs Listings (PR K, issue #564 Foundation Phase 1) ─────────────────
@@ -402,7 +427,7 @@ class SalesListingPair(BaseModel):
     deal_address: str | None = None
 
     listing_id: int | None = None
-    listing_source: str | None = None       # 'avito' / 'cian' / 'yandex' / 'n1' / 'domklik'
+    listing_source: str | None = None  # 'avito' / 'cian' / 'yandex' / 'n1' / 'domklik'
     listing_source_url: str | None = None
     listing_date: date | None = None
     listing_price_rub: int | None = None
@@ -424,15 +449,15 @@ class SalesVsListingsResponse(BaseModel):
     linkage rate и медианный discount.
     """
 
-    street: str | None                      # извлечённое имя улицы, None если не извлеклось
-    period_months: int                      # окно поиска сделок
-    window_days: int                        # окно matching listing → deal
-    area_tolerance: float                   # 0.15 = ±15% по area_m2
-    total_deals: int                        # количество всех matching ДКП в улице/период
-    deals_with_listings: int                # сколько имеют связанный listing
-    linkage_rate_pct: float                 # deals_with_listings / total_deals * 100
-    median_discount_pct: float | None       # медиана по парам с listing
-    pairs: list[SalesListingPair]           # все пары, sorted by deal_date DESC
+    street: str | None  # извлечённое имя улицы, None если не извлеклось
+    period_months: int  # окно поиска сделок
+    window_days: int  # окно matching listing → deal
+    area_tolerance: float  # 0.15 = ±15% по area_m2
+    total_deals: int  # количество всех matching ДКП в улице/период
+    deals_with_listings: int  # сколько имеют связанный listing
+    linkage_rate_pct: float  # deals_with_listings / total_deals * 100
+    median_discount_pct: float | None  # медиана по парам с listing
+    pairs: list[SalesListingPair]  # все пары, sorted by deal_date DESC
 
 
 # ── Account quota (#quota) ──────────────────────────────────────────────────
@@ -441,7 +466,7 @@ class SalesVsListingsResponse(BaseModel):
 class QuotaStatus(BaseModel):
     """Статус квоты оценок для текущего аккаунта (GET /api/v1/trade-in/quota)."""
 
-    limit: int       # MONTHLY_LIMIT = 15
-    used: int        # использовано в текущем месяце
-    remaining: int   # max(0, limit - used)
+    limit: int  # MONTHLY_LIMIT = 15
+    used: int  # использовано в текущем месяце
+    remaining: int  # max(0, limit - used)
     unlimited: bool  # True для admin / kopylov / без заголовка

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"]))