2026-08-05 18:16:58 +00:00
8 changed files with 775 additions and 6 deletions
--- a/tradein-mvp/backend/app/api/v1/admin.py
+++ b/tradein-mvp/backend/app/api/v1/admin.py
@ -72,6 +72,7 @@ from app.services import cian_session as cian_session_svc
 from app.services import domclick_session as domclick_session_svc
 from app.services import proxy_rotation as proxy_rotation_svc
 from app.services import scrape_runs as runs_mod
+from app.services.estimator import LISTINGS_FRESH_DAYS
 from app.services.geocoder import geocode, known_city_hint
 from app.services.proxy_pool import clear_source_bans
 from app.services.scheduler import has_running_run
@ -2473,6 +2474,12 @@ async def update_scraper_pacing(
 class SourceCoverage(BaseModel):
    source: str
    active_count: int
+    # #2660: «активно» ≠ «живо». is_active снимается только деактиватором протухших,
+    # а он покрывает не все источники — на проде (2026-08-05) cian показывал 18 530
+    # активных при 12 683 не виденных 14+ дней. Из-за этого #2574 месяц читалась как
+    # «всё собирается». Не прячем протухшее из счётчика, а отдаём ВТОРЫМ числом
+    # рядом — тогда «активно» перестаёт читаться как «живо».
+    stale_count: int
    fields: dict[str, float]  # field_name -> fill% (0..100, round 1)


@ -2487,6 +2494,9 @@ class HousesCoverage(BaseModel):
 class DataQualityResponse(BaseModel):
    sources: list[SourceCoverage]
    houses: HousesCoverage
+    # Порог «не виделись N дней» для stale_count — отдаём в ответе, чтобы UI
+    # подписывал число, а не хардкодил порог у себя вторым определением.
+    stale_days: int


 # Поля listings для fill%-аудита. Каждый кортеж: (имя_поля, SQL-выражение IS NOT NULL).
@ -2516,6 +2526,10 @@ def get_data_quality(
    living_area_m2, ceiling_height (cian), ceiling_height_m (avito), metro_stations.
    houses: total, avito_validated_at%, rating_score%, house_type%.
    house_reviews: общий count.
+
+    #2660: рядом с active_count отдаётся stale_count — сколько из «активных» не
+    виделись LISTINGS_FRESH_DAYS дней (last_seen_at). Порог отдаётся в ответе
+    (stale_days), чтобы UI не заводил второе определение.
    """
    # Строим single-pass SELECT для listings полей через FILTER-агрегаты.
    # Структура: COUNT(*) FILTER (WHERE <expr>) / NULLIF(COUNT(*), 0) * 100
@ -2523,10 +2537,17 @@ def get_data_quality(
    filter_exprs = ", ".join(
        f"COUNT(*) FILTER (WHERE {expr}) AS f_{name}" for name, expr in _DQ_LISTING_FIELDS
    )
+    # last_seen_at, а не scraped_at: счётчик отвечает буквально на «сколько не
+    # виделись». На проде две колонки не расходятся (замер 2026-08-05: 0 активных
+    # строк с разницей ≥ суток), но семантика счётчика — про «видели», и колонка
+    # должна называть ровно её.
    sql_listings = text(f"""
        SELECT
            source,
            COUNT(*) AS active_count,
+            COUNT(*) FILTER (
+                WHERE last_seen_at <= NOW() - (:fresh_days || ' days')::interval
+            ) AS stale_count,
            {filter_exprs}
        FROM listings
        WHERE is_active = true
@ -2534,7 +2555,7 @@ def get_data_quality(
        ORDER BY source
    """)

-    rows = db.execute(sql_listings).mappings().all()
+    rows = db.execute(sql_listings, {"fresh_days": LISTINGS_FRESH_DAYS}).mappings().all()

    sources: list[SourceCoverage] = []
    for row in rows:
@ -2548,6 +2569,7 @@ def get_data_quality(
            SourceCoverage(
                source=row["source"],
                active_count=int(row["active_count"]),
+                stale_count=int(row["stale_count"] or 0),
                fields=fields,
            )
        )
@ -2575,7 +2597,7 @@ def get_data_quality(
        reviews_count=reviews_count,
    )

-    return DataQualityResponse(sources=sources, houses=houses)
+    return DataQualityResponse(sources=sources, houses=houses, stale_days=LISTINGS_FRESH_DAYS)


 # ── Proxy pool: хранилище + bulk-загрузка / список (#2161) ───────────────────
--- a/tradein-mvp/backend/app/api/v1/trade_in.py
+++ b/tradein-mvp/backend/app/api/v1/trade_in.py
@ -750,7 +750,15 @@ def cache_stats(db: Annotated[Session, Depends(get_db)]) -> dict[str, object]:
        trade_in_estimates с непустым address; NULL при отсутствии адресов.
        NB: это честный best-effort по persisted оценкам, а не hit-rate реального
        кэша (отдельного счётчика попаданий не ведём).
+
+    #2660: listings_active сам по себе врал — «активно» на проде не означает
+    «живо» (деактиватор протухших покрывает не все источники). Рядом отдаём
+    listings_active_stale — сколько из них не виделись listings_stale_days
+    (= LISTINGS_FRESH_DAYS эстиматора; прод 2026-08-05: 37 900 активных при
+    20 935 не виденных 14+ дней). Счётчик не прячем, а разделяем.
    """
+    from app.services.estimator import LISTINGS_FRESH_DAYS
+
    row = (
        db.execute(
            text(
@ -760,6 +768,10 @@ def cache_stats(db: Annotated[Session, Depends(get_db)]) -> dict[str, object]:
              (SELECT count(*) FROM geocode_cache WHERE expires_at > NOW())
                  AS geocode_cache_fresh,
              (SELECT count(*) FROM listings WHERE is_active) AS listings_active,
+              (SELECT count(*) FROM listings
+                WHERE is_active
+                  AND last_seen_at <= NOW() - (:fresh_days || ' days')::interval)
+                  AS listings_active_stale,
              (SELECT max(scraped_at) FROM listings) AS listings_last_scraped,
              (SELECT count(*) FROM deals) AS deals,
              (SELECT count(*) FROM gendesign_cad_buildings) AS cad_buildings,
@ -776,12 +788,15 @@ def cache_stats(db: Annotated[Session, Depends(get_db)]) -> dict[str, object]:
                   WHERE address IS NOT NULL AND address <> ''
                 ) t) AS repeat_address_pct
            """
-            )
+            ),
+            {"fresh_days": LISTINGS_FRESH_DAYS},
        )
        .mappings()
        .fetchone()
    )
-    return dict(row) if row else {}
+    # Порог отдаём рядом с числом — чтобы UI подписывал «не виделись N дней»,
+    # а не заводил второе определение свежести у себя.
+    return (dict(row) | {"listings_stale_days": LISTINGS_FRESH_DAYS}) if row else {}


 # ── Stage 4a: house info + IMV benchmark для UI ───────────────────────────────
--- a/tradein-mvp/backend/app/services/location_index.py
+++ b/tradein-mvp/backend/app/services/location_index.py
@ -13,9 +13,13 @@
 POI-score его не улавливал (POI ranking ≠ цена).

 НОВЫЙ ПОКАЗАТЕЛЬ (location index):
-  location_index_pct = (медиана ₽/м² сопоставимых активных листингов в радиусе точки −
+  location_index_pct = (медиана ₽/м² сопоставимых листингов в радиусе точки −
                         медиана ₽/м² по всему ЕКБ) / медиана по ЕКБ * 100

+«Сопоставимые» = ровно тот же пул, что берёт эстиматор (#2660): активные И свежие
+(scraped_at в пределах LISTINGS_FRESH_DAYS — `is_active` на проде не равно «живо») И
+только вторичка (гард #1186 — девелоперский прайс новостроек завышал обе медианы).
+
 Самообновляем (те же `listings`, что уже скрейпятся под estimator), интерпретируем напрямую
 ("район на N% дороже/дешевле среднего по городу"), устойчив к выбросам (percentile_cont(0.5) —
 медиана самой природой игнорирует единичные экстремумы, в отличие от mean/min/max), и НЕ зажат
@ -45,6 +49,11 @@ from typing import Any
 from pydantic import BaseModel
 from sqlalchemy import text

+# #2660: окно свежести берём ИЗ эстиматора — единственное определение в проекте.
+# Дублировать значение здесь нельзя: две константы разъедутся при первой же
+# перекалибровке, и витрина начнёт показывать другой пул, чем считает цена.
+from app.services.estimator import LISTINGS_FRESH_DAYS
+
 logger = logging.getLogger(__name__)

 # ── Гео-охват продукта: только Екатеринбург ──────────────────────────────────
@ -159,6 +168,43 @@ def _pct_deviation(local_median_ppm2: float, city_median_ppm2: float) -> float:
 # price_per_m2 BETWEEN sanity-границы — не бизнес-калибровка, а защита от битых строк
 # (см. _PRICE_PER_M2_SANITY_MIN/MAX выше).
 #
+# #2660 свежесть + сегмент — оба предиката ЗЕРКАЛЯТ _COMMON_WHERE эстиматора.
+# Вклад у них РАЗНЫЙ, и не тот, на который легко подумать. Прод-разложение
+# (2026-08-05, пул location_index — bbox ЕКБ + sanity ₽/м² + geo_precision):
+#
+#     было (только is_active)   30 222 строк   172 984 ₽/м²
+#     + только свежесть         11 453 строк   163 363 ₽/м²
+#     + только сегмент          11 219 строк   147 632 ₽/м²
+#     стало (оба)                7 715 строк   147 368 ₽/м²
+#
+#   - listing_segment guard (#1186) — ЭТО и есть исправление смещения: из −14.8%
+#     сдвига городской медианы он даёт −14.7 п.п. Девелоперский прайс новостроек
+#     завышал и локальную, и городскую медиану. NULL = legacy вторичка до м.011.
+#     Мертвецы, кстати, живут почти целиком тут же: из 18 769 протухших строк
+#     пула 15 265 — новостройки, и гард выносит их заодно.
+#   - scraped_at > NOW() - LISTINGS_FRESH_DAYS — даёт ПОВЕРХ сегмента всего
+#     −0.18 п.п. Для ЭТОЙ метрики он не коррекция смещения, а СТРАХОВКА на
+#     будущее (пул совпадает с пулом цены; если завтра протухнет вторичка —
+#     виджет не соврёт), и страховка не бесплатная: выбрасывает 3 504 вторичных
+#     строки, из которых 2 724 — живые объявления, отскрейпленные 15-30 дней
+#     назад. Пул −31%, шум растёт: на центре ЕКБ (r=800) n падает 423 → 86, а
+#     сам индекс гуляет по выбору окна на 12-14 п.п. (7д +75.7% / 14д +77.0% /
+#     21д +79.1% / 30д +64.7%) — при n=86 это в пределах шума выборки медианы.
+#     Размен «меньше смещения ↔ больше дисперсии» сделан осознанно: старое число
+#     было предвзятым, новое — шумным, но честным. Окно менять здесь НЕ надо,
+#     LISTINGS_FRESH_DAYS живёт в estimator.py (см. импорт выше).
+#
+# НОВЫЙ РЕЖИМ ОТКАЗА (знать обязательно): свежесть связала витрину со здоровьем
+# СБОРА. Встанет скрейпинг на LISTINGS_FRESH_DAYS — городская выборка не наберёт
+# MIN_SAMPLE_SIZE, и "insufficient_data" прилетит ВСЕМ пользователям разом; до
+# этой правки виджет продолжал бы показывать устаревшее число. Учитывая, что
+# #2574 — ровно месяц молчаливой поломки сбора, сценарий не гипотетический.
+# Деградация честная (прочерк, а не выдуманное число), но она теперь массовая.
+#
+# Порог MIN_SAMPLE_SIZE после сужения пула набирается реже, но лестница радиусов
+# упирается в отказ редко — прод-симуляция на 246 реальных точках оценок:
+# insufficient_data 0 → 1 точка (0.4%), 800м хватает 241 точке из 246.
+#
 # bbox-фильтр (lat/lon) — сопоставимые листинги считаются ТОЛЬКО по Екатеринбургу, даже если
 # сам продукт уже скрейпит соседние города области (city-sweep): географию location_index
 # явно ограничил владелец продукта.
@ -173,6 +219,8 @@ _MEDIAN_PPM2_LOCAL_SQL = text(
      AND price_per_m2 IS NOT NULL
      AND price_per_m2 BETWEEN CAST(:price_min AS integer) AND CAST(:price_max AS integer)
      AND (geo_precision IS DISTINCT FROM 'city')
+      AND scraped_at > NOW() - (:fresh_days || ' days')::interval
+      AND (listing_segment IS NULL OR listing_segment = 'vtorichka')
      AND lat BETWEEN CAST(:bbox_south AS double precision)
                  AND CAST(:bbox_north AS double precision)
      AND lon BETWEEN CAST(:bbox_west AS double precision)
@ -196,6 +244,8 @@ _MEDIAN_PPM2_CITYWIDE_SQL = text(
      AND price_per_m2 IS NOT NULL
      AND price_per_m2 BETWEEN CAST(:price_min AS integer) AND CAST(:price_max AS integer)
      AND (geo_precision IS DISTINCT FROM 'city')
+      AND scraped_at > NOW() - (:fresh_days || ' days')::interval
+      AND (listing_segment IS NULL OR listing_segment = 'vtorichka')
      AND lat BETWEEN CAST(:bbox_south AS double precision)
                  AND CAST(:bbox_north AS double precision)
      AND lon BETWEEN CAST(:bbox_west AS double precision)
@ -235,6 +285,7 @@ def _local_median_ppm2(db: Any, lat: float, lon: float, radius_m: int) -> tuple[
                "lat": lat,
                "lon": lon,
                "radius_m": radius_m,
+                "fresh_days": LISTINGS_FRESH_DAYS,
                "price_min": _PRICE_PER_M2_SANITY_MIN,
                "price_max": _PRICE_PER_M2_SANITY_MAX,
                "bbox_south": _EKB_BBOX_SOUTH,
@ -257,6 +308,7 @@ def _citywide_median_ppm2(db: Any) -> tuple[float | None, int]:
        db.execute(
            _MEDIAN_PPM2_CITYWIDE_SQL,
            {
+                "fresh_days": LISTINGS_FRESH_DAYS,
                "price_min": _PRICE_PER_M2_SANITY_MIN,
                "price_max": _PRICE_PER_M2_SANITY_MAX,
                "bbox_south": _EKB_BBOX_SOUTH,
--- a/tradein-mvp/backend/data/sql/211_sales_vs_listings_segment_guard.sql
+++ b/tradein-mvp/backend/data/sql/211_sales_vs_listings_segment_guard.sql
@ -0,0 +1,204 @@
+-- 211_sales_vs_listings_segment_guard.sql
+-- Purpose: #2660 — «медианный торг» на витрине /sales-vs-listings считался с участием
+-- ПЕРВИЧКИ. CTE window_listings в street_sales_vs_listings() (миграция 205) читает
+-- listings без сегментного гарда #1186, поэтому в пару к ДКП-сделке вторички мог
+-- встать лот застройщика: девелоперский прайс (фиксированный, без торга) формирует
+-- показываемый пользователю процент скидки.
+--
+-- Прод-замер (2026-08-05, БД tradein):
+--   - популяция кандидатов на пейринг (окно period_months+6 = 30 мес, price_rub > 0):
+--     93 241 строк, из них novostroyki 25 428 (27.3%), vtorichka 66 256,
+--     listing_segment IS NULL 1 557 (legacy до м.011 — считаем вторичкой);
+--   - симуляция по 20 самым «густым» (улица, комнаты) ЕКБ (2 120 сделок):
+--     до гарда 1 219 сделок получили listing-match, из них 299 (24.5%) — против
+--     новостройки; медианный торг −18.18%. После гарда: 994 матча, медиана −17.11%
+--     (+1.07 п.п. — торг перестаёт «утяжеляться» за счёт первички).
+--
+-- Что НЕ меняется и почему:
+--   - `is_active` в window_listings НЕ добавляем — осознанно (см. 067/205): функция
+--     намеренно смотрит и снятые объявления, иначе к сделке нечего подставить
+--     (объявление снимают ПОСЛЕ продажи — активные для пейринга бесполезны).
+--   - Свежесть здесь тоже не при чём по той же причине: пейринг привязан к дате
+--     сделки (window_days ± grace), а не к «сейчас».
+--   - Сигнатура функции НЕ меняется (те же 7 аргументов, что после м.205).
+--     CREATE OR REPLACE FUNCTION с ИЗМЕНЁННЫМ списком типов создал бы ВТОРУЮ
+--     перегрузку вместо замены (грабли #2627, разобраны в шапке 205) — здесь
+--     список типов побайтово тот же, значит замена in-place, оверлоада не будет.
+--     DROP FUNCTION намеренно НЕ делаем: 205 уже дропнула старую 6-арг сигнатуру,
+--     а дропать текущую 7-арг перед CREATE — лишний оконный простой для caller'а.
+--
+-- Сегментный предикат — канонический #1186 (тот же, что в _COMMON_WHERE эстиматора
+-- и в 109_asking_to_sold_ratio_segment_filter.sql):
+--   (l.listing_segment IS NULL OR l.listing_segment = 'vtorichka')
+-- NULL = legacy вторичка до миграции 011 (rosreestr/avito/yandex без сегмента), её
+-- отбрасывать нельзя — потеряли бы 1 557 валидных кандидатов.
+--
+-- ЗАВИСИМОСТИ: 205 (текущее тело + 7-арг сигнатура), listings.listing_segment (м.011).
+-- Deploy order: после 210. Идемпотентность: CREATE OR REPLACE + COMMENT ON — оба
+-- по конструкции re-run safe, ре-apply просто перезапишет то же тело.
+-- Caller (app/api/v1/trade_in.py, /sales-vs-listings) не меняется — сигнатура и
+-- RETURNS TABLE те же.
+
+BEGIN;
+
+CREATE OR REPLACE FUNCTION street_sales_vs_listings(
+    p_street_pattern  text,
+    p_area_m2         numeric,
+    p_rooms           integer,
+    p_window_days     integer DEFAULT 180,
+    p_area_tolerance  numeric DEFAULT 0.15,
+    p_period_months   integer DEFAULT 24,
+    p_target_city     text    DEFAULT NULL
+)
+RETURNS TABLE (
+    deal_id                bigint,
+    deal_date              date,
+    deal_price_rub         bigint,
+    deal_price_per_m2      integer,
+    deal_area_m2           numeric,
+    deal_rooms             integer,
+    deal_floor             integer,
+    deal_address           text,
+    listing_id             bigint,
+    listing_source         text,
+    listing_source_url     text,
+    listing_date           date,
+    listing_price_rub      bigint,
+    listing_price_per_m2   integer,
+    listing_area_m2        numeric,
+    days_listing_to_deal   integer,
+    discount_pct           numeric
+)
+LANGUAGE sql
+STABLE
+AS $$
+    WITH window_deals AS (
+        -- Сделки в улице + период. Фильтр по rooms + area + (#2583 H4) city.
+        SELECT
+            d.id           AS deal_id,
+            d.deal_date    AS deal_date,
+            d.price_rub    AS deal_price_rub,
+            d.price_per_m2 AS deal_price_per_m2,
+            d.area_m2      AS deal_area_m2,
+            d.rooms        AS deal_rooms,
+            d.floor        AS deal_floor,
+            d.address      AS deal_address
+        FROM deals d
+        WHERE d.source = 'rosreestr'
+          AND d.address ILIKE p_street_pattern
+          AND d.rooms = p_rooms
+          AND d.area_m2 BETWEEN p_area_m2 * (1.0 - p_area_tolerance)
+                            AND p_area_m2 * (1.0 + p_area_tolerance)
+          AND d.deal_date > NOW() - (p_period_months || ' months')::interval
+          AND d.price_rub > 0
+          -- #2583 H4: deals.city заполнена на 100% — строгое равенство.
+          -- NULL p_target_city (город вне словаря) → фильтр не применяется.
+          AND (p_target_city IS NULL OR LOWER(d.city) = LOWER(p_target_city))
+    ),
+    window_listings AS (
+        -- Кандидаты-listings на той же улице, rooms exact, area ±tolerance,
+        -- (#2583 H4) тот же город что deals-сторона, (#2660) только вторичка.
+        SELECT
+            l.id                 AS listing_id,
+            l.source             AS listing_source,
+            l.source_url         AS listing_source_url,
+            l.listing_date       AS listing_date,
+            l.price_rub          AS listing_price_rub,
+            l.price_per_m2       AS listing_price_per_m2,
+            l.area_m2            AS listing_area_m2,
+            l.rooms              AS listing_rooms,
+            COALESCE(l.listing_date, l.scraped_at::date) AS listing_event_date
+        FROM listings l
+        WHERE l.address ILIKE p_street_pattern
+          AND l.rooms = p_rooms
+          AND l.area_m2 BETWEEN p_area_m2 * (1.0 - p_area_tolerance)
+                            AND p_area_m2 * (1.0 + p_area_tolerance)
+          AND l.price_rub > 0
+          AND COALESCE(l.listing_date, l.scraped_at::date)
+                > NOW() - ((p_period_months + 6) || ' months')::interval
+          -- #2583 H4: listings.city заполнена ЧАСТИЧНО (прод: avito 63%,
+          -- yandex 19%, cian 4.6%, domklik 0.6%, n1 0%) — NULL считается "своим"
+          -- (симметрично asking_to_sold_ratio.py #2583 H2), иначе строгий
+          -- фильтр выбросил бы почти все listings кроме avito.
+          AND (p_target_city IS NULL OR l.city IS NULL OR LOWER(l.city) = LOWER(p_target_city))
+          -- #2660 novostroyki guard (#1186): к ДКП-сделке вторички нельзя
+          -- подставлять лот застройщика — девелоперский прайс не торгуется и
+          -- уводит показываемый «медианный торг». Прод: 27.3% кандидатов —
+          -- первичка. NULL = legacy вторичка до м.011, оставляем.
+          AND (l.listing_segment IS NULL OR l.listing_segment = 'vtorichka')
+    ),
+    paired AS (
+        -- LEFT JOIN: сохраняем все сделки даже если нет listing match.
+        -- Для каждой сделки выбираем listing с listing_date ближайший
+        -- к deal_date (предпочтительно перед сделкой).
+        SELECT DISTINCT ON (wd.deal_id)
+            wd.deal_id,
+            wd.deal_date,
+            wd.deal_price_rub,
+            wd.deal_price_per_m2,
+            wd.deal_area_m2,
+            wd.deal_rooms,
+            wd.deal_floor,
+            wd.deal_address,
+            wl.listing_id,
+            wl.listing_source,
+            wl.listing_source_url,
+            wl.listing_date,
+            wl.listing_price_rub,
+            wl.listing_price_per_m2,
+            wl.listing_area_m2,
+            (wd.deal_date - wl.listing_event_date)::integer AS days_listing_to_deal,
+            CASE
+                WHEN wl.listing_price_rub IS NOT NULL AND wl.listing_price_rub > 0
+                THEN ROUND(
+                    (wd.deal_price_rub - wl.listing_price_rub)::numeric
+                    / wl.listing_price_rub * 100,
+                    2
+                )
+                ELSE NULL
+            END AS discount_pct
+        FROM window_deals wd
+        LEFT JOIN window_listings wl
+          ON wl.listing_event_date
+                BETWEEN (wd.deal_date - (p_window_days || ' days')::interval)::date
+                    AND (wd.deal_date + interval '30 days')::date
+        ORDER BY
+            wd.deal_id,
+            -- prefer listing event дата перед сделкой и ближе к ней
+            CASE WHEN wl.listing_event_date IS NULL THEN 1 ELSE 0 END,
+            CASE WHEN wl.listing_event_date <= wd.deal_date THEN 0 ELSE 1 END,
+            ABS((wd.deal_date - wl.listing_event_date))
+    )
+    SELECT
+        deal_id,
+        deal_date,
+        deal_price_rub,
+        deal_price_per_m2,
+        deal_area_m2,
+        deal_rooms,
+        deal_floor,
+        deal_address,
+        listing_id,
+        listing_source,
+        listing_source_url,
+        listing_date,
+        listing_price_rub,
+        listing_price_per_m2,
+        listing_area_m2,
+        days_listing_to_deal,
+        discount_pct
+    FROM paired
+    ORDER BY deal_date DESC;
+$$;
+
+COMMENT ON FUNCTION street_sales_vs_listings(text, numeric, integer, integer, numeric, integer, text) IS
+  'Pairs (ДКП-сделка, listing) для улицы. PR K / issue #564 Foundation Phase 1, '
+  'city-filter #2583 H4 (миграция 205), segment-guard #2660/#1186 (миграция 211). '
+  'Per-street matching: address ILIKE, area ±tolerance, rooms exact, window_days '
+  'до даты сделки (+30д grace), city-scope (p_target_city, deals строго / listings '
+  'терпимо к NULL), listings — только вторичка (listing_segment IS NULL или '
+  'vtorichka). Возвращает LEFT JOIN — сделки без listing match имеют '
+  'listing_* = NULL. discount_pct = (deal - listing) / listing * 100. '
+  'is_active намеренно НЕ фильтруется: снятые объявления и есть материал пейринга.';
+
+COMMIT;
--- a/tradein-mvp/backend/tests/services/test_location_index.py
+++ b/tradein-mvp/backend/tests/services/test_location_index.py
@ -138,6 +138,65 @@ def test_median_queries_exclude_city_centroid_and_bound_bbox() -> None:
        assert "bbox_west" in sql_text and "bbox_east" in sql_text


+# ── #2660: свежесть + сегмент (зеркало _COMMON_WHERE эстиматора) ─────────────
+
+
+def test_median_queries_filter_stale_listings() -> None:
+    """#2660: `is_active` на проде НЕ означает «живо» — деактиватор протухших покрывает не
+    все источники. Без предиката свежести медиана считалась по мертвецам: прод-замер по
+    пулу location_index (bbox ЕКБ) — 172 984 ₽/м² по всему активному пулу против
+    163 363 по свежим. Оба запроса (локальный и общегородской) должны фильтровать
+    симметрично, иначе индекс = отношение двух разных популяций."""
+    for sql in (lc._MEDIAN_PPM2_LOCAL_SQL, lc._MEDIAN_PPM2_CITYWIDE_SQL):
+        sql_text = str(sql.text)
+        assert "scraped_at > NOW() - (:fresh_days || ' days')::interval" in sql_text
+
+
+def test_median_queries_exclude_novostroyki() -> None:
+    """Сегментный гард #1186 — тот же канонический предикат, что в _COMMON_WHERE
+    эстиматора. NULL = legacy вторичка до м.011 и должен остаться в пуле."""
+    for sql in (lc._MEDIAN_PPM2_LOCAL_SQL, lc._MEDIAN_PPM2_CITYWIDE_SQL):
+        sql_text = str(sql.text)
+        assert "(listing_segment IS NULL OR listing_segment = 'vtorichka')" in sql_text
+
+
+def test_freshness_window_is_the_estimator_constant_not_a_copy() -> None:
+    """Второе определение свежести разъедется с эстиматором при первой перекалибровке —
+    витрина начнёт показывать пул, отличный от того, на котором считается цена.
+
+    Проверяем ИСХОДНИК, а не равенство значений: CPython кэширует малые int (−5…256),
+    поэтому `lc.LISTINGS_FRESH_DAYS is estimator.LISTINGS_FRESH_DAYS` проходит и для
+    скопированного литерала `LISTINGS_FRESH_DAYS = 14` — то есть стережёт не то, что
+    обещает. Отсутствие присваивания в модуле = константа может быть только импортом.
+    """
+    import inspect
+
+    from app.services import estimator
+
+    assert "LISTINGS_FRESH_DAYS =" not in inspect.getsource(
+        lc
+    ), "константа скопирована в location_index — она должна ИМПОРТИРОВАТЬСЯ из estimator"
+    assert lc.LISTINGS_FRESH_DAYS == estimator.LISTINGS_FRESH_DAYS
+
+
+def test_both_median_queries_bind_the_same_freshness_window() -> None:
+    """Свежесть должна реально долетать в оба запроса bind-параметром (а не остаться
+    только в тексте SQL) и быть ОДИНАКОВОЙ — иначе локальная и городская медианы
+    считаются по разным окнам, и индекс — отношение несопоставимых величин."""
+    db = _FakeDB(
+        [
+            _FakeResult(scalar_value=0),  # poi mirror empty
+            _FakeResult(mapping_one={"median_ppm2": 150_000.0, "n": 4000}),  # citywide
+            _FakeResult(mapping_one={"median_ppm2": 165_000.0, "n": 25}),  # radius[0]
+        ]
+    )
+    lc.compute_location_index(db, lat=_LAT_IN_EKB, lon=_LON_IN_EKB)
+    citywide_params = db.executed[1][1]
+    local_params = db.executed[2][1]
+    assert citywide_params["fresh_days"] == lc.LISTINGS_FRESH_DAYS
+    assert local_params["fresh_days"] == lc.LISTINGS_FRESH_DAYS
+
+
 # ── _fetch_nearby_poi (qualitative "что рядом" list) ─────────────────────────


--- a/tradein-mvp/backend/tests/test_admin_stale_counters.py
+++ b/tradein-mvp/backend/tests/test_admin_stale_counters.py
@ -0,0 +1,205 @@
+"""#2660: админ-счётчики честно разделяют «активно» и «из них не виделись N дней».
+
+Почему это баг, а не косметика: `is_active` снимается только деактиватором протухших,
+а он покрывает не все источники — прод (2026-08-05) показывал cian 18 530 «активных»
+при 12 683 не виденных 14+ дней и 10 212 не виденных 30+. Именно поэтому #2574 месяц
+читалась как «всё собирается». Счётчик не прячем (иначе потеряем видимость проблемы) —
+отдаём вторым числом рядом.
+
+Покрыто (db мокается, NO live network/DB):
+  - GET /api/v1/admin/scraper/data-quality — stale_count per source + stale_days;
+  - GET /api/v1/trade-in/cache-stats       — listings_active_stale + listings_stale_days;
+  - порог свежести = LISTINGS_FRESH_DAYS эстиматора (не второе определение);
+  - оба запроса считают протухшее по last_seen_at И только среди is_active.
+"""
+
+from __future__ import annotations
+
+import os
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from typing import Any
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+
+class _FakeResult:
+    def __init__(
+        self,
+        *,
+        rows: list[dict] | None = None,
+        one: dict | None = None,
+        scalar_value: Any = None,
+    ) -> None:
+        self._rows = rows or []
+        self._one = one
+        self._scalar = scalar_value
+
+    def scalar(self) -> Any:
+        return self._scalar
+
+    def mappings(self) -> Any:
+        outer = self
+
+        class _Mappings:
+            def all(self) -> list[dict]:
+                return outer._rows
+
+            def one(self) -> dict:
+                assert outer._one is not None
+                return outer._one
+
+            def fetchone(self) -> dict | None:
+                return outer._one
+
+        return _Mappings()
+
+
+class _FakeDB:
+    """Session stand-in: execute() отдаёт очередь заготовленных результатов."""
+
+    def __init__(self, results: list[_FakeResult]) -> None:
+        self._results = list(results)
+        self.executed: list[tuple[Any, Any]] = []
+
+    def execute(self, clause: Any, params: Any = None) -> _FakeResult:
+        self.executed.append((clause, params))
+        return self._results.pop(0)
+
+
+# Одна строка coverage-выборки: source + счётчики + fill-агрегаты по всем полям.
+def _dq_row(source: str, active: int, stale: int) -> dict[str, Any]:
+    from app.api.v1.admin import _DQ_LISTING_FIELDS
+
+    row: dict[str, Any] = {"source": source, "active_count": active, "stale_count": stale}
+    for name, _expr in _DQ_LISTING_FIELDS:
+        row[f"f_{name}"] = active  # 100% fill — не предмет этого теста
+    return row
+
+
+_HOUSES_ROW = {"total": 10, "validated_cnt": 5, "rating_cnt": 5, "house_type_cnt": 5}
+
+
+def _make_client(db: _FakeDB, router_module: str, prefix: str) -> TestClient:
+    import importlib
+
+    from app.core.db import get_db
+
+    module = importlib.import_module(router_module)
+    app = FastAPI()
+    app.include_router(module.router, prefix=prefix)
+    app.dependency_overrides[get_db] = lambda: db
+    return TestClient(app)
+
+
+# ── GET /scraper/data-quality ────────────────────────────────────────────────
+
+
+@pytest.fixture
+def dq_db() -> _FakeDB:
+    return _FakeDB(
+        [
+            _FakeResult(
+                rows=[
+                    # Прод-пропорции 2026-08-05: cian почти весь протухший, avito живой.
+                    _dq_row("cian", 18_530, 12_683),
+                    _dq_row("avito", 4_975, 0),
+                ]
+            ),
+            _FakeResult(one=_HOUSES_ROW),
+            _FakeResult(scalar_value=0),
+        ]
+    )
+
+
+def test_data_quality_reports_stale_next_to_active(dq_db: _FakeDB) -> None:
+    """«Активно» остаётся как было, рядом появляется «из них не виделись N дней»."""
+    client = _make_client(dq_db, "app.api.v1.admin", "/api/v1/admin")
+    r = client.get("/api/v1/admin/scraper/data-quality")
+    assert r.status_code == 200
+    body = r.json()
+    by_source = {s["source"]: s for s in body["sources"]}
+    assert by_source["cian"]["active_count"] == 18_530
+    assert by_source["cian"]["stale_count"] == 12_683
+    # Живой источник — протухших нет, счётчик не «шумит» на здоровых данных.
+    assert by_source["avito"]["active_count"] == 4_975
+    assert by_source["avito"]["stale_count"] == 0
+
+
+def test_data_quality_exposes_the_threshold_used(dq_db: _FakeDB) -> None:
+    """Порог отдаётся в ответе — UI подписывает число, а не хардкодит своё N."""
+    from app.services.estimator import LISTINGS_FRESH_DAYS
+
+    client = _make_client(dq_db, "app.api.v1.admin", "/api/v1/admin")
+    body = client.get("/api/v1/admin/scraper/data-quality").json()
+    assert body["stale_days"] == LISTINGS_FRESH_DAYS
+
+
+def test_data_quality_counts_stale_by_last_seen_within_active_only(dq_db: _FakeDB) -> None:
+    """Протухшее считается по last_seen_at (буквально «не виделись») и ТОЛЬКО среди
+    активных — иначе счётчик смешает деактивированные строки и перестанет быть
+    сопоставимым с active_count."""
+    from app.services.estimator import LISTINGS_FRESH_DAYS
+
+    client = _make_client(dq_db, "app.api.v1.admin", "/api/v1/admin")
+    client.get("/api/v1/admin/scraper/data-quality")
+    clause, params = dq_db.executed[0]
+    sql = str(clause)
+    assert "COUNT(*) FILTER (" in sql
+    assert "last_seen_at <= NOW() - (:fresh_days || ' days')::interval" in sql
+    assert "WHERE is_active = true" in sql
+    assert params == {"fresh_days": LISTINGS_FRESH_DAYS}
+
+
+# ── GET /trade-in/cache-stats ────────────────────────────────────────────────
+
+
+def _cache_stats_row() -> dict[str, Any]:
+    return {
+        "geocode_cache": 1,
+        "geocode_cache_fresh": 1,
+        "listings_active": 37_900,
+        "listings_active_stale": 20_935,
+        "listings_last_scraped": None,
+        "deals": 0,
+        "cad_buildings": 0,
+        "house_metadata": 0,
+        "estimates_total": 0,
+        "avg_median_price": None,
+        "repeat_address_pct": None,
+    }
+
+
+def test_cache_stats_splits_active_and_stale() -> None:
+    from app.services.estimator import LISTINGS_FRESH_DAYS
+
+    db = _FakeDB([_FakeResult(one=_cache_stats_row())])
+    client = _make_client(db, "app.api.v1.trade_in", "/api/v1/trade-in")
+    body = client.get("/api/v1/trade-in/cache-stats").json()
+    assert body["listings_active"] == 37_900
+    assert body["listings_active_stale"] == 20_935
+    assert body["listings_stale_days"] == LISTINGS_FRESH_DAYS
+
+
+def test_cache_stats_counts_stale_by_last_seen_within_active_only() -> None:
+    from app.services.estimator import LISTINGS_FRESH_DAYS
+
+    db = _FakeDB([_FakeResult(one=_cache_stats_row())])
+    client = _make_client(db, "app.api.v1.trade_in", "/api/v1/trade-in")
+    client.get("/api/v1/trade-in/cache-stats")
+    clause, params = db.executed[0]
+    sql = str(clause)
+    assert "last_seen_at <= NOW() - (:fresh_days || ' days')::interval" in sql
+    assert "listings_active_stale" in sql
+    assert params == {"fresh_days": LISTINGS_FRESH_DAYS}
+
+
+def test_cache_stats_empty_db_returns_empty_dict_not_a_fake_threshold() -> None:
+    """Нет строки — отдаём {}, а не {listings_stale_days: N} с пустыми счётчиками
+    (иначе UI отрисует подпись к несуществующему числу)."""
+    db = _FakeDB([_FakeResult(one=None)])
+    client = _make_client(db, "app.api.v1.trade_in", "/api/v1/trade-in")
+    assert client.get("/api/v1/trade-in/cache-stats").json() == {}
--- a/tradein-mvp/backend/tests/test_migration_211_sales_vs_listings_segment_guard.py
+++ b/tradein-mvp/backend/tests/test_migration_211_sales_vs_listings_segment_guard.py
@ -0,0 +1,207 @@
+"""Static guards для миграции 211 (сегментный гард #1186 в street_sales_vs_listings(), #2660).
+
+Прод применяет data/sql построчно строго (ON_ERROR_STOP), полный DB-прогон требует живой
+БД — здесь фиксируем структурные инварианты:
+  - транзакционность + идемпотентность (CREATE OR REPLACE, без DROP/TRUNCATE);
+  - гард #1186 стоит именно в window_listings (listings-сторона), а не в window_deals
+    (у deals нет listing_segment) и не в paired;
+  - `is_active` НЕ появился — его отсутствие осознанно (снятые объявления и есть
+    материал пейринга), см. шапку 205/211;
+  - сигнатура НЕ изменилась относительно 205 — иначе CREATE OR REPLACE создаст ВТОРУЮ
+    перегрузку вместо замены (грабли #2627), и caller получит старое тело;
+  - city-предикаты #2583 H4 (обе стороны JOIN) на месте — не потеряны при переносе тела;
+  - RETURNS TABLE не переименована — endpoint читает колонки по имени.
+"""
+
+from __future__ import annotations
+
+import re
+from pathlib import Path
+
+_SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql"
+_MIGRATION_205 = _SQL_DIR / "205_sales_vs_listings_city_filter.sql"
+_MIGRATION_211 = _SQL_DIR / "211_sales_vs_listings_segment_guard.sql"
+
+_SIGNATURE_TYPES = ("text", "numeric", "integer", "integer", "numeric", "integer", "text")
+_SEGMENT_GUARD = "(l.listing_segment is null or l.listing_segment = 'vtorichka')"
+
+
+def _sql(path: Path = _MIGRATION_211) -> str:
+    return path.read_text(encoding="utf-8")
+
+
+def _executable_sql(path: Path = _MIGRATION_211) -> str:
+    """SQL без построчных `--`-комментариев — только исполняемый код."""
+    lines = []
+    for raw in _sql(path).splitlines():
+        code = raw.split("--", 1)[0]
+        if code.strip():
+            lines.append(code)
+    return "\n".join(lines)
+
+
+def _flat(text: str) -> str:
+    return re.sub(r"\s+", " ", text).strip().lower()
+
+
+def _function_body(path: Path = _MIGRATION_211) -> str:
+    """Тело функции между `AS $$` и `$$;`, без `--`-комментариев, нормализованное.
+
+    Отдельно от _executable_sql(): COMMENT ON FUNCTION — тоже исполняемый SQL, и его
+    текст описывает фильтры словами (в т.ч. «is_active намеренно не фильтруется»).
+    Проверки «такого предиката в запросе нет» обязаны смотреть только на тело.
+    """
+    sql = _sql(path)
+    start = sql.index("AS $$")
+    end = sql.index("$$;", start)
+    body = "\n".join(
+        line.split("--", 1)[0] for line in sql[start:end].splitlines() if line.split("--", 1)[0]
+    )
+    return _flat(body)
+
+
+def _function_signature(path: Path) -> str:
+    """Блок `CREATE OR REPLACE FUNCTION ...(...)` до RETURNS TABLE, нормализованный."""
+    sql = _sql(path)
+    start = sql.index("CREATE OR REPLACE FUNCTION street_sales_vs_listings(")
+    end = sql.index("RETURNS TABLE", start)
+    return _flat(sql[start:end])
+
+
+def test_migration_211_exists() -> None:
+    assert _MIGRATION_211.exists(), f"missing migration: {_MIGRATION_211}"
+
+
+def test_migration_211_is_transactional() -> None:
+    sql = _sql()
+    assert "BEGIN;" in sql
+    assert "COMMIT;" in sql
+
+
+def test_migration_211_no_destructive_ddl() -> None:
+    """Только CREATE OR REPLACE FUNCTION + COMMENT — таблицы не трогаем."""
+    flat = _flat(_executable_sql())
+    assert "drop table" not in flat
+    assert "alter table" not in flat
+    assert "truncate" not in flat
+    assert "delete from" not in flat
+
+
+def test_migration_211_segment_guard_present_in_window_listings() -> None:
+    """Гард #1186 — канонический предикат, НЕ `= 'vtorichka'` без NULL-ветки
+    (NULL = legacy вторичка до м.011; прод: 1 557 таких кандидатов)."""
+    flat = _flat(_executable_sql())
+    assert _SEGMENT_GUARD in flat
+    # Гард стоит в listings-CTE, а не где-то ниже по тексту (в paired/финальном SELECT).
+    window_listings_pos = flat.index("window_listings as")
+    paired_pos = flat.index("paired as")
+    guard_pos = flat.index(_SEGMENT_GUARD)
+    assert window_listings_pos < guard_pos < paired_pos
+
+
+def test_migration_211_does_not_add_is_active_filter() -> None:
+    """Отсутствие is_active — ОСОЗНАННОЕ (#2660): функция намеренно смотрит и снятые
+    объявления, иначе к сделке нечего подставить (объявление снимают ПОСЛЕ продажи)."""
+    assert "is_active" not in _function_body()
+
+
+def test_migration_211_does_not_add_freshness_filter() -> None:
+    """Свежесть здесь тоже не при чём: пейринг привязан к дате СДЕЛКИ (window_days),
+    а не к «сейчас». Регресс-гард против копипасты предиката из location_index."""
+    body = _function_body()
+    assert "l.scraped_at > now()" not in body
+    assert "last_seen_at" not in body
+
+
+def test_migration_211_signature_identical_to_205_no_new_overload() -> None:
+    """#2627: CREATE OR REPLACE с ИЗМЕНЁННЫМ списком типов создаёт ВТОРУЮ перегрузку,
+    а не замену — caller продолжит попадать в старое тело. Сигнатура 211 обязана
+    совпадать с 205 побайтово (после нормализации пробелов)."""
+    assert _function_signature(_MIGRATION_211) == _function_signature(_MIGRATION_205)
+
+
+def test_migration_211_does_not_drop_current_signature() -> None:
+    """DROP текущей 7-арг сигнатуры не нужен (замена и так in-place) и вреден —
+    между DROP и CREATE caller получил бы `function does not exist`."""
+    flat = _flat(_executable_sql())
+    new_sig = "street_sales_vs_listings(" + ", ".join(_SIGNATURE_TYPES) + ")"
+    assert f"drop function if exists {new_sig}" not in flat
+    assert "drop function" not in flat
+
+
+def test_migration_211_exactly_one_function_definition() -> None:
+    """Ровно одно CREATE OR REPLACE — иначе последнее молча перетрёт предыдущее."""
+    flat = _flat(_executable_sql())
+    assert flat.count("create or replace function street_sales_vs_listings") == 1
+
+
+def test_migration_211_comment_targets_the_7arg_signature() -> None:
+    """COMMENT ON FUNCTION со списком типов — иначе комментарий сядет на
+    несуществующий оверлоад / упадёт на ambiguity."""
+    flat = _flat(_executable_sql())
+    new_sig = "street_sales_vs_listings(" + ", ".join(_SIGNATURE_TYPES) + ")"
+    assert f"comment on function {new_sig}" in flat
+
+
+def test_migration_211_keeps_city_predicates_from_205() -> None:
+    """Тело переносится целиком — city-скоуп #2583 H4 (обе стороны JOIN) не потерян."""
+    flat = _flat(_executable_sql())
+    assert "(p_target_city is null or lower(d.city) = lower(p_target_city))" in flat
+    assert (
+        "(p_target_city is null or l.city is null or lower(l.city) = lower(p_target_city))" in flat
+    )
+
+
+def test_migration_211_return_table_shape_unchanged() -> None:
+    """RETURNS TABLE читается endpoint'ом по имени через .mappings() — любое
+    переименование/удаление сломало бы response mapping без явного сигнала."""
+    sql = _sql()
+    for col in (
+        "deal_id",
+        "deal_date",
+        "deal_price_rub",
+        "deal_price_per_m2",
+        "deal_area_m2",
+        "deal_rooms",
+        "deal_floor",
+        "deal_address",
+        "listing_id",
+        "listing_source",
+        "listing_source_url",
+        "listing_date",
+        "listing_price_rub",
+        "listing_price_per_m2",
+        "listing_area_m2",
+        "days_listing_to_deal",
+        "discount_pct",
+    ):
+        assert col in sql, f"missing column in RETURNS TABLE: {col}"
+
+
+def test_migration_211_no_psycopg_cast_trap() -> None:
+    """Регресс-гард против copy-paste `:param::type` (в чистом .sql bind-параметров нет,
+    но файл переносится в Python-тесты/скрипты)."""
+    assert not re.search(r":\w+::", _sql())
+
+
+def test_migration_211_body_matches_205_except_the_guard() -> None:
+    """Тело — байт-в-байт копия 205 плюс ровно один новый предикат. Защищает от
+    случайной потери логики при ручном переносе большого SQL-блока."""
+
+    def body(path: Path) -> list[str]:
+        sql = _sql(path)
+        start = sql.index("AS $$")
+        end = sql.index("$$;", start)
+        # Только исполняемые строки, без комментариев — их мы правим осознанно.
+        return [
+            _flat(line.split("--", 1)[0])
+            for line in sql[start:end].splitlines()
+            if line.split("--", 1)[0].strip()
+        ]
+
+    old_body = body(_MIGRATION_205)
+    new_body = body(_MIGRATION_211)
+    added = [line for line in new_body if line not in old_body]
+    removed = [line for line in old_body if line not in new_body]
+    assert added == ["and " + _SEGMENT_GUARD], added
+    assert removed == [], removed
--- a/tradein-mvp/backend/tests/test_scraper_admin_apis.py
+++ b/tradein-mvp/backend/tests/test_scraper_admin_apis.py
@ -357,10 +357,12 @@ def _make_dq_db_mock() -> MagicMock:
    """Мок db-сессии для GET /scraper/data-quality."""
    db = MagicMock()

-    # listings rows: avito=1000 active, cian=500 active
+    # listings rows: avito=1000 active (все свежие), cian=500 active / 300 протухших
+    # (#2660: stale_count идёт рядом с active_count — «активно» ≠ «живо»).
    avito_row = {
        "source": "avito",
        "active_count": 1000,
+        "stale_count": 0,
        "f_description": 900,
        "f_photo_urls": 980,
        "f_address": 1000,
@ -375,6 +377,7 @@ def _make_dq_db_mock() -> MagicMock:
    cian_row = {
        "source": "cian",
        "active_count": 500,
+        "stale_count": 300,
        "f_description": 450,
        "f_photo_urls": 490,
        "f_address": 500,
@ -435,6 +438,7 @@ def test_data_quality_shape(client: TestClient) -> None:

    avito = sources["avito"]
    assert avito["active_count"] == 1000
+    assert avito["stale_count"] == 0  # #2660
    fields = avito["fields"]
    assert 0.0 <= fields["description"] <= 100.0
    assert fields["description"] == pytest.approx(90.0, abs=0.1)
@ -443,6 +447,7 @@ def test_data_quality_shape(client: TestClient) -> None:

    cian = sources["cian"]
    assert cian["active_count"] == 500
+    assert cian["stale_count"] == 300  # #2660: 60% «активных» на деле не виделись
    assert cian["fields"]["ceiling_height"] == pytest.approx(70.0, abs=0.1)
    assert cian["fields"]["ceiling_height_m"] == pytest.approx(0.0, abs=0.1)