gendesign/backend/app/services/analytics_queries.py

"""SQL queries for /api/v1/analytics endpoints.

One function per endpoint. All return plain dicts/lists ready for JSON.
Region 66 = Sverdlovskaya oblast. Developer 6208_0 = PRINZIP.
"""

from __future__ import annotations

from decimal import Decimal
from typing import Any

from sqlalchemy import text
from sqlalchemy.orm import Session


def _f(value: Any) -> float | None:
    if value is None:
        return None
    if isinstance(value, Decimal):
        return float(value)
    return value


def market_pulse(db: Session, region_code: int = 66) -> list[dict[str, Any]]:
    rows = (
        db.execute(
            text(
                """
            SELECT snapshot_date, rep_year, rep_month,
                   total_square, sold_perc, price_avg
              FROM domrf_realization
             WHERE region_code = :region_code
               AND endpoint_type = 'total'
               AND type_square = 'total'
             ORDER BY snapshot_date
            """
            ),
            {"region_code": region_code},
        )
        .mappings()
        .all()
    )
    return [
        {
            "snapshot_date": r["snapshot_date"].isoformat(),
            "rep_year": r["rep_year"],
            "rep_month": r["rep_month"],
            "total_square_th_sqm": _f(r["total_square"]),
            "sold_perc": _f(r["sold_perc"]),
            "price_avg": _f(r["price_avg"]),
        }
        for r in rows
    ]


def quartirography(db: Session, source: str, region_id: int = 66) -> list[dict[str, Any]]:
    """source: 'portfolio' (что строится) or 'deals' (реально покупают)."""
    if source == "portfolio":
        rows = (
            db.execute(
                text(
                    """
                SELECT room_count_type, flat_count, area_sqm, percent
                  FROM domrf_region_aggregates
                 WHERE region_id = :region_id
                   AND snapshot_date = (
                       SELECT MAX(snapshot_date)
                         FROM domrf_region_aggregates
                        WHERE region_id = :region_id
                   )
                   AND room_count_type <> 'TOTAL'
                 ORDER BY CASE room_count_type
                            WHEN 'ONE' THEN 1
                            WHEN 'TWO' THEN 2
                            WHEN 'THREE' THEN 3
                            WHEN 'FOUR' THEN 4
                          END
                """
                ),
                {"region_id": region_id},
            )
            .mappings()
            .all()
        )
        return [
            {
                "bucket": {
                    "ONE": "1-к",
                    "TWO": "2-к",
                    "THREE": "3-к",
                    "FOUR": "4+",
                }.get(r["room_count_type"], r["room_count_type"]),
                "flat_count": r["flat_count"],
                "area_sqm": _f(r["area_sqm"]),
                "percent": r["percent"],
                "avg_area": _f(r["area_sqm"] / r["flat_count"]) if r["flat_count"] else None,
            }
            for r in rows
        ]

    # deals: bucketize Rosreestr area into 5 segments (студия, 1-к, 2-к, 3-к, 4+).
    # Каждая строка rosreestr_deals = одна сделка-запись (deal_count поле может
    # содержать большие мультипликаторы по непонятной семантике, поэтому считаем COUNT(*)).
    rows = (
        db.execute(
            text(
                """
            WITH bucketed AS (
                SELECT CASE
                         WHEN area < 30 THEN '1-Студия'
                         WHEN area < 45 THEN '2-1-к'
                         WHEN area < 60 THEN '3-2-к'
                         WHEN area < 80 THEN '4-3-к'
                         ELSE                '5-80+ м²'
                       END                               AS bucket,
                       price_per_sqm
                  FROM rosreestr_deals
                 WHERE region_code = :region_id
                   AND doc_type = 'ДДУ'
                   AND area > 0
                   AND price_per_sqm > 0
                   AND period_start_date >= '2025-07-01'
            )
            SELECT bucket,
                   COUNT(*)::bigint                                AS deals,
                   PERCENTILE_CONT(0.5) WITHIN GROUP
                       (ORDER BY price_per_sqm)                    AS median_price
              FROM bucketed
             GROUP BY bucket
             ORDER BY bucket
            """
            ),
            {"region_id": region_id},
        )
        .mappings()
        .all()
    )
    pretty = {
        "1-Студия": "Студии 15-30",
        "2-1-к": "1-к 30-45",
        "3-2-к": "2-к 45-60",
        "4-3-к": "3-к 60-80",
        "5-80+ м²": "80+ м²",
    }
    total = sum(r["deals"] or 0 for r in rows) or 1
    return [
        {
            "bucket": pretty[r["bucket"]],
            "deals": int(r["deals"] or 0),
            "percent": round((r["deals"] or 0) * 100 / total, 1),
            "median_price": _f(r["median_price"]),
        }
        for r in rows
    ]


