bot-backend
14f3ef2019
Co-authored-by: bot-backend <bot-backend@gendsgn.local> Co-committed-by: bot-backend <bot-backend@gendsgn.local>
292 lines
11 KiB
Python
292 lines
11 KiB
Python
"""Velocity-anomaly detection per ЖК (Issue #17).
|
||
|
||
Detects жилые комплексы whose recent sales velocity has dropped значительно
|
||
versus their own trailing history. Deterministic (pure statistics — z-score +
|
||
percent-change over the per-object monthly series). No ML, no external calls.
|
||
|
||
Data source
|
||
-----------
|
||
`domrf_kn_sale_graph` — monthly per-object sale series (type='apartments').
|
||
Columns used: obj_id, report_month, realised (monthly units sold — FLOW, not
|
||
cumulative — verified on prod 2026-06: values rise and fall month-to-month),
|
||
snapshot_date.
|
||
|
||
Object metadata (comm_name / dev_name / region_cd) is joined from
|
||
`domrf_kn_objects` (latest snapshot per obj_id).
|
||
|
||
Real-data gotchas (verified on prod)
|
||
-------------------------------------
|
||
1. **Snapshot coverage is uneven.** `MAX(snapshot_date)` is NOT safe: the two
|
||
newest snapshots on prod (2026-06-03, 2026-04-27) are partial scrapes with
|
||
only 2–7 objects. The richest snapshots carry ~440 objects. We therefore
|
||
pick the snapshot with the MOST distinct objects (ties broken by recency)
|
||
and run the whole analysis within that single consistent snapshot.
|
||
2. **All sale_graph objects are region 66** today, but we still filter by
|
||
region via domrf_kn_objects.region_cd so the endpoint stays correct if
|
||
other regions are scraped later.
|
||
3. **History is ~9 months** in the richest snapshot, so `lookback_months`
|
||
defaults to 9 and we require ≥6 months (issue acceptance) before scoring.
|
||
|
||
Algorithm (per object, within the chosen snapshot)
|
||
--------------------------------------------------
|
||
- Order months descending; take the most recent 3 as the "recent" window and
|
||
everything older (within lookback) as the "prior" window.
|
||
- recent_velocity = mean(realised) over recent 3 months.
|
||
- prior_velocity = mean(realised) over prior months.
|
||
- prior_std = sample stddev of prior months.
|
||
- z = (recent_velocity - prior_velocity) / prior_std (negative ⇒ slowdown)
|
||
- drop_pct = (recent_velocity - prior_velocity) / prior_velocity * 100
|
||
- An object is an alert when z <= -min_zscore AND drop_pct <= -min_drop_pct
|
||
(both gates: statistical significance AND material magnitude — keeps the
|
||
false-positive rate down on noisy low-volume objects, per acceptance ≤10%).
|
||
|
||
Severity
|
||
--------
|
||
- 'high' : z <= -3.0 OR drop_pct <= -60
|
||
- 'medium' : otherwise (still past the alert gates)
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from decimal import Decimal
|
||
from typing import Any
|
||
|
||
from sqlalchemy import text
|
||
from sqlalchemy.orm import Session
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Minimum months of history before an object is eligible (issue: ignore <6 mo).
|
||
_MIN_MONTHS_HISTORY = 6
|
||
|
||
# Recent window size (rolling 3-mo mean), per issue spec.
|
||
_RECENT_WINDOW = 3
|
||
|
||
# Minimum prior observations needed for a meaningful stddev / z-score.
|
||
_MIN_PRIOR_OBS = 3
|
||
|
||
# Default magnitude gate alongside the z-score gate. A pure z-score alert on a
|
||
# low-volume object (prior mean ~2/mo, stddev ~1) fires on tiny absolute moves;
|
||
# requiring a real percent drop suppresses those false positives.
|
||
_DEFAULT_MIN_DROP_PCT = 30.0
|
||
|
||
# Severity thresholds.
|
||
_HIGH_ZSCORE = 3.0
|
||
_HIGH_DROP_PCT = 60.0
|
||
|
||
|
||
def _f(value: Any) -> float | None:
|
||
if value is None:
|
||
return None
|
||
if isinstance(value, Decimal):
|
||
return float(value)
|
||
return float(value)
|
||
|
||
|
||
def _best_snapshot(db: Session, *, sale_type: str = "apartments") -> Any | None:
|
||
"""Pick the sale_graph snapshot with the most distinct objects.
|
||
|
||
NOT ``MAX(snapshot_date)`` — the freshest snapshots on prod are frequently
|
||
partial scrapes (2–7 objects). The richest snapshot gives the broadest,
|
||
most consistent cohort. Ties broken by recency.
|
||
"""
|
||
return db.execute(
|
||
text(
|
||
"""
|
||
SELECT snapshot_date
|
||
FROM domrf_kn_sale_graph
|
||
WHERE type = :sale_type
|
||
GROUP BY snapshot_date
|
||
ORDER BY COUNT(DISTINCT obj_id) DESC, snapshot_date DESC
|
||
LIMIT 1
|
||
"""
|
||
),
|
||
{"sale_type": sale_type},
|
||
).scalar()
|
||
|
||
|
||
def detect_velocity_anomalies(
|
||
db: Session,
|
||
region_code: int = 66,
|
||
lookback_months: int = 9,
|
||
min_zscore: float = 2.0,
|
||
min_drop_pct: float = _DEFAULT_MIN_DROP_PCT,
|
||
limit: int = 100,
|
||
) -> dict[str, Any]:
|
||
"""Return ЖК whose recent velocity dropped significantly vs their history.
|
||
|
||
Args:
|
||
db: SQLAlchemy session (sync).
|
||
region_code: filter via domrf_kn_objects.region_cd (66 = Свердловская обл.).
|
||
lookback_months: months of history to consider (recent + prior windows).
|
||
min_zscore: absolute z-score threshold; alert when z <= -min_zscore.
|
||
min_drop_pct: absolute percent-drop threshold; alert when drop_pct <= -value.
|
||
limit: max alerts returned (ordered by z-score, most severe first).
|
||
|
||
Returns:
|
||
``{"snapshot_date": str|None, "region_code": int, "params": {...},
|
||
"alerts": [...]}``. ``alerts`` is empty when there is no usable
|
||
snapshot or no object clears both gates.
|
||
"""
|
||
snapshot = _best_snapshot(db)
|
||
if snapshot is None:
|
||
logger.info("velocity_alerts: no sale_graph snapshot available")
|
||
return {
|
||
"snapshot_date": None,
|
||
"region_code": region_code,
|
||
"params": {
|
||
"lookback_months": lookback_months,
|
||
"min_zscore": min_zscore,
|
||
"min_drop_pct": min_drop_pct,
|
||
},
|
||
"alerts": [],
|
||
}
|
||
|
||
# One pass: window the per-object monthly series, compute recent/prior
|
||
# aggregates, score, then join names + filter by region. All bind params use
|
||
# CAST(:x AS type) per repo convention (psycopg v3 ignores :x::type).
|
||
rows = (
|
||
db.execute(
|
||
text(
|
||
"""
|
||
WITH anchor AS (
|
||
-- Anchor the lookback to the LATEST report_month present in
|
||
-- this snapshot, NOT snapshot_date. On prod the scrape date
|
||
-- (snapshot_date) can lag the newest data month by months,
|
||
-- so anchoring to snapshot_date would silently truncate the
|
||
-- prior window and starve the z-score (prior_n < min).
|
||
SELECT MAX(report_month) AS max_month
|
||
FROM domrf_kn_sale_graph
|
||
WHERE type = 'apartments'
|
||
AND snapshot_date = :snap
|
||
),
|
||
series AS (
|
||
SELECT g.obj_id, g.report_month, g.realised
|
||
FROM domrf_kn_sale_graph g
|
||
CROSS JOIN anchor a
|
||
WHERE g.type = 'apartments'
|
||
AND g.snapshot_date = :snap
|
||
AND g.report_month > (
|
||
a.max_month - CAST(:lookback_interval 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 <= :recent_window) AS recent_mean,
|
||
AVG(realised) FILTER (WHERE rn > :recent_window) AS prior_mean,
|
||
STDDEV_SAMP(realised) FILTER (WHERE rn > :recent_window) AS prior_std,
|
||
COUNT(realised) FILTER (WHERE rn > :recent_window) AS prior_n
|
||
FROM ranked
|
||
GROUP BY obj_id, n_months
|
||
),
|
||
scored AS (
|
||
SELECT
|
||
w.obj_id,
|
||
w.recent_mean,
|
||
w.prior_mean,
|
||
w.prior_std,
|
||
CASE WHEN w.prior_std > 0
|
||
THEN (w.recent_mean - w.prior_mean) / w.prior_std
|
||
END AS zscore,
|
||
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 >= :min_months
|
||
AND w.prior_n >= :min_prior_obs
|
||
AND w.prior_mean > 0
|
||
AND w.prior_std > 0
|
||
)
|
||
SELECT
|
||
s.obj_id,
|
||
o.comm_name,
|
||
o.dev_name,
|
||
s.recent_mean,
|
||
s.prior_mean,
|
||
s.zscore,
|
||
s.drop_pct
|
||
FROM scored s
|
||
JOIN LATERAL (
|
||
SELECT comm_name, dev_name, region_cd
|
||
FROM domrf_kn_objects
|
||
WHERE obj_id = s.obj_id
|
||
ORDER BY snapshot_date DESC
|
||
LIMIT 1
|
||
) o ON TRUE
|
||
WHERE o.region_cd = :region_code
|
||
AND s.zscore <= :neg_zscore
|
||
AND s.drop_pct <= :neg_drop_pct
|
||
ORDER BY s.zscore ASC
|
||
LIMIT :limit
|
||
"""
|
||
),
|
||
{
|
||
"snap": snapshot,
|
||
"lookback_interval": f"{lookback_months} months",
|
||
"recent_window": _RECENT_WINDOW,
|
||
"min_months": _MIN_MONTHS_HISTORY,
|
||
"min_prior_obs": _MIN_PRIOR_OBS,
|
||
"region_code": region_code,
|
||
"neg_zscore": -abs(min_zscore),
|
||
"neg_drop_pct": -abs(min_drop_pct),
|
||
"limit": limit,
|
||
},
|
||
)
|
||
.mappings()
|
||
.all()
|
||
)
|
||
|
||
alerts: list[dict[str, Any]] = []
|
||
for r in rows:
|
||
z = _f(r["zscore"])
|
||
drop = _f(r["drop_pct"])
|
||
severity = (
|
||
"high"
|
||
if (z is not None and z <= -_HIGH_ZSCORE)
|
||
or (drop is not None and drop <= -_HIGH_DROP_PCT)
|
||
else "medium"
|
||
)
|
||
alerts.append(
|
||
{
|
||
"obj_id": int(r["obj_id"]),
|
||
"comm_name": r["comm_name"],
|
||
"dev_name": r["dev_name"],
|
||
"recent_velocity_pm": round(_f(r["recent_mean"]) or 0.0, 2),
|
||
"prior_velocity_pm": round(_f(r["prior_mean"]) or 0.0, 2),
|
||
"zscore": round(z, 2) if z is not None else None,
|
||
"drop_pct": round(drop, 1) if drop is not None else None,
|
||
"severity": severity,
|
||
}
|
||
)
|
||
|
||
logger.info(
|
||
"velocity_alerts: snapshot=%s region=%d lookback=%dmo z>=%.1f drop>=%.0f%% -> %d alerts",
|
||
snapshot,
|
||
region_code,
|
||
lookback_months,
|
||
min_zscore,
|
||
min_drop_pct,
|
||
len(alerts),
|
||
)
|
||
|
||
return {
|
||
"snapshot_date": snapshot.isoformat() if snapshot else None,
|
||
"region_code": region_code,
|
||
"params": {
|
||
"lookback_months": lookback_months,
|
||
"min_zscore": min_zscore,
|
||
"min_drop_pct": min_drop_pct,
|
||
},
|
||
"alerts": alerts,
|
||
}
|