fix(trade-in): привести SOLD-сторону коэффициента asking→sold к сегодняшнему дню

Числитель ratio — сырые цены сделок Росреестра за трейлинг-12 месяцев (на проде
фактическое окно 2025-10..2026-04, центр ≈ декабрь 2025), знаменатель — медиана
объявлений за LISTINGS_FRESH_DAYS, то есть сегодня. На растущем рынке ratio
занижен ровно на дрейф за половину окна, и прод отдаёт цену позапрошлого
квартала. Это и есть «систематическая недооценка на свежих сделках».

Каждая сделка домножается на sber-фактор приведения к последнему месяцу ряда —
тот же приём и та же карта регион→ряд (sber_region_series_name), что estimator
уже применяет к ДКП-коридору (#794). До этой правки коридор был приведён к
сегодня, а ratio — нет; теперь обе стороны на одной временной шкале.

Двойного учёта нет: _sber_time_factor в estimator зовётся только в
_fetch_dkp_corridor (estimator.py:2219, 2291) и уходит в clamp ASKING-медианы
(_apply_corridor_clamp, estimator.py:3989); ratio из asking_to_sold_ratios
множится отдельно на estimator.py:4058. Один множитель на одно число дважды
не ложится.

Прод-замер 2026-09-16 (read-only, окно 2025-10..2026-04):
  77 Москва   n=68 731  sold_median 323 529 → 360 330  ×1.1137
  50 область  n=35 264  sold_median 162 303 → 179 314  ×1.1048
  66 Свердл.  n=33 907  sold_median  94 655 →  99 059  ×1.0465
Ожидаемые ratio: 0.729→0.811 (77), 0.811→0.896 (50), 0.728→0.762 (66) — все
ниже 1.0, клампа estimate_expected_sold_le_asking не задевают.

Откат без деплоя — ASKING_RATIO_TIME_ADJUST_ENABLED=false (factor=1.0 для
каждой сделки, байт-в-байт прежнее поведение). Если ряда для города нет вовсе,
сделка не выбрасывается: factor=1.0 плюс счётчик
sber_time_adjust_regions_missing_series и WARNING, а не тишина.

Замороженный регресс-гейт не двигается: фикстура несёт готовый ratio и через
этот таск не ходит.

Claude-Session: https://claude.ai/code/session_01NQb6WeJtagZwZnUsSjDizs

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

@@ -961,6 +961,27 @@ class Settings(BaseSettings):
     # ENV: ASKING_RATIO_PPM2_MAX.
     asking_ratio_ppm2_max: int = 1_200_000
 
+    # Time-adjust для SOLD-стороны asking→sold ratio (rollback-флаг, без релиза).
+    # ПРОБЛЕМА: числитель ratio (deals, trailing 12 months) — сырые цены сделок,
+    # знаменатель (listings, LISTINGS_FRESH_DAYS) — сегодняшний рынок. На растущем
+    # рынке медиана сделок системно отстаёт от медианы объявлений, т.е. ratio занижен
+    # ровно на дрейф рынка за половину окна. Прод-замер 2026-09-16 (окно сделок
+    # 2025-10..2026-04, ряд real_estate_deals своего региона, clamp 0.7/1.6):
+    #   регион 77  n=68 731  sold_median 323 529 → 360 330  (×1.1137, factor 1.077..1.147)
+    #   регион 50  n=35 264  sold_median 162 303 → 179 314  (×1.1048, factor 1.065..1.138)
+    #   регион 66  n=33 907  sold_median  94 655 →  99 059  (×1.0465, factor 1.032..1.062)
+    # Ожидаемые ratio после первого пересчёта: 0.729→0.811 (77), 0.811→0.896 (50),
+    # 0.728→0.762 (66) — все ниже 1.0, клампа estimate_expected_sold_le_asking не задевают.
+    # NB: бэктест меряет предсказание против СЫРЫХ исторических цен сделок, поэтому его
+    # bias после включения уедет в плюс примерно на эти же 4-11%. Это не регресс: прод
+    # обязан отдавать цену СЕГОДНЯШНЕГО рынка, а таргет бэктеста — цена декабря 2025.
+    # При True каждая сделка домножается на sber-фактор приведения (тот же
+    # ряд/карту, что estimator._SBER_REGION_SERIES, см. app.tasks.asking_to_sold_ratio)
+    # к последнему доступному месяцу серии перед расчётом медианы. False — байт-в-байт
+    # прежнее поведение (для быстрого отката без деплоя, сравнения до/после).
+    # ENV: ASKING_RATIO_TIME_ADJUST_ENABLED.
+    asking_ratio_time_adjust_enabled: bool = True
+
     # SSRF-защита для admin scrape endpoints (#756).
     # Список хостов которым разрешено передавать абсолютные URL в параметрах *_url.
     # Относительные пути (без netloc) проходят без проверки — хост подставляется

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

@@ -103,6 +103,103 @@ _AREA_ROOMS_BUCKET_SQL = (
 )
 
 
+# ── #time-adjust: приведение SOLD-стороны к сегодняшнему дню (rollback за settings) ──
+# ПРОБЛЕМА: sold-медиана (deal_side/deal_global/deal_geo) — сырые цены сделок за
+# трейлинг-12 мес (на проде фактическое окно сентябрь 2025 - апрель 2026, медиана
+# ~декабрь 2025), ask-медиана — объявления за LISTINGS_FRESH_DAYS (сегодня). На растущем
+# рынке это занижает ratio (числитель отстаёт от знаменателя во времени). Фикс — тот же
+# приём, что estimator._sber_time_factor применяет к ДКП-коридору (#794): каждая сделка
+# домножается на factor = idx[последний доступный месяц серии] / idx[месяц сделки]
+# (с clamp SBER_TIME_FACTOR_MIN/MAX). Серия и её имя — ТА ЖЕ карта sber_region_series_name
+# (app.services.estimator), НЕ вторая карта — импортируется ЛЕНИВО внутри
+# recompute_asking_to_sold_ratios(), т.к. estimator.py импортирует area_bucket ИЗ этого
+# модуля на верхнем уровне (см. выше) — top-level импорт в обратную сторону дал бы цикл.
+#
+# Двойной учёт (проверено, см. PR-описание/vault): estimator._fetch_dkp_corridor тоже
+# зовёт _sber_time_factor, но на ЖИВОЙ per-request ДКП-коридор (отдельный SQL, отдельная
+# цель — condition-guard corridor-clamp поверх ASKING-медианы, estimator.py:3989-4002),
+# а не на ratio из asking_to_sold_ratios. Эта таблица и коридор — независимые артефакты
+# по одному сырому источнику (deals); один и тот же множитель здесь и там НЕ перемножается
+# на одно и то же число дважды.
+#
+# Дашборд выбирается В SQL по приоритету SBER_COEFF_DASHBOARDS (bind-массив
+# :sber_dashboards, array_position — первый непустой побеждает, тот же порядок, что
+# estimator._load_sber_index_series перебирает по одному дашборду). Приведение
+# управляется флагом settings.asking_ratio_time_adjust_enabled (:time_adjust_enabled) —
+# False даёт factor=1.0 для КАЖДОЙ сделки (байт-в-байт прежнее поведение, откат без
+# деплоя). Если серии для города вообще нет (sber_bounds пуст) — тоже factor=1.0; это
+# НЕ ошибка (сделка не выбрасывается), но recompute_asking_to_sold_ratios логирует
+# факт счётчиком, а не молча (см. _sber_series_missing ниже). Если для конкретного
+# месяца сделки нет точки — берётся ближайший БОЛЕЕ РАННИЙ месяц серии (или самый
+# ранний, если сделка старше начала серии) — то же правило, что estimator._sber_time_factor.
+_SBER_FACTOR_CTES = """
+    sber_series_raw AS (
+        SELECT period_month, index_value_rub_m2, dashboard,
+               array_position(CAST(:sber_dashboards AS text[]), dashboard) AS dash_priority
+        FROM sber_price_index
+        WHERE city = CAST(:sber_city AS text)
+          AND (segment IS NULL OR segment ILIKE '%вторичн%')
+          AND dashboard = ANY(CAST(:sber_dashboards AS text[]))
+    ),
+    sber_best_dashboard AS (
+        SELECT dashboard FROM sber_series_raw
+        WHERE dash_priority IS NOT NULL
+        ORDER BY dash_priority
+        LIMIT 1
+    ),
+    sber_series AS (
+        SELECT r.period_month, r.index_value_rub_m2
+        FROM sber_series_raw r
+        JOIN sber_best_dashboard b ON r.dashboard = b.dashboard
+    ),
+    sber_bounds AS (
+        SELECT MAX(period_month) AS latest_month, MIN(period_month) AS earliest_month
+        FROM sber_series
+    ),
+    sber_latest_value AS (
+        SELECT s.index_value_rub_m2 AS latest_value
+        FROM sber_series s JOIN sber_bounds b ON s.period_month = b.latest_month
+    ),
+    sber_earliest_value AS (
+        SELECT s.index_value_rub_m2 AS earliest_value
+        FROM sber_series s JOIN sber_bounds b ON s.period_month = b.earliest_month
+    )"""
+
+# LEFT JOIN'ы к sber_* CTE выше — вставляются в `FROM deals` ДО `WHERE` (JOIN не может
+# идти после WHERE). ON TRUE — CTE содержат максимум одну строку (не декартово произведение).
+# snb — LATERAL «ближайший месяц серии <= месяца сделки», коррелирован по bare deal_date
+# (единственная таблица в запросе с этой колонкой — алиас не нужен).
+_SBER_FACTOR_JOINS = """
+          LEFT JOIN sber_bounds sb ON TRUE
+          LEFT JOIN sber_latest_value slv ON TRUE
+          LEFT JOIN sber_earliest_value sev ON TRUE
+          LEFT JOIN LATERAL (
+              SELECT s.index_value_rub_m2 AS base_value
+              FROM sber_series s
+              WHERE s.period_month <= date_trunc('month', deal_date)::date
+              ORDER BY s.period_month DESC
+              LIMIT 1
+          ) snb ON TRUE
+"""
+
+# Сам фактор — 1.0 при выключенном флаге, при пустой серии, при сделке новее последнего
+# месяца серии (без экстраполяции вперёд — симметрично estimator._sber_time_factor).
+# GREATEST/LEAST — те же клампы SBER_TIME_FACTOR_MIN/MAX, что estimator применяет к
+# ДКП-коридору (bind :factor_min/:factor_max — значения ОТТУДА, не второй набор констант).
+_SBER_FACTOR_EXPR = """CASE
+            WHEN NOT CAST(:time_adjust_enabled AS boolean) THEN 1.0
+            WHEN sb.latest_month IS NULL THEN 1.0
+            WHEN date_trunc('month', deal_date)::date >= sb.latest_month THEN 1.0
+            ELSE GREATEST(
+                CAST(:factor_min AS double precision),
+                LEAST(
+                    CAST(:factor_max AS double precision),
+                    slv.latest_value / NULLIF(COALESCE(snb.base_value, sev.earliest_value), 0)
+                )
+            )
+        END"""
+
+
 def area_bucket(area_m2: float) -> int:
     """Python-двойник _AREA_ROOMS_BUCKET_SQL (границы ИДЕНТИЧНЫ, #2620).
 
@@ -165,14 +262,18 @@ _DELETE_SQL_REGION = text(
 # (безопасно от SQL-инъекций; CAST не нужен — psycopg v3 передаёт int напрямую).
 _REDERIVE_SQL = text(
     f"""
-    WITH
-    -- SOLD медианы по бакетам комнат за трейлинг-12мес (ДКП Росреестра).
+    WITH{_SBER_FACTOR_CTES},
+    -- SOLD медианы по бакетам комнат за трейлинг-12мес (ДКП Росреестра). #time-adjust:
+    -- price_per_m2 домножен на sber-фактор приведения к последнему месяцу серии.
     deal_side AS (
         SELECT
             LEAST(GREATEST(rooms, 0), 4)                                  AS rooms_bucket,
-            percentile_cont(0.5) WITHIN GROUP (ORDER BY price_per_m2)     AS sold_median,
+            percentile_cont(0.5) WITHIN GROUP (
+                ORDER BY price_per_m2 * ({_SBER_FACTOR_EXPR})
+            )                                                              AS sold_median,
             COUNT(*)                                                      AS n_deals
         FROM deals
+        {_SBER_FACTOR_JOINS}
         WHERE source = 'rosreestr'
           AND rooms IS NOT NULL
           AND city ILIKE :asking_city  -- #C2 SOLD-сторона на ЕКБ (match asking-рынок)
@@ -244,9 +345,12 @@ _REDERIVE_SQL = text(
     -- SOLD медиана по ВСЕМ комнатам (без бакет-фильтра) за трейлинг-12мес — для global row.
     deal_global AS (
         SELECT
-            percentile_cont(0.5) WITHIN GROUP (ORDER BY price_per_m2) AS sold_median,
+            percentile_cont(0.5) WITHIN GROUP (
+                ORDER BY price_per_m2 * ({_SBER_FACTOR_EXPR})
+            )                                                          AS sold_median,
             COUNT(*)                                                  AS n_deals
         FROM deals
+        {_SBER_FACTOR_JOINS}
         WHERE source = 'rosreestr'
           AND rooms IS NOT NULL
           AND city ILIKE :asking_city  -- #C2 SOLD-сторона на ЕКБ (match asking-рынок)
@@ -380,8 +484,11 @@ _GEOM_WARN_SHARE: float = 0.25
 # и деривация разъехались бы при первой же правке одного из них). Состав гардов тот же,
 # что у ЕКБ-деривации (12-мес окно, ppm²-полоса, свежесть #2656, novostroyki #1186,
 # area_m2 IS NOT NULL #2620) — меняется ТОЛЬКО гео-согласование.
-_DEAL_FROM_WHERE_REGION = """
+_DEAL_FROM_REGION = """
         FROM deals
+"""
+
+_DEAL_WHERE_REGION = """
         WHERE source = 'rosreestr'
           AND rooms IS NOT NULL
           AND region_code = CAST(:region_code AS int)
@@ -389,6 +496,10 @@ _DEAL_FROM_WHERE_REGION = """
           AND deal_date >= CURRENT_DATE - INTERVAL '12 months'
 """
 
+# #time-adjust: FROM и WHERE разведены — deal_geo (ниже) вставляет sber-JOIN'ы МЕЖДУ
+# ними (JOIN обязан стоять до WHERE), deal_all (stats, без time-adjust) склеивает как раньше.
+_DEAL_FROM_WHERE_REGION = _DEAL_FROM_REGION + _DEAL_WHERE_REGION
+
 _ASK_FROM_WHERE_REGION = """
         FROM listings
         WHERE is_active
@@ -402,14 +513,19 @@ _ASK_FROM_WHERE_REGION = """
 
 # Ячеечные CTE — ОБЩИЕ для stats-запроса (счётчики + гард) и для самой деривации,
 # чтобы решение «писать / не писать» принималось РОВНО по тем ячейкам, которые потом считаются.
-_CELL_CTES_REGION = f"""
+_CELL_CTES_REGION = f"""{_SBER_FACTOR_CTES},
+    -- #time-adjust: price_per_m2 домножен на sber-фактор ДО попадания в deal_cell/
+    -- deal_cell_bucket — приведение проезжает через весь взвешенный (Ласпейрес) расчёт
+    -- региона автоматически, отдельно трогать deal_cell/per_bucket не нужно.
     deal_geo AS (
         SELECT
             FLOOR(ST_Y(geom) / {_CELL_LAT_DEG})                            AS cell_lat,
             FLOOR(ST_X(geom) / {_CELL_LON_DEG})                            AS cell_lon,
             LEAST(GREATEST(rooms, 0), 4)                                   AS rooms_bucket,
-            price_per_m2
-        {_DEAL_FROM_WHERE_REGION}          AND geom IS NOT NULL
+            price_per_m2 * ({_SBER_FACTOR_EXPR})                           AS price_per_m2
+        {_DEAL_FROM_REGION}
+        {_SBER_FACTOR_JOINS}
+        {_DEAL_WHERE_REGION}          AND geom IS NOT NULL
     ),
     ask_geo AS (
         SELECT
@@ -715,11 +831,73 @@ def recompute_asking_to_sold_ratios(db: Session, run_id: int) -> dict[str, int]:
         # деградировало явно (строк нет → эстиматор без коэффициента).
         "geo_regions_written": 0,
         "geo_regions_skipped": 0,
+        # #time-adjust: сколько регионов реально получили sber-приведение SOLD-стороны
+        # (серия найдена) vs посчитаны с factor=1.0 (флаг выключен ИЛИ серии нет вовсе).
+        "sber_time_adjust_regions_applied": 0,
+        "sber_time_adjust_regions_missing_series": 0,
     }
+    # #time-adjust: ленивый импорт — estimator.py импортирует area_bucket ИЗ этого модуля
+    # на верхнем уровне, top-level импорт в обратную сторону дал бы цикл (см. комментарий
+    # у _SBER_FACTOR_CTES выше).
+    from app.services.estimator import (
+        SBER_COEFF_DASHBOARDS,
+        SBER_TIME_FACTOR_MAX,
+        SBER_TIME_FACTOR_MIN,
+        sber_region_series_name,
+    )
+
+    sber_dashboards = list(SBER_COEFF_DASHBOARDS)
+    time_adjust_enabled = bool(settings.asking_ratio_time_adjust_enabled)
+
+    def _sber_params(sber_city: str) -> dict[str, object]:
+        return {
+            "sber_city": sber_city,
+            "sber_dashboards": sber_dashboards,
+            "time_adjust_enabled": time_adjust_enabled,
+            "factor_min": SBER_TIME_FACTOR_MIN,
+            "factor_max": SBER_TIME_FACTOR_MAX,
+        }
+
+    def _check_sber_series(region_code: int, sber_city: str) -> None:
+        """Пункт 3 задачи: если серии для города вообще нет — не молча, счётчик+лог."""
+        if not time_adjust_enabled:
+            return
+        try:
+            row = (
+                db.execute(
+                    text(
+                        """
+                        SELECT COUNT(*) AS n FROM sber_price_index
+                        WHERE city = CAST(:sber_city AS text)
+                          AND dashboard = ANY(CAST(:sber_dashboards AS text[]))
+                        """
+                    ),
+                    {"sber_city": sber_city, "sber_dashboards": sber_dashboards},
+                )
+                .mappings()
+                .first()
+            )
+        except Exception as exc:  # pragma: no cover — defensive, graceful
+            logger.warning("sber series presence-check failed (graceful): %s", exc)
+            row = None
+        n = int(row["n"]) if row and row.get("n") is not None else 0
+        if n:
+            counters["sber_time_adjust_regions_applied"] += 1
+        else:
+            counters["sber_time_adjust_regions_missing_series"] += 1
+            logger.warning(
+                "asking_to_sold_ratio region_code=%d: sber_price_index серии нет "
+                "(city=%s) — SOLD-сторона считается БЕЗ time-adjust (factor=1.0)",
+                region_code,
+                sber_city,
+            )
+
     try:
         # DELETE + re-derive INSERT в одной транзакции (НЕ коммитим между ними —
         # таблица не должна остаться пустой, если INSERT упадёт). Регион 66 —
         # прежняя ЕКБ-деривация байт-в-байт; остальные регионы — цикл ниже (#3512).
+        ekb_sber_city = sber_region_series_name(66)
+        _check_sber_series(66, ekb_sber_city)
         db.execute(_DELETE_SQL)
         db.execute(
             _REDERIVE_SQL,
@@ -728,14 +906,18 @@ def recompute_asking_to_sold_ratios(db: Session, run_id: int) -> dict[str, int]:
                 "ppm2_max": settings.asking_ratio_ppm2_max,
                 "asking_city": _ASKING_CITY_PATTERN,
                 "fresh_days": LISTINGS_FRESH_DAYS,
+                **_sber_params(ekb_sber_city),
             },
         )
         for region_code in _OTHER_REGION_CODES:
+            region_sber_city = sber_region_series_name(region_code)
+            _check_sber_series(region_code, region_sber_city)
             params = {
                 "region_code": region_code,
                 "ppm2_min": _PPM2_MIN,
                 "ppm2_max": settings.asking_ratio_ppm2_max,
                 "fresh_days": LISTINGS_FRESH_DAYS,
+                **_sber_params(region_sber_city),
             }
             # Сначала состав выборки (#3529) — он же решает, писать ли регион вообще.
             stats_row = db.execute(_GEO_STATS_SQL_REGION, params).mappings().first()

tradein-mvp/backend/tests/test_asking_to_sold_ratio.py

@@ -295,6 +295,22 @@ def test_migration_080_derivation_is_subset_of_refresh_sql() -> None:
         """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 _drop_sber_time_adjust(s: str) -> str:
+        """Remove the #time-adjust sber-factor CTEs/joins/multiplication (absent in 080).
+
+        Verbatim string removal (not regex) — the exact same constants the refresh SQL is
+        built from, so this stays correct even if the CTE text is edited later.
+        """
+        s = s.replace(ratio_mod._SBER_FACTOR_CTES, "")
+        s = s.replace("WITH,", "WITH")
+        s = s.replace(ratio_mod._SBER_FACTOR_JOINS, "")
+        s = s.replace(f"price_per_m2 * ({ratio_mod._SBER_FACTOR_EXPR})", "price_per_m2")
+        # percentile_cont(...) WITHIN GROUP (\n  ORDER BY price_per_m2\n) — multi-line
+        # formatting around the (now-removed) factor leaves whitespace touching the
+        # parens that the 080 seed's single-line form doesn't have; cosmetic only.
+        s = re.sub(r"GROUP\s*\(\s*ORDER BY price_per_m2\s*\)", "GROUP (ORDER BY price_per_m2)", s)
+        return s
+
     def _drop_region_code_column(s: str) -> str:
         """Remove the #3512 region_code column/literal (absent in the 080 seed).
 
@@ -311,7 +327,9 @@ def test_migration_080_derivation_is_subset_of_refresh_sql() -> None:
                 _normalise_bucket_expr(
                     _drop_freshness_guard(
                         _drop_region_code_column(
-                            _drop_area_not_null_guard(_drop_city_guard(_drop_segment_guard(s)))
+                            _drop_area_not_null_guard(
+                                _drop_city_guard(_drop_segment_guard(_drop_sber_time_adjust(s)))
+                            )
                         )
                     )
                 )
@@ -508,7 +526,9 @@ def test_counter_logic_with_fake_db(monkeypatch: pytest.MonkeyPatch) -> None:
     assert db.committed is True
     # #3529: у каждого прочего региона теперь ТРИ стейтмента — geo-stats SELECT,
     # DELETE и (если гард пропустил) INSERT; счётчики несут гео-разрез.
-    expected_calls = 2 + 3 * len(ratio_mod._OTHER_REGION_CODES) + 1
+    # #time-adjust: +1 sber-series presence-check SELECT для EKB и для каждого региона
+    # (_check_sber_series) — EKB теперь 3 стейтмента (было 2), прочий регион — 4 (было 3).
+    expected_calls = 3 + 4 * len(ratio_mod._OTHER_REGION_CODES) + 1
     assert len(db.executed) == expected_calls
     assert db.executed[-1][0] is ratio_mod._COUNTERS_SQL
     assert out["geo_regions_written"] == len(ratio_mod._OTHER_REGION_CODES)