def pipeline_by_year(db: Session, region_code: int = 66) -> list[dict[str, Any]]:
    rows = (
        db.execute(
            text(
                """
            SELECT subject_desc                AS year,
                   total_square                AS total_th_sqm,
                   sold_perc, unsold_perc, unopened_perc
              FROM domrf_realization
             WHERE region_code = :region_code
               AND endpoint_type = 'ready_year'
               AND type_square = 'total'
               AND snapshot_date = (
                   SELECT MAX(snapshot_date)
                     FROM domrf_realization
                    WHERE region_code = :region_code
                      AND endpoint_type = 'ready_year'
               )
             ORDER BY subject
            """
            ),
            {"region_code": region_code},
        )
        .mappings()
        .all()
    )
    return [
        {
            "year": r["year"],
            "total_th_sqm": _f(r["total_th_sqm"]),
            "sold_perc": _f(r["sold_perc"]),
            "unsold_perc": _f(r["unsold_perc"]),
            "unopened_perc": _f(r["unopened_perc"]),
        }
        for r in rows
    ]


def districts(db: Session) -> list[dict[str, Any]]:
    rows = (
        db.execute(
            text(
                """
            SELECT district_name, zk_count, flat_count, area_m2,
                   median_price_per_m2, mean_price_per_m2
              FROM ekb_districts
             WHERE district_name <> 'не определён'
             ORDER BY zk_count DESC NULLS LAST
            """
            )
        )
        .mappings()
        .all()
    )
    return [
        {
            "district_name": r["district_name"],
            "zk_count": r["zk_count"],
            "flat_count": r["flat_count"],
            "area_m2": _f(r["area_m2"]),
            "median_price_per_m2": _f(r["median_price_per_m2"]),
            "mean_price_per_m2": _f(r["mean_price_per_m2"]),
        }
        for r in rows
    ]


def yandex_listings(db: Session) -> dict[str, Any]:
    rows = (
        db.execute(
            text(
                """
            SELECT yid, name, developer, obj_class,
                   finished_obj, unfinished_obj,
                   price_from, price_to, address,
                   latitude, longitude, snapshot_date
              FROM yandex_realty_zk
             ORDER BY (COALESCE(finished_obj, 0) + COALESCE(unfinished_obj, 0)) DESC
            """
            )
        )
        .mappings()
        .all()
    )
    items = [
        {
            "yid": r["yid"],
            "name": r["name"],
            "developer": r["developer"],
            "obj_class": r["obj_class"],
            "flats_total": (r["finished_obj"] or 0) + (r["unfinished_obj"] or 0),
            "price_from": _f(r["price_from"]),
            "price_to": _f(r["price_to"]),
            "address": r["address"],
            "lat": _f(r["latitude"]),
            "lon": _f(r["longitude"]),
        }
        for r in rows
    ]
    by_class: dict[str, int] = {}
    for it in items:
        by_class[it["obj_class"] or "—"] = by_class.get(it["obj_class"] or "—", 0) + 1
    return {
        "snapshot_date": rows[0]["snapshot_date"].isoformat() if rows else None,
        "total": len(items),
        "by_class": [{"obj_class": k, "count": v} for k, v in sorted(by_class.items())],
        "items": items,
    }


