"""Trade-In Estimator — реальное SQL aggregation поверх listings + deals.

Заменяет старый _mock_estimate() из api/v1/trade_in.py.

Алгоритм:
1. Geocode address → (lat, lon)
2. SELECT listings с фильтрами:
   - PostGIS ST_DWithin (geom, point, 1000m) — радиус поиска
   - source ≠ avito (у Avito фейковые anchor-jitter координаты — не гео-аналог)
   - rooms = target_rooms (точное совпадение)
   - area_m2 BETWEEN target × 0.85 AND target × 1.15
   - scraped_at > NOW() - 14 days (свежие)
   - is_active = true
3. Tukey outlier filter (1.5 × IQR rule)
4. Median / Q1 / Q3 / count → confidence
5. То же для deals (period = 12 mo).
6. Сохранить в trade_in_estimates + вернуть AggregatedEstimate
"""

from __future__ import annotations

import asyncio
import hashlib
import json
import logging
import math
import re
import time
from datetime import UTC, datetime, timedelta
from typing import Any
from uuid import uuid4

from sqlalchemy import text
from sqlalchemy.orm import Session

from app.core.config import settings
from app.schemas.trade_in import (
    AggregatedEstimate,
    AnalogLot,
    AvitoImvSummary,
    CianValuationSummary,
    DkpCorridor,
    PriceTrendPoint,
    TradeInEstimateInput,
)
from app.services.dadata import DadataAddressResult
from app.services.dadata import clean_address as dadata_clean_address
from app.services.geocoder import GeocodeResult, geocode
from app.services.house_metadata import get_house_metadata
from app.services.matching.houses import match_house_readonly, match_or_create_house
from app.services.scrapers.avito_imv import (
    IMVAddressNotFoundError,
    IMVAuthError,
    IMVEvaluation,
    IMVTransientError,
    compute_imv_cache_key,
    evaluate_via_imv,
    save_imv_evaluation,
)
from app.services.scrapers.cian_valuation import (
    CianValuationResult,
    estimate_via_cian_valuation,
)
from app.services.scrapers.yandex_valuation import (
    YandexValuationResult,
    YandexValuationScraper,
)

logger = logging.getLogger(__name__)


# ── Constants ────────────────────────────────────────────────────────────────
DEFAULT_RADIUS_M = 1000  # ПО ВСТРЕЧЕ ПТИЦЫ: «локация не дальше 800-1000 м»
FALLBACK_RADIUS_M = 2000
AREA_TOLERANCE = 0.15  # ±15% площади
MAX_ANALOGS_PER_ADDRESS = 5  # анти-bias: не больше 5 лотов с одного адреса
MIN_ANALOGS_PER_SOURCE = 5  # гарантированный минимум на live source
LISTINGS_FRESH_DAYS = 14  # объявления не старше 14 дней
DEALS_PERIOD_MONTHS = 12  # сделки за последний год

# #699: санитизация ДКП-выбросов (Росреестр `deals`). В сырых сделках встречаются
# нерыночные/битые записи — доли, сделки с обременением, опечатки этажа/площади —
# которые шумят actual_deals (display) и dkp_corridor/expected_sold. Абсолютные
# guard-bands (НЕ относительные) для ЕКБ-вторички: рынок ~100–400 К/м² (ср. пороги
# _band_haircut 180/350К), премиум до ~680К. Нижняя/верхняя границы заведомо вне
# рынка — режут «4.98 М за 125 м²» = 39.7 К/м² и т.п. Этаж: ЕКБ-максимум ~52 эт.
DEAL_MIN_PPM2 = 50_000  # ниже = не arms-length (доля/обременение/ошибка)
DEAL_MAX_PPM2 = 800_000  # выше премиума → опечатка/коммерция
DEAL_MAX_FLOOR = 60  # выше реального максимума ЕКБ → битый этаж (напр. floor:100)

# Когорта по году постройки — типизация массовой застройки РФ.
# Используется как hard-filter в Tier 0 _fetch_analogs (PR 9, 2026-05-24).
# Если target_year не задан — cohort = None → фильтр отключён, Tier 0 пропускается.
COHORTS = (
    # (cohort_name, year_min_inclusive, year_max_inclusive)
    ("khrushchev", 1955, 1969),  # Хрущёвки 5-эт
    ("brezhnev", 1970, 1989),  # Брежневка кирпич/панель 9–12-эт
    (
        "late_soviet",
        1990,
        1999,
    ),  # Поздний СССР (no overlap; first-match would never pick old range)
    ("2000s", 2000, 2010),  # Ранние новостройки
    ("modern", 2011, 2100),  # Современные ЖК
)
# Минимум аналогов чтобы остаться на Tier 0 (с cohort); ниже — fallback на Tier A.
MIN_ANALOGS_TIER_0 = 5


def _target_cohort_range(year_built: int | None) -> tuple[int, int] | None:
    """Maps a target year to its cohort year range [min, max] inclusive.

    Returns None if year_built is None — caller will skip cohort filter.
    Picks the FIRST matching cohort (so 1988 → 'brezhnev', not 'late_soviet').
    """
    if year_built is None:
        return None
    for _name, ymin, ymax in COHORTS:
        if ymin <= year_built <= ymax:
            return (ymin, ymax)
    # Out-of-range год (например, 1900 или 2050) — cohort фильтр не применяем,
    # лучше показать что есть в радиусе, чем 0 результатов.
    return None


# Маппинг наших house_type → словарь Avito-IMV (внешний source). НЕ путать с
# _REPAIR_COEF (heuristic-множитель ниже).
_IMV_HOUSE_TYPE_MAP: dict[str | None, str | None] = {
    "panel": "panel",
    "brick": "brick",
    "monolith": "monolith",
    "monolith_brick": "monolith_brick",
    "monolithic": "monolith",
    "block": "block",
    "wood": "wood",
    None: None,
}
_IMV_REPAIR_MAP: dict[str | None, str | None] = {
    "needs_repair": "required",
    "standard": "cosmetic",
    "good": "euro",
    "excellent": "designer",
    None: None,
}

# Множители к медиане по состоянию ремонта. Аналоги в выборке — микс состояний;
# коэффициент сдвигает оценку под ремонт целевой квартиры (встреча Птицы: ремонт
# влияет на цену).
#
# WARNING: tunable МАРКЕТ-ЭВРИСТИКА, НЕ data-derived (issue #7). Вывести из данных пока
# нельзя: listings.repair_state покрыт только ~2% (coverage вырастет после #621 backfill),
# а медианы по нему confounded by area (немонотонны). Baseline = standard = 1.00 (no-op:
# было 0.98, срезало каждую «стандартную» оценку на 2% — пофикшено). Пересмотреть при
# coverage > 20% и наборе достаточной выборки по каждому bucket-у (#7).
# После #621: repair_state нормализован → needs_repair/standard/good/excellent на инgesте.
_REPAIR_COEF: dict[str, float] = {
    "needs_repair": 0.94,  # требует ремонта — ниже рынка
    "standard": 1.00,  # baseline
    "good": 1.05,
    "excellent": 1.10,  # евроремонт — выше рынка
}
_REPAIR_LABEL: dict[str | None, str] = {
    "needs_repair": "требует ремонта",
    "standard": "стандартный ремонт",
    "good": "хороший ремонт",
    "excellent": "евроремонт",
}


def _repair_coefficient(repair_state: str | None) -> float:
    """Множитель к медиане по состоянию ремонта. None → 1.0 (без поправки)."""
    if not repair_state:
        return 1.0
    return _REPAIR_COEF.get(repair_state, 1.0)


# ── Asking→sold correction ratio lookup (#648 Stage 3) ──────────────────────
# Таблица asking_to_sold_ratios (migration 080) хранит per-rooms коэффициент
# ratio = median(SOLD ppm²) / median(ASKING ppm²) (~0.72–0.93). Estimator
# домножает ASKING-медиану на этот ratio, получая параллельную expected_sold
# цену (релевантную для выкупа). Headline asking-медиана НЕ меняется.
#
# Кэш: tiny in-process dict {bucket: (ratio, basis, fetched_monotonic)} с TTL.
# Ratio дрейфует медленно (refresh-задача раз в сутки, Stage 4), поэтому 300с
# TTL более чем достаточно и снимает по SELECT'у с каждой оценки. Single-worker
# uvicorn/scheduler — GIL делает dict-доступ atomic enough (без явного lock).
_ASKING_SOLD_RATIO_CACHE_TTL_S = 300.0
_asking_sold_ratio_cache: dict[int, tuple[float | None, str | None, float]] = {}


