gendesign/backend/app/services/analytics/ddu_price_indicator.py

"""ДДУ price indicator — ARN-framework applied to primary market (Issue #99).

Reads ``mv_ddu_price_indicator`` (migration 152) and serves an ARN-mirror
response: per-quarter price indicator of the PRIMARY market (новостройки/ДДУ)
for Свердловская область (region 66), broken down by area bucket.

This is a deterministic statistical indicator (median price/m² ratios), not a
forecast. See the migration header for the full MVP-scope rationale; the short
version of the honest simplifications versus the full ARN matrix:

- Granularity = subject (region 66), NOT cad_quarter — primary-market ДДУ rows
  are far too sparse per cad_quarter after the 2025 rosreestr aggregation change.
- period_type = 'Q' only (source publishes quarterly; no monthly granularity).
- Window starts 2025-Q2 (hard data-regime break at 2025-Q1).
- calc_method 1 (basis) and 2 (previous) only. Method 3 (year-ago) is BLOCKED:
  there is no comparable pre-2025-Q2 primary-market data.

The endpoint accepts the same request body shape as the ARN API (indicators,
calculationMethod, periodFrom/To, areaRanges, federalSubject) so the frontend
can reuse one UI pattern. Unsupported ARN options are reported in a ``notes``
list rather than silently ignored.
"""

from __future__ import annotations

import logging
from decimal import Decimal
from typing import Any

from sqlalchemy import text
from sqlalchemy.exc import OperationalError
from sqlalchemy.orm import Session

logger = logging.getLogger(__name__)

# ARN area-range labels (poquartirnaya area, m²). Index = area_bucket id.
_BUCKET_LABELS: dict[int, str] = {
    0: "Все площади",
    1: "0–25 м²",
    2: "25–40 м²",
    3: "40–60 м²",
    4: "60–80 м²",
    5: "80–100 м²",
    6: "100+ м²",
}

# Calc methods supported by the MVP. 3 (year-ago) is intentionally absent.
_CALC_BASIS = 1
_CALC_PREVIOUS = 2
_SUPPORTED_METHODS = (_CALC_BASIS, _CALC_PREVIOUS)

# Only Свердловская область (66) and the Уральский federal district are backed
# by data today; other ARN federalSubject / federalDistrict values are reported
# as unsupported.
_SUPPORTED_SUBJECT = "66"


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


def _normalize_subjects(federal_subject: list[str] | None) -> tuple[bool, list[str]]:
    """Return (subject_supported, unsupported_subjects)."""
    if not federal_subject:
        return True, []
    unsupported = [s for s in federal_subject if str(s) != _SUPPORTED_SUBJECT]
    supported = any(str(s) == _SUPPORTED_SUBJECT for s in federal_subject)
    return supported, unsupported