def top_developers(db: Session, region_code: int = 66, limit: int = 15) -> list[dict[str, Any]]:
    """Top developers in Sverdl by sqm + Δ sold% over the available history.

    Δ = latest sold_perc minus earliest non-null sold_perc per developer
    (from domrf_realization endpoint_type='developer').
    """
    rows = (
        db.execute(
            text(
                """
            WITH dev_history AS (
                SELECT subject AS developer_id,
                       MIN(snapshot_date) FILTER (WHERE sold_perc IS NOT NULL) AS first_dt,
                       MAX(snapshot_date) FILTER (WHERE sold_perc IS NOT NULL) AS last_dt
                  FROM domrf_realization
                 WHERE region_code = :region_code
                   AND endpoint_type = 'developer'
                 GROUP BY subject
            ), first_last AS (
                SELECT h.developer_id,
                       (SELECT sold_perc FROM domrf_realization r
                         WHERE r.region_code = :region_code
                           AND r.endpoint_type = 'developer'
                           AND r.subject = h.developer_id
                           AND r.snapshot_date = h.first_dt
                           AND r.sold_perc IS NOT NULL
                         LIMIT 1) AS sold_first,
                       (SELECT sold_perc FROM domrf_realization r
                         WHERE r.region_code = :region_code
                           AND r.endpoint_type = 'developer'
                           AND r.subject = h.developer_id
                           AND r.snapshot_date = h.last_dt
                           AND r.sold_perc IS NOT NULL
                         LIMIT 1) AS sold_last,
                       h.first_dt, h.last_dt
                  FROM dev_history h
            )
            SELECT m.developer_id, m.developer_name,
                   m.jk_count, m.jk_flats_total,
                   m.sverdl_sqm, m.sverdl_sold_pct,
                   m.avg_area_sqm, m.pct_one, m.pct_three_plus,
                   fl.sold_first, fl.sold_last,
                   (fl.sold_last - fl.sold_first)              AS sold_delta_pp,
                   fl.first_dt, fl.last_dt
              FROM v_developer_full_metrics m
              LEFT JOIN first_last fl ON fl.developer_id = m.developer_id
             WHERE m.sverdl_sqm IS NOT NULL
             ORDER BY m.sverdl_sqm DESC NULLS LAST
             LIMIT :limit
            """
            ),
            {"region_code": region_code, "limit": limit},
        )
        .mappings()
        .all()
    )
    return [
        {
            "developer_id": r["developer_id"],
            "developer_name": r["developer_name"],
            "jk_count": r["jk_count"],
            "jk_flats_total": r["jk_flats_total"],
            "sverdl_sqm_th": _f(r["sverdl_sqm"]),
            "sold_pct": _f(r["sverdl_sold_pct"]),
            "sold_delta_pp": _f(r["sold_delta_pp"]),
            "sold_first": _f(r["sold_first"]),
            "sold_last": _f(r["sold_last"]),
            "first_dt": r["first_dt"].isoformat() if r["first_dt"] else None,
            "last_dt": r["last_dt"].isoformat() if r["last_dt"] else None,
            "avg_area_sqm": _f(r["avg_area_sqm"]),
            "pct_one": _f(r["pct_one"]),
            "pct_three_plus": _f(r["pct_three_plus"]),
        }
        for r in rows
    ]


def developer_detail(db: Session, developer_id: str) -> dict[str, Any] | None:
    row = (
        db.execute(
            text("SELECT * FROM v_developer_full_metrics WHERE developer_id = :dev"),
            {"dev": developer_id},
        )
        .mappings()
        .first()
    )
    if not row:
        return None
    return {
        "developer_id": row["developer_id"],
        "developer_name": row["developer_name"],
        "jk_count": row["jk_count"],
        "jk_flats_total": row["jk_flats_total"],
        "jk_sqm_total": _f(row["jk_sqm_total"]),
        "jk_ekb": row["jk_ekb"],
        "jk_completed": row["jk_completed"],
        "jk_in_progress": row["jk_in_progress"],
        "jk_escrow": row["jk_escrow"],
        "agg_flats_total": row["agg_flats_total"],
        "agg_one_room": row["agg_one_room"],
        "agg_two_room": row["agg_two_room"],
        "agg_three_room": row["agg_three_room"],
        "agg_four_plus": row["agg_four_plus"],
        "pct_one": _f(row["pct_one"]),
        "pct_three_plus": _f(row["pct_three_plus"]),
        "avg_area_sqm": _f(row["avg_area_sqm"]),
        "sverdl_sqm_th": _f(row["sverdl_sqm"]),
        "sverdl_sold_pct": _f(row["sverdl_sold_pct"]),
        "sverdl_unsold_pct": _f(row["sverdl_unsold_pct"]),
        "sverdl_price_avg": _f(row["sverdl_price_avg"]),
    }


