fix(tradein/v2): canonical sources_used — kill «ИСТОЧНИКОВ X/7» POST↔GET desync

Единый helper estimator._canonical_sources(analogs, valuation_flags) — источник
правды для sources_used, зовётся идентично на POST (estimate_quality) и
GET-rehydrate (get_estimate) из ТЕХ ЖЕ persisted analogs + оценочных флагов.

Root cause: три расходящиеся деривации — POST брал sources_used из top-N
analogs_lots, GET возвращал persisted-колонку (радиусный набор + quarter_index),
source_counts на GET считался из persisted analogs. Источник (напр. cian) мог
попасть в source_counts, но не в sources_used → счётчик «X/7» прыгал 4→5 между
POST-ответом и reload(GET).

- sources_used = {листинговые из persisted analogs} ∪ {avito_imv/yandex_valuation/
  cian_valuation}. Детерминированно отсортирован.
- source_counts на POST теперь тоже из analogs_lots (не полной metadata-выборки)
  → инвариант source_counts.keys() ⊆ sources_used на POST и GET.
- POST персистит канонический sources_used в колонку (history/PDF консистентны для
  новых строк); GET рехайдрейтит его же helper'ом — чинит и СТАРЫЕ строки
  (листинговая часть пересобирается из analogs, quarter_index/радиусный шум
  отбрасывается фильтром оценочных флагов).

Оценочные флаги персистятся в колонке sources_used и читаются оттуда на GET —
реконструкция не требуется.

Repro 451de30b: до — sources_used=[avito,avito_imv,domklik,yandex], source_counts
имеет cian (не в sources_used); после — sources_used=[avito,avito_imv,cian,
domklik,yandex] (5/7), counts.keys() ⊆ sources_used.

Part of #2087 (M1).

tradein-mvp/backend/app/api/v1/trade_in.py

