gendesign/tradein-mvp/backend/app/services/buildings_query.py

"""SQL-билдеры страницы «доля квартир дома в продаже» (мигр. 143; окно 45д мигр. 146).

Читаем готовый view v_building_sale_share — логику доли НЕ пересчитываем.
Два процента на дом:
  - sale_share_pct      — active_secondary / flat_count_effective (только активные);
  - sale_share_pct_45d  — distinct vtorichka за последние 45д (активные ИЛИ недавно
    закрытые) / flat_count_effective.
Дом проходит фильтр, если ЛЮБОЙ из двух процентов в диапазоне [min_pct, max_pct].

Все запросы параметризованы (SQLAlchemy text + bind params, CAST(:x AS type) —
psycopg v3). Сортировка/колонка гистограммы — строго по whitelist, user-ввод в SQL
не интерполируется.
"""

from __future__ import annotations

# share_desc / share_45d_desc / active_desc / count_45d_desc / exposure_desc /
# price_asc / price_desc → безопасный ORDER BY. house_id как стабильный tiebreaker
# для детерминированной пагинации.
_SORT_SQL: dict[str, str] = {
    "share_desc": "sale_share_pct DESC NULLS LAST, house_id",
    "share_45d_desc": "sale_share_pct_45d DESC NULLS LAST, house_id",
    "active_desc": "active_secondary DESC NULLS LAST, house_id",
    "count_45d_desc": "listings_45d DESC NULLS LAST, house_id",
    "exposure_desc": "avg_days_on_market DESC NULLS LAST, house_id",
    "price_asc": "median_price_rub ASC NULLS LAST, house_id",
    "price_desc": "median_price_rub DESC NULLS LAST, house_id",
}
ALLOWED_SORTS: frozenset[str] = frozenset(_SORT_SQL)

# Гистограмма распределения процента — фиксированный порядок корзин.
HISTOGRAM_BUCKETS: tuple[str, ...] = (
    "0-5",
    "5-10",
    "10-20",
    "20-30",
    "30-50",
    "50-100",
    "100+",
)

# Колонки view, по которым можно строить гистограмму (whitelist против инъекции).
_HISTOGRAM_COLUMNS: frozenset[str] = frozenset({"sale_share_pct", "sale_share_pct_45d"})

# Колонки view, отдаваемые в список (в порядке SELECT). median_* округляем до
# bigint — percentile_cont отдаёт double precision (может быть x.5), а Pydantic
# int-поле на дробном float падает. over_100 — implausible-флаг по любому из %.
_LIST_COLUMNS = (
    "house_id, "
    "COALESCE(short_address, full_address, address) AS address, "
    "lat, lon, "
    "sale_share_pct, "
    "sale_share_pct_45d, "
    "(sale_share_pct > 100 OR sale_share_pct_45d > 100) AS over_100, "
    "active_secondary, "
    "listings_45d, "
    "flat_count_effective, "
    "gar_match_method, "
    "CAST(round(median_price_rub) AS bigint) AS median_price_rub, "
    "CAST(round(median_price_per_m2) AS bigint) AS median_price_per_m2, "
    "avg_days_on_market, "
    "year_built, house_type, total_floors, series_name, is_emergency"
)


def build_sale_share_query(
    *,
    min_pct: float = 5.0,
    max_pct: float | None = None,
    city: str | None = None,
    price_min: int | None = None,
    price_max: int | None = None,
    year_min: int | None = None,
    year_max: int | None = None,
    house_type: str | None = None,
    sort: str = "share_desc",
    limit: int = 200,
) -> tuple[str, dict[str, object]]:
    """SELECT домов из v_building_sale_share по фильтрам.

    Дом проходит, если ЛЮБОЙ из двух процентов (now / 45д) попадает в диапазон
    [min_pct, max_pct]. NULL-процент (неправдоподобный знаменатель) не проходит
    сравнение, поэтому отдельный `IS NOT NULL` не нужен — рядки без обоих % выпадают.
    """
    where: list[str] = []
    args: dict[str, object] = {}

    # Каждая ветка требует, чтобы ИМЕННО её % был в диапазоне; building qualifies
    # on either. Без max_pct сводится к `(pct_now>=min OR pct_45d>=min)`.
    args["min_pct"] = min_pct
    qual_now = "sale_share_pct >= CAST(:min_pct AS numeric)"
    qual_45d = "sale_share_pct_45d >= CAST(:min_pct AS numeric)"
    if max_pct is not None:
        args["max_pct"] = max_pct
        qual_now += " AND sale_share_pct <= CAST(:max_pct AS numeric)"
        qual_45d += " AND sale_share_pct_45d <= CAST(:max_pct AS numeric)"
    where.append(f"(({qual_now}) OR ({qual_45d}))")

    if city:
        # case-insensitive substring по всем адресным полям (город может быть в любом).
        where.append(
            "(COALESCE(short_address, '') || ' ' || COALESCE(full_address, '') "
            "|| ' ' || COALESCE(address, '')) ILIKE CAST(:city_like AS text)"
        )
        args["city_like"] = f"%{city}%"

    if price_min is not None:
        where.append("median_price_rub >= CAST(:price_min AS bigint)")
        args["price_min"] = price_min
    if price_max is not None:
        where.append("median_price_rub <= CAST(:price_max AS bigint)")
        args["price_max"] = price_max

    if year_min is not None:
        where.append("year_built >= CAST(:year_min AS integer)")
        args["year_min"] = year_min
    if year_max is not None:
        where.append("year_built <= CAST(:year_max AS integer)")
        args["year_max"] = year_max

    if house_type:
        where.append("house_type = CAST(:house_type AS text)")
        args["house_type"] = house_type

    order_sql = _SORT_SQL.get(sort, _SORT_SQL["share_desc"])
    args["limit"] = max(1, min(int(limit), 1000))

    sql = (
        f"SELECT {_LIST_COLUMNS} "
        "FROM v_building_sale_share "
        f"WHERE {' AND '.join(where)} "
        f"ORDER BY {order_sql} "
        "LIMIT CAST(:limit AS integer)"
    )
    return sql, args