def developer_history(
    db: Session,
    developer_ids: list[str],
    region_code: int = 66,
) -> list[dict[str, Any]]:
    """Per-month sold_perc for one or more developers in the region."""
    rows = (
        db.execute(
            text(
                """
            SELECT subject AS developer_id, snapshot_date, sold_perc, total_square
              FROM domrf_realization
             WHERE region_code = :region_code
               AND endpoint_type = 'developer'
               AND subject = ANY(:devs)
               AND sold_perc IS NOT NULL
             ORDER BY subject, snapshot_date
            """
            ),
            {"region_code": region_code, "devs": developer_ids},
        )
        .mappings()
        .all()
    )
    return [
        {
            "developer_id": r["developer_id"],
            "snapshot_date": r["snapshot_date"].isoformat(),
            "sold_perc": _f(r["sold_perc"]),
            "total_th_sqm": _f(r["total_square"]),
        }
        for r in rows
    ]


def developer_portfolio(db: Session, developer_id: str) -> list[dict[str, Any]]:
    rows = (
        db.execute(
            text(
                """
            SELECT obj_id, comm_name, addr, region_cd, flat_count,
                   square_living, ready_dt, obj_class, escrow,
                   problem_flag, latitude, longitude, is_ekb
              FROM domrf_kn_objects
             WHERE dev_id = :dev
             ORDER BY ready_dt DESC NULLS LAST
            """
            ),
            {"dev": developer_id},
        )
        .mappings()
        .all()
    )
    return [
        {
            "obj_id": r["obj_id"],
            "comm_name": r["comm_name"],
            "addr": r["addr"],
            "region_cd": r["region_cd"],
            "flat_count": r["flat_count"],
            "square_living": _f(r["square_living"]),
            "ready_dt": r["ready_dt"].isoformat() if r["ready_dt"] else None,
            "obj_class": r["obj_class"],
            "escrow": r["escrow"],
            "problem_flag": r["problem_flag"],
            "lat": _f(r["latitude"]),
            "lon": _f(r["longitude"]),
            "is_ekb": r["is_ekb"],
        }
        for r in rows
    ]


def prinzip_district_distribution(
    db: Session, developer_id: str = "6208_0"
) -> list[dict[str, Any]]:
    """Spatial-join PRINZIP buildings to ЕКБ districts via lat/lon polygons.

    Без полигонов района: используем bbox-эвристику EKB и группируем по nearest district
    через простой COUNT — но в таблице нет геометрии районов. Возвращаем сводку
    с фолбэком на district_name='не определён', основанную на текстовых известных
    PRINZIP-проектах. Для MVP — заглушка из памяти, чтобы UI не зависел от неполного
    spatial-join. TODO: добавить geometry в ekb_districts.
    """
    # Hard-coded from PRINZIP_Strategy_Apr27 — verified mapping.
    known: list[dict[str, Any]] = [
        {"district_name": "Октябрьский", "prinzip_zk": 6, "share_in_district_pct": 6.7},
        {"district_name": "Верх-Исетский", "prinzip_zk": 4, "share_in_district_pct": 2.6},
        {"district_name": "Ленинский", "prinzip_zk": 4, "share_in_district_pct": 1.9},
        {"district_name": "Кировский", "prinzip_zk": 2, "share_in_district_pct": 1.7},
        {"district_name": "Орджоникидзевский", "prinzip_zk": 1, "share_in_district_pct": 0.7},
        {"district_name": "Академический", "prinzip_zk": 0, "share_in_district_pct": 0.0},
        {"district_name": "Чкаловский", "prinzip_zk": 0, "share_in_district_pct": 0.0},
        {"district_name": "Железнодорожный", "prinzip_zk": 0, "share_in_district_pct": 0.0},
    ]
    return known


