feat(sitefinder): rosreestr cadastral-value loader + 0.7x denoise of quarter price index

Quarterly Rosreestr dataset_КАДАСТРСТОИМОСТЬ (cadastral value) gives a
per-cad-quarter cadastral ₽/m². Registered deals priced below 0.7x that
quarter's cadastral median are dropped from mv_quarter_price_per_m2 as
noise (data-entry junk / non-arm's-length deals that drag medians down).

- 178_schema_cadastral_value.sql: rosreestr_cadastral_value (+ staging),
  region-66-scoped, idempotent.
- 178b_load_cadastral_value.py: psycopg v3 staging COPY -> typed region-66
  INSERT, idempotent skip-if-loaded. Run MANUALLY by operator after deploy.
- 179_mv_quarter_price_cadastral_floor.sql: recreates BOTH MVs (CASCADE
  drops the dependent index MV). Adds cad_floor CTE + LEFT JOIN guard to
  mv_quarter_price_per_m2; mv_quarter_price_index recreated verbatim.

Safe to auto-apply on prod before any cadastral data exists: empty
rosreestr_cadastral_value -> cad_floor has 0 rows -> floor_ppm2 IS NULL for
every quarter -> predicate keeps ALL deals -> MV byte-identical to the
pre-floor (95) definition. Proven by a rolled-back dry-run (66:% EXCEPT
both directions = 0). With the experiment cad source loaded the floor
moves 480 of 1972 quarters (median 66788 -> 67350).

NN bumped from spec's 172/173 to 178/178b/179 (172-177 already on main).

data/sql/178_schema_cadastral_value.sql

@@ -0,0 +1,77 @@
+-- 178_schema_cadastral_value.sql
+-- Rosreestr open dataset: dataset_КАДАСТРСТОИМОСТЬ (cadastral value).
+-- Source URL pattern (region group 60–92, contains Свердловская обл. = 66):
+--   https://rosreestr.gov.ru/.../dataset_КАДАСТРСТОИМОСТЬ_r-r_60-92_y_{YYYY}_q_{Q}.csv
+-- Encoding: UTF-8. Separator: '~'. QUOTE '"'. NULL '' (empty string).
+-- Header columns (verified, in order):
+--   number ~ realestate_type_code ~ purpose_code ~ area ~ region_code ~ district ~
+--   city ~ quarter_cad_number ~ street ~ registration_quarter ~ application_quarter ~
+--   cost ~ cost_per_area ~ period
+--
+-- Purpose: cadastral ₽/m² floor per cadastral quarter, used to denoise the deals
+--   price index (mv_quarter_price_per_m2) — registered deals priced below
+--   0.7 × the cadastral median for their quarter are dropped as data noise.
+--   realestate_type_code='002001003000' (помещения/квартиры) — same code as
+--   rosreestr_deals; quarter_cad_number JOIN-compatible with rosreestr_deals.
+--   `number` = count of aggregated objects collapsed into that source row.
+--
+-- Scope: region-66-scoped — the .py loader (178b_load_cadastral_value.py) filters
+--   region_code='66' on INSERT. The 60–92 CSV is loaded once per quarter MANUALLY
+--   by an operator after deploy (this migration only creates the empty schema).
+--
+-- Dependencies: none. Auto-applied on prod via deploy.yml (_schema_migrations).
+-- Idempotent: CREATE TABLE/INDEX IF NOT EXISTS — safe to re-apply.
+
+BEGIN;
+
+-- Typed target table (region-66-scoped after loader filter).
+CREATE TABLE IF NOT EXISTS rosreestr_cadastral_value (
+    id                    BIGSERIAL    PRIMARY KEY,
+    source_quarter        VARCHAR(8)   NOT NULL,        -- '2026Q1' for traceability
+    -- Source columns (typed)
+    number                INTEGER,                      -- aggregated object count
+    realestate_type_code  TEXT,                         -- '002001003000' = помещения
+    purpose_code          TEXT,
+    area                  NUMERIC,                      -- м²
+    region_code           SMALLINT,                     -- 66 = Свердловская обл.
+    district              TEXT,
+    city                  TEXT,
+    quarter_cad_number    VARCHAR(30),                  -- '66:41:0506001' — JOIN-ключ
+    street                TEXT,
+    registration_quarter  TEXT,
+    application_quarter   TEXT,
+    cost                  NUMERIC,                      -- кадастровая стоимость, ₽
+    cost_per_area         NUMERIC,                      -- кадастровая ₽/m²
+    period                TEXT,
+    -- Audit
+    loaded_at             TIMESTAMPTZ  NOT NULL DEFAULT NOW()
+);
+
+-- Lookup index for the cad_floor CTE (filter by quarter + apartment type code).
+CREATE INDEX IF NOT EXISTS rosreestr_cadastral_value_cq_type_idx
+    ON rosreestr_cadastral_value (quarter_cad_number, realestate_type_code);
+
+-- Idempotent skip-if-loaded check in the loader is by source_quarter.
+CREATE INDEX IF NOT EXISTS rosreestr_cadastral_value_source_quarter_idx
+    ON rosreestr_cadastral_value (source_quarter);
+
+-- Staging: один общий буфер. Перед каждым кварталом TRUNCATE.
+-- 14 source columns, all TEXT, in CSV header order — typed cast happens on INSERT.
+CREATE UNLOGGED TABLE IF NOT EXISTS rosreestr_cadastral_value_staging (
+    number                TEXT,
+    realestate_type_code  TEXT,
+    purpose_code          TEXT,
+    area                  TEXT,
+    region_code           TEXT,
+    district              TEXT,
+    city                  TEXT,
+    quarter_cad_number    TEXT,
+    street                TEXT,
+    registration_quarter  TEXT,
+    application_quarter   TEXT,
+    cost                  TEXT,
+    cost_per_area         TEXT,
+    period                TEXT
+);
+
+COMMIT;

data/sql/178b_load_cadastral_value.py

@@ -0,0 +1,157 @@
+"""Загрузчик кварталов dataset_КАДАСТРСТОИМОСТЬ (кадастровая стоимость) в gendesign.
+
+Region-66-scoped: при INSERT фильтруем region_code='66' (Свердловская обл.), хотя
+исходный CSV покрывает регионы 60-92. Источник — открытый квартальный датасет
+Росреестра. Заливается через psycopg v3 (staging COPY → typed INSERT).
+
+Назначение: cost_per_area (кадастровая ₽/m²) даёт «пол» цены на квартал, по которому
+denoise-миграция 179 отбрасывает зашумлённые сделки rosreestr_deals дешевле
+0.7x кадастрового медианного ₽/m².
+
+Idempotent: пропускает source_quarter, уже залитый в rosreestr_cadastral_value.
+Запускается ВРУЧНУЮ оператором после деплоя; затем REFRESH двух MV (см. 179).
+
+Usage:
+    cd backend
+    uv run python ../data/sql/178b_load_cadastral_value.py 2026Q1
+    uv run python ../data/sql/178b_load_cadastral_value.py            # все из JOBS
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+import time
+from pathlib import Path
+
+import psycopg
+
+PG_DSN = os.environ.get(
+    "TUNNEL_DSN",
+    "postgresql://gendesign:2J2SBPMKuS998fiwhtQqDhMI@localhost:15432/gendesign",
+)
+
+RAW = Path(__file__).resolve().parent.parent / "raw"
+
+# Датасет всегда UTF-8, delimiter '~', header columns в порядке staging-таблицы.
+DELIMITER = "~"
+
+# (source_quarter, csv_filename)
+JOBS: list[tuple[str, str]] = [
+    ("2026Q1", "dataset_КАДАСТРСТОИМОСТЬ_r-r_60-92_y_2026_q_1.csv"),
+]
+
+
+# Region-66-scoped typed INSERT. Staging — всё TEXT, поэтому каждое числовое поле
+# проходит regex-guard + NULLIF перед ::cast (мусорные строки → NULL, не падаем).
+INSERT_SQL = """
+INSERT INTO rosreestr_cadastral_value (
+    source_quarter,
+    number, realestate_type_code, purpose_code, area, region_code,
+    district, city, quarter_cad_number, street,
+    registration_quarter, application_quarter, cost, cost_per_area, period
+)
+SELECT
+    %s,
+    CASE WHEN number ~ '^[0-9]+$' THEN number::INTEGER ELSE NULL END,
+    NULLIF(realestate_type_code, ''),
+    NULLIF(purpose_code, ''),
+    CASE WHEN area ~ '^[0-9]+(\\.[0-9]+)?$' AND area::NUMERIC > 0
+         THEN area::NUMERIC ELSE NULL END,
+    region_code::SMALLINT,
+    NULLIF(district, ''),
+    NULLIF(city, ''),
+    NULLIF(quarter_cad_number, ''),
+    NULLIF(street, ''),
+    NULLIF(registration_quarter, ''),
+    NULLIF(application_quarter, ''),
+    CASE WHEN cost ~ '^[0-9]+(\\.[0-9]+)?$' AND cost::NUMERIC > 0
+         THEN cost::NUMERIC ELSE NULL END,
+    CASE WHEN cost_per_area ~ '^[0-9]+(\\.[0-9]+)?$' AND cost_per_area::NUMERIC > 0
+         THEN cost_per_area::NUMERIC ELSE NULL END,
+    NULLIF(period, '')
+FROM rosreestr_cadastral_value_staging
+WHERE region_code = '66'
+  AND NULLIF(quarter_cad_number, '') IS NOT NULL
+"""
+
+
+def main() -> int:
+    targets = sys.argv[1:] or [j[0] for j in JOBS]
+    selected = [j for j in JOBS if j[0] in targets]
+    if not selected:
+        print(f"Нет matching кварталов в JOBS. Targets: {targets}")
+        return 1
+
+    conn = psycopg.connect(PG_DSN, connect_timeout=15)
+    try:
+        for sq, fname in selected:
+            with conn.cursor() as cur:
+                cur.execute(
+                    "SELECT COUNT(*) FROM rosreestr_cadastral_value "
+                    "WHERE source_quarter = %s",
+                    (sq,),
+                )
+                already = cur.fetchone()[0]
+            if already:
+                print(
+                    f"=== {sq} SKIP — уже {already} строк "
+                    f"(DELETE FROM rosreestr_cadastral_value "
+                    f"WHERE source_quarter='{sq}' чтобы перезалить)"
+                )
+                continue
+            csv_path = RAW / fname
+            if not csv_path.exists():
+                print(f"=== {sq} SKIP — нет файла {csv_path}")
+                continue
+            size_mb = csv_path.stat().st_size / 1e6
+            print(f"=== {sq}  ({fname}, sep='{DELIMITER}', {size_mb:.1f} MB) ===")
+            t0 = time.time()
+            with conn.cursor() as cur:
+                cur.execute("TRUNCATE rosreestr_cadastral_value_staging")
+                copy_sql = (
+                    f"COPY rosreestr_cadastral_value_staging FROM STDIN "
+                    f"(FORMAT csv, HEADER true, DELIMITER '{DELIMITER}', "
+                    f"QUOTE '\"', NULL '')"
+                )
+                with cur.copy(copy_sql) as copy:
+                    with open(csv_path, "rb") as f:
+                        while True:
+                            chunk = f.read(8 * 1024 * 1024)
+                            if not chunk:
+                                break
+                            copy.write(chunk)
+                t_copy = time.time() - t0
+                cur.execute(INSERT_SQL, (sq,))
+                inserted = cur.rowcount
+            conn.commit()
+            t_total = time.time() - t0
+            print(
+                f"  COPY {t_copy:.0f}s, INSERT (region 66) {inserted} rows, "
+                f"total {t_total:.0f}s"
+            )
+
+        # Final summary
+        with conn.cursor() as cur:
+            cur.execute(
+                """
+                SELECT source_quarter,
+                       COUNT(*) AS rows,
+                       COUNT(*) FILTER (
+                           WHERE realestate_type_code = '002001003000'
+                       ) AS apt_rows
+                  FROM rosreestr_cadastral_value
+                 GROUP BY source_quarter
+                 ORDER BY source_quarter
+                """
+            )
+            print("\n=== Итоги по rosreestr_cadastral_value (region 66) ===")
+            for sq, rows, apt_rows in cur.fetchall():
+                print(f"  {sq}  rows={rows:>9}  apartments={apt_rows:>9}")
+    finally:
+        conn.close()
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

data/sql/179_mv_quarter_price_cadastral_floor.sql

@@ -0,0 +1,223 @@
+-- 179_mv_quarter_price_cadastral_floor.sql
+-- Denoise the deals price index by a cadastral-value floor.
+--
+-- Chain (unchanged consumers):
+--   rosreestr_deals
+--     → mv_quarter_price_per_m2   (95_mv_quarter_price.sql)      ← MODIFIED here
+--     → mv_quarter_price_index    (99a_quarter_price_index.sql)  ← recreated as-is
+--     → tradein FDW quarter_price_index (reads mv_quarter_price_index by name)
+--
+-- What changes: mv_quarter_price_per_m2 gains a cadastral floor. Registered deals
+-- priced below 0.7 × the cadastral median ₽/m² for their cad-quarter are dropped
+-- as noise (data-entry junk / non-arm's-length deals drag medians down).
+-- Floor source: rosreestr_cadastral_value (schema 178, loaded MANUALLY by
+-- 178b_load_cadastral_value.py — region-66-scoped, apartment code 002001003000).
+--
+-- Why this file recreates BOTH MVs:
+--   DROP MATERIALIZED VIEW mv_quarter_price_per_m2 CASCADE also drops the dependent
+--   mv_quarter_price_index. We must recreate the index MV (verbatim from 99a) to
+--   restore the FDW contract that CASCADE removed.
+--
+-- *** CRITICAL SAFETY — no-op when rosreestr_cadastral_value is empty ***
+--   On prod this migration auto-applies (deploy.yml / _schema_migrations) BEFORE
+--   any cadastral data is loaded — the table exists (178) but is EMPTY. With zero
+--   rows, the cad_floor CTE produces zero rows → the LEFT JOIN yields
+--   floor_ppm2 IS NULL for EVERY quarter → the predicate
+--   `cf.floor_ppm2 IS NULL OR price_m2 >= cf.floor_ppm2` keeps ALL deals. The
+--   resulting mv_quarter_price_per_m2 is byte-identical to the pre-floor (95)
+--   definition. The floor only takes effect after an operator loads cadastral
+--   data and REFRESHes the MVs. Verified by a rolled-back dry-run (EXCEPT both
+--   directions = 0 rows when the cad source is empty).
+--
+-- Idempotent: DROP ... IF EXISTS ... CASCADE + recreate. Re-apply guarded by
+--   _schema_migrations in prod. First REFRESH is non-concurrent (empty MV).
+-- Dependencies: rosreestr_deals, rosreestr_cadastral_value (178).
+-- Output columns / unique index names / REFRESH / COMMENT preserved exactly
+--   (FDW + 99a depend on the column contract).
+
+BEGIN;
+
+-- ---------------------------------------------------------------------------
+-- 1. Drop per_m2 MV (CASCADE also drops mv_quarter_price_index — recreated below).
+-- ---------------------------------------------------------------------------
+DROP MATERIALIZED VIEW IF EXISTS mv_quarter_price_per_m2 CASCADE;
+
+-- ---------------------------------------------------------------------------
+-- 2. Recreate mv_quarter_price_per_m2 = body from 95_mv_quarter_price.sql,
+--    plus the cadastral floor (cad_floor CTE + LEFT JOIN in `filtered`).
+-- ---------------------------------------------------------------------------
+CREATE MATERIALIZED VIEW mv_quarter_price_per_m2 AS
+WITH cad_floor AS (
+    -- Cadastral median ₽/m² per quarter × 0.7 → noise floor.
+    -- Zero rows when rosreestr_cadastral_value is empty → floor disabled (no-op).
+    SELECT quarter_cad_number,
+           0.7 * percentile_cont(0.5) WITHIN GROUP (ORDER BY cost_per_area) AS floor_ppm2
+    FROM rosreestr_cadastral_value
+    WHERE realestate_type_code = '002001003000'
+      AND cost_per_area IS NOT NULL AND cost_per_area > 0
+    GROUP BY quarter_cad_number
+),
+dataset AS (
+    -- Resolve price_m2: use pre-computed field, fall back to deal_price / area
+    SELECT
+        quarter_cad_number,
+        CASE
+            WHEN price_per_sqm IS NOT NULL           THEN price_per_sqm
+            WHEN deal_price IS NOT NULL AND area > 0 THEN deal_price / area
+            ELSE NULL
+        END                 AS price_m2,
+        period_start_date   AS deal_date
+    FROM rosreestr_deals
+    WHERE realestate_type_code = '002001003000'
+      AND period_start_date >= NOW() - INTERVAL '24 months'
+),
+filtered AS (
+    -- Apply outlier filter once; reused by both aggregate CTEs.
+    -- Cadastral-floor denoise: drop deals below 0.7 × cadastral median ₽/m².
+    -- floor_ppm2 IS NULL (no cadastral row for the quarter) → keep the deal.
+    SELECT quarter_cad_number, price_m2, deal_date
+    FROM dataset
+    LEFT JOIN cad_floor cf USING (quarter_cad_number)
+    WHERE price_m2 BETWEEN 30000 AND 800000
+      AND (cf.floor_ppm2 IS NULL OR price_m2 >= cf.floor_ppm2)
+),
+agg_main AS (
+    -- Primary aggregation: full 24-month window
+    SELECT
+        quarter_cad_number,
+        COUNT(*)                                                AS deals_count,
+        PERCENTILE_CONT(0.25) WITHIN GROUP (ORDER BY price_m2) AS p25,
+        PERCENTILE_CONT(0.50) WITHIN GROUP (ORDER BY price_m2) AS median,
+        PERCENTILE_CONT(0.75) WITHIN GROUP (ORDER BY price_m2) AS p75,
+        AVG(price_m2)::numeric(12, 2)                          AS mean,
+        MAX(deal_date)                                         AS last_deal_date
+    FROM filtered
+    GROUP BY quarter_cad_number
+    HAVING COUNT(*) >= 3
+),
+agg_windows AS (
+    -- Sub-window medians per quarter.
+    -- PERCENTILE_CONT does not accept a FILTER clause in PG16, so we pre-filter
+    -- per window inside separate conditional aggregations using a CASE that
+    -- returns NULL for out-of-window rows (NULLs are excluded from PERCENTILE_CONT).
+    SELECT
+        quarter_cad_number,
+        PERCENTILE_CONT(0.50) WITHIN GROUP (
+            ORDER BY CASE WHEN deal_date > NOW() - INTERVAL '6 months'
+                          THEN price_m2 ELSE NULL END
+        )                                                       AS median_6m,
+        PERCENTILE_CONT(0.50) WITHIN GROUP (
+            ORDER BY CASE WHEN deal_date > NOW() - INTERVAL '12 months'
+                          THEN price_m2 ELSE NULL END
+        )                                                       AS median_12m
+    FROM filtered
+    GROUP BY quarter_cad_number
+)
+SELECT
+    m.quarter_cad_number,
+    m.deals_count,
+    m.p25,
+    m.median,
+    m.p75,
+    m.mean,
+    m.last_deal_date,
+    w.median_6m,
+    w.median_12m,
+    -- median_24m == median (window is already 24mo); kept for API symmetry
+    m.median                                                    AS median_24m
+FROM agg_main m
+JOIN agg_windows w USING (quarter_cad_number)
+WITH NO DATA;
+
+-- UNIQUE index required for REFRESH MATERIALIZED VIEW CONCURRENTLY.
+-- Created on empty MV → instant (no data lock).
+CREATE UNIQUE INDEX mv_quarter_price_pk
+    ON mv_quarter_price_per_m2 (quarter_cad_number);
+
+-- Initial populate (non-concurrent — CONCURRENTLY not allowed on empty MV).
+REFRESH MATERIALIZED VIEW mv_quarter_price_per_m2;
+
+COMMENT ON MATERIALIZED VIEW mv_quarter_price_per_m2 IS
+    'Per-cad_quarter (quarter_cad_number) price aggregation из rosreestr_deals. '
+    'Filter: realestate_type_code=002001003000 (новостройки/ДДУ), rolling 24 months, '
+    'outlier filter 30K–800K руб/м², HAVING >= 3 deals. '
+    'All-Russia scope (~52K rows expected). '
+    'Refresh: weekly via celery beat (TBD, Issue #33 PR C).';
+
+-- ---------------------------------------------------------------------------
+-- 3. Recreate mv_quarter_price_index = body from 99a_quarter_price_index.sql,
+--    UNCHANGED (restores what the CASCADE above dropped).
+-- ---------------------------------------------------------------------------
+DROP MATERIALIZED VIEW IF EXISTS mv_quarter_price_index CASCADE;
+
+CREATE MATERIALIZED VIEW mv_quarter_price_index AS
+WITH
+-- All 66:% quarters from source MV (Sverdlovsk oblast scope)
+all_66_quarters AS (
+    SELECT
+        quarter_cad_number,
+        deals_count,
+        median,
+        substring(quarter_cad_number FOR 5) AS district_prefix
+    FROM mv_quarter_price_per_m2
+    WHERE quarter_cad_number LIKE '66:%'
+),
+
+-- City normalizer: equal-weight median-of-quarter-medians for EKB (66:41:%)
+-- Uses only quarters with deals_count >= 5 to suppress noise.
+city_median_cte AS (
+    SELECT
+        percentile_cont(0.5) WITHIN GROUP (ORDER BY median) AS city_median
+    FROM all_66_quarters
+    WHERE quarter_cad_number LIKE '66:41:%'
+      AND deals_count >= 5
+),
+
+-- District fallback: median of quarter medians per district prefix,
+-- also restricted to deals_count >= 5 quarters.
+district_medians AS (
+    SELECT
+        district_prefix,
+        percentile_cont(0.5) WITHIN GROUP (ORDER BY median) AS district_median
+    FROM all_66_quarters
+    WHERE deals_count >= 5
+    GROUP BY district_prefix
+)
+
+SELECT
+    q.quarter_cad_number,
+    CASE
+        WHEN q.deals_count >= 5                    THEN q.median / c.city_median
+        WHEN dm.district_median IS NOT NULL        THEN dm.district_median / c.city_median
+        ELSE 1.0
+    END::double precision                                       AS price_index,
+    q.deals_count                                               AS n_deals,
+    CASE
+        WHEN q.deals_count >= 5                    THEN 'quarter'
+        WHEN dm.district_median IS NOT NULL        THEN 'district'
+        ELSE 'city_fallback'
+    END::text                                                   AS basis,
+    NOW()                                                       AS computed_at
+FROM all_66_quarters q
+CROSS JOIN city_median_cte c
+LEFT JOIN district_medians dm ON dm.district_prefix = q.district_prefix
+WITH NO DATA;
+
+-- UNIQUE index required for REFRESH MATERIALIZED VIEW CONCURRENTLY.
+-- Also serves as O(1) PK lookup for the estimator (#647-3).
+CREATE UNIQUE INDEX IF NOT EXISTS mv_quarter_price_index_uq
+    ON mv_quarter_price_index (quarter_cad_number);
+
+-- Initial populate (non-concurrent — MV is empty, CONCURRENTLY not allowed on empty MV).
+REFRESH MATERIALIZED VIEW mv_quarter_price_index;
+
+COMMENT ON MATERIALIZED VIEW mv_quarter_price_index IS
+    'Per-cadastral-quarter price index normalized to EKB city median (=1.0). '
+    'Source: mv_quarter_price_per_m2. Scope: all 66:% (Sverdlovsk oblast) quarters. '
+    'city_median = percentile_cont(0.5) over 66:41:% quarters with deals_count>=5. '
+    'basis: quarter (>=5 deals) | district (sparse, district fallback) | city_fallback. '
+    'Refresh: CONCURRENTLY safe (UNIQUE index on quarter_cad_number). '
+    'Consumer: estimator service (#647-3) reads O(1) by quarter_cad_number. '
+    'Issue: #760.';
+
+COMMIT;