fix(tradein/estimator): savepoint+retry вокруг ratio-lookup — transient не гасит expected_sold (#2265)

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

@@ -263,46 +263,63 @@ def _get_asking_sold_ratio(
 
     ratio: float | None = None
     basis: str | None = None
-    try:
+    # #2265 D2: SAVEPOINT + 1 retry вокруг ratio-lookup. Транзиентный сбой —
-        row = db.execute(
+    # обычно poisoned tx (InFailedSqlTransaction) от вышестоящего graceful-except:
-            text(
+    # первый же SELECT падает, и раньше это гасило expected_sold для КОНКРЕТНОЙ
-                """
+    # оценки (3/224 за 14 дней при живых ratio-бакетах). Оборачиваем в begin_nested:
-                SELECT ratio, basis FROM asking_to_sold_ratios
+    # при сбое savepoint откатывается, tx чистится (db.rollback), retry идёт на
-                WHERE rooms_bucket = CAST(:b AS int) AND district = ''
+    # здоровом соединении. begin_nested также локализует benign-сбой (таблицы нет
-                """
+    # на свежей/старой БД, миграция 080 не применена) — без отравления outer tx.
-            ),
+    last_exc: Exception | None = None
-            {"b": bucket},
+    ok = False
-        ).fetchone()
+    for attempt in range(2):
-        if row is None:
-            # Бакет тонкий (n<30 при seed'е) или отсутствует → global fallback (-1).
-            row = db.execute(
-                text(
-                    """
-                    SELECT ratio, basis FROM asking_to_sold_ratios
-                    WHERE rooms_bucket = -1 AND district = ''
-                    """
-                ),
-            ).fetchone()
-        if row is not None and row.ratio is not None:
-            ratio = float(row.ratio)
-            basis = row.basis
-    except Exception as exc:
-        # Таблицы может не быть на свежей/старой БД (миграция 080 не применена),
-        # либо транзакция в сбойном состоянии — graceful: без sold-коррекции.
-        # ОБЯЗАТЕЛЬНО rollback: неудачный SELECT помечает транзакцию
-        # InFailedSqlTransaction, и без отката следующий statement упал бы → 500.
-        #
-        # НЕ кэшируем этот None: ошибка транзиентна (poisoned tx от вышестоящего
-        # graceful-except, миг. лаг, коннект-хиккап). Раньше строка кэша ниже
-        # писалась безусловно → один сбой отравлял _asking_sold_ratio_cache[bucket]
-        # = (None, None) на весь TTL (300с) и молча гасил expected_sold («ожидаемая
-        # цена сделки» → «—») для ВСЕХ оценок этого rooms-бакета на воркере до
-        # истечения TTL. Ранний return без записи в кэш → следующая оценка ретраит.
-        logger.debug("asking_to_sold_ratio lookup skipped (graceful, NOT cached): %s", exc)
         try:
-            db.rollback()
+            with db.begin_nested():
-        except Exception:
+                row = db.execute(
-            pass
+                    text(
+                        """
+                        SELECT ratio, basis FROM asking_to_sold_ratios
+                        WHERE rooms_bucket = CAST(:b AS int) AND district = ''
+                        """
+                    ),
+                    {"b": bucket},
+                ).fetchone()
+                if row is None:
+                    # Бакет тонкий (n<30 при seed'е) или отсутствует → global (-1).
+                    row = db.execute(
+                        text(
+                            """
+                            SELECT ratio, basis FROM asking_to_sold_ratios
+                            WHERE rooms_bucket = -1 AND district = ''
+                            """
+                        ),
+                    ).fetchone()
+                if row is not None and row.ratio is not None:
+                    ratio = float(row.ratio)
+                    basis = row.basis
+            ok = True
+            break
+        except Exception as exc:
+            last_exc = exc
+            # begin_nested откатывает СВОЙ savepoint; но если outer tx уже была
+            # poisoned до savepoint — сам SAVEPOINT не встаёт → чистим всю tx,
+            # чтобы retry шёл на здоровом соединении.
+            try:
+                db.rollback()
+            except Exception:
+                pass
+            if attempt == 0:
+                logger.debug("asking_to_sold_ratio lookup failed, retrying on clean tx: %s", exc)
+                continue
+
+    if not ok:
+        # Оба захода упали — graceful без sold-коррекции. НЕ кэшируем этот None:
+        # ошибка транзиентна (poisoned tx, миг. лаг, коннект-хиккап). Раньше строка
+        # кэша ниже писалась безусловно → один сбой отравлял
+        # _asking_sold_ratio_cache[bucket] = (None, None) на весь TTL (300с) и молча
+        # гасил expected_sold для ВСЕХ оценок бакета до истечения TTL (#2175). Ранний
+        # return без записи в кэш → следующая оценка ретраит.
+        logger.debug("asking_to_sold_ratio lookup skipped (graceful, NOT cached): %s", last_exc)
         return None, None
 
     # Кэшируем ТОЛЬКО успешный lookup. ratio может быть None (строки нет —
@@ -4537,53 +4554,149 @@ def _apply_corridor_clamp(
     return new_ppm2, new_price, new_low, new_high, True
 
 
-# ── #2087 H4: кросс-source физический дедуп аналогов ─────────────────────────
+# ── #2087 H4 / #2265: кросс-source физический дедуп аналогов ──────────────────
 # Radius-дедуп (_RN_DUP_WINDOW) ловит только повторы ВНУТРИ одного source
 # (source, source_id). Один физический лот, кросс-запостенный на avito+cian+
 # domklik, имеет РАЗНЫЕ source → переживает radius-дедуп → считается несколько
 # раз → раздувает n_analogs И cv (→ шире коридор), может смещать медиану.
 # _dedup_cross_source схлопывает такие дубли по физическому ключу ДО агрегации.
+#
+# #2265: адресный fallback-ключ ужесточён. Раньше building = lower + схлопнутая
+# пунктуация ВСЕГО адреса → три source-формата одного дома дают три разных
+# building-компонента и один физлот считается тремя. Прод-пример (оценка
+# d5fc3841): один лот 33.6 м²/3 этаж/4.0М на ул. Сыромолотова показан как 3
+# аналога, т.к. cadnum пуст у всех троих, а адреса форматированы по-разному:
+#   cian    «р-н Кировский, мкр. ЖБИ, улица Сыромолотова, …»
+#   domklik «Россия, Екатеринбург, ЖБИ м-н, улица Сыромолотова…»
+#   yandex  «Екатеринбург, улица Сыромолотова, 11»
+# domklik building_cadastral_number ВСЕГДА пуст (0/6296), yandex/avito/cian —
+# 17/29/33% → cadnum-ключ между источниками тоже не совпадает.
+#
+# Фикс: из адреса извлекаем нормализованный street_token (тип улицы + город +
+# район + «Россия» выброшены) и строим композит (street_token, floor, area,
+# price). house_no СОЗНАТЕЛЬНО не входит в equality-ключ — он присутствует
+# непоследовательно (yandex несёт номер, cian/domklik часто нет), включение
+# помешало бы слить именно тот кросс-пост, ради которого дедуп существует.
+# Разные физлоты остаются разными через floor/area/price (требование «не
+# over-merge»). Дедуп union-find по ЛЮБОМУ совпавшему ключу (cadnum-композит ИЛИ
+# street-композит) → закрывает и «у одного cadnum есть, у другого нет, адрес
+# совпадает». Два РАЗНЫХ кадастра (авторитетный сигнал разных зданий) НЕ
+# сливаются, даже если street-композит совпал.
 _DEDUP_PRICE_BUCKET_RUB = 100_000
-_ADDR_NORM_RE = re.compile(r"[^0-9a-zа-яё]+", flags=re.IGNORECASE)
+
+# Типы улиц (для отсечения от названия). District-маркеры (мкр/м-н/квартал/
+# район) СЮДА НЕ входят — иначе «мкр. ЖБИ, улица Сыромолотова» матчилось бы на
+# «мкр ЖБИ» вместо реальной улицы; они убираются как шум в _ADDR_NOISE_WORDS.
+_STREET_TYPE_WORDS = (
+    "улица",
+    "ул",
+    "переулок",
+    "пер",
+    "проспект",
+    "просп",
+    "пр-кт",
+    "пр-т",
+    "пр",
+    "бульвар",
+    "б-р",
+    "бул",
+    "проезд",
+    "шоссе",
+    "ш",
+    "набережная",
+    "наб",
+    "площадь",
+    "пл",
+    "аллея",
+    "тупик",
+    "тракт",
+    "линия",
+)
+# Длинные типы раньше коротких, чтобы «улица» матчилось раньше «ул».
+_STREET_TYPE_ALT = "|".join(re.escape(w) for w in sorted(_STREET_TYPE_WORDS, key=len, reverse=True))
+# «<тип>[.] <название>»: название = 1-3 буквенных слова, до запятой/цифры.
+_STREET_NAME_RE = re.compile(
+    rf"(?:^|[,\s])(?:{_STREET_TYPE_ALT})\.?\s+([а-яё][а-яё\-]+(?:\s+[а-яё][а-яё\-]+){{0,2}})",
+    flags=re.IGNORECASE,
+)
+# Слова-маркеры дома, которые могли попасть в захват названия (напр. «ленина д 5»).
+_HOUSE_MARKER_WORDS = frozenset(
+    {"дом", "д", "корпус", "корп", "к", "строение", "стр", "литера", "лит", "владение", "вл"}
+)
+# Шум-токены (город/страна/район/микрорайон) на случай, если попали в название.
+_ADDR_NOISE_WORDS = frozenset(
+    {
+        "россия",
+        "рф",
+        "екатеринбург",
+        "г",
+        "город",
+        "область",
+        "обл",
+        "район",
+        "р-н",
+        "мкр",
+        "м-н",
+        "микрорайон",
+        "квартал",
+    }
+)
 
 
-def _norm_addr_for_dedup(addr: str | None) -> str:
+def _extract_street_token(addr: str | None) -> str:
-    """Нормализация адреса для физического ключа: lower + схлопнутая пунктуация.
+    """Нормализованный уличный токен для дедуп-ключа (#2265).
 
-    Кросс-source адреса форматируются по-разному («ул. Ленина, д.5» vs
+    Из адреса произвольного source-формата вытаскивает название улицы без типа
-    «Ленина 5») — нормализуем к сравнимому виду. Пустой/None → "".
+    (улица/пер/пр…), города, района и «Россия». Не распознан / пусто → "".
     """
     if not addr:
         return ""
-    return _ADDR_NORM_RE.sub(" ", addr.strip().lower()).strip()
+    m = _STREET_NAME_RE.search(addr.strip().lower())
+    if not m:
+        return ""
+    words = [
+        w
+        for w in m.group(1).split()
+        if w and w not in _HOUSE_MARKER_WORDS and w not in _ADDR_NOISE_WORDS
+    ]
+    return " ".join(words).strip()
 
 
-def _phys_dedup_key(lot: dict[str, Any]) -> tuple[str, Any, int, int] | None:
+def _lot_dedup_components(
-    """Физический ключ лота (building, floor, area_bucket, price_bucket).
+    lot: dict[str, Any],
+) -> tuple[str, tuple[str, Any, int, int] | None, tuple[str, Any, int, int] | None]:
+    """(cadnum, cadnum-композит, street-композит) лота для union-find дедупа.
 
-    None, если нет данных для ключа (нет площади/цены/building) — такой лот
+    Каждый композит = (building, floor, area_bucket, price_bucket) или None,
-    НЕ дедупим (оставляем как уникальный, чтобы не сливать неоднозначное).
+    если нет соответствующего сигнала. area_bucket = round(area_m2) (~±0.5 м² —
-    building = building_cadastral_number (надёжнее) ИЛИ нормализованный address.
+    гасит 66.9 vs 67.0), price_bucket = round(price_rub / _DEDUP_PRICE_BUCKET_RUB)
-    area_bucket = round(area_m2) (~±0.5 м² — поглощает округление 66.9 vs 67.0).
+    (у настоящего кросс-поста цена идентична, допуск лишь гасит округление).
-    price_bucket = round(price_rub / _DEDUP_PRICE_BUCKET_RUB): у настоящего
+    Нет площади/цены → оба None (лот уникален, не дедупится).
-    кросс-поста цена идентична (тот же продавец), допуск лишь гасит округление.
     """
     area = lot.get("area_m2")
     price = lot.get("price_rub")
     if not area or not price:
-        return None
+        return "", None, None
+    floor = lot.get("floor")
+    area_b = round(float(area))
+    price_b = round(float(price) / _DEDUP_PRICE_BUCKET_RUB)
     cad = lot.get("building_cadastral_number")
-    building = (
+    cad_s = str(cad).strip() if cad and str(cad).strip() else ""
-        str(cad).strip() if cad and str(cad).strip() else _norm_addr_for_dedup(lot.get("address"))
+    cad_key = (cad_s, floor, area_b, price_b) if cad_s else None
-    )
+    street = _extract_street_token(lot.get("address"))
-    if not building:
+    street_key = (street, floor, area_b, price_b) if street else None
-        return None
+    return cad_s, cad_key, street_key
-    return (
+
-        building,
+
-        lot.get("floor"),
+def _phys_dedup_key(lot: dict[str, Any]) -> tuple[str, Any, int, int] | None:
-        round(float(area)),
+    """Первичный физический ключ (building, floor, area_bucket, price_bucket).
-        round(float(price) / _DEDUP_PRICE_BUCKET_RUB),
+
-    )
+    building = cadnum (надёжнее) ИЛИ street_token (#2265). None, если нет
+    площади/цены или не из чего построить building. Сохраняет 4-кортежную форму
+    (canonical-ключ; union-find в _dedup_cross_source использует оба композита).
+    """
+    _cad_s, cad_key, street_key = _lot_dedup_components(lot)
+    return cad_key or street_key
 
 
 def _dedup_rep_key(lot: dict[str, Any]) -> tuple[float, int, str, str]:
@@ -4599,36 +4712,88 @@ def _dedup_rep_key(lot: dict[str, Any]) -> tuple[float, int, str, str]:
 
 
 def _dedup_cross_source(lots: list[dict[str, Any]]) -> list[dict[str, Any]]:
-    """Схлопывает кросс-source дубли одного физического лота (#2087 H4).
+    """Схлопывает кросс-source дубли одного физического лота (#2087 H4 / #2265).
 
-    Из группы с общим физическим ключом (_phys_dedup_key) оставляет ОДНОГО
+    Union-find по физическим композитам: два лота — дубли, если совпал ЛЮБОЙ
-    представителя (свежайший scraped_at), НЕ суммирует. Порядок вывода — по
+    ключ — cadnum-композит ИЛИ street-композит (_lot_dedup_components). Из группы
-    первому появлению группы (сохраняет relevance-сортировку из _fetch_analogs
+    оставляет ОДНОГО представителя (свежайший scraped_at), НЕ суммирует. Порядок
-    для top-N UI). Представитель несёт свой source → n_analogs/median/cv/
+    вывода — по первому появлению группы (сохраняет relevance-сортировку из
-    source_counts/sources_used считаются по физическим лотам («лот считается
+    _fetch_analogs для top-N UI). Представитель несёт свой source → n_analogs/
-    один раз»; source кросс-поста, не ставший представителем, выпадает из
+    median/cv/source_counts/sources_used считаются по физическим лотам («лот
-    выборки — по нему нет независимой ценовой точки). Флаг OFF → no-op
+    считается один раз»; source кросс-поста, не ставший представителем, выпадает
-    (байт-идентичный проход). Лоты без ключа (нет площади/цены/building) не
+    из выборки — по нему нет независимой ценовой точки).
-    дедупятся — остаются уникальными.
+
+    #2265: cadnum-композит обрабатывается первым (авторитетный), затем street-
+    композит — но union по street НЕ сливает два РАЗНЫХ непустых кадастра
+    (разные здания). Флаг OFF → no-op (байт-идентичный проход). Лоты без
+    площади/цены не дедупятся — остаются уникальными.
     """
     if not settings.estimate_dedup_analogs_enabled or len(lots) < 2:
         return lots
 
-    groups: dict[Any, list[dict[str, Any]]] = {}
+    n = len(lots)
-    order: list[Any] = []
+    parent = list(range(n))
-    for lot in lots:
+    comps = [_lot_dedup_components(lot) for lot in lots]
-        key: Any = _phys_dedup_key(lot)
+    # cadnum группы (root → кадастр «» если группа без/со смешанным кадастром).
-        if key is None:
+    root_cad = [c[0] for c in comps]
-            key = ("__nodup__", id(lot))  # неоднозначный → уникальный псевдоключ
+
-        if key not in groups:
+    def find(x: int) -> int:
-            groups[key] = []
+        while parent[x] != x:
-            order.append(key)
+            parent[x] = parent[parent[x]]  # path-halving
-        groups[key].append(lot)
+            x = parent[x]
+        return x
+
+    def union(a: int, b: int) -> None:
+        ra, rb = find(a), find(b)
+        if ra == rb:
+            return
+        ca, cb = root_cad[ra], root_cad[rb]
+        if ca and cb and ca != cb:
+            return  # разные кадастры → разные здания, не сливаем
+        merged = ca or cb
+        # меньший индекс — корень: сохраняет first-seen порядок групп.
+        if ra < rb:
+            parent[rb] = ra
+            root_cad[ra] = merged
+        else:
+            parent[ra] = rb
+            root_cad[rb] = merged
+
+    # 1) cadnum-композит — авторитетный (тот же кадастр = то же здание).
+    cad_owner: dict[tuple[str, Any, int, int], int] = {}
+    for i, (_cad_s, cad_key, _street_key) in enumerate(comps):
+        if cad_key is None:
+            continue
+        prev = cad_owner.get(cad_key)
+        if prev is None:
+            cad_owner[cad_key] = i
+        else:
+            union(prev, i)
+
+    # 2) street-композит — union при отсутствии cadnum-конфликта.
+    street_owner: dict[tuple[str, Any, int, int], int] = {}
+    for i, (_cad_s, _cad_key, street_key) in enumerate(comps):
+        if street_key is None:
+            continue
+        prev = street_owner.get(street_key)
+        if prev is None:
+            street_owner[street_key] = i
+        else:
+            union(prev, i)
+
+    groups: dict[int, list[dict[str, Any]]] = {}
+    order: list[int] = []
+    for i, lot in enumerate(lots):
+        root = find(i)
+        if root not in groups:
+            groups[root] = []
+            order.append(root)
+        groups[root].append(lot)
 
     result: list[dict[str, Any]] = []
     collapsed = 0
-    for key in order:
+    for root in order:
-        group = groups[key]
+        group = groups[root]
         if len(group) == 1:
             result.append(group[0])
         else:
@@ -4636,7 +4801,7 @@ def _dedup_cross_source(lots: list[dict[str, Any]]) -> list[dict[str, Any]]:
             collapsed += len(group) - 1
     if collapsed:
         logger.info(
-            "cross-source dedup #2087 H4: %d lots → %d unique (-%d dups)",
+            "cross-source dedup #2087/#2265: %d lots → %d unique (-%d dups)",
             len(lots),
             len(result),
             collapsed,

tradein-mvp/backend/tests/test_asking_sold_ratio_cache_poison.py

@@ -12,6 +12,8 @@ No DB: fake-Session (monkeypatched .execute/.rollback). DATABASE_URL нужен
 """
 
 import os
+from collections.abc import Iterator
+from contextlib import contextmanager
 from types import SimpleNamespace
 from typing import Any
 
@@ -37,6 +39,11 @@ class _OkDB:
         self._row = SimpleNamespace(ratio=ratio, basis=basis)
         self.rolled_back = False
 
+    @contextmanager
+    def begin_nested(self) -> Iterator[Any]:
+        # #2265 D2: ratio-lookup обёрнут в SAVEPOINT — моделируем CM.
+        yield SimpleNamespace()
+
     def execute(self, *_a: Any, **_k: Any) -> _Result:
         return _Result(self._row)
 
@@ -50,6 +57,10 @@ class _BoomDB:
     def __init__(self) -> None:
         self.rolled_back = False
 
+    @contextmanager
+    def begin_nested(self) -> Iterator[Any]:
+        yield SimpleNamespace()
+
     def execute(self, *_a: Any, **_k: Any) -> _Result:
         raise RuntimeError("InFailedSqlTransaction")
 

tradein-mvp/backend/tests/test_estimator_dedup_cross_source_2087.py

@@ -179,3 +179,128 @@ def test_dedup_lots_without_key_stay_unique(monkeypatch: pytest.MonkeyPatch) ->
     ]
     out = estimator._dedup_cross_source(lots)
     assert len(out) == 2
+
+
+# --------------------------------------------------------------------------- #
+# #2265 — устойчивый адресный street-токен (кросс-source форматы)
+# --------------------------------------------------------------------------- #
+
+
+def test_extract_street_token_across_source_formats() -> None:
+    # Три РЕАЛЬНЫХ прод-формата одного дома (ул. Сыромолотова) → один токен.
+    cian = estimator._extract_street_token("р-н Кировский, мкр. ЖБИ, улица Сыромолотова, 11")
+    domklik = estimator._extract_street_token("Россия, Екатеринбург, ЖБИ м-н, улица Сыромолотова")
+    yandex = estimator._extract_street_token("Екатеринбург, улица Сыромолотова, 11")
+    assert cian == domklik == yandex == "сыромолотова"
+
+
+def test_extract_street_token_strips_house_and_noise() -> None:
+    assert estimator._extract_street_token("Екатеринбург, ул. Ленина, д. 5") == "ленина"
+    assert estimator._extract_street_token("проспект Космонавтов 15") == "космонавтов"
+    assert estimator._extract_street_token("пер. Банковский, 8а") == "банковский"
+    # Нет типа улицы → распознать нельзя → пусто (безопасный fallback, лот уникален).
+    assert estimator._extract_street_token("Екатеринбург, ЖК Дружба") == ""
+    assert estimator._extract_street_token("") == ""
+    assert estimator._extract_street_token(None) == ""
+
+
+def test_dedup_collapses_three_source_formats(monkeypatch: pytest.MonkeyPatch) -> None:
+    # Прод-дефект (оценка d5fc3841): один физлот 33.6 м²/этаж 3/4.0М показан
+    # как 3 аналога — cadnum пуст у всех троих, адреса в трёх форматах.
+    monkeypatch.setattr(estimator.settings, "estimate_dedup_analogs_enabled", True)
+    lots = [
+        _lot(
+            source="cian",
+            address="р-н Кировский, мкр. ЖБИ, улица Сыромолотова, 11",
+            area=33.6,
+            price=4_000_000,
+            floor=3,
+            cad=None,
+        ),
+        _lot(
+            source="domklik",
+            address="Россия, Екатеринбург, ЖБИ м-н, улица Сыромолотова",
+            area=33.6,
+            price=4_000_000,
+            floor=3,
+            cad=None,
+        ),
+        _lot(
+            source="yandex",
+            address="Екатеринбург, улица Сыромолотова, 11",
+            area=33.6,
+            price=4_000_000,
+            floor=3,
+            cad=None,
+        ),
+    ]
+    out = estimator._dedup_cross_source(lots)
+    assert len(out) == 1  # один физлот, не три
+
+
+def test_dedup_merges_cadnum_asymmetry_same_street(monkeypatch: pytest.MonkeyPatch) -> None:
+    # У одного cadnum есть, у другого нет, адрес совпадает → сливаются по
+    # street-композиту (#2265: union по любому совпавшему ключу).
+    monkeypatch.setattr(estimator.settings, "estimate_dedup_analogs_enabled", True)
+    lots = [
+        _lot(
+            source="cian",
+            address="улица Сыромолотова, 11",
+            area=33.6,
+            price=4_000_000,
+            floor=3,
+            cad="66:41:0601011:1234",
+        ),
+        _lot(
+            source="domklik",
+            address="ЖБИ м-н, улица Сыромолотова",
+            area=33.6,
+            price=4_000_000,
+            floor=3,
+            cad=None,
+        ),
+    ]
+    out = estimator._dedup_cross_source(lots)
+    assert len(out) == 1
+
+
+def test_dedup_distinct_cadnums_same_metrics_do_not_merge(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # Две башни на одной улице с одинаковыми floor/area/price, но РАЗНЫМИ
+    # кадастрами — авторитетный сигнал разных зданий → НЕ сливаем.
+    monkeypatch.setattr(estimator.settings, "estimate_dedup_analogs_enabled", True)
+    lots = [
+        _lot(
+            source="cian",
+            address="улица Сыромолотова, 11",
+            area=33.6,
+            price=4_000_000,
+            floor=3,
+            cad="66:41:0601011:1234",
+        ),
+        _lot(
+            source="avito",
+            address="улица Сыромолотова, 13",
+            area=33.6,
+            price=4_000_000,
+            floor=3,
+            cad="66:41:0601011:9999",
+        ),
+    ]
+    out = estimator._dedup_cross_source(lots)
+    assert len(out) == 2
+
+
+def test_dedup_same_street_different_metrics_stay_distinct(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # Разные лоты на одной улице с разной ценой/площадью/этажом → остаются
+    # разными (защита от over-merge при отсутствии cadnum).
+    monkeypatch.setattr(estimator.settings, "estimate_dedup_analogs_enabled", True)
+    lots = [
+        _lot(source="cian", address="улица Сыромолотова, 11", area=33.6, price=4_000_000, floor=3),
+        _lot(source="avito", address="улица Сыромолотова, 11", area=45.0, price=6_500_000, floor=8),
+    ]
+    out = estimator._dedup_cross_source(lots)
+    assert len(out) == 2

tradein-mvp/backend/tests/test_estimator_ratio_savepoint_retry_2265.py

@@ -0,0 +1,103 @@
+"""Unit tests for SAVEPOINT + retry вокруг ratio-lookup (#2265 D2).
+
+Транзиентный сбой первого ratio-SELECT (обычно poisoned tx,
+InFailedSqlTransaction, от вышестоящего graceful-except) раньше гасил
+expected_sold для КОНКРЕТНОЙ оценки (3/224 за 14 дней при живых бакетах).
+Фикс: begin_nested + 1 retry на чистом savepoint. Проверяем:
+  - первый lookup падает → второй проходит → возвращается ratio (recover);
+  - оба захода падают → (None, None) и кэш НЕ отравлен (#2175 держится);
+  - happy-path — ровно один begin_nested, без rollback.
+
+No DB / network: session — лёгкий fake с side-effect'ами на execute.
+DATABASE_URL нужен только для импорта app.core.config.Settings.
+"""
+
+import os
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from contextlib import contextmanager
+from typing import Any
+from unittest.mock import MagicMock
+
+from app.services import estimator
+
+
+class _Row:
+    """Строка результата с полями .ratio/.basis (как row из fetchone())."""
+
+    def __init__(self, ratio: float | None, basis: str | None) -> None:
+        self.ratio = ratio
+        self.basis = basis
+
+
+class _FakeSession:
+    """Session-заглушка: execute отдаёт side-effect'ы по очереди.
+
+    Exception в списке → execute бросает; иначе — result, чей fetchone() вернёт
+    заготовленную строку. begin_nested — CM, rollback — счётчик.
+    """
+
+    def __init__(self, execute_effects: list[Any]) -> None:
+        self._effects = list(execute_effects)
+        self.rollback_count = 0
+        self.begin_nested_count = 0
+
+    @contextmanager
+    def begin_nested(self):  # type: ignore[no-untyped-def]
+        self.begin_nested_count += 1
+        yield MagicMock()
+
+    def execute(self, *_args: Any, **_kwargs: Any) -> Any:
+        eff = self._effects.pop(0)
+        if isinstance(eff, Exception):
+            raise eff
+        result = MagicMock()
+        result.fetchone.return_value = eff
+        return result
+
+    def rollback(self) -> None:
+        self.rollback_count += 1
+
+
+def test_ratio_retry_recovers_after_first_lookup_fails() -> None:
+    estimator._asking_sold_ratio_cache.clear()
+    poisoned = RuntimeError("current transaction is aborted (InFailedSqlTransaction)")
+    db = _FakeSession([poisoned, _Row(0.84, "per_rooms")])
+
+    ratio, basis = estimator._get_asking_sold_ratio(db, rooms=1)  # type: ignore[arg-type]
+
+    assert ratio == 0.84
+    assert basis == "per_rooms"
+    assert db.begin_nested_count == 2  # первый заход упал, второй прошёл
+    assert db.rollback_count >= 1  # чистка tx перед retry
+    assert 1 in estimator._asking_sold_ratio_cache  # успех кэшируется
+
+
+def test_ratio_both_attempts_fail_returns_none_and_not_cached() -> None:
+    estimator._asking_sold_ratio_cache.clear()
+    db = _FakeSession([RuntimeError("boom-1"), RuntimeError("boom-2")])
+
+    ratio, basis = estimator._get_asking_sold_ratio(db, rooms=2)  # type: ignore[arg-type]
+
+    assert ratio is None
+    assert basis is None
+    assert 2 not in estimator._asking_sold_ratio_cache  # транзиент НЕ отравляет кэш (#2175)
+    assert db.begin_nested_count == 2
+    assert db.rollback_count >= 2
+
+
+def test_ratio_happy_path_single_attempt_no_rollback() -> None:
+    estimator._asking_sold_ratio_cache.clear()
+    db = _FakeSession([_Row(0.9, "global_fallback")])
+
+    ratio, basis = estimator._get_asking_sold_ratio(db, rooms=0)  # type: ignore[arg-type]
+
+    assert ratio == 0.9
+    assert basis == "global_fallback"
+    assert db.begin_nested_count == 1  # ни одного retry
+    assert db.rollback_count == 0
+    cached = estimator._asking_sold_ratio_cache.get(0)
+    assert cached is not None
+    assert cached[0] == 0.9
+    assert cached[1] == "global_fallback"