def prinzip_insights() -> dict[str, Any]:
    """Static text/recommendations from PRINZIP_Strategy_Apr27 (knowledge graph)."""
    return {
        "headline": (
            "PRINZIP — velocity-лидер Свердл (sold% +33пп за 14 мес), "
            "но портфель смещён в сегмент инвесторских студий-однушек, "
            "тогда как рынок голосует деньгами за семейные 60-90 м² "
            "и премиум 80+."
        ),
        "key_gaps": [
            {
                "label": "Средний метраж",
                "prinzip": 38.1,
                "market": 49.0,
                "brusnika": 60.0,
                "forum": 61.0,
                "unit": "м²",
            },
            {
                "label": "Доля 1-к",
                "prinzip": 75.4,
                "market": 52.0,
                "brusnika": 47.0,
                "forum": 44.3,
                "unit": "%",
            },
            {
                "label": "Доля 3-к+",
                "prinzip": 5.4,
                "market": 13.0,
                "brusnika": 18.1,
                "forum": 21.5,
                "unit": "%",
            },
            {
                "label": "sold% Свердл",
                "prinzip": 48.0,
                "market": 29.0,
                "brusnika": 47.0,
                "forum": 54.0,
                "unit": "%",
            },
        ],
        "priorities": [
            {
                "rank": 1,
                "title": "Семейные 60-90 м² (3-к)",
                "why": (
                    "Дефицит в портфеле (5% vs Брусника 18%, рынок 13%). "
                    "Реальные сделки Q3'25-Q1'26: 3-к 60-80 м² = 8% сделок "
                    "при медиане 126 934 ₽/м². Средний чек ≈ 10.5 М ₽ — "
                    "выше текущих 6.15 М CRM."
                ),
            },
            {
                "rank": 2,
                "title": "Премиум 100-150 м²",
                "why": (
                    "37% реальных ДДУ-сделок Свердл в сегменте 80+ м² "
                    "при медиане 139 382 ₽/м², средний чек 20 М ₽. "
                    "Премиум кад.кварталы: 66:41:0701011 (медиана 424K), "
                    "66:41:0106113 (172K), 66:41:0704044 (149K)."
                ),
            },
        ],
        "where_to_build": [
            {
                "district": "Академический",
                "why": (
                    "330 ЖК / 82К квартир — самый большой кластер ЕКБ, "
                    "PRINZIP отсутствует (0%). Семейный сегмент молодых покупателей."
                ),
            },
            {
                "district": "Верх-Исетский (расширение)",
                "why": (
                    "Кад.квартал 66:41:0106113 — ср.метраж 113 м² × 172K ₽/м², "
                    "ниша бизнес 80-130 м²."
                ),
            },
            {
                "district": "Чкаловский / Железнодорожный",
                "why": (
                    "Растущие районы, 0% PRINZIP, низкая конкуренция. "
                    "Тест 60-80 м² без премиума."
                ),
            },
        ],
        "what_to_avoid": [
            (
                "Однушки 30-40 м² — переразвитый сегмент Свердл "
                "(рынок строит 52% таких, доля сделок падает)."
            ),
            (
                "Проекты со сдачей 2028+ на эскроу — 66-89% unsold, "
                "рынок не рассчитывается на дальний горизонт."
            ),
        ],
        "benchmarks": [
            {
                "name": "Брусника",
                "model": ("350 тыс м² × sold 47% × Δ +11пп. 3-к доля 18%, ср. метраж 60 м²."),
            },
            {
                "name": "Холдинг Форум-групп",
                "model": (
                    "113 тыс м² × sold 54% × Δ +21пп лидер velocity. " "3-к доля 21.5%, ср. 61 м²."
                ),
            },
        ],
    }