Merge pull request 'feat(tradein/estimator): кросс-source физдедуп аналогов за флагом (#2087 H4)' (#2173) from fix/tradein-analog-dedup-cv into main

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

@@ -244,6 +244,33 @@ 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 ON (#2173): бэктест #1966 OFF vs
+    # ON accuracy-идентичен (MAPE 13.89%, coverage 83.33%, bias −3.83%, median width/cv
+    # без изменений), меняется только user-visible n_analogs — перестаёт быть раздутым
+    # кросс-постингом ×3. ENV: ESTIMATE_DEDUP_ANALOGS_ENABLED (=false откатывает).
+    estimate_dedup_analogs_enabled: bool = True
+
     # ── #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

@@ -1952,6 +1952,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)
 
@@ -4471,6 +4476,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_backtest_regression_gate.py

@@ -25,8 +25,11 @@ import math
 import os
 from pathlib import Path
 
+import pytest
+
 os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
 
+from app.services import estimator
 from scripts.backtest_estimator import load_fixture, replay_fixture
 
 _FIXTURES = Path(__file__).parent / "fixtures"
@@ -73,7 +76,16 @@ def test_fixture_and_baseline_committed() -> None:
     assert _BASELINE_PATH.exists(), f"frozen baseline missing: {_BASELINE_PATH}"
 
 
-def test_backtest_regression_gate() -> None:
+def test_backtest_regression_gate(monkeypatch: pytest.MonkeyPatch) -> None:
+    # The frozen fixture records the injected-callback control flow captured with
+    # cross-source physical dedup OFF (#2087 H4 was a no-op default at capture time).
+    # Dedup is now default ON (#2173), but this gate is a byte-identical REPLAY of a
+    # frozen OFF capture — with dedup active _dedup_cross_source would trim listings
+    # before quarter_indexes_lookup and the recorded call sequence would diverge. Pin
+    # the flag OFF so the replay follows the captured control flow. (Accuracy-neutral:
+    # backtest #1966 OFF vs ON is identical; dedup only trims user-visible n_analogs,
+    # so the OFF baseline stays the valid spine regression reference.)
+    monkeypatch.setattr(estimator.settings, "estimate_dedup_analogs_enabled", False)
     fixture = load_fixture(_FIXTURE_PATH)
     baseline = json.loads(_BASELINE_PATH.read_text(encoding="utf-8"))
     # Round-trip the replay output through JSON before comparing: the committed

tradein-mvp/backend/tests/test_estimator_dedup_cross_source_2087.py

@@ -0,0 +1,181 @@
+"""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
+
+
+# --------------------------------------------------------------------------- #
+# Флаг: default ON (#2173) vs явный OFF (byte-identical)
+# --------------------------------------------------------------------------- #
+
+
+def test_dedup_default_is_on() -> None:
+    # #2173: default флага теперь ON (в проде дедуп активен без ENV-override).
+    # Без monkeypatch settings отражает config-default → дедуп схлопывает
+    # кросс-пост-тройку одного физлота. Явно НЕ монкипатчим — проверяем именно
+    # поведение по умолчанию.
+    assert estimator.settings.estimate_dedup_analogs_enabled is True
+    lots = [
+        _lot(source="avito"),
+        _lot(source="cian"),
+        _lot(source="domklik"),
+    ]
+    out = estimator._dedup_cross_source(lots)
+    assert len(out) == 1  # дедуп активен по умолчанию
+
+
+def test_dedup_flag_off_is_noop(monkeypatch: pytest.MonkeyPatch) -> None:
+    # OFF-путь проверяем ЯВНЫМ override (не полагаясь на default, который теперь ON):
+    # при флаге OFF проход байт-идентичен — возвращается тот же объект.
+    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

tradein-mvp/backend/tests/test_estimator_n_analogs_priced.py

@@ -26,9 +26,13 @@ from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any
 
+import pytest
+
 # Settings требует DATABASE_URL при инициализации (fail-fast, C-3).
 os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
 
+from app.services import estimator
+
 # Переиспользуем harness _run_estimate / _make_payload из anchor-теста (полный
 # estimate путь со всеми I/O застабленными). anchor_comps=[] + anchor_tier=None
 # направляет estimate в radius-aggregation ветку (где живёт фикс #2).
@@ -39,6 +43,16 @@ _h = importlib.util.module_from_spec(_spec)
 _spec.loader.exec_module(_h)
 
 
+@pytest.fixture(autouse=True)
+def _pin_dedup_off(monkeypatch: pytest.MonkeyPatch) -> None:
+    # Эти тесты проверяют инвариант #2 «n_analogs = число priced-аналогов»,
+    # ортогональный кросс-source физдедупу (#2087 H4, default ON с #2173).
+    # Синтетические аналоги делят один адрес/этаж/площадь → дедуп схлопнул бы
+    # близкие по цене как кросс-посты и исказил счётчик. Пиним флаг OFF, чтобы
+    # мерить именно #2 в изоляции (не полагаясь на прод-default, который теперь ON).
+    monkeypatch.setattr(estimator.settings, "estimate_dedup_analogs_enabled", False)
+
+
 def _listing(price_per_m2: float | None, area_m2: float = 60.0) -> dict[str, Any]:
     """Радиусный аналог; price_per_m2=None моделирует листинг без ₽/м².
 

tradein-mvp/backend/tests/test_same_building_anchor.py

@@ -24,7 +24,9 @@ os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import anyio
+import pytest
 
+from app.services import estimator
 from app.services.estimator import (
     _compute_same_building_anchor,
     _geocode_is_coarse,
@@ -1045,8 +1047,15 @@ def test_anchor_n_analogs_syncs_to_anchor_count() -> None:
     assert "по 4 аналогам" in (est.confidence_explanation or "")
 
 
-def test_radius_path_n_analogs_unchanged() -> None:
-    """#695 контроль: без якоря n_analogs = радиусный счёт (поведение не меняется)."""
+def test_radius_path_n_analogs_unchanged(monkeypatch: pytest.MonkeyPatch) -> None:
+    """#695 контроль: без якоря n_analogs = радиусный счёт (поведение не меняется).
+
+    Проверяет radius-passthrough vs anchor-override — ортогонально кросс-source
+    дедупу (#2087 H4, default ON с #2173). 5 ИДЕНТИЧНЫХ листингов — это ровно
+    патологический кросс-пост, который дедуп схлопнул бы в 1; пиним флаг OFF,
+    чтобы мерить именно #695-инвариант, не полагаясь на прод-default (теперь ON).
+    """
+    monkeypatch.setattr(estimator.settings, "estimate_dedup_analogs_enabled", False)
     radius = [_make_listing(price_per_m2=200_000.0) for _ in range(5)]
     est = _run_estimate(anchor_comps=[], anchor_tier=None, radius_analogs=radius)
     assert est.n_analogs == 5