gendesign/backend/tests/sql/test_ddu_price_indicator.py

"""Integration test for Issue #99 — ДДУ price indicator SQL logic.

Builds a synthetic temp table mirroring rosreestr_deals and runs the exact
bucketing + index computation from migration 152 / ddu_price_indicator against
a real PostgreSQL. Proves:

1. Per-unit area bucketing uses area/deal_count (packaged ДДУ), not raw area.
2. index_basis is median / first-period-median; index_previous is median /
   previous-PRESENT-period median (min_deals=10 gates noisy quarters out).
3. prev_period_value honestly records which period index_previous compares to
   (may NOT be the literal previous quarter if one was filtered out).

psycopg v3 only. CAST(:x AS type) everywhere. Skips cleanly off-CI.
"""

import os
from decimal import Decimal

import psycopg
import pytest


def _get_dsn() -> str:
    raw = os.environ.get("TEST_DATABASE_URL") or os.environ.get(
        "DATABASE_URL",
        "postgresql://gendesign@localhost:15432/gendesign",
    )
    return raw.replace("+psycopg", "")


def _db_reachable() -> tuple[bool, str]:
    try:
        with psycopg.connect(_get_dsn(), connect_timeout=3):
            return True, ""
    except Exception as e:
        return False, str(e)


_DB_OK, _DB_ERR = _db_reachable()
pytestmark = pytest.mark.skipif(
    not _DB_OK,
    reason=(
        "Нет доступной postgres БД (TEST_DATABASE_URL/DATABASE_URL) — "
        f"тест #99 пропущен: {_DB_ERR}"
    ),
)


# Core indicator query under test — mirrors migration 152 (reads temp table rd).
_INDICATOR_SQL = """
WITH per_unit AS (
    SELECT date_trunc('quarter', period_start_date)::date AS quarter_start,
           (area / deal_count) AS area_per_unit,
           CASE WHEN price_per_sqm IS NOT NULL THEN price_per_sqm
                WHEN deal_price IS NOT NULL AND area > 0 THEN deal_price/area END AS price_m2
      FROM rd
     WHERE region_code = 66 AND realestate_type_code = '002001003000'
       AND doc_type = 'ДДУ' AND deal_count > 0 AND area > 0
       AND period_start_date >= DATE '2025-04-01'
),
filtered AS (
    SELECT quarter_start, area_per_unit, price_m2 FROM per_unit
     WHERE area_per_unit BETWEEN 10 AND 300 AND price_m2 BETWEEN 30000 AND 800000
),
bucketed AS (
    SELECT quarter_start, price_m2,
           CASE WHEN area_per_unit < 25 THEN 1 WHEN area_per_unit < 40 THEN 2
                WHEN area_per_unit < 60 THEN 3 WHEN area_per_unit < 80 THEN 4
                WHEN area_per_unit < 100 THEN 5 ELSE 6 END AS area_bucket
      FROM filtered
    UNION ALL
    SELECT quarter_start, price_m2, 0 FROM filtered
),
agg AS (
    SELECT area_bucket, quarter_start, COUNT(*) AS deals_count,
           PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY price_m2) AS med
      FROM bucketed GROUP BY area_bucket, quarter_start
    HAVING COUNT(*) >= 10
),
indexed AS (
    SELECT area_bucket, quarter_start, deals_count, med,
           FIRST_VALUE(med) OVER w AS basis_med,
           LAG(med) OVER w AS prev_med,
           LAG(quarter_start) OVER w AS prev_q
      FROM agg WINDOW w AS (PARTITION BY area_bucket ORDER BY quarter_start)
)
SELECT area_bucket,
       (EXTRACT(YEAR FROM quarter_start)::int::text || '-Q'
            || EXTRACT(QUARTER FROM quarter_start)::int::text) AS period_value,
       deals_count,
       ROUND((med / NULLIF(basis_med, 0))::numeric, 4) AS index_basis,
       ROUND((med / NULLIF(prev_med, 0))::numeric, 4) AS index_previous,
       CASE WHEN prev_q IS NOT NULL THEN
            (EXTRACT(YEAR FROM prev_q)::int::text || '-Q'
                || EXTRACT(QUARTER FROM prev_q)::int::text) END AS prev_period_value
  FROM indexed
 ORDER BY area_bucket, period_value
"""


@pytest.fixture()
def conn():
    with psycopg.connect(_get_dsn(), autocommit=False) as c:
        yield c
        c.rollback()


