From 4d063d682772bf3fdb7acbb86398776fefcef424 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Sun, 28 Jun 2026 19:45:35 +0300 Subject: [PATCH] feat(sitefinder): rosreestr cadastral-value loader + 0.7x denoise of quarter price index MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 77 ++++++ data/sql/178b_load_cadastral_value.py | 157 ++++++++++++ .../179_mv_quarter_price_cadastral_floor.sql | 223 ++++++++++++++++++ 3 files changed, 457 insertions(+) create mode 100644 data/sql/178_schema_cadastral_value.sql create mode 100644 data/sql/178b_load_cadastral_value.py create mode 100644 data/sql/179_mv_quarter_price_cadastral_floor.sql diff --git a/data/sql/178_schema_cadastral_value.sql b/data/sql/178_schema_cadastral_value.sql new file mode 100644 index 00000000..1d202358 --- /dev/null +++ b/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; diff --git a/data/sql/178b_load_cadastral_value.py b/data/sql/178b_load_cadastral_value.py new file mode 100644 index 00000000..7a3e6ccf --- /dev/null +++ b/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()) diff --git a/data/sql/179_mv_quarter_price_cadastral_floor.sql b/data/sql/179_mv_quarter_price_cadastral_floor.sql new file mode 100644 index 00000000..12c689da --- /dev/null +++ b/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; -- 2.45.3