lekss361/gendesign
1
0
Fork 0
Code Issues 57 Pull requests Projects Releases Packages Wiki Activity Actions

feat(forecasting): monthly sales series builder for §9.6 (#951c)

Browse source 
Add backend/app/services/forecasting/sales_series.py — deterministic monthly
sold-units (+area+avg price) time series per SegmentSpec, the Y-axis foundation
for the §9.6 key-rate regression. Two sources with distinct authority:
objective_corpus_room_month (class/district aggregates, full ДДУ+ДКП, count-
weighted price ×1000 тыс→₽) and objective_lots (room×area×price segmentation,
GROUP BY date_trunc registration_date; survivorship caveat documented).

Pure DB-free helpers price_bucket_of / room_area_bucket_of / log_diff /
fill_month_grid, unit-tested. In-SQL CASE bucketing shares the same threshold
constants as the helpers (bound params) → no Python↔SQL drift. Months with no
sales → units=0 (real zero), area/price → None. Graceful empty/thin → low
confidence, zero-filled grid. Mirrors macro_series.py (PR2). room_bucket
semantics are source-specific (documented on SegmentSpec). 62 tests, ruff clean.
This commit is contained in:
Light1YT 2026-06-03 10:50:28 +05:00
parent ace3b99508
commit e3bb125910
3 changed files with 1276 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

25
backend/app/services/forecasting/__init__.py
View file

@ -6,13 +6,14 @@
классификатор режима ставки, лаговые помощники. Всё ДЕТЕРМИНИРОВАННО, БЕЗ LLM.
Слои (по PR):
• macro_series (этот PR, #951b) — monthly макро-ряд + классификатор режима ставки.
• sales_series (later PR) — monthly ряд продаж по локации.
• macro_series (#951b) — monthly макро-ряд + классификатор режима ставки (X-ось §9.6).
• sales_series (#951c) — monthly ряд продаж по сегменту (Y-ось §9.6).
• rate_sensitivity (later PR) — §9.6 чувствительность продаж к key_rate.
• macro_coefficient (later PR) — §9.5 макро-коэффициент.
Источник макро-данных — таблица macro_indicator через тонкий reader
backend/app/services/site_finder/macro.py (переиспользуем, не дублируем).
Источники данных:
• макро — таблица macro_indicator через reader site_finder/macro.py (reuse).
• продажи — objective_corpus_room_month / objective_lots (см. sales_series).
"""
from __future__ import annotations
@ -24,11 +25,27 @@ from app.services.forecasting.macro_series import (
is_confounded_window,
macro_at_lag,
)
from app.services.forecasting.sales_series import (
SalesSeries,
SegmentSpec,
build_sales_series,
fill_month_grid,
log_diff,
price_bucket_of,
room_area_bucket_of,
)
__all__ = [
"MonthlyMacro",
"SalesSeries",
"SegmentSpec",
"build_sales_series",
"classify_regime",
"fill_month_grid",
"get_monthly_macro",
"is_confounded_window",
"log_diff",
"macro_at_lag",
"price_bucket_of",
"room_area_bucket_of",
]

610
backend/app/services/forecasting/sales_series.py Normal file
View file

@ -0,0 +1,610 @@
"""Monthly ряд продаж (sold units + area + avg price) по сегменту — Y-ось §9.6.
#951 (Site Finder v2 / «GG-форсайт» ТЗ §9.6), EPIC 7 «Чувствительность к ключевой
ставке», sub-PR 1 (#951c). Это **data-independent фундамент** Y-оси (зависимая
переменная) для регрессии §9.6 «продажи ↔ ключевая ставка» (later PR3). Парный к
PR2 (macro_series.py — X-ось/regressor). Всё ДЕТЕРМИНИРОВАННО, БЕЗ LLM.
Что строим: для SegmentSpec (любой subset class/room_bucket/district/price_bucket)
— непрерывный MONTHLY ряд (units, area_m2, avg_price_per_m2) из ОДНОГО из двух
источников Объектива, в frozen-dataclass SalesSeries.
Два источника (см. data/sql/68_schema_objective.sql), у каждого своя зона
авторитетности:
• Source A `objective_corpus_room_month` (monthly aggregate, long-формат
месяц×корпус×room_bucket). Колонки сделок: deals_total_count /
deals_total_vol_m2 / deals_total_avg_price_thousand_rub_per_m2. class —
Title-case ("Комфорт"). АВТОРИТЕТЕН для агрегатов класс/район (полные
помесячные сделки ДДУ+ДКП, не зависит от текущего листинга).
• Source B `objective_lots` (per-lot, даты сделок). COUNT(*) + SUM(area_pd) +
взвешенная цена, GROUP BY date_trunc('month', registration_date). class —
lowercase ("комфорт") → нормализуем регистр при матчинге. АВТОРИТЕТЕН для
тонкой сегментации room×area×price (Source A не несёт price-bucket / точную
площадь лота). КАВЕАТ ВЫЖИВАЕМОСТИ: objective_lots — это последний UPSERT-
снапшот per-lot; видны лишь лоты, до сих пор присутствующие в выгрузке. Лот,
проданный и затем выпавший из листинга, в исторических месяцах недоучтён →
Source B занижает старые месяцы (survivorship bias). Для агрегатов класс/
район поэтому предпочтителен Source A; Source B — когда нужна price/area-
сегментация, которую A не даёт.
Соглашение по дате: каждый месяц ряда — ПЕРВОЕ число (YYYY-MM-01). Source B
нормализует registration_date через date_trunc('month', …); Source A's
report_month уже 1-е число. Переиспользуем _month_start/_shift_months/_month_grid
из macro_series (PR2) — не дублируем сеточную логику.
КРИТИЧНО (units=0 vs None): месяц без сделок → units=0 (НАСТОЯЩИЙ ноль: «честно
продали 0» — это данные, а не пропуск; downstream-регрессии нужен непрерывный
ряд счётчиков). Площадь/цена в таком месяце → None (среднее по пустому множеству
не определено — НЕ 0). Это зеркалит дух market_metrics: «нет данных ≠ 0».
Graceful-on-thin-data: пустая таблица / сбой БД → ряд по сетке месяцев с
units=0, area/price=None, confidence='low' (НЕ crash).
psycopg v3 / SQLAlchemy text: bind-параметры ВСЕГДА через CAST(:x AS type) —
никогда :x::type (парсер psycopg3 даёт SyntaxError на :name::type).
"""
from __future__ import annotations
import logging
import math
from dataclasses import dataclass
from datetime import date
from typing import Any, Literal
from sqlalchemy import text
from sqlalchemy.orm import Session
# Переиспользуем сеточные helper'ы PR2 (не дублируем) — они module-level в
# macro_series, хоть и не в __all__ пакета.
from app.services.forecasting.macro_series import _month_grid, _month_start, _shift_months
logger = logging.getLogger(__name__)
SalesSource = Literal["corpus_room_month", "objective_lots"]
Confidence = Literal["high", "medium", "low"]
# ── Named-константы ───────────────────────────────────────────────────────────
# Глубина ряда по умолчанию (месяцев назад). 48 мес ≈ 4 года — зеркалит
# _DEFAULT_MONTHS_BACK макро-ряда (PR2): регрессия §9.6 join-ит sales↔macro
# месяц-в-месяц, окна должны совпадать по длине.
_DEFAULT_MONTHS_BACK: int = 48
# Пороги confidence по числу НЕнулевых месяцев (месяцев с реальными сделками).
# Регрессия §9.6 (sales ~ key_rate + лаги) требует достаточной вариации Y:
# • ≥24 ненулевых мес → 'high' (≥2 года сделок — устойчивый фит лагов)
# • ≥12 ненулевых мес → 'medium' (≥1 год — фит возможен, но шумнее)
# • иначе → 'low' (тонкий сегмент — регрессии не доверять)
# Считаем именно НЕнулевые: длинный хвост zero-месяцев (мёртвый сегмент) не
# добавляет информации регрессии, поэтому в порог не идёт.
_CONF_HIGH_MIN_NONZERO_MONTHS: int = 24
_CONF_MEDIUM_MIN_NONZERO_MONTHS: int = 12
# ── EKB price-bands (₽/м²) ────────────────────────────────────────────────────
# Фиксированные пороги цены за м² (рубли) для price_bucket_of. Источник — рынок
# новостроек ЕКБ 2024-2025 (objective_lots.price_per_m2_rub): здоровое тело
# распределения ~110-220 тыс ₽/м². Берём 4 band'а квартильного духа с КРУГЛЫМИ
# порогами (стабильнее «плавающих» квантилей — детерминированный bucket, не
# зависящий от выборки):
# эконом : < 120 000 ₽/м² (нижний сегмент / окраины)
# комфорт : 120k–160k ₽/м² (массовое тело рынка)
# бизнес : 160k–220k ₽/м² (верх-средний)
# премиум : ≥ 220 000 ₽/м² (премиальные ЖК / центр)
# Границы ВКЛючительны слева (lo ≤ x < hi); подобрано под §9.6 price-сегментацию,
# а НЕ под маркетинговый класс ЖК (потому ключи нейтральные, не путать с `class`).
_PRICE_BAND_ECONOMY_MAX: float = 120_000.0
_PRICE_BAND_COMFORT_MAX: float = 160_000.0
_PRICE_BAND_BUSINESS_MAX: float = 220_000.0
PRICE_BUCKET_ECONOMY: str = "эконом"
PRICE_BUCKET_COMFORT: str = "комфорт"
PRICE_BUCKET_BUSINESS: str = "бизнес"
PRICE_BUCKET_PREMIUM: str = "премиум"
PRICE_BUCKET_UNKNOWN: str = "unknown"
# room×area bucket-ключи — зеркало analytics_queries._BUCKET_PRETTY (§9.6 ждёт
# «Z m²»-подписи). rooms_int — primary axis (mirror _elasticity_per_bucket_coef:
# 0/студия→Студии, 1→1-к, 2→2-к, 3→3-к, 4|5+→80+). area_pd — вторичная страховка
# для пограничных случаев (большие 1-2-к по площади уезжают в 80+).
ROOM_AREA_BUCKET_STUDIO: str = "Студии 15-30"
ROOM_AREA_BUCKET_1K: str = "1-к 30-45"
ROOM_AREA_BUCKET_2K: str = "2-к 45-60"
ROOM_AREA_BUCKET_3K: str = "3-к 60-80"
ROOM_AREA_BUCKET_LARGE: str = "80+ м²"
ROOM_AREA_BUCKET_UNKNOWN: str = "unknown"
# Площадь (м²), с которой лот считается «80+» независимо от комнатности —
# зеркалит верхний bucket _BUCKET_PRETTY ('5-80+ м²' / area_per_unit >= 80).
_LARGE_AREA_THRESHOLD_M2: float = 80.0
@dataclass(frozen=True)
class SegmentSpec:
"""Спецификация сегмента — любой subset осей; None = агрегировать по оси.
obj_class передаётся в «человеческом» регистре; SQL нормализует регистр под
источник (Source A Title-case / Source B lowercase), поэтому здесь регистр
не важен. price_bucket — ключ из price_bucket_of (только Source B; для
Source A игнорируется — у месячного агрегата нет per-lot цены).
ВАЖНО (room_bucket source-specific!): семантика зависит от source:
• Source A (corpus_room_month): СЫРОЙ Objective room_bucket —
"студия" | "1" | "2" | "3" | "4" | "5+".
• Source B (objective_lots): метка из room_area_bucket_of —
напр. "2-к 45-60", "80+ м²".
Передача A-ключа в Source B (или наоборот) → тихий пустой ряд (0 строк).
Для агрегатов класс/район бери Source A; для room×area×price — Source B.
"""
obj_class: str | None = None
room_bucket: str | None = None
district: str | None = None
price_bucket: str | None = None
def as_dict(self) -> dict[str, str | None]:
return {
"obj_class": self.obj_class,
"room_bucket": self.room_bucket,
"district": self.district,
"price_bucket": self.price_bucket,
}
@dataclass(frozen=True)
class SalesSeries:
"""Monthly ряд продаж по сегменту (ТЗ §9.6, Y-ось регрессии).
Все списки выровнены по индексу months (одной длины). units[i] — НАСТОЯЩИЙ
счётчик (0 = честный ноль продаж, не пропуск). area_m2[i]/avg_price_per_m2[i]
— None в месяцах без сделок (среднее по пустому не определено).
"""
months: list[date]
units: list[int]
area_m2: list[float | None]
avg_price_per_m2: list[float | None]
n_months: int
source: SalesSource
segment: dict[str, str | None]
confidence: Confidence
def as_dict(self) -> dict[str, Any]:
return {
"months": [m.isoformat() for m in self.months],
"units": list(self.units),
"area_m2": [_round_or_none(a, 1) for a in self.area_m2],
"avg_price_per_m2": [_round_or_none(p, 0) for p in self.avg_price_per_m2],
"n_months": self.n_months,
"source": self.source,
"segment": dict(self.segment),
"confidence": self.confidence,
}
def _round_or_none(value: float | None, digits: int) -> float | None:
return round(value, digits) if value is not None else None
# ──────────────────────────────────────────────────────────────────────────────
# Pure-логика — без БД, полностью юнит-тестируемо.
# ──────────────────────────────────────────────────────────────────────────────
def price_bucket_of(price_per_m2: float | int | None) -> str:
"""Отнести цену за м² (₽) к фиксированному EKB price-band'у.
Детерминированные круглые пороги (_PRICE_BAND_*): эконом < 120k ≤ комфорт <
160k ≤ бизнес < 220k ≤ премиум. None / непозитивная цена → 'unknown'
(нет осмысленного band'а — не подмешиваем в реальные сегменты). PURE.
Args:
price_per_m2: цена за квадратный метр в рублях.
Returns:
Один из PRICE_BUCKET_* ключей.
"""
if price_per_m2 is None:
return PRICE_BUCKET_UNKNOWN
p = float(price_per_m2)
if p <= 0:
return PRICE_BUCKET_UNKNOWN
if p < _PRICE_BAND_ECONOMY_MAX:
return PRICE_BUCKET_ECONOMY
if p < _PRICE_BAND_COMFORT_MAX:
return PRICE_BUCKET_COMFORT
if p < _PRICE_BAND_BUSINESS_MAX:
return PRICE_BUCKET_BUSINESS
return PRICE_BUCKET_PREMIUM
def room_area_bucket_of(rooms_int: int | None, area_pd: float | int | None) -> str:
"""Отнести лот к room×area bucket'у (зеркало _BUCKET_PRETTY / §9.6 «Z m²»).
Primary axis — rooms_int (mirror _elasticity_per_bucket_coef): 0/студия→Студии,
1→1-к, 2→2-к, 3→3-к, 4+→80+. Вторичная страховка: лот площадью ≥ 80 м²
уезжает в верхний bucket '80+ м²' независимо от комнатности (так же, как
_BUCKET_PRETTY бакетит area_per_unit ≥ 80 в '5-80+ м²'). rooms_int=None И
area=None → 'unknown'; если комнат нет, но площадь известна — решаем по площади
(≥80 → большой, иначе unknown: без комнатности тонкий формат не определить).
PURE.
Args:
rooms_int: число комнат (objective: 0=студия), либо None.
area_pd: площадь лота, м² (для верхней границы), либо None.
Returns:
Один из ROOM_AREA_BUCKET_* ключей.
"""
area = float(area_pd) if area_pd is not None else None
if area is not None and area >= _LARGE_AREA_THRESHOLD_M2:
return ROOM_AREA_BUCKET_LARGE
if rooms_int is None:
return ROOM_AREA_BUCKET_UNKNOWN
if rooms_int <= 0:
return ROOM_AREA_BUCKET_STUDIO
if rooms_int == 1:
return ROOM_AREA_BUCKET_1K
if rooms_int == 2:
return ROOM_AREA_BUCKET_2K
if rooms_int == 3:
return ROOM_AREA_BUCKET_3K
return ROOM_AREA_BUCKET_LARGE
def log_diff(series: list[float | int | None]) -> list[float | None]:
"""Δln ряда для регрессии §9.6: out[t] = ln(x_t) − ln(x_{t-1}).
Логарифм-разность стационаризует ряд продаж (рост в %). Правила None:
• out[0] = None ВСЕГДА (нет предыдущей точки).
• если x_t или x_{t-1} равны None или ≤ 0 → out[t] = None (ln(0)/ln(neg)
= −inf/undefined; 0 продаж — валидное наблюдение для уровня, но не для
Δln, поэтому помечаем пропуском, а не −inf, чтобы не отравить регрессию).
Длина выхода = длине входа. PURE.
Args:
series: ряд значений (обычно units или area), None/0 допустимы.
Returns:
Список Δln той же длины; [0] и любые undefined-точки = None.
"""
out: list[float | None] = []
for i, cur in enumerate(series):
if i == 0:
out.append(None)
continue
prev = series[i - 1]
if cur is None or prev is None:
out.append(None)
continue
cur_f = float(cur)
prev_f = float(prev)
if cur_f <= 0 or prev_f <= 0:
out.append(None)
continue
out.append(math.log(cur_f) - math.log(prev_f))
return out
def fill_month_grid(
by_month: dict[date, tuple[int, float | None, float | None]],
grid: list[date],
) -> tuple[list[int], list[float | None], list[float | None]]:
"""Разложить разрежённые помесячные точки на непрерывную сетку.
Каждая точка by_month[month] = (units, area_m2, avg_price). Месяц сетки,
которого нет в by_month → (0, None, None): units=0 — НАСТОЯЩИЙ ноль («продали
0»), area/price=None (нет сделок → среднего нет). Ключи нормализуются к 1-му
числу (страховка). PURE, не мутирует вход.
Args:
by_month: {month1st: (units, area_m2, avg_price_per_m2)}.
grid: непрерывный ASC-список 1-х чисел месяцев (из _month_grid).
Returns:
(units, area_m2, avg_price_per_m2) — три списка длины len(grid).
"""
normalised = {_month_start(m): v for m, v in by_month.items()}
units: list[int] = []
area: list[float | None] = []
price: list[float | None] = []
for month in grid:
point = normalised.get(month)
if point is None:
units.append(0)
area.append(None)
price.append(None)
else:
u, a, p = point
units.append(int(u))
area.append(a)
price.append(p)
return units, area, price
def _confidence(units: list[int]) -> Confidence:
"""Confidence по числу НЕнулевых месяцев (см. _CONF_* пороги)."""
nonzero = sum(1 for u in units if u > 0)
if nonzero >= _CONF_HIGH_MIN_NONZERO_MONTHS:
return "high"
if nonzero >= _CONF_MEDIUM_MIN_NONZERO_MONTHS:
return "medium"
return "low"
# ──────────────────────────────────────────────────────────────────────────────
# SQL — Source A (objective_corpus_room_month, monthly aggregate)
# ──────────────────────────────────────────────────────────────────────────────
# Source A: суммируем помесячные сделки (ДДУ+ДКП = deals_total_*) по report_month.
# Авторитетен для агрегатов класс/район. Фильтры spec опциональны через
# CAST(:x AS text) IS NULL OR col = :x. class — Title-case в этой таблице, но
# матчим через LOWER(...)=LOWER(...) (регистронезависимо — устойчиво к вариациям
# регистра spec). Площадь — SUM(deals_total_vol_m2); цена — count-weighted AVG
# уже усреднённой колонки (sql.md: AVG по pre-aggregated строкам неверен →
# SUM(avg*cnt)/NULLIF(SUM(cnt),0)). Цена в таблице в тыс.₽/м² → ×1000 к ₽/м²
# (единый масштаб с Source B price_per_m2_rub).
_SOURCE_A_SQL = text(
"""
SELECT
crm.report_month AS month,
SUM(crm.deals_total_count) AS units,
SUM(crm.deals_total_vol_m2) AS area_m2,
SUM(
crm.deals_total_avg_price_thousand_rub_per_m2
* crm.deals_total_count
) / NULLIF(SUM(crm.deals_total_count), 0) * 1000.0 AS avg_price_per_m2
FROM objective_corpus_room_month crm
WHERE crm.report_month >= CAST(:since AS date)
AND (CAST(:cls AS text) IS NULL OR LOWER(crm.class) = LOWER(CAST(:cls AS text)))
AND (CAST(:district AS text) IS NULL OR crm.district = CAST(:district AS text))
AND (
CAST(:room_bucket AS text) IS NULL
OR crm.room_bucket = CAST(:room_bucket AS text)
)
GROUP BY crm.report_month
"""
)
# ──────────────────────────────────────────────────────────────────────────────
# SQL — Source B (objective_lots, per-lot deal dates)
# ──────────────────────────────────────────────────────────────────────────────
# Source B: COUNT(*) проданных лотов по месяцу РЕГИСТРАЦИИ сделки
# (date_trunc('month', registration_date) — есть индекс objective_lots_reg_idx).
# Авторитетен для room×area×price-сегментации (несёт area_pd / price_per_m2_rub
# per-lot, чего Source A не даёт). class — lowercase в этой таблице → матчим
# LOWER(...)=LOWER(...). room×area bucket и price bucket считаем В SQL теми же
# порогами, что pure-helpers (держим логику зеркальной — пороги ниже = константы
# _LARGE_AREA_THRESHOLD_M2 / _PRICE_BAND_*; при правке менять в ОБОИХ местах).
# Цена — средневзвешенная по площади: SUM(price*area)/NULLIF(SUM(area),0)
# (корректнее простого AVG по лотам разной площади).
# КАВЕАТ ВЫЖИВАЕМОСТИ задокументирован в module docstring.
_SOURCE_B_SQL = text(
"""
WITH sold AS (
SELECT
CAST(date_trunc('month', ol.registration_date) AS date) AS month,
ol.area_pd AS area_pd,
ol.price_per_m2_rub AS price_per_m2,
CASE
WHEN ol.area_pd >= CAST(:large_area AS numeric)
THEN CAST(:b_large AS text)
WHEN ol.rooms_int IS NULL THEN CAST(:b_unknown AS text)
WHEN ol.rooms_int <= 0 THEN CAST(:b_studio AS text)
WHEN ol.rooms_int = 1 THEN CAST(:b_1k AS text)
WHEN ol.rooms_int = 2 THEN CAST(:b_2k AS text)
WHEN ol.rooms_int = 3 THEN CAST(:b_3k AS text)
ELSE CAST(:b_large AS text)
END AS room_area_bucket,
CASE
WHEN ol.price_per_m2_rub IS NULL
OR ol.price_per_m2_rub <= 0 THEN CAST(:p_unknown AS text)
WHEN ol.price_per_m2_rub < CAST(:p_economy_max AS numeric)
THEN CAST(:p_economy AS text)
WHEN ol.price_per_m2_rub < CAST(:p_comfort_max AS numeric)
THEN CAST(:p_comfort AS text)
WHEN ol.price_per_m2_rub < CAST(:p_business_max AS numeric)
THEN CAST(:p_business AS text)
ELSE CAST(:p_premium AS text)
END AS price_bucket
FROM objective_lots ol
WHERE ol.premise_kind = CAST(:premise_kind AS text)
AND ol.registration_date IS NOT NULL
AND ol.registration_date >= CAST(:since AS date)
AND (CAST(:cls AS text) IS NULL OR LOWER(ol.class) = LOWER(CAST(:cls AS text)))
AND (
CAST(:district AS text) IS NULL
OR ol.district = CAST(:district AS text)
)
)
SELECT
month,
COUNT(*) AS units,
SUM(area_pd) AS area_m2,
SUM(price_per_m2 * area_pd)
/ NULLIF(SUM(area_pd), 0) AS avg_price_per_m2
FROM sold
WHERE (CAST(:room_bucket AS text) IS NULL OR room_area_bucket = CAST(:room_bucket AS text))
AND (CAST(:price_bucket AS text) IS NULL OR price_bucket = CAST(:price_bucket AS text))
GROUP BY month
"""
)
# premise_kind по умолчанию для Source B (жилые квартиры — единственный сегмент,
# по которому считаем продажи §9.6; зеркалит market_metrics default).
_DEFAULT_PREMISE_KIND: str = "квартира"
def build_sales_series(
db: Session,
*,
spec: SegmentSpec,
source: SalesSource,
months_back: int = _DEFAULT_MONTHS_BACK,
premise_kind: str = _DEFAULT_PREMISE_KIND,
) -> SalesSeries:
"""Собрать monthly ряд продаж по сегменту из выбранного источника.
Ряд строится по НЕПРЕРЫВНОЙ сетке месяцев [start .. текущий] (а не только по
месяцам со сделками): месяц без продаж → units=0 (настоящий ноль), area/price
= None. Это даёт регрессии §9.6 регулярный шаг по времени.
Source выбирает зону авторитетности (см. module docstring):
• 'corpus_room_month' — агрегаты класс/район (полные сделки ДДУ+ДКП).
• 'objective_lots' — тонкая room×area×price-сегментация (но survivorship
bias на старых месяцах — каведат в module docstring).
Graceful: при сбое БД / пустых данных возвращается ряд по сетке с units=0,
area/price=None, confidence='low' (НЕ crash). Пустой ряд (months=[]) — только
если сетка пуста (months_back < 0).
Args:
db: SQLAlchemy sync Session.
spec: сегмент (любой subset осей; None-поля = агрегировать по оси).
source: 'corpus_room_month' | 'objective_lots'.
months_back: глубина ряда (по умолчанию _DEFAULT_MONTHS_BACK).
premise_kind: тип помещения для Source B (по умолчанию 'квартира').
Returns:
SalesSeries (всегда; [] months только при пустой сетке).
"""
today = date.today()
start = _shift_months(today, -max(0, months_back))
grid = _month_grid(start, _month_start(today))
segment = spec.as_dict()
if not grid:
return SalesSeries(
months=[],
units=[],
area_m2=[],
avg_price_per_m2=[],
n_months=0,
source=source,
segment=segment,
confidence="low",
)
if source == "corpus_room_month":
by_month = _query_source_a(db, spec=spec, since=start)
else:
by_month = _query_source_b(db, spec=spec, since=start, premise_kind=premise_kind)
units, area_m2, avg_price = fill_month_grid(by_month, grid)
confidence = _confidence(units)
logger.info(
"build_sales_series: source=%s months=%d nonzero=%d "
"class=%s room=%s district=%s price=%s confidence=%s",
source,
len(grid),
sum(1 for u in units if u > 0),
spec.obj_class,
spec.room_bucket,
spec.district,
spec.price_bucket,
confidence,
)
return SalesSeries(
months=grid,
units=units,
area_m2=area_m2,
avg_price_per_m2=avg_price,
n_months=len(grid),
source=source,
segment=segment,
confidence=confidence,
)
def _query_source_a(
db: Session, *, spec: SegmentSpec, since: date
) -> dict[date, tuple[int, float | None, float | None]]:
"""Source A (corpus_room_month) → {month1st: (units, area, avg_price)}.
Graceful → {} при сбое/пустых данных. price_bucket в spec для Source A
игнорируется (агрегат не несёт per-lot цены) — фиксируется логом.
"""
if spec.price_bucket is not None:
logger.info(
"build_sales_series: price_bucket=%s ignored for source=corpus_room_month "
"(monthly aggregate carries no per-lot price)",
spec.price_bucket,
)
params = {
"since": since,
"cls": spec.obj_class,
"district": spec.district,
"room_bucket": spec.room_bucket,
}
try:
rows = db.execute(_SOURCE_A_SQL, params).mappings().all()
except Exception:
logger.exception("build_sales_series: source A query failed")
return {}
return _rows_to_by_month(rows)
def _query_source_b(
db: Session, *, spec: SegmentSpec, since: date, premise_kind: str
) -> dict[date, tuple[int, float | None, float | None]]:
"""Source B (objective_lots) → {month1st: (units, area, avg_price)}.
Graceful → {} при сбое/пустых данных. Передаёт bucket-пороги/-метки в SQL
(зеркало pure-helpers), чтобы room×area / price сегментация считалась тем же
правилом и в БД, и в Python.
"""
params = {
"since": since,
"premise_kind": premise_kind,
"cls": spec.obj_class,
"district": spec.district,
"room_bucket": spec.room_bucket,
"price_bucket": spec.price_bucket,
# bucket-пороги (зеркало _LARGE_AREA_THRESHOLD_M2 / _PRICE_BAND_*).
"large_area": _LARGE_AREA_THRESHOLD_M2,
"p_economy_max": _PRICE_BAND_ECONOMY_MAX,
"p_comfort_max": _PRICE_BAND_COMFORT_MAX,
"p_business_max": _PRICE_BAND_BUSINESS_MAX,
# bucket-метки (зеркало ROOM_AREA_BUCKET_* / PRICE_BUCKET_*).
"b_studio": ROOM_AREA_BUCKET_STUDIO,
"b_1k": ROOM_AREA_BUCKET_1K,
"b_2k": ROOM_AREA_BUCKET_2K,
"b_3k": ROOM_AREA_BUCKET_3K,
"b_large": ROOM_AREA_BUCKET_LARGE,
"b_unknown": ROOM_AREA_BUCKET_UNKNOWN,
"p_economy": PRICE_BUCKET_ECONOMY,
"p_comfort": PRICE_BUCKET_COMFORT,
"p_business": PRICE_BUCKET_BUSINESS,
"p_premium": PRICE_BUCKET_PREMIUM,
"p_unknown": PRICE_BUCKET_UNKNOWN,
}
try:
rows = db.execute(_SOURCE_B_SQL, params).mappings().all()
except Exception:
logger.exception("build_sales_series: source B query failed")
return {}
return _rows_to_by_month(rows)
def _rows_to_by_month(
rows: list[Any],
) -> dict[date, tuple[int, float | None, float | None]]:
"""Свести строки (month, units, area_m2, avg_price_per_m2) в {month1st: tuple}.
Площадь/цена → None если NULL (нет сделок с известным значением). units
приводим к int (≥0). Строки с month IS NULL пропускаем (защита от мусора).
"""
out: dict[date, tuple[int, float | None, float | None]] = {}
for r in rows:
month = r["month"]
if month is None:
continue
units = int(r["units"] or 0)
area = float(r["area_m2"]) if r["area_m2"] is not None else None
price = float(r["avg_price_per_m2"]) if r["avg_price_per_m2"] is not None else None
out[_month_start(month)] = (units, area, price)
return out

645
backend/tests/services/forecasting/test_sales_series.py Normal file
View file

@ -0,0 +1,645 @@
"""Unit-тесты monthly ряда продаж по сегменту (#951c, ТЗ §9.6, Y-ось регрессии).
Чистые тесты (без живой БД):
• price_bucket_of — границы band'ов (включительно слева), None/≤0 → 'unknown'.
• room_area_bucket_of — rooms→bucket, area≥80 override, unknown-кейсы.
• log_diff — Δln, [0]=None всегда, ноль/None/neg → None (не −inf), длина.
• fill_month_grid — zero-fill месяцев (units=0 НАСТОЯЩИЙ, area/price=None).
• SalesSeries.as_dict / SegmentSpec.as_dict — округление + None survive.
• build_sales_series через MagicMock-сессию: правильная таблица (Source A vs B),
GROUP BY date_trunc для Source B, CAST(:x AS type) не :x::type, case-handling
класса (LOWER=LOWER), zero-fill месяцев, тиры confidence, graceful empty → low.
psycopg v3 правило проверяется явно: bind-параметры — CAST(:x AS type).
"""
from __future__ import annotations
import datetime as dt
import math
import os
from unittest.mock import MagicMock
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
from app.services.forecasting.macro_series import _month_grid, _month_start, _shift_months
from app.services.forecasting.sales_series import (
PRICE_BUCKET_BUSINESS,
PRICE_BUCKET_COMFORT,
PRICE_BUCKET_ECONOMY,
PRICE_BUCKET_PREMIUM,
PRICE_BUCKET_UNKNOWN,
ROOM_AREA_BUCKET_1K,
ROOM_AREA_BUCKET_2K,
ROOM_AREA_BUCKET_3K,
ROOM_AREA_BUCKET_LARGE,
ROOM_AREA_BUCKET_STUDIO,
ROOM_AREA_BUCKET_UNKNOWN,
SalesSeries,
SegmentSpec,
_confidence,
build_sales_series,
fill_month_grid,
log_diff,
price_bucket_of,
room_area_bucket_of,
)
# ── pure: price_bucket_of ─────────────────────────────────────────────────────
class TestPriceBucketOf:
def test_economy_below_120k(self) -> None:
assert price_bucket_of(90_000) == PRICE_BUCKET_ECONOMY
assert price_bucket_of(119_999) == PRICE_BUCKET_ECONOMY
def test_comfort_band(self) -> None:
# Граница 120k включительна слева (lo ≤ x < hi).
assert price_bucket_of(120_000) == PRICE_BUCKET_COMFORT
assert price_bucket_of(159_999) == PRICE_BUCKET_COMFORT
def test_business_band(self) -> None:
assert price_bucket_of(160_000) == PRICE_BUCKET_BUSINESS
assert price_bucket_of(219_999) == PRICE_BUCKET_BUSINESS
def test_premium_at_and_above_220k(self) -> None:
assert price_bucket_of(220_000) == PRICE_BUCKET_PREMIUM
assert price_bucket_of(500_000) == PRICE_BUCKET_PREMIUM
def test_none_is_unknown(self) -> None:
assert price_bucket_of(None) == PRICE_BUCKET_UNKNOWN
def test_zero_and_negative_unknown(self) -> None:
# Цена ≤ 0 бессмысленна → unknown (не подмешиваем в реальные band'ы).
assert price_bucket_of(0) == PRICE_BUCKET_UNKNOWN
assert price_bucket_of(-5) == PRICE_BUCKET_UNKNOWN
def test_float_input(self) -> None:
assert price_bucket_of(155_500.75) == PRICE_BUCKET_COMFORT
# ── pure: room_area_bucket_of ─────────────────────────────────────────────────
class TestRoomAreaBucketOf:
def test_studio_zero_rooms(self) -> None:
# objective: 0 = студия.
assert room_area_bucket_of(0, 25.0) == ROOM_AREA_BUCKET_STUDIO
def test_studio_negative_treated_as_studio(self) -> None:
assert room_area_bucket_of(-1, 22.0) == ROOM_AREA_BUCKET_STUDIO
def test_one_room(self) -> None:
assert room_area_bucket_of(1, 38.0) == ROOM_AREA_BUCKET_1K
def test_two_rooms(self) -> None:
assert room_area_bucket_of(2, 55.0) == ROOM_AREA_BUCKET_2K
def test_three_rooms(self) -> None:
assert room_area_bucket_of(3, 70.0) == ROOM_AREA_BUCKET_3K
def test_four_plus_rooms_large(self) -> None:
assert room_area_bucket_of(4, 95.0) == ROOM_AREA_BUCKET_LARGE
assert room_area_bucket_of(5, 120.0) == ROOM_AREA_BUCKET_LARGE
def test_area_override_pushes_small_rooms_to_large(self) -> None:
# Площадь ≥ 80 м² → '80+' независимо от комнатности (зеркало _BUCKET_PRETTY).
assert room_area_bucket_of(1, 85.0) == ROOM_AREA_BUCKET_LARGE
assert room_area_bucket_of(2, 80.0) == ROOM_AREA_BUCKET_LARGE
def test_area_just_below_threshold_keeps_room_bucket(self) -> None:
# 79.9 < 80 → решаем по комнатности.
assert room_area_bucket_of(2, 79.9) == ROOM_AREA_BUCKET_2K
def test_rooms_none_area_none_unknown(self) -> None:
assert room_area_bucket_of(None, None) == ROOM_AREA_BUCKET_UNKNOWN
def test_rooms_none_large_area_is_large(self) -> None:
# Комнат нет, но площадь ≥ 80 → большой (area override срабатывает первым).
assert room_area_bucket_of(None, 90.0) == ROOM_AREA_BUCKET_LARGE
def test_rooms_none_small_area_unknown(self) -> None:
# Без комнатности тонкий формат не определить → unknown.
assert room_area_bucket_of(None, 40.0) == ROOM_AREA_BUCKET_UNKNOWN
def test_rooms_known_area_none(self) -> None:
# Площадь неизвестна → решаем чисто по комнатности.
assert room_area_bucket_of(1, None) == ROOM_AREA_BUCKET_1K
# ── pure: log_diff ────────────────────────────────────────────────────────────
class TestLogDiff:
def test_first_element_always_none(self) -> None:
assert log_diff([10, 20, 30])[0] is None
def test_basic_log_difference(self) -> None:
out = log_diff([10, 20])
assert out[0] is None
assert out[1] is not None
assert math.isclose(out[1], math.log(20) - math.log(10))
def test_length_matches_input(self) -> None:
assert len(log_diff([1, 2, 3, 4, 5])) == 5
def test_zero_current_is_none(self) -> None:
# ln(0) = −inf → помечаем None (0 продаж — валидный уровень, не Δln).
out = log_diff([10, 0, 10])
assert out[1] is None # cur=0
assert out[2] is None # prev=0
def test_none_in_series_yields_none(self) -> None:
out = log_diff([10, None, 30])
assert out[1] is None # cur=None
assert out[2] is None # prev=None
def test_negative_yields_none(self) -> None:
# ln(neg) не определён → None.
out = log_diff([10, -5])
assert out[1] is None
def test_empty(self) -> None:
assert log_diff([]) == []
def test_single_element(self) -> None:
assert log_diff([42]) == [None]
def test_no_minus_inf_anywhere(self) -> None:
# Гарантия: ни одна точка не −inf/nan (главная цель zero-handling).
out = log_diff([0, 5, 0, 8, None, 3])
for v in out:
assert v is None or (math.isfinite(v))
# ── pure: fill_month_grid ─────────────────────────────────────────────────────
class TestFillMonthGrid:
def test_zero_fill_missing_months(self) -> None:
grid = _month_grid(dt.date(2024, 1, 1), dt.date(2024, 3, 1))
by_month = {dt.date(2024, 2, 1): (5, 250.0, 150_000.0)}
units, area, price = fill_month_grid(by_month, grid)
# Январь и март без сделок → units=0 (НАСТОЯЩИЙ ноль), area/price=None.
assert units == [0, 5, 0]
assert area == [None, 250.0, None]
assert price == [None, 150_000.0, None]
def test_zero_is_real_not_none(self) -> None:
# Ключевое отличие: пропущенный месяц = 0 units (не None) — 0 это данные.
grid = _month_grid(dt.date(2024, 1, 1), dt.date(2024, 1, 1))
units, _area, _price = fill_month_grid({}, grid)
assert units == [0]
assert units[0] == 0 and units[0] is not None
def test_present_month_passes_through(self) -> None:
grid = [dt.date(2024, 5, 1)]
by_month = {dt.date(2024, 5, 1): (12, 600.0, 140_000.0)}
units, area, price = fill_month_grid(by_month, grid)
assert (units, area, price) == ([12], [600.0], [140_000.0])
def test_area_price_none_when_units_present_but_value_missing(self) -> None:
# Сделки есть, но area/price NULL в источнике → None сохраняется.
grid = [dt.date(2024, 5, 1)]
by_month = {dt.date(2024, 5, 1): (3, None, None)}
units, area, price = fill_month_grid(by_month, grid)
assert units == [3]
assert area == [None]
assert price == [None]
def test_keys_normalised_to_first_of_month(self) -> None:
# Ключ-середина месяца нормализуется к 1-му числу.
grid = [dt.date(2024, 5, 1)]
by_month = {dt.date(2024, 5, 17): (7, 350.0, 130_000.0)}
units, area, price = fill_month_grid(by_month, grid)
assert units == [7]
assert area == [350.0]
assert price == [130_000.0]
def test_does_not_mutate_input(self) -> None:
by_month = {dt.date(2024, 5, 1): (1, 50.0, 100_000.0)}
fill_month_grid(by_month, [dt.date(2024, 5, 1)])
assert by_month == {dt.date(2024, 5, 1): (1, 50.0, 100_000.0)}
# ── pure: _confidence ─────────────────────────────────────────────────────────
class TestConfidence:
def test_high_at_24_nonzero(self) -> None:
assert _confidence([1] * 24) == "high"
def test_medium_at_12_nonzero(self) -> None:
assert _confidence([1] * 12) == "medium"
def test_low_below_12_nonzero(self) -> None:
assert _confidence([1] * 11) == "low"
def test_zeros_do_not_count(self) -> None:
# 30 месяцев, но только 5 ненулевых → low (хвост нулей не информативен).
units = [1] * 5 + [0] * 25
assert _confidence(units) == "low"
def test_high_with_zeros_mixed(self) -> None:
# 24 ненулевых + сколько угодно нулей → high (порог по ненулевым).
units = [1] * 24 + [0] * 10
assert _confidence(units) == "high"
def test_empty_is_low(self) -> None:
assert _confidence([]) == "low"
# ── SalesSeries / SegmentSpec as_dict ─────────────────────────────────────────
class TestAsDict:
def test_sales_series_rounds_and_serialises(self) -> None:
s = SalesSeries(
months=[dt.date(2024, 1, 1), dt.date(2024, 2, 1)],
units=[5, 0],
area_m2=[250.456, None],
avg_price_per_m2=[150_123.7, None],
n_months=2,
source="objective_lots",
segment={
"obj_class": "комфорт",
"room_bucket": None,
"district": None,
"price_bucket": None,
},
confidence="low",
)
d = s.as_dict()
assert d["months"] == ["2024-01-01", "2024-02-01"]
assert d["units"] == [5, 0]
assert d["area_m2"] == [250.5, None]
assert d["avg_price_per_m2"] == [150_124, None]
assert d["n_months"] == 2
assert d["source"] == "objective_lots"
assert d["confidence"] == "low"
assert d["segment"]["obj_class"] == "комфорт"
def test_units_zero_survives_as_zero(self) -> None:
# as_dict не должен превращать 0 в None.
s = SalesSeries(
months=[dt.date(2024, 1, 1)],
units=[0],
area_m2=[None],
avg_price_per_m2=[None],
n_months=1,
source="corpus_room_month",
segment={},
confidence="low",
)
assert s.as_dict()["units"] == [0]
def test_segment_spec_as_dict_subset(self) -> None:
spec = SegmentSpec(obj_class="Комфорт", district="Автовокзал")
assert spec.as_dict() == {
"obj_class": "Комфорт",
"room_bucket": None,
"district": "Автовокзал",
"price_bucket": None,
}
# ── build_sales_series: MagicMock-сессия (форма SQL + zero-fill + graceful) ────
def _result(rows: list[dict]) -> MagicMock:
"""Результат db.execute(...).mappings().all() → rows (list of dict-like)."""
res = MagicMock()
res.mappings.return_value.all.return_value = rows
return res
def _sql_of(db: MagicMock, call_idx: int = 0) -> str:
return str(db.execute.call_args_list[call_idx].args[0])
def _params_of(db: MagicMock, call_idx: int = 0) -> dict:
return db.execute.call_args_list[call_idx].args[1]
class TestBuildSalesSeriesSourceShape:
def test_source_a_queries_corpus_table(self) -> None:
db = MagicMock()
db.execute.return_value = _result([])
build_sales_series(
db,
spec=SegmentSpec(obj_class="Комфорт"),
source="corpus_room_month",
months_back=3,
)
sql = _sql_of(db)
assert "objective_corpus_room_month" in sql
assert "deals_total_count" in sql
assert "deals_total_vol_m2" in sql
assert "deals_total_avg_price_thousand_rub_per_m2" in sql
def test_source_b_queries_lots_with_date_trunc_groupby(self) -> None:
db = MagicMock()
db.execute.return_value = _result([])
build_sales_series(
db,
spec=SegmentSpec(),
source="objective_lots",
months_back=3,
)
sql = _sql_of(db)
assert "objective_lots" in sql
# Source B группирует по месяцу РЕГИСТРАЦИИ через date_trunc.
assert "date_trunc('month', ol.registration_date)" in sql
assert "GROUP BY month" in sql
assert "COUNT(*)" in sql
def test_source_a_uses_cast_not_double_colon(self) -> None:
db = MagicMock()
db.execute.return_value = _result([])
build_sales_series(
db,
spec=SegmentSpec(obj_class="Бизнес", district="Центр", room_bucket="2"),
source="corpus_room_month",
months_back=3,
)
sql = _sql_of(db)
assert "CAST(:since AS date)" in sql
assert "CAST(:cls AS text)" in sql
assert "CAST(:district AS text)" in sql
assert "CAST(:room_bucket AS text)" in sql
# psycopg v3 trap: никаких :name::type.
assert "::" not in sql
def test_source_b_uses_cast_not_double_colon(self) -> None:
db = MagicMock()
db.execute.return_value = _result([])
build_sales_series(
db,
spec=SegmentSpec(price_bucket="комфорт"),
source="objective_lots",
months_back=3,
)
sql = _sql_of(db)
assert "CAST(:since AS date)" in sql
assert "CAST(:premise_kind AS text)" in sql
assert "CAST(:large_area AS numeric)" in sql
assert "CAST(:price_bucket AS text)" in sql
assert "::" not in sql
def test_class_case_insensitive_match_both_sources(self) -> None:
# Source A — Title-case в БД, Source B — lowercase: оба матчат LOWER=LOWER.
db = MagicMock()
db.execute.return_value = _result([])
build_sales_series(
db,
spec=SegmentSpec(obj_class="Комфорт"),
source="corpus_room_month",
months_back=1,
)
assert "LOWER(crm.class) = LOWER(CAST(:cls AS text))" in _sql_of(db)
db2 = MagicMock()
db2.execute.return_value = _result([])
build_sales_series(
db2,
spec=SegmentSpec(obj_class="комфорт"),
source="objective_lots",
months_back=1,
)
assert "LOWER(ol.class) = LOWER(CAST(:cls AS text))" in _sql_of(db2)
def test_source_a_params_pass_spec(self) -> None:
db = MagicMock()
db.execute.return_value = _result([])
build_sales_series(
db,
spec=SegmentSpec(obj_class="Комфорт", district="Уралмаш", room_bucket="1"),
source="corpus_room_month",
months_back=12,
)
params = _params_of(db)
assert params["cls"] == "Комфорт"
assert params["district"] == "Уралмаш"
assert params["room_bucket"] == "1"
assert isinstance(params["since"], dt.date)
def test_source_b_passes_bucket_thresholds_and_labels(self) -> None:
db = MagicMock()
db.execute.return_value = _result([])
build_sales_series(
db,
spec=SegmentSpec(),
source="objective_lots",
months_back=1,
)
params = _params_of(db)
# Пороги/метки bucket'ов передаются параметрами (зеркало pure-helpers).
assert params["large_area"] == 80.0
assert params["p_economy_max"] == 120_000.0
assert params["b_studio"] == ROOM_AREA_BUCKET_STUDIO
assert params["p_premium"] == PRICE_BUCKET_PREMIUM
assert params["premise_kind"] == "квартира"
def test_source_b_custom_premise_kind(self) -> None:
db = MagicMock()
db.execute.return_value = _result([])
build_sales_series(
db,
spec=SegmentSpec(),
source="objective_lots",
months_back=1,
premise_kind="нежилое",
)
assert _params_of(db)["premise_kind"] == "нежилое"
class TestBuildSalesSeriesLogic:
def test_continuous_grid_with_zero_fill(self) -> None:
today = dt.date.today()
target = _shift_months(today, -1)
db = MagicMock()
db.execute.return_value = _result(
[{"month": target, "units": 7, "area_m2": 350.0, "avg_price_per_m2": 145_000.0}]
)
out = build_sales_series(
db,
spec=SegmentSpec(),
source="objective_lots",
months_back=3,
)
# Непрерывная сетка 4 месяца (-3..0).
assert out.n_months == 4
assert len(out.months) == 4
assert out.months == sorted(out.months)
idx = out.months.index(target)
assert out.units[idx] == 7
assert out.area_m2[idx] == 350.0
assert out.avg_price_per_m2[idx] == 145_000.0
# Месяцы без сделок → units=0 (настоящий), area/price=None.
other = [i for i in range(out.n_months) if i != idx]
for i in other:
assert out.units[i] == 0
assert out.area_m2[i] is None
assert out.avg_price_per_m2[i] is None
def test_source_a_price_scaled_to_rub_per_m2(self) -> None:
# Source A SQL умножает тыс.₽/м² на 1000 — проверяем наличие масштаба в SQL.
db = MagicMock()
db.execute.return_value = _result([])
build_sales_series(
db,
spec=SegmentSpec(),
source="corpus_room_month",
months_back=1,
)
assert "* 1000.0" in _sql_of(db)
def test_confidence_high_with_24_nonzero(self) -> None:
today = dt.date.today()
rows = [
{
"month": _shift_months(today, -k),
"units": 3,
"area_m2": 100.0,
"avg_price_per_m2": 130_000.0,
}
for k in range(24)
]
db = MagicMock()
db.execute.return_value = _result(rows)
out = build_sales_series(
db,
spec=SegmentSpec(),
source="objective_lots",
months_back=30,
)
assert out.confidence == "high"
def test_confidence_medium_with_12_nonzero(self) -> None:
today = dt.date.today()
rows = [
{
"month": _shift_months(today, -k),
"units": 2,
"area_m2": 80.0,
"avg_price_per_m2": 120_000.0,
}
for k in range(12)
]
db = MagicMock()
db.execute.return_value = _result(rows)
out = build_sales_series(
db,
spec=SegmentSpec(),
source="objective_lots",
months_back=20,
)
assert out.confidence == "medium"
def test_confidence_low_thin_data(self) -> None:
today = dt.date.today()
rows = [
{
"month": _shift_months(today, -k),
"units": 1,
"area_m2": 40.0,
"avg_price_per_m2": 110_000.0,
}
for k in range(3)
]
db = MagicMock()
db.execute.return_value = _result(rows)
out = build_sales_series(
db,
spec=SegmentSpec(),
source="objective_lots",
months_back=12,
)
assert out.confidence == "low"
def test_segment_recorded_in_result(self) -> None:
db = MagicMock()
db.execute.return_value = _result([])
spec = SegmentSpec(
obj_class="Комфорт", room_bucket="2", district="Центр", price_bucket="бизнес"
)
out = build_sales_series(
db,
spec=spec,
source="objective_lots",
months_back=1,
)
assert out.segment == spec.as_dict()
assert out.source == "objective_lots"
class TestBuildSalesSeriesGraceful:
def test_empty_data_returns_zero_filled_low(self) -> None:
db = MagicMock()
db.execute.return_value = _result([])
out = build_sales_series(
db,
spec=SegmentSpec(),
source="objective_lots",
months_back=2,
)
assert out.n_months == 3 # -2..0
assert out.units == [0, 0, 0]
assert out.area_m2 == [None, None, None]
assert out.avg_price_per_m2 == [None, None, None]
assert out.confidence == "low"
def test_db_exception_graceful_source_a(self) -> None:
db = MagicMock()
db.execute.side_effect = RuntimeError("db down")
out = build_sales_series(
db,
spec=SegmentSpec(),
source="corpus_room_month",
months_back=2,
)
# Ряд по сетке всё равно, zero-filled, low (НЕ crash).
assert out.n_months == 3
assert out.units == [0, 0, 0]
assert out.confidence == "low"
def test_db_exception_graceful_source_b(self) -> None:
db = MagicMock()
db.execute.side_effect = RuntimeError("db down")
out = build_sales_series(
db,
spec=SegmentSpec(),
source="objective_lots",
months_back=1,
)
assert out.n_months == 2
assert out.units == [0, 0]
assert out.confidence == "low"
def test_negative_months_back_clamps_to_single_month(self) -> None:
# months_back<0 клампится к 0 (как PR2: -max(0, months_back)) → один
# текущий месяц, валидный объект, low (НЕ crash, НЕ отрицательная сетка).
db = MagicMock()
db.execute.return_value = _result([])
out = build_sales_series(
db,
spec=SegmentSpec(),
source="objective_lots",
months_back=-5,
)
assert out.n_months == 1
assert out.months == [_month_start(dt.date.today())]
assert out.units == [0]
assert out.confidence == "low"
def test_month_back_zero_single_month(self) -> None:
db = MagicMock()
db.execute.return_value = _result([])
out = build_sales_series(
db,
spec=SegmentSpec(),
source="objective_lots",
months_back=0,
)
assert out.n_months == 1 # только текущий месяц
assert out.units == [0]