-- 080_asking_to_sold_ratios.sql
-- #648 Stage 2 — persist the per-rooms asking→sold correction ratio.
--
-- ПРОБЛЕМА (#648). Оценщик (estimator.py) предсказывает цену квартиры по МЕДИАНЕ
--   *активных asking* цен аналогов (Tukey 1.5×IQR → median ₽/m²). Backtest-харнес
--   (scripts/backtest_estimator.py, Stage 1, merged) показал, что asking систематически
--   на +20-30% выше реально зарегистрированной цены ДКП (deals, source='rosreestr').
--   Per-rooms множитель ratio = median(SOLD ppm²) / median(ASKING ppm²) (≈0.73-0.85,
--   глобально ≈0.81 — sold ≈ 81% от asking) убирает этот перекос: out-of-sample bias
--   на held-out ДКП ушёл с +20.4% до −4.0%. Stage 2 = СОХРАНИТЬ этот ratio, чтобы
--   Stage 3 (estimator) читал его и домножал asking-median прогноз на sold/asking.
--
-- СОЗДАЁТ:
--   asking_to_sold_ratios — таблица коэффициентов (rooms_bucket, district) → ratio.
--     rooms_bucket: 0=студия,1,2,3,4(=4+); СПЕЦ-строка rooms_bucket=-1 = global fallback.
--     district: ЗАРЕЗЕРВИРОВАНО для #647 (geo-разбивка); в #648 ВСЕГДА '' (часть PK,
--       поэтому NOT NULL DEFAULT '' — '' можно положить в PK, NULL нельзя).
--
-- МЕТОДОЛОГИЯ (повторяет intent харнеса, см. _derive_room_ratios / _percentile):
--   sold_median = percentile_cont(0.5) по deals.price_per_m2, source='rosreestr',
--     ppm² ∈ [30000, 600000], deal_date >= CURRENT_DATE − 12 months, бакет по
--     LEAST(GREATEST(rooms,0),4). (percentile_cont = линейно-интерполированная медиана,
--     соответствует _percentile(sorted, 0.5) в харнесе.)
--   ask_median  = percentile_cont(0.5) по listings.price_per_m2, is_active,
--     ppm² ∈ [30000, 600000], тот же бакет.
--   ratio = sold_median / ask_median.  window_months = 12.
--
--   MIN-SAMPLE (вариант (a) — skip): бакет с n_deals < 30 ИЛИ n_listings < 30 НЕ получает
--     собственную per_rooms-строку. Estimator при чтении делает fallback: строки бакета
--     нет → берёт строку rooms_bucket = -1 (global). Так таблица не overfit'ит тонкие
--     бакеты (студия/3к/4+ обычно тонкие; 1к/2к — плотные). Глобальная -1 строка
--     пишется ВСЕГДА (basis='global_fallback'); per_rooms-строки — только для бакетов,
--     прошедших порог 30/30 (basis='per_rooms').
--   GUARD: бакет с ask_median 0/NULL пропускается (деление невозможно) — это и так
--     отсекается порогом n_listings>=30, но HAVING ask_median IS NOT NULL AND >0 страхует.
--
-- CAVEATS (важно — те же, что в шапке харнеса):
--   (a) TIME-MISMATCH: ask_median считается по ТЕКУЩИМ активным listings, а sold_median —
--       по сделкам за ПРОШЕДШИЕ 12 мес. Это не point-in-time: рынок дрейфует, ratio слегка
--       смещён. Честный point-in-time потребует listing_source_snapshots (#570), когда
--       накопится история asking «на дату сделки».
--   (b) ДКП = ЗАРЕГИСТРИРОВАННАЯ цена (что стороны заявили в Росреестр); может расходиться
--       с реальной (налоговая оптимизация, сделки со связанными лицами).
--   (c) PER-ROOMS ONLY: district зарезервирован под #647 (geo-разбивка), пока всегда ''.
--
-- STAGE-4 REUSE: блок derivation ниже (CTE deal_side / ask_side / per_bucket / global_row +
--   финальный INSERT ... ON CONFLICT DO UPDATE) — это ровно та логика, которую переиспользует
--   refresh-задача `asking_to_sold_ratio_refresh` (Stage 4): она прогоняет тот же
--   INSERT...SELECT по расписанию, чтобы ratio следовал за рынком. ON CONFLICT DO UPDATE
--   делает повторный прогон идемпотентным (refresh = re-seed). При вынесении в app-код
--   bind-параметры (порог, окно, ppm²-полоса) подаются через CAST(:x AS type) (psycopg v3).
--
-- ЗАВИСИМОСТИ: deals, listings (002_core_tables.sql). Применяется под gate _schema_migrations
--   (#637): ровно один раз по NN-порядку. Idempotent (CREATE TABLE IF NOT EXISTS,
--   ON CONFLICT DO UPDATE) — безопасно ре-apply / dry-run (BEGIN/ROLLBACK).
-- Apply after: 079_listing_source_history.sql

BEGIN;

-- ── asking_to_sold_ratios — коэффициенты asking→sold на (бакет комнат × район) ──
-- district NOT NULL DEFAULT '' — часть PK; #647 заполнит район, #648 всегда ''.
-- sold_median/ask_median nullable — диагностика; ratio NOT NULL (строку без ratio не пишем).
CREATE TABLE IF NOT EXISTS asking_to_sold_ratios (
    rooms_bucket  int          NOT NULL,             -- 0=студия..4=4+; -1 = global fallback row
    district      text         NOT NULL DEFAULT '',  -- RESERVED for #647 (always '' in #648)
    ratio         numeric      NOT NULL,             -- sold_median_ppm2 / ask_median_ppm2
    sold_median   bigint,                            -- median(deals.price_per_m2), диагностика
    ask_median    bigint,                            -- median(listings.price_per_m2), диагностика
    n_deals       int          NOT NULL,             -- сделок в окне/бакете
    n_listings    int          NOT NULL,             -- активных listings в бакете
    window_months int          NOT NULL,             -- окно сделок (12)
    basis         text         NOT NULL,             -- 'per_rooms' | 'global_fallback'
    computed_at   timestamptz  NOT NULL DEFAULT now(),
    PRIMARY KEY (rooms_bucket, district)
);

COMMENT ON TABLE asking_to_sold_ratios IS
    'Per-rooms asking→sold коэффициент (#648): ratio = median(SOLD ppm²)/median(ASKING ppm²). '
    'rooms_bucket 0=студия..4=4+; -1 = global fallback (basis=global_fallback, пишется всегда). '
    'Per-rooms строки только при n_deals>=30 AND n_listings>=30 (иначе estimator читает -1). '
    'district зарезервирован под #647 (geo), в #648 всегда ''''. '
    'Caveat: ask=ТЕКУЩИЕ listings vs sold=сделки за 12 мес (не point-in-time); ДКП=registered. '
    'Refresh — Stage 4 asking_to_sold_ratio_refresh переиспользует derivation ниже.';

-- ── Derivation + seed ─────────────────────────────────────────────────────────
-- Вынесено как один INSERT...SELECT с CTE-«сторонами» (deal_side / ask_side), чтобы
-- Stage 4 (asking_to_sold_ratio_refresh) переиспользовал ровно эту логику. Окно сделок
-- = трейлинг 12 мес; listings — текущие активные. ppm²-полоса [30000,600000] и бакет
-- LEAST(GREATEST(rooms,0),4) — байт-в-байт как в харнесе (PPM2_MIN/PPM2_MAX, _bucketize_rooms).
WITH
-- SOLD медианы по бакетам комнат за трейлинг-12мес (ДКП Росреестра).
deal_side AS (
    SELECT
        LEAST(GREATEST(rooms, 0), 4)                                  AS rooms_bucket,
        percentile_cont(0.5) WITHIN GROUP (ORDER BY price_per_m2)     AS sold_median,
        COUNT(*)                                                      AS n_deals
    FROM deals
    WHERE source = 'rosreestr'
      AND rooms IS NOT NULL
      AND price_per_m2 BETWEEN 30000 AND 600000
      AND deal_date >= CURRENT_DATE - INTERVAL '12 months'
    GROUP BY LEAST(GREATEST(rooms, 0), 4)
),
-- ASKING медианы по бакетам комнат среди ТЕКУЩИХ активных объявлений.
ask_side AS (
    SELECT
        LEAST(GREATEST(rooms, 0), 4)                                  AS rooms_bucket,
        percentile_cont(0.5) WITHIN GROUP (ORDER BY price_per_m2)     AS ask_median,
        COUNT(*)                                                      AS n_listings
    FROM listings
    WHERE is_active
      AND rooms IS NOT NULL
      AND price_per_m2 BETWEEN 30000 AND 600000
      -- novostroyki guard (#1364/#1186): NULL = legacy вторичка до м.011.
      -- Secondary-only asking — sold-сторона ДКП вторичка; новостройки исказили бы ratio.
      AND (listing_segment IS NULL OR listing_segment = 'vtorichka')
    GROUP BY LEAST(GREATEST(rooms, 0), 4)
),
-- Per-rooms строки: только бакеты с обеими сторонами, прошедшие порог 30/30 и ask>0.
-- Тонкие бакеты (n<30) сюда НЕ попадают → estimator делает fallback на -1.
per_bucket AS (
    SELECT
        d.rooms_bucket,
        ''::text                                  AS district,
        (d.sold_median / a.ask_median)::numeric   AS ratio,
        round(d.sold_median)::bigint              AS sold_median,
        round(a.ask_median)::bigint               AS ask_median,
        d.n_deals::int                            AS n_deals,
        a.n_listings::int                         AS n_listings,
        12                                        AS window_months,
        'per_rooms'::text                         AS basis
    FROM deal_side d
    JOIN ask_side a USING (rooms_bucket)
    WHERE d.n_deals >= 30
      AND a.n_listings >= 30
      AND a.ask_median IS NOT NULL
      AND a.ask_median > 0
      AND d.sold_median IS NOT NULL          -- divide-safety (n_deals>=30 это и так гарантирует)
      AND d.sold_median > 0
),
-- SOLD медиана по ВСЕМ комнатам (без бакет-фильтра) за трейлинг-12мес — для global row.
deal_global AS (
    SELECT
        percentile_cont(0.5) WITHIN GROUP (ORDER BY price_per_m2) AS sold_median,
        COUNT(*)                                                  AS n_deals
    FROM deals
    WHERE source = 'rosreestr'
      AND rooms IS NOT NULL
      AND price_per_m2 BETWEEN 30000 AND 600000
      AND deal_date >= CURRENT_DATE - INTERVAL '12 months'
),
-- ASKING медиана по ВСЕМ активным listings (без бакет-фильтра) — для global row.
ask_global AS (
    SELECT
        percentile_cont(0.5) WITHIN GROUP (ORDER BY price_per_m2) AS ask_median,
        COUNT(*)                                                  AS n_listings
    FROM listings
    WHERE is_active
      AND rooms IS NOT NULL
      AND price_per_m2 BETWEEN 30000 AND 600000
      -- novostroyki guard (#1364/#1186): NULL = legacy вторичка до м.011.
      -- Secondary-only asking — sold-сторона ДКП вторичка; новостройки исказили бы ratio.
      AND (listing_segment IS NULL OR listing_segment = 'vtorichka')
),
-- Global fallback строка rooms_bucket=-1 (пишется всегда, если ask>0).
global_row AS (
    SELECT
        -1                                        AS rooms_bucket,
        ''::text                                  AS district,
        (d.sold_median / a.ask_median)::numeric   AS ratio,
        round(d.sold_median)::bigint              AS sold_median,
        round(a.ask_median)::bigint               AS ask_median,
        d.n_deals::int                            AS n_deals,
        a.n_listings::int                         AS n_listings,
        12                                        AS window_months,
        'global_fallback'::text                   AS basis
    FROM deal_global d
    CROSS JOIN ask_global a
    WHERE a.ask_median IS NOT NULL
      AND a.ask_median > 0
      AND d.sold_median IS NOT NULL          -- защита от пустого окна сделок (иначе ratio=NULL)
      AND d.sold_median > 0
)
INSERT INTO asking_to_sold_ratios (
    rooms_bucket, district, ratio, sold_median, ask_median,
    n_deals, n_listings, window_months, basis
)
SELECT rooms_bucket, district, ratio, sold_median, ask_median,
       n_deals, n_listings, window_months, basis FROM global_row
UNION ALL
SELECT rooms_bucket, district, ratio, sold_median, ask_median,
       n_deals, n_listings, window_months, basis FROM per_bucket
ON CONFLICT (rooms_bucket, district) DO UPDATE SET
    ratio         = EXCLUDED.ratio,
    sold_median   = EXCLUDED.sold_median,
    ask_median    = EXCLUDED.ask_median,
    n_deals       = EXCLUDED.n_deals,
    n_listings    = EXCLUDED.n_listings,
    window_months = EXCLUDED.window_months,
    basis         = EXCLUDED.basis,
    computed_at   = now();

COMMIT;