feat(objective): diff-gated objective_lots_history writer trigger (#307 OBJ-6)

data/sql/164_objective_lots_history_writer.sql

@@ -0,0 +1,170 @@
+-- 164_objective_lots_history_writer.sql
+-- #307 OBJ-6 — diff-gated objective_lots_history writer (DB-level trigger).
+--
+-- ПРОБЛЕМА (ground-truth, измерено на prod 2026-06-17):
+--   objective_lots_history содержит 974 071 строк, но это НЕ delta-tracking, а wholesale-дамп:
+--     snapshot_date | rows
+--       2026-05-17  | 324 217
+--       2026-05-19  | 324 510
+--       2026-06-03  | 325 344
+--   = 3 даты × ~324k = по строке на КАЖДЫЙ лот в КАЖДЫЙ sync, вне зависимости от того, менялось
+--   ли что-то. Источник: data/sql/70_parse_objective_raw.py пишет напрямую
+--     INSERT INTO objective_lots_history (...) ... ON CONFLICT (objective_lot_id, snapshot_date)
+--       DO NOTHING
+--   для каждого лота при каждом sync (Module_Objective_Parser). Никакого diff нет — поэтому
+--   таблица раздувается, а реальные изменения (price delta, free→reserved→sold, экспозиция)
+--   утопают в шуме одинаковых снимков. Аудит SiteFinder_V2_Groundtruth_Audit_0602 уже отметил
+--   «снапшотов всего 2-3 → динамика остатков data-starved».
+--
+-- ЦЕЛЬ OBJ-6: каждый sync лота фиксирует в history строку ТОЛЬКО когда реально изменилось
+--   tracked-поле (цена за м², суммарные цены, status, is_sold) относительно ПОСЛЕДНЕГО снимка
+--   этого лота. Это даёт чистую дельта-историю: price delta, переход статуса
+--   free→reserved→sold, длительность экспозиции — без раздувания одинаковыми строками.
+--
+-- ПОЧЕМУ ТРИГГЕР (а не правка python-sync):
+--   • Параллельный воркер сейчас редактирует objective-mapper (70_parse_objective_raw.py) →
+--     избегаем file-overlap, не трогаем python.
+--   • BEFORE/AFTER ROW-триггер на objective_lots ловит ВСЕ пути записи разом: основной
+--     ON CONFLICT (objective_lot_id) DO UPDATE python-парсера (UPDATE-путь), INSERT новых лотов,
+--     ручной SQL, будущие загрузчики. Один инвариант в БД вместо логики в каждом писателе.
+--   • Diff-gate в БД дёшев: один индексный lookup последнего снимка по
+--     objective_lots_history_lot_idx (objective_lot_id, snapshot_date DESC) — LIMIT 1.
+--
+-- ТРИГГЕР: AFTER INSERT OR UPDATE FOR EACH ROW на objective_lots.
+--   AFTER (не BEFORE): пишем в ДРУГУЮ таблицу (history), строка objective_lots уже
+--   зафиксирована — побочная вставка, не мутация NEW. FOR EACH ROW: нужен per-lot diff.
+--
+-- TRACKED-ПОЛЯ (diff-gate) — ровно «экономика лота», ради которой ведётся история:
+--   price_per_m2_rub, price_calculated_total_rub, price_offer_total_rub, status, is_sold.
+--   Сравнение через IS DISTINCT FROM — корректно трактует NULL (NULL vs NULL = НЕ изменилось,
+--   value vs NULL = изменилось). Если последнего снимка НЕТ (новый лот) → пишем всегда.
+--
+-- price_delta_pct в history: вычисляем snapshot-over-snapshot
+--   (NEW.price_per_m2_rub / prev.price_per_m2_rub − 1) × 100, округление до 2 знаков.
+--   Это ОТЛИЧАЕТСЯ от objective_lots.price_delta_pct (внутренняя «Дельта цена, %» Объектива
+--   из исходного отчёта) — здесь нужна именно дельта между снимками для velocity-аналитики.
+--   Если prev отсутствует или prev.price_per_m2_rub = 0/NULL → price_delta_pct = NULL.
+--
+-- КОЛОНКИ history (точно по схеме objective_lots_history, ничего не выдумываем):
+--   objective_lot_id, snapshot_date, status, is_sold, price_calculated_total_rub,
+--   price_per_m2_rub, price_offer_total_rub, price_delta_pct, area_pd, contract_date,
+--   registration_date, contract_type, bank_name, raw_id, complex_id
+--   (id/fetched_at — defaults: nextval / now()).
+--
+-- ON CONFLICT (objective_lot_id, snapshot_date) DO UPDATE — НЕ DO NOTHING:
+--   если за тот же день уже есть снимок (в т.ч. от python-парсера) и пришло НОВОЕ значение
+--   tracked-поля, обновляем снимок этого дня свежими цифрами + пересчитанным delta. Это чинит
+--   баг текущего python-писателя (DO NOTHING терял внутридневное изменение). Дубликат-строка
+--   при коллизии не создаётся (защищён UNIQUE-констрейнтом).
+--
+-- СОСУЩЕСТВОВАНИЕ С PYTHON-ПИСАТЕЛЕМ: пока 70_parse_objective_raw.py продолжает свой прямой
+--   INSERT ... DO NOTHING, триггер безопасен — его собственный INSERT уходит в DO UPDATE при
+--   коллизии snapshot_date, дубль не создаётся. Чистый эффект: со следующего sync новые/
+--   изменённые лоты получают корректную diff-строку, а wholesale-INSERT парсера для
+--   НЕизменившихся лотов перекрывается DO NOTHING (его строка остаётся, но триггер её не
+--   плодит). FOLLOW-UP (backend-задача, отдельный PR): убрать прямой history-INSERT из
+--   70_parse_objective_raw.py, оставив запись только за триггером — тогда history станет
+--   полностью diff-gated. Эта миграция самодостаточна и безопасна без той правки.
+--
+-- BACKFILL не делаем: существующие 974k строк — исторический wholesale-дамп, не трогаем
+--   (никаких DROP/TRUNCATE/DELETE). Триггер влияет только на будущие записи.
+--
+-- ПРОИЗВОДИТЕЛЬНОСТЬ на bulk-sync (~1.1M upsert): на строку — один индексный lookup
+--   последнего снимка (Index Scan по objective_lots_history_lot_idx, LIMIT 1) + диф 5 полей.
+--   Для НЕизменившихся лотов (типичный sync — меняется меньшинство) — ранний выход без INSERT.
+--   Пренебрежимо относительно стоимости самого upsert в objective_lots.
+--
+-- Idempotent: CREATE OR REPLACE FUNCTION + DROP TRIGGER IF EXISTS перед CREATE TRIGGER
+--   (PG16 не имеет CREATE OR REPLACE TRIGGER — паттерн как в 129_cad_parcels_quarter_cad_trigger).
+--   BEGIN/COMMIT — атомарно. Re-apply чист. Auto-applied deploy.yml через _schema_migrations.
+--
+-- Deploy order: чистая СХЕМА, без зависимости от кода — безопасно катить в любой момент.
+
+BEGIN;
+
+-- ── 1. Триггер-функция: diff-gated запись снимка лота в objective_lots_history ──────────
+CREATE OR REPLACE FUNCTION objective_lots_write_history() RETURNS TRIGGER AS $$
+DECLARE
+    prev objective_lots_history%ROWTYPE;
+    has_prev boolean := false;
+    delta_pct numeric := NULL;
+BEGIN
+    -- Последний снимок этого лота (любой более ранней или той же даты) — для diff и delta.
+    -- Index Scan по objective_lots_history_lot_idx (objective_lot_id, snapshot_date DESC).
+    SELECT * INTO prev
+    FROM objective_lots_history
+    WHERE objective_lot_id = NEW.objective_lot_id
+      AND snapshot_date <= NEW.snapshot_date
+    ORDER BY snapshot_date DESC
+    LIMIT 1;
+    has_prev := FOUND;
+
+    -- Diff-gate: пишем только если снимка ещё нет ИЛИ изменилось хоть одно tracked-поле.
+    -- IS DISTINCT FROM — NULL-safe (NULL vs NULL = не изменилось).
+    IF has_prev
+       AND prev.price_per_m2_rub            IS NOT DISTINCT FROM NEW.price_per_m2_rub
+       AND prev.price_calculated_total_rub  IS NOT DISTINCT FROM NEW.price_calculated_total_rub
+       AND prev.price_offer_total_rub       IS NOT DISTINCT FROM NEW.price_offer_total_rub
+       AND prev.status                      IS NOT DISTINCT FROM NEW.status
+       AND prev.is_sold                     IS NOT DISTINCT FROM NEW.is_sold
+    THEN
+        -- Ничего из tracked не поменялось → не плодим снимок.
+        RETURN NULL;  -- AFTER-триггер: возвращаемое значение игнорируется, RETURN NULL = no-op.
+    END IF;
+
+    -- snapshot-over-snapshot дельта цены за м² (для velocity-аналитики).
+    IF has_prev
+       AND prev.price_per_m2_rub IS NOT NULL
+       AND prev.price_per_m2_rub <> 0
+       AND NEW.price_per_m2_rub IS NOT NULL
+    THEN
+        delta_pct := round((NEW.price_per_m2_rub / prev.price_per_m2_rub - 1) * 100, 2);
+    END IF;
+
+    INSERT INTO objective_lots_history (
+        objective_lot_id, snapshot_date, status, is_sold,
+        price_calculated_total_rub, price_per_m2_rub, price_offer_total_rub,
+        price_delta_pct, area_pd, contract_date, registration_date,
+        contract_type, bank_name, raw_id, complex_id
+    ) VALUES (
+        NEW.objective_lot_id, NEW.snapshot_date, NEW.status, NEW.is_sold,
+        NEW.price_calculated_total_rub, NEW.price_per_m2_rub, NEW.price_offer_total_rub,
+        delta_pct, NEW.area_pd, NEW.contract_date, NEW.registration_date,
+        NEW.contract_type, NEW.bank_name, NEW.raw_id, NEW.complex_id
+    )
+    ON CONFLICT (objective_lot_id, snapshot_date) DO UPDATE SET
+        status                     = EXCLUDED.status,
+        is_sold                    = EXCLUDED.is_sold,
+        price_calculated_total_rub = EXCLUDED.price_calculated_total_rub,
+        price_per_m2_rub           = EXCLUDED.price_per_m2_rub,
+        price_offer_total_rub      = EXCLUDED.price_offer_total_rub,
+        price_delta_pct            = EXCLUDED.price_delta_pct,
+        area_pd                    = EXCLUDED.area_pd,
+        contract_date              = EXCLUDED.contract_date,
+        registration_date          = EXCLUDED.registration_date,
+        contract_type              = EXCLUDED.contract_type,
+        bank_name                  = EXCLUDED.bank_name,
+        raw_id                     = EXCLUDED.raw_id,
+        complex_id                 = EXCLUDED.complex_id,
+        fetched_at                 = now();
+
+    RETURN NULL;  -- AFTER ROW trigger — результат не используется.
+END;
+$$ LANGUAGE plpgsql;
+
+COMMENT ON FUNCTION objective_lots_write_history() IS
+    'AFTER INSERT OR UPDATE-триггер objective_lots (#307 OBJ-6): diff-gated запись снимка в '
+    'objective_lots_history. Пишет строку ТОЛЬКО при изменении tracked-полей (price_per_m2_rub, '
+    'price_calculated_total_rub, price_offer_total_rub, status, is_sold) против последнего '
+    'снимка лота (IS DISTINCT FROM, NULL-safe). Считает snapshot-over-snapshot price_delta_pct. '
+    'ON CONFLICT(objective_lot_id,snapshot_date) DO UPDATE — внутридневное изменение обновляет '
+    'снимок дня, дубль не создаётся. Заменяет wholesale-дамп python-парсера на дельта-историю.';
+
+-- ── 2. Триггер AFTER INSERT OR UPDATE на objective_lots ─────────────────────────────────
+-- PG16 не имеет CREATE OR REPLACE TRIGGER → DROP IF EXISTS + CREATE (idempotent, паттерн 129).
+DROP TRIGGER IF EXISTS objective_lots_write_history ON objective_lots;
+CREATE TRIGGER objective_lots_write_history
+    AFTER INSERT OR UPDATE ON objective_lots
+    FOR EACH ROW EXECUTE FUNCTION objective_lots_write_history();
+
+COMMIT;