def _insert_quarter(cur, q_start, bucket_area, price_m2, n) -> None:
    """Insert n single-flat ДДУ rows at given per-unit area + price/m²."""
    rows = [
        ("002001003000", "ДДУ", 66, q_start, bucket_area, 1, price_m2)
        for _ in range(n)
    ]
    cur.executemany(
        "INSERT INTO rd (realestate_type_code, doc_type, region_code, "
        "period_start_date, area, deal_count, price_per_sqm) "
        "VALUES (%s, %s, %s, %s, %s, %s, %s)",
        rows,
    )


def _setup(cur: psycopg.Cursor) -> None:
    cur.execute(
        """
        CREATE TEMP TABLE rd (
            realestate_type_code text, doc_type varchar, region_code smallint,
            period_start_date date, area numeric, deal_count int, price_per_sqm numeric,
            deal_price numeric
        ) ON COMMIT DROP;
        """
    )
    # Bucket 3 (40-60 m²): three quarters, rising median, all >=10 deals.
    _insert_quarter(cur, "2025-04-01", 50, 100000, 12)  # basis
    _insert_quarter(cur, "2025-07-01", 50, 110000, 12)  # +10% vs basis & prev
    _insert_quarter(cur, "2025-10-01", 50, 121000, 12)  # +10% vs prev
    # Bucket 4 (60-80 m²): 2025-Q3 present, 2025-Q4 SPARSE (<10 → filtered),
    # 2026-Q1 present. index_previous for 2026-Q1 must compare to 2025-Q3.
    _insert_quarter(cur, "2025-07-01", 70, 150000, 11)
    _insert_quarter(cur, "2025-10-01", 70, 999999, 3)   # below min_deals → dropped
    _insert_quarter(cur, "2026-01-01", 70, 165000, 11)
    # Packaged-deal trap: one row area=350 deal_count=7 → per-unit 50 m² (bucket 3),
    # NOT bucket 6. Price chosen mid-range so it doesn't move the median much.
    cur.execute(
        "INSERT INTO rd (realestate_type_code, doc_type, region_code, "
        "period_start_date, area, deal_count, price_per_sqm) "
        "VALUES ('002001003000','ДДУ',66,'2025-04-01',350,7,100000)"
    )


def _rows_by_key(rows) -> dict[tuple[int, str], tuple]:
    return {(r[0], r[1]): r for r in rows}


def test_basis_and_previous_index(conn: psycopg.Connection) -> None:
    cur = conn.cursor()
    _setup(cur)
    cur.execute(_INDICATOR_SQL)
    by = _rows_by_key(cur.fetchall())
    # Bucket 3 rising 100k→110k→121k.
    assert by[(3, "2025-Q2")][3] == Decimal("1.0000")  # basis self
    assert by[(3, "2025-Q3")][4] == Decimal("1.1000")  # prev: 110/100
    assert by[(3, "2025-Q4")][4] == Decimal("1.1000")  # prev: 121/110
    assert by[(3, "2025-Q4")][3] == Decimal("1.2100")  # basis: 121/100


def test_packaged_deal_bucketed_by_per_unit_area(conn: psycopg.Connection) -> None:
    """area=350/deal_count=7 → 50 m² → bucket 3, never bucket 6."""
    cur = conn.cursor()
    _setup(cur)
    cur.execute(_INDICATOR_SQL)
    by = _rows_by_key(cur.fetchall())
    # Bucket 6 should have NO rows (the only 100+ candidate was the packaged
    # row, which correctly lands in bucket 3).
    assert not any(k[0] == 6 for k in by), "packaged ДДУ leaked into 100+ bucket"
    # Bucket 3 basis quarter deal count includes the 7-unit packaged row.
    assert by[(3, "2025-Q2")][2] >= 12


def test_prev_period_value_skips_filtered_quarter(conn: psycopg.Connection) -> None:
    """Bucket 4: 2025-Q4 is below min_deals and dropped; 2026-Q1's
    index_previous must compare against 2025-Q3 (honest prev_period_value),
    not the literal previous quarter.
    """
    cur = conn.cursor()
    _setup(cur)
    cur.execute(_INDICATOR_SQL)
    by = _rows_by_key(cur.fetchall())
    assert (4, "2025-Q4") not in by, "sparse quarter should be filtered by min_deals"
    q1 = by[(4, "2026-Q1")]
    assert q1[5] == "2025-Q3", f"prev_period_value must skip dropped Q4, got {q1[5]}"
    # 165000 / 150000 = 1.1000
    assert q1[4] == Decimal("1.1000")