gendesign/backend/tests/sql/test_velocity_alerts.py

"""Integration test for Issue #17 — velocity-anomaly detection SQL logic.

Builds synthetic temp tables mirroring domrf_kn_sale_graph + domrf_kn_objects
and runs the exact windowing / z-score query from
``app.services.analytics.velocity_alerts`` against a real PostgreSQL. Proves:

1. A ЖК with a sharp recent drop is flagged (z <= -2.0 AND drop_pct <= -30%).
2. A ЖК with stable velocity is NOT flagged (no false positive).
3. The lookback window anchors to the latest report_month (not snapshot_date),
   so a stale scrape date does not starve the prior window.

Uses psycopg v3 (never psycopg2). All bind params use CAST(:x AS type).
Skips cleanly off-CI when no Postgres is reachable.
"""

import os

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"тест #17 пропущен: {_DB_ERR}"
    ),
)


# Core query under test — kept byte-identical in spirit to velocity_alerts.py.
# Reads from temp tables sg (sale_graph) + obj (objects).
_DETECT_SQL = """
WITH anchor AS (
    SELECT MAX(report_month) AS max_month
      FROM sg WHERE type = 'apartments' AND snapshot_date = %(snap)s
),
series AS (
    SELECT g.obj_id, g.report_month, g.realised
      FROM sg g CROSS JOIN anchor a
     WHERE g.type = 'apartments' AND g.snapshot_date = %(snap)s
       AND g.report_month > (a.max_month - CAST(%(lookback)s AS interval))
),
ranked AS (
    SELECT obj_id, realised,
           ROW_NUMBER() OVER (PARTITION BY obj_id ORDER BY report_month DESC) AS rn,
           COUNT(*) OVER (PARTITION BY obj_id) AS n_months
      FROM series
),
windows AS (
    SELECT obj_id, n_months,
           AVG(realised) FILTER (WHERE rn <= 3) AS recent_mean,
           AVG(realised) FILTER (WHERE rn > 3)  AS prior_mean,
           STDDEV_SAMP(realised) FILTER (WHERE rn > 3) AS prior_std,
           COUNT(*) FILTER (WHERE rn > 3) AS prior_n
      FROM ranked GROUP BY obj_id, n_months
),
scored AS (
    SELECT w.obj_id, w.recent_mean, w.prior_mean,
           CASE WHEN w.prior_std > 0 THEN (w.recent_mean - w.prior_mean)/w.prior_std END AS z,
           CASE WHEN w.prior_mean > 0
                THEN (w.recent_mean - w.prior_mean)/w.prior_mean*100.0 END AS drop_pct
      FROM windows w
     WHERE w.n_months >= 6 AND w.prior_n >= 3 AND w.prior_mean > 0 AND w.prior_std > 0
)
SELECT s.obj_id, ROUND(s.z::numeric, 2) AS z, ROUND(s.drop_pct::numeric, 1) AS drop_pct
  FROM scored s
  JOIN LATERAL (SELECT region_cd FROM obj WHERE obj_id = s.obj_id LIMIT 1) o ON TRUE
 WHERE o.region_cd = 66 AND s.z <= -2.0 AND s.drop_pct <= -30.0
 ORDER BY s.z ASC
"""


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


def _setup(cur: psycopg.Cursor) -> None:
    cur.execute(
        """
        CREATE TEMP TABLE sg (
            obj_id bigint, report_month date, type text,
            realised int, snapshot_date date
        ) ON COMMIT DROP;
        CREATE TEMP TABLE obj (obj_id bigint, region_cd int) ON COMMIT DROP;
        """
    )
    snap = "2026-04-28"  # stale scrape date; data months end 2025-12 (4-mo gap)
    months = [
        "2025-04-01", "2025-05-01", "2025-06-01", "2025-07-01", "2025-08-01",
        "2025-09-01", "2025-10-01", "2025-11-01", "2025-12-01",
    ]
    # obj 1 — sharp drop: prior ~15/mo, recent ~4/mo  -> alert
    dropper = [16, 14, 15, 17, 13, 14, 5, 4, 3]
    # obj 2 — stable ~10/mo  -> NO alert
    stable = [10, 9, 11, 10, 12, 9, 10, 11, 10]
    rows = []
    for m, v in zip(months, dropper, strict=True):
        rows.append((1, m, "apartments", v, snap))
    for m, v in zip(months, stable, strict=True):
        rows.append((2, m, "apartments", v, snap))
    cur.executemany(
        "INSERT INTO sg (obj_id, report_month, type, realised, snapshot_date) "
        "VALUES (%s, %s, %s, %s, %s)",
        rows,
    )
    cur.executemany("INSERT INTO obj (obj_id, region_cd) VALUES (%s, %s)", [(1, 66), (2, 66)])


def test_sharp_drop_is_flagged(conn: psycopg.Connection) -> None:
    cur = conn.cursor()
    _setup(cur)
    cur.execute(_DETECT_SQL, {"snap": "2026-04-28", "lookback": "9 months"})
    alerts = cur.fetchall()
    flagged_ids = {r[0] for r in alerts}
    # obj 1 (dropper) must be flagged; obj 2 (stable) must not.
    assert 1 in flagged_ids, f"expected dropper flagged, got {alerts}"
    assert 2 not in flagged_ids, f"stable obj must not be a false positive, got {alerts}"
    # And the drop must be a strong negative z-score with a material drop%.
    dropper_row = next(r for r in alerts if r[0] == 1)
    assert dropper_row[1] <= -2.0  # z
    assert dropper_row[2] <= -30.0  # drop_pct


def test_lookback_anchors_to_latest_data_month(conn: psycopg.Connection) -> None:
    """If the window anchored to snapshot_date (2026-04-28) instead of the
    latest report_month (2025-12), the 4-month gap would trim the prior window
    below prior_n>=3 and the dropper would NOT be flagged. This guards the fix.
    """
    cur = conn.cursor()
    _setup(cur)
    cur.execute(_DETECT_SQL, {"snap": "2026-04-28", "lookback": "9 months"})
    flagged_ids = {r[0] for r in cur.fetchall()}
    assert 1 in flagged_ids, "anchor-to-latest-month regression: dropper lost"