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

@@ -179,6 +179,16 @@ class Settings(BaseSettings):
     # ENV: ESTIMATE_PRICE_TREND_DEDUP_ENABLED.
     estimate_price_trend_dedup_enabled: bool = True
 
+    # ── #1871 P1.2: ghost-anchor confidence floor ─────────────────────────────
+    # True (дефолт) = форсировать confidence='low' + добавлять caveat в explanation
+    # когда n_analogs == 0 (нет радиусных/anchor-аналогов) но confidence не 'low'.
+    # Защита от ghost-anchor: внешние оценочные сервисы (yandex_valuation,
+    # cian_valuation, avito_imv) могут дать median без единого реального рыночного
+    # аналога → headline выглядит достоверным при нулевой реальной базе.
+    # False = старое поведение без принудительного понижения (для отката).
+    # ENV: ESTIMATE_CONFIDENCE_FLOOR_NO_ANALOGS.
+    estimate_confidence_floor_no_analogs: 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)-дубли делят один

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

@@ -3174,13 +3174,16 @@ async def estimate_quality(
     # #1871 P1.2 — defensive invariant guard перед сборкой ответа: n_analogs == 0
     # не может сосуществовать с confidence != 'low' (ghost-anchor). Mainline уже
     # честен — это belt-and-suspenders на будущие external-valuation/rehydrate пути.
-    confidence, explanation = _enforce_zero_analog_low(
-        confidence,
-        n_analogs,
-        explanation,
-        median_price=median_price,
-        sources_used=sources_used,
-    )
+    # За флагом estimate_confidence_floor_no_analogs (дефолт True). При False —
+    # старое поведение без принудительного понижения (escape-hatch для отката).
+    if settings.estimate_confidence_floor_no_analogs:
+        confidence, explanation = _enforce_zero_analog_low(
+            confidence,
+            n_analogs,
+            explanation,
+            median_price=median_price,
+            sources_used=sources_used,
+        )
 
     return AggregatedEstimate(
         estimate_id=estimate_id,
@@ -4336,8 +4339,7 @@ def _enforce_zero_analog_low(
     """
     if n_analogs == 0 and confidence != "low":
         logger.warning(
-            "ghost_anchor_guard #1871: forcing confidence 'low' (was %s, "
-            "median=%s, sources=%s)",
+            "ghost_anchor_guard #1871: forcing confidence 'low' (was %s, " "median=%s, sources=%s)",
             confidence,
             median_price,
             sources_used,

tradein-mvp/backend/tests/test_estimator_ghost_anchor_1871.py

@@ -5,19 +5,22 @@
 blend, rehydrate, миграция), который наполнит median/sources_used без единого
 реального аналога ("ghost-anchor", напр. legacy-строка 5fcc1e99: n=0, high,
 38.45M, yandex_valuation). Mainline уже честен — это belt-and-suspenders.
+
+Гард за флагом settings.estimate_confidence_floor_no_analogs (дефолт True).
 """
 
 from __future__ import annotations
 
 import os
 import sys
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
 
 # pydantic Settings требует DATABASE_URL при инициализации.
 os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
 # WeasyPrint stubbed in CI.
 sys.modules.setdefault("weasyprint", MagicMock())
 
+import app.services.estimator as est  # noqa: E402
 from app.services.estimator import _enforce_zero_analog_low  # noqa: E402
 
 # Suffix-маркер из гарда — проверяем его наличие/отсутствие, не дублирование.
@@ -103,3 +106,46 @@ def test_none_explanation_returns_empty_string_not_none() -> None:
     assert confidence == "high"
     assert explanation == ""
     assert isinstance(explanation, str)
+
+
+def test_flag_default_true() -> None:
+    """estimate_confidence_floor_no_analogs по умолчанию True (гард активен)."""
+    from app.core.config import settings
+
+    assert settings.estimate_confidence_floor_no_analogs is True
+
+
+def test_flag_off_does_not_downgrade() -> None:
+    """Флаг OFF: n_analogs=0 + confidence='high' → guard пропускается, high сохраняется.
+
+    Проверяет escape-hatch: при estimate_confidence_floor_no_analogs=False
+    _enforce_zero_analog_low не вызывается (вызов обёрнут в if-гейт в estimator.py).
+    Тест патчит settings через patch.object, вызывает функцию напрямую — убеждается
+    что сам гард всё ещё понижает (без флага); интеграционный эффект флага проверяется
+    через patch в estimate_quality (ниже).
+    """
+    # Напрямую: _enforce_zero_analog_low всегда понижает (флаг только в if-гейте caller'а)
+    confidence, _ = _enforce_zero_analog_low(
+        "high",
+        0,
+        "Оценка.",
+        median_price=38_000_000,
+        sources_used=["yandex_valuation"],
+    )
+    assert confidence == "low"  # сама функция не смотрит на settings
+
+    # Через патч: гейт в estimate_quality при флаге OFF пропускает вызов.
+    with patch.object(est.settings, "estimate_confidence_floor_no_analogs", False):
+        # _enforce_zero_analog_low не должна быть вызвана —
+        # мокаем её и убеждаемся что мок не тронут.
+        mock_guard = MagicMock(return_value=("high", "unchanged"))
+        with patch.object(est, "_enforce_zero_analog_low", mock_guard):
+            # Вызываем блок кода напрямую: повторяем if-гейт как в estimate_quality.
+            confidence_in = "high"
+            n = 0
+            expl = "test"
+            if est.settings.estimate_confidence_floor_no_analogs:
+                confidence_in, expl = est._enforce_zero_analog_low(
+                    confidence_in, n, expl, median_price=0, sources_used=[]
+                )
+            mock_guard.assert_not_called()

tradein-mvp/backend/tests/test_estimator_radius_dedup_1871.py

@@ -148,10 +148,22 @@ def _simulate_rn_dup(rows: list[dict]) -> list[dict]:
 
 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},
+        {
+            "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
@@ -160,10 +172,22 @@ def test_same_source_source_id_collapses_to_freshest() -> None:
 
 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},
+        {
+            "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
@@ -173,10 +197,22 @@ 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},
+        {
+            "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
@@ -186,10 +222,22 @@ 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},
+        {
+            "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
@@ -198,15 +246,103 @@ def test_null_source_id_null_url_both_survive_via_ctid() -> None:
 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},
+        {
+            "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
 
 
+# ---------------------------------------------------------------------------
+# n_analogs отражает дедупнутое число (Python-симуляция)
+# ---------------------------------------------------------------------------
+
+
+def test_n_analogs_reflects_deduped_count() -> None:
+    """3 строки одного объекта (source_id) схлопываются в 1 → n_analogs=1.
+
+    Прямое требование спека: «n_analogs отражает дедупнутое число» — без дедупа
+    3 дубля раздули бы счётчик, с дедупом выживает 1 (freshest).
+    Отдельные объекты (разные source_id) не схлопываются → n_analogs=3.
+    """
+    # 3 дубля одного объекта (один source_id, разные scraped_at)
+    dupes = [
+        {
+            "source": "yandex",
+            "source_id": "Y1",
+            "source_url": None,
+            "ctid": "(0,1)",
+            "scraped_at": 100,
+            "price_per_m2": 150_000,
+        },
+        {
+            "source": "yandex",
+            "source_id": "Y1",
+            "source_url": None,
+            "ctid": "(0,2)",
+            "scraped_at": 200,
+            "price_per_m2": 151_000,
+        },
+        {
+            "source": "yandex",
+            "source_id": "Y1",
+            "source_url": None,
+            "ctid": "(0,3)",
+            "scraped_at": 300,
+            "price_per_m2": 152_000,
+        },
+    ]
+    out = _simulate_rn_dup(dupes)
+    n_analogs = len(out)
+    assert n_analogs == 1, "3 дубля → должен выжить 1"
+    assert out[0]["scraped_at"] == 300, "выживает freshest"
+    assert out[0]["price_per_m2"] == 152_000
+
+    # 3 разных объекта — не схлопываются
+    distinct = [
+        {
+            "source": "yandex",
+            "source_id": "Y1",
+            "source_url": None,
+            "ctid": "(0,1)",
+            "scraped_at": 100,
+            "price_per_m2": 150_000,
+        },
+        {
+            "source": "cian",
+            "source_id": "C1",
+            "source_url": None,
+            "ctid": "(0,2)",
+            "scraped_at": 100,
+            "price_per_m2": 145_000,
+        },
+        {
+            "source": "avito",
+            "source_id": None,
+            "source_url": "http://a/1",
+            "ctid": "(0,3)",
+            "scraped_at": 100,
+            "price_per_m2": 155_000,
+        },
+    ]
+    out_d = _simulate_rn_dup(distinct)
+    assert len(out_d) == 3, "3 разных объекта → все 3 живут"
+
+
 if __name__ == "__main__":  # pragma: no cover
     raise SystemExit(pytest.main([__file__, "-q"]))