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

Production audit (2026-05-24, estimate d4ec4610) showed analogs mixing
new high-rises (31-этаж, ~2015+) with target 1978 Khrushchev/Brezhnev-era
9-floor building — relevance_score gave too little weight to year mismatch.
Median price drifted; CV at 32%.

Add cohort hard-filter as new Tier 0 in fallback cascade:
  Tier 0 (NEW): 1km + ±15% area + year cohort match
  Tier A:       1km + ±15% area  (no cohort — graceful drop if Tier 0 < 5)
  Tier B:       2km + ±15% area
  Tier C:       2km + ±25% area

5 cohorts (khrushchev 1955-69, brezhnev 1970-89, late_soviet 1985-99,
2000s 2000-10, modern 2011+). target_year=None or out-of-range -> cohort
skipped, Tier 0 not attempted.

Listings with year_built IS NULL pass through cohort filter (gentle --
don't penalize scrapers with incomplete data).

No schema/API changes. relevance_score weights untouched (separate concern).
This commit is contained in:
lekss361 2026-05-24 15:51:15 +03:00
parent a0d50bd282
commit 1d4d58538d

View file

@ -64,6 +64,37 @@ MIN_ANALOGS_PER_SOURCE = 5 # гарантированный минимум н
LISTINGS_FRESH_DAYS = 14 # объявления не старше 14 дней LISTINGS_FRESH_DAYS = 14 # объявления не старше 14 дней
DEALS_PERIOD_MONTHS = 12 # сделки за последний год 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), # Брежневка кирпич/панель 912-эт
('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: if target_house_type is None:
target_house_type = house_meta.house_type target_house_type = house_meta.house_type
# 3. House-match: S → H → W tiered lookup (see _fetch_analogs docstring). # 3. Four-tier fallback (PR 9 — added Tier 0 with cohort filter):
# Radius fallback still applies when W tier has < 5 results at 1km. # 0) 1km + ±15% area + cohort match (year_built — если задан)
listings, fallback_used, analog_tier = _fetch_analogs( # a) 1km + ±15% area (без cohort — drop fallback)
db, lat=geo.lat, lon=geo.lon, rooms=payload.rooms, area=payload.area_m2, # b) 2km + ±15% area (fallback_used = True)
radius_m=DEFAULT_RADIUS_M, # c) 2km + ±25% area (fallback_used = True, area_widened = True)
full_address=geo.full_address, cohort_range = _target_cohort_range(target_year)
year_built=target_year, house_type=target_house_type,
total_floors=payload.total_floors, 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 area_widened = False
if len(listings) < 5: if len(listings) < 5:
@ -977,6 +1032,8 @@ def _fetch_analogs(
area_tolerance: float = AREA_TOLERANCE, area_tolerance: float = AREA_TOLERANCE,
year_built: int | None = None, house_type: str | None = None, year_built: int | None = None, house_type: str | None = None,
total_floors: int | 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. # TODO: когда listings получит колонку house_id_fk — добавить ext_house_id JOIN для Tier S.
) -> tuple[list[dict[str, Any]], bool, str]: ) -> tuple[list[dict[str, Any]], bool, str]:
"""SELECT аналогов — трёхуровневый house-match (S → H → W). """SELECT аналогов — трёхуровневый house-match (S → H → W).
@ -999,6 +1056,11 @@ def _fetch_analogs(
3. Оставшиеся слоты заполняются из остальных кандидатов по relevance. 3. Оставшиеся слоты заполняются из остальных кандидатов по relevance.
4. Итоговый список отсортирован по relevance, LIMIT 50. 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: Returns:
(list_of_listings_as_dicts, fallback_radius_used_flag, tier) (list_of_listings_as_dicts, fallback_radius_used_flag, tier)
tier: 'S' | 'H' | 'W' tier: 'S' | 'H' | 'W'
@ -1216,6 +1278,16 @@ def _fetch_analogs(
AND is_active = true AND is_active = true
AND scraped_at > NOW() - (:fresh_days || ' days')::interval AND scraped_at > NOW() - (:fresh_days || ' days')::interval
AND price_rub > 0 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 после -- 2026-05-23: Avito coords теперь real (PR #487 убрал jitter после
-- C-5 audit). Listings с NULL coords отфильтруются через ST_DWithin -- C-5 audit). Listings с NULL coords отфильтруются через ST_DWithin
-- (geom IS NULL не matches). geocode-missing-listings backfill -- (geom IS NULL не matches). geocode-missing-listings backfill
@ -1246,6 +1318,8 @@ def _fetch_analogs(
"target_year": year_built, "target_year": year_built,
"target_house_type": house_type, "target_house_type": house_type,
"max_per_addr": MAX_ANALOGS_PER_ADDRESS, "max_per_addr": MAX_ANALOGS_PER_ADDRESS,
"cohort_year_min": cohort_year_min, # NEW
"cohort_year_max": cohort_year_max, # NEW
}, },
).mappings().all() ).mappings().all()