def get_ddu_indicator(
    db: Session,
    *,
    calculation_method: int = _CALC_PREVIOUS,
    period_from: str | None = None,
    period_to: str | None = None,
    area_ranges: list[int] | None = None,
    federal_subject: list[str] | None = None,
    indicators: list[str] | None = None,
) -> dict[str, Any]:
    """ARN-mirror response for the primary-market ДДУ price indicator.

    Args:
        db: SQLAlchemy session (sync).
        calculation_method: 1 = basis (vs first period), 2 = previous quarter.
        period_from / period_to: inclusive ARN-style bounds, e.g. '2025-Q2'..'2026-Q1'.
            None = no bound on that side.
        area_ranges: ARN area-bucket ids to include (1..6); 0 = all-area headline.
            None / empty → all buckets present in the MV.
        federal_subject: ARN subject codes. Only '66' is backed by data.
        indicators: ARN indicator types (M/Q/H/Y). Only 'Q' is supported.

    Returns:
        ``{"meta": {...}, "table": [...], "graph": [...], "notes": [...]}``.
        ``table`` rows carry both the chosen-method ``index`` and the raw median;
        ``graph`` is the headline (bucket 0) series for charting.
    """
    notes: list[str] = []

    method = calculation_method if calculation_method in _SUPPORTED_METHODS else _CALC_PREVIOUS
    if calculation_method not in _SUPPORTED_METHODS:
        notes.append(
            f"calculationMethod={calculation_method} не поддержан "
            f"(MVP: 1=базисный, 2=предыдущий; 3=годом-ранее недоступен — нет "
            f"сопоставимых данных первички до 2025-Q2). Использован method=2 (предыдущий)."
        )

    if indicators:
        unsupported_ind = [i for i in indicators if str(i).upper() != "Q"]
        if unsupported_ind:
            notes.append(
                f"indicators={unsupported_ind} не поддержаны (источник rosreestr "
                f"публикует данные поквартально; доступен только 'Q')."
            )

    subject_ok, unsupported_subjects = _normalize_subjects(federal_subject)
    if unsupported_subjects:
        notes.append(
            f"federalSubject={unsupported_subjects} вне покрытия "
            f"(данные первички только по 66 — Свердловская обл.)."
        )
    if not subject_ok:
        # Asked exclusively for unsupported subjects → empty result, explained.
        return {
            "meta": {
                "market": "primary_ddu",
                "region_code": int(_SUPPORTED_SUBJECT),
                "calculation_method": method,
                "period_type": "Q",
            },
            "table": [],
            "graph": [],
            "notes": notes,
        }

    index_col = "index_basis" if method == _CALC_BASIS else "index_previous"

    # area_ranges filter (optional). Validate ids to a safe int list — never
    # interpolate into SQL; bound via ANY(CAST(:buckets AS int[])).
    bucket_filter = ""
    params: dict[str, Any] = {}
    if area_ranges:
        clean_buckets = sorted({int(b) for b in area_ranges if 0 <= int(b) <= 6})
        if clean_buckets:
            bucket_filter = "AND area_bucket = ANY(CAST(:buckets AS int[]))"
            params["buckets"] = clean_buckets

    period_filter = ""
    if period_from:
        period_filter += " AND period_value >= :period_from"
        params["period_from"] = period_from
    if period_to:
        period_filter += " AND period_value <= :period_to"
        params["period_to"] = period_to

    try:
        rows = (
            db.execute(
                text(
                    f"""
                    SELECT
                        area_bucket,
                        period_value,
                        period_type,
                        deals_count,
                        median_price_per_m2,
                        index_basis,
                        index_previous,
                        prev_period_value,
                        {index_col} AS chosen_index
                      FROM mv_ddu_price_indicator
                     WHERE 1 = 1
                       {bucket_filter}
                       {period_filter}
                     ORDER BY area_bucket, period_value
                    """
                ),
                params,
            )
            .mappings()
            .all()
        )
    except OperationalError:
        # Most likely the MV does not exist yet (migration 152 not applied).
        logger.exception("ddu_indicator: query failed (mv_ddu_price_indicator missing?)")
        raise

    table: list[dict[str, Any]] = []
    graph: list[dict[str, Any]] = []
    for r in rows:
        bucket = int(r["area_bucket"])
        item = {
            "area_bucket": bucket,
            "area_label": _BUCKET_LABELS.get(bucket, str(bucket)),
            "period_value": r["period_value"],
            "period_type": r["period_type"],
            "deals_count": int(r["deals_count"]),
            "median_price_per_m2": _f(r["median_price_per_m2"]),
            "index": _f(r["chosen_index"]),
            "index_basis": _f(r["index_basis"]),
            "index_previous": _f(r["index_previous"]),
            "prev_period_value": r["prev_period_value"],
        }
        table.append(item)
        if bucket == 0:
            graph.append(
                {
                    "period_value": r["period_value"],
                    "index": _f(r["chosen_index"]),
                    "median_price_per_m2": _f(r["median_price_per_m2"]),
                    "deals_count": int(r["deals_count"]),
                }
            )

    logger.info(
        "ddu_indicator: method=%d buckets=%s period=%s..%s -> %d rows",
        method,
        params.get("buckets", "all"),
        period_from or "*",
        period_to or "*",
        len(table),
    )

    return {
        "meta": {
            "market": "primary_ddu",
            "region_code": int(_SUPPORTED_SUBJECT),
            "calculation_method": method,
            "period_type": "Q",
            "source": "rosreestr_deals (ДДУ, realestate_type_code=002001003000)",
            "arn_comparison": {
                "note": (
                    "ARN/НСПД считает вторичку; здесь первичка (новостройки). "
                    "Для новостроек рост индекса ≠ инфляция — это сдвиг ассортимента + цен."
                ),
                "ekb_q1_2026_arn_secondary": 1.03,
            },
        },
        "table": table,
        "graph": graph,
        "notes": notes,
    }


def refresh_ddu_price_indicator(db: Session, *, concurrently: bool = True) -> int:
    """REFRESH MATERIALIZED VIEW mv_ddu_price_indicator.

    Mirrors refresh_layout_velocity(): CONCURRENTLY by default (needs the unique
    index from migration 152), with a one-shot fallback to non-concurrent if the
    MV is not yet populated. Not wired into Celery beat in this PR — manual /
    admin invocation, like layout_velocity_refresh.

    Returns the row count after refresh (observability).
    """
    try:
        if concurrently:
            db.execute(text("REFRESH MATERIALIZED VIEW CONCURRENTLY mv_ddu_price_indicator"))
        else:
            db.execute(text("REFRESH MATERIALIZED VIEW mv_ddu_price_indicator"))
        db.commit()
    except OperationalError as e:
        if concurrently and "cannot refresh materialized view" in str(e).lower():
            logger.warning(
                "ddu_indicator CONCURRENTLY failed (MV not populated), falling back"
            )
            db.rollback()
            db.execute(text("REFRESH MATERIALIZED VIEW mv_ddu_price_indicator"))
            db.commit()
        else:
            raise
    row = db.execute(text("SELECT COUNT(*) FROM mv_ddu_price_indicator")).first()
    count = int(row[0]) if row else 0
    logger.info("mv_ddu_price_indicator refreshed: %d rows", count)
    return count