feat(tradein): source-quota in estimator analogs (fix Cian/Yandex starvation) (#491)

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

@@ -55,6 +55,8 @@ logger = logging.getLogger(__name__)
 DEFAULT_RADIUS_M = 1000   # ПО ВСТРЕЧЕ ПТИЦЫ: «локация не дальше 800-1000 м»
 FALLBACK_RADIUS_M = 2000
 AREA_TOLERANCE = 0.15        # ±15% площади
+MAX_ANALOGS_PER_ADDRESS = 5  # анти-bias: не больше 5 лотов с одного адреса
+MIN_ANALOGS_PER_SOURCE = 5   # гарантированный минимум на live source
 LISTINGS_FRESH_DAYS = 14     # объявления не старше 14 дней
 DEALS_PERIOD_MONTHS = 12     # сделки за последний год
 
@@ -785,45 +787,91 @@ def _fetch_analogs(
     и совпадение типа дома. Так аналог «рядом + та же эпоха дома» побеждает
     аналог «чуть ближе, но дом на 30 лет старше».
 
+    Стратифицированная выборка (Approach B):
+    1. SQL вытягивает до 300 кандидатов с per-address row_number (cap MAX_ANALOGS_PER_ADDRESS).
+    2. Python гарантирует MIN_ANALOGS_PER_SOURCE слотов каждому live source.
+    3. Оставшиеся слоты заполняются из остальных кандидатов по relevance.
+    4. Итоговый список отсортирован по relevance, LIMIT 50.
+
     Returns:
         (list_of_listings_as_dicts, fallback_radius_used_flag)
     """
     rows = db.execute(
         text(
             """
+            WITH base AS (
+                SELECT
+                    source, source_url, address, lat, lon,
+                    rooms, area_m2, floor, total_floors,
+                    price_rub, price_per_m2,
+                    listing_date, days_on_market, photo_urls,
+                    scraped_at,
+                    ST_Distance(geom::geography, ST_MakePoint(:lon, :lat)::geography)
+                        AS distance_m,
+                    (
+                        ST_Distance(geom::geography, ST_MakePoint(:lon, :lat)::geography)
+                            / 1000.0
+                        -- CAST обязателен: target_year / target_house_type приходят NULL
+                        -- без типа → PostgreSQL "could not determine data type of parameter"
+                        -- (AmbiguousParameter). Явный тип снимает неоднозначность.
+                        + CASE
+                            WHEN CAST(:target_year AS integer) IS NOT NULL
+                                 AND year_built IS NOT NULL
+                            THEN abs(year_built - CAST(:target_year AS integer)) / 12.0
+                            ELSE 0
+                          END
+                        + CASE
+                            WHEN CAST(:target_house_type AS text) IS NOT NULL
+                                 AND house_type IS NOT NULL
+                                 AND house_type <> CAST(:target_house_type AS text)
+                            THEN 1.5
+                            ELSE 0
+                          END
+                    ) AS relevance_score,
+                    row_number() OVER (
+                        PARTITION BY address
+                        ORDER BY (
+                            ST_Distance(geom::geography, ST_MakePoint(:lon, :lat)::geography)
+                                / 1000.0
+                            + CASE
+                                WHEN CAST(:target_year AS integer) IS NOT NULL
+                                     AND year_built IS NOT NULL
+                                THEN abs(year_built - CAST(:target_year AS integer)) / 12.0
+                                ELSE 0
+                              END
+                            + CASE
+                                WHEN CAST(:target_house_type AS text) IS NOT NULL
+                                     AND house_type IS NOT NULL
+                                     AND house_type <> CAST(:target_house_type AS text)
+                                THEN 1.5
+                                ELSE 0
+                              END
+                        )
+                    ) AS rn_addr
+                FROM listings
+                WHERE ST_DWithin(geom::geography, ST_MakePoint(:lon, :lat)::geography, :radius)
+                  AND rooms = :rooms
+                  AND area_m2 BETWEEN :area_min AND :area_max
+                  AND is_active = true
+                  AND scraped_at > NOW() - (:fresh_days || ' days')::interval
+                  AND price_rub > 0
+                  -- 2026-05-23: Avito coords теперь real (PR #487 убрал jitter после
+                  -- C-5 audit). Listings с NULL coords отфильтруются через ST_DWithin
+                  -- (geom IS NULL → не matches). geocode-missing-listings backfill
+                  -- подтягивает координаты для address-only Avito листингов.
+            )
             SELECT
                 source, source_url, address, lat, lon,
                 rooms, area_m2, floor, total_floors,
                 price_rub, price_per_m2,
                 listing_date, days_on_market, photo_urls,
                 scraped_at,
-                ST_Distance(geom::geography, ST_MakePoint(:lon, :lat)::geography) AS distance_m
-            FROM listings
-            WHERE ST_DWithin(geom::geography, ST_MakePoint(:lon, :lat)::geography, :radius)
-              AND rooms = :rooms
-              AND area_m2 BETWEEN :area_min AND :area_max
-              AND is_active = true
-              AND scraped_at > NOW() - (:fresh_days || ' days')::interval
-              AND price_rub > 0
-              -- 2026-05-23: Avito coords теперь real (PR #487 убрал jitter после
-              -- C-5 audit). Listings с NULL coords отфильтруются через ST_DWithin
-              -- (geom IS NULL → не matches). geocode-missing-listings backfill
-              -- подтягивает координаты для address-only Avito листингов.
-            ORDER BY (
-                -- distance_m — это SELECT-алиас. В ORDER BY-ВЫРАЖЕНИИ (не голым
-                -- термом) PostgreSQL трактует имя как входную колонку listings,
-                -- которой нет → "column distance_m does not exist". Инлайним ST_Distance.
-                ST_Distance(geom::geography, ST_MakePoint(:lon, :lat)::geography) / 1000.0
-                -- CAST обязателен: target_year / target_house_type приходят NULL
-                -- без типа → PostgreSQL "could not determine data type of parameter"
-                -- (AmbiguousParameter). Явный тип снимает неоднозначность.
-                + CASE WHEN CAST(:target_year AS integer) IS NOT NULL AND year_built IS NOT NULL
-                       THEN abs(year_built - CAST(:target_year AS integer)) / 12.0 ELSE 0 END
-                + CASE WHEN CAST(:target_house_type AS text) IS NOT NULL AND house_type IS NOT NULL
-                            AND house_type <> CAST(:target_house_type AS text)
-                       THEN 1.5 ELSE 0 END
-            )
-            LIMIT 50
+                distance_m,
+                relevance_score
+            FROM base
+            WHERE rn_addr <= :max_per_addr
+            ORDER BY relevance_score
+            LIMIT 300
             """
         ),
         {
@@ -836,10 +884,45 @@ def _fetch_analogs(
             "fresh_days": LISTINGS_FRESH_DAYS,
             "target_year": year_built,
             "target_house_type": house_type,
+            "max_per_addr": MAX_ANALOGS_PER_ADDRESS,
         },
     ).mappings().all()
 
-    return [dict(r) for r in rows], radius_m > DEFAULT_RADIUS_M
+    candidates: list[dict[str, Any]] = [dict(r) for r in rows]
+
+    # Stratified quota: гарантируем MIN_ANALOGS_PER_SOURCE слотов каждому source.
+    # Candidates уже отсортированы по relevance_score (лучшие первые) из SQL.
+    guaranteed: list[dict[str, Any]] = []
+    guaranteed_ids: set[int] = set()  # по object id, не по внешнему ключу
+    by_source: dict[str, list[dict[str, Any]]] = {}
+    for row in candidates:
+        src = row.get("source") or "unknown"
+        by_source.setdefault(src, []).append(row)
+
+    for _src, src_rows in by_source.items():
+        quota = min(len(src_rows), MIN_ANALOGS_PER_SOURCE)
+        for row in src_rows[:quota]:
+            if id(row) not in guaranteed_ids:
+                guaranteed.append(row)
+                guaranteed_ids.add(id(row))
+
+    # Оставшиеся слоты из candidates, которые ещё не попали в guaranteed.
+    remaining_slots = 50 - len(guaranteed)
+    remainder: list[dict[str, Any]] = []
+    if remaining_slots > 0:
+        for row in candidates:
+            if id(row) not in guaranteed_ids:
+                remainder.append(row)
+                if len(remainder) >= remaining_slots:
+                    break
+
+    result = guaranteed + remainder
+    # Финальная сортировка по relevance (candidates из SQL уже отсортированы,
+    # но guaranteed + remainder смешиваются). relevance_score присутствует в каждом dict.
+    result.sort(key=lambda r: r.get("relevance_score") or 0.0)
+    result = result[:50]
+
+    return result, radius_m > DEFAULT_RADIUS_M
 
 
 def _fetch_deals(

tradein-mvp/backend/tests/test_estimator_source_quota.py

@@ -0,0 +1,189 @@
+"""Tests for _fetch_analogs per-address cap and per-source quota (source starvation fix).
+
+Regression: Монтёрская 8/2 — 91 Avito listings с distance=0 выдавливали
+Cian/Yandex/N1 из топ-50, т.к. pure-distance sort + LIMIT 50.
+Fix: MAX_ANALOGS_PER_ADDRESS cap в SQL + MIN_ANALOGS_PER_SOURCE quota в Python.
+"""
+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 UTC, datetime
+from typing import Any
+from unittest.mock import MagicMock
+
+# ── Helpers ───────────────────────────────────────────────────────────────────
+
+def _make_listing(
+    *,
+    source: str,
+    address: str,
+    distance_m: float,
+    relevance_score: float | None = None,
+    price_rub: float = 5_000_000.0,
+    area_m2: float = 38.0,
+    rooms: int = 1,
+) -> dict[str, Any]:
+    """Construct a minimal listing dict mimicking DB mapping output."""
+    if relevance_score is None:
+        relevance_score = distance_m / 1000.0
+    return {
+        "source": source,
+        "source_url": f"https://{source}.ru/offer/1",
+        "address": address,
+        "lat": 56.838,
+        "lon": 60.595,
+        "rooms": rooms,
+        "area_m2": area_m2,
+        "floor": 3,
+        "total_floors": 16,
+        "price_rub": price_rub,
+        "price_per_m2": price_rub / area_m2,
+        "listing_date": datetime(2026, 5, 1),
+        "days_on_market": 10,
+        "photo_urls": [],
+        "scraped_at": datetime(2026, 5, 20, tzinfo=UTC),
+        "distance_m": distance_m,
+        "relevance_score": relevance_score,
+    }
+
+
+def _make_db_mock(rows: list[dict[str, Any]]) -> MagicMock:
+    """Build a Session mock where db.execute().mappings().all() returns rows."""
+    db = MagicMock()
+    db.execute.return_value.mappings.return_value.all.return_value = rows
+    return db
+
+
+# ── Test 1: per-address cap ───────────────────────────────────────────────────
+
+def test_address_cap_limits_per_address_listings() -> None:
+    """_fetch_analogs caps at MAX_ANALOGS_PER_ADDRESS listings from a single address.
+
+    SQL already applies rn_addr <= MAX_ANALOGS_PER_ADDRESS via window function.
+    This test verifies the Python post-processing does not accidentally bypass
+    the cap by confirming that when SQL returns exactly MAX_ANALOGS_PER_ADDRESS
+    rows per address, the result contains no more than that.
+    """
+    from app.services.estimator import MAX_ANALOGS_PER_ADDRESS, _fetch_analogs
+
+    # SQL has already applied rn_addr <= MAX_ANALOGS_PER_ADDRESS.
+    # Simulate: SQL returns exactly MAX_ANALOGS_PER_ADDRESS avito rows (cap enforced).
+    addr = "ул. Монтёрская, 8/2"
+    sql_rows = [
+        _make_listing(source="avito", address=addr, distance_m=0.0, relevance_score=float(i))
+        for i in range(MAX_ANALOGS_PER_ADDRESS)
+    ]
+    db = _make_db_mock(sql_rows)
+
+    result, fallback_used = _fetch_analogs(
+        db, lat=56.838, lon=60.595, rooms=1, area=38.0, radius_m=1000
+    )
+
+    avito_from_addr = [r for r in result if r["source"] == "avito" and r["address"] == addr]
+    assert len(avito_from_addr) <= MAX_ANALOGS_PER_ADDRESS, (
+        f"Expected at most {MAX_ANALOGS_PER_ADDRESS} avito from same address, "
+        f"got {len(avito_from_addr)}"
+    )
+    assert fallback_used is False
+
+
+# ── Test 2: source quota (regression for Cian starvation) ────────────────────
+
+def test_source_quota_prevents_cian_starvation() -> None:
+    """MIN_ANALOGS_PER_SOURCE guarantees Cian is represented despite Avito dominance.
+
+    Regression: Монтёрская 8/2 — 60 Avito @ distance=0 + 8 Cian @ distance=200m.
+    Before fix: LIMIT 50 → 50 Avito, 0 Cian.
+    After fix: result contains >= min(8, MIN_ANALOGS_PER_SOURCE) Cian.
+    """
+    from app.services.estimator import MIN_ANALOGS_PER_SOURCE, _fetch_analogs
+
+    # SQL already applied address cap. Simulate SQL result after cap:
+    # 5 avito (cap applied to large block), 8 cian (different address, 200m away).
+    avito_rows = [
+        _make_listing(source="avito", address="ул. Монтёрская, 8/2", distance_m=0.0,
+                      relevance_score=float(i) * 0.01)
+        for i in range(5)
+    ]
+    cian_rows = [
+        _make_listing(source="cian", address="ул. Монтёрская, 1", distance_m=200.0,
+                      relevance_score=0.2 + float(i) * 0.01)
+        for i in range(8)
+    ]
+    # SQL returns avito first (better relevance), then cian
+    sql_rows = avito_rows + cian_rows
+    db = _make_db_mock(sql_rows)
+
+    result, _ = _fetch_analogs(
+        db, lat=56.838, lon=60.595, rooms=1, area=38.0, radius_m=1000
+    )
+
+    cian_count = sum(1 for r in result if r["source"] == "cian")
+    expected_min = min(8, MIN_ANALOGS_PER_SOURCE)
+    assert cian_count >= expected_min, (
+        f"Expected >= {expected_min} Cian in result, got {cian_count}. "
+        "Source starvation bug not fixed."
+    )
+
+
+# ── Test 3: no source starvation when quota > supply ─────────────────────────
+
+def test_source_quota_includes_all_when_supply_below_min() -> None:
+    """When a source has fewer listings than MIN_ANALOGS_PER_SOURCE, all are included.
+
+    Seed: 5 avito (after cap) + 3 cian @ 300m. All 3 cian must appear in result.
+    """
+    from app.services.estimator import _fetch_analogs
+
+    avito_rows = [
+        _make_listing(source="avito", address="ул. Монтёрская, 8/2", distance_m=0.0,
+                      relevance_score=float(i) * 0.01)
+        for i in range(5)
+    ]
+    cian_rows = [
+        _make_listing(source="cian", address="ул. Монтёрская, 3", distance_m=300.0,
+                      relevance_score=0.3 + float(i) * 0.01)
+        for i in range(3)
+    ]
+    sql_rows = avito_rows + cian_rows
+    db = _make_db_mock(sql_rows)
+
+    result, _ = _fetch_analogs(
+        db, lat=56.838, lon=60.595, rooms=1, area=38.0, radius_m=1000
+    )
+
+    cian_count = sum(1 for r in result if r["source"] == "cian")
+    assert cian_count == 3, (
+        f"All 3 Cian listings (below MIN quota) must be included, got {cian_count}"
+    )
+    assert len(result) == 8  # 5 avito + 3 cian
+
+
+# ── Test 4: fallback signal preserved ────────────────────────────────────────
+
+def test_fallback_signal_reflects_radius() -> None:
+    """_fetch_analogs returns correct fallback_used boolean based on radius_m.
+
+    fallback_used=False when radius_m == DEFAULT_RADIUS_M (1000).
+    fallback_used=True  when radius_m == FALLBACK_RADIUS_M (2000).
+    """
+    from app.services.estimator import DEFAULT_RADIUS_M, FALLBACK_RADIUS_M, _fetch_analogs
+
+    rows = [
+        _make_listing(source="avito", address="ул. Ленина, 1", distance_m=100.0,
+                      relevance_score=0.1),
+    ]
+
+    db_default = _make_db_mock(rows)
+    _, fallback_default = _fetch_analogs(
+        db_default, lat=56.838, lon=60.595, rooms=1, area=38.0, radius_m=DEFAULT_RADIUS_M
+    )
+    assert fallback_default is False, "radius == DEFAULT should produce fallback_used=False"
+
+    db_fallback = _make_db_mock(rows)
+    _, fallback_wide = _fetch_analogs(
+        db_fallback, lat=56.838, lon=60.595, rooms=1, area=38.0, radius_m=FALLBACK_RADIUS_M
+    )
+    assert fallback_wide is True, "radius == FALLBACK should produce fallback_used=True"