Merge pull request 'fix(estimator): radius (source,source_id) dedup + split-house corridor disclosure (audit #1871 P2)' (#1880) from audit/1871-p2-estimator-dedup-split into main

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

@@ -179,6 +179,30 @@ class Settings(BaseSettings):
     # ENV: ESTIMATE_PRICE_TREND_DEDUP_ENABLED.
     estimate_price_trend_dedup_enabled: bool = True
 
+    # ── #1871 P2: radius-tier (source, source_id) dedup ───────────────────────
+    # Radius-путь _fetch_analogs (Tier S/H/W) кэпит только per-address
+    # (rn_addr <= MAX_ANALOGS_PER_ADDRESS), но (source, source_id)-дубли делят один
+    # address и выживают на разных rn_addr рангах → раздувают n_analogs (prod
+    # 2026-06-23: yandex 48, cian 9, n1 5 excess). Anchor-путь дедупит по
+    # (source, source_id) — radius нет. True (дефолт) добавляет rn_dup=1 фильтр в
+    # каждом тире (freshest scraped_at на (source, source_id|source_url|ctid)).
+    # False = старое поведение без radius-дедупа (для отката). Окно rn_dup всегда
+    # в SQL (безвредно без фильтра); флаг управляет только применением фильтра.
+    # ENV: ESTIMATE_RADIUS_DEDUP_ENABLED.
+    estimate_radius_dedup_enabled: bool = True
+
+    # ── #1871 P2: split-дома wide-corridor disclosure (default OFF) ────────────
+    # Tier A (same-building) матчит по address-regex (намеренно НЕ house_id — дом
+    # дробится на несколько house_id). На split-доме разной этажности comp_min..max
+    # растягивается через несколько ценовых режимов → коридор range_low/high
+    # 148%/170%. Коридор честно широкий, но юзер видит 170% без объяснения. True +
+    # Tier A + corridor_pct > threshold → понижаем confidence на ступень и дописываем
+    # disclosure в explanation. НЕ трогает point/median/range. default False
+    # (no-regress, включаем осознанно). ENV: ESTIMATE_WIDE_CORRIDOR_DISCLOSURE_ENABLED.
+    estimate_wide_corridor_disclosure_enabled: bool = False
+    # Порог ширины коридора (range_high-range_low)/median_price для disclosure (60%).
+    estimate_wide_corridor_threshold: float = 0.6
+
     # ── Mera-audit fix-1: Cian valuation sanity bounds ────────────────────────
     # API-ответ Cian иногда возвращает garbage-значения (999_999 или 9_999_999_999).
     # sale_price_rub вне [min, max] → результат отбрасывается (return None, не кэшируется).

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

@@ -2492,6 +2492,25 @@ async def estimate_quality(
             # которых построена оценка (= показываемый analogs, capped 10). Для anchor-
             # пути это и есть «полное число найденных» по контракту n_analogs (#698).
             n_analogs = anchor["n"]
+            # ── #1871 P2: split-дома wide-corridor disclosure (за флагом, OFF) ──
+            # Tier A матчит по address-regex (намеренно НЕ house_id — дом дробится
+            # на несколько house_id). На split-доме разной этажности comp_min..max
+            # растягивается через несколько ценовых режимов → коридор 148%/170%.
+            # Коридор честно широкий (НЕ сужаем); только понижаем confidence на
+            # ступень и дописываем disclosure. НЕ трогаем point/median/range.
+            if (
+                settings.estimate_wide_corridor_disclosure_enabled
+                and anchor_tier == "A"
+                and median_price > 0
+            ):
+                corridor_pct = (range_high - range_low) / median_price
+                if corridor_pct > settings.estimate_wide_corridor_threshold:
+                    confidence = _downgrade_confidence(confidence)
+                    explanation = (explanation or "") + (
+                        " Широкий ценовой диапазон: дом разбит на секции разной "
+                        "этажности — оценка ориентировочная, уточните по конкретной "
+                        "секции."
+                    )
 
     # ── #651: IMV / Yandex BLEND (killer accuracy fix) — SECONDARY, Tier D only ──
     # Радиусная медиана системно занижает премиум/видовые квартиры (нет class/
@@ -3670,6 +3689,27 @@ _COMMON_WHERE = """
 # ~1280). Не удалять — Tier W не использует _COMMON_WHERE из-за inline
 # relevance_score CASE expressions. Both code paths must stay in sync.
 
+# ── #1871 P2: radius-tier (source, source_id) dedup ──────────────────────────
+# Окно rn_dup в base-CTE каждого radius-тира: (source, source_id)-дубли делят
+# один address и переживают per-address rn_addr cap → раздувают n_analogs.
+# Anchor-путь дедупит по (source, source_id) (:1556); radius — нет. Ключ:
+# source + (id:source_id | url:source_url | ctid:ctid) namespaced-тегом (защита
+# от value-collision между id и url, симметрично anchor dedup). freshest по
+# scraped_at. ВСЕ тиры читают FROM listings (физическая таблица, без JOIN) →
+# source_id / source_url / ctid доступны в base-CTE даже без явного SELECT.
+# psycopg3: source_id::text / ctid::text — это COLUMN::type (не bind-параметр),
+# работает; запрещён только :name::type на bind-параметрах (их тут нет).
+_RN_DUP_WINDOW = """
+                        row_number() OVER (
+                            PARTITION BY source,
+                                CASE
+                                    WHEN source_id IS NOT NULL THEN 'id:' || source_id::text
+                                    WHEN source_url IS NOT NULL THEN 'url:' || source_url
+                                    ELSE 'ctid:' || ctid::text
+                                END
+                            ORDER BY scraped_at DESC
+                        ) AS rn_dup"""
+
 
 def _fetch_analogs(
     db: Session,
@@ -3720,6 +3760,11 @@ def _fetch_analogs(
     """
     area_min = area * (1 - area_tolerance)
     area_max = area * (1 + area_tolerance)
+    # #1871 P2: (source, source_id) dedup в radius-тирах. rn_dup-окно всегда в SQL
+    # (безвредно без фильтра); статический фрагмент управляет только применением
+    # `AND rn_dup = 1` в outer WHERE. Это SQL-литерал (static), НЕ data — psycopg3
+    # bind-параметры не задействованы, инъекции нет.
+    dup_filter = "AND rn_dup = 1" if settings.estimate_radius_dedup_enabled else ""
     base_params: dict[str, Any] = {
         "rooms": rooms,
         "area_min": area_min,
@@ -3747,7 +3792,10 @@ def _fetch_analogs(
                         {_ANALOG_SELECT_COLS},
                         0.0 AS distance_m,
                         0.0 AS relevance_score,
-                        row_number() OVER (PARTITION BY address ORDER BY scraped_at DESC) AS rn_addr
+                        row_number() OVER (
+                            PARTITION BY address ORDER BY scraped_at DESC
+                        ) AS rn_addr,
+{_RN_DUP_WINDOW}
                     FROM listings
                     WHERE house_id_fk = CAST(:target_house_id AS bigint)
                     {_COMMON_WHERE}
@@ -3761,6 +3809,7 @@ def _fetch_analogs(
                     building_cadastral_number
                 FROM base
                 WHERE rn_addr <= :max_per_addr
+                {dup_filter}
                 ORDER BY scraped_at DESC
                 LIMIT 300
                 """
@@ -3798,7 +3847,10 @@ def _fetch_analogs(
                         {_ANALOG_SELECT_COLS},
                         0.0 AS distance_m,
                         0.0 AS relevance_score,
-                        row_number() OVER (PARTITION BY address ORDER BY scraped_at DESC) AS rn_addr
+                        row_number() OVER (
+                            PARTITION BY address ORDER BY scraped_at DESC
+                        ) AS rn_addr,
+{_RN_DUP_WINDOW}
                     FROM listings
                     WHERE address ILIKE :short_addr_prefix
                     {_COMMON_WHERE}
@@ -3812,6 +3864,7 @@ def _fetch_analogs(
                     building_cadastral_number
                 FROM base
                 WHERE rn_addr <= :max_per_addr
+                {dup_filter}
                 ORDER BY scraped_at DESC
                 LIMIT 300
                 """
@@ -3881,7 +3934,8 @@ def _fetch_analogs(
                                     ELSE 0
                                   END
                             )
-                        ) AS rn_addr
+                        ) AS rn_addr,
+{_RN_DUP_WINDOW}
                     FROM listings
                     WHERE ST_DWithin(geom::geography, ST_MakePoint(:lon, :lat)::geography, :radius)
                       AND (geo_precision IS DISTINCT FROM 'city')
@@ -3900,6 +3954,7 @@ def _fetch_analogs(
                     building_cadastral_number
                 FROM base
                 WHERE rn_addr <= :max_per_addr
+                {dup_filter}
                 ORDER BY relevance_score
                 LIMIT 300
                 """
@@ -3944,7 +3999,7 @@ def _fetch_analogs(
     tier_w_rows = (
         db.execute(
             text(
-                """
+                f"""
             WITH base AS (
                 SELECT
                     source, source_url, address, lat, lon,
@@ -3994,7 +4049,8 @@ def _fetch_analogs(
                                 ELSE 0
                               END
                         )
-                    ) AS rn_addr
+                    ) AS rn_addr,
+{_RN_DUP_WINDOW}
                 FROM listings
                 WHERE ST_DWithin(geom::geography, ST_MakePoint(:lon, :lat)::geography, :radius)
                   AND (geo_precision IS DISTINCT FROM 'city')
@@ -4036,6 +4092,7 @@ def _fetch_analogs(
                 relevance_score
             FROM base
             WHERE rn_addr <= :max_per_addr
+            {dup_filter}
             ORDER BY relevance_score
             LIMIT 300
             """
@@ -4294,6 +4351,15 @@ def _enforce_zero_analog_low(
     return confidence, explanation or ""
 
 
+def _downgrade_confidence(confidence: str) -> str:
+    """#1871 P2 — понизить confidence на одну ступень: high→medium→low→low.
+
+    Зеркалит inline downgrade-map в _compute_confidence (concentration bias) и
+    coarse-geo gate. Неизвестные значения возвращаются как есть (defensive).
+    """
+    return {"high": "medium", "medium": "low", "low": "low"}.get(confidence, confidence)
+
+
 def _compute_confidence(
     n_analogs: int,
     median_ppm2: float,

tradein-mvp/backend/tests/test_estimator_radius_dedup_1871.py

@@ -0,0 +1,212 @@
+"""#1871 P2 — radius-tier (source, source_id) dedup в _fetch_analogs.
+
+Radius-путь кэпит только per-address (rn_addr <= MAX_ANALOGS_PER_ADDRESS), но
+(source, source_id)-дубли делят один address и выживают на разных rn_addr рангах →
+раздувают n_analogs. Anchor-путь дедупит по (source, source_id); radius — нет.
+
+Эти тесты проверяют SQL-фрагмент (rn_dup-окно всегда; `AND rn_dup = 1` фильтр за
+флагом) во ВСЕХ четырёх тирах (S-canonical, S-fallback, H, W) + дедупную семантику
+окна на уровне Python-симуляции (freshest scraped_at, namespaced ключ, NULL-guard).
+Полный radius-путь требует PostGIS+БД, поэтому SQL проверяется на сгенерированном
+тексте (как test_estimator_price_trend_dedup.py).
+"""
+
+from __future__ import annotations
+
+import os
+from unittest.mock import MagicMock, patch
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+import pytest
+
+import app.services.estimator as est
+
+
+def _capture_tier_sql(*, dedup_enabled: bool) -> list[str]:
+    """Прогоняет _fetch_analogs с mock db, возвращает SQL-текст каждого тира.
+
+    target_house_id + short_addr + year/floors заданы так, чтобы SQL ВСЕХ четырёх
+    тиров отрендерился (каждый тир возвращает [] → fallthrough к следующему).
+    """
+    captured: list[str] = []
+    db = MagicMock()
+
+    def side_effect(*args, **kwargs):  # type: ignore[no-untyped-def]
+        captured.append(str(args[0].text))
+        result = MagicMock()
+        result.mappings.return_value.all.return_value = []
+        return result
+
+    db.execute.side_effect = side_effect
+
+    with patch.object(est.settings, "estimate_radius_dedup_enabled", dedup_enabled):
+        est._fetch_analogs(
+            db,
+            lat=56.83,
+            lon=60.6,
+            rooms=2,
+            area=50.0,
+            radius_m=2000,
+            full_address="г Екатеринбург, ул Малышева, д 30",
+            year_built=2010,
+            house_type="монолит",
+            total_floors=20,
+            target_house_id=123,
+        )
+    return captured
+
+
+# ---------------------------------------------------------------------------
+# Флаг по умолчанию
+# ---------------------------------------------------------------------------
+
+
+def test_radius_dedup_flag_default_true() -> None:
+    from app.core.config import settings
+
+    assert settings.estimate_radius_dedup_enabled is True
+
+
+# ---------------------------------------------------------------------------
+# SQL-фрагмент во всех тирах
+# ---------------------------------------------------------------------------
+
+
+def test_all_four_tiers_render() -> None:
+    sqls = _capture_tier_sql(dedup_enabled=True)
+    assert len(sqls) == 4, "ожидаем S-canonical, S-fallback, H, W"
+
+
+def test_rn_dup_window_present_in_every_tier_when_enabled() -> None:
+    sqls = _capture_tier_sql(dedup_enabled=True)
+    for i, sql in enumerate(sqls):
+        assert "AS rn_dup" in sql, f"tier#{i} без rn_dup-окна"
+        assert "rn_dup = 1" in sql, f"tier#{i} без `AND rn_dup = 1` фильтра"
+
+
+def test_dup_filter_dropped_when_disabled_but_window_stays() -> None:
+    """Флаг OFF: фильтр `AND rn_dup = 1` убран, окно остаётся (безвредно)."""
+    sqls = _capture_tier_sql(dedup_enabled=False)
+    assert len(sqls) == 4
+    for i, sql in enumerate(sqls):
+        assert "AS rn_dup" in sql, f"tier#{i}: окно должно оставаться даже при OFF"
+        assert "rn_dup = 1" not in sql, f"tier#{i}: фильтр должен исчезнуть при OFF"
+
+
+def test_rn_dup_key_namespaced_and_freshest() -> None:
+    """Ключ окна namespaced (id:/url:/ctid:) + ORDER BY scraped_at DESC."""
+    sql = _capture_tier_sql(dedup_enabled=True)[0]
+    assert "PARTITION BY source" in sql
+    assert "'id:' || source_id::text" in sql
+    assert "'url:' || source_url" in sql
+    assert "'ctid:' || ctid::text" in sql
+    # freshest: окно сортируется по scraped_at DESC → rn_dup=1 это свежайшая строка
+    win_start = sql.find("AS rn_dup")
+    win_region = sql[max(0, win_start - 400) : win_start]
+    assert "ORDER BY scraped_at DESC" in win_region
+
+
+# ---------------------------------------------------------------------------
+# psycopg3-инвариант: column::type OK, :bind::type запрещён
+# ---------------------------------------------------------------------------
+
+
+def test_no_bind_param_double_colon_cast() -> None:
+    """В сгенерированном SQL нет :name::type (только column::type, что легально)."""
+    import re
+
+    bad = re.compile(r":[a-z_]+::[a-z]")
+    for i, sql in enumerate(_capture_tier_sql(dedup_enabled=True)):
+        assert not bad.search(sql), f"tier#{i}: найден запрещённый :bind::type"
+
+
+# ---------------------------------------------------------------------------
+# Семантика дедупа окна (Python-симуляция row_number() OVER PARTITION/ORDER)
+# ---------------------------------------------------------------------------
+
+
+def _simulate_rn_dup(rows: list[dict]) -> list[dict]:
+    """Эмулирует SQL-окно rn_dup=1: per (source, key) freshest scraped_at.
+
+    Зеркалит SQL: ключ = source + namespaced(id:source_id | url:source_url |
+    ctid:ctid). NULL source_id+source_url → разные ctid → разные группы (обе живут).
+    """
+    groups: dict[tuple, dict] = {}
+    for r in rows:
+        if r.get("source_id") is not None:
+            key = (r["source"], "id", r["source_id"])
+        elif r.get("source_url") is not None:
+            key = (r["source"], "url", r["source_url"])
+        else:
+            key = (r["source"], "ctid", r["ctid"])
+        cur = groups.get(key)
+        if cur is None or r["scraped_at"] > cur["scraped_at"]:
+            groups[key] = r
+    return list(groups.values())
+
+
+def test_same_source_source_id_collapses_to_freshest() -> None:
+    rows = [
+        {"source": "yandex", "source_id": "A1", "source_url": None, "ctid": "(0,1)",
+         "scraped_at": 100, "price_per_m2": 150_000},
+        {"source": "yandex", "source_id": "A1", "source_url": None, "ctid": "(0,2)",
+         "scraped_at": 200, "price_per_m2": 151_000},
+    ]
+    out = _simulate_rn_dup(rows)
+    assert len(out) == 1
+    assert out[0]["scraped_at"] == 200, "должна выжить freshest (scraped_at=200)"
+
+
+def test_different_source_id_both_survive() -> None:
+    rows = [
+        {"source": "cian", "source_id": "X", "source_url": None, "ctid": "(0,1)",
+         "scraped_at": 100, "price_per_m2": 140_000},
+        {"source": "cian", "source_id": "Y", "source_url": None, "ctid": "(0,2)",
+         "scraped_at": 100, "price_per_m2": 145_000},
+    ]
+    out = _simulate_rn_dup(rows)
+    assert len(out) == 2
+
+
+def test_same_source_id_different_source_both_survive() -> None:
+    """(source, source_id) — PARTITION включает source: одинаковый id у разных
+    источников НЕ схлопывается."""
+    rows = [
+        {"source": "cian", "source_id": "Z", "source_url": None, "ctid": "(0,1)",
+         "scraped_at": 100, "price_per_m2": 140_000},
+        {"source": "yandex", "source_id": "Z", "source_url": None, "ctid": "(0,2)",
+         "scraped_at": 100, "price_per_m2": 145_000},
+    ]
+    out = _simulate_rn_dup(rows)
+    assert len(out) == 2
+
+
+def test_null_source_id_null_url_both_survive_via_ctid() -> None:
+    """CASE-guard: 2 строки с NULL source_id И NULL source_url → ctid-tiebreak →
+    обе сохранены (не схлопываются по NULL)."""
+    rows = [
+        {"source": "rosreestr", "source_id": None, "source_url": None, "ctid": "(0,1)",
+         "scraped_at": 100, "price_per_m2": 130_000},
+        {"source": "rosreestr", "source_id": None, "source_url": None, "ctid": "(0,2)",
+         "scraped_at": 100, "price_per_m2": 132_000},
+    ]
+    out = _simulate_rn_dup(rows)
+    assert len(out) == 2
+
+
+def test_null_source_id_falls_back_to_source_url() -> None:
+    """source_id NULL, но одинаковый source_url → схлопываются (url-ключ)."""
+    rows = [
+        {"source": "avito", "source_id": None, "source_url": "http://a/1", "ctid": "(0,1)",
+         "scraped_at": 100, "price_per_m2": 120_000},
+        {"source": "avito", "source_id": None, "source_url": "http://a/1", "ctid": "(0,2)",
+         "scraped_at": 300, "price_per_m2": 121_000},
+    ]
+    out = _simulate_rn_dup(rows)
+    assert len(out) == 1
+    assert out[0]["scraped_at"] == 300
+
+
+if __name__ == "__main__":  # pragma: no cover
+    raise SystemExit(pytest.main([__file__, "-q"]))

tradein-mvp/backend/tests/test_estimator_split_corridor_1871.py

@@ -0,0 +1,145 @@
+"""#1871 P2 — split-дома wide-corridor disclosure (за флагом, default OFF).
+
+Tier A (same-building) матчит по address-regex (намеренно НЕ house_id — дом дробится
+на несколько house_id). На split-доме разной этажности comp_min..comp_max растягивается
+через несколько ценовых режимов → коридор range_low/range_high доходит до 148%/170%.
+Коридор честно широкий (НЕ сужаем); disclosure только понижает confidence на ступень и
+дописывает объяснение, НЕ трогая point/median/range.
+
+Полный estimate-путь застаблен через harness из test_same_building_anchor.py
+(_run_estimate). Плюс unit на helper _downgrade_confidence.
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import os
+from pathlib import Path
+from typing import Any
+from unittest.mock import patch
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
+
+import pytest
+
+from app.core.config import settings
+from app.services.estimator import _downgrade_confidence
+
+# Переиспользуем harness _run_estimate / _make_payload из anchor-теста (полный
+# estimate путь со всеми I/O застабленными).
+_ANCHOR_TEST = Path(__file__).parent / "test_same_building_anchor.py"
+_spec = importlib.util.spec_from_file_location("_anchor_harness", _ANCHOR_TEST)
+assert _spec is not None and _spec.loader is not None
+_h = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(_h)
+
+
+# Split-дом: низкая секция (130k) … видовая высокая секция (360k) → коридор > 60%.
+_WIDE_SPLIT_COMPS: list[dict[str, Any]] = [
+    {"price_per_m2": 130_000, "area_m2": 60.0, "rooms": 2},
+    {"price_per_m2": 150_000, "area_m2": 62.0, "rooms": 2},
+    {"price_per_m2": 240_000, "area_m2": 58.0, "rooms": 2},
+    {"price_per_m2": 360_000, "area_m2": 61.0, "rooms": 2},
+]
+
+
+def _payload():  # type: ignore[no-untyped-def]
+    return _h._make_payload(area=60.0, rooms=2)
+
+
+def _run(*, flag: bool, anchor_tier: str, comps=None, threshold: float = 0.6):  # type: ignore[no-untyped-def]
+    with (
+        patch.object(settings, "estimate_wide_corridor_disclosure_enabled", flag),
+        patch.object(settings, "estimate_wide_corridor_threshold", threshold),
+    ):
+        return _h._run_estimate(
+            anchor_comps=_WIDE_SPLIT_COMPS if comps is None else comps,
+            anchor_tier=anchor_tier,
+            payload=_payload(),
+        )
+
+
+_DISCLOSURE_MARKER = "разбит на секции"
+
+
+# ---------------------------------------------------------------------------
+# Флаг по умолчанию
+# ---------------------------------------------------------------------------
+
+
+def test_disclosure_flag_default_off() -> None:
+    assert settings.estimate_wide_corridor_disclosure_enabled is False
+
+
+def test_threshold_default_0_6() -> None:
+    assert settings.estimate_wide_corridor_threshold == 0.6
+
+
+# ---------------------------------------------------------------------------
+# Disclosure поведение
+# ---------------------------------------------------------------------------
+
+
+def test_tier_a_wide_corridor_flag_on_downgrades_and_discloses() -> None:
+    """Tier A + corridor_pct > threshold + флаг ON → confidence понижен + disclosure."""
+    on = _run(flag=True, anchor_tier="A")
+    pct = (on.range_high_rub - on.range_low_rub) / on.median_price_rub
+    assert pct > 0.6, "fixture должен давать коридор > 60%"
+    assert _DISCLOSURE_MARKER in (on.confidence_explanation or "")
+    # confidence понижен относительно flag-OFF baseline (medium → low).
+    off = _run(flag=False, anchor_tier="A")
+    assert _downgrade_confidence(off.confidence) == on.confidence
+    assert on.confidence != off.confidence
+
+
+def test_flag_off_no_change() -> None:
+    """Флаг OFF → ничего не делает (ни confidence, ни explanation)."""
+    off = _run(flag=False, anchor_tier="A")
+    assert _DISCLOSURE_MARKER not in (off.confidence_explanation or "")
+
+
+def test_does_not_touch_point_median_range() -> None:
+    """Disclosure НЕ трогает point/median/range — только confidence+explanation."""
+    off = _run(flag=False, anchor_tier="A")
+    on = _run(flag=True, anchor_tier="A")
+    assert on.median_price_rub == off.median_price_rub
+    assert on.range_low_rub == off.range_low_rub
+    assert on.range_high_rub == off.range_high_rub
+
+
+def test_below_threshold_no_disclosure() -> None:
+    """corridor_pct ниже порога → без изменений (порог поднят выше фактического)."""
+    high_threshold = _run(flag=True, anchor_tier="A", threshold=5.0)
+    assert _DISCLOSURE_MARKER not in (high_threshold.confidence_explanation or "")
+
+
+def test_non_tier_a_not_affected() -> None:
+    """Не-Tier-A (Tier C) → disclosure не применяется (даже с флагом ON)."""
+    c = _run(flag=True, anchor_tier="C")
+    assert _DISCLOSURE_MARKER not in (c.confidence_explanation or "")
+
+
+# ---------------------------------------------------------------------------
+# _downgrade_confidence helper
+# ---------------------------------------------------------------------------
+
+
+def test_downgrade_high_to_medium() -> None:
+    assert _downgrade_confidence("high") == "medium"
+
+
+def test_downgrade_medium_to_low() -> None:
+    assert _downgrade_confidence("medium") == "low"
+
+
+def test_downgrade_low_stays_low() -> None:
+    assert _downgrade_confidence("low") == "low"
+
+
+def test_downgrade_unknown_passthrough() -> None:
+    # defensive: незнакомое значение возвращается как есть
+    assert _downgrade_confidence("weird") == "weird"
+
+
+if __name__ == "__main__":  # pragma: no cover
+    raise SystemExit(pytest.main([__file__, "-q"]))