def _get_asking_sold_ratio(db: Session, rooms: int | None) -> tuple[float | None, str | None]:
    """Возвращает (ratio, basis) asking→sold для бакета комнат.

    bucket = min(max(rooms or 0, 0), 4). Сначала ищем per-rooms строку
    (district=''), при отсутствии — global fallback (rooms_bucket=-1). Если
    таблицы нет / пуста / любая ошибка → (None, None), НЕ raise (graceful:
    estimator продолжает без sold-коррекции, headline asking-медиана отдаётся).

    Кэшируется на бакет с TTL _ASKING_SOLD_RATIO_CACHE_TTL_S.
    """
    bucket = min(max(rooms or 0, 0), 4)

    cached = _asking_sold_ratio_cache.get(bucket)
    if cached is not None:
        ratio, basis, fetched = cached
        if (time.monotonic() - fetched) < _ASKING_SOLD_RATIO_CACHE_TTL_S:
            return ratio, basis

    ratio: float | None = None
    basis: str | None = None
    try:
        row = db.execute(
            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 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 (как в sibling-helper'ах _get_or_fetch_*): неудачный
        # SELECT помечает транзакцию InFailedSqlTransaction, и без отката следующий
        # statement (_fetch_deals) упал бы → 500. Откат держит shared session чистой
        # для последующего INSERT. rollback тоже guard'им (соединение могло умереть).
        logger.debug("asking_to_sold_ratio lookup skipped (graceful): %s", exc)
        try:
            db.rollback()
        except Exception:
            pass
        ratio, basis = None, None

    _asking_sold_ratio_cache[bucket] = (ratio, basis, time.monotonic())
    return ratio, basis


# ── Avito IMV cache lookup (Stage 3) ────────────────────────────────────────
IMV_CACHE_TTL_HOURS = 24

# Префиксы в адресе, которые Avito-геокодер не распознаёт (не жилые назначения).
# Пример: "Склад, ул. Заводская, д. 44-а" → "ул. Заводская, д. 44-а"
_NOISE_PREFIX_RE = re.compile(
    r"(Склад|Гараж|Подсобка|Нежилое|Помещение|Цех),\s*",
    flags=re.IGNORECASE,
)

YANDEX_VALUATION_CACHE_TTL_HOURS = 24
YANDEX_VALUATION_DEFAULT_CATEGORY = "APARTMENT"
YANDEX_VALUATION_DEFAULT_TYPE = "SELL"


async def _get_or_fetch_imv_cached(
    db: Session,
    *,
    address: str,
    rooms: int,
    area_m2: float,
    floor: int,
    floor_at_home: int,
    house_type: str,
    renovation_type: str,
    has_balcony: bool,
    has_loggia: bool,
    estimate_id_for_link: Any = None,
) -> IMVEvaluation | None:
    """Cached IMV lookup. TTL 24h по cache_key (sha256 of address + params).

    1. compute cache_key
    2. SELECT из avito_imv_evaluations WHERE cache_key = :ck AND fetched_at > NOW() - 24h
    3. Если hit → возвращаем reconstructed IMVEvaluation
    4. Cache miss → call evaluate_via_imv, save_imv_evaluation, return

    Graceful: на любой error возвращаем None (estimator продолжает без IMV).
    """
    try:
        cache_key = compute_imv_cache_key(
            address,
            rooms,
            area_m2,
            floor,
            floor_at_home,
            house_type,
            renovation_type,
            has_balcony,
            has_loggia,
        )

        existing = (
            db.execute(
                text(
                    """
                SELECT id, cache_key, address, rooms, area_m2, floor, floor_at_home,
                       house_type, renovation_type, has_balcony, has_loggia,
                       lat, lon, geo_hash, avito_address_id, avito_location_id,
                       avito_metro_id, avito_district_id,
                       recommended_price, lower_price, higher_price, market_count,
                       raw_response, fetched_at
                FROM avito_imv_evaluations
                WHERE cache_key = :ck
                  AND fetched_at > NOW() - (:ttl_hours || ' hours')::interval
                ORDER BY fetched_at DESC
                LIMIT 1
                """
                ),
                {"ck": cache_key, "ttl_hours": IMV_CACHE_TTL_HOURS},
            )
            .mappings()
            .first()
        )

        if existing is not None:
            logger.info(
                "imv: cache HIT key=%s recommended=%d",
                cache_key[:8],
                existing["recommended_price"],
            )
            from app.services.scrapers.avito_imv import IMVGeo

            return IMVEvaluation(
                cache_key=existing["cache_key"],
                address=existing["address"],
                rooms=existing["rooms"],
                area_m2=float(existing["area_m2"]),
                floor=existing["floor"],
                floor_at_home=existing["floor_at_home"],
                house_type=existing["house_type"],
                renovation_type=existing["renovation_type"],
                has_balcony=existing["has_balcony"],
                has_loggia=existing["has_loggia"],
                geo=IMVGeo(
                    geo_hash=existing["geo_hash"] or "",
                    lat=existing["lat"],
                    lon=existing["lon"],
                    avito_address_id=existing["avito_address_id"],
                    avito_location_id=existing["avito_location_id"],
                    avito_metro_id=existing["avito_metro_id"],
                    avito_district_id=existing["avito_district_id"],
                ),
                recommended_price=existing["recommended_price"],
                lower_price=existing["lower_price"],
                higher_price=existing["higher_price"],
                market_count=existing["market_count"],
                raw_response=existing.get("raw_response"),
            )

        # Cache miss — fresh fetch
        logger.info("imv: cache MISS key=%s — fetching fresh", cache_key[:8])
        result = await evaluate_via_imv(
            address=address,
            rooms=rooms,
            area_m2=area_m2,
            floor=floor,
            floor_at_home=floor_at_home,
            house_type=house_type,
            renovation_type=renovation_type,
            has_balcony=has_balcony,
            has_loggia=has_loggia,
        )
        save_imv_evaluation(db, result, estimate_id=estimate_id_for_link)
        logger.info(
            "imv: fresh recommended=%d range=(%d, %d) count=%d",
            result.recommended_price,
            result.lower_price,
            result.higher_price,
            result.market_count or 0,
        )
        return result

    except IMVAddressNotFoundError as e:
        logger.warning("imv: address not found in Avito geocoder: %s", e)
        # Retry once with noise prefixes stripped (e.g. "Склад, ул. X" → "ул. X")
        cleaned = _NOISE_PREFIX_RE.sub("", address)
        if cleaned != address:
            logger.info(
                "imv: retry with cleaned address %r → %r",
                address[:60],
                cleaned[:60],
            )
            try:
                result = await evaluate_via_imv(
                    address=cleaned,
                    rooms=rooms,
                    area_m2=area_m2,
                    floor=floor,
                    floor_at_home=floor_at_home,
                    house_type=house_type,
                    renovation_type=renovation_type,
                    has_balcony=has_balcony,
                    has_loggia=has_loggia,
                )
                save_imv_evaluation(db, result, estimate_id=estimate_id_for_link)
                logger.info(
                    "imv: retry OK recommended=%d range=(%d, %d) count=%d",
                    result.recommended_price,
                    result.lower_price,
                    result.higher_price,
                    result.market_count or 0,
                )
                return result
            except IMVAddressNotFoundError:
                logger.warning("imv: cleaned address also not found — giving up")
            except Exception as retry_exc:
                logger.warning("imv: retry failed: %s", retry_exc)
        return None
    except IMVAuthError as e:
        logger.error(
            "imv: auth/quota error — manual action required: %s",
            e,
        )
        return None
    except IMVTransientError as e:
        logger.warning("imv: transient error, skipping retry in estimator context: %s", e)
        return None
    except Exception as e:
        logger.warning("imv: fetch failed — estimator продолжает без IMV: %s", e)
        return None


# ── Yandex Valuation cache lookup (Stage 8) ─────────────────────────────────


def _yandex_valuation_cache_key(address: str, offer_category: str, offer_type: str) -> str:
    """SHA256 cache key for Yandex Valuation lookups."""
    payload = f"{address}|{offer_category}|{offer_type}"
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()


async def _get_or_fetch_yandex_valuation_cached(
    db: Session,
    *,
    address: str,
    offer_category: str = YANDEX_VALUATION_DEFAULT_CATEGORY,
    offer_type: str = YANDEX_VALUATION_DEFAULT_TYPE,
) -> YandexValuationResult | None:
    """Cached Yandex Valuation lookup. TTL 24h via external_valuations table.

    Returns None on any error / cache miss + fetch failure — caller continues
    without Yandex enrichment (graceful degradation).
    """
    cache_key = _yandex_valuation_cache_key(address, offer_category, offer_type)

    # Cache lookup
    try:
        cached = (
            db.execute(
                text(
                    """
                SELECT raw_payload, fetched_at
                FROM external_valuations
                WHERE source = 'yandex_valuation'
                  AND cache_key = :ck
                  AND expires_at > NOW()
                ORDER BY fetched_at DESC
                LIMIT 1
                """
                ),
                {"ck": cache_key},
            )
            .mappings()
            .first()
        )
    except Exception as e:
        logger.warning("yandex_valuation: cache lookup failed: %s", e)
        cached = None

    if cached is not None and cached.get("raw_payload"):
        try:
            payload_dict = (
                cached["raw_payload"]
                if isinstance(cached["raw_payload"], dict)
                else json.loads(cached["raw_payload"])
            )
            logger.info(
                "yandex_valuation: cache HIT key=%s items=%d",
                cache_key[:8],
                len(payload_dict.get("history_items", [])),
            )
            return YandexValuationResult.model_validate(payload_dict)
        except Exception as e:
            logger.warning("yandex_valuation: cache deserialize failed — refetching: %s", e)

    # Fresh fetch
    try:
        async with YandexValuationScraper() as scraper:
            result = await scraper.fetch_house_history(
                address=address,
                offer_category=offer_category,
                offer_type=offer_type,
            )
    except Exception as e:
        logger.warning("yandex_valuation: fetch failed — estimator продолжает без Yandex: %s", e)
        return None

    if result is None:
        logger.info("yandex_valuation: empty result for address=%s", address[:60])
        return None

    # Save to cache (UPSERT on (source, cache_key))
    try:
        db.execute(
            text(
                """
                INSERT INTO external_valuations (
                    source, cache_key, address,
                    raw_payload,
                    fetched_at, expires_at
                ) VALUES (
                    'yandex_valuation', :ck, :addr,
                    CAST(:payload AS jsonb),
                    NOW(), NOW() + (:ttl_hours || ' hours')::interval
                )
                ON CONFLICT (source, cache_key) DO UPDATE
                    SET raw_payload = EXCLUDED.raw_payload,
                        fetched_at = NOW(),
                        expires_at = NOW() + (:ttl_hours || ' hours')::interval
                """
            ),
            {
                "ck": cache_key,
                "addr": address,
                "payload": json.dumps(result.model_dump(mode="json"), ensure_ascii=False),
                "ttl_hours": YANDEX_VALUATION_CACHE_TTL_HOURS,
            },
        )
        db.commit()
        logger.info(
            "yandex_valuation: fresh fetch saved key=%s items=%d",
            cache_key[:8],
            len(result.history_items),
        )
    except Exception as e:
        logger.warning("yandex_valuation: cache save failed (continuing): %s", e)
        db.rollback()

    return result


def _save_yandex_history_items(
    db: Session,
    result: YandexValuationResult,
) -> int:
    """Persist history items to house_placement_history. Returns saved count.

    Resolves house_id ONCE per result via match_or_create_house() using the
    valuation page's address + meta (year_built/total_floors). All items from
    the same page share that house_id.

    Confidence pipeline:
      method_confidence  <- match_or_create_house (1.0 cadastr/source, 0.9 fp, 0.7 geo, 1.0 new)
      final_confidence   = method_confidence

    Idempotent via UNIQUE (source, ext_item_id); ext_item_id synthesized from
    (address|publish_date|area|floor|prices) hash.

    Batch semantics: single try/except; on any failure the batch rolls back.
    """
    if not result.history_items:
        return 0

    # Resolve house ONCE per page. Synthetic ext_id = sha256(address)[:16]
    # — stable across re-runs, distinguishes pages for different addresses.
    address_seed = (result.address or "").strip().lower()
    house_ext_id = (
        hashlib.sha256(address_seed.encode("utf-8")).hexdigest()[:16] if address_seed else "unknown"
    )

    try:
        house_id, method_confidence, method = match_or_create_house(
            db,
            ext_source="yandex_valuation",
            ext_id=house_ext_id,
            address=result.address,
            year_built=result.house.year_built,
        )
    except Exception as e:
        logger.warning(
            "yandex_valuation: house resolution failed for address=%r: %s"
            " — saving with house_id=NULL",
            result.address,
            e,
        )
        db.rollback()
        house_id = None
        method_confidence = 0.0
        method = "fail"

    logger.info(
        "yandex_valuation: house resolved house_id=%s method=%s confidence=%.2f addr=%r",
        house_id,
        method,
        method_confidence,
        result.address,
    )

    rows = []
    for item in result.history_items:
        ext_seed = (
            f"{result.address}|{item.publish_date}|{item.area_m2}|{item.floor}|"
            f"{item.start_price}|{item.last_price}"
        )
        ext_item_id = hashlib.sha256(ext_seed.encode("utf-8")).hexdigest()[:32]
        rows.append(
            {
                "ext_id": ext_item_id,
                "house_id": house_id,
                "rooms": item.rooms,
                "area": item.area_m2,
                "floor": item.floor,
                "total_floors": result.house.total_floors,
                "start_price": item.start_price,
                "last_price": item.last_price,
                "publish_date": item.publish_date,
                "removed_date": item.removed_date,
                "exposure": item.exposure_days,
                "confidence": float(method_confidence),
                "notes": f"match_method={method}" if method != "fail" else None,
                "raw": json.dumps(item.model_dump(mode="json"), ensure_ascii=False),
            }
        )

    sql = text(
        """
        INSERT INTO house_placement_history (
            source, ext_item_id, house_id,
            rooms, area_m2, floor, total_floors,
            start_price, start_price_date,
            last_price, last_price_date,
            removed_date,
            exposure_days,
            source_confidence, notes,
            raw_payload
        ) VALUES (
            'yandex_valuation', :ext_id, :house_id,
            :rooms, :area, :floor, :total_floors,
            :start_price, :publish_date,
            :last_price, :publish_date,
            :removed_date,
            :exposure,
            :confidence, :notes,
            CAST(:raw AS jsonb)
        )
        ON CONFLICT (source, ext_item_id) DO NOTHING
        """
    )

    try:
        for row in rows:
            db.execute(sql, row)
        db.commit()
        return len(rows)
    except Exception as e:
        logger.warning(
            "yandex_valuation: failed to save history batch (%d items): %s",
            len(rows),
            e,
        )
        db.rollback()
        return 0


# ── #651: IMV / Yandex blend (killer accuracy fix) ─────────────────────────────


def _fetch_house_imv_anchor(
    db: Session,
    *,
    target_house_id: int | None,
    rooms: int | None,
    area: float | None,
) -> dict[str, Any] | None:
    """Достаёт РЕАЛЬНУЮ Avito IMV-оценку target-дома из `house_imv_evaluations`.

    В отличие от `avito_imv_evaluations` (keyed estimate_id — пустая, on-demand
    скрейп), `house_imv_evaluations` популирована (~2951 домов, fresh) и keyed по
    house_id. Резолвим строку: WHERE house_id = target_house_id, предпочитаем
    запись с ближайшими rooms+area (минимизируем |Δrooms|*10 + |Δarea%|), иначе
    самую свежую (fetched_at DESC). Best-effort: None при любой ошибке / отсутствии
    house_id / пустой таблице — estimator продолжает на гео-tier'ах (no regress).

    Returns dict {recommended_price, lower_price, higher_price, market_count,
    rooms, area_m2} или None.
    """
    if target_house_id is None:
        return None
    try:
        row = (
            db.execute(
                text(
                    """
                    SELECT recommended_price, lower_price, higher_price,
                           market_count, rooms, area_m2
                    FROM house_imv_evaluations
                    WHERE house_id = CAST(:hid AS bigint)
                      AND recommended_price > 0
                      -- Band-guard: строка пригодна как anchor только при правдоподобном
                      -- совпадении rooms/area. Иначе studio-only IMV запись «прилипала»
                      -- к 3-комн. target'у (ORDER BY всё равно вернёт LIMIT 1) и при
                      -- anchor > median×1.15 раздувала blend. NULL target/row → не
                      -- режем (graceful, нет данных для сравнения).
                      AND (
                        CAST(:rooms AS integer) IS NULL OR rooms IS NULL
                        OR abs(rooms - CAST(:rooms AS integer)) <= 1
                      )
                      AND (
                        CAST(:area AS double precision) IS NULL
                        OR area_m2 IS NULL OR area_m2 <= 0
                        OR area_m2 BETWEEN CAST(:area AS double precision) * 0.7
                                       AND CAST(:area AS double precision) * 1.3
                      )
                    ORDER BY
                      -- ближе по комнатам и площади → меньше score; NULL target → 0
                      (CASE WHEN CAST(:rooms AS integer) IS NOT NULL AND rooms IS NOT NULL
                            THEN abs(rooms - CAST(:rooms AS integer)) * 10 ELSE 0 END)
                      + (CASE WHEN CAST(:area AS double precision) IS NOT NULL
                                   AND area_m2 IS NOT NULL AND area_m2 > 0
                              THEN abs(area_m2 - CAST(:area AS double precision))
                                   / area_m2 * 100 ELSE 0 END) ASC,
                      fetched_at DESC
                    LIMIT 1
                    """
                ),
                {"hid": target_house_id, "rooms": rooms, "area": area},
            )
            .mappings()
            .first()
        )
    except Exception as exc:  # pragma: no cover — defensive
        logger.warning("house_imv anchor lookup failed (graceful): %s", exc)
        return None
    return dict(row) if row is not None else None


def _apply_imv_blend(
    *,
    median_price: int,
    range_high: int,
    median_ppm2: float,
    area: float,
    anchor_total: int | None,
    anchor_higher: int | None,
    weight: float,
    threshold: float,
) -> tuple[int, int, float, bool, int | None]:
    """Чистая (testable без БД) blend-трансформация для #651.

    Если есть надёжный якорь A (`anchor_total`, ПОЛНАЯ цена за квартиру) и он
    выше median_price × threshold (сигнал занижения) → поднимаем медиану до
    blend = median*(1-w) + A*w и расширяем верх диапазона до max(range_high,
    anchor_higher или A). ОДНОНАПРАВЛЕННО: только повышаем (баг — занижение).
    Если A ниже медианы — медиану НЕ трогаем, но диапазон можем расширить, чтобы
    включить A (информативность). Null-guard: при anchor_total=None — no-op.

    Returns (new_median_price, new_range_high, new_median_ppm2, blended,
             anchor_used_total).
    """
    if anchor_total is None or anchor_total <= 0 or median_price <= 0 or area <= 0:
        return median_price, range_high, median_ppm2, False, None

    blended = False
    new_median = median_price
    new_ppm2 = median_ppm2
    if anchor_total > median_price * threshold:
        w = max(0.0, min(1.0, weight))
        new_median = round(median_price * (1.0 - w) + anchor_total * w)
        new_ppm2 = new_median / area
        blended = True

    # Расширяем верх диапазона: предпочитаем верхнюю границу IMV-коридора, иначе сам
    # якорь. Только вверх (никогда не сужаем).
    range_top_candidate = anchor_higher if (anchor_higher and anchor_higher > 0) else anchor_total
    new_range_high = max(range_high, range_top_candidate, new_median)

    return new_median, new_range_high, new_ppm2, blended, anchor_total


def _fetch_dkp_corridor(
    db: Session,
    *,
    address: str | None,
    rooms: int | None,
    area: float | None,
    period_months: int = DEALS_PERIOD_MONTHS,
    area_tolerance: float = AREA_TOLERANCE,
) -> dict[str, Any] | None:
    """#652: коридор ₽/м² по реальным ДКП-сделкам Росреестра для target.

    Reuse паттерна street-deals (api/v1/trade_in.py): извлекаем улицу из адреса,
    фильтруем `deals` (source='rosreestr', та же rooms, площадь ±tolerance, окно
    period_months) и нормализуем per-m². Возвращаем low/median/high ₽/м².
    ADVISORY — caller не клампит, только сурфейсит + опциональная пометка.
    Best-effort: None при отсутствии улицы / сделок / любой ошибке.
    """
    if not address or rooms is None or not area:
        return None
    street_name = extract_street_name(address)
    if not street_name:
        return None
    area_min = area * (1.0 - area_tolerance)
    area_max = area * (1.0 + area_tolerance)
    try:
        rows = (
            db.execute(
                text(
                    """
                    SELECT price_per_m2
                    FROM deals
                    WHERE source = 'rosreestr'
                      AND address ILIKE :street_pattern
                      AND rooms = CAST(:rooms AS integer)
                      AND area_m2 BETWEEN :area_min AND :area_max
                      AND deal_date > NOW()
                          - (CAST(:period_months AS integer) || ' months')::interval
                      AND price_per_m2 > 0
                      -- #699: режем нерыночные ppm²-выбросы из коридора expected_sold
                      AND price_per_m2 BETWEEN :ppm_min AND :ppm_max
                    """
                ),
                {
                    "street_pattern": "%" + street_name + "%",
                    "rooms": rooms,
                    "area_min": area_min,
                    "area_max": area_max,
                    "period_months": period_months,
                    "ppm_min": DEAL_MIN_PPM2,
                    "ppm_max": DEAL_MAX_PPM2,
                },
            )
            .mappings()
            .all()
        )
    except Exception as exc:  # pragma: no cover — defensive
        logger.warning("dkp_corridor lookup failed (graceful): %s", exc)
        return None

    ppm2_values = sorted(float(r["price_per_m2"]) for r in rows if r["price_per_m2"])
    if not ppm2_values:
        return None
    return {
        "count": len(ppm2_values),
        "low_ppm2": int(ppm2_values[0]),
        "median_ppm2": int(_percentile(ppm2_values, 0.5)),
        "high_ppm2": int(ppm2_values[-1]),
        "period_months": period_months,
    }


# ── #651/#652 v2: same-building anchor (validated, 55 golden cases) ─────────────
#
# Радиусная медиана ₽/м² системно занижает премиум/видовые квартиры — она мешает
# топовый дом с массовой застройкой рядом. v2 строит PRIMARY якорь из комплов ТОГО
# ЖЕ ДОМА (Tier A), similarity-weighted по площади+комнатам, с premium-uplift и
# hard guardrail. Industry-grounded (Fannie Mae «same-project comps preferred» +
# inverse-adjustment weighting + FSD-as-confidence). Полностью за флагом.

# Street-alias map: ЕКБ-специфичные расхождения между golden/source-адресами и БД.
# Golden «Ткачёва 13» = БД «Ткачей 13» — без алиаса 0 комплов для 5 business-кейсов
# (Clever Park). Ключи/значения уже ё→е-нормализованы и lowercase. Расширяемо.
# ВАЖНО: применяется к ЦЕЛЬНОМУ street_core (после strip type-words), не к токену.
_STREET_ALIAS_MAP: dict[str, str] = {
    "ткачева": "ткачей",  # «ул. Ткачёва» (родит. падеж) ↔ БД «ул. Ткачей»
}

# Street-type токены: канонизируем (drop type-слово, оставляем имя). Реальные prod-
# адреса ставят тип ДО или ПОСЛЕ имени («улица Ткачей» И «Олимпийская наб.») — поэтому
# дропаем тип ОТКУДА УГОДНО в строке, а не только ведущий keyword. Лемматизированы по
# точкам/окончаниям: матчим точное слово ИЛИ «<core>.»-сокращение.
_STREET_TYPE_TOKENS: frozenset[str] = frozenset(
    {
        "улица",
        "ул",
        "у",
        "переулок",
        "пер",
        "проспект",
        "пр",
        "прт",
        "пркт",
        "пр-кт",
        "пр-т",
        "проезд",
        "бульвар",
        "бр",
        "б-р",
        "набережная",
        "наб",
        "шоссе",
        "ш",
        "площадь",
        "пл",
        "тракт",
        "аллея",
        "тупик",
    }
)

# Административные токены-маркеры: всё, что относится к городу/району/мкр — НЕ часть
# имени дома. Дропаем токен-маркер ВМЕСТЕ со следующим за ним словом-значением
# («р-н Октябрьский», «мкр Парковый», «г Екатеринбург»). Города ЕКБ-агломерации тоже
# чистим (стоят как ведущий токен «Екатеринбург,»/«Первоуральск,»).
_ADMIN_MARKER_TOKENS: frozenset[str] = frozenset(
    {"рн", "р-н", "район", "мкр", "микрорайон", "г", "гор", "город", "обл", "область"}
)
_CITY_TOKENS: frozenset[str] = frozenset(
    {"екатеринбург", "первоуральск", "березовский", "верхняя", "пышма", "среднеуральск", "россия"}
)

# Литера корпуса, прилипшая к номеру: «16а», «204г», «57а» → base + letter (рус/лат).
_HOUSE_LETTER_RE = re.compile(r"^(?P<num>\d+)\s*(?P<letter>[а-яa-z])?$", flags=re.UNICODE)
# Разбивка на токены: слова/числа, отбрасывая пунктуацию (',', '.', '·', '/', '-').
_TOKEN_SPLIT_RE = re.compile(r"[\s,;·]+", flags=re.UNICODE)


def _normalize_building_key(
    address: str | None,
) -> tuple[str | None, int | None, str | None]:
    """Нормализует адрес в robust-ключ «того же дома»: (street_core, base_no, letter).

    Инвариантен к prod-форматам ЕКБ-вторички:
      - ё→е, lowercase.
      - отрезает город / «р-н …» / «мкр …» / «г. …» / «· …»-хвост (district-suffix).
      - дропает street-type слова ОТКУДА УГОДНО (улица/ул./наб./набережная/пр./… —
        тип может стоять ДО или ПОСЛЕ имени: «улица Ткачей» И «Олимпийская наб.»);
        остаток alpha-токенов = street_core («бориса ельцина», «сакко и ванцетти»).
      - base_no = первый числовой токен ПОСЛЕ имени улицы, толерантно к
        «Ткачей, 13» = «Ткачей,13» = «Ткачей,д. 13» = «Ткачей дом 13».
      - letter = прилипшая литера корпуса («16а»→'а', «204г»→'г'); «/N» и «кN»
        (корпус) схлопываются к base (тот же дом). Литеры — РАЗНЫЕ дома (204г ≠ 204д).
      - street_core прогоняется через _STREET_ALIAS_MAP (ткачева→ткачей).

    Returns (street_core, base_no, letter) — любой элемент None если не извлёкся.
    Best-effort: при пустом адресе → (None, None, None).
    """
    if not address:
        return None, None, None
    norm = address.replace("ё", "е").replace("Ё", "Е").lower()
    # Отрезаем «· …»-хвост (Avito-формат «… · р-н Центр»): district всегда после «·».
    norm = norm.split("·")[0]
    # «д.»/«дом» перед номером → пробел (чтобы числовой токен встал отдельно).
    norm = re.sub(r"\b(?:д\.?|дом)\s*(?=\d)", " ", norm, flags=re.UNICODE)

    raw = [t for t in _TOKEN_SPLIT_RE.split(norm) if t]

    # 1. Вычищаем admin-маркеры (+следующее за ними значение) и города.
    cleaned: list[str] = []
    skip_next = False
    for tok in raw:
        if skip_next:
            skip_next = False
            continue
        bare = tok.rstrip(".")
        if bare in _ADMIN_MARKER_TOKENS:
            skip_next = True  # дропаем и сам маркер, и следующее слово-значение
            continue
        if bare in _CITY_TOKENS:
            continue
        cleaned.append(tok)

    # 2. House-токен = ПОСЛЕДНИЙ токен, начинающийся с цифры. Берём последний (а не
    #    первый), чтобы числа ВНУТРИ имени улицы («8 Марта», «1905 года») не съелись
    #    как номер дома — настоящий номер всегда в хвосте, за именем. Хвост за номером
    #    (корпус «/N», «кN») игнорируем; всё остальное — токены улицы.
    base_no: int | None = None
    letter: str | None = None
    house_idx: int | None = None
    for i, tok in enumerate(cleaned):
        if tok[0].isdigit():
            head = re.split(r"[/\\]", tok, maxsplit=1)[0]  # «4/2» → «4»; корпус отброшен
            head = re.split(r"к\d", head, maxsplit=1)[0]  # «105к1» → «105»
            m = _HOUSE_LETTER_RE.match(head)
            if m:
                base_no = int(m.group("num"))
                letter = m.group("letter") or None
                house_idx = i

    # 3. street_core = токены до номера дома, минус type-слова (улица/наб./пр./…) и
    #    минус сам house-токен. Числовые префиксы имени («8 марта») сохраняем.
    street_tokens = [
        tok
        for i, tok in enumerate(cleaned)
        if i != house_idx and tok.rstrip(".") not in _STREET_TYPE_TOKENS
    ]
    street_core = " ".join(street_tokens).strip() or None
    if street_core:
        street_core = _STREET_ALIAS_MAP.get(street_core, street_core)

    return street_core, base_no, letter


def _anchor_comp_from_row(r: Any) -> dict[str, Any]:
    """Строит comp-dict из строки SQL same-building/micro-radius (#694).

    Несёт 5 числовых полей для _compute_same_building_anchor
    (price_per_m2/area_m2/rooms/floor/total_floors) + display-поля listings
    (address/source/source_url/price_rub/listing_date/days_on_market/photo_urls/
    lat/lon), чтобы UI-аналоги отражали ИМЕННО комплы, на которых построен якорь,
    а не радиусные (cheaper/empty). Display-ключи best-effort: SELECT их тянет, но
    helper устойчив к их отсутствию (тестовые моки могут давать только числа).
    """
    return {
        "price_per_m2": int(r["price_per_m2"]),
        "area_m2": float(r["area_m2"]) if r.get("area_m2") is not None else None,
        "rooms": int(r["rooms"]) if r.get("rooms") is not None else None,
        "floor": int(r["floor"]) if r.get("floor") is not None else None,
        "total_floors": int(r["total_floors"]) if r.get("total_floors") is not None else None,
        "address": r.get("address"),
        "source": r.get("source"),
        "source_url": r.get("source_url"),
        "price_rub": int(r["price_rub"]) if r.get("price_rub") is not None else None,
        "listing_date": r.get("listing_date"),
        "days_on_market": r.get("days_on_market"),
        "photo_urls": r.get("photo_urls"),
        "lat": float(r["lat"]) if r.get("lat") is not None else None,
        "lon": float(r["lon"]) if r.get("lon") is not None else None,
    }


def _fetch_anchor_comps(
    db: Session,
    *,
    address: str | None,
    target_house_id: int | None,
    lat: float | None,
    lon: float | None,
    rooms: int | None,
    area: float | None,
    listing_segment: str | None = None,
) -> tuple[list[dict[str, Any]], str | None]:
    """Тированный набор комплов для same-building якоря. Стоп на 1-м тире с ≥ min_comps.

    Tier A — SAME BUILDING: normalized street + base house no (+ литера если есть).
      RELAXED rooms (без фильтра), БЕЗ area±15%. Не группируем по house_id_fk —
      один дом дробится на несколько fk (Хохрякова 48 = 7085/9878/12797).
    Tier C — micro-radius ≤500m (ST_DWithin) + same listing_segment + rooms match
      + area±25%. (Tier B «тот же ЖК» — skip: complex_id/cian_zhk_url ненадёжны.)
    Tier D — фолбэк: None tier (caller остаётся на радиусном median-пути).

    Excludes lots без price_per_m2. is_active=true. Best-effort: ([], None) на ошибке.

    Returns (comps, tier) где tier ∈ {'A','C', None}. comps — list dict с
    ключами price_per_m2 (int>0), area_m2 (float|None), rooms (int|None),
    floor (int|None), total_floors (int|None) — последние два для floor-веса (#680-WB).
    """
    min_comps = settings.estimate_sb_min_comps

    # ── Tier A: same building ────────────────────────────────────────────────
    street, base_no, letter = _normalize_building_key(address)
    if street and base_no is not None:
        # Numeric-boundary regex: дом 204 не матчит 2040/1204; литера при наличии
        # обязательна (204г ≠ 204д). Корпус «/N» допускаем (тот же дом). ё→е в SQL
        # для symmetry с нормализатором. psycopg v3: bind через :param, оператор ~.
        if letter:
            house_re = rf"(^|[^0-9]){base_no}\s*{letter}([^а-яёa-z0-9/]|/|$)"
        else:
            house_re = rf"(^|[^0-9]){base_no}([^а-яёa-z0-9/]|/|$)"
        try:
            rows = (
                db.execute(
                    text(
                        """
                        SELECT price_per_m2, area_m2, rooms, floor, total_floors,
                               address, source, source_url, price_rub, listing_date,
                               days_on_market, photo_urls, lat, lon
                        FROM listings
                        WHERE is_active = true
                          AND price_per_m2 > 0
                          AND lower(translate(address, 'ёЁ', 'ее')) LIKE :street_like
                          AND lower(translate(address, 'ёЁ', 'ее')) ~ :house_re
                        """
                    ),
                    {
                        "street_like": "%" + street + "%",
                        "house_re": house_re,
                    },
                )
                .mappings()
                .all()
            )
        except Exception as exc:  # pragma: no cover — defensive
            logger.warning("anchor Tier A lookup failed (graceful): %s", exc)
            try:
                db.rollback()
            except Exception:
                pass
            rows = []
        comps = [_anchor_comp_from_row(r) for r in rows if r["price_per_m2"]]
        if len(comps) >= min_comps:
            logger.info(
                "anchor tier=A street=%r base=%s letter=%s → %d comps",
                street,
                base_no,
                letter,
                len(comps),
            )
            return comps, "A"

    # ── Tier C: micro-radius ≤500m + same segment + rooms + area±25% ─────────
    if lat is not None and lon is not None and rooms is not None and area:
        try:
            rows = (
                db.execute(
                    text(
                        """
                        SELECT price_per_m2, area_m2, rooms, floor, total_floors,
                               address, source, source_url, price_rub, listing_date,
                               days_on_market, photo_urls, lat, lon
                        FROM listings
                        WHERE is_active = true
                          AND price_per_m2 > 0
                          AND rooms = CAST(:rooms AS integer)
                          AND area_m2 BETWEEN :area_min AND :area_max
                          AND (
                            CAST(:segment AS text) IS NULL
                            OR listing_segment IS NULL
                            OR listing_segment = CAST(:segment AS text)
                          )
                          AND geom IS NOT NULL
                          AND ST_DWithin(
                                geom::geography,
                                ST_MakePoint(:lon, :lat)::geography,
                                500
                          )
                        """
                    ),
                    {
                        "rooms": rooms,
                        "area_min": area * 0.75,
                        "area_max": area * 1.25,
                        "segment": listing_segment,
                        "lon": lon,
                        "lat": lat,
                    },
                )
                .mappings()
                .all()
            )
        except Exception as exc:  # pragma: no cover — defensive
            logger.warning("anchor Tier C lookup failed (graceful): %s", exc)
            try:
                db.rollback()
            except Exception:
                pass
            rows = []
        comps = [_anchor_comp_from_row(r) for r in rows if r["price_per_m2"]]
        if len(comps) >= min_comps:
            logger.info("anchor tier=C micro-radius → %d comps", len(comps))
            return comps, "C"

    # Tier D — caller fallback (радиусный median-путь без anchor).
    return [], None


def _band_haircut(anchor_ppm2: float) -> float:
    """asking→sold haircut, banded по ppm² (class-label в prod пуст — band на цену).

    Премиум (высокий ppm²) торгуется плотнее → меньше скидка; эконом — больше.
    Пороги ЕКБ-вторички: ≥350k → 4%; 180-350k → 5%; <180k → 7%. Дефолт из config.
    """
    if anchor_ppm2 >= 350_000:
        return 0.04
    if anchor_ppm2 >= 180_000:
        return settings.asking_to_sold_haircut  # 5% mid
    return 0.07


def _compute_same_building_anchor(
    comps: list[dict[str, Any]],
    *,
    area_target: float,
    rooms_target: int | None,
    tier: str,
    sigma: float,
    rooms_boost: float,
    floor_target: int | None = None,
    total_floors_target: int | None = None,
    floor_sigma: float = 0.0,
) -> dict[str, Any] | None:
    """Чистая (testable без БД) свёртка комплов в anchor-оценку.

    1. similarity-weighted mean ppm²: w_i = exp(−(ln(area_i/area_target))²/(2σ²))
       × (rooms_boost если rooms_i==rooms_target) × FLOOR-вес. area_i=None →
       area-вес 1.0 (нейтрально). FLOOR-вес (#680-WB) — Gaussian по ОТНОСИТЕЛЬНОЙ
       вертикальной позиции pos=floor/total_floors: exp(−(pos_i−pos_t)²/(2σ_f²)),
       σ_f=floor_sigma. Прижимает якорь к комплам с похожим этажом (1-й/последний
       и видовые этажи реально отличаются по цене). floor_sigma=0 ИЛИ нет floor у
       target/компла → вес 1.0 (выключено / нейтрально — no regress).
    2. PREMIUM uplift (class-free): target — топ-юнит ДОМА (area ≥ p66 площадей
       комплов) И rooms_target ≥ медианы комнат комплов И tier == 'A' → берём
       weighted ~p70 верхний квантиль ppm² (консервативно, только вверх). Условие
       по комнатам (#680-WB) не даёт мелкокомнатному юниту во флагман-доме унаследовать
       headline-премию флагмана (3к/153 в доме с 4к-флагманом ≠ цена флагмана).
    3. haircut asking→sold (banded по anchor ppm²): anchor_sold = anchor×(1−haircut).
    4. FSD = 0.07 + 0.25·CV(comp ppm²) + tier_penalty + n_penalty; range полуширина
       = k·fsd. confidence-банд по fsd.

    Returns dict {anchor_ppm2, anchor_sold_ppm2, fsd, confidence, n, cv,
    comp_min_ppm2, used_uplift, haircut} или None если комплов нет.
    """
    if not comps:
        return None
    ppm2 = [float(c["price_per_m2"]) for c in comps if c.get("price_per_m2")]
    if not ppm2:
        return None
    n = len(ppm2)

    # target relative vertical position (None → floor-вес отключён/нейтрален).
    target_pos: float | None = None
    if floor_sigma > 0 and floor_target and total_floors_target and total_floors_target > 0:
        target_pos = floor_target / total_floors_target

    # 1. similarity weights (area × rooms × floor-position)
    weights: list[float] = []
    for c in comps:
        a = c.get("area_m2")
        if a and area_target > 0:
            area_w = math.exp(-((math.log(a / area_target)) ** 2) / (2.0 * sigma * sigma))
        else:
            area_w = 1.0  # площадь неизвестна → нейтральный area-вес
        rooms_match = rooms_target is not None and c.get("rooms") == rooms_target
        rooms_w = rooms_boost if rooms_match else 1.0
        floor_w = 1.0
        if target_pos is not None:
            cf = c.get("floor")
            ctf = c.get("total_floors")
            if cf and ctf and ctf > 0:
                comp_pos = cf / ctf
                floor_w = math.exp(
                    -((comp_pos - target_pos) ** 2) / (2.0 * floor_sigma * floor_sigma)
                )
            # компл без floor/total_floors → нейтральный floor-вес 1.0
        weights.append(area_w * rooms_w * floor_w)
    wsum = sum(weights)
    if wsum > 0:
        anchor = sum(w * p for w, p in zip(weights, ppm2, strict=True)) / wsum
    else:
        anchor = _percentile(sorted(ppm2), 0.5)

    # 2. premium uplift — топ-юнит дома (площадь ≥ p66 И комнаты ≥ медианы) И Tier A
    #    → weighted p70. Условие по комнатам отсекает мелкие юниты во флагман-домах.
    used_uplift = False
    areas = [c.get("area_m2") for c in comps if c.get("area_m2")]
    comp_rooms = [c.get("rooms") for c in comps if c.get("rooms") is not None]
    if tier == "A" and areas and area_target > 0:
        p66_area = _percentile(sorted(areas), 0.66)
        rooms_ok = True
        if rooms_target is not None and comp_rooms:
            median_rooms = _percentile(sorted(comp_rooms), 0.5)
            rooms_ok = rooms_target >= median_rooms
        if area_target >= p66_area and rooms_ok:
            p70 = _percentile(sorted(ppm2), 0.70)
            if p70 > anchor:
                anchor = p70
                used_uplift = True

    # 3. asking→sold haircut (banded)
    haircut = _band_haircut(anchor)
    anchor_sold = anchor * (1.0 - haircut)

    # 4. FSD-диапазон (tight). CV = std/mean comp ppm².
    mean_ppm2 = sum(ppm2) / n
    if mean_ppm2 > 0 and n >= 2:
        var = sum((p - mean_ppm2) ** 2 for p in ppm2) / n
        cv = math.sqrt(var) / mean_ppm2
    else:
        cv = 0.0
    tier_penalty = {"A": 0.0, "C": 0.05}.get(tier, 0.09)
    n_penalty = 0.05 if n < 3 else (0.02 if n < 5 else 0.0)
    fsd = 0.07 + 0.25 * cv + tier_penalty + n_penalty

    if fsd <= 0.13:
        confidence = "high"
    elif fsd <= 0.20:
        confidence = "medium"
    else:
        confidence = "low"

    return {
        "anchor_ppm2": anchor,
        "anchor_sold_ppm2": anchor_sold,
        "fsd": fsd,
        "confidence": confidence,
        "n": n,
        "cv": cv,
        "comp_min_ppm2": min(ppm2),
        "comp_max_ppm2": max(ppm2),
        "used_uplift": used_uplift,
        "haircut": haircut,
    }


# ── #693 prod-fix: coarse-geocode detector (DaData-independent) ─────────────
# Дом всегда оканчивается номером (1-3 цифры, опц. литера корпуса). Centroid
# города/региона его НЕ содержит. Прод-сигнал грубости геокода, работающий БЕЗ
# DaData: на проде DaData может быть off (token не сконфигурирован) → dadata.qc_geo
# всегда None → старый гейт #707 (условный на qc_geo>=2) НЕ срабатывал НИКОГДА,
# даже для региона/города. Кэш-агностичен — смотрит на geo.full_address, каким бы
# провайдером/кэшем он ни был получен.
#
# Граница (?<!\d)\d{1,3}(?!\d) исключает почтовый индекс (6 цифр) и год постройки
# (4 цифры) — они не матчат «дом». Числовые улицы («8 Марта») матчат свой номер →
# НЕ даунгрейдятся (консервативно: реальная улица, аналоги рядом есть; принцип
# #707 — никаких ложных downgrade важнее, чем отлов всех coarse-кейсов).
_HOUSE_NUMBER_RE = re.compile(r"(?<!\d)\d{1,3}[а-яёa-z]?(?!\d)", flags=re.IGNORECASE)


def _geocode_is_coarse(geo: GeocodeResult) -> bool:
    """True если геокод разрешился лишь до centroid'а НП/города/региона (без дома).

    Два сигнала (OR):
      1. provider confidence == 'locality' — явный centroid-маркер (на будущее:
         текущие провайдеры его не эмитят, но enum/кэш-колонка допускают, и так
         геокодер можно улучшить позже без правки гейта).
      2. в geo.full_address нет house-number токена (1-3 цифры) → геокодер не дошёл
         до дома, вернул центр НП/города/региона.
    Гарантирует ZERO ложных downgrade на реальных адресах: у любого реального дома
    есть номер → токен матчится → не coarse.
    """
    if geo.confidence == "locality":
        return True
    return _HOUSE_NUMBER_RE.search(geo.full_address or "") is None


# ── Time-budget guard (#654) ────────────────────────────────────────────────
async def _with_budget(coro: Any, budget_s: float, *, label: str) -> Any:
    """Await `coro` under an asyncio.wait_for() time budget.

    On timeout the coroutine is cancelled and we return None — mapping a slow
    upstream onto the SAME graceful "None" path these enrichments already take
    on network error, so a single slow source degrades the estimate instead of
    blowing the gateway read timeout (#654: opaque Caddy 502/504).

    budget_s <= 0 disables the guard (await directly) — escape hatch via config.
    """
    if budget_s is None or budget_s <= 0:
        return await coro
    try:
        return await asyncio.wait_for(coro, timeout=budget_s)
    except TimeoutError:
        # asyncio.TimeoutError is an alias of builtin TimeoutError (py3.11+).
        logger.warning("%s exceeded %.1fs budget — degrading to None (#654)", label, budget_s)
        return None


# ── Public ───────────────────────────────────────────────────────────────────
async def estimate_quality(
    payload: TradeInEstimateInput, db: Session, created_by: str | None = None
) -> AggregatedEstimate:
    """Главная функция — оценка квартиры по реальным данным.

    PR M / #564 Phase 3: rosreestr_deals **included** в actual_deals output.
    Stale NOTE 2026-05-24 (про ДДУ contamination) устарел — importer
    `import-rosreestr.sh` после PR-A 2026-05-24 фильтрует doc_type='ДКП',
    ДДУ первички исключены. Deals идут в `actual_deals` JSONB поле
    AggregatedEstimate с tier classification (T0_per_house / T1_per_street)
    — frontend может разделять confidence в UI.

    Returns:
        AggregatedEstimate с estimate_id, медианой, диапазоном, аналогами.
    """
    # 1. Geocode (#654: time-budgeted — Yandex/Nominatim retry chain can stack
    #    multiple network round-trips + 1s Nominatim rate-limit sleeps).
    geo: GeocodeResult | None = None
    if payload.address:
        geo = await _with_budget(
            geocode(payload.address, db),
            settings.estimate_geocode_budget_s,
            label="geocode",
        )

    if geo is None:
        # Без координат не можем искать через PostGIS. Возвращаем low confidence.
        logger.warning("geocode failed for %s — returning low-confidence estimate", payload.address)
        return _empty_estimate(payload, db, reason="address_not_geocoded", created_by=created_by)

    # 1b. DaData enrichment (PR Q1) — on-demand cleanup для target адреса.
    # Best-effort: graceful None при отсутствии credentials / quota / fail.
    # Дополняет geocode результатом kadastr_num + canonical form + nearest metro.
    dadata: DadataAddressResult | None = None
    try:
        dadata = await dadata_clean_address(payload.address)
    except Exception as exc:  # pragma: no cover — defensive
        logger.warning("dadata: unexpected error (graceful): %s", exc)

    # 1c. #6 House-match: резолвим target в КАНОНИЧЕСКИЙ house_id (read-only, без
    #     создания записи). Это даёт детерминированный Tier S «тот же дом» через
    #     listings.house_id_fk (99% покрытие), точнее хрупкого address-string match.
    #     cadastr от DaData → cadastr_exact tier заработает по мере backfill houses.
    #     Best-effort: None при любой ошибке, estimator продолжает на гео-tier'ах.
    target_house_id: int | None = None
    try:
        match = match_house_readonly(
            db,
            address=(dadata.canonical_address if dadata else None) or geo.full_address,
            lat=geo.lat,
            lon=geo.lon,
            cadastral_number=(dadata.house_cadnum if dadata else None),
        )
        if match is not None:
            target_house_id = match[0]
            logger.info(
                "estimate target → house_id=%s via %s (conf=%.2f)", match[0], match[2], match[1]
            )
    except Exception as exc:  # pragma: no cover — defensive
        logger.warning("target house match failed (graceful): %s", exc)

    # 2. #392: обогащаем год / тип дома из картографии (OSM Overpass), если
    #    пользователь их не указал — это улучшает house-match аналогов (#6).
    #    Best-effort: при недоступности OSM target_* остаются None.
    #    #654: time-budgeted — Overpass httpx timeout 15s сам по себе близок к
    #    gateway-таймауту; деградируем в None при превышении budget.
    target_year = payload.year_built
    target_house_type = payload.house_type
    if target_year is None or target_house_type is None:
        house_meta = await _with_budget(
            get_house_metadata(geo.lat, geo.lon, db),
            settings.estimate_house_meta_timeout_s,
            label="house_metadata(overpass)",
        )
        if house_meta is not None:
            if target_year is None:
                target_year = house_meta.year_built
            if target_house_type is None:
                target_house_type = house_meta.house_type

    # 3. Four-tier fallback (PR 9 — added Tier 0 with cohort filter):
    #    0) 1km + ±15% area + cohort match (year_built — если задан)
    #    a) 1km + ±15% area  (без cohort — drop fallback)
    #    b) 2km + ±15% area  (fallback_used = True)
    #    c) 2km + ±25% area  (fallback_used = True, area_widened = True)
    cohort_range = _target_cohort_range(target_year)

    if cohort_range is not None:
        cy_min, cy_max = cohort_range
        listings_tier0, _, analog_tier = _fetch_analogs(
            db,
            lat=geo.lat,
            lon=geo.lon,
            rooms=payload.rooms,
            area=payload.area_m2,
            radius_m=DEFAULT_RADIUS_M,
            full_address=geo.full_address,
            target_house_id=target_house_id,
            year_built=target_year,
            house_type=target_house_type,
            total_floors=payload.total_floors,
            cohort_year_min=cy_min,
            cohort_year_max=cy_max,
        )
    else:
        listings_tier0 = []
        analog_tier = "W"

    if len(listings_tier0) >= MIN_ANALOGS_TIER_0:
        listings = listings_tier0
        fallback_used = False
    else:
        # Tier 0 пуст/мал — graceful fallback на Tier A без cohort
        listings, fallback_used, analog_tier = _fetch_analogs(
            db,
            lat=geo.lat,
            lon=geo.lon,
            rooms=payload.rooms,
            area=payload.area_m2,
            radius_m=DEFAULT_RADIUS_M,
            full_address=geo.full_address,
            target_house_id=target_house_id,
            year_built=target_year,
            house_type=target_house_type,
            total_floors=payload.total_floors,
        )
    area_widened = False

    if len(listings) < 5:
        listings_wide, _, analog_tier_wide = _fetch_analogs(
            db,
            lat=geo.lat,
            lon=geo.lon,
            rooms=payload.rooms,
            area=payload.area_m2,
            radius_m=FALLBACK_RADIUS_M,
            full_address=geo.full_address,
            target_house_id=target_house_id,
            year_built=target_year,
            house_type=target_house_type,
            total_floors=payload.total_floors,
        )
        if len(listings_wide) > len(listings):
            listings = listings_wide
            fallback_used = True
            analog_tier = analog_tier_wide

    # Tier C: если даже на 2км мало — расширяем area tolerance до ±25%
    # (актуально для отдалённых районов / новостроек с нестандартной планировкой)
    if len(listings) < 3:
        listings_widearea, _, analog_tier_wa = _fetch_analogs(
            db,
            lat=geo.lat,
            lon=geo.lon,
            rooms=payload.rooms,
            area=payload.area_m2,
            radius_m=FALLBACK_RADIUS_M,
            area_tolerance=0.25,
            full_address=geo.full_address,
            target_house_id=target_house_id,
            year_built=target_year,
            house_type=target_house_type,
            total_floors=payload.total_floors,
        )
        if len(listings_widearea) > len(listings):
            listings = listings_widearea
            fallback_used = True
            area_widened = True
            analog_tier = analog_tier_wa

    # 3. Outlier filter
    listings_clean = _filter_outliers(listings)

    # 4. Aggregation
    if listings_clean:
        prices_ppm2 = sorted(lot["price_per_m2"] for lot in listings_clean if lot["price_per_m2"])
        median_ppm2 = _percentile(prices_ppm2, 0.5)
        q1_ppm2 = _percentile(prices_ppm2, 0.25)
        q3_ppm2 = _percentile(prices_ppm2, 0.75)
        median_price = int(median_ppm2 * payload.area_m2)
        range_low = int(q1_ppm2 * payload.area_m2)
        range_high = int(q3_ppm2 * payload.area_m2)
        n_analogs = len(listings_clean)
    else:
        median_ppm2 = 0
        median_price = 0
        range_low = 0
        range_high = 0
        n_analogs = 0

    # 4b. Поправка на состояние ремонта (встреча Птицы: ремонт влияет на цену).
    # Аналоги — микс состояний; коэффициент сдвигает оценку под ремонт клиента.
    repair_coef = _repair_coefficient(payload.repair_state)
    repair_note = ""
    if listings_clean and repair_coef != 1.0:
        median_price = int(median_price * repair_coef)
        range_low = int(range_low * repair_coef)
        range_high = int(range_high * repair_coef)
        median_ppm2 = median_ppm2 * repair_coef
        pct = round((repair_coef - 1.0) * 100)
        repair_note = (
            f" Цена скорректирована на состояние ремонта "
            f"({_REPAIR_LABEL.get(payload.repair_state, '')} {pct:+d}%)."
        )

    # 4c. Asking→sold коррекция (#648 Stage 3) — PURELY ADDITIVE. Headline
    # median_price/range_*/median_ppm2 (ASKING активных объявлений) НЕ трогаем;
    # вычисляем ПАРАЛЛЕЛЬНУЮ expected_sold цену = asking × per-rooms ratio
    # (asking_to_sold_ratios, migration 080). Это релевантная для выкупа цена
    # сделки (backtest #648 S1: bias asking-медианы +20% → −4% на held-out ДКП).
    # NOTE: actual_deals (#564) остаётся ИНФОРМАЦИОННЫМ и НЕ подмешивается в
    # headline — sold-коррекция здесь единственный sold-сигнал (без double-count).
    # NOTE: expected_sold_* (= asking × ratio) выводятся НЕ здесь, а ПОСЛЕ #651
    # IMV-blend (ниже), который мутирует median_price/median_ppm2/range_high. Иначе
    # expected_sold остаётся pre-blend → asking 75M / sold 45M (бессмысленная скидка
    # в HeroSummary) и stale-значения persist'ятся в trade_in_estimates. Здесь только
    # резолвим ratio/basis (нужны для confidence/explanation и null-guard).
    asking_to_sold_ratio, ratio_basis = _get_asking_sold_ratio(db, payload.rooms)
    expected_sold_per_m2: int | None = None
    expected_sold_price: int | None = None
    expected_sold_range_low: int | None = None
    expected_sold_range_high: int | None = None
    # Не было ratio (нет таблицы/бакета) — не вводим в заблуждение пустым basis.
    if asking_to_sold_ratio is None:
        ratio_basis = None

    confidence, explanation = _compute_confidence(
        n_analogs,
        median_ppm2,
        q1_ppm2 if listings_clean else 0,
        q3_ppm2 if listings_clean else 0,
        fallback_used,
        area_widened,
        listings=listings_clean,
    )

    # Tier note — информируем пользователя о качестве house-match
    tier_note = ""
    if analog_tier == "S":
        tier_note = " (аналоги из того же дома)"
    elif analog_tier == "H":
        tf_str = f"{payload.total_floors}-эт." if payload.total_floors is not None else ""
        yr_str = f"{target_year}±15 г." if target_year else ""
        parts_str = ", ".join(p for p in [yr_str, tf_str] if p)
        tier_note = f" (аналоги из домов того же класса: {parts_str})" if parts_str else ""
    else:
        tier_note = " (нет аналогов в том же доме/классе — расширили поиск)"

    explanation = (explanation or "") + tier_note + repair_note

    # ── Stage 3: Avito IMV evaluation as 5-th source (on-demand cached) ──
    imv_eval: IMVEvaluation | None = None
    imv_house_type = _IMV_HOUSE_TYPE_MAP.get(target_house_type)
    imv_renovation = _IMV_REPAIR_MAP.get(payload.repair_state)
    # IMV требует: address, rooms, area, floor, floor_at_home, house_type, renovation_type.
    # Если payload не содержит required fields — skip IMV (graceful).
    if (
        geo is not None
        and geo.full_address
        and payload.rooms is not None
        and payload.area_m2
        and payload.floor is not None
        and payload.total_floors is not None
        and imv_house_type is not None
        and imv_renovation is not None
    ):
        imv_eval = await _get_or_fetch_imv_cached(
            db,
            address=geo.full_address,
            rooms=payload.rooms,
            area_m2=payload.area_m2,
            floor=payload.floor,
            floor_at_home=payload.total_floors,
            house_type=imv_house_type,
            renovation_type=imv_renovation,
            has_balcony=bool(payload.has_balcony),
            has_loggia=False,  # payload не разделяет балкон/лоджия → дефолт False
        )

    # Include IMV в sources_used если получили
    sources_used_pre = sorted({lot.get("source") for lot in listings_clean if lot.get("source")})
    if imv_eval is not None:
        sources_used_pre = sorted(set(sources_used_pre) | {"avito_imv"})

    # ── Stage 8: Yandex Valuation as on-demand source (anonymous, cached 24h) ──
    # #654: главный латентность-подозреваемый. Этот источник UNGATED — бежит на
    # КАЖДОЙ оценке (в т.ч. без floor/total_floors), а его внутренний httpx
    # timeout 30s + curl_cffi impersonation + sleep_between_requests могут одни
    # превысить gateway read timeout → opaque 502. Оборачиваем в budget: при
    # превышении → None (cache-hit путь быстрый, timeout бьёт только по медленному
    # cache-miss fetch). Деградация идентична сетевой ошибке внутри функции.
    yandex_val: YandexValuationResult | None = None
    if geo is not None and geo.full_address:
        yandex_val = await _with_budget(
            _get_or_fetch_yandex_valuation_cached(db, address=geo.full_address),
            settings.estimate_yandex_valuation_timeout_s,
            label="yandex_valuation",
        )
        if yandex_val is not None:
            sources_used_pre = sorted(set(sources_used_pre) | {"yandex_valuation"})
            saved_hist = _save_yandex_history_items(db, yandex_val)
            logger.info(
                "yandex_valuation: history items processed=%d saved=%d"
                " (house_id=NULL — matching deferred)",
                len(yandex_val.history_items),
                saved_hist,
            )

    # ── Stage 9: Cian Valuation as 7th source (on-demand, 24h cached, graceful if no cookies) ──
    cian_val: CianValuationResult | None = None
    if (
        geo is not None
        and geo.full_address
        and payload.rooms is not None
        and payload.area_m2
        and payload.floor is not None
        and payload.total_floors is not None
    ):
        try:
            cian_val = await estimate_via_cian_valuation(
                db,
                address=geo.full_address,
                total_area=payload.area_m2,
                rooms_count=payload.rooms,
                floor=payload.floor,
                total_floors=payload.total_floors,
                repair_type="cosmetic",
                deal_type="sale",
                use_cache=True,
            )
            if cian_val is not None and cian_val.sale_price_rub:
                sources_used_pre = sorted(set(sources_used_pre) | {"cian_valuation"})
                logger.info(
                    "cian_valuation: price=%s accuracy=%s house_id=%s",
                    cian_val.sale_price_rub,
                    cian_val.sale_accuracy,
                    cian_val.external_house_id,
                )
        except Exception as exc:
            logger.warning("cian_valuation: lookup failed (graceful): %s", exc)

    # ── #651/#652 v2: SAME-BUILDING ANCHOR (PRIMARY, validated) ──────────────
    # Якорь из комплов ТОГО ЖЕ ДОМА (Tier A) / micro-radius (Tier C). Когда он
    # сработал — он ЗАМЕНЯЕТ радиусную медиану для median_price/ppm²/range (premium
    # больше не размывается массовой застройкой). За флагом; OFF ⇒ точно старое
    # поведение. listing_segment для Tier C берём из самого частого среди аналогов.
    anchor_tier: str | None = None
    # #694: комплы того же дома, на которых РЕАЛЬНО построен headline-якорь.
    # Заполняется только когда anchor мутировал median (ниже) — тогда UI-аналоги
    # строятся из них, а не из радиусных listings_clean (cheaper/empty).
    anchor_comps_used: list[dict[str, Any]] = []
    # #691: гейт НЕ требует радиусных аналогов (listings_clean) / median_price>0.
    # На проде геокод часто = None → ST_DWithin не находит радиусные комплы →
    # median=0, и same-building якорь скипался, ХОТЯ комплы того же дома есть
    # (_fetch_anchor_comps резолвит дом по payload.address без гео — Ткачёва 13,
    # Сакко 99). Запускаем по resolved-таргету (area + raw-адрес/geo); если комплов
    # реально нет — _compute_same_building_anchor вернёт None и мутации не будет
    # (поведение идентично старому). Новый гейт ⊇ старого: при непустом radius
    # адрес всегда есть, так что ранее-якорившиеся кейсы якорятся по-прежнему.
    if (
        settings.estimate_same_building_anchor_enabled
        and payload.area_m2
        and (payload.address or (geo is not None and geo.full_address))
    ):
        seg_counts: dict[str, int] = {}
        for lot in listings_clean:
            seg = lot.get("listing_segment")
            if seg:
                seg_counts[seg] = seg_counts.get(seg, 0) + 1
        target_segment = max(seg_counts, key=seg_counts.get) if seg_counts else None

        comps, anchor_tier = _fetch_anchor_comps(
            db,
            # payload.address (сырой ввод) ПЕРВИЧЕН: на проде геокод часто
            # возвращает None (lat/lon/canonical=None), и geo.full_address пуст —
            # тогда same-building Tier A не находил дом и падал в радиус. Raw-адрес
            # всегда есть и именно на нём валидирован нормализатор (#677/#679).
            address=payload.address or geo.full_address,
            target_house_id=target_house_id,
            lat=geo.lat,
            lon=geo.lon,
            rooms=payload.rooms,
            area=payload.area_m2,
            listing_segment=target_segment,
        )
        anchor = _compute_same_building_anchor(
            comps,
            area_target=payload.area_m2,
            rooms_target=payload.rooms,
            tier=anchor_tier or "",
            sigma=settings.estimate_sb_area_sigma,
            rooms_boost=settings.estimate_sb_rooms_match_boost,
            floor_target=payload.floor,
            total_floors_target=payload.total_floors,
            floor_sigma=settings.estimate_sb_floor_sigma,
        )
        if anchor is not None:
            # #694: якорь мутирует headline — UI-аналоги должны отражать ЭТИ комплы.
            anchor_comps_used = comps
            # Headline = recommended ASKING price (комплы — активные объявления;
            # golden-реалы — asking). Берём anchor_ppm2 (PRE-haircut), НЕ
            # anchor_sold_ppm2. asking→sold скидка применяется единственным
            # механизмом — per-rooms asking_to_sold_ratio (migration 080) ниже,
            # дающим distinct expected_sold. Двойная скидка (band-haircut здесь +
            # ratio там) давала median == expected_sold (#677/#681 collision).
            est_ppm2 = anchor["anchor_ppm2"]
            # PREMIUM GUARDRAIL (hard): не ниже минимального same-building ppm² (−tol).
            # Только Tier A/C (комплы реально из дома/микрорайона). Эконом — no-op
            # (est уже ≥ floor), премиум — поднимает если mean занизил. ASKING-space.
            floor_ppm2 = anchor["comp_min_ppm2"] * (1.0 - settings.estimate_sb_guardrail_tol)
            if est_ppm2 < floor_ppm2:
                est_ppm2 = floor_ppm2
            # POINT = anchor_asking × area × repair_coef (repair уже применён к старой
            # median; здесь применяем к свежему якорю — заменяем headline целиком).
            new_ppm2 = est_ppm2 * repair_coef
            point = int(new_ppm2 * payload.area_m2)
            # FSD-диапазон (tight): симметричный вокруг point, k·fsd полуширина.
            half = settings.estimate_fsd_k * anchor["fsd"]
            new_range_low = int(point * max(0.0, 1.0 - half))
            new_range_high = int(point * (1.0 + half))
            # Диапазон должен ПОКРЫВАТЬ same-building спред комплов (sold-adjusted) и
            # удовлетворять low ≤ point ≤ high. Внутридомовая дисперсия (этаж/вид) —
            # реальный разброс цены в доме; честный диапазон обязан её включать
            # (иначе видовой топ-юнит вылетает за range_high — residual miss спека).
            # comp spread в ASKING-пространстве (комплы — активные объявления). range_high
            # покрывает RAW comp max — честно показываем верх дома (видовой/топ-юнит),
            # иначе он вылетает за диапазон. range_low — RAW comp min (asking-space):
            # headline теперь asking, band-haircut больше не применяется (sold-скидка —
            # единственным механизмом per-rooms ratio ниже).
            spread_low = int(anchor["comp_min_ppm2"] * payload.area_m2)
            spread_high = int(anchor["comp_max_ppm2"] * payload.area_m2)
            new_range_low = min(new_range_low, spread_low, point)
            new_range_high = max(new_range_high, spread_high, point)

            logger.info(
                "sb_anchor: tier=%s n=%d radius_median_ppm2=%d → anchor_asking_ppm2=%d"
                " (uplift=%s haircut=%.2f) point %d → %d",
                anchor_tier,
                anchor["n"],
                int(median_ppm2),
                int(est_ppm2),
                anchor["used_uplift"],
                anchor["haircut"],
                median_price,
                point,
            )
            median_ppm2 = new_ppm2
            median_price = point
            range_low = new_range_low
            range_high = new_range_high
            confidence = anchor["confidence"]
            tier_label = "того же дома" if anchor_tier == "A" else "ближайшего окружения (≤500 м)"
            # #695: когда якорь построил headline, explanation описывает ИМЕННО якорные
            # комплы (anchor['n'] из tier_label). Радиусный base-текст («Найдено N из M
            # разных адресов») и analog_tier tier_note относятся к ДРУГОЙ выборке и
            # противоречат якорю (n_analogs≠anchor['n']; «разных адресов» vs «того же
            # дома») — поэтому ЗАМЕНЯЕМ, а не конкатенируем. repair_note сохраняем
            # (ортогонален — поправка на ремонт). confidence уже = anchor["confidence"].
            explanation = (
                f"Оценка построена по {anchor['n']} аналогам из {tier_label}"
                f"{' (топ-уровень в доме)' if anchor['used_uplift'] else ''}."
            ) + repair_note
            # #695 (QA fixup): когда якорь подменяет headline и список аналогов на
            # комплы дома, n_analogs ДОЛЖЕН считаться по anchor-популяции, а не по
            # радиусу (listings_clean). Иначе все UI-счётчики («Аналогов N»,
            # «Показано N из M», «по N аналогам», гистограмма) расходятся: radius=4 vs
            # anchor=5 → артефакт «Показано 5 из 4». anchor['n'] = число комплов, на
            # которых построена оценка (= показываемый analogs, capped 10). Для anchor-
            # пути это и есть «полное число найденных» по контракту n_analogs (#698).
            n_analogs = anchor["n"]

    # ── #651: IMV / Yandex BLEND (killer accuracy fix) — SECONDARY, Tier D only ──
    # Радиусная медиана системно занижает премиум/видовые квартиры (нет class/
    # segment-коррекции). Берём РЕАЛЬНЫЙ Avito IMV target-дома из house_imv_evaluations
    # (avito_imv_evaluations пуст — keyed estimate_id, on-demand), используем как
    # anchor: если IMV recommended_price > median × threshold — поднимаем медиану
    # blend'ом и расширяем верх диапазона. Всё за флагом + null-guard (no-op без IMV).
    # ВАЖНО (v2): IMV-blend выполняется ТОЛЬКО когда same-building anchor НЕ сработал
    # (anchor_tier is None) — не накладываем blend поверх уже-построенного якоря дома.
    avito_imv_summary: AvitoImvSummary | None = None
    if (
        anchor_tier is None
        and settings.estimate_imv_blend_enabled
        and listings_clean
        and median_price > 0
    ):
        imv_anchor = _fetch_house_imv_anchor(
            db,
            target_house_id=target_house_id,
            rooms=payload.rooms,
            area=payload.area_m2,
        )
        # Anchor chain: prefer Avito IMV recommended; fall back to on-demand imv_eval.
        anchor_total: int | None = None
        anchor_higher: int | None = None
        anchor_label: str | None = None
        if imv_anchor is not None and imv_anchor.get("recommended_price"):
            anchor_total = int(imv_anchor["recommended_price"])
            anchor_higher = (
                int(imv_anchor["higher_price"]) if imv_anchor.get("higher_price") else None
            )
            anchor_label = "оценке Avito IMV"
            avito_imv_summary = AvitoImvSummary(
                recommended_price=anchor_total,
                lower_price=(
                    int(imv_anchor["lower_price"]) if imv_anchor.get("lower_price") else None
                ),
                higher_price=anchor_higher,
                market_count=(
                    int(imv_anchor["market_count"]) if imv_anchor.get("market_count") else None
                ),
            )
        elif imv_eval is not None and imv_eval.recommended_price:
            # on-demand IMV (avito_imv_evaluations) — fallback, обычно пуст
            anchor_total = int(imv_eval.recommended_price)
            anchor_higher = int(imv_eval.higher_price) if imv_eval.higher_price else None
            anchor_label = "оценке Avito IMV"
            avito_imv_summary = AvitoImvSummary(
                recommended_price=anchor_total,
                lower_price=int(imv_eval.lower_price) if imv_eval.lower_price else None,
                higher_price=anchor_higher,
                market_count=imv_eval.market_count,
            )

        if anchor_total is not None:
            new_median, new_range_high, new_ppm2, blended, anchor_used = _apply_imv_blend(
                median_price=median_price,
                range_high=range_high,
                median_ppm2=median_ppm2,
                area=payload.area_m2,
                anchor_total=anchor_total,
                anchor_higher=anchor_higher,
                weight=settings.estimate_imv_blend_weight,
                threshold=settings.estimate_imv_blend_threshold,
            )
            if blended:
                logger.info(
                    "imv_blend: median %d → %d (anchor=%d w=%.2f) range_high %d → %d",
                    median_price,
                    new_median,
                    anchor_used,
                    settings.estimate_imv_blend_weight,
                    range_high,
                    new_range_high,
                )
                median_price = new_median
                median_ppm2 = new_ppm2
                explanation = (explanation or "") + (
                    f" Оценка скорректирована по {anchor_label} "
                    f"({anchor_used / 1_000_000:.1f} млн ₽)."
                )
                sources_used_pre = sorted(set(sources_used_pre) | {"avito_imv"})
            # Диапазон расширяем даже если медиану не двигали (информативность).
            range_high = new_range_high

    # Display-only Avito IMV summary, когда headline построен same-building якорем
    # (IMV-blend выше пропущен). Якорь дома — primary; IMV остаётся cross-check в UI.
    if anchor_tier is not None and avito_imv_summary is None:
        imv_anchor_disp = _fetch_house_imv_anchor(
            db,
            target_house_id=target_house_id,
            rooms=payload.rooms,
            area=payload.area_m2,
        )
        if imv_anchor_disp is not None and imv_anchor_disp.get("recommended_price"):
            avito_imv_summary = AvitoImvSummary(
                recommended_price=int(imv_anchor_disp["recommended_price"]),
                lower_price=(
                    int(imv_anchor_disp["lower_price"])
                    if imv_anchor_disp.get("lower_price")
                    else None
                ),
                higher_price=(
                    int(imv_anchor_disp["higher_price"])
                    if imv_anchor_disp.get("higher_price")
                    else None
                ),
                market_count=(
                    int(imv_anchor_disp["market_count"])
                    if imv_anchor_disp.get("market_count")
                    else None
                ),
            )

    # 4c (cont.). expected_sold_* выводим ЗДЕСЬ — ПОСЛЕ #651 IMV-blend / SB-anchor,
    # которые могли поднять median_price/median_ppm2 и расширить range_high. Применяем
    # ratio к POST-якорным значениям → asking (median_price_rub) и sold
    # (expected_sold_price_rub) консистентны в HeroSummary, и в DB persist'ятся
    # свежие значения (no stale «скидки»). range_low берём как есть.
    # Headline (median_price/ppm²/range) — ASKING-space во ВСЕХ ветках (радиус,
    # SB-якорь, IMV-blend): якорь теперь берёт anchor_ppm2 (pre-haircut), blend
    # работает в asking. asking→sold скидка — ЕДИНСТВЕННЫМ механизмом per-rooms
    # ratio: expected_sold = headline × ratio → DISTINCT, строго ниже median (когда
    # ratio<1). Null-guard: нет ratio (нет migration-080 строки) → expected_sold_*
    # остаются None → UI не показывает «−N%» badge (не фабрикуем).
    if asking_to_sold_ratio is not None and listings_clean:
        expected_sold_per_m2 = round(median_ppm2 * asking_to_sold_ratio)
        expected_sold_price = round(median_price * asking_to_sold_ratio)
        expected_sold_range_low = round(range_low * asking_to_sold_ratio)
        expected_sold_range_high = round(range_high * asking_to_sold_ratio)

    # ── #652: ДКП-коридор реальных сделок (ADVISORY + soft sanity-bound) ─────
    dkp_corridor: DkpCorridor | None = None
    dkp_raw = _fetch_dkp_corridor(
        db,
        address=geo.full_address,
        rooms=payload.rooms,
        area=payload.area_m2,
    )
    if dkp_raw is not None:
        dkp_corridor = DkpCorridor(**dkp_raw)
        # Soft sanity-bound: если итоговая медиана ₽/м² заметно вне коридора —
        # текстовая пометка (без хард-клампа). slack ±25% — широко, чтобы не
        # шуметь на тонких выборках.
        slack = 0.25
        if median_ppm2 and dkp_raw["count"] >= 3:
            if median_ppm2 > dkp_raw["high_ppm2"] * (1.0 + slack):
                explanation = (explanation or "") + (
                    " Оценка выше коридора реальных сделок Росреестра по улице."
                )
            elif median_ppm2 < dkp_raw["low_ppm2"] * (1.0 - slack):
                explanation = (explanation or "") + (
                    " Оценка ниже коридора реальных сделок Росреестра по улице."
                )

    # ── #693: coarse-geo downgrade ──────────────────────────────────────────
    # Когда DaData дала только грубую точность (settlement/city/region/unknown,
    # qc_geo >= 2), PostGIS-радиус крутится вокруг центроида НП и собирает
    # аналоги из случайных частей города → уверенная «medium» вводит в
    # заблуждение. Помечаем как ориентировочную (low). КОНСЕРВАТИВНО:
    #   1) есть позитивный сигнал грубости (dadata.qc_geo >= 2 ИЛИ _geocode_is_coarse);
    #   2) НЕ сработал якорь «ТОГО ЖЕ ДОМА» (Tier A) — он стоит на реальных комплах
    #      дома (#691 path) и геоцентроид не важен. ВАЖНО (#693 QA-fail 2/3): Tier C
    #      (micro-radius ≤500 м) — это НЕ «тот же дом», а случайные квартиры в 500 м
    #      вокруг геокода. Если геокод грубый (мусор → ЕКБ-центроид), Tier C набирает
    #      ≥5 комплов в плотном центре и РАНЬШЕ блокировал downgrade (guard был
    #      `anchor_tier is None`) → канонический кейс «фывапролд 999» оставался medium.
    #      Теперь защищаем ТОЛЬКО Tier A; Tier C при грубом геокоде — тоже понижаем.
    #   3) есть радиусное число (median_price > 0) — что квалифицировать.
    #   ИСТОЧНИКИ сигнала грубости (OR): dadata.qc_geo>=2 (если DaData активна) ИЛИ
    #   _geocode_is_coarse(geo) — прод-фолбэк, работает БЕЗ DaData.
    dadata_coarse = dadata is not None and dadata.qc_geo is not None and dadata.qc_geo >= 2
    geo_coarse = _geocode_is_coarse(geo)
    if (dadata_coarse or geo_coarse) and median_price > 0:
        # #693 QA diag (2/3): на проде QA просил залогировать сигналы геокода для
        # негеокодируемых адресов — фиксируем, что увидел гейт.
        logger.info(
            "coarse-geo gate #693: dadata_coarse=%s geo_coarse=%s anchor_tier=%s "
            "median=%s geo.provider=%s geo.confidence=%s geo.full_address=%r geo=(%.5f,%.5f)",
            dadata_coarse,
            geo_coarse,
            anchor_tier,
            median_price,
            geo.provider,
            geo.confidence,
            geo.full_address,
            geo.lat,
            geo.lon,
        )
    if (dadata_coarse or geo_coarse) and anchor_tier != "A" and median_price > 0:
        # Уровень грубости берём из DaData (точнее), иначе generic «населённого пункта».
        if dadata_coarse:
            _coarse_label = {2: "населённого пункта", 3: "города", 4: "региона"}.get(
                dadata.qc_geo, "населённого пункта"
            )
        else:
            _coarse_label = "населённого пункта"
        confidence = "low"
        explanation = (explanation or "") + (
            f" Адрес определён лишь до уровня {_coarse_label} — точные координаты "
            "дома найти не удалось, поэтому оценка ориентировочная (аналоги взяты "
            "по широкой окрестности)."
        )

    # 5. Deals — ДКП-only sales (вторичка) из rosreestr_deals.
    #    Importer фильтрует doc_type='ДКП' (PR-A 2026-05-24), ДДУ застройщиков
    #    исключены — больше не скёюят median вторички ~110-120 К/м².
    deals = _fetch_deals(
        db,
        lat=geo.lat,
        lon=geo.lon,
        rooms=payload.rooms,
        area=payload.area_m2,
        radius_m=DEFAULT_RADIUS_M,
    )

    # 6. Сохраняем в trade_in_estimates
    estimate_id = uuid4()
    now = datetime.now(tz=UTC)
    expires_at = now + timedelta(hours=24)

    # #694: когда same-building якорь сработал, headline построен на комплах того
    # же дома (anchor_comps_used) — показываем ИХ, а не радиусные listings_clean
    # (другие/дешевле/пусто → premium headline «не подтверждён» аналогами).
    if anchor_tier is not None and anchor_comps_used:
        analogs_lots = [_anchor_comp_to_analog(c) for c in anchor_comps_used[:10]]
    else:
        analogs_lots = [_listing_to_analog(lot) for lot in listings_clean[:10]]
    deals_lots = [_deal_to_analog(d) for d in deals[:10]]
    freshness_pre = _compute_freshness_minutes(listings_clean)
    # DaData enrichment (PR Q1) — заполняется только если service отработал.
    # При DaData = None все колонки идут в DB как NULL (graceful).
    dadata_metro_json = (
        json.dumps(dadata.metro, ensure_ascii=False)
        if dadata is not None and dadata.metro
        else None
    )
    db.execute(
        text(
            """
            INSERT INTO trade_in_estimates (
                id, address, lat, lon,
                area_m2, rooms, floor, total_floors,
                year_built, house_type, repair_state, has_balcony,
                ownership_type, has_mortgage, client_name, client_phone,
                median_price, range_low, range_high, median_price_per_m2,
                confidence, confidence_explanation, n_analogs,
                analogs, actual_deals,
                sources_used, data_freshness_minutes,
                canonical_address, house_cadnum, house_fias_id,
                dadata_qc_geo, dadata_qc_house, dadata_metro,
                expected_sold_price, expected_sold_range_low,
                expected_sold_range_high, expected_sold_per_m2,
                asking_to_sold_ratio, ratio_basis,
                created_by,
                expires_at
            ) VALUES (
                CAST(:id AS uuid),
                :address, :lat, :lon,
                :area, :rooms, :floor, :total_floors,
                :year_built, :house_type, :repair_state, :has_balcony,
                :ownership_type, :has_mortgage, :client_name, :client_phone,
                :median_price, :range_low, :range_high, :median_ppm2,
                :confidence, :explanation, :n_analogs,
                CAST(:analogs_json AS jsonb),
                CAST(:deals_json AS jsonb),
                CAST(:sources_json AS jsonb),
                :freshness,
                :canonical_address, :house_cadnum, :house_fias_id,
                :dadata_qc_geo, :dadata_qc_house,
                CAST(:dadata_metro_json AS jsonb),
                :expected_sold_price, :expected_sold_range_low,
                :expected_sold_range_high, :expected_sold_per_m2,
                :asking_to_sold_ratio, :ratio_basis,
                :created_by,
                :expires_at
            )
            """
        ),
        {
            "id": str(estimate_id),
            "address": geo.full_address,
            "lat": geo.lat,
            "lon": geo.lon,
            "area": payload.area_m2,
            "rooms": payload.rooms,
            "floor": payload.floor,
            "total_floors": payload.total_floors,
            "year_built": target_year,
            "house_type": target_house_type,
            "repair_state": payload.repair_state,
            "has_balcony": payload.has_balcony,
            "ownership_type": payload.ownership_type,
            "has_mortgage": payload.has_mortgage,
            "client_name": payload.client_name,
            "client_phone": payload.client_phone,
            "median_price": median_price,
            "range_low": range_low,
            "range_high": range_high,
            "median_ppm2": int(median_ppm2),
            "confidence": confidence,
            "explanation": explanation,
            "n_analogs": n_analogs,
            "analogs_json": json.dumps(
                [a.model_dump(mode="json") for a in analogs_lots], ensure_ascii=False
            ),
            "deals_json": json.dumps(
                [a.model_dump(mode="json") for a in deals_lots], ensure_ascii=False
            ),
            "sources_json": json.dumps(sources_used_pre, ensure_ascii=False),
            "freshness": freshness_pre,
            "canonical_address": dadata.canonical_address if dadata else None,
            "house_cadnum": dadata.house_cadnum if dadata else None,
            "house_fias_id": dadata.house_fias_id if dadata else None,
            "dadata_qc_geo": dadata.qc_geo if dadata else None,
            "dadata_qc_house": dadata.qc_house if dadata else None,
            "dadata_metro_json": dadata_metro_json,
            "expected_sold_price": expected_sold_price,
            "expected_sold_range_low": expected_sold_range_low,
            "expected_sold_range_high": expected_sold_range_high,
            "expected_sold_per_m2": expected_sold_per_m2,
            "asking_to_sold_ratio": asking_to_sold_ratio,
            "ratio_basis": ratio_basis,
            "created_by": created_by,
            "expires_at": expires_at,
        },
    )

    # Link saved IMV evaluation к этому estimate_id атомарно с основным INSERT
    # (closes finding #4 from 2026-05-24 audit — prior code committed estimate first,
    # then UPDATEd IMV in a separate tx, racing against concurrent estimators
    # sharing the same cache_key).
    if imv_eval is not None:
        db.execute(
            text(
                """
                UPDATE avito_imv_evaluations
                SET estimate_id = CAST(:estimate_id AS uuid)
                WHERE cache_key = :cache_key
                  AND (estimate_id IS NULL OR estimate_id = CAST(:estimate_id AS uuid))
                """
            ),
            {"estimate_id": str(estimate_id), "cache_key": imv_eval.cache_key},
        )

    db.commit()

    logger.info(
        "estimate: id=%s addr=%s rooms=%d area=%.1f → median=%d (n=%d, conf=%s)%s%s",
        estimate_id,
        geo.full_address[:60],
        payload.rooms,
        payload.area_m2,
        median_price,
        n_analogs,
        confidence,
        f" imv={imv_eval.recommended_price}" if imv_eval else "",
        f" cian={cian_val.sale_price_rub}" if cian_val and cian_val.sale_price_rub else "",
    )

    sources_used = sorted({lot.source for lot in analogs_lots if lot.source})
    if imv_eval is not None:
        sources_used = sorted(set(sources_used) | {"avito_imv"})
    if yandex_val is not None:
        sources_used = sorted(set(sources_used) | {"yandex_valuation"})
    if cian_val is not None and cian_val.sale_price_rub:
        sources_used = sorted(set(sources_used) | {"cian_valuation"})
    freshness_min = _compute_freshness_minutes(listings_clean)
    last_scraped_at = _compute_last_scraped_at(listings_clean)
    # Месячный ₽/м² тренд целевого дома (web TREND chart) — best-effort, None если нет данных.
    price_trend_raw = _fetch_price_trend(db, target_house_id=target_house_id)
    price_trend = (
        [PriceTrendPoint(month=p["month"], ppm2=p["ppm2"]) for p in price_trend_raw]
        if price_trend_raw
        else None
    )

    return AggregatedEstimate(
        estimate_id=estimate_id,
        median_price_rub=median_price,
        range_low_rub=range_low,
        range_high_rub=range_high,
        median_price_per_m2=int(median_ppm2),
        confidence=confidence,
        confidence_explanation=explanation,
        n_analogs=n_analogs,
        period_months=DEALS_PERIOD_MONTHS,
        analogs=analogs_lots,
        actual_deals=deals_lots,
        expires_at=expires_at,
        target_address=geo.full_address,
        target_lat=geo.lat,
        target_lon=geo.lon,
        sources_used=sources_used,
        data_freshness_minutes=freshness_min,
        last_scraped_at=last_scraped_at,
        price_trend=price_trend,
        est_days_on_market=_estimate_days_on_market(listings_clean, deals),
        cian_valuation=(
            CianValuationSummary(
                sale_price_rub=int(cian_val.sale_price_rub) if cian_val.sale_price_rub else None,
                rent_price_rub=int(cian_val.rent_price_rub) if cian_val.rent_price_rub else None,
                chart=[
                    {
                        "date": p.get("month_date") or p.get("date") or "",
                        "price": p["price"],
                    }
                    for p in (cian_val.chart or [])
                    if p.get("price") is not None
                ],
                chart_change_pct=cian_val.chart_change_pct,
                chart_change_direction=(
                    cian_val.chart_change_direction
                    if cian_val.chart_change_direction in {"increase", "decrease", "neutral"}
                    else None
                ),
            )
            if cian_val is not None
            else None
        ),
        avito_imv=avito_imv_summary,
        dkp_corridor=dkp_corridor,
        expected_sold_price_rub=expected_sold_price,
        expected_sold_range_low_rub=expected_sold_range_low,
        expected_sold_range_high_rub=expected_sold_range_high,
        expected_sold_per_m2=expected_sold_per_m2,
        asking_to_sold_ratio=asking_to_sold_ratio,
        ratio_basis=ratio_basis,
        area_m2=payload.area_m2,
        rooms=payload.rooms,
        floor=payload.floor,
        total_floors=payload.total_floors,
        year_built=target_year,
        house_type=target_house_type,
        repair_state=payload.repair_state,
        has_balcony=payload.has_balcony,
        canonical_address=dadata.canonical_address if dadata else None,
        house_cadnum=dadata.house_cadnum if dadata else None,
        house_fias_id=dadata.house_fias_id if dadata else None,
        metro_nearest=(dadata.metro if dadata and dadata.metro else []),
        address_precision=_qc_geo_to_precision(dadata.qc_geo if dadata else None),
    )


def _qc_geo_to_precision(qc_geo: int | None) -> str | None:
    # DaData qc_geo: 0=exact(house), 1=street, 2=settlement, 3=city, 4=region, 5=unknown
    if qc_geo is None:
        return None
    if qc_geo == 0:
        return "house"
    if qc_geo == 1:
        return "street"
    return "approximate"


def _estimate_days_on_market(
    listings: list[dict[str, Any]], deals: list[dict[str, Any]]
) -> int | None:
    """Прогноз срока продажи — медиана days_on_market по аналогам/сделкам.

    Возвращает None если ни у одного аналога нет данных о сроке экспозиции
    (наши парсеры не всегда его отдают — честно показываем «нет данных»).
    """
    values = [
        int(lot["days_on_market"])
        for lot in (*listings, *deals)
        if lot.get("days_on_market") and int(lot["days_on_market"]) > 0
    ]
    if len(values) < 3:
        return None
    values.sort()
    return values[len(values) // 2]


def _compute_freshness_minutes(lots: list[dict[str, Any]]) -> int | None:
    """Минут с последнего парсинга — для UI «обновлено N мин назад»."""
    if not lots:
        return None
    from datetime import datetime as _dt

    now = _dt.now(tz=UTC)
    scraped = [lot.get("scraped_at") or lot.get("listing_date") for lot in lots]
    scraped_dt: list[datetime] = []
    for s in scraped:
        if s is None:
            continue
        # listings rows из mappings — scraped_at это datetime, не date
        if hasattr(s, "tzinfo"):
            scraped_dt.append(s if s.tzinfo else s.replace(tzinfo=UTC))
    if not scraped_dt:
        return None
    return int((now - max(scraped_dt)).total_seconds() / 60)


def _compute_last_scraped_at(lots: list[dict[str, Any]]) -> datetime | None:
    """Абсолютный timestamp самого свежего парсинга среди аналогов (для UI).

    Дополняет _compute_freshness_minutes (относительные минуты): отдаёт точную
    дату/время, чтобы фронт мог отрендерить «обновлено DD.MM HH:MM». None если
    ни у одного лота нет scraped_at/listing_date с tzinfo (graceful)."""
    if not lots:
        return None
    scraped = [lot.get("scraped_at") or lot.get("listing_date") for lot in lots]
    scraped_dt: list[datetime] = []
    for s in scraped:
        if s is None:
            continue
        if hasattr(s, "tzinfo"):
            scraped_dt.append(s if s.tzinfo else s.replace(tzinfo=UTC))
    return max(scraped_dt) if scraped_dt else None


def _fetch_price_trend(
    db: Session,
    *,
    target_house_id: int | None,
    months: int = 24,
    min_points: int = 3,
) -> list[dict[str, Any]] | None:
    """Месячный ₽/м² тренд для целевого дома (web TREND chart) — best-effort.

    Предпочитает `houses_price_dynamics` (house_id, month_date, price_per_sqm) —
    готовая помесячная серия. В prod эта таблица пока ПУСТА, поэтому fallback —
    агрегация `house_placement_history` по месяцам (медиана COALESCE(last_price,
    start_price)/area_m2, дата = COALESCE(last_price_date, start_price_date)).

    Возвращает список [{month: 'YYYY-MM', ppm2: int}, ...] (≤ `months` точек,
    ASC по месяцу) или None если house_id не задан / точек < `min_points` /
    любая ошибка (graceful — фронт скрывает chart, без регрессий).
    """
    if target_house_id is None:
        return None

    # ── Source 1 (preferred): houses_price_dynamics ──────────────────────────
    try:
        rows = (
            db.execute(
                text(
                    """
                    SELECT to_char(month_date, 'YYYY-MM') AS month,
                           round(
                               percentile_cont(0.5) WITHIN GROUP (ORDER BY price_per_sqm)
                           )::int AS ppm2
                    FROM houses_price_dynamics
                    WHERE house_id = CAST(:hid AS bigint)
                      AND price_per_sqm > 0
                      AND month_date > (CURRENT_DATE
                          - (CAST(:months AS integer) || ' months')::interval)
                    GROUP BY month_date
                    ORDER BY month_date ASC
                    """
                ),
                {"hid": target_house_id, "months": months},
            )
            .mappings()
            .all()
        )
    except Exception as exc:  # pragma: no cover — defensive
        logger.warning("price_trend houses_price_dynamics lookup failed (graceful): %s", exc)
        try:
            db.rollback()
        except Exception:
            pass
        rows = []

    trend = [{"month": r["month"], "ppm2": int(r["ppm2"])} for r in rows if r["ppm2"]]
    if len(trend) >= min_points:
        logger.info(
            "price_trend house_id=%s source=houses_price_dynamics → %d points",
            target_house_id,
            len(trend),
        )
        return trend

    # ── Source 2 (fallback): aggregate house_placement_history by month ──────
    try:
        rows = (
            db.execute(
                text(
                    """
                    SELECT to_char(
                               date_trunc('month',
                                          COALESCE(last_price_date, start_price_date)),
                               'YYYY-MM'
                           ) AS month,
                           round(
                               percentile_cont(0.5) WITHIN GROUP (
                                   ORDER BY COALESCE(last_price, start_price)
                                            / NULLIF(area_m2, 0)
                               )
                           )::int AS ppm2
                    FROM house_placement_history
                    WHERE house_id = CAST(:hid AS bigint)
                      AND area_m2 > 0
                      AND COALESCE(last_price, start_price) > 0
                      AND COALESCE(last_price_date, start_price_date) IS NOT NULL
                      AND COALESCE(last_price_date, start_price_date) > (CURRENT_DATE
                          - (CAST(:months AS integer) || ' months')::interval)
                    GROUP BY 1
                    ORDER BY 1 ASC
                    """
                ),
                {"hid": target_house_id, "months": months},
            )
            .mappings()
            .all()
        )
    except Exception as exc:  # pragma: no cover — defensive
        logger.warning("price_trend house_placement_history lookup failed (graceful): %s", exc)
        try:
            db.rollback()
        except Exception:
            pass
        rows = []

    trend = [{"month": r["month"], "ppm2": int(r["ppm2"])} for r in rows if r["ppm2"]]
    if len(trend) >= min_points:
        logger.info(
            "price_trend house_id=%s source=house_placement_history → %d points",
            target_house_id,
            len(trend),
        )
        return trend

    logger.info(
        "price_trend house_id=%s → only %d points (<%d) → None",
        target_house_id,
        len(trend),
        min_points,
    )
    return None


# ── Internals ────────────────────────────────────────────────────────────────

# Compiled regexes for _extract_short_addr — module-level for performance.

# Strips leading admin prefixes: «Россия», «Свердловская область», «г. Екатеринбург» etc.
_ADMIN_PREFIX_RE = re.compile(
    r"^(?:"
    r"\s*(?:Россия|РФ|Российская\s+Федерация)\s*,?\s*|"
    r"\s*[А-Яа-яёЁ][А-Яа-яёЁ\s-]+(?:\s+(?:обл(?:асть)?|р-н|район|округ|край|республика))\.?\s*,?\s*|"
    r"\s*(?:г(?:ород)?|гор)\.?\s*[А-Яа-яёЁ][А-Яа-яёЁ\s-]+\s*,?\s*|"
    r"\s*[А-Я][а-яё-]+(?:\s+[А-Я][а-яё-]+)?\s*,?\s*"
    r")+",
    flags=re.UNICODE,
)

# Recognizes start of a street keyword.
_STREET_START_RE = re.compile(
    r"(?:ул\.|улица|пр\.|пр-т|проспект|пер\.|переулок|"
    r"б-р|бульвар|ш\.|шоссе|наб\.|набережная|проезд|тракт|пл\.|площадь|"
    r"мкр\.?|микрорайон)\s+",
    flags=re.IGNORECASE | re.UNICODE,
)

# Drops trailing apartment / office / corpus noise from the end.
_TRAILING_NOISE_RE = re.compile(
    r"\s*,\s*(?:кв\.?\s*\d+|корп\.?\s*\w+|оф\.?\s*\d+|пом\.?\s*\d+|подъезд\s*\d+).*$",
    flags=re.IGNORECASE | re.UNICODE,
)


def _extract_short_addr(full_address: str | None) -> str | None:
    """Извлекает «улица + номер дома» из полного адреса для поиска в том же доме.

    Примеры:
      "Свердловская область, г. Екатеринбург, ул. Заводская, д. 44-а" → "ул. Заводская, д. 44-а"
      "Россия, Екатеринбург, ул. Малышева, 1"                          → "ул. Малышева, 1"
      "РФ, Свердловская обл., Екатеринбург, ул. Ленина, 5, кв. 12"    → "ул. Ленина, 5"
      "г. Екатеринбург, проспект Ленина, 50"                           → "проспект Ленина, 50"
      "Екатеринбург, ул. Крауля, 48/2"                                 → "ул. Крауля, 48/2"

    Алгоритм:
    1. Отрезаем trailing кв./корп./оф. noise.
    2. Ищем первый street-keyword токен (ул./пр./пер. и т.д.) — возвращаем с него.
    3. Fallback: агрессивно strip admin-prefix regex, вернуть остаток.
    4. None если строка пустая или нечего возвращать.
    """
    if not full_address:
        return None
    s = full_address.strip()
    s = _TRAILING_NOISE_RE.sub("", s)

    # Find first street-keyword position and return from there.
    m = _STREET_START_RE.search(s)
    if m:
        return s[m.start() :].strip(" ,.")

    # Fallback: strip known admin prefixes, return whatever remains.
    s = _ADMIN_PREFIX_RE.sub("", s)
    return s.strip(" ,.") or None


# Ищет keyword типа улицы (ул./улица/пр./проспект/...) в адресе.
# Работает для FORWARD и REVERSE форматов Nominatim.
_STREET_KW_RE = re.compile(
    r"(?<![А-Яа-яёЁa-zA-Z])"
    r"(?:ул\.|улица|пр\.|пр-т|проспект|пер\.|переулок|"
    r"б-р|бульвар|ш\.|шоссе|наб\.|набережная|проезд|тракт|"
    r"пл\.|площадь|мкр\.|мкр|микрорайон)"
    r"\s+",
    flags=re.IGNORECASE | re.UNICODE,
)

# После keyword: 1-3 слова имени улицы со стопом на запятую или номер дома.
# Поддерживает "8 Марта" (цифра + слово) и "Большая Конюшенная" (несколько слов).
_STREET_NAME_RE = re.compile(
    r"^([0-9]+\s+[А-Яа-яёЁ][А-Яа-яёЁ-]+"
    r"|[А-Яа-яёЁ][А-Яа-яёЁ-]+(?:\s+[А-Яа-яёЁ][А-Яа-яёЁ-]+){0,2})"
    r"(?=,|\s+(?:д\.?\s*)?\d|\s*$)",
    flags=re.UNICODE,
)


def extract_street_name(full_address: str | None) -> str | None:
    """Извлекает чистое имя улицы из адреса в FORWARD или REVERSE формате.

    Примеры:
      "Екатеринбург, ул. Космонавтов, 50"                      → "Космонавтов"
      "80, улица 8 Марта, Артек, ..., Россия"                  → "8 Марта"
      "проспект Ленина 50"                                      → "Ленина"
      "Россия, Екатеринбург, ул. Малышева, 1"                  → "Малышева"
      "ул. Большая Конюшенная, 25"                              → "Большая Конюшенная"
      ""                                                        → None

    Алгоритм:
    1. Ищем street-keyword (ул/улица/пр/проспект/...) — case-insensitive.
    2. После keyword берём 1-3 слова до запятой или номера дома.
    3. Если keyword не нашёлся — пытаемся первый capitalized токен с
       поиском до запятой или номера (fallback для адресов без keyword'а).

    Returns None если ничего не извлеклось.
    """
    if not full_address or not full_address.strip():
        return None

    s = full_address.strip()

    # 1. Keyword-based extraction (работает для обоих форматов: forward и reverse)
    m = _STREET_KW_RE.search(s)
    if m:
        rest = s[m.end() :].lstrip()
        nm = _STREET_NAME_RE.match(rest)
        if nm:
            return nm.group(1).strip()

    # 2. Fallback: нет keyword — пробуем первый capitalized токен
    # Используется для "Большая Конюшенная, 25" без "ул."
    nm = _STREET_NAME_RE.match(s)
    if nm:
        candidate = nm.group(1).strip()
        # Отсеиваем очевидные административные слова
        bad = {"Россия", "Москва", "Санкт-Петербург", "область", "район", "округ", "край"}
        if not any(b in candidate for b in bad):
            return candidate

    return None


def _stratify_candidates(candidates: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Стратифицированная выборка Approach B — гарантирует MIN_ANALOGS_PER_SOURCE слотов.

    Candidates должны быть уже отсортированы по relevance_score (ASC).
    """
    guaranteed: list[dict[str, Any]] = []
    guaranteed_ids: set[int] = set()
    by_source: dict[str, list[dict[str, Any]]] = {}
    for row in candidates:
        src = row.get("source") or "unknown"
        by_source.setdefault(src, []).append(row)

    for _src, src_rows in by_source.items():
        quota = min(len(src_rows), MIN_ANALOGS_PER_SOURCE)
        for row in src_rows[:quota]:
            if id(row) not in guaranteed_ids:
                guaranteed.append(row)
                guaranteed_ids.add(id(row))

    remaining_slots = 50 - len(guaranteed)
    remainder: list[dict[str, Any]] = []
    if remaining_slots > 0:
        for row in candidates:
            if id(row) not in guaranteed_ids:
                remainder.append(row)
                if len(remainder) >= remaining_slots:
                    break

    result = guaranteed + remainder
    result.sort(key=lambda r: r.get("relevance_score") or 0.0)
    return result[:50]


_ANALOG_SELECT_COLS = """
    source, source_url, address, lat, lon,
    rooms, area_m2, floor, total_floors,
    price_rub, price_per_m2,
    listing_date, days_on_market, photo_urls,
    scraped_at
"""

_COMMON_WHERE = """
    AND rooms = :rooms
    AND area_m2 BETWEEN :area_min AND :area_max
    AND is_active = true
    AND scraped_at > NOW() - (:fresh_days || ' days')::interval
    AND price_rub > 0
    -- Когортный фильтр (PR 10): применяется к Tier S и Tier H через _COMMON_WHERE.
    -- Tier W имеет свою копию этого блока в inline SQL. Если cohort_year_min IS NULL —
    -- фильтр прозрачен. CAST обязателен — psycopg3 prepared statement не выводит
    -- тип $N при IS NULL в predicate (см. PR #518 fix).
    AND (
        CAST(:cohort_year_min AS integer) IS NULL
        OR year_built IS NULL
        OR year_built BETWEEN CAST(:cohort_year_min AS integer)
                          AND CAST(:cohort_year_max AS integer)
    )
"""
# Note: Tier W has its own inline copy of the cohort clause (PR #519 line
# ~1280). Не удалять — Tier W не использует _COMMON_WHERE из-за inline
# relevance_score CASE expressions. Both code paths must stay in sync.


def _fetch_analogs(
    db: Session,
    *,
    lat: float,
    lon: float,
    rooms: int,
    area: float,
    radius_m: int,
    full_address: str | None = None,
    area_tolerance: float = AREA_TOLERANCE,
    year_built: int | None = None,
    house_type: str | None = None,
    total_floors: int | None = None,
    cohort_year_min: int | None = None,  # NEW: lower bound year_built inclusive
    cohort_year_max: int | None = None,  # NEW: upper bound year_built inclusive
    target_house_id: int | None = None,  # #6: canonical house for same-building Tier S
) -> tuple[list[dict[str, Any]], bool, str]:
    """SELECT аналогов — трёхуровневый house-match (S → H → W).

    **Tier S (same building):** сначала канонический match по house_id_fk
    (если задан target_house_id) — детерминированно «тот же дом»; fallback —
    address ILIKE prefix-match по short_addr. Если ≥3 результатов → tier='S'.

    **Tier H (same class):** PostGIS + rooms + area + year ±15 + total_floors ±30%.
    Если ≥5 результатов → возвращаем; tier='H'.
    Пропускается если year_built или total_floors неизвестны.

    **Tier W (wide / current):** текущая логика без year/floors WHERE фильтра.
    tier='W'.

    House-match relevance_score используется для сортировки в Tier H и W.

    Стратифицированная выборка (Approach B):
    1. SQL вытягивает до 300 кандидатов с per-address row_number (cap MAX_ANALOGS_PER_ADDRESS).
    2. Python гарантирует MIN_ANALOGS_PER_SOURCE слотов каждому live source.
    3. Оставшиеся слоты заполняются из остальных кандидатов по relevance.
    4. Итоговый список отсортирован по relevance, LIMIT 50.

    Когортный фильтр (PR 9): если переданы cohort_year_min/max — добавляется
    hard-filter WHERE year_built BETWEEN min AND max OR year_built IS NULL.
    NULL допускается чтобы не отсеивать листинги с неизвестным годом
    (типично для Avito anonymous-address объявлений).

    Returns:
        (list_of_listings_as_dicts, fallback_radius_used_flag, tier)
        tier: 'S' | 'H' | 'W'
    """
    area_min = area * (1 - area_tolerance)
    area_max = area * (1 + area_tolerance)
    base_params: dict[str, Any] = {
        "rooms": rooms,
        "area_min": area_min,
        "area_max": area_max,
        "fresh_days": LISTINGS_FRESH_DAYS,
        "max_per_addr": MAX_ANALOGS_PER_ADDRESS,
        "cohort_year_min": cohort_year_min,
        "cohort_year_max": cohort_year_max,
    }

    # ── Tier S (canonical): same building via house_id_fk ─────────────────────
    # #6: детерминированный «тот же дом» через канонический house-граф. 99%
    # listings слинкованы (listings.house_id_fk → houses.id), что несравнимо
    # надёжнее address-string match'а между разнородными источниками
    # (Avito/Cian/Yandex форматируют адрес по-разному). Если target_house_id
    # неизвестен или результатов <3 — падаем на address-prefix Tier S ниже.
    if target_house_id is not None:
        tier_sc_params = {**base_params, "target_house_id": target_house_id}
        tier_sc_rows = (
            db.execute(
                text(
                    f"""
                WITH base AS (
                    SELECT
                        {_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
                    FROM listings
                    WHERE house_id_fk = CAST(:target_house_id AS bigint)
                    {_COMMON_WHERE}
                )
                SELECT
                    source, source_url, address, lat, lon,
                    rooms, area_m2, floor, total_floors,
                    price_rub, price_per_m2,
                    listing_date, days_on_market, photo_urls,
                    scraped_at, distance_m, relevance_score
                FROM base
                WHERE rn_addr <= :max_per_addr
                ORDER BY scraped_at DESC
                LIMIT 300
                """
                ),
                tier_sc_params,
            )
            .mappings()
            .all()
        )

        tier_sc = [dict(r) for r in tier_sc_rows]
        if len(tier_sc) >= 3:
            logger.info(
                "analogs tier=S(canonical) house_id=%s → %d results",
                target_house_id,
                len(tier_sc),
            )
            return _stratify_candidates(tier_sc), radius_m > DEFAULT_RADIUS_M, "S"

    # ── Tier S (fallback): same building via address prefix ───────────────────
    short_addr = _extract_short_addr(full_address)

    if short_addr:
        tier_s_params = {
            **base_params,
            "short_addr_prefix": short_addr + "%",
        }

        tier_s_rows = (
            db.execute(
                text(
                    f"""
                WITH base AS (
                    SELECT
                        {_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
                    FROM listings
                    WHERE address ILIKE :short_addr_prefix
                    {_COMMON_WHERE}
                )
                SELECT
                    source, source_url, address, lat, lon,
                    rooms, area_m2, floor, total_floors,
                    price_rub, price_per_m2,
                    listing_date, days_on_market, photo_urls,
                    scraped_at, distance_m, relevance_score
                FROM base
                WHERE rn_addr <= :max_per_addr
                ORDER BY scraped_at DESC
                LIMIT 300
                """
                ),
                tier_s_params,
            )
            .mappings()
            .all()
        )

        tier_s = [dict(r) for r in tier_s_rows]
        if len(tier_s) >= 3:
            logger.info(
                "analogs tier=S addr_prefix=%r → %d results",
                short_addr,
                len(tier_s),
            )
            return _stratify_candidates(tier_s), radius_m > DEFAULT_RADIUS_M, "S"

    # ── Tier H: same class (year ±15, total_floors ±30%) ─────────────────────
    if year_built is not None and total_floors is not None:
        year_min = year_built - 15
        year_max = year_built + 15
        tf_min = math.floor(total_floors * 0.7)
        tf_max = math.ceil(total_floors * 1.3)

        tier_h_rows = (
            db.execute(
                text(
                    f"""
                WITH base AS (
                    SELECT
                        {_ANALOG_SELECT_COLS},
                        ST_Distance(geom::geography, ST_MakePoint(:lon, :lat)::geography)
                            AS distance_m,
                        (
                            ST_Distance(geom::geography, ST_MakePoint(:lon, :lat)::geography)
                                / 1000.0
                            + CASE
                                WHEN year_built IS NOT NULL
                                THEN abs(year_built - CAST(:target_year AS integer)) / 12.0
                                ELSE 0
                              END
                            + CASE
                                WHEN CAST(:target_house_type AS text) IS NOT NULL
                                     AND house_type IS NOT NULL
                                     AND house_type <> CAST(:target_house_type AS text)
                                THEN 1.5
                                ELSE 0
                              END
                        ) AS relevance_score,
                        row_number() OVER (
                            PARTITION BY address
                            ORDER BY (
                                ST_Distance(geom::geography, ST_MakePoint(:lon, :lat)::geography)
                                    / 1000.0
                                + CASE
                                    WHEN year_built IS NOT NULL
                                    THEN abs(year_built - CAST(:target_year AS integer)) / 12.0
                                    ELSE 0
                                  END
                                + CASE
                                    WHEN CAST(:target_house_type AS text) IS NOT NULL
                                         AND house_type IS NOT NULL
                                         AND house_type <> CAST(:target_house_type AS text)
                                    THEN 1.5
                                    ELSE 0
                                  END
                            )
                        ) AS rn_addr
                    FROM listings
                    WHERE ST_DWithin(geom::geography, ST_MakePoint(:lon, :lat)::geography, :radius)
                    {_COMMON_WHERE}
                      AND total_floors BETWEEN CAST(:tf_min AS integer)
                                           AND CAST(:tf_max AS integer)
                      AND year_built BETWEEN CAST(:year_min AS integer)
                                        AND CAST(:year_max AS integer)
                )
                SELECT
                    source, source_url, address, lat, lon,
                    rooms, area_m2, floor, total_floors,
                    price_rub, price_per_m2,
                    listing_date, days_on_market, photo_urls,
                    scraped_at, distance_m, relevance_score
                FROM base
                WHERE rn_addr <= :max_per_addr
                ORDER BY relevance_score
                LIMIT 300
                """
                ),
                {
                    **base_params,
                    "lat": lat,
                    "lon": lon,
                    "radius": radius_m,
                    "target_year": year_built,
                    "target_house_type": house_type,
                    "tf_min": tf_min,
                    "tf_max": tf_max,
                    "year_min": year_min,
                    "year_max": year_max,
                },
            )
            .mappings()
            .all()
        )

        tier_h = [dict(r) for r in tier_h_rows]
        if len(tier_h) >= 5:
            logger.info(
                "analogs tier=H year=%d±15 tf=%d-%d → %d results",
                year_built,
                tf_min,
                tf_max,
                len(tier_h),
            )
            return _stratify_candidates(tier_h), radius_m > DEFAULT_RADIUS_M, "H"

        logger.info(
            "analogs tier=H year=%d±15 tf=%d-%d → only %d (fallthrough to W)",
            year_built,
            tf_min,
            tf_max,
            len(tier_h),
        )

    # ── Tier W: wide (current logic, year/floors only in relevance sort) ──────
    tier_w_rows = (
        db.execute(
            text(
                """
            WITH base AS (
                SELECT
                    source, source_url, address, lat, lon,
                    rooms, area_m2, floor, total_floors,
                    price_rub, price_per_m2,
                    listing_date, days_on_market, photo_urls,
                    scraped_at,
                    ST_Distance(geom::geography, ST_MakePoint(:lon, :lat)::geography)
                        AS distance_m,
                    (
                        ST_Distance(geom::geography, ST_MakePoint(:lon, :lat)::geography)
                            / 1000.0
                        -- CAST обязателен: target_year / target_house_type приходят NULL
                        -- без типа → PostgreSQL "could not determine data type of parameter"
                        -- (AmbiguousParameter). Явный тип снимает неоднозначность.
                        + CASE
                            WHEN CAST(:target_year AS integer) IS NOT NULL
                                 AND year_built IS NOT NULL
                            THEN abs(year_built - CAST(:target_year AS integer)) / 12.0
                            ELSE 0
                          END
                        + CASE
                            WHEN CAST(:target_house_type AS text) IS NOT NULL
                                 AND house_type IS NOT NULL
                                 AND house_type <> CAST(:target_house_type AS text)
                            THEN 1.5
                            ELSE 0
                          END
                    ) AS relevance_score,
                    row_number() OVER (
                        PARTITION BY address
                        ORDER BY (
                            ST_Distance(geom::geography, ST_MakePoint(:lon, :lat)::geography)
                                / 1000.0
                            + CASE
                                WHEN CAST(:target_year AS integer) IS NOT NULL
                                     AND year_built IS NOT NULL
                                THEN abs(year_built - CAST(:target_year AS integer)) / 12.0
                                ELSE 0
                              END
                            + CASE
                                WHEN CAST(:target_house_type AS text) IS NOT NULL
                                     AND house_type IS NOT NULL
                                     AND house_type <> CAST(:target_house_type AS text)
                                THEN 1.5
                                ELSE 0
                              END
                        )
                    ) AS rn_addr
                FROM listings
                WHERE ST_DWithin(geom::geography, ST_MakePoint(:lon, :lat)::geography, :radius)
                  AND rooms = :rooms
                  AND area_m2 BETWEEN :area_min AND :area_max
                  AND is_active = true
                  AND scraped_at > NOW() - (:fresh_days || ' days')::interval
                  AND price_rub > 0
                  -- Когортный фильтр (PR 9): отсеивает разные эпохи застройки
                  -- (хрущёвка vs новостройка). Если cohort_year_min IS NULL —
                  -- фильтр прозрачен. CAST обязателен — psycopg3 prepared statement
                  -- не выводит тип $N при IS NULL в predicate (см. PR #518 fix).
                  AND (
                      CAST(:cohort_year_min AS integer) IS NULL
                      OR year_built IS NULL
                      OR year_built BETWEEN CAST(:cohort_year_min AS integer)
                                        AND CAST(:cohort_year_max AS integer)
                  )
                  -- 2026-05-23: Avito coords теперь real (PR #487 убрал jitter после
                  -- C-5 audit). Listings с NULL coords отфильтруются через ST_DWithin
                  -- (geom IS NULL → не matches). geocode-missing-listings backfill
                  -- подтягивает координаты для address-only Avito листингов.
            )
            SELECT
                source, source_url, address, lat, lon,
                rooms, area_m2, floor, total_floors,
                price_rub, price_per_m2,
                listing_date, days_on_market, photo_urls,
                scraped_at,
                distance_m,
                relevance_score
            FROM base
            WHERE rn_addr <= :max_per_addr
            ORDER BY relevance_score
            LIMIT 300
            """
            ),
            {
                "lat": lat,
                "lon": lon,
                "radius": radius_m,
                "rooms": rooms,
                "area_min": area_min,
                "area_max": area_max,
                "fresh_days": LISTINGS_FRESH_DAYS,
                "target_year": year_built,
                "target_house_type": house_type,
                "max_per_addr": MAX_ANALOGS_PER_ADDRESS,
                "cohort_year_min": cohort_year_min,  # NEW
                "cohort_year_max": cohort_year_max,  # NEW
            },
        )
        .mappings()
        .all()
    )

    candidates: list[dict[str, Any]] = [dict(r) for r in tier_w_rows]
    logger.info("analogs tier=W radius=%dm → %d candidates", radius_m, len(candidates))
    return _stratify_candidates(candidates), radius_m > DEFAULT_RADIUS_M, "W"


def _is_plausible_deal(
    price_per_m2: float | None, floor: int | None, total_floors: int | None
) -> bool:
    """#699: True если ДКП-сделка правдоподобна (не выброс по ppm²/этажу).

    Абсолютные guard-bands (см. DEAL_* константы). None-поля не судим (keep —
    нечем сравнивать). Этаж > total_floors физически невозможен → drop.
    """
    if price_per_m2 is not None and not (DEAL_MIN_PPM2 <= price_per_m2 <= DEAL_MAX_PPM2):
        return False
    if floor is not None:
        if floor < 1 or floor > DEAL_MAX_FLOOR:
            return False
        if total_floors is not None and total_floors > 0 and floor > total_floors:
            return False
    return True


def _fetch_deals(
    db: Session, *, lat: float, lon: float, rooms: int, area: float, radius_m: int
) -> list[dict[str, Any]]:
    rows = (
        db.execute(
            text(
                """
            SELECT
                source, address, lat, lon,
                rooms, area_m2, floor, total_floors,
                price_rub, price_per_m2,
                deal_date, days_on_market,
                kadastr_num,
                ST_Distance(geom::geography, ST_MakePoint(:lon, :lat)::geography) AS distance_m
            FROM deals
            WHERE ST_DWithin(geom::geography, ST_MakePoint(:lon, :lat)::geography, :radius)
              AND rooms = :rooms
              AND area_m2 BETWEEN :area_min AND :area_max
              AND deal_date > NOW() - (:months || ' months')::interval
              AND price_rub > 0
            ORDER BY deal_date DESC
            LIMIT 30
            """
            ),
            {
                "lat": lat,
                "lon": lon,
                "radius": radius_m,
                "rooms": rooms,
                "area_min": area * (1 - AREA_TOLERANCE),
                "area_max": area * (1 + AREA_TOLERANCE),
                "months": DEALS_PERIOD_MONTHS,
            },
        )
        .mappings()
        .all()
    )

    # #699: отсекаем ДКП-выбросы (битый этаж / нерыночный ppm²) до выдачи в
    # actual_deals и expected_sold — иначе floor:100 / 39.7К-м² шумят демо.
    deals = [dict(r) for r in rows]
    clean = [
        d
        for d in deals
        if _is_plausible_deal(d.get("price_per_m2"), d.get("floor"), d.get("total_floors"))
    ]
    if len(clean) < len(deals):
        logger.info("deals sanitize #699: %d → %d (dropped outliers)", len(deals), len(clean))
    return clean


def _filter_outliers(lots: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Tukey IQR rule: исключаем точки вне [Q1 - 1.5×IQR, Q3 + 1.5×IQR]."""
    if len(lots) < 5:
        return lots  # на маленькой выборке нечего фильтровать

    prices = sorted(lot["price_per_m2"] for lot in lots if lot.get("price_per_m2"))
    if len(prices) < 4:
        return lots

    q1 = _percentile(prices, 0.25)
    q3 = _percentile(prices, 0.75)
    iqr = q3 - q1
    low = q1 - 1.5 * iqr
    high = q3 + 1.5 * iqr

    # None-safe: listings.price_per_m2 is nullable, и lot.get(..., 0) вернёт None
    # (а не дефолт 0) когда ключ ПРИСУТСТВУЕТ со значением None → low <= None <= high
    # бросает TypeError в Python 3. Лоты без цены судить как outlier нечем — оставляем их.
    clean = []
    for lot in lots:
        ppm2 = lot.get("price_per_m2")
        if ppm2 is None:
            clean.append(lot)  # нечего сравнивать — keep
            continue
        if low <= ppm2 <= high:
            clean.append(lot)  # priced лот внутри Tukey-границ — keep
        # else: priced outlier за пределами границ — drop
    if len(clean) < len(lots):
        logger.info("outlier filter: %d → %d (Q1=%d Q3=%d)", len(lots), len(clean), q1, q3)
    return clean


def _percentile(sorted_values: list[float], p: float) -> float:
    """Linear interpolation percentile (не округляем — оставляем float)."""
    if not sorted_values:
        return 0.0
    if len(sorted_values) == 1:
        return float(sorted_values[0])
    n = len(sorted_values)
    rank = p * (n - 1)
    lo = int(rank)
    hi = min(lo + 1, n - 1)
    frac = rank - lo
    return sorted_values[lo] + (sorted_values[hi] - sorted_values[lo]) * frac


def _compute_confidence(
    n_analogs: int,
    median_ppm2: float,
    q1: float,
    q3: float,
    fallback_radius_used: bool,
    area_widened: bool = False,
    listings: list[dict] | None = None,
) -> tuple[str, str]:
    """Confidence + explanation string.

    Уровень определяется по количеству уникальных адресов, а не по raw n_analogs.
    Это защищает от overstated confidence когда много лотов из одного здания
    (например, MIN_ANALOGS_PER_SOURCE=5 + same-building bias).

    high   — unique_addr ≥ 7 AND IQR/median < 0.15
    medium — unique_addr ≥ 4 OR (unique_addr ≥ 2 AND IQR/median < 0.25)
    low    — иначе

    Downgrade на один уровень если avg_lots_per_addr > 2.5 (concentration bias).
    """
    if median_ppm2 == 0:
        return "low", "Не найдено аналогов — попробуйте уточнить адрес или расширить параметры."

    # Вычисляем метрики уникальных адресов
    if listings:
        unique_addrs = {
            (lot.get("address") or "").strip().lower() for lot in listings if lot.get("address")
        }
        unique_addr_count = len(unique_addrs)
        avg_lots_per_addr = n_analogs / max(unique_addr_count, 1)
    else:
        unique_addr_count = n_analogs  # fallback: считаем каждый лот уникальным
        avg_lots_per_addr = 1.0

    iqr = q3 - q1
    iqr_pct = iqr / median_ppm2 if median_ppm2 > 0 else 1.0
    notes = []
    if fallback_radius_used:
        notes.append("расширили радиус до 2 км")
    if area_widened:
        notes.append("расширили допуск по площади до ±25%")
    fallback_note = f" ({', '.join(notes)} из-за нехватки данных)" if notes else ""

    # Базовый уровень по уникальным адресам
    if unique_addr_count >= 7 and iqr_pct < 0.15:
        base = "high"
    elif unique_addr_count >= 4:
        base = "medium"
    elif unique_addr_count >= 2 and iqr_pct < 0.25:
        base = "medium"
    else:
        base = "low"

    # Downgrade на один шаг если слишком много лотов сконцентрировано на малом числе адресов
    if avg_lots_per_addr > 2.5 and base != "low":
        downgrade_map = {"high": "medium", "medium": "low"}
        downgraded = downgrade_map[base]
        explanation = (
            f"Найдено {n_analogs} аналогов из {unique_addr_count} разных адресов, "
            f"разброс цены ±{int(iqr_pct * 100 / 2)}% от медианы{fallback_note}. "
            f"Снижена точность (≥2.5 лотов на адрес — возможен bias)."
        )
        return downgraded, explanation

    explanation = (
        f"Найдено {n_analogs} аналогов из {unique_addr_count} разных адресов, "
        f"разброс цены ±{int(iqr_pct * 100 / 2)}% от медианы{fallback_note}."
    )
    return base, explanation


def _listing_to_analog(row: dict[str, Any]) -> AnalogLot:
    return AnalogLot(
        address=row.get("address") or "",
        area_m2=float(row.get("area_m2") or 0),
        rooms=int(row.get("rooms") or 0),
        floor=row.get("floor"),
        total_floors=row.get("total_floors"),
        price_rub=int(row["price_rub"]),
        price_per_m2=int(row.get("price_per_m2") or 0),
        listing_date=row.get("listing_date"),
        days_on_market=row.get("days_on_market"),
        photo_url=(row["photo_urls"] or [None])[0] if row.get("photo_urls") else None,
        source=row.get("source"),
        source_url=row.get("source_url"),
        distance_m=int(row["distance_m"]) if row.get("distance_m") is not None else None,
        lat=float(row["lat"]) if row.get("lat") is not None else None,
        lon=float(row["lon"]) if row.get("lon") is not None else None,
    )


def _anchor_comp_to_analog(c: dict[str, Any]) -> AnalogLot:
    """Same-building/micro-radius comp → AnalogLot (#694).

    Когда якорь сработал, headline построен на этих комплах — показываем их в UI.
    ROBUST к отсутствию display-полей: реальные комплы несут address/source/price_rub
    (SELECT их тянет), но тестовые моки и legacy-строки могут давать только числа
    (price_per_m2/area_m2/rooms). price_rub тогда вычисляем ppm²×area
    (fallback на ppm² если area 0/None — без zero-result/crash). distance_m не
    значим для same-building → None, если явно не передан.
    """
    ppm2 = int(c.get("price_per_m2") or 0)
    area = float(c.get("area_m2") or 0)
    price_rub_raw = c.get("price_rub")
    if price_rub_raw is not None:
        price_rub = int(price_rub_raw)
    elif area > 0:
        price_rub = round(ppm2 * area)
    else:
        # area неизвестна — деградируем до ppm² (никогда 0/crash при ppm²>0).
        price_rub = ppm2
    photo_urls = c.get("photo_urls")
    return AnalogLot(
        address=c.get("address") or "",
        area_m2=area,
        rooms=int(c.get("rooms") or 0),
        floor=c.get("floor"),
        total_floors=c.get("total_floors"),
        price_rub=price_rub,
        price_per_m2=ppm2,
        listing_date=c.get("listing_date"),
        days_on_market=c.get("days_on_market"),
        photo_url=(photo_urls or [None])[0] if photo_urls else None,
        source=c.get("source"),
        source_url=c.get("source_url"),
        distance_m=int(c["distance_m"]) if c.get("distance_m") is not None else None,
        lat=float(c["lat"]) if c.get("lat") is not None else None,
        lon=float(c["lon"]) if c.get("lon") is not None else None,
    )


def _deal_to_analog(row: dict[str, Any]) -> AnalogLot:
    """deals не имеют photo_url — упрощённо.

    Tier classification (PR M / #564 Phase 3):
        T0_per_house — kadastr_num exact match (НЕ доступно в open dataset Росреестра)
        T1_per_street — street-level only (default для всех ДКП open dataset)
    """
    kad = row.get("kadastr_num")
    # Per-house tier требует kadastr_num типа "66:41:0204016:10" (с участком).
    # Street-only patterns: "66:41:0000000:0" или NULL → T1.
    tier = "T0_per_house" if kad and not kad.endswith(":0") else "T1_per_street"
    return AnalogLot(
        address=row.get("address") or "",
        area_m2=float(row.get("area_m2") or 0),
        rooms=int(row.get("rooms") or 0),
        floor=row.get("floor"),
        total_floors=row.get("total_floors"),
        price_rub=int(row["price_rub"]),
        price_per_m2=int(row.get("price_per_m2") or 0),
        listing_date=row.get("deal_date"),
        days_on_market=row.get("days_on_market"),
        photo_url=None,
        source=row.get("source"),
        source_url=None,  # rosreestr сделки без публичной ссылки
        distance_m=int(row["distance_m"]) if row.get("distance_m") is not None else None,
        lat=float(row["lat"]) if row.get("lat") is not None else None,
        lon=float(row["lon"]) if row.get("lon") is not None else None,
        tier=tier,
    )


def _empty_estimate(
    payload: TradeInEstimateInput, db: Session, *, reason: str, created_by: str | None = None
) -> AggregatedEstimate:
    """Fallback когда нет данных для оценки.

    Сохраняет запись в БД (confidence='low', пустые analogs/deals), чтобы GET /estimate/{id}
    не возвращал 404. C-4 security audit.
    """
    estimate_id = uuid4()
    now = datetime.now(tz=UTC)
    expires_at = now + timedelta(hours=24)

    db.execute(
        text(
            """
            INSERT INTO trade_in_estimates (
                id, address,
                area_m2, rooms, floor, total_floors,
                year_built, house_type, repair_state, has_balcony,
                ownership_type, has_mortgage, client_name, client_phone,
                median_price, range_low, range_high, median_price_per_m2,
                confidence, confidence_explanation, n_analogs,
                analogs, actual_deals,
                sources_used,
                created_by,
                expires_at
            ) VALUES (
                CAST(:id AS uuid), :address,
                :area, :rooms, :floor, :total_floors,
                :year_built, :house_type, :repair_state, :has_balcony,
                :ownership_type, :has_mortgage, :client_name, :client_phone,
                0, 0, 0, 0,
                'low', :explanation, 0,
                '[]'::jsonb, '[]'::jsonb,
                '[]'::jsonb,
                :created_by,
                :expires_at
            )
            """
        ),
        {
            "id": str(estimate_id),
            "address": payload.address,
            "area": payload.area_m2,
            "rooms": payload.rooms,
            "floor": payload.floor,
            "total_floors": payload.total_floors,
            "year_built": payload.year_built,
            "house_type": payload.house_type,
            "repair_state": payload.repair_state,
            "has_balcony": payload.has_balcony,
            "ownership_type": payload.ownership_type,
            "has_mortgage": payload.has_mortgage,
            "client_name": payload.client_name,
            "client_phone": payload.client_phone,
            "explanation": reason,
            "created_by": created_by,
            "expires_at": expires_at,
        },
    )
    db.commit()
    logger.info(
        "empty_estimate: id=%s reason=%s addr=%s", estimate_id, reason, payload.address[:60]
    )

    return AggregatedEstimate(
        estimate_id=estimate_id,
        median_price_rub=0,
        range_low_rub=0,
        range_high_rub=0,
        median_price_per_m2=0,
        confidence="low",
        confidence_explanation=reason,
        n_analogs=0,
        period_months=DEALS_PERIOD_MONTHS,
        analogs=[],
        actual_deals=[],
        expires_at=expires_at,
        cian_valuation=None,
        # Адрес не геокодирован (DaData не отрабатывала) → точность неизвестна.
        address_precision=None,
    )