feat(tradein): estimator — Yandex Valuation as on-demand 6th source

Stage 8 of YandexRealtyScraper v1 (Wave 6).

Adds Yandex Valuation tool (anonymous, no cookies) as on-demand enrichment
source in estimate_quality. Mirrors the Avito IMV pattern with 24h cache,
graceful degradation, and idempotent history persistence.

estimator.py:
- _yandex_valuation_cache_key(address, category, type) -> sha256
- _get_or_fetch_yandex_valuation_cached(db, address, ...) -> cache via
  external_valuations(source='yandex_valuation'). Any error -> None +
  log warning + estimator continues.
- _save_yandex_history_items(db, result) -> INSERT into
  house_placement_history(source='yandex_valuation') with synthesized
  ext_item_id (sha256 of address+date+area+floor+prices, first 32 chars).
  Idempotent via UNIQUE (source, ext_item_id). house_id stays NULL -
  estimator doesn't compute target_house_id yet (matching pipeline TBD).
- estimate_quality: call after IMV block, BEFORE deals fetch. Adds
  'yandex_valuation' to sources_used if Yandex returned a result.

Tests: 11 unit tests (cache key determinism, hit/miss, fetch error,
empty result, history item save / dedup / partial error). All pass.
Ruff clean (no new errors introduced).

Constants: YANDEX_VALUATION_CACHE_TTL_HOURS=24, default category=APARTMENT,
default type=SELL.

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

@@ -19,6 +19,7 @@
 
 from __future__ import annotations
 
+import hashlib
 import json
 import logging
 from datetime import UTC, datetime, timedelta
@@ -38,6 +39,10 @@ from app.services.scrapers.avito_imv import (
     evaluate_via_imv,
     save_imv_evaluation,
 )
