gendesign/data/sql/126_domrf_housing_macro.sql

-- 126_domrf_housing_macro.sql
-- #946 (Site Finder v2 / макро-слой, ТЗ §7.10 «Жильё / эскроу dom.рф → macro_indicator»).
-- Агрегирует УЖЕ заскрейпленные domrf_kn_sale_graph + domrf_kn_objects в месячные
-- ряды macro_indicator (source='domrf', frequency='monthly'). ЧИСТЫЙ SQL-aggregation —
-- скрейпер НЕ нужен (данные уже в БД, см. м.43/51). Зеркалит domrf-backfill-блок м.123
-- (INSERT...SELECT...ON CONFLICT для ипотечных ставок). Никаких внешних вызовов, без LLM.
--
-- Что строим (3 измеренных жилищных макро-сигнала §7.10):
--   1. housing_price_per_m2  — ПЛОЩАДЬ-взвешенная средняя цена ₽/м² по (регион, месяц).
--   2. new_contracts_volume  — Σ заключённых ДДУ по (регион, месяц) (измеренный объём dom.рф).
--   3. escrow_objects_count  — кол-во объектов с escrow=TRUE в СВЕЖЕМ снапшоте (текущий срез).
--   regions: 'sverdl' (region_cd=66, Свердловская) и 'rf' (все регионы агрегировано) — оба.
--
-- §7.11 (EMISS: ИПЦ/реальные доходы Росстата) — ОТЛОЖЕН и НЕ входит в эту миграцию:
--   fedstat.ru за WAF (блокирует автоматический забор), источник пока недоступен. key_rate
--   (LIVE-скрапер ставки ЦБ) — отдельный follow-up #945-PR B, тоже не здесь.
--
-- КРИТИЧНО — dedup по свежему снапшоту (иначе двойной счёт):
--   • domrf_kn_sale_graph НАКАПЛИВАЕТ снапшоты (одна (obj_id, report_month, type) пара
--     повторяется на каждый snapshot_date — UNIQUE(obj_id,report_month,type,snapshot_date)).
--     Наивный SUM(contracted) сложил бы ВСЕ снапшоты одного месяца → кратное завышение.
--     Поэтому СНАЧАЛА DISTINCT ON (obj_id, report_month) ORDER BY ... snapshot_date DESC —
--     ровно одна (последняя) строка на ЖК×месяц, и только ПОТОМ агрегируем.
--   • domrf_kn_objects тоже снапшотится → берём свежий снапшот на obj_id тем же
--     DISTINCT ON (obj_id) ORDER BY obj_id, snapshot_date DESC (как supply_layers.py / м.121).
--   • Фильтр type='apartments' — parking/нежилое из sale_graph НЕ должно загрязнять жилищный
--     ₽/м² и ДДУ-объём (так же фильтруют м.120 и v_bucket_success_score / м.86).
--
-- Площадь-взвешенная средняя (обязательна по .claude/rules/sql.md — наивный AVG(price_avg)
--   занижает, мелкие ЖК перевешивают): SUM(price_avg*area_sq)/NULLIF(SUM(area_sq),0).
--   NULLIF(...,0) — защита от деления на ноль (all-zero месяц → NULL, не фейковый 0).
--
-- region-split — ДВА ОТДЕЛЬНЫХ INSERT'а на показатель (НЕ GROUPING SETS):
--   'sverdl' = WHERE region_cd=66; 'rf' = БЕЗ фильтра региона (все регионы агрегировано).
--   Раздельные стейтменты выбраны намеренно: семантика каждого региона явная, нет риска
--   NULL-метки региона из rollup-строки GROUPING SETS. Зеркалит стиль дискретных INSERT'ов м.123.
--
-- ЭСКРОУ-КАВЕАТ (важно): из dom.рф заскрейплен ТОЛЬКО булев флаг domrf_kn_objects.escrow
--   (объект продаётся через эскроу — да/нет). Денежные остатки/наполнение эскроу-счетов
--   (₽ на счетах, раскрытие, покрытие кредита) НЕ скрейпятся и НЕ доступны. Поэтому
--   escrow_objects_count — это СЧЁТ ОБЪЕКТОВ под эскроу, НЕ сумма денег. Документировано
--   в comment каждой соответствующей строки.
--
-- Зависимости (источники, НЕ дропаются — продолжают писать своими импортёрами):
--   • domrf_kn_sale_graph  (м.51) — obj_id, report_month, type, contracted, area_sq,
--                                    price_avg, snapshot_date.
--   • domrf_kn_objects     (м.43) — obj_id, region_cd (66=Свердловская), escrow, snapshot_date.
--   • macro_indicator      (м.123) — целевая таблица, PK (indicator_type, region, obs_date).
--
-- Deploy: auto-applied by deploy.yml через _schema_migrations tracking (ровно один раз по NN).
--   Idempotent (должен ре-applied чисто ДВАЖДЫ): INSERT ... ON CONFLICT DO UPDATE SET
--   value/comment/updated_at — повторный прогон обновит value, дублей по PK не будет.
--   Никаких DROP/TRUNCATE, новой таблицы НЕ создаём (переиспользуем macro_indicator).
-- Migration order: эта SQL-миграция применяется ПЕРВОЙ; backend macro-reader — следом.

