feat(tradein-estimator): cohort filter + Tier 0 for analog selection quality (#519)

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

@@ -64,6 +64,37 @@ MIN_ANALOGS_PER_SOURCE = 5   # гарантированный минимум н
 LISTINGS_FRESH_DAYS = 14     # объявления не старше 14 дней
 DEALS_PERIOD_MONTHS = 12     # сделки за последний год
 
+# Когорта по году постройки — типизация массовой застройки РФ.
+# Используется как hard-filter в Tier 0 _fetch_analogs (PR 9, 2026-05-24).
+# Если target_year не задан — cohort = None → фильтр отключён, Tier 0 пропускается.
+COHORTS = (
+    # (cohort_name, year_min_inclusive, year_max_inclusive)
+    ('khrushchev',  1955, 1969),  # Хрущёвки 5-эт
+    ('brezhnev',    1970, 1989),  # Брежневка кирпич/панель 9–12-эт
+    ('late_soviet', 1985, 1999),  # Поздний СССР (overlap с brezhnev intentional)
+    ('2000s',       2000, 2010),  # Ранние новостройки
+    ('modern',      2011, 2100),  # Современные ЖК
+)
+# Минимум аналогов чтобы остаться на Tier 0 (с cohort); ниже — fallback на Tier A.
+MIN_ANALOGS_TIER_0 = 5
+
+
+def _target_cohort_range(year_built: int | None) -> tuple[int, int] | None:
+    """Maps a target year to its cohort year range [min, max] inclusive.
+
+    Returns None if year_built is None — caller will skip cohort filter.
+    Picks the FIRST matching cohort (so 1988 → 'brezhnev', not 'late_soviet').
+    """
+    if year_built is None:
+        return None
+    for _name, ymin, ymax in COHORTS:
+        if ymin <= year_built <= ymax:
+            return (ymin, ymax)
+    # Out-of-range год (например, 1900 или 2050) — cohort фильтр не применяем,
+    # лучше показать что есть в радиусе, чем 0 результатов.
+    return None
+
+
 # Поправочные коэффициенты на состояние ремонта. Аналоги в выборке — микс
 # состояний (≈ "стандартный/косметический"), коэффициент сдвигает медиану под
 # конкретный ремонт целевой квартиры. Встреча Птицы: ремонт влияет на цену.
@@ -494,15 +525,39 @@ async def estimate_quality(
             if target_house_type is None:
                 target_house_type = house_meta.house_type
 
-    # 3. House-match: S → H → W tiered lookup (see _fetch_analogs docstring).
-    #    Radius fallback still applies when W tier has < 5 results at 1km.
-    listings, fallback_used, analog_tier = _fetch_analogs(
-        db, lat=geo.lat, lon=geo.lon, rooms=payload.rooms, area=payload.area_m2,
-        radius_m=DEFAULT_RADIUS_M,
-        full_address=geo.full_address,
-        year_built=target_year, house_type=target_house_type,
-        total_floors=payload.total_floors,
-    )
+    # 3. Four-tier fallback (PR 9 — added Tier 0 with cohort filter):
+    #    0) 1km + ±15% area + cohort match (year_built — если задан)
+    #    a) 1km + ±15% area  (без cohort — drop fallback)
+    #    b) 2km + ±15% area  (fallback_used = True)
+    #    c) 2km + ±25% area  (fallback_used = True, area_widened = True)
+    cohort_range = _target_cohort_range(target_year)
+
+    if cohort_range is not None:
+        cy_min, cy_max = cohort_range
+        listings_tier0, _, analog_tier = _fetch_analogs(
+            db, lat=geo.lat, lon=geo.lon, rooms=payload.rooms, area=payload.area_m2,
+            radius_m=DEFAULT_RADIUS_M,
+            full_address=geo.full_address,
+            year_built=target_year, house_type=target_house_type,
+            total_floors=payload.total_floors,
+            cohort_year_min=cy_min, cohort_year_max=cy_max,
+        )
+    else:
+        listings_tier0 = []
+        analog_tier = 'W'
+
+    if len(listings_tier0) >= MIN_ANALOGS_TIER_0:
+        listings = listings_tier0
+        fallback_used = False
+    else:
+        # Tier 0 пуст/мал — graceful fallback на Tier A без cohort
+        listings, fallback_used, analog_tier = _fetch_analogs(
+            db, lat=geo.lat, lon=geo.lon, rooms=payload.rooms, area=payload.area_m2,
+            radius_m=DEFAULT_RADIUS_M,
+            full_address=geo.full_address,
+            year_built=target_year, house_type=target_house_type,
+            total_floors=payload.total_floors,
+        )
     area_widened = False
 
     if len(listings) < 5:
@@ -977,6 +1032,8 @@ def _fetch_analogs(
     area_tolerance: float = AREA_TOLERANCE,
     year_built: int | None = None, house_type: str | None = None,
     total_floors: int | None = None,
+    cohort_year_min: int | None = None,   # NEW: lower bound year_built inclusive
+    cohort_year_max: int | None = None,   # NEW: upper bound year_built inclusive
     # TODO: когда listings получит колонку house_id_fk — добавить ext_house_id JOIN для Tier S.
 ) -> tuple[list[dict[str, Any]], bool, str]:
     """SELECT аналогов — трёхуровневый house-match (S → H → W).
@@ -999,6 +1056,11 @@ def _fetch_analogs(
     3. Оставшиеся слоты заполняются из остальных кандидатов по relevance.
     4. Итоговый список отсортирован по relevance, LIMIT 50.
 
+    Когортный фильтр (PR 9): если переданы cohort_year_min/max — добавляется
+    hard-filter WHERE year_built BETWEEN min AND max OR year_built IS NULL.
+    NULL допускается чтобы не отсеивать листинги с неизвестным годом
+    (типично для Avito anonymous-address объявлений).
+
     Returns:
         (list_of_listings_as_dicts, fallback_radius_used_flag, tier)
         tier: 'S' | 'H' | 'W'
@@ -1216,6 +1278,16 @@ def _fetch_analogs(
                   AND is_active = true
                   AND scraped_at > NOW() - (:fresh_days || ' days')::interval
                   AND price_rub > 0
+                  -- Когортный фильтр (PR 9): отсеивает разные эпохи застройки
+                  -- (хрущёвка vs новостройка). Если cohort_year_min IS NULL —
+                  -- фильтр прозрачен. CAST обязателен — psycopg3 prepared statement
+                  -- не выводит тип $N при IS NULL в predicate (см. PR #518 fix).
+                  AND (
+                      CAST(:cohort_year_min AS integer) IS NULL
+                      OR year_built IS NULL
+                      OR year_built BETWEEN CAST(:cohort_year_min AS integer)
+                                        AND CAST(:cohort_year_max AS integer)
+                  )
                   -- 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
@@ -1246,6 +1318,8 @@ def _fetch_analogs(
             "target_year": year_built,
             "target_house_type": house_type,
             "max_per_addr": MAX_ANALOGS_PER_ADDRESS,
+            "cohort_year_min": cohort_year_min,   # NEW
+            "cohort_year_max": cohort_year_max,   # NEW
         },
     ).mappings().all()