+from app.services.scrapers.yandex_valuation import (
+    YandexValuationResult,
+    YandexValuationScraper,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -94,6 +99,10 @@ def _repair_coefficient(repair_state: str | None) -> float:
 # ── Avito IMV cache lookup (Stage 3) ────────────────────────────────────────
 IMV_CACHE_TTL_HOURS = 24
 
+YANDEX_VALUATION_CACHE_TTL_HOURS = 24
+YANDEX_VALUATION_DEFAULT_CATEGORY = "APARTMENT"
+YANDEX_VALUATION_DEFAULT_TYPE = "SELL"
+
 
 async def _get_or_fetch_imv_cached(
     db: Session,
@@ -200,6 +209,187 @@ async def _get_or_fetch_imv_cached(
         return None
 
 
+# ── Yandex Valuation cache lookup (Stage 8) ─────────────────────────────────
+
+def _yandex_valuation_cache_key(
+    address: str, offer_category: str, offer_type: str
+) -> str:
+    """SHA256 cache key for Yandex Valuation lookups."""
+    payload = f"{address}|{offer_category}|{offer_type}"
+    return hashlib.sha256(payload.encode("utf-8")).hexdigest()
+
+
+async def _get_or_fetch_yandex_valuation_cached(
+    db: Session,
+    *,
+    address: str,
+    offer_category: str = YANDEX_VALUATION_DEFAULT_CATEGORY,
+    offer_type: str = YANDEX_VALUATION_DEFAULT_TYPE,
+) -> YandexValuationResult | None:
+    """Cached Yandex Valuation lookup. TTL 24h via external_valuations table.
+
+    Returns None on any error / cache miss + fetch failure — caller continues
+    without Yandex enrichment (graceful degradation).
+    """
+    cache_key = _yandex_valuation_cache_key(address, offer_category, offer_type)
+
+    # Cache lookup
+    try:
+        cached = db.execute(
+            text(
+                """
+                SELECT raw_payload, fetched_at
+                FROM external_valuations
+                WHERE source = 'yandex_valuation'
+                  AND cache_key = :ck
+                  AND expires_at > NOW()
+                ORDER BY fetched_at DESC
+                LIMIT 1
+                """
+            ),
+            {"ck": cache_key},
+        ).mappings().first()
+    except Exception as e:
+        logger.warning("yandex_valuation: cache lookup failed: %s", e)
+        cached = None
+
+    if cached is not None and cached.get("raw_payload"):
+        try:
+            payload_dict = (
+                cached["raw_payload"]
+                if isinstance(cached["raw_payload"], dict)
+                else json.loads(cached["raw_payload"])
+            )
+            logger.info(
+                "yandex_valuation: cache HIT key=%s items=%d",
+                cache_key[:8],
+                len(payload_dict.get("history_items", [])),
+            )
+            return YandexValuationResult.model_validate(payload_dict)
+        except Exception as e:
+            logger.warning("yandex_valuation: cache deserialize failed — refetching: %s", e)
+
+    # Fresh fetch
+    try:
+        async with YandexValuationScraper() as scraper:
+            result = await scraper.fetch_house_history(
+                address=address,
+                offer_category=offer_category,
+                offer_type=offer_type,
+            )
+    except Exception as e:
+        logger.warning(
+            "yandex_valuation: fetch failed — estimator продолжает без Yandex: %s", e
+        )
+        return None
+
+    if result is None:
+        logger.info("yandex_valuation: empty result for address=%s", address[:60])
+        return None
+
+    # Save to cache (UPSERT on (source, cache_key))
+    try:
+        db.execute(
+            text(
+                """
+                INSERT INTO external_valuations (
+                    source, cache_key, address,
+                    raw_payload,
+                    fetched_at, expires_at
+                ) VALUES (
+                    'yandex_valuation', :ck, :addr,
+                    CAST(:payload AS jsonb),
+                    NOW(), NOW() + (:ttl_hours || ' hours')::interval
+                )
+                ON CONFLICT (source, cache_key) DO UPDATE
+                    SET raw_payload = EXCLUDED.raw_payload,
+                        fetched_at = NOW(),
+                        expires_at = NOW() + (:ttl_hours || ' hours')::interval
+                """
+            ),
+            {
+                "ck": cache_key,
+                "addr": address,
+                "payload": json.dumps(result.model_dump(mode="json"), ensure_ascii=False),
+                "ttl_hours": YANDEX_VALUATION_CACHE_TTL_HOURS,
+            },
+        )
+        db.commit()
+        logger.info(
+            "yandex_valuation: fresh fetch saved key=%s items=%d",
+            cache_key[:8],
+            len(result.history_items),
+        )
+    except Exception as e:
+        logger.warning("yandex_valuation: cache save failed (continuing): %s", e)
+        db.rollback()
+
+    return result
+
+
+def _save_yandex_history_items(
+    db: Session,
+    result: YandexValuationResult,
+) -> int:
+    """Persist history items to house_placement_history. Returns saved count.
+
+    house_id stays NULL — estimator doesn't compute target_house_id yet.
+    Idempotent via UNIQUE (source, ext_item_id); we synthesize ext_item_id from
+    (address|date|area|floor) hash since Yandex history items don't carry an
+    explicit ID.
+    """
+    saved = 0
+    for item in result.history_items:
+        # Synthesize stable ext_item_id (no native ID in valuation page)
+        ext_seed = (
+            f"{result.address}|{item.publish_date}|{item.area_m2}|{item.floor}|"
+            f"{item.start_price}|{item.last_price}"
+        )
+        ext_item_id = hashlib.sha256(ext_seed.encode("utf-8")).hexdigest()[:32]
+        try:
+            db.execute(
+                text(
+                    """
+                    INSERT INTO house_placement_history (
+                        source, ext_item_id,
+                        rooms, area_m2, floor,
+                        start_price, start_price_date,
+                        last_price, last_price_date,
+                        exposure_days,
+                        raw_payload
+                    ) VALUES (
+                        'yandex_valuation', :ext_id,
+                        :rooms, :area, :floor,
+                        :start_price, :publish_date,
+                        :last_price, :publish_date,
+                        :exposure,
+                        CAST(:raw AS jsonb)
+                    )
+                    ON CONFLICT (source, ext_item_id) DO NOTHING
+                    """
+                ),
+                {
+                    "ext_id": ext_item_id,
+                    "rooms": item.rooms,
+                    "area": item.area_m2,
+                    "floor": item.floor,
+                    "start_price": item.start_price,
+                    "last_price": item.last_price,
+                    "publish_date": item.publish_date,
+                    "exposure": item.exposure_days,
+                    "raw": json.dumps(item.model_dump(mode="json"), ensure_ascii=False),
+                },
+            )
+            saved += 1
+        except Exception as e:
+            logger.warning("yandex_valuation: failed to save history item: %s", e)
+            db.rollback()
+            continue
+    if saved:
+        db.commit()
+    return saved
+
+
 # ── Public ───────────────────────────────────────────────────────────────────
 async def estimate_quality(
     payload: TradeInEstimateInput, db: Session
@@ -341,6 +531,21 @@ async def estimate_quality(
     if imv_eval is not None:
         sources_used_pre = sorted(set(sources_used_pre) | {"avito_imv"})
 
+    # ── Stage 8: Yandex Valuation as on-demand source (anonymous, cached 24h) ──
+    yandex_val: YandexValuationResult | None = None
+    if geo is not None and geo.full_address:
+        yandex_val = await _get_or_fetch_yandex_valuation_cached(
+            db, address=geo.full_address,
+        )
+        if yandex_val is not None:
+            sources_used_pre = sorted(set(sources_used_pre) | {"yandex_valuation"})
+            saved_hist = _save_yandex_history_items(db, yandex_val)
+            logger.info(
+                "yandex_valuation: history items processed=%d saved=%d"
+                " (house_id=NULL — matching deferred)",
+                len(yandex_val.history_items), saved_hist,
+            )
+
     # 5. Deals — фактические сделки за период
     deals = _fetch_deals(
         db, lat=geo.lat, lon=geo.lon, rooms=payload.rooms, area=payload.area_m2,
@@ -454,6 +659,8 @@ async def estimate_quality(
     sources_used = sorted({lot.source for lot in analogs_lots if lot.source})
     if imv_eval is not None:
         sources_used = sorted(set(sources_used) | {"avito_imv"})
+    if yandex_val is not None:
+        sources_used = sorted(set(sources_used) | {"yandex_valuation"})
     freshness_min = _compute_freshness_minutes(listings_clean)
 
     return AggregatedEstimate(

tradein-mvp/backend/tests/test_estimator_yandex_integration.py

@@ -0,0 +1,197 @@
+"""Tests for Yandex Valuation integration in estimator.py."""
+import os
+
+# Settings requires DATABASE_URL at init time. Set dummy DSN before any app import.
+os.environ.setdefault("DATABASE_URL", "postgresql://test:test@localhost/test_db")
+
+from datetime import date
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from app.services.estimator import (
+    YANDEX_VALUATION_CACHE_TTL_HOURS,
+    _get_or_fetch_yandex_valuation_cached,
+    _save_yandex_history_items,
+    _yandex_valuation_cache_key,
+)
+from app.services.scrapers.yandex_valuation import (
+    ValuationHistoryItem,
+    ValuationHouseMeta,
+    YandexValuationResult,
+)
+
+
+def _sample_result(address: str = "Екатеринбург, ул. Учителей, 18") -> YandexValuationResult:
+    return YandexValuationResult(
+        address=address,
+        offer_category="APARTMENT",
+        offer_type="SELL",
+        page=1,
+        source_url=(
+            f"https://realty.yandex.ru/otsenka-kvartiry-po-adresu-onlayn/?address={address}"
+        ),
+        house=ValuationHouseMeta(
+            year_built=1981,
+            total_floors=9,
+            house_type="panel",
+            ceiling_height=2.5,
+            has_lift=True,
+            total_objects=12,
+        ),
+        history_items=[
+            ValuationHistoryItem(
+                area_m2=36.8,
+                rooms=1,
+                floor=8,
+                start_price=4_500_000,
+                start_price_per_m2=122_283,
+                last_price=4_400_000,
+                last_price_per_m2=119_566,
+                publish_date=date(2026, 2, 17),
+                exposure_days=96,
+                status="В продаже",
+            ),
+            ValuationHistoryItem(
+                area_m2=64.4,
+                rooms=2,
+                floor=1,
+                start_price=6_580_000,
+                start_price_per_m2=102_174,
+                last_price=6_100_000,
+                last_price_per_m2=94_721,
+                publish_date=date(2026, 2, 2),
+                exposure_days=111,
+                status="В продаже",
+            ),
+        ],
+    )
+
+
+def test_cache_key_deterministic():
+    k1 = _yandex_valuation_cache_key("addr X", "APARTMENT", "SELL")
+    k2 = _yandex_valuation_cache_key("addr X", "APARTMENT", "SELL")
+    assert k1 == k2
+    assert len(k1) == 64
+
+
+def test_cache_key_different_for_different_inputs():
+    a = _yandex_valuation_cache_key("addr X", "APARTMENT", "SELL")
+    b = _yandex_valuation_cache_key("addr Y", "APARTMENT", "SELL")
+    c = _yandex_valuation_cache_key("addr X", "HOUSE", "SELL")
+    d = _yandex_valuation_cache_key("addr X", "APARTMENT", "RENT")
+    assert len({a, b, c, d}) == 4
+
+
+def test_ttl_default_24h():
+    assert YANDEX_VALUATION_CACHE_TTL_HOURS == 24
+
+
+@pytest.mark.asyncio
+async def test_cache_hit_returns_deserialized_result():
+    """When cache row exists and hasn't expired, return parsed YandexValuationResult."""
+    db = MagicMock()
+    result = _sample_result()
+    db.execute.return_value.mappings.return_value.first.return_value = {
+        "raw_payload": result.model_dump(mode="json"),
+        "fetched_at": None,
+    }
+    with patch("app.services.estimator.YandexValuationScraper") as mock_scraper_cls:
+        out = await _get_or_fetch_yandex_valuation_cached(db, address=result.address)
+    mock_scraper_cls.assert_not_called()  # no fetch on hit
+    assert out is not None
+    assert out.address == result.address
+    assert len(out.history_items) == 2
+
+
+@pytest.mark.asyncio
+async def test_cache_miss_triggers_fetch_and_persist():
+    """On cache miss, fetch via scraper + INSERT into external_valuations."""
+    db = MagicMock()
+    db.execute.return_value.mappings.return_value.first.return_value = None  # miss
+    fresh = _sample_result()
+    fake_scraper = MagicMock()
+    fake_scraper.__aenter__ = AsyncMock(return_value=fake_scraper)
+    fake_scraper.__aexit__ = AsyncMock(return_value=False)
+    fake_scraper.fetch_house_history = AsyncMock(return_value=fresh)
+    with patch("app.services.estimator.YandexValuationScraper", return_value=fake_scraper):
+        out = await _get_or_fetch_yandex_valuation_cached(db, address=fresh.address)
+    assert out is fresh
+    fake_scraper.fetch_house_history.assert_awaited_once()
+    # 2 DB calls: 1 SELECT (miss) + 1 INSERT/UPSERT
+    assert db.execute.call_count >= 2
+
+
+@pytest.mark.asyncio
+async def test_fetch_error_returns_none_gracefully():
+    db = MagicMock()
+    db.execute.return_value.mappings.return_value.first.return_value = None
+    fake_scraper = MagicMock()
+    fake_scraper.__aenter__ = AsyncMock(return_value=fake_scraper)
+    fake_scraper.__aexit__ = AsyncMock(return_value=False)
+    fake_scraper.fetch_house_history = AsyncMock(side_effect=RuntimeError("network"))
+    with patch("app.services.estimator.YandexValuationScraper", return_value=fake_scraper):
+        out = await _get_or_fetch_yandex_valuation_cached(db, address="addr")
+    assert out is None
+
+
+@pytest.mark.asyncio
+async def test_fetch_returns_none_propagates_none():
+    db = MagicMock()
+    db.execute.return_value.mappings.return_value.first.return_value = None
+    fake_scraper = MagicMock()
+    fake_scraper.__aenter__ = AsyncMock(return_value=fake_scraper)
+    fake_scraper.__aexit__ = AsyncMock(return_value=False)
+    fake_scraper.fetch_house_history = AsyncMock(return_value=None)
+    with patch("app.services.estimator.YandexValuationScraper", return_value=fake_scraper):
+        out = await _get_or_fetch_yandex_valuation_cached(db, address="addr")
+    assert out is None
+
+
+def test_save_history_items_inserts_each():
+    db = MagicMock()
+    result = _sample_result()
+    saved = _save_yandex_history_items(db, result)
+    assert saved == 2
+    # 2 INSERTs + 1 commit
+    assert db.execute.call_count == 2
+    db.commit.assert_called_once()
+
+
+def test_save_history_items_empty_no_commit():
+    db = MagicMock()
+    result = YandexValuationResult(
+        address="x",
+        offer_category="APARTMENT",
+        offer_type="SELL",
+        page=1,
+        source_url="https://x",
+        house=ValuationHouseMeta(),
+        history_items=[],
+    )
+    saved = _save_yandex_history_items(db, result)
+    assert saved == 0
+    db.execute.assert_not_called()
+    db.commit.assert_not_called()
+
+
+def test_save_history_items_ext_id_stable_across_calls():
+    """Same result → same ext_item_id sequence → idempotent UPSERT."""
+    db1 = MagicMock()
+    db2 = MagicMock()
+    result = _sample_result()
+    _save_yandex_history_items(db1, result)
+    _save_yandex_history_items(db2, result)
+    # Both got same call params (ext_id derived from same seed)
+    ext_ids_1 = [c.args[1]["ext_id"] for c in db1.execute.call_args_list]
+    ext_ids_2 = [c.args[1]["ext_id"] for c in db2.execute.call_args_list]
+    assert ext_ids_1 == ext_ids_2
+
+
+def test_save_history_items_db_error_continues():
+    """One item failing doesn't abort the others."""
+    db = MagicMock()
+    db.execute.side_effect = [RuntimeError("first row fails"), None]
+    result = _sample_result()
+    saved = _save_yandex_history_items(db, result)
+    assert saved == 1  # second item still saved