BEGIN;

-- ════════════════════════════════════════════════════════════════════════════════
-- Повторяющиеся CTE-формы (физически продублированы в каждом стейтменте — SQL НЕ делит
-- CTE между стейтментами; держим формулу единой для аудита):
--   sg_latest  := DISTINCT ON (obj_id, report_month) ... WHERE type='apartments'
--                 ORDER BY obj_id, report_month, snapshot_date DESC   -- свежий снапшот на ЖК×месяц
--   obj_region := DISTINCT ON (obj_id) obj_id, region_cd
--                 ORDER BY obj_id, snapshot_date DESC NULLS LAST       -- свежий снапшот объекта
-- ════════════════════════════════════════════════════════════════════════════════

-- ── 1a. housing_price_per_m2 / sverdl — площадь-взвешенная ₽/м² (region_cd=66) ───
WITH sg_latest AS (
    SELECT DISTINCT ON (sg.obj_id, sg.report_month)
        sg.obj_id, sg.report_month, sg.price_avg, sg.area_sq
      FROM domrf_kn_sale_graph AS sg
     WHERE sg.type = 'apartments'
     ORDER BY sg.obj_id, sg.report_month, sg.snapshot_date DESC
),
obj_region AS (
    SELECT DISTINCT ON (o.obj_id) o.obj_id, o.region_cd
      FROM domrf_kn_objects AS o
     ORDER BY o.obj_id, o.snapshot_date DESC NULLS LAST
)
INSERT INTO macro_indicator
    (indicator_type, region, obs_date, value, source, frequency, unit, comment)
SELECT
    'housing_price_per_m2'                                   AS indicator_type,
    'sverdl'                                                 AS region,
    s.report_month                                          AS obs_date,
    SUM(s.price_avg * s.area_sq) / NULLIF(SUM(s.area_sq), 0) AS value,
    'domrf'                                                  AS source,
    'monthly'                                                AS frequency,
    '₽/м²'                                                   AS unit,
    'ТЗ §7.10: площадь-взвешенная средняя цена первички '
    || 'SUM(price_avg*area_sq)/SUM(area_sq) из domrf_kn_sale_graph '
    || '(type=apartments, latest snapshot per ЖК×месяц), region_cd=66. Источник dom.рф.' AS comment
FROM sg_latest AS s
JOIN obj_region AS o ON o.obj_id = s.obj_id
WHERE o.region_cd = 66
  AND s.price_avg IS NOT NULL
  AND s.area_sq IS NOT NULL
GROUP BY s.report_month
HAVING SUM(s.area_sq) > 0
ON CONFLICT (indicator_type, region, obs_date) DO UPDATE
SET value      = EXCLUDED.value,
    source     = EXCLUDED.source,
    frequency  = EXCLUDED.frequency,
    unit       = EXCLUDED.unit,
    comment    = EXCLUDED.comment,
    updated_at = now();

-- ── 1b. housing_price_per_m2 / rf — площадь-взвешенная ₽/м² (все регионы) ────────
-- join к objects НЕ нужен (region-фильтра нет) — агрегируем прямо по sg_latest.
WITH sg_latest AS (
    SELECT DISTINCT ON (sg.obj_id, sg.report_month)
        sg.obj_id, sg.report_month, sg.price_avg, sg.area_sq
      FROM domrf_kn_sale_graph AS sg
     WHERE sg.type = 'apartments'
     ORDER BY sg.obj_id, sg.report_month, sg.snapshot_date DESC
)
INSERT INTO macro_indicator
    (indicator_type, region, obs_date, value, source, frequency, unit, comment)