def build_listings_query(house_id: int) -> tuple[str, dict[str, object]]:
    """Активные вторичные объявления одного дома (ORDER BY price_rub).

    Гео-фильтр (≤300м от geom дома) против mis-bucketing листингов в чужой дом —
    тот же порог, что в v_building_sale_share (мигр.150). Листинги/дома без geom — КЕЕП.
    """
    sql = (
        "SELECT l.id AS listing_id, l.source, l.source_url, l.price_rub, l.price_per_m2, "
        "l.rooms, CAST(l.area_m2 AS double precision) AS area_m2, l.floor, l.total_floors, "
        "l.days_on_market, l.listing_date, l.address "
        "FROM listings l JOIN houses h ON h.id = l.house_id_fk "
        "WHERE l.house_id_fk = CAST(:house_id AS bigint) "
        "AND l.is_active AND l.listing_segment = 'vtorichka' "
        "AND (l.geom IS NULL OR h.geom IS NULL OR ST_DistanceSphere(l.geom, h.geom) <= 300) "
        "ORDER BY l.price_rub ASC NULLS LAST, l.id"
    )
    return sql, {"house_id": house_id}


def build_summary_scalars_query() -> str:
    """Скаляры сводки: всего домов, с знаменателем (любой %), coverage, max/p95 обоих %."""
    return (
        "SELECT "
        "count(*) AS total_secondary_buildings, "
        "count(*) FILTER (WHERE sale_share_pct IS NOT NULL OR sale_share_pct_45d IS NOT NULL) "
        "AS buildings_with_denominator, "
        "COALESCE(round("
        "100.0 * count(*) FILTER "
        "(WHERE sale_share_pct IS NOT NULL OR sale_share_pct_45d IS NOT NULL) "
        "/ NULLIF(count(*), 0), 1), 0) AS coverage_pct, "
        "max(sale_share_pct) AS max_pct, "
        "percentile_cont(0.95) WITHIN GROUP (ORDER BY sale_share_pct) AS p95_pct, "
        "max(sale_share_pct_45d) AS max_pct_45d, "
        "percentile_cont(0.95) WITHIN GROUP (ORDER BY sale_share_pct_45d) AS p95_pct_45d "
        "FROM v_building_sale_share"
    )


def build_summary_histogram_query(column: str = "sale_share_pct") -> str:
    """Гистограмма по фиксированным корзинам (только non-null) для одного из двух %.

    column — whitelist {sale_share_pct, sale_share_pct_45d}; чужое → fallback на
    sale_share_pct (user-ввод в SQL не интерполируется).
    """
    col = column if column in _HISTOGRAM_COLUMNS else "sale_share_pct"
    return (
        "SELECT "
        "CASE "
        f"WHEN {col} < 5 THEN '0-5' "
        f"WHEN {col} < 10 THEN '5-10' "
        f"WHEN {col} < 20 THEN '10-20' "
        f"WHEN {col} < 30 THEN '20-30' "
        f"WHEN {col} < 50 THEN '30-50' "
        f"WHEN {col} <= 100 THEN '50-100' "
        "ELSE '100+' "
        "END AS bucket, "
        "count(*) AS count "
        "FROM v_building_sale_share "
        f"WHERE {col} IS NOT NULL "
        "GROUP BY bucket"
    )