bot-backend
90332e9827
All checks were successful
Deploy Trade-In / changes (push) Successful in 9s
Deploy Trade-In / build-frontend (push) Has been skipped
Deploy Trade-In / build-browser (push) Has been skipped
Deploy Trade-In / test (push) Successful in 2m36s
Deploy Trade-In / build-backend (push) Successful in 1m0s
Deploy Trade-In / deploy (push) Successful in 1m28s
340 lines
22 KiB
Python
340 lines
22 KiB
Python
"""Daily recompute of the asking→sold correction ratios (#648 Stage 4).
|
||
|
||
ПРОБЛЕМА: asking_to_sold_ratios (migration 080) засеяна один раз derivation-CTE
|
||
(медиана SOLD ДКП за 12 мес vs медиана ASKING активных listings, per-rooms + global -1
|
||
fallback). Оценщик (estimator.py, Stage 3) читает её per-estimate (cache 300s) и домножает
|
||
asking-median прогноз на sold/asking. По мере ночного импорта новых ДКП-сделок
|
||
(rosreestr_dkp_import) ratio устаревает — нужен периодический пересчёт по тому же derivation.
|
||
|
||
TRUE-MIRROR REFRESH (флаг database-expert): 080-seed использовал ON CONFLICT DO UPDATE,
|
||
который при повторном прогоне ОБНОВЛЯЕТ существующие строки, но НЕ удаляет per-rooms строки
|
||
бакетов, упавших ниже порога 30/30 на более позднем прогоне (StaLE rows). Поэтому refresh
|
||
сначала DELETE FROM asking_to_sold_ratios WHERE district = '' (все строки #648), затем
|
||
заново гоняет ту же 080-derivation INSERT...SELECT. DELETE+INSERT в ОДНОЙ транзакции
|
||
(атомарно — таблица никогда не пуста mid-refresh). После commit таблица == свежий re-seed.
|
||
|
||
Задача синхронная (DB-only, никаких внешних HTTP-вызовов) — запускается kit-scheduler'ом
|
||
через product_handlers._job_asking_to_sold_ratio (run_in_executor), по образцу
|
||
snapshot_listing_sources / import_rosreestr_dkp.
|
||
|
||
Окно расписания 06:00-07:00 UTC — ПОСЛЕ rosreestr_dkp_import (04:00-06:00 UTC), чтобы
|
||
refresh потреблял свежие ДКП-сделки того же дня.
|
||
|
||
SQL derivation ниже повторяет seed в data/sql/080_asking_to_sold_ratios.sql (deal_side /
|
||
ask_side / per_bucket + deal_global / ask_global / global_row: трейлинг-12мес окно, ppm²-полоса
|
||
[_PPM2_MIN, settings.asking_ratio_ppm2_max] (default [30000,1200000]), порог n_deals>=30 AND
|
||
n_listings>=30 для per_rooms, global -1 строка всегда). ON CONFLICT убран — DELETE идёт первым,
|
||
конфликтов нет (повторный прогон в одной tx невозможен, refresh = re-seed по семантике).
|
||
|
||
#2620 — ОДНО ПРЕДНАМЕРЕННОЕ РАСХОЖДЕНИЕ с 080: deal_side бакетится по
|
||
LEAST(GREATEST(rooms,0),4), а ask_side — по _AREA_ROOMS_BUCKET_SQL (площадь, та же формула,
|
||
что deals.rooms получает при импорте). Причина — deals.rooms НЕ настоящая комнатность
|
||
(Росреестр её не отдаёт), это синтетика из площади; сравнивать её с РЕАЛЬНЫМИ комнатами
|
||
listings значило сравнивать разные классификации. Замер на проде (2026-08, #2620) показал
|
||
миграцию 23-55% объявлений между бакетами при таком сравнении — не только в бакете «4+»
|
||
(который к тому же обрезан обрезкой ELSE 4, тогда как listings.rooms доходит до 10) — и
|
||
это и была причина ratio>1 в бакете 4+ (см. _AREA_ROOMS_BUCKET_SQL ниже).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
|
||
from sqlalchemy import text
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.config import settings
|
||
from app.services import scrape_runs as runs_mod
|
||
|
||
# Нижняя граница ppm² — отсекает нежилые/технические сделки; не меняется.
|
||
_PPM2_MIN: int = 30_000
|
||
|
||
# #C2 — исторически asking-сторона (listings) была покрыта скрейпом ТОЛЬКО по ЕКБ, а
|
||
# миграция 177 залила ДКП-сделки по всей обл.66 (368 городов) → sold-медиана смешивала
|
||
# дешёвую область с ЕКБ-asking и обваливала ratio (0.877→0.62, «выкупная» −29% системно).
|
||
# Скоупили SOLD-сторону (deal_side/deal_global) на ЕКБ, чтобы sold и asking считались по
|
||
# ОДНОМУ рынку.
|
||
#
|
||
# #2583 H2 (аудит, 2026-08): oblast-развёртки заработали 12 июля — областные объявления
|
||
# попали в знаменатель (ask_side/ask_global) без городского скоупа, а sold-сторона
|
||
# осталась скоуплена на ЕКБ → асимметрия вернулась с другой стороны (дешёвая область
|
||
# занижает ask-медиану → ratio завышен на 2.5-5.3% по всем бакетам, выкупные цены
|
||
# системно переплачены). Теперь ask_side/ask_global ТОЖЕ скоупятся этим паттерном
|
||
# (предикат `city IS NULL OR city ILIKE :asking_city` — см. комментарий на месте в CTE
|
||
# ниже) — симметрично deal-стороне. Когда появится per-city ratio через зарезервированный
|
||
# столбец `district` (#647), эта константа станет per-city параметром для обеих сторон.
|
||
_ASKING_CITY_PATTERN: str = "%Екатеринбург%"
|
||
# Верхняя граница берётся из settings.asking_ratio_ppm2_max (default 1_200_000).
|
||
# QA-note: точное значение сверить с `SELECT max(price_per_m2) FROM deals
|
||
# WHERE source='rosreestr'` на проде — ceiling должен быть > max(ppm²) premium-сделок.
|
||
|
||
# #2620 — синтетический "бакет комнат по площади", ИСТОЧНИК ИСТИНЫ:
|
||
# tradein-mvp/deploy/import-rosreestr.sh (Росреестр не отдаёт комнатность — deals.rooms
|
||
# синтезируется из area_m2 при импорте ровно этим CASE). Три представления ОДНОЙ формулы —
|
||
# держи границы (30/44/62/85) в синхроне при правке: shell (import-rosreestr.sh) → SQL
|
||
# (эта константа, ask_side ниже) → Python (area_bucket() ниже, estimator.py rekey #2620-2).
|
||
_AREA_ROOMS_BUCKET_SQL = (
|
||
"CASE WHEN area_m2 < 30 THEN 0 WHEN area_m2 < 44 THEN 1 "
|
||
"WHEN area_m2 < 62 THEN 2 WHEN area_m2 < 85 THEN 3 ELSE 4 END"
|
||
)
|
||
|
||
|
||
def area_bucket(area_m2: float) -> int:
|
||
"""Python-двойник _AREA_ROOMS_BUCKET_SQL (границы ИДЕНТИЧНЫ, #2620).
|
||
|
||
Используется estimator.py при ПРИМЕНЕНИИ ratio (не только при расчёте здесь) —
|
||
ratio_resolver должен ключевать по ТОМУ ЖЕ area-бакету, что и ask_side при
|
||
деривации, иначе mismatch просто переезжает из расчёта в применение (прод-замер
|
||
ревьюера #2620: 310/1038 = 29.9% исторических запросов легли бы в другой бакет
|
||
при rooms-ключе vs area-ключе).
|
||
"""
|
||
if area_m2 < 30:
|
||
return 0
|
||
if area_m2 < 44:
|
||
return 1
|
||
if area_m2 < 62:
|
||
return 2
|
||
if area_m2 < 85:
|
||
return 3
|
||
return 4
|
||
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# ── True-mirror cleanup: drop all #648 rows before re-derivation ──────────────
|
||
# district = '' — все строки #648 (district зарезервирован под #647, пока всегда '').
|
||
# Удаляем ПЕРЕД re-derive, чтобы бакеты, упавшие ниже порога 30/30, не оставались
|
||
# stale (ON CONFLICT DO UPDATE такие строки бы не тронул). В одной транзакции с INSERT.
|
||
_DELETE_SQL = text(
|
||
"""
|
||
DELETE FROM asking_to_sold_ratios
|
||
WHERE district = ''
|
||
"""
|
||
)
|
||
|
||
# ── Derivation + re-seed (БАЙТ-В-БАЙТ из 080, ON CONFLICT убран — DELETE идёт первым) ──
|
||
# deal_side / ask_side / per_bucket + deal_global / ask_global / global_row:
|
||
# sold_median = percentile_cont(0.5) по deals.price_per_m2 (source='rosreestr',
|
||
# ppm² ∈ [_PPM2_MIN, settings.asking_ratio_ppm2_max], deal_date >= CURRENT_DATE − 12 months),
|
||
# бакет LEAST(GREATEST(rooms,0),4) (rooms уже синтетика-из-площади при импорте, см. #2620
|
||
# комментарий у _AREA_ROOMS_BUCKET_SQL выше).
|
||
# ask_median = percentile_cont(0.5) по listings.price_per_m2
|
||
# (is_active, та же ppm²-полоса [_PPM2_MIN, asking_ratio_ppm2_max], тот же город что
|
||
# SOLD-сторона — city IS NULL OR city ILIKE :asking_city, #2583 H2). Бакет —
|
||
# _AREA_ROOMS_BUCKET_SQL (площадь, #2620), НЕ listings.rooms — см. комментарий там.
|
||
# per_rooms строки — только при n_deals>=30 AND n_listings>=30 AND ask>0 AND sold>0.
|
||
# global -1 строка (basis='global_fallback') — всегда (если ask>0 AND sold>0). window_months=12.
|
||
# Порог/окно — литералы; ppm²-полоса передаётся bind-параметрами :ppm2_min/:ppm2_max
|
||
# (безопасно от SQL-инъекций; CAST не нужен — psycopg v3 передаёт int напрямую).
|
||
_REDERIVE_SQL = text(
|
||
f"""
|
||
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 city ILIKE :asking_city -- #C2 SOLD-сторона на ЕКБ (match asking-рынок)
|
||
AND price_per_m2 BETWEEN :ppm2_min AND :ppm2_max
|
||
AND deal_date >= CURRENT_DATE - INTERVAL '12 months'
|
||
GROUP BY LEAST(GREATEST(rooms, 0), 4)
|
||
),
|
||
-- ASKING медианы по ТОМУ ЖЕ area-бакету, что deal_side (#2620) — НЕ по listings.rooms.
|
||
-- deals.rooms — синтетика из площади (Росреестр её не отдаёт), listings.rooms — реальная
|
||
-- комнатность; сравнение area-бакета с area-бакетом (не area-бакета с real-rooms-бакетом)
|
||
-- убирает миграцию объявлений между бакетами (23-55% строк на проде, 2026-08, #2620) —
|
||
-- включая инверсию ratio>1 в бакете «4+» (deals.rooms обрезан ELSE 4, а listings.rooms
|
||
-- нет: 110/782 пяти- и более комнатных объявлений раньше схлопывались в бакет 4).
|
||
ask_side AS (
|
||
SELECT
|
||
{_AREA_ROOMS_BUCKET_SQL} 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
|
||
-- #2620 hardening: area_m2 IS NULL falls into the CASE ELSE branch (bucket 4)
|
||
-- of _AREA_ROOMS_BUCKET_SQL — a latent "everything unmeasured looks like a big
|
||
-- flat" trap. Excluded explicitly instead of relying on ELSE-as-junk-drawer.
|
||
AND area_m2 IS NOT NULL
|
||
AND price_per_m2 BETWEEN :ppm2_min AND :ppm2_max
|
||
-- novostroyki guard (#1186): NULL = legacy вторичка до м.011
|
||
AND (listing_segment IS NULL OR listing_segment = 'vtorichka')
|
||
-- #2583 H2: скоупим ASKING-сторону на тот же город, что и SOLD-сторона
|
||
-- (симметрично deal_side выше) — иначе дешёвые oblast-объявления (развёртки
|
||
-- с 12 июля) занижают ask-медиану и завышают ratio. city IS NULL считается
|
||
-- "своим" (не отбрасывается) НАМЕРЕННО: listings.city заполнена пока только у
|
||
-- Авито (Циан/Домклик/Яндекс — NULL, #2598/#2606), симметричный
|
||
-- `city ILIKE :asking_city` без IS NULL выбросил бы ~70% выборки. По мере
|
||
-- роста покрытия колонки этот предикат сам ужесточается без правок кода; когда
|
||
-- покрытие станет полным — заменить на строго симметричный `city ILIKE :asking_city`.
|
||
AND (city IS NULL OR city ILIKE :asking_city)
|
||
GROUP BY {_AREA_ROOMS_BUCKET_SQL}
|
||
),
|
||
-- 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 city ILIKE :asking_city -- #C2 SOLD-сторона на ЕКБ (match asking-рынок)
|
||
AND price_per_m2 BETWEEN :ppm2_min AND :ppm2_max
|
||
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
|
||
-- #2620 hardening: same area_m2 IS NOT NULL as ask_side — keeps the global-row
|
||
-- population consistent with the per-bucket rows it's a fallback for.
|
||
AND area_m2 IS NOT NULL
|
||
AND price_per_m2 BETWEEN :ppm2_min AND :ppm2_max
|
||
-- novostroyki guard (#1186): NULL = legacy вторичка до м.011
|
||
AND (listing_segment IS NULL OR listing_segment = 'vtorichka')
|
||
-- #2583 H2: тот же городской скоуп, что и ask_side выше (см. комментарий там
|
||
-- про причину city IS NULL == "свой" и #2598/#2606).
|
||
AND (city IS NULL OR city ILIKE :asking_city)
|
||
),
|
||
-- 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
|
||
"""
|
||
)
|
||
|
||
# ── Post-insert counters ──────────────────────────────────────────────────────
|
||
# Считываем итог из таблицы (всё ещё в той же транзакции — до commit): сколько строк
|
||
# записано всего, сколько per_rooms, был ли использован global -1 fallback.
|
||
_COUNTERS_SQL = text(
|
||
"""
|
||
SELECT
|
||
COUNT(*) AS rows_written,
|
||
COUNT(*) FILTER (WHERE basis = 'per_rooms') AS per_rooms_rows,
|
||
COUNT(*) FILTER (WHERE rooms_bucket = -1) AS used_global_fallback
|
||
FROM asking_to_sold_ratios
|
||
WHERE district = ''
|
||
"""
|
||
)
|
||
|
||
|
||
def recompute_asking_to_sold_ratios(db: Session, run_id: int) -> dict[str, int]:
|
||
"""Пересчитать asking_to_sold_ratios (TRUE-MIRROR refresh, #648 Stage 4).
|
||
|
||
Sync (вызывается scheduler-триггером в executor, как snapshot_listing_sources).
|
||
В ОДНОЙ транзакции (атомарно — таблица никогда не пуста mid-refresh):
|
||
1. DELETE FROM asking_to_sold_ratios WHERE district = '' — снести stale-строки.
|
||
2. Заново прогнать 080-derivation INSERT...SELECT (per_rooms при 30/30 + global -1).
|
||
Затем counters из таблицы, commit, mark_done. Семантика == re-seed миграции 080.
|
||
|
||
Финализирует scrape_runs (mark_done / mark_failed) и пишет counters.
|
||
|
||
LIMITATION (#2620, честно задокументировано — не гард, а факт данных): sold-сторона
|
||
(deals) НЕ имеет маркера новостройка/вторичка — Росреестр таким свойством ДКП не
|
||
делится, а listing_segment (гард #1186) существует только у listings. ask_side/ask_global
|
||
отфильтрованы на вторичку, deal_side/deal_global — нет. Замер на проде (2026-08, #2620):
|
||
доля сделок с year_built >= 2020 (грубый прокси новостройки) — 44.1% в бакете «4+» против
|
||
23.1% в бакетах 1-3 — заметный перекос, но year_built НЕ идентифицирует первичку/вторичку
|
||
(продажа квартиры 2021 года постройки в 2026м — легитимная вторичка), поэтому фильтр по
|
||
году НЕ добавлен (создал бы новую, столь же спекулятивную асимметрию). Area-бакет-фикс
|
||
ниже (см. _AREA_ROOMS_BUCKET_SQL) сам по себе убрал инверсию ratio>1 в бакете «4+»
|
||
(0.8315 на замере прод-данных 2026-08, было 1.0257) — снятие миграции между бакетами было
|
||
root cause, а не новостройки.
|
||
|
||
Returns {"rows_written": N, "per_rooms_rows": M, "used_global_fallback": 0|1}.
|
||
"""
|
||
counters: dict[str, int] = {
|
||
"rows_written": 0,
|
||
"per_rooms_rows": 0,
|
||
"used_global_fallback": 0,
|
||
}
|
||
try:
|
||
# DELETE + re-derive INSERT в одной транзакции (НЕ коммитим между ними —
|
||
# таблица не должна остаться пустой, если INSERT упадёт).
|
||
db.execute(_DELETE_SQL)
|
||
db.execute(
|
||
_REDERIVE_SQL,
|
||
{
|
||
"ppm2_min": _PPM2_MIN,
|
||
"ppm2_max": settings.asking_ratio_ppm2_max,
|
||
"asking_city": _ASKING_CITY_PATTERN,
|
||
},
|
||
)
|
||
|
||
row = db.execute(_COUNTERS_SQL).mappings().first()
|
||
if row is not None:
|
||
counters["rows_written"] = int(row["rows_written"] or 0)
|
||
counters["per_rooms_rows"] = int(row["per_rooms_rows"] or 0)
|
||
counters["used_global_fallback"] = int(row["used_global_fallback"] or 0)
|
||
|
||
db.commit()
|
||
runs_mod.mark_done(db, run_id, counters)
|
||
logger.info(
|
||
"recompute_asking_to_sold_ratios run_id=%d done: "
|
||
"rows_written=%d per_rooms_rows=%d used_global_fallback=%d",
|
||
run_id,
|
||
counters["rows_written"],
|
||
counters["per_rooms_rows"],
|
||
counters["used_global_fallback"],
|
||
)
|
||
return counters
|
||
except Exception as exc:
|
||
logger.exception("recompute_asking_to_sold_ratios run_id=%d failed", run_id)
|
||
db.rollback()
|
||
runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters)
|
||
raise
|