SELECT
    'housing_price_per_m2'                                   AS indicator_type,
    'rf'                                                     AS region,
    s.report_month                                          AS obs_date,
    SUM(s.price_avg * s.area_sq) / NULLIF(SUM(s.area_sq), 0) AS value,
    'domrf'                                                  AS source,
    'monthly'                                                AS frequency,
    '₽/м²'                                                   AS unit,
    'ТЗ §7.10: площадь-взвешенная средняя цена первички '
    || 'SUM(price_avg*area_sq)/SUM(area_sq) из domrf_kn_sale_graph '
    || '(type=apartments, latest snapshot per ЖК×месяц), все регионы. Источник dom.рф.' AS comment
FROM sg_latest AS s
WHERE s.price_avg IS NOT NULL
  AND s.area_sq IS NOT NULL
GROUP BY s.report_month
HAVING SUM(s.area_sq) > 0
ON CONFLICT (indicator_type, region, obs_date) DO UPDATE
SET value      = EXCLUDED.value,
    source     = EXCLUDED.source,
    frequency  = EXCLUDED.frequency,
    unit       = EXCLUDED.unit,
    comment    = EXCLUDED.comment,
    updated_at = now();

-- ── 2a. new_contracts_volume / sverdl — Σ ДДУ (region_cd=66) ─────────────────────
-- Измеренный месячный объём ДДУ (contracted, type=apartments) с тем же dedup-по-снапшоту.
WITH sg_latest AS (
    SELECT DISTINCT ON (sg.obj_id, sg.report_month)
        sg.obj_id, sg.report_month, sg.contracted
      FROM domrf_kn_sale_graph AS sg
     WHERE sg.type = 'apartments'
     ORDER BY sg.obj_id, sg.report_month, sg.snapshot_date DESC
),
obj_region AS (
    SELECT DISTINCT ON (o.obj_id) o.obj_id, o.region_cd
      FROM domrf_kn_objects AS o
     ORDER BY o.obj_id, o.snapshot_date DESC NULLS LAST
)
INSERT INTO macro_indicator
    (indicator_type, region, obs_date, value, source, frequency, unit, comment)
SELECT
    'new_contracts_volume'                                  AS indicator_type,
    'sverdl'                                                AS region,
    s.report_month                                         AS obs_date,
    SUM(s.contracted)                                      AS value,
    'domrf'                                                AS source,
    'monthly'                                              AS frequency,
    'ДДУ/мес'                                              AS unit,
    'ТЗ §7.10: измеренный месячный объём ДДУ (SUM(contracted), type=apartments, '
    || 'latest snapshot per ЖК×месяц), region_cd=66. Источник domrf_kn_sale_graph (dom.рф).' AS comment
FROM sg_latest AS s
JOIN obj_region AS o ON o.obj_id = s.obj_id
WHERE o.region_cd = 66
  AND s.contracted IS NOT NULL
GROUP BY s.report_month
HAVING SUM(s.contracted) IS NOT NULL
ON CONFLICT (indicator_type, region, obs_date) DO UPDATE
SET value      = EXCLUDED.value,
    source     = EXCLUDED.source,
    frequency  = EXCLUDED.frequency,
    unit       = EXCLUDED.unit,
    comment    = EXCLUDED.comment,
    updated_at = now();

-- ── 2b. new_contracts_volume / rf — Σ ДДУ (все регионы) ─────────────────────────
WITH sg_latest AS (
    SELECT DISTINCT ON (sg.obj_id, sg.report_month)
        sg.obj_id, sg.report_month, sg.contracted
      FROM domrf_kn_sale_graph AS sg
     WHERE sg.type = 'apartments'
     ORDER BY sg.obj_id, sg.report_month, sg.snapshot_date DESC
)
INSERT INTO macro_indicator
    (indicator_type, region, obs_date, value, source, frequency, unit, comment)
SELECT
    'new_contracts_volume'                                  AS indicator_type,
    'rf'                                                    AS region,
    s.report_month                                         AS obs_date,
    SUM(s.contracted)                                      AS value,
    'domrf'                                                AS source,
    'monthly'                                              AS frequency,
    'ДДУ/мес'                                              AS unit,
    'ТЗ §7.10: измеренный месячный объём ДДУ (SUM(contracted), type=apartments, '
    || 'latest snapshot per ЖК×месяц), все регионы. Источник domrf_kn_sale_graph (dom.рф).' AS comment
FROM sg_latest AS s
WHERE s.contracted IS NOT NULL
GROUP BY s.report_month
HAVING SUM(s.contracted) IS NOT NULL
ON CONFLICT (indicator_type, region, obs_date) DO UPDATE
SET value      = EXCLUDED.value,
    source     = EXCLUDED.source,
    frequency  = EXCLUDED.frequency,
    unit       = EXCLUDED.unit,
    comment    = EXCLUDED.comment,
    updated_at = now();