@@ -235,6 +235,7 @@ def get_estimate(
     _assert_estimate_access(row.created_by, x_authenticated_user)
 
     from app.services.estimator import (
+        _canonical_sources,
         _cv_from_ppm2,
         _fetch_dkp_corridor,
         _fetch_house_imv_anchor,
@@ -252,6 +253,15 @@ def get_estimate(
     # в строке. created_at берём из колонки (persisted).
     cv = _cv_from_ppm2([a.price_per_m2 for a in analogs])
     source_counts = _source_counts([a.source for a in analogs])
+    # #2087 (M1): sources_used рехайдрейтим тем же helper'ом, что и POST — из ТЕХ ЖЕ
+    # persisted analogs (листинговая часть) ∪ оценочных флагов, вычитанных из
+    # persisted-колонки sources_used (avito_imv/yandex_valuation/cian_valuation).
+    # Так один estimate_id → идентичный sources_used на POST/GET/reload, а
+    # source_counts.keys() ⊆ sources_used (обе части — из одного набора analogs).
+    # Чинит рассинхром и на СТАРЫХ строках (где колонка хранит радиусный набор +
+    # quarter_index): листинговая часть пересобирается из analogs, служебный шум
+    # отбрасывается фильтром helper'а.
+    sources_used = _canonical_sources((a.source for a in analogs), (row.sources_used or []))
 
     # #696: POST-only производные поля (price_trend / avito_imv / dkp_corridor /
     # last_scraped_at) на GET-rehydrate ранее были null → на shared-link / PDF /
@@ -315,7 +325,7 @@ def get_estimate(
         target_address=row.address,
         target_lat=row.lat,
         target_lon=row.lon,
-        sources_used=row.sources_used or [],
+        sources_used=sources_used,
         data_freshness_minutes=row.data_freshness_minutes,
         # #696 — POST-only производные поля, пересчитанные на rehydrate (см. выше).
         last_scraped_at=last_scraped_at,

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

@@ -26,7 +26,7 @@ import logging
 import math
 import re
 import time
-from collections.abc import Callable
+from collections.abc import Callable, Iterable
 from dataclasses import dataclass
 from datetime import UTC, date, datetime, timedelta
 from typing import Any
@@ -1839,6 +1839,40 @@ def _source_counts(sources: list[str | None]) -> dict[str, int]:
     return dict(sorted(counts.items()))
 
 
+# Оценочные (не-листинговые) источники канона «7» источников. Листинговые
+# (avito/cian/yandex/domklik) выводятся динамически из реальных analog-карточек;
+# эти три — внешние оценки, которые не порождают analog-строк.
+_CANONICAL_VALUATION_SOURCES: frozenset[str] = frozenset(
+    {"avito_imv", "yandex_valuation", "cian_valuation"}
+)
+
+
+def _canonical_sources(
+    analog_sources: Iterable[str | None],
+    valuation_flags: Iterable[str | None],
+) -> list[str]:
+    """Единый источник правды для sources_used — #2087 (M1).
+
+    sources_used = {листинговые источники, РЕАЛЬНО присутствующие в persisted
+    analogs} ∪ {оценочные источники (avito_imv/yandex_valuation/cian_valuation),
+    которые действительно использованы}. Детерминированно отсортирован.
+
+    Вызывается идентично на POST (estimate_quality) и GET-rehydrate
+    (api.v1.trade_in.get_estimate) из ТЕХ ЖЕ persisted analogs, поэтому один
+    estimate_id → идентичный sources_used на POST/GET/reload. Гарантирует
+    инвариант source_counts.keys() ⊆ sources_used: и counts, и листинговая часть
+    sources_used считаются из одного и того же набора analog-источников.
+
+    valuation_flags фильтруется до канонических оценочных источников — можно
+    безопасно передать сырой persisted-массив sources_used (листинговые сорта и
+    служебные метки вроде quarter_index отбрасываются, т.к. листинговая часть
+    берётся из analogs).
+    """
+    listing = {s for s in analog_sources if s}
+    valuation = {s for s in valuation_flags if s in _CANONICAL_VALUATION_SOURCES}
+    return sorted(listing | valuation)
+
+
 @dataclass
 class PricingResult:
     """Return type of _price_from_inputs — все переменные, нужные estimate_quality после блока."""
@@ -2917,7 +2951,6 @@ async def estimate_quality(
     expected_sold_range_high = pr.expected_sold_range_high
     asking_to_sold_ratio = pr.asking_to_sold_ratio
     ratio_basis = pr.ratio_basis
-    sources_used_pre = pr.sources_used_pre
     listings_clean = pr.listings_clean
     cv = pr.cv
 
@@ -2952,9 +2985,21 @@ async def estimate_quality(
         analogs_lots = [_listing_to_analog(lot) for lot in listings_clean[:10]]
         metadata_lots = listings_clean
     deals_lots = [_deal_to_analog(d) for d in deals[:10]]
-    # #2043 (BE-1): счётчики по источнику считаем по ПОЛНОЙ выборке (metadata_lots
+    # #2087 (M1): единый канонический sources_used + source_counts. ОБА считаем по
-    # — anchor-комплы или радиусные listings_clean, ДО top-N отсечки на UI).
+    # persisted analogs_lots (top-N, ровно то, что уходит в колонку analogs и
-    source_counts = _source_counts([lot.get("source") for lot in metadata_lots])
+    # рехайдрейтится на GET) — не по полной metadata-выборке. Иначе POST-ответ
+    # (полная выборка) расходился бы с GET-reload (persisted top-N): источник мог
+    # оказаться в source_counts, но не в sources_used → счётчик «X/7» прыгал.
+    # Инвариант: source_counts.keys() ⊆ sources_used (общий набор analog-источников).
+    valuation_flags: set[str] = set()
+    if imv_eval is not None:
+        valuation_flags.add("avito_imv")
+    if yandex_val is not None:
+        valuation_flags.add("yandex_valuation")
+    if cian_val is not None and cian_val.sale_price_rub:
+        valuation_flags.add("cian_valuation")
+    sources_used = _canonical_sources((lot.source for lot in analogs_lots), valuation_flags)
+    source_counts = _source_counts([lot.source for lot in analogs_lots])
     freshness_pre = _compute_freshness_minutes(metadata_lots)
     # DaData enrichment (PR Q1) — заполняется только если service отработал.
     # При DaData = None все колонки идут в DB как NULL (graceful).
@@ -3035,7 +3080,11 @@ async def estimate_quality(
             "deals_json": json.dumps(
                 [a.model_dump(mode="json") for a in deals_lots], ensure_ascii=False
             ),
-            "sources_json": json.dumps(sources_used_pre, ensure_ascii=False),
+            # #2087 (M1): персистим КАНОНИЧЕСКИЙ sources_used (листинговые из
+            # analogs ∪ оценочные флаги) — тот же массив, что в POST-ответе. GET
+            # рехайдрейтит его же (валид. флаги читаются из этой колонки), history
+            # и PDF получают консистентный «X/7» без quarter_index/радиусного шума.
+            "sources_json": json.dumps(sources_used, ensure_ascii=False),
             "freshness": freshness_pre,
             "canonical_address": dadata.canonical_address if dadata else None,
             "house_cadnum": dadata.house_cadnum if dadata else None,
@@ -3086,13 +3135,9 @@ async def estimate_quality(
         f" cian={cian_val.sale_price_rub}" if cian_val and cian_val.sale_price_rub else "",
     )
 
-    sources_used = sorted({lot.source for lot in analogs_lots if lot.source})
+    # #2087 (M1): sources_used / source_counts уже посчитаны выше (канонически, из
-    if imv_eval is not None:
+    # persisted analogs_lots) — не пересчитываем, чтобы POST-ответ был байт-в-байт
-        sources_used = sorted(set(sources_used) | {"avito_imv"})
+    # тем, что персистится и рехайдрейтится на GET.
-    if yandex_val is not None:
-        sources_used = sorted(set(sources_used) | {"yandex_valuation"})
-    if cian_val is not None and cian_val.sale_price_rub:
-        sources_used = sorted(set(sources_used) | {"cian_valuation"})
     freshness_min = _compute_freshness_minutes(metadata_lots)
     last_scraped_at = _compute_last_scraped_at(metadata_lots)
     # Месячный ₽/м² тренд целевого дома (web TREND chart) — best-effort, None если нет данных.

tradein-mvp/backend/tests/test_estimate_idor.py

@@ -143,6 +143,9 @@ def _stub_precision_and_pdf():
         # persisted analogs. Empty analogs in the fixture → None / {} (real behaviour).
         _cv_from_ppm2=lambda *a, **k: None,
         _source_counts=lambda *a, **k: {},
+        # #2087 (M1): GET-rehydrate derives canonical sources_used via the shared
+        # helper. Empty analogs + no valuation flags → [] (real behaviour).
+        _canonical_sources=lambda *a, **k: [],
     )
     real_estimator = sys.modules.get("app.services.estimator")
     sys.modules["app.services.estimator"] = estimator_stub  # type: ignore[assignment]

tradein-mvp/backend/tests/test_estimator_source_metrics_2043.py

@@ -81,6 +81,64 @@ def test_source_counts_keys_sorted() -> None:
     assert list(counts.keys()) == ["avito", "cian", "yandex"]
 
 
+# --------------------------------------------------------------------------- #
+# _canonical_sources — #2087 (M1) single source of truth for sources_used
+# --------------------------------------------------------------------------- #
+
+
+def test_canonical_sources_listing_plus_valuation() -> None:
+    # Listing sources come from analogs; valuation flags are unioned on top.
+    result = estimator._canonical_sources(
+        ["avito", "cian", "avito"], {"avito_imv", "yandex_valuation"}
+    )
+    assert result == ["avito", "avito_imv", "cian", "yandex_valuation"]
+
+
+def test_canonical_sources_deterministic_same_inputs_same_output() -> None:
+    # Same set of analogs + flags → identical, sorted sources_used (POST == GET).
+    analogs = ["yandex", "avito", "domklik", "avito"]
+    flags = {"cian_valuation", "avito_imv"}
+    a = estimator._canonical_sources(analogs, flags)
+    b = estimator._canonical_sources(list(reversed(analogs)), set(flags))
+    assert a == b
+    assert a == ["avito", "avito_imv", "cian_valuation", "domklik", "yandex"]
+
+
+def test_canonical_sources_skips_none_and_empty_analogs() -> None:
+    result = estimator._canonical_sources(["avito", None, "", "cian"], set())
+    assert result == ["avito", "cian"]
+
+
+def test_canonical_sources_filters_non_canonical_valuation_flags() -> None:
+    # Passing a raw persisted column (listing sorts + quarter_index) as the
+    # valuation arg must NOT leak listing/service labels — only the 3 canonical
+    # valuation sources survive. Listing part comes exclusively from analogs.
+    raw_column = ["avito", "domklik", "quarter_index", "avito_imv", "yandex"]
+    result = estimator._canonical_sources(["cian"], raw_column)
+    # From analogs: cian. From column: only avito_imv is a canonical valuation.
+    assert result == ["avito_imv", "cian"]
+    assert "quarter_index" not in result
+    assert "domklik" not in result  # listing label in the column is ignored
+
+
+def test_canonical_source_counts_keys_subset_of_sources_used() -> None:
+    # Invariant #2087: every key of source_counts (derived from the same analog
+    # sources) is present in sources_used.
+    analog_sources = ["avito", "cian", "avito", "yandex", None, ""]
+    sources_used = estimator._canonical_sources(analog_sources, {"avito_imv"})
+    source_counts = estimator._source_counts(analog_sources)
+    assert set(source_counts.keys()) <= set(sources_used)
+    # And every listing source present in analogs is reflected in sources_used.
+    for s in analog_sources:
+        if s:
+            assert s in sources_used
+
+
+def test_canonical_sources_empty_is_empty_list() -> None:
+    assert estimator._canonical_sources([], []) == []
+    assert estimator._canonical_sources([None, ""], ["quarter_index"]) == []
+
+
 # --------------------------------------------------------------------------- #
 # _price_from_inputs — cv surfaced on both paths
 # --------------------------------------------------------------------------- #