fix(tradein/estimate): фильтр свежести в якоре дома и знаменателе коэффициента выкупа (#2656)

Цена местами строилась на объявлениях, которых никто не видел месяц. Главный
радиусный путь эстиматора нёс `scraped_at > NOW() - LISTINGS_FRESH_DAYS дней`
(_COMMON_WHERE), а четыре денежные выборки — нет:

- `_fetch_anchor_comps` Tier A и Tier C: якорь ЗАМЕЩАЕТ headline
  (median_ppm2/median_price/n_analogs), т.е. правит цену напрямую;
- `asking_to_sold_ratio` ask_side и ask_global: знаменатель коэффициента,
  на который умножается expected_sold_price.

`is_active` свежесть не заменяет — он означает разное у разных источников
(TTL деактивации 30 дней, NULL-сегмент не деактивируется никогда), а
`scraped_at` одно и то же. Прод: 21 132 из 37 497 активных строк протухли по
14-дневной мерке самого эстиматора и были полностью годны для якоря.

Окно вынесено в app.core.config.LISTINGS_FRESH_DAYS (одно место на всех):
держать его в estimator.py нельзя — тот сам импортирует area_bucket из
app.tasks.asking_to_sold_ratio, обратный импорт дал бы цикл.

Второй половиной — залипший anchor_tier: он оставался "C"/"A", когда якорь не
был построен, и молча глушил IMV-blend, quarter-index (#764 Guard-1a),
radius-floor и corridor-clamp-exempt Tier A. Теперь сбрасывается явно, у
источника, для всех трёх причин (None из _compute_same_building_anchor, гейт
Tier C #1795, low-conf гейт #audit-1).

Обе половины одним PR намеренно: порознь они дадут два заметных скачка цены
вместо одного меньшего (якорная половина −3.63%, знаменатель +1.25%,
вместе −2.42% от суммы выкупа на 1040 реальных оценках).

Тесты: tests/test_freshness_filter_2656.py — предикат во всех четырёх местах,
единственность константы, бинд :fresh_days, протухший комп не в пуле якоря,
сброс anchor_tier и разглушённый IMV-blend. Все 7 краснеют без фикса
(проверено git stash).

tradein-mvp/backend/app/core/config.py

@@ -30,6 +30,20 @@ _AUTH_DB_DEFAULT_NAME = "auth"
 # Роль приложения из data/sql/auth/002_auth_app_role.sql (least privilege).
 _AUTH_DB_DEFAULT_USER = "auth_app"
 
+# ── Окно свежести объявлений (#2656) ──────────────────────────────────────────
+# ЕДИНСТВЕННОЕ место, где живёт это число. Читают: estimator (_COMMON_WHERE
+# радиусного пути, inline-копия Tier W, оба SQL якоря дома) и ночной пересчёт
+# asking_to_sold_ratios (знаменатель коэффициента выкупа). Лежит здесь, а не в
+# estimator.py, потому что estimator сам импортирует area_bucket из
+# app.tasks.asking_to_sold_ratio — обратный импорт дал бы цикл.
+#
+# ЗАЧЕМ фильтр вообще: `is_active` означает РАЗНОЕ для разных источников (TTL
+# деактивации 30 дней у cian/yandex-вторички, NULL-сегмент не деактивируется
+# никогда), а `scraped_at` — одно и то же. Ослабление этого окна или подмена
+# `scraped_at` на `last_seen_at` впускает в ценовые выборки объявления, которых
+# никто не видел месяц (прод 2026-08: 21 132 из 37 497 активных строк).
+LISTINGS_FRESH_DAYS = 14  # объявления не старше 14 дней
+
 
 class Settings(BaseSettings):
     model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", extra="ignore")

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

@@ -53,7 +53,7 @@ from scraper_kit.providers.yandex.valuation import (
 from sqlalchemy import text
 from sqlalchemy.orm import Session
 
-from app.core.config import Settings, settings
+from app.core.config import LISTINGS_FRESH_DAYS, Settings, settings
 from app.schemas.trade_in import (
     AggregatedEstimate,
     AnalogLot,
@@ -90,7 +90,9 @@ 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 дней
+# LISTINGS_FRESH_DAYS (окно свежести объявлений) живёт в app.core.config — #2656:
+# его читает и ночной пересчёт asking_to_sold_ratios, а импортировать estimator
+# оттуда нельзя (estimator сам тянет area_bucket из этой задачи → цикл импорта).
 DEALS_PERIOD_MONTHS = 12  # сделки за последний год
 
 # #oblast-D (non-EKB deals-headline-fallback): минимум ДКП-сделок, чтобы
@@ -1857,7 +1859,9 @@ def _fetch_anchor_comps(
       ненадёжны.)
     Tier D — фолбэк: None tier (caller остаётся на радиусном median-пути).
 
-    Excludes lots без price_per_m2. is_active=true. Best-effort: ([], None) на ошибке.
+    Excludes lots без price_per_m2. is_active=true И scraped_at свежее
+    LISTINGS_FRESH_DAYS (#2656 — то же окно, что у радиусного пути; см.
+    комментарий в SQL Tier A). Best-effort: ([], None) на ошибке.
 
     Returns (comps, tier) где tier ∈ {'A','C', None}. comps — list dict с
     ключами price_per_m2 (int>0), area_m2 (float|None), rooms (int|None),
@@ -1883,6 +1887,15 @@ def _fetch_anchor_comps(
                                listing_segment, source_id
                         FROM listings
                         WHERE is_active = true
+                          -- #2656: то же окно свежести, что несёт _COMMON_WHERE
+                          -- радиусного пути. Без него якорь строился на
+                          -- объявлениях, которых никто не видел месяц (is_active
+                          -- живёт по TTL деактивации 30д, у NULL-сегмента — вообще
+                          -- бессрочно), и они ЗАМЕЩАЛИ headline (median_ppm2/
+                          -- median_price/n_analogs ниже), т.е. двигали деньги
+                          -- напрямую. Прод 2026-08: 31 дом из 952 получал якорь
+                          -- Tier A исключительно за счёт протухших строк.
+                          AND scraped_at > NOW() - (:fresh_days || ' days')::interval
                           AND price_per_m2 > 0
                           AND lower(translate(address, 'ёЁ', 'ее')) LIKE :street_like
                           AND lower(translate(address, 'ёЁ', 'ее')) ~ :house_re
@@ -1900,6 +1913,7 @@ def _fetch_anchor_comps(
                         "lon": lon,
                         "lat": lat,
                         "radius": ANCHOR_TIER_A_RADIUS_M,
+                        "fresh_days": LISTINGS_FRESH_DAYS,
                     },
                 )
                 .mappings()
@@ -1975,6 +1989,9 @@ def _fetch_anchor_comps(
                                days_on_market, photo_urls, lat, lon
                         FROM listings
                         WHERE is_active = true
+                          -- #2656: окно свежести, симметрично Tier A выше и
+                          -- _COMMON_WHERE радиусного пути (см. комментарий там).
+                          AND scraped_at > NOW() - (:fresh_days || ' days')::interval
                           AND price_per_m2 > 0
                           AND rooms = CAST(:rooms AS integer)
                           AND area_m2 BETWEEN :area_min AND :area_max
@@ -1994,6 +2011,7 @@ def _fetch_anchor_comps(
                         "area_max": area * 1.25,
                         "lon": lon,
                         "lat": lat,
+                        "fresh_days": LISTINGS_FRESH_DAYS,
                     },
                 )
                 .mappings()
@@ -2682,7 +2700,18 @@ def _price_from_inputs(
                 gate_thin,
             )
             anchor = None
-            anchor_tier = None
+
+    # #2656: якорь не построен — сбрасываем tier-флаг ЯВНО. Причин три:
+    # _compute_same_building_anchor вернула None (комплов меньше min_comps, в т.ч.
+    # после MAD-клипа), гейт Tier C #1795 или low-conf гейт #audit-1 выше. Раньше
+    # сброс делал только последний из трёх, и в остальных случаях anchor_tier
+    # залипал равным anchor_tier_fetched ("C"/"A") при радиусном headline. Флаг
+    # читают IMV-blend (`anchor_tier is None`, ниже), quarter-index #764 Guard-1a,
+    # radius-floor от ДКП-коридора, corridor-clamp (Tier A exempt) и api_analog_tier
+    # — залипший флаг молча глушил их все, будто headline построил якорь. Замер
+    # #2656: 24 оценки без якоря до и после правки всё равно меняли цену из-за него.
+    if anchor is None:
+        anchor_tier = None
 
     if anchor is not None:
         # #694: якорь мутирует headline — UI-аналоги должны отражать ЭТИ комплы.
@@ -3172,6 +3201,10 @@ def _price_from_inputs(
     # and blocked this fallback even with a large, valid ДКП corridor
     # available (observed: 677 deals for one fixture case). `anchor is None`
     # is the ground truth of whether the anchor actually produced a headline.
+    # #2656 update: тот залипший флаг теперь сбрасывается у источника (см. `if
+    # anchor is None: anchor_tier = None` в anchor-блоке выше), т.е. два условия
+    # стали эквивалентны. Гард оставлен на `anchor is None` НАМЕРЕННО — это
+    # по-прежнему прямая проверка факта «якорь дал headline», а не производный флаг.
     if (
         median_ppm2 <= 0
         and anchor is None

tradein-mvp/backend/app/tasks/asking_to_sold_ratio.py

@@ -26,7 +26,14 @@ ask_side / per_bucket + deal_global / ask_global / global_row: трейлинг-
 n_listings>=30 для per_rooms, global -1 строка всегда). ON CONFLICT убран — DELETE идёт первым,
 конфликтов нет (повторный прогон в одной tx невозможен, refresh = re-seed по семантике).
 
-#2620 — ОДНО ПРЕДНАМЕРЕННОЕ РАСХОЖДЕНИЕ с 080: deal_side бакетится по
+#2656 — ВТОРОЕ ПРЕДНАМЕРЕННОЕ РАСХОЖДЕНИЕ с 080 (первое — #2620 ниже): ask_side/ask_global
+несут фильтр свежести `scraped_at > NOW() - LISTINGS_FRESH_DAYS дней` — тот же, что эстиматор
+применяет к ЧИСЛИТЕЛЮ (_COMMON_WHERE). Без него знаменатель считался по бессрочной популяции
+объявлений, а числитель — по 14-дневной, т.е. коэффициент калибровался на одном рынке, а
+применялся к другому. Замер на проде (2026-08, #2656): глобальный коэффициент −0.49%, бакеты
+44-62 +2.43% и 62-85 +4.58%; NULL-сегмент в знаменателе схлопывается с 674 строк до 20.
+
+#2620 — ПЕРВОЕ ПРЕДНАМЕРЕННОЕ РАСХОЖДЕНИЕ с 080: deal_side бакетится по
 LEAST(GREATEST(rooms,0),4), а ask_side — по _AREA_ROOMS_BUCKET_SQL (площадь, та же формула,
 что deals.rooms получает при импорте). Причина — deals.rooms НЕ настоящая комнатность
 (Росреестр её не отдаёт), это синтетика из площади; сравнивать её с РЕАЛЬНЫМИ комнатами
@@ -43,7 +50,7 @@ import logging
 from sqlalchemy import text
 from sqlalchemy.orm import Session
 
-from app.core.config import settings
+from app.core.config import LISTINGS_FRESH_DAYS, settings
 from app.services import scrape_runs as runs_mod
 
 # Нижняя граница ppm² — отсекает нежилые/технические сделки; не меняется.
@@ -119,7 +126,8 @@ _DELETE_SQL = text(
 #     бакет LEAST(GREATEST(rooms,0),4) (rooms уже синтетика-из-площади при импорте, см. #2620
 #     комментарий у _AREA_ROOMS_BUCKET_SQL выше).
 #   ask_median  = percentile_cont(0.5) по listings.price_per_m2
-#     (is_active, та же ppm²-полоса [_PPM2_MIN, asking_ratio_ppm2_max], тот же город что
+#     (is_active + свежесть scraped_at ≤ LISTINGS_FRESH_DAYS (#2656), та же ppm²-полоса
+#     [_PPM2_MIN, asking_ratio_ppm2_max], тот же город что
 #     SOLD-сторона — city IS NULL OR city ILIKE :asking_city, #2583 H2). Бакет —
 #     _AREA_ROOMS_BUCKET_SQL (площадь, #2620), НЕ listings.rooms — см. комментарий там.
 #   per_rooms строки — только при n_deals>=30 AND n_listings>=30 AND ask>0 AND sold>0.
@@ -156,6 +164,13 @@ _REDERIVE_SQL = text(
             COUNT(*)                                                      AS n_listings
         FROM listings
         WHERE is_active
+          -- #2656: окно свежести — то же самое, что эстиматор применяет к ЧИСЛИТЕЛЮ
+          -- (_COMMON_WHERE, LISTINGS_FRESH_DAYS). Без него числитель оценки считался
+          -- по 14-дневной популяции, а знаменатель коэффициента — по бессрочной:
+          -- калибровка и применение по разным рынкам. `is_active` для этого не годится
+          -- — он означает разное у разных источников (TTL деактивации 30д, NULL-сегмент
+          -- не деактивируется никогда: 97.4% таких строк протухшие и при этом дорогие).
+          AND scraped_at > NOW() - (:fresh_days || ' days')::interval
           AND rooms IS NOT NULL
           -- #2620 hardening: area_m2 IS NULL falls into the CASE ELSE branch (bucket 4)
           -- of _AREA_ROOMS_BUCKET_SQL — a latent "everything unmeasured looks like a big
@@ -216,6 +231,9 @@ _REDERIVE_SQL = text(
             COUNT(*)                                                  AS n_listings
         FROM listings
         WHERE is_active
+          -- #2656: то же окно свежести, что и в ask_side выше (см. комментарий там)
+          -- — global-строка должна считаться по той же популяции, что per-bucket.
+          AND scraped_at > NOW() - (:fresh_days || ' days')::interval
           AND rooms IS NOT NULL
           -- #2620 hardening: same area_m2 IS NOT NULL as ask_side — keeps the global-row
           -- population consistent with the per-bucket rows it's a fallback for.
@@ -313,6 +331,7 @@ def recompute_asking_to_sold_ratios(db: Session, run_id: int) -> dict[str, int]:
                 "ppm2_min": _PPM2_MIN,
                 "ppm2_max": settings.asking_ratio_ppm2_max,
                 "asking_city": _ASKING_CITY_PATTERN,
+                "fresh_days": LISTINGS_FRESH_DAYS,
             },
         )
 

tradein-mvp/backend/scripts/backtest_estimator.py

@@ -1462,8 +1462,15 @@ def load_fixture(path: str) -> dict[str, Any]:
     return json.loads(Path(path).read_text(encoding="utf-8"))
 
 
+_UNRECORDED = object()  # sentinel: «на лишний вызов ответа нет» → RuntimeError
+
+
 def _make_call_stub(
-    calls: list[Any], *, label: str, coerce: Callable[[Any], Any]
+    calls: list[Any],
+    *,
+    label: str,
+    coerce: Callable[[Any], Any],
+    on_exhausted: Any = _UNRECORDED,
 ) -> Callable[[Any], Any]:
     """Build an ORDER-based (FIFO) replay stub from recorded ``[arg, return]`` pairs.
 
@@ -1478,6 +1485,17 @@ def _make_call_stub(
     stays correct if a call site ever loops. Calling the stub MORE times than
     recorded raises RuntimeError — control flow diverged from capture.
 
+    ``on_exhausted`` (#2656) ослабляет ЭТО последнее правило для конкретной
+    callable: значение возвращается вместо RuntimeError, когда фикстура записала
+    меньше вызовов, чем сделал реплей. Нужно, когда правка РАЗБЛОКИРОВАЛА путь,
+    которого при захвате фикстуры не было (сброс залипшего ``anchor_tier`` открыл
+    quarter-index-гейт `Guard-1a` на 19 из 277 сделок) — у прод-фикстуры на такие
+    вызовы ответа нет и взять его негде, пока фикстуру не перезахватят с прода.
+    Использовать ТОЛЬКО для lookup'ов, у которых «промах» — валидное состояние
+    (quarter-index: None/{} = индекса нет → блок no-op). ``ratio_resolver``
+    остаётся строгим НАМЕРЕННО: лишний вызов там означал бы, что реплей взял
+    другой коэффициент выкупа, т.е. молча другие деньги.
+
     ``coerce`` maps each JSON-plain recorded return back to the live callable's
     return type (tuple / dict) so unpacking at the call site behaves identically.
     """
@@ -1487,10 +1505,13 @@ def _make_call_stub(
     def _stub(_arg: Any) -> Any:
         nonlocal idx
         if idx >= len(returns):
-            raise RuntimeError(
-                f"{label}: replay made call #{idx + 1} but fixture recorded only "
-                f"{len(returns)} — control flow diverged from capture"
-            )
+            if on_exhausted is _UNRECORDED:
+                raise RuntimeError(
+                    f"{label}: replay made call #{idx + 1} but fixture recorded only "
+                    f"{len(returns)} — control flow diverged from capture"
+                )
+            idx += 1
+            return on_exhausted
         ret = returns[idx]
         idx += 1
         return ret
@@ -1530,11 +1551,23 @@ def replay_fixture(fixture: dict[str, Any]) -> dict[str, Any]:
         kw["ratio_resolver"] = _make_call_stub(
             rec.get("ratio_calls") or [], label="ratio_resolver", coerce=_coerce_ratio_return
         )
+        # #2656: quarter-index-lookup'ы отвечают «промах» (None / {}) на вызовы, которых
+        # в фикстуре нет — см. _make_call_stub. Сброс залипшего anchor_tier открыл гейт
+        # Guard-1a на 19 из 277 сделок, а прод-фикстура захвачена ДО правки. Следствие,
+        # которое надо знать при чтении метрик: для этих 19 сделок квартальный индекс в
+        # реплее НЕ применяется вовсе (в проде — применился бы), т.е. гейт занижает
+        # эффект правки. Уйдёт при следующем перезахвате фикстуры с прода.
         kw["quarter_index_lookup"] = _make_call_stub(
-            rec.get("qi_calls") or [], label="quarter_index_lookup", coerce=_coerce_qi_return
+            rec.get("qi_calls") or [],
+            label="quarter_index_lookup",
+            coerce=_coerce_qi_return,
+            on_exhausted=None,
         )
         kw["quarter_indexes_lookup"] = _make_call_stub(
-            rec.get("qis_calls") or [], label="quarter_indexes_lookup", coerce=_coerce_qis_return
+            rec.get("qis_calls") or [],
+            label="quarter_indexes_lookup",
+            coerce=_coerce_qis_return,
+            on_exhausted={},
         )
 
         pr = m._price_from_inputs(**kw)

tradein-mvp/backend/tests/test_asking_to_sold_ratio.py

@@ -208,6 +208,11 @@ def test_migration_080_derivation_is_subset_of_refresh_sql() -> None:
     #2620 hardening: the refresh also adds `AND area_m2 IS NOT NULL` to ask_side/ask_global
     (NULL area_m2 would fall into the CASE ELSE branch = bucket 4 — a latent trap). Absent
     in the 080 seed; dropped here the same way as the other guards above.
+
+    #2656: the refresh now also adds the freshness predicate (`scraped_at > NOW() -
+    LISTINGS_FRESH_DAYS days`) to each ask_* CTE — the same window the estimator applies to
+    the NUMERATOR (_COMMON_WHERE). Absent in the 080 seed; normalised away here, asserted
+    on its own in tests/test_freshness_filter_2656.py.
     """
     seed_sql = _MIGRATION_080.read_text("utf-8")
     # Extract the WITH … (up to the ON CONFLICT) from the seed.
@@ -269,11 +274,17 @@ def test_migration_080_derivation_is_subset_of_refresh_sql() -> None:
         """
         return re.sub(r"AND\s+area_m2\s+IS\s+NOT\s+NULL", "", s)
 
+    def _drop_freshness_guard(s: str) -> str:
+        """Remove the #2656 freshness predicate (absent in the 080 seed)."""
+        return re.sub(r"AND\s+scraped_at\s*>\s*NOW\(\)\s*-\s*\(:fresh_days[^\n]*?interval", "", s)
+
     def _norm(s: str) -> str:
         return _strip_sql(
             _normalise_ppm2(
                 _normalise_bucket_expr(
-                    _drop_area_not_null_guard(_drop_city_guard(_drop_segment_guard(s)))
+                    _drop_freshness_guard(
+                        _drop_area_not_null_guard(_drop_city_guard(_drop_segment_guard(s)))
+                    )
                 )
             )
         )

tradein-mvp/backend/tests/test_estimator_price_spine.py

@@ -200,8 +200,10 @@ def test_same_building_anchor_tier_a_mutates_headline() -> None:
 def test_tier_c_corridor_gate_suppresses_anchor() -> None:
     """Tier C anchor ppm2 >> corridor_high × mult → anchor suppressed.
 
-    anchor_tier remains "C" in the result (gate sets anchor=None but doesn't
-    reset anchor_tier); headline stays at the radius median.
+    #2656: anchor_tier теперь СБРАСЫВАЕТСЯ в None (раньше гейт ставил anchor=None,
+    но оставлял залипший "C" — и этот флаг молча глушил IMV-blend/quarter-index/
+    radius-floor, будто headline построил якорь). Headline как и раньше остаётся
+    радиусной медианой.
     """
     # 5 comps at 300k ppm2; corridor_high=150k; gate threshold=150k×1.5=225k.
     # 300k > 225k → suppressed.
@@ -215,8 +217,8 @@ def test_tier_c_corridor_gate_suppresses_anchor() -> None:
         ratio=None,
     )
 
-    # Tier C gate sets anchor=None but leaves anchor_tier="C".
-    assert pr.anchor_tier == "C"
+    # #2656: гейт ставит anchor=None → tier-флаг сбрасывается вместе с ним.
+    assert pr.anchor_tier is None
     # Headline was NOT mutated by the suppressed anchor — stays at radius median.
     assert pr.median_price == radius_median_price
     # anchor_comps_used stays empty (anchor didn't fire).

tradein-mvp/backend/tests/test_freshness_filter_2656.py

@@ -0,0 +1,305 @@
+"""#2656 — фильтр свежести объявлений там, где он двигает деньги.
+
+Дыра (найдена аудитом 2026-08, код читан на origin/main): главный радиусный путь
+эстиматора нёс `scraped_at > NOW() - LISTINGS_FRESH_DAYS дней` (_COMMON_WHERE), а
+ЧЕТЫРЕ другие выборки, которые тоже двигают цену, — нет:
+
+  1. `_fetch_anchor_comps` Tier A (same building) — якорь ЗАМЕЩАЕТ headline
+     (median_ppm2/median_price/n_analogs), т.е. правит цену напрямую;
+  2. `_fetch_anchor_comps` Tier C (микро-радиус ≤500 м) — то же замещение;
+  3. `asking_to_sold_ratio.ask_side`  — знаменатель коэффициента выкупа;
+  4. `asking_to_sold_ratio.ask_global` — он же, global-фолбэк.
+
+`is_active` не заменяет свежесть: он означает РАЗНОЕ у разных источников (TTL
+деактивации 30 дней у cian/yandex-вторички, NULL-сегмент не деактивируется никогда),
+а `scraped_at` — одно и то же. Прод 2026-08: 21 132 из 37 497 «активных» строк
+протухли по 14-дневной мерке самого эстиматора и при этом были полностью годны
+для якоря.
+
+Пятый (побочный) дефект, вскрытый тем же замером: `anchor_tier` оставался равным
+"C", когда якорь НЕ был построен, и этот залипший флаг молча глушил IMV-blend,
+quarter-index и radius-floor — см. вторую половину файла.
+
+Эти тесты — защита от снятия предиката: именно так дыра и появилась (в главном
+пути фильтр есть, в якоре и знаменателе забыли). Проверка на красноту сделана
+`git stash` на файлах реализации — все 7 тестов падают без фикса.
+"""
+
+from __future__ import annotations
+
+import inspect
+import os
+import re
+from datetime import UTC, datetime, timedelta
+from typing import Any
+from unittest.mock import MagicMock
+
+# Settings требует DATABASE_URL при инициализации (fail-fast, C-3).
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from app.core.config import LISTINGS_FRESH_DAYS
+from app.services import estimator as est_mod
+from app.services.estimator import _fetch_anchor_comps, _price_from_inputs
+from app.services.geocoder import GeocodeResult
+from app.tasks import asking_to_sold_ratio as ratio_mod
+
+# Ровно тот предикат, что стоит в _COMMON_WHERE радиусного пути. Сверяем ПОДСТРОКОЙ,
+# а не регуляркой: цель — чтобы во всех местах стоял ОДИН И ТОТ ЖЕ текст, а не пять
+# похожих вариантов, которые потом разъедутся по смыслу.
+_PREDICATE = "AND scraped_at > NOW() - (:fresh_days || ' days')::interval"
+
+_ANCHOR_SRC = inspect.getsource(_fetch_anchor_comps)
+_TIER_C_MARK = "# ── Tier C: micro-radius"
+_TIER_A_SRC = _ANCHOR_SRC[: _ANCHOR_SRC.index(_TIER_C_MARK)]
+_TIER_C_SRC = _ANCHOR_SRC[_ANCHOR_SRC.index(_TIER_C_MARK) :]
+
+_REDERIVE_SQL = str(ratio_mod._REDERIVE_SQL.text)
+_ASK_SIDE_SQL = _REDERIVE_SQL[
+    _REDERIVE_SQL.index("ask_side AS") : _REDERIVE_SQL.index("per_bucket AS")
+]
+_ASK_GLOBAL_SQL = _REDERIVE_SQL[
+    _REDERIVE_SQL.index("ask_global AS") : _REDERIVE_SQL.index("global_row AS")
+]
+
+
+# ── 1. Предикат стоит во всех четырёх местах ─────────────────────────────────
+
+
+def test_freshness_predicate_present_in_all_four_places() -> None:
+    """Все четыре денежные выборки несут ТОТ ЖЕ предикат свежести, что _COMMON_WHERE.
+
+    Falsifiable: на непропатченном коде FALSE во всех четырёх (там только
+    `is_active = true`), TRUE после правки.
+    """
+    places = {
+        "_COMMON_WHERE (эталон, радиусный путь)": est_mod._COMMON_WHERE,
+        "_fetch_anchor_comps Tier A": _TIER_A_SRC,
+        "_fetch_anchor_comps Tier C": _TIER_C_SRC,
+        "asking_to_sold_ratio ask_side": _ASK_SIDE_SQL,
+        "asking_to_sold_ratio ask_global": _ASK_GLOBAL_SQL,
+    }
+    missing = [name for name, sql in places.items() if _PREDICATE not in sql]
+    assert not missing, f"фильтр свежести #2656 снят в: {missing}"
+
+
+def test_freshness_window_lives_in_one_place() -> None:
+    """Окно берётся из ОДНОЙ константы (app.core.config.LISTINGS_FRESH_DAYS), не из литерала.
+
+    Иначе значения разъедутся — ровно так дыра #2656 и появилась. Проверяем и то, что
+    оба модуля импортируют имя из config (а не завели свою копию), и то, что в SQL нет
+    зашитого «14 days».
+    """
+    assert LISTINGS_FRESH_DAYS == 14  # текущее значение; меняется в одном месте
+    for mod in (est_mod, ratio_mod):
+        src = inspect.getsource(mod)
+        assert re.search(
+            r"from app\.core\.config import [^\n]*LISTINGS_FRESH_DAYS", src
+        ), f"{mod.__name__}: LISTINGS_FRESH_DAYS не импортирован из app.core.config"
+        assert not re.search(
+            r"^LISTINGS_FRESH_DAYS\s*=", src, re.M
+        ), f"{mod.__name__}: локальная копия константы — значение разъедется"
+    assert "'14 days'" not in _REDERIVE_SQL
+    assert "'14 days'" not in _ANCHOR_SRC
+
+
+def test_freshness_param_bound_in_both_anchor_tiers() -> None:
+    """Оба тира якоря БИНДЯТ :fresh_days — иначе SQL упадёт на невязанном параметре.
+
+    Отдельный тест от предиката: забыть параметр в dict — ровно та ошибка, которую
+    статическая проверка текста SQL не ловит, а прод ловит 500-й.
+    """
+    for name, src in (("Tier A", _TIER_A_SRC), ("Tier C", _TIER_C_SRC)):
+        assert '"fresh_days": LISTINGS_FRESH_DAYS,' in src, f"{name}: :fresh_days не забинден"
+
+
+# ── 2. Протухший комп не попадает в пул якоря, свежий попадает ───────────────
+
+_ADDRESS = "Екатеринбург, ул. Евгения Савкова, 29"
+_LAT, _LON = 56.838, 60.595
+
+
+def _listing_row(
+    *, source_id: str, days_ago: int, price_per_m2: float = 150_000.0
+) -> dict[str, Any]:
+    """Строка listings в форме .mappings() + служебный scraped_at для заглушки ниже."""
+    return {
+        "price_per_m2": price_per_m2,
+        "area_m2": 50.0,
+        "rooms": 2,
+        "floor": 3,
+        "total_floors": 9,
+        "address": _ADDRESS,
+        "source": "avito",
+        "source_url": f"https://avito.ru/offer/{source_id}",
+        "price_rub": int(price_per_m2 * 50.0),
+        "listing_date": None,
+        "days_on_market": 15,
+        "photo_urls": [],
+        "lat": _LAT,
+        "lon": _LON,
+        "listing_segment": "vtorichka",
+        "source_id": source_id,
+        "scraped_at": datetime.now(tz=UTC) - timedelta(days=days_ago),
+    }
+
+
+def _predicate_aware_db(rows: list[dict[str, Any]]) -> MagicMock:
+    """Session-заглушка, ИСПОЛНЯЮЩАЯ предикат свежести вместо Postgres.
+
+    Настоящей БД в тестах нет, поэтому отсев по `scraped_at` делает сама заглушка —
+    но ТОЛЬКО если SQL реально несёт предикат И :fresh_days забинден. Без фикса
+    предиката в SQL нет → протухшие строки доезжают до пула якоря → тест краснеет.
+    Это эмуляция ровно одного WHERE-условия, не мини-Postgres: остальные предикаты
+    (адрес/гео/сегмент) заданы данными фикстуры так, чтобы они тривиально проходили.
+    """
+    db = MagicMock()
+
+    def _execute(stmt: Any, params: dict[str, Any] | None = None) -> MagicMock:
+        sql = str(stmt)
+        kept = rows
+        if _PREDICATE in sql:
+            cutoff = datetime.now(tz=UTC) - timedelta(days=params["fresh_days"])
+            kept = [r for r in rows if r["scraped_at"] > cutoff]
+        result = MagicMock()
+        result.mappings.return_value.all.return_value = kept
+        return result
+
+    db.execute.side_effect = _execute
+    return db
+
+
+def test_stale_comps_excluded_fresh_kept_in_anchor_pool() -> None:
+    """4 свежих + 3 протухших компа → в якорь идут только 4 свежих.
+
+    Прод-мотивация: протухшие строки в сегменте, из которого строится якорь, в среднем
+    на 7.7% дороже живых (дорогое висит дольше и первым выпадает из выдачи), т.е.
+    фантом в тонком пуле тянет якорь ВВЕРХ. Замер #2656: 31 дом из 952 получал якорь
+    Tier A исключительно за счёт протухших.
+    """
+    rows = [_listing_row(source_id=f"f{i}", days_ago=1) for i in range(4)]
+    rows += [_listing_row(source_id=f"s{i}", days_ago=40, price_per_m2=260_000.0) for i in range(3)]
+
+    comps, tier = _fetch_anchor_comps(
+        _predicate_aware_db(rows),
+        address=_ADDRESS,
+        target_house_id=None,
+        lat=_LAT,
+        lon=_LON,
+        rooms=2,
+        area=50.0,
+    )
+
+    assert tier == "A"
+    assert len(comps) == 4
+    assert all(c["price_per_m2"] == 150_000.0 for c in comps), "протухший комп попал в якорь"
+
+
+def test_thin_pool_loses_anchor_when_only_stale_comps_remain() -> None:
+    """3 свежих + 5 протухших: было ≥ min_comps=4 → якоря больше нет (Tier D фолбэк).
+
+    Это и есть заявленная цена правки: 139 из 1014 реальных оценок теряют якорь и
+    уезжают на радиусный путь / ДКП-фолбэк. Честнее, чем якорь на объявлениях,
+    которых никто не видел месяц.
+    """
+    rows = [_listing_row(source_id=f"f{i}", days_ago=2) for i in range(3)]
+    rows += [_listing_row(source_id=f"s{i}", days_ago=45) for i in range(5)]
+
+    comps, tier = _fetch_anchor_comps(
+        _predicate_aware_db(rows),
+        address=_ADDRESS,
+        target_house_id=None,
+        lat=_LAT,
+        lon=_LON,
+        rooms=2,
+        area=50.0,
+    )
+
+    # Tier A не набрал порог; Tier C на тех же строках — тоже (заглушка отдаёт их же).
+    assert tier is None
+    assert comps == []
+
+
+# ── 3. anchor_tier сбрасывается, когда якорь не построен ─────────────────────
+
+
+def _geo() -> GeocodeResult:
+    return GeocodeResult(
+        lat=_LAT, lon=_LON, full_address="Екатеринбург, ул. Ленина, 5", provider="dadata"
+    )
+
+
+def _price(
+    *,
+    anchor_comps: list[dict[str, Any]],
+    anchor_tier_fetched: str | None,
+    imv_anchor: dict[str, Any] | None = None,
+) -> est_mod.PricingResult:
+    listings = [
+        {"price_per_m2": 200_000.0, "address": f"ул. Ленина, {i + 5}", "source": "avito"}
+        for i in range(6)
+    ]
+    return _price_from_inputs(
+        listings=listings,
+        area_m2=50.0,
+        rooms=2,
+        repair_state=None,
+        floor=5,
+        total_floors=9,
+        target_year=None,
+        analog_tier="W",
+        fallback_used=False,
+        area_widened=False,
+        anchor_comps=anchor_comps,
+        anchor_tier_fetched=anchor_tier_fetched,
+        dkp_raw=None,
+        imv_anchor=imv_anchor,
+        imv_eval=None,
+        yandex_val_present=False,
+        cian_val_present=False,
+        ratio_resolver=lambda _ppm2: (None, None),
+        quarter_index_lookup=lambda _q: None,
+        quarter_indexes_lookup=lambda _qs: {},
+        target_house_cadnum=None,
+        dadata_coarse=False,
+        geo=_geo(),
+        dadata_qc_geo=None,
+    )
+
+
+def _anchor_comp(ppm2: float) -> dict[str, Any]:
+    return {"price_per_m2": ppm2, "area_m2": 50.0, "rooms": 2, "floor": 3, "total_floors": 9}
+
+
+def test_anchor_tier_reset_when_anchor_not_built() -> None:
+    """Якорь не построен (комплов меньше min_comps) → anchor_tier=None, а не залипшая "C".
+
+    Побочный дефект, вскрытый замером #2656: сброс делал только low-conf гейт, а когда
+    _compute_same_building_anchor возвращала None сама, флаг оставался равным
+    anchor_tier_fetched. Дальше по коду он читается как «headline построил якорь».
+    """
+    pr = _price(
+        anchor_comps=[_anchor_comp(150_000.0), _anchor_comp(155_000.0)], anchor_tier_fetched="C"
+    )
+
+    assert pr.anchor_tier is None
+    assert pr.anchor_comps_used == []
+    assert pr.median_ppm2 == 200_000.0  # headline остался радиусным
+
+
+def test_sticky_anchor_tier_no_longer_mutes_imv_blend() -> None:
+    """Тот же случай + IMV-якорь: blend (гейт `anchor_tier is None`) обязан сработать.
+
+    Это и есть денежное последствие залипшего флага: 24 оценки, у которых якоря нет ни
+    до, ни после правки, всё равно меняли цену — IMV-blend молча глушился. Falsifiable:
+    без фикса anchor_tier == "C" → blend пропускается → median_price остаётся 10 000 000.
+    """
+    imv = {"recommended_price": 14_000_000}  # > медианы (10 млн) × threshold 1.15
+    pr = _price(
+        anchor_comps=[_anchor_comp(150_000.0), _anchor_comp(155_000.0)],
+        anchor_tier_fetched="C",
+        imv_anchor=imv,
+    )
+
+    assert pr.anchor_tier is None
+    assert pr.median_price > 10_000_000, "IMV-blend заглушен залипшим anchor_tier"
+    assert pr.avito_imv_summary is not None