-- ── 3a. escrow_objects_count / sverdl — кол-во объектов под эскроу (текущий срез) ──
-- ЕДИНСТВЕННАЯ текущая точка на регион: count объектов с escrow=TRUE в СВЕЖЕМ снапшоте.
--   obj_latest — свежий снапшот на obj_id (escrow-флаг + region_cd + snapshot_date).
--   obs_date = САМЫЙ свежий snapshot_date в наборе (текущий срез; НЕ месячный ряд, но
--   frequency='monthly' для единообразия macro_indicator — это одна актуальная обсервация).
--   HAVING MAX(snapshot_date) IS NOT NULL — guard: пустой набор → нет строки (obs_date NOT NULL
--   в PK иначе упал бы на NULL). На prod 457 sverdl-объектов есть, но guard явный/idempotent.
-- КАВЕАТ (в comment): escrow — БУЛЕВ ФЛАГ «продаётся через эскроу», а НЕ деньги. Денежные
--   остатки/наполнение эскроу-счетов dom.рф НЕ скрейпит → метрика = СЧЁТ ОБЪЕКТОВ.
WITH obj_latest AS (
    SELECT DISTINCT ON (o.obj_id)
        o.obj_id, o.region_cd, o.escrow, o.snapshot_date
      FROM domrf_kn_objects AS o
     ORDER BY o.obj_id, o.snapshot_date DESC NULLS LAST
)
INSERT INTO macro_indicator
    (indicator_type, region, obs_date, value, source, frequency, unit, comment)
SELECT
    'escrow_objects_count'                                  AS indicator_type,
    'sverdl'                                                AS region,
    MAX(snapshot_date)                                     AS obs_date,
    COUNT(*) FILTER (WHERE escrow IS TRUE)                 AS value,
    'domrf'                                                AS source,
    'monthly'                                              AS frequency,
    'объектов'                                             AS unit,
    'ТЗ §7.10: число объектов с escrow=TRUE в свежем снапшоте domrf_kn_objects '
    || '(текущий срез), region_cd=66. КАВЕАТ: dom.рф скрейпит ТОЛЬКО булев флаг '
    || 'эскроу, НЕ денежные остатки счетов — это СЧЁТ ОБЪЕКТОВ, не сумма денег.' AS comment
FROM obj_latest
WHERE region_cd = 66
HAVING MAX(snapshot_date) IS NOT NULL
ON CONFLICT (indicator_type, region, obs_date) DO UPDATE
SET value      = EXCLUDED.value,
    source     = EXCLUDED.source,
    frequency  = EXCLUDED.frequency,
    unit       = EXCLUDED.unit,
    comment    = EXCLUDED.comment,
    updated_at = now();

-- ── 3b. escrow_objects_count / rf — кол-во объектов под эскроу (все регионы) ─────
WITH obj_latest AS (
    SELECT DISTINCT ON (o.obj_id)
        o.obj_id, o.region_cd, o.escrow, o.snapshot_date
      FROM domrf_kn_objects AS o
     ORDER BY o.obj_id, o.snapshot_date DESC NULLS LAST
)
INSERT INTO macro_indicator
    (indicator_type, region, obs_date, value, source, frequency, unit, comment)
SELECT
    'escrow_objects_count'                                  AS indicator_type,
    'rf'                                                    AS region,
    MAX(snapshot_date)                                     AS obs_date,
    COUNT(*) FILTER (WHERE escrow IS TRUE)                 AS value,
    'domrf'                                                AS source,
    'monthly'                                              AS frequency,
    'объектов'                                             AS unit,
    'ТЗ §7.10: число объектов с escrow=TRUE в свежем снапшоте domrf_kn_objects '
    || '(текущий срез), все регионы. КАВЕАТ: dom.рф скрейпит ТОЛЬКО булев флаг '
    || 'эскроу, НЕ денежные остатки счетов — это СЧЁТ ОБЪЕКТОВ, не сумма денег.' AS comment
FROM obj_latest
HAVING MAX(snapshot_date) IS NOT NULL
ON CONFLICT (indicator_type, region, obs_date) DO UPDATE
SET value      = EXCLUDED.value,
    source     = EXCLUDED.source,
    frequency  = EXCLUDED.frequency,
    unit       = EXCLUDED.unit,
    comment    = EXCLUDED.comment,
    updated_at = now();

COMMIT;