feat(tradein/estimator): кросс-source физдедуп аналогов за флагом (#2087 H4)

Один физический лот кросс-постится на avito+cian+domklik (разные source и
source_id) → existing radius-дедуп по (source, source_id) его НЕ ловит →
n_analogs И source_counts раздуты кросс-постами (аудит #2087: лот
80м²/265000₽/м² = N1+Домклик+Циан ×3; «14 аналогов» → ~6-7 уникальных).

_dedup_cross_source схлопывает дубли по физическому ключу (building_cadastral
| нормализованный address + floor + area-bucket round(m²) + price-bucket
round(₽/100k)) ДО подсчёта n_analogs/median/cv, оставляя свежайшего (scraped_at)
представителя. За флагом estimate_dedup_analogs_enabled (default OFF =
байт-идентично; регресс-гейт зелёный).

Бэктест #1966 (400 ДКП, full spine, OFF vs ON, тот же сэмпл): MAPE 13.89%→13.89%,
coverage 83.33%→83.33%, bias −3.83%→−3.83%, median width 0.743→0.743, median cv
0.0988→0.0988, avg n_analogs 27.64→27.57 (дедуп отработал 107× на 335 оценках).
Кросс-посты имеют identical price → нулевой вклад в дисперсию → cv/коридор НЕ
сужаются: это фикс ЧЕСТНОСТИ СЧЁТА (n_analogs не раздут, source_counts по
физлотам), accuracy-нейтральный, а НЕ рычаг cv-сужения (рычаг cv→коридор —
estimate_sb_clip_after_weight, уже default ON). Default OFF — narrowing не
продемонстрирован; флаг готов к canary/монкипатч-бэктесту.

Refs #2087

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

@@ -244,6 +244,31 @@ class Settings(BaseSettings):
     # ENV: ESTIMATE_RADIUS_DEDUP_ENABLED.
     estimate_radius_dedup_enabled: bool = True
 
+    # ── #2087 H4: кросс-source физический дедуп аналогов ──────────────────────
+    # Radius-дедуп выше ловит только повторы ВНУТРИ одного source (source, source_id).
+    # Один физический лот кросс-постится на avito+cian+domklik (разные source, разные
+    # source_id) → radius-дедуп его НЕ схлопывает → он считается несколько раз →
+    # раздувает n_analogs И cv (→ шире коридор), может смещать медиану. Прод-аудит
+    # #2087: лот 80м²/265000₽/м² = N1+Домклик+Циан (×3); «14 аналогов» → ~6-7 уникальных.
+    # True схлопывает дубли по ФИЗИЧЕСКОМУ ключу до подсчёта n_analogs/median/cv:
+    #   building (building_cadastral_number | нормализованный address)
+    #   + floor + area_bucket (round(area_m2), ~±0.5 м²)
+    #   + price_bucket (round(price_rub / 100000), ~±0.5% @21М / ~±2% @2.5М).
+    # Из группы остаётся ОДИН представитель (свежайший scraped_at), НЕ суммируем;
+    # n_analogs/median/cv/source_counts/sources_used считаются по физическим лотам
+    # («лот считается один раз»).
+    #
+    # Бэктест #1966 (400 ДКП, radius-путь, full spine, OFF vs ON): MAPE 13.89% →
+    # 13.89%, coverage 83.33% → 83.33%, bias −3.83% → −3.83%, median width 0.743 →
+    # 0.743, median cv 0.0988 → 0.0988; avg n_analogs 27.64 → 27.57. Дедуп отработал
+    # 107× на 335 оценках, но снимает лишь identical-price кросс-посты (дубли имеют
+    # ТУ ЖЕ цену → нулевой вклад в дисперсию) → cv/коридор НЕ сужаются. Это фикс
+    # ЧЕСТНОСТИ СЧЁТА (n_analogs не раздут ×3 кросс-постами, source_counts по
+    # физлотам), accuracy-нейтральный, а НЕ рычаг сужения cv (рычаг cv→коридор —
+    # estimate_sb_clip_after_weight, уже ON). Default False (байт-идентично) — флаг
+    # готов к монкипатч-бэктесту/canary. ENV: ESTIMATE_DEDUP_ANALOGS_ENABLED.
+    estimate_dedup_analogs_enabled: bool = False
+
     # ── #1871 P2: split-дома wide-corridor disclosure (default ON, порог 1.2) ──
     # Tier A (same-building) матчит по address-regex (намеренно НЕ house_id — дом
     # дробится на несколько house_id). На split-доме разной этажности comp_min..max

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

@@ -1942,6 +1942,11 @@ def _price_from_inputs(
     direct unit testing. All DB lookups are injected via callables or pre-fetched
     values; behavior is byte-identical to the original block.
     """
+    # 2b. Кросс-source физический дедуп (#2087 H4) — ДО outlier-фильтра и
+    # агрегации, чтобы n_analogs/median/cv считались по уникальным лотам, а не
+    # по кросс-постам. No-op при estimate_dedup_analogs_enabled=False.
+    listings = _dedup_cross_source(listings)
+
     # 3. Outlier filter
     listings_clean = _filter_outliers(listings)
 
@@ -4429,6 +4434,113 @@ def _apply_corridor_clamp(
     return new_ppm2, new_price, new_low, new_high, True
 
 
+# ── #2087 H4: кросс-source физический дедуп аналогов ─────────────────────────
+# Radius-дедуп (_RN_DUP_WINDOW) ловит только повторы ВНУТРИ одного source
+# (source, source_id). Один физический лот, кросс-запостенный на avito+cian+
+# domklik, имеет РАЗНЫЕ source → переживает radius-дедуп → считается несколько
+# раз → раздувает n_analogs И cv (→ шире коридор), может смещать медиану.
+# _dedup_cross_source схлопывает такие дубли по физическому ключу ДО агрегации.
+_DEDUP_PRICE_BUCKET_RUB = 100_000
+_ADDR_NORM_RE = re.compile(r"[^0-9a-zа-яё]+", flags=re.IGNORECASE)
+
+
+def _norm_addr_for_dedup(addr: str | None) -> str:
+    """Нормализация адреса для физического ключа: lower + схлопнутая пунктуация.
+
+    Кросс-source адреса форматируются по-разному («ул. Ленина, д.5» vs
+    «Ленина 5») — нормализуем к сравнимому виду. Пустой/None → "".
+    """
+    if not addr:
+        return ""
+    return _ADDR_NORM_RE.sub(" ", addr.strip().lower()).strip()
+
+
+def _phys_dedup_key(lot: dict[str, Any]) -> tuple[str, Any, int, int] | None:
+    """Физический ключ лота (building, floor, area_bucket, price_bucket).
+
+    None, если нет данных для ключа (нет площади/цены/building) — такой лот
+    НЕ дедупим (оставляем как уникальный, чтобы не сливать неоднозначное).
+    building = building_cadastral_number (надёжнее) ИЛИ нормализованный address.
+    area_bucket = round(area_m2) (~±0.5 м² — поглощает округление 66.9 vs 67.0).
+    price_bucket = round(price_rub / _DEDUP_PRICE_BUCKET_RUB): у настоящего
+    кросс-поста цена идентична (тот же продавец), допуск лишь гасит округление.
+    """
+    area = lot.get("area_m2")
+    price = lot.get("price_rub")
+    if not area or not price:
+        return None
+    cad = lot.get("building_cadastral_number")
+    building = (
+        str(cad).strip() if cad and str(cad).strip() else _norm_addr_for_dedup(lot.get("address"))
+    )
+    if not building:
+        return None
+    return (
+        building,
+        lot.get("floor"),
+        round(float(area)),
+        round(float(price) / _DEDUP_PRICE_BUCKET_RUB),
+    )
+
+
+def _dedup_rep_key(lot: dict[str, Any]) -> tuple[float, int, str, str]:
+    """Ключ выбора представителя из группы-дубля (больше = предпочтительнее).
+
+    Свежайший scraped_at → есть фото → source → source_url (детерминированный
+    tie-break). Свежайшая цена наиболее актуальна для оценки.
+    """
+    sa = lot.get("scraped_at")
+    sa_ord = sa.timestamp() if isinstance(sa, datetime) else 0.0
+    has_photo = 1 if lot.get("photo_urls") else 0
+    return (sa_ord, has_photo, lot.get("source") or "", lot.get("source_url") or "")
+
+
+def _dedup_cross_source(lots: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """Схлопывает кросс-source дубли одного физического лота (#2087 H4).
+
+    Из группы с общим физическим ключом (_phys_dedup_key) оставляет ОДНОГО
+    представителя (свежайший scraped_at), НЕ суммирует. Порядок вывода — по
+    первому появлению группы (сохраняет relevance-сортировку из _fetch_analogs
+    для top-N UI). Представитель несёт свой source → n_analogs/median/cv/
+    source_counts/sources_used считаются по физическим лотам («лот считается
+    один раз»; source кросс-поста, не ставший представителем, выпадает из
+    выборки — по нему нет независимой ценовой точки). Флаг OFF → no-op
+    (байт-идентичный проход). Лоты без ключа (нет площади/цены/building) не
+    дедупятся — остаются уникальными.
+    """
+    if not settings.estimate_dedup_analogs_enabled or len(lots) < 2:
+        return lots
+
+    groups: dict[Any, list[dict[str, Any]]] = {}
+    order: list[Any] = []
+    for lot in lots:
+        key: Any = _phys_dedup_key(lot)
+        if key is None:
+            key = ("__nodup__", id(lot))  # неоднозначный → уникальный псевдоключ
+        if key not in groups:
+            groups[key] = []
+            order.append(key)
+        groups[key].append(lot)
+
+    result: list[dict[str, Any]] = []
+    collapsed = 0
+    for key in order:
+        group = groups[key]
+        if len(group) == 1:
+            result.append(group[0])
+        else:
+            result.append(max(group, key=_dedup_rep_key))
+            collapsed += len(group) - 1
+    if collapsed:
+        logger.info(
+            "cross-source dedup #2087 H4: %d lots → %d unique (-%d dups)",
+            len(lots),
+            len(result),
+            collapsed,
+        )
+    return result
+
+
 def _filter_outliers(lots: list[dict[str, Any]]) -> list[dict[str, Any]]:
     """Tukey IQR rule: исключаем точки вне [Q1 - k×IQR, Q3 + k×IQR].
 

tradein-mvp/backend/tests/test_estimator_dedup_cross_source_2087.py

@@ -0,0 +1,164 @@
+"""Unit tests for cross-source physical dedup of analogs (#2087 H4).
+
+Проверяет:
+  - физический ключ (_phys_dedup_key) — что образует / не образует;
+  - _dedup_cross_source: кросс-source дубли схлопываются в 1 представителя
+    (свежайший scraped_at), не-дубли не трогаются, флаг OFF = байт-идентично;
+  - выбор представителя детерминирован (freshest scraped_at).
+
+No DB / network: помощники оперируют plain-dict'ами. DATABASE_URL нужен только
+для импорта app.core.config.Settings (тот же паттерн, что в test_estimator_pure_units).
+"""
+
+import os
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from datetime import UTC, datetime
+
+import pytest
+
+from app.services import estimator
+
+
+def _lot(
+    *,
+    source: str,
+    address: str = "ул. Ленина 5",
+    area: float = 67.0,
+    price: int = 21_995_000,
+    floor: int | None = 4,
+    scraped_at: datetime | None = None,
+    cad: str | None = None,
+    source_url: str | None = None,
+    photo_urls: list[str] | None = None,
+) -> dict:
+    return {
+        "source": source,
+        "address": address,
+        "area_m2": area,
+        "price_rub": price,
+        "price_per_m2": int(price / area) if area else 0,
+        "floor": floor,
+        "scraped_at": scraped_at or datetime(2026, 6, 1, tzinfo=UTC),
+        "building_cadastral_number": cad,
+        "source_url": source_url,
+        "photo_urls": photo_urls,
+    }
+
+
+# --------------------------------------------------------------------------- #
+# _phys_dedup_key
+# --------------------------------------------------------------------------- #
+
+
+def test_phys_key_prefers_cadastral_over_address() -> None:
+    lot = _lot(source="avito", address="ул. Ленина 5", cad="66:41:0601025:94")
+    key = estimator._phys_dedup_key(lot)
+    assert key is not None
+    assert key[0] == "66:41:0601025:94"  # building = cad, не address
+
+
+def test_phys_key_normalizes_address_when_no_cad() -> None:
+    # Тот же адрес, разное форматирование (регистр/пунктуация/пробелы) —
+    # punctuation-only нормализация схлопывает их к одному building-компоненту.
+    a = estimator._phys_dedup_key(_lot(source="avito", address="Екатеринбург, ул. Ленина, д. 5"))
+    b = estimator._phys_dedup_key(_lot(source="cian", address="екатеринбург  ул ленина д 5"))
+    assert a is not None and b is not None
+    assert a[0] == b[0]
+
+
+def test_phys_key_different_streets_do_not_merge() -> None:
+    # Токенное отличие (разная улица) НЕ должно сливаться — защита от ложного merge.
+    a = estimator._phys_dedup_key(_lot(source="avito", address="ул. Ленина 5"))
+    b = estimator._phys_dedup_key(_lot(source="cian", address="ул. Мира 5"))
+    assert a is not None and b is not None
+    assert a[0] != b[0]
+
+
+def test_phys_key_none_without_area_or_price() -> None:
+    assert estimator._phys_dedup_key(_lot(source="avito", area=0)) is None
+    assert estimator._phys_dedup_key(_lot(source="avito", price=0)) is None
+
+
+def test_phys_key_area_bucket_tolerates_rounding() -> None:
+    # 66.9 и 67.0 → один и тот же area_bucket (round → 67)
+    a = estimator._phys_dedup_key(_lot(source="avito", area=66.9))
+    b = estimator._phys_dedup_key(_lot(source="cian", area=67.0))
+    assert a is not None and b is not None
+    assert a[2] == b[2] == 67
+
+
+# --------------------------------------------------------------------------- #
+# _dedup_cross_source — флаг OFF (byte-identical)
+# --------------------------------------------------------------------------- #
+
+
+def test_dedup_flag_off_is_noop(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(estimator.settings, "estimate_dedup_analogs_enabled", False)
+    lots = [
+        _lot(source="avito"),
+        _lot(source="cian"),
+        _lot(source="domklik"),
+    ]
+    out = estimator._dedup_cross_source(lots)
+    assert out is lots  # тот же объект → нулевое влияние на выборку
+
+
+# --------------------------------------------------------------------------- #
+# _dedup_cross_source — флаг ON
+# --------------------------------------------------------------------------- #
+
+
+def test_dedup_collapses_cross_source_triple(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(estimator.settings, "estimate_dedup_analogs_enabled", True)
+    # Один физлот 67м²/21.995М на floor 4 — кросс-пост avito+cian+domklik.
+    lots = [
+        _lot(source="avito", scraped_at=datetime(2026, 6, 1, tzinfo=UTC)),
+        _lot(source="cian", scraped_at=datetime(2026, 6, 3, tzinfo=UTC)),
+        _lot(source="domklik", scraped_at=datetime(2026, 6, 2, tzinfo=UTC)),
+    ]
+    out = estimator._dedup_cross_source(lots)
+    assert len(out) == 1
+    # представитель = свежайший scraped_at (cian, 06-03)
+    assert out[0]["source"] == "cian"
+
+
+def test_dedup_keeps_distinct_units(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(estimator.settings, "estimate_dedup_analogs_enabled", True)
+    # Разные физлоты: разный этаж / площадь / цена / здание → не дубли.
+    lots = [
+        _lot(source="avito", floor=4, area=67.0, price=21_995_000),
+        _lot(source="cian", floor=7, area=67.0, price=21_995_000),  # другой этаж
+        _lot(source="yandex", floor=4, area=80.0, price=21_200_000),  # другая площадь
+        _lot(source="avito", floor=4, area=67.0, price=25_500_000),  # другая цена
+        _lot(source="cian", address="ул. Мира 3", floor=4, area=67.0),  # другой дом
+    ]
+    out = estimator._dedup_cross_source(lots)
+    assert len(out) == 5  # ничего не схлопнулось
+
+
+def test_dedup_preserves_first_seen_order(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(estimator.settings, "estimate_dedup_analogs_enabled", True)
+    # relevance-порядок из _fetch_analogs должен сохраниться: группа занимает
+    # позицию первого появления.
+    lots = [
+        _lot(source="avito", address="ул. Мира 3", area=50.0, price=10_000_000),
+        _lot(source="avito", address="ул. Ленина 5", area=67.0, price=21_995_000),
+        _lot(source="cian", address="ул. Ленина 5", area=67.0, price=21_995_000),  # dup #2
+        _lot(source="yandex", address="ул. Гагарина 9", area=90.0, price=30_000_000),
+    ]
+    out = estimator._dedup_cross_source(lots)
+    assert len(out) == 3
+    assert [o["address"] for o in out] == ["ул. Мира 3", "ул. Ленина 5", "ул. Гагарина 9"]
+
+
+def test_dedup_lots_without_key_stay_unique(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(estimator.settings, "estimate_dedup_analogs_enabled", True)
+    # Нет площади → ключ None → не дедупим, даже если оба «пустые».
+    lots = [
+        _lot(source="avito", area=0),
+        _lot(source="cian", area=0),
+    ]
+    out = estimator._dedup_cross_source(lots)
+    assert len(out) == 2