lekss361/gendesign
1
0
Fork 0
Code Issues 57 Pull requests Projects Releases Packages Wiki Activity Actions

feat(tradein): ppm²-tier segmentation of asking→sold ratio (#928, flag OFF) #934

Merged
bot-reviewer merged 2 commits from feat/tradein-ppm2-tier-ratio into main 2026-05-31 21:22:40 +00:00
Conversation 0 Commits 2 Files changed 6 +1118 -44
6 changed files with 1118 additions and 44 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
tradein-mvp/backend/app/core/config.py
View file

@ -149,6 +149,18 @@ class Settings(BaseSettings):
# ENV: ASKING_RATIO_PPM2_MAX.
asking_ratio_ppm2_max: int = 1_200_000
# ppm2-tier asking->sold lookup (#928) -- gate + empirical-Bayes shrink parameter.
# tier_aware_ratio_enabled=False (default) -> byte-identical flag-OFF path via
# asking_to_sold_ratios (old table). True -> tiered lookup from
# asking_to_sold_ratios_tiered (requires migration 098 applied).
# ENV: TIER_AWARE_RATIO_ENABLED.
tier_aware_ratio_enabled: bool = False
# Empirical-Bayes shrink weight denominator: w = n_deals / (n_deals + K).
# K=150 -> cell with 150 deals gets w=0.5 (50% own tier ratio, 50% rooms-'all').
# Higher K -> more shrinkage toward 'all' ratio (conservative); lower -> more tier.
# ENV: TIER_RATIO_SHRINK_K.
tier_ratio_shrink_k: int = 150
# SSRF-защита для admin scrape endpoints (#756).
# Список хостов которым разрешено передавать абсолютные URL в параметрах *_url.
# Относительные пути (без netloc) проходят без проверки — хост подставляется

231
tradein-mvp/backend/app/services/estimator.py
View file

@ -194,68 +194,223 @@ def _repair_coefficient(repair_state: str | None) -> float:
# TTL более чем достаточно и снимает по SELECT'у с каждой оценки. Single-worker
# uvicorn/scheduler — GIL делает dict-доступ atomic enough (без явного lock).
_ASKING_SOLD_RATIO_CACHE_TTL_S = 300.0
_asking_sold_ratio_cache: dict[int, tuple[float | None, str | None, float]] = {}
# Cache key: (bucket, tier_or_None) — tier=None for flag-OFF / no-anchor path (#928).
_asking_sold_ratio_cache: dict[tuple[int, str | None], tuple[float | None, str | None, float]] = {}
def _get_asking_sold_ratio(db: Session, rooms: int | None) -> tuple[float | None, str | None]:
def _get_asking_sold_ratio(
db: Session,
rooms: int | None,
anchor_ppm2: float | None = None,
) -> tuple[float | None, str | None]:
"""Возвращает (ratio, basis) asking→sold для бакета комнат.
bucket = min(max(rooms or 0, 0), 4). Сначала ищем per-rooms строку
(district=''), при отсутствии — global fallback (rooms_bucket=-1). Если
таблицы нет / пуста / любая ошибка → (None, None), НЕ raise (graceful:
estimator продолжает без sold-коррекции, headline asking-медиана отдаётся).
bucket = min(max(rooms or 0, 0), 4).
Кэшируется на бакет с TTL _ASKING_SOLD_RATIO_CACHE_TTL_S.
Flag-OFF (settings.tier_aware_ratio_enabled=False) или anchor_ppm2=None:
Запрос к asking_to_sold_ratios (старая таблица) — байт-в-байт как до #928.
Fallback: per-rooms строка → global -1.
Flag-ON и anchor_ppm2 задан (#928 ppm2-tier path):
1. Читаем t33/t66 из asking_to_sold_tier_bounds для (bucket, '').
Нет bounds-строки → fallback к 'all' строке (шаг 3).
2. tier = 'low'/'mid'/'high' по anchor_ppm2 vs t33/t66.
Читаем asking_to_sold_ratios_tiered (bucket,'',tier) → tier_ratio, n_deals.
Shrink: w = n_deals/(n_deals + settings.tier_ratio_shrink_k).
Также читаем (bucket,'','all') как shrink-target.
shrunk = tier_ratio * w + all_ratio * (1 - w).
basis = f'per_rooms_tier:{tier}'.
3. Fallback (bucket,'','all') → basis='per_rooms_all'.
4. Fallback (-1,'','all') → basis='global_all'.
5. Ничего → (None, None).
Таблицы нет / любая ошибка → (None, None), НЕ raise (graceful).
Кэшируется на ключ (bucket, tier|None) с TTL _ASKING_SOLD_RATIO_CACHE_TTL_S.
"""
bucket = min(max(rooms or 0, 0), 4)
cached = _asking_sold_ratio_cache.get(bucket)
if cached is not None:
ratio, basis, fetched = cached
if (time.monotonic() - fetched) < _ASKING_SOLD_RATIO_CACHE_TTL_S:
return ratio, basis
# Determine tier key for cache:
# flag-OFF or no anchor_ppm2 → use None (legacy path, reads old table).
tier_key: str | None = None
use_tier_path = settings.tier_aware_ratio_enabled and anchor_ppm2 is not None
cache_key: tuple[int, str | None] = (bucket, tier_key) # may be updated below
if not use_tier_path:
# ── Flag-OFF path: byte-identical behavior (legacy asking_to_sold_ratios) ──
# Use a distinct key "_legacy" to avoid collision with the flag-ON Step-3 'all'
# fallback which also caches under (bucket, None). Without this, a process that
# flips tier_aware_ratio_enabled ON without restart could serve the old-table ratio
# from the flag-OFF entry for up to the TTL — wrong table.
cache_key = (bucket, "_legacy")
cached = _asking_sold_ratio_cache.get(cache_key)
if cached is not None:
ratio, basis, fetched = cached
if (time.monotonic() - fetched) < _ASKING_SOLD_RATIO_CACHE_TTL_S:
return ratio, basis
ratio: float | None = None
basis: str | None = None
try:
row = db.execute(
text(
"""
SELECT ratio, basis FROM asking_to_sold_ratios
WHERE rooms_bucket = CAST(:b AS int) AND district = ''
"""
),
{"b": bucket},
).fetchone()
if row is None:
# Бакет тонкий (n<30 при seed'е) или отсутствует → global fallback (-1).
row = db.execute(
text(
"""
SELECT ratio, basis FROM asking_to_sold_ratios
WHERE rooms_bucket = -1 AND district = ''
"""
),
).fetchone()
if row is not None and row.ratio is not None:
ratio = float(row.ratio)
basis = row.basis
except Exception as exc:
# Таблицы может не быть на свежей/старой БД (миграция 080 не применена),
# либо транзакция в сбойном состоянии — graceful: без sold-коррекции.
# ОБЯЗАТЕЛЬНО rollback: неудачный SELECT помечает транзакцию
# InFailedSqlTransaction, и без отката следующий statement упал бы → 500.
logger.debug("asking_to_sold_ratio lookup skipped (graceful): %s", exc)
try:
db.rollback()
except Exception:
pass
ratio, basis = None, None
_asking_sold_ratio_cache[cache_key] = (ratio, basis, time.monotonic())
return ratio, basis
# ── Flag-ON path: ppm2-tier lookup (#928) ───────────────────────────────────
ratio: float | None = None
basis: str | None = None
try:
row = db.execute(
# Step 1: read bounds for this bucket.
bounds_row = db.execute(
text(
"""
SELECT ratio, basis FROM asking_to_sold_ratios
SELECT t33, t66 FROM asking_to_sold_tier_bounds
WHERE rooms_bucket = CAST(:b AS int) AND district = ''
"""
),
{"b": bucket},
).fetchone()
if row is None:
# Бакет тонкий (n<30 при seed'е) или отсутствует → global fallback (-1).
row = db.execute(
if bounds_row is not None:
t33 = float(bounds_row.t33)
t66 = float(bounds_row.t66)
tier_key = "low" if anchor_ppm2 < t33 else ("mid" if anchor_ppm2 < t66 else "high")
cache_key = (bucket, tier_key)
# Check cache for this (bucket, tier) key.
cached = _asking_sold_ratio_cache.get(cache_key)
if cached is not None:
c_ratio, c_basis, fetched = cached
if (time.monotonic() - fetched) < _ASKING_SOLD_RATIO_CACHE_TTL_S:
return c_ratio, c_basis
# Step 2: look up tier row + 'all' row for shrink.
tier_row = db.execute(
text(
"""
SELECT ratio, basis FROM asking_to_sold_ratios
WHERE rooms_bucket = -1 AND district = ''
SELECT ratio, n_deals FROM asking_to_sold_ratios_tiered
WHERE rooms_bucket = CAST(:b AS int)
AND district = ''
AND ppm2_tier = CAST(:tier AS text)
"""
),
{"b": bucket, "tier": tier_key},
).fetchone()
if row is not None and row.ratio is not None:
ratio = float(row.ratio)
basis = row.basis
all_row = db.execute(
text(
"""
SELECT ratio FROM asking_to_sold_ratios_tiered
WHERE rooms_bucket = CAST(:b AS int)
AND district = ''
AND ppm2_tier = 'all'
"""
),
{"b": bucket},
).fetchone()
if tier_row is not None and all_row is not None:
tier_ratio = float(tier_row.ratio)
n_deals = int(tier_row.n_deals)
all_ratio = float(all_row.ratio)
k = settings.tier_ratio_shrink_k
w = n_deals / (n_deals + k)
ratio = tier_ratio * w + all_ratio * (1.0 - w)
basis = f"per_rooms_tier:{tier_key}"
_asking_sold_ratio_cache[cache_key] = (ratio, basis, time.monotonic())
return ratio, basis
# Step 3: fallback to 'all' for this bucket.
cache_key = (bucket, None)
cached = _asking_sold_ratio_cache.get(cache_key)
if cached is not None:
c_ratio, c_basis, fetched = cached
if (time.monotonic() - fetched) < _ASKING_SOLD_RATIO_CACHE_TTL_S:
return c_ratio, c_basis
all_row = db.execute(
text(
"""
SELECT ratio FROM asking_to_sold_ratios_tiered
WHERE rooms_bucket = CAST(:b AS int)
AND district = ''
AND ppm2_tier = 'all'
"""
),
{"b": bucket},
).fetchone()
if all_row is not None:
ratio = float(all_row.ratio)
basis = "per_rooms_all"
_asking_sold_ratio_cache[cache_key] = (ratio, basis, time.monotonic())
return ratio, basis
# Step 4: global fallback (-1, 'all').
global_key: tuple[int, str | None] = (-1, None)
cached = _asking_sold_ratio_cache.get(global_key)
if cached is not None:
c_ratio, c_basis, fetched = cached
if (time.monotonic() - fetched) < _ASKING_SOLD_RATIO_CACHE_TTL_S:
return c_ratio, c_basis
global_row = db.execute(
text(
"""
SELECT ratio FROM asking_to_sold_ratios_tiered
WHERE rooms_bucket = -1
AND district = ''
AND ppm2_tier = 'all'
"""
),
).fetchone()
if global_row is not None:
ratio = float(global_row.ratio)
basis = "global_all"
_asking_sold_ratio_cache[global_key] = (ratio, basis, time.monotonic())
return ratio, basis
except Exception as exc:
# Таблицы может не быть на свежей/старой БД (миграция 080 не применена),
# либо транзакция в сбойном состоянии — graceful: без sold-коррекции.
# ОБЯЗАТЕЛЬНО rollback (как в sibling-helper'ах _get_or_fetch_*): неудачный
# SELECT помечает транзакцию InFailedSqlTransaction, и без отката следующий
# statement (_fetch_deals) упал бы → 500. Откат держит shared session чистой
# для последующего INSERT. rollback тоже guard'им (соединение могло умереть).
logger.debug("asking_to_sold_ratio lookup skipped (graceful): %s", exc)
# Таблицы нет / ошибка → graceful (migration 098 ещё не применена).
logger.debug("asking_to_sold_ratio tiered lookup skipped (graceful): %s", exc)
try:
db.rollback()
except Exception:
pass
ratio, basis = None, None
_asking_sold_ratio_cache[bucket] = (ratio, basis, time.monotonic())
return ratio, basis
# Step 5: nothing found.
return None, None
# ── Avito IMV cache lookup (Stage 3) ────────────────────────────────────────
@ -1861,7 +2016,17 @@ async def estimate_quality(
# expected_sold остаётся pre-blend → asking 75M / sold 45M (бессмысленная скидка
# в HeroSummary) и stale-значения persist'ятся в trade_in_estimates. Здесь только
# резолвим ratio/basis (нужны для confidence/explanation и null-guard).
asking_to_sold_ratio, ratio_basis = _get_asking_sold_ratio(db, payload.rooms)
# #928: pass median_ppm2 (best proxy available at this point — anchor_ppm2 from
# same-building anchor computed below, but that's post-call). median_ppm2 = 0 when
# no radius analogs yet; tier lookup uses it as the ppm2 placement signal.
# NOTE(#928): tier placement uses the pre-anchor radius median (median_ppm2), not the
# same-building Tukey anchor that #928 ideally specifies (anchor is computed post-call).
# Coarse 3-tier bucketing is robust to this for the golden case; revisit (resolve ratio
# after anchor) before flipping tier_aware_ratio_enabled ON — validate in the held-out
# backtest.
asking_to_sold_ratio, ratio_basis = _get_asking_sold_ratio(
db, payload.rooms, anchor_ppm2=median_ppm2 if median_ppm2 > 0 else None
)
expected_sold_per_m2: int | None = None
expected_sold_price: int | None = None
expected_sold_range_low: int | None = None

247
tradein-mvp/backend/app/tasks/asking_to_sold_ratio.py
View file

@ -185,6 +185,229 @@ _COUNTERS_SQL = text(
)
# ── Tiered refresh (#928): DELETE + re-derive for tier tables ────────────────
# атомарно DELETE WHERE district='' ПЕРЕД re-derive INSERT (те же семантика, что
# для asking_to_sold_ratios выше: stale строки не оставляем).
_DELETE_TIERED_ROWS_SQL = text(
"""
DELETE FROM asking_to_sold_ratios_tiered WHERE district = ''
"""
)
_DELETE_TIER_BOUNDS_SQL = text(
"""
DELETE FROM asking_to_sold_tier_bounds WHERE district = ''
"""
)
# ── Tiered derivation (BYTE-CONSISTENT с migration 098) ──────────────────────
# Те же CTE что в 098: bounds -> deal_tiered/ask_tiered -> tier_rows + all_rows + global_row.
# Bind params: :ppm2_min / :ppm2_max (как в _REDERIVE_SQL выше).
# CAST не нужен -- psycopg v3 передаёт int напрямую.
_REDERIVE_BOUNDS_SQL = text(
"""
WITH bounds AS (
SELECT
LEAST(GREATEST(rooms, 0), 4) AS rooms_bucket,
percentile_cont(0.3333) WITHIN GROUP (ORDER BY price_per_m2)::bigint AS t33,
percentile_cont(0.6667) WITHIN GROUP (ORDER BY price_per_m2)::bigint AS t66,
COUNT(*)::int AS n_listings
FROM listings
WHERE is_active
AND rooms IS NOT NULL
AND price_per_m2 BETWEEN :ppm2_min AND :ppm2_max
GROUP BY LEAST(GREATEST(rooms, 0), 4)
HAVING COUNT(*) > 0
)
INSERT INTO asking_to_sold_tier_bounds (rooms_bucket, district, t33, t66, n_listings)
SELECT rooms_bucket, ''::text, t33, t66, n_listings FROM bounds
"""
)
_REDERIVE_TIERED_SQL = text(
"""
WITH
bnd AS (
SELECT rooms_bucket, t33, t66 FROM asking_to_sold_tier_bounds WHERE district = ''
),
deal_tiered AS (
SELECT
LEAST(GREATEST(d.rooms, 0), 4) AS rooms_bucket,
CASE
WHEN d.price_per_m2 < b.t33 THEN 'low'
WHEN d.price_per_m2 < b.t66 THEN 'mid'
ELSE 'high'
END AS ppm2_tier,
d.price_per_m2
FROM deals d
JOIN bnd b ON b.rooms_bucket = LEAST(GREATEST(d.rooms, 0), 4)
WHERE d.source = 'rosreestr'
AND d.rooms IS NOT NULL
AND d.price_per_m2 BETWEEN :ppm2_min AND :ppm2_max
AND d.deal_date >= CURRENT_DATE - INTERVAL '12 months'
),
ask_tiered AS (
SELECT
LEAST(GREATEST(l.rooms, 0), 4) AS rooms_bucket,
CASE
WHEN l.price_per_m2 < b.t33 THEN 'low'
WHEN l.price_per_m2 < b.t66 THEN 'mid'
ELSE 'high'
END AS ppm2_tier,
l.price_per_m2
FROM listings l
JOIN bnd b ON b.rooms_bucket = LEAST(GREATEST(l.rooms, 0), 4)
WHERE l.is_active
AND l.rooms IS NOT NULL
AND l.price_per_m2 BETWEEN :ppm2_min AND :ppm2_max
),
deal_side_tier AS (
SELECT
rooms_bucket,
ppm2_tier,
percentile_cont(0.5) WITHIN GROUP (ORDER BY price_per_m2) AS sold_median,
COUNT(*)::int AS n_deals
FROM deal_tiered
GROUP BY rooms_bucket, ppm2_tier
),
ask_side_tier AS (
SELECT
rooms_bucket,
ppm2_tier,
percentile_cont(0.5) WITHIN GROUP (ORDER BY price_per_m2) AS ask_median,
COUNT(*)::int AS n_listings
FROM ask_tiered
GROUP BY rooms_bucket, ppm2_tier
),
tier_rows AS (
SELECT
d.rooms_bucket,
''::text AS district,
d.ppm2_tier,
(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,
a.n_listings,
12 AS window_months,
'per_rooms_tier'::text AS basis
FROM deal_side_tier d
JOIN ask_side_tier a USING (rooms_bucket, ppm2_tier)
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
AND d.sold_median > 0
),
deal_side_all 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(*)::int AS n_deals
FROM deals
WHERE source = 'rosreestr'
AND rooms IS NOT NULL
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)
),
ask_side_all 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(*)::int AS n_listings
FROM listings
WHERE is_active
AND rooms IS NOT NULL
AND price_per_m2 BETWEEN :ppm2_min AND :ppm2_max
GROUP BY LEAST(GREATEST(rooms, 0), 4)
),
all_rows AS (
SELECT
d.rooms_bucket,
''::text AS district,
'all'::text AS ppm2_tier,
(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,
a.n_listings,
12 AS window_months,
'per_rooms_all'::text AS basis
FROM deal_side_all d
JOIN ask_side_all 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
AND d.sold_median > 0
),
deal_global AS (
SELECT
percentile_cont(0.5) WITHIN GROUP (ORDER BY price_per_m2) AS sold_median,
COUNT(*)::int AS n_deals
FROM deals
WHERE source = 'rosreestr'
AND rooms IS NOT NULL
AND price_per_m2 BETWEEN :ppm2_min AND :ppm2_max
AND deal_date >= CURRENT_DATE - INTERVAL '12 months'
),
ask_global AS (
SELECT
percentile_cont(0.5) WITHIN GROUP (ORDER BY price_per_m2) AS ask_median,
COUNT(*)::int AS n_listings
FROM listings
WHERE is_active
AND rooms IS NOT NULL
AND price_per_m2 BETWEEN :ppm2_min AND :ppm2_max
),
global_row AS (
SELECT
-1 AS rooms_bucket,
''::text AS district,
'all'::text AS ppm2_tier,
(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,
a.n_listings,
12 AS window_months,
'global_all'::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
AND d.sold_median > 0
)
INSERT INTO asking_to_sold_ratios_tiered (
rooms_bucket, district, ppm2_tier, ratio, sold_median, ask_median,
n_deals, n_listings, window_months, basis
)
SELECT rooms_bucket, district, ppm2_tier, ratio, sold_median, ask_median,
n_deals, n_listings, window_months, basis FROM global_row
UNION ALL
SELECT rooms_bucket, district, ppm2_tier, ratio, sold_median, ask_median,
n_deals, n_listings, window_months, basis FROM all_rows
UNION ALL
SELECT rooms_bucket, district, ppm2_tier, ratio, sold_median, ask_median,
n_deals, n_listings, window_months, basis FROM tier_rows
"""
)
# ── Post-insert counters (tiered) ─────────────────────────────────────────────
_TIERED_COUNTERS_SQL = text(
"""
SELECT
(SELECT COUNT(*) FROM asking_to_sold_ratios_tiered WHERE district = '')
AS tiered_rows_written,
(SELECT COUNT(*) FROM asking_to_sold_tier_bounds WHERE district = '')
AS tier_bounds_written
"""
)
def recompute_asking_to_sold_ratios(db: Session, run_id: int) -> dict[str, int]:
"""Пересчитать asking_to_sold_ratios (TRUE-MIRROR refresh, #648 Stage 4).
@ -202,6 +425,8 @@ def recompute_asking_to_sold_ratios(db: Session, run_id: int) -> dict[str, int]:
"rows_written": 0,
"per_rooms_rows": 0,
"used_global_fallback": 0,
"tiered_rows_written": 0,
"tier_bounds_written": 0,
}
try:
# DELETE + re-derive INSERT в одной транзакции (НЕ коммитим между ними —
@ -218,15 +443,35 @@ def recompute_asking_to_sold_ratios(db: Session, run_id: int) -> dict[str, int]:
counters["per_rooms_rows"] = int(row["per_rooms_rows"] or 0)
counters["used_global_fallback"] = int(row["used_global_fallback"] or 0)
# Tiered refresh (#928): DELETE + re-derive bounds + tier rows в той же транзакции.
# Атомарно — таблицы никогда не пустые mid-refresh (всё в одном commit ниже).
db.execute(_DELETE_TIERED_ROWS_SQL)
db.execute(_DELETE_TIER_BOUNDS_SQL)
db.execute(
_REDERIVE_BOUNDS_SQL,
{"ppm2_min": _PPM2_MIN, "ppm2_max": settings.asking_ratio_ppm2_max},
)
db.execute(
_REDERIVE_TIERED_SQL,
{"ppm2_min": _PPM2_MIN, "ppm2_max": settings.asking_ratio_ppm2_max},
)
tier_row = db.execute(_TIERED_COUNTERS_SQL).mappings().first()
if tier_row is not None:
counters["tiered_rows_written"] = int(tier_row["tiered_rows_written"] or 0)
counters["tier_bounds_written"] = int(tier_row["tier_bounds_written"] 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",
"rows_written=%d per_rooms_rows=%d used_global_fallback=%d "
"tiered_rows_written=%d tier_bounds_written=%d",
run_id,
counters["rows_written"],
counters["per_rooms_rows"],
counters["used_global_fallback"],
counters["tiered_rows_written"],
counters["tier_bounds_written"],
)
return counters
except Exception as exc:

301
tradein-mvp/backend/data/sql/098_asking_to_sold_ratios_tiered.sql Normal file
View file

@ -0,0 +1,301 @@
-- 098_asking_to_sold_ratios_tiered.sql
-- #928 -- ppm2-tier segmentation of the asking->sold ratio pool.
--
-- ПРОБЛЕМА (#928). Единый per-rooms ratio (migration 080, asking_to_sold_ratios)
-- смешивает в одном коэффициенте квартиры разных ценовых сегментов. Backtest
-- показал, что ratio варьируется ~0.91 (эконом) -> 0.99 (премиум): global ratio
-- ~0.80 = средневзвешенное. ppm2-тир: binning listings ВНУТРИ каждого rooms_bucket
-- на три равных квартиля (33.33/66.67) -> три tier-строки с ratio + empirical-Bayes
-- shrink к per-rooms 'all' (K=150).
--
-- ПОЧЕМУ ПАРАЛЛЕЛЬНАЯ ТАБЛИЦА, а не ADD COLUMN на asking_to_sold_ratios:
-- Flag-OFF путь: SELECT ratio, basis FROM asking_to_sold_ratios
-- WHERE rooms_bucket=B AND district='' -> fetchone(). Добавление tier-строк
-- в ту же таблицу с другим PK -> fetchone() вернул бы произвольную строку ->
-- тихая продовая регрессия. Параллельная таблица изолирует tier-данные;
-- старый путь (flag OFF) остаётся байт-в-байт идентичным.
--
-- СОЗДАЁТ (additive, idempotent -- без DROP/ALTER любых существующих объектов):
-- asking_to_sold_tier_bounds -- ppm2-квантили (33.33/66.67 pct) per бакет.
-- asking_to_sold_ratios_tiered -- per-tier ratio, 'all' агрегаты, global -1.
-- ppm2_tier: 'low'|'mid'|'high' (tier) или 'all' (per-rooms агрегат / global).
-- basis: 'per_rooms_tier' | 'per_rooms_all' | 'global_all'.
--
-- МЕТОДОЛОГИЯ:
-- bounds: percentile_cont(0.3333/0.6667) по listings.price_per_m2
-- (is_active, ppm2 in [30000,1200000]) per rooms_bucket.
-- Tier rows: listings и deals биннятся в low/mid/high по t33/t66 СВОЕГО бакета.
-- ratio = percentile_cont(0.5) deals / percentile_cont(0.5) listings в ячейке.
-- Только ячейки n_deals>=30 AND n_listings>=30 AND ask_median>0 AND sold_median>0.
-- 'all' строки: per-rooms 080-логика с полосой [30000,1200000], basis='per_rooms_all'.
-- global строка: (-1,'','all') 080-логика, basis='global_all'.
--
-- ВАЛИДАЦИЯ (продовые числа 2026-05-31): все 15 ячеек (5 rooms x 3 tier) имеют
-- n_deals>=169, n_listings>=583. rb=2 t33=129167 t66=163870;
-- rb=2 low~0.911, mid~0.978, high~0.991; rb=2 'all'~0.803; global 'all'~0.790.
--
-- ЗАВИСИМОСТИ: deals, listings (002_core_tables.sql).
-- Apply after: 097_index_hygiene.sql
BEGIN;
-- asking_to_sold_tier_bounds -- ppm2-квантили активных listings per бакет --------
-- t33/t66 -- 33.33/66.67 pct ppm2 в бакете. Refresh атомарно (DELETE + re-derive).
CREATE TABLE IF NOT EXISTS asking_to_sold_tier_bounds (
rooms_bucket int NOT NULL, -- 0=студия..4=4+
district text NOT NULL DEFAULT '', -- RESERVED для #647; всегда ''
t33 bigint NOT NULL, -- 33.33-й pct ppm2 listings в бакете
t66 bigint NOT NULL, -- 66.67-й pct ppm2 listings в бакете
n_listings int NOT NULL, -- число активных listings в бакете
computed_at timestamptz NOT NULL DEFAULT now(),
PRIMARY KEY (rooms_bucket, district)
);
COMMENT ON TABLE asking_to_sold_tier_bounds IS
'ppm2-квантили активных listings per rooms_bucket (#928): t33/t66 = 33.33/66.67 pct '
'price_per_m2 среди is_active AND ppm2 in [30000,1200000]. Используется estimator-ом '
'для tier-assignment (low/mid/high) при флаге tier_aware_ratio_enabled=True. '
'Refresh -- asking_to_sold_ratio_refresh DELETE WHERE district='''' + re-derive. '
'district RESERVED под #647 (geo), всегда ''''. '
'Параллельная к asking_to_sold_ratios (#928 rationale: fetchone-safety).';
-- asking_to_sold_ratios_tiered -- tier-коэффициенты asking->sold (#928) -----------
-- ppm2_tier: 'low'|'mid'|'high' = tier-строки; 'all' = per-rooms агрегат / global.
-- basis: 'per_rooms_tier'|'per_rooms_all'|'global_all'.
-- MIN-SAMPLE: ячейки n_deals<30 OR n_listings<30 не пишутся -> estimator fallback.
CREATE TABLE IF NOT EXISTS asking_to_sold_ratios_tiered (
rooms_bucket int NOT NULL, -- 0=студия..4=4+; -1 = global
district text NOT NULL DEFAULT '', -- RESERVED под #647; всегда ''
ppm2_tier text NOT NULL, -- 'low'|'mid'|'high' | 'all'
ratio numeric NOT NULL, -- sold_median_ppm2 / ask_median_ppm2
sold_median bigint, -- медиана deals.price_per_m2
ask_median bigint, -- медиана listings.price_per_m2
n_deals int NOT NULL, -- сделок в ячейке за 12 мес
n_listings int NOT NULL, -- активных listings в ячейке
window_months int NOT NULL, -- окно сделок (12)
basis text NOT NULL, -- 'per_rooms_tier'|'per_rooms_all'|'global_all'
computed_at timestamptz NOT NULL DEFAULT now(),
PRIMARY KEY (rooms_bucket, district, ppm2_tier)
);
COMMENT ON TABLE asking_to_sold_ratios_tiered IS
'Per-rooms-tier asking->sold коэффициент (#928): ratio = median(SOLD)/median(ASKING) '
'внутри ppm2-тира бакета. ppm2_tier: low/mid/high (tier), all (per-rooms / global). '
'rooms_bucket 0=студия..4=4+; -1 = global fallback (ppm2_tier=''all'', basis=''global_all''). '
'Tier-строки только при n_deals>=30 AND n_listings>=30. '
'Параллельная к asking_to_sold_ratios: flag OFF читает OLD таблицу (fetchone-safety). '
'district RESERVED под #647 (geo), всегда ''''. '
'Refresh -- asking_to_sold_ratio_refresh атомарно DELETE WHERE district='''' + re-derive.';
-- Seed derivation ----------------------------------------------------------------
-- ppm2-полоса [30000, 1200000] -- settings-default (#767). Окно = трейлинг 12 мес.
-- ON CONFLICT DO UPDATE -- идемпотентный re-apply / dry-run.
-- ppm² band literals MUST equal _PPM2_MIN (30000) and settings.asking_ratio_ppm2_max
-- default (1_200_000) — see app/tasks/asking_to_sold_ratio.py. If that setting's DEFAULT
-- changes, update these literals (or fresh-seed vs daily-refresh diverge).
-- Step 1: bounds (ppm2-квантили по listings per бакет) ---------------------------
WITH bounds AS (
SELECT
LEAST(GREATEST(rooms, 0), 4) AS rooms_bucket,
percentile_cont(0.3333) WITHIN GROUP (ORDER BY price_per_m2)::bigint AS t33,
percentile_cont(0.6667) WITHIN GROUP (ORDER BY price_per_m2)::bigint AS t66,
COUNT(*)::int AS n_listings
FROM listings
WHERE is_active
AND rooms IS NOT NULL
AND price_per_m2 BETWEEN 30000 AND 1200000
GROUP BY LEAST(GREATEST(rooms, 0), 4)
HAVING COUNT(*) > 0
)
INSERT INTO asking_to_sold_tier_bounds (rooms_bucket, district, t33, t66, n_listings)
SELECT rooms_bucket, ''::text, t33, t66, n_listings FROM bounds
ON CONFLICT (rooms_bucket, district) DO UPDATE SET
t33 = EXCLUDED.t33,
t66 = EXCLUDED.t66,
n_listings = EXCLUDED.n_listings,
computed_at = now();
-- Step 2: tier rows (low/mid/high) + 'all' rows + global row --------------------
WITH
-- Bounds уже в таблице (Step 1). Читаем для join-а с listings и deals.
bnd AS (
SELECT rooms_bucket, t33, t66 FROM asking_to_sold_tier_bounds WHERE district = ''
),
deal_tiered AS (
SELECT
LEAST(GREATEST(d.rooms, 0), 4) AS rooms_bucket,
CASE
WHEN d.price_per_m2 < b.t33 THEN 'low'
WHEN d.price_per_m2 < b.t66 THEN 'mid'
ELSE 'high'
END AS ppm2_tier,
d.price_per_m2
FROM deals d
JOIN bnd b ON b.rooms_bucket = LEAST(GREATEST(d.rooms, 0), 4)
WHERE d.source = 'rosreestr'
AND d.rooms IS NOT NULL
AND d.price_per_m2 BETWEEN 30000 AND 1200000
AND d.deal_date >= CURRENT_DATE - INTERVAL '12 months'
),
ask_tiered AS (
SELECT
LEAST(GREATEST(l.rooms, 0), 4) AS rooms_bucket,
CASE
WHEN l.price_per_m2 < b.t33 THEN 'low'
WHEN l.price_per_m2 < b.t66 THEN 'mid'
ELSE 'high'
END AS ppm2_tier,
l.price_per_m2
FROM listings l
JOIN bnd b ON b.rooms_bucket = LEAST(GREATEST(l.rooms, 0), 4)
WHERE l.is_active
AND l.rooms IS NOT NULL
AND l.price_per_m2 BETWEEN 30000 AND 1200000
),
deal_side_tier AS (
SELECT
rooms_bucket,
ppm2_tier,
percentile_cont(0.5) WITHIN GROUP (ORDER BY price_per_m2) AS sold_median,
COUNT(*)::int AS n_deals
FROM deal_tiered
GROUP BY rooms_bucket, ppm2_tier
),
ask_side_tier AS (
SELECT
rooms_bucket,
ppm2_tier,
percentile_cont(0.5) WITHIN GROUP (ORDER BY price_per_m2) AS ask_median,
COUNT(*)::int AS n_listings
FROM ask_tiered
GROUP BY rooms_bucket, ppm2_tier
),
-- Tier строки -- только ячейки с n>=30 с обеих сторон.
tier_rows AS (
SELECT
d.rooms_bucket,
''::text AS district,
d.ppm2_tier,
(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,
a.n_listings,
12 AS window_months,
'per_rooms_tier'::text AS basis
FROM deal_side_tier d
JOIN ask_side_tier a USING (rooms_bucket, ppm2_tier)
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
AND d.sold_median > 0
),
-- 'all' строки: повторяет 080-логику с полосой [30000,1200000].
deal_side_all 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(*)::int AS n_deals
FROM deals
WHERE source = 'rosreestr'
AND rooms IS NOT NULL
AND price_per_m2 BETWEEN 30000 AND 1200000
AND deal_date >= CURRENT_DATE - INTERVAL '12 months'
GROUP BY LEAST(GREATEST(rooms, 0), 4)
),
ask_side_all 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(*)::int AS n_listings
FROM listings
WHERE is_active
AND rooms IS NOT NULL
AND price_per_m2 BETWEEN 30000 AND 1200000
GROUP BY LEAST(GREATEST(rooms, 0), 4)
),
all_rows AS (
SELECT
d.rooms_bucket,
''::text AS district,
'all'::text AS ppm2_tier,
(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,
a.n_listings,
12 AS window_months,
'per_rooms_all'::text AS basis
FROM deal_side_all d
JOIN ask_side_all 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
AND d.sold_median > 0
),
deal_global AS (
SELECT
percentile_cont(0.5) WITHIN GROUP (ORDER BY price_per_m2) AS sold_median,
COUNT(*)::int AS n_deals
FROM deals
WHERE source = 'rosreestr'
AND rooms IS NOT NULL
AND price_per_m2 BETWEEN 30000 AND 1200000
AND deal_date >= CURRENT_DATE - INTERVAL '12 months'
),
ask_global AS (
SELECT
percentile_cont(0.5) WITHIN GROUP (ORDER BY price_per_m2) AS ask_median,
COUNT(*)::int AS n_listings
FROM listings
WHERE is_active
AND rooms IS NOT NULL
AND price_per_m2 BETWEEN 30000 AND 1200000
),
global_row AS (
SELECT
-1 AS rooms_bucket,
''::text AS district,
'all'::text AS ppm2_tier,
(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,
a.n_listings,
12 AS window_months,
'global_all'::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
AND d.sold_median > 0
)
INSERT INTO asking_to_sold_ratios_tiered (
rooms_bucket, district, ppm2_tier, ratio, sold_median, ask_median,
n_deals, n_listings, window_months, basis
)
SELECT rooms_bucket, district, ppm2_tier, ratio, sold_median, ask_median,
n_deals, n_listings, window_months, basis FROM global_row
UNION ALL
SELECT rooms_bucket, district, ppm2_tier, ratio, sold_median, ask_median,
n_deals, n_listings, window_months, basis FROM all_rows
UNION ALL
SELECT rooms_bucket, district, ppm2_tier, ratio, sold_median, ask_median,
n_deals, n_listings, window_months, basis FROM tier_rows
ON CONFLICT (rooms_bucket, district, ppm2_tier) 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;

291
tradein-mvp/backend/tests/services/test_asking_sold_tier_ratio.py Normal file
View file

@ -0,0 +1,291 @@
"""Unit tests for ppm2-tier asking->sold ratio lookup (#928).
Tests for the _get_asking_sold_ratio helper with tier_aware_ratio_enabled flag.
Five scenarios:
1. flag OFF -> uses old asking_to_sold_ratios table (byte-identical to pre-#928).
2. flag ON, anchor in low band -> returns shrunk tier ratio, basis startswith 'per_rooms_tier'.
3. flag ON, no tier row for that cell -> fallback to 'all' (basis 'per_rooms_all').
4. flag ON, no 'all' row either -> global (-1) fallback (basis 'global_all').
5. flag ON, no tables / empty -> (None, None), no raise.
Static callers audit: _get_asking_sold_ratio gained anchor_ppm2 param (default None),
so all existing call sites using positional (db, rooms) continue to work unchanged.
"""
from __future__ import annotations
import os
from typing import Any
from unittest.mock import MagicMock, patch
# Settings requires DATABASE_URL at init time. Set dummy DSN before any app import.
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
class _FakeRow:
"""Stand-in for a SQLAlchemy Row."""
def __init__(self, **kw: Any) -> None:
for k, v in kw.items():
setattr(self, k, v)
def _clear_ratio_cache() -> None:
from app.services import estimator
estimator._asking_sold_ratio_cache.clear()
# ── helper: build a MagicMock db returning consecutive fetchone() results ─────
def _db_sequence(rows: list[Any]) -> MagicMock:
"""MagicMock db whose execute().fetchone() calls return `rows` in order."""
db = MagicMock()
db.execute.return_value.fetchone.side_effect = rows
return db
# ── 1. Flag OFF -> old table, ignores anchor_ppm2 ────────────────────────────
def test_flag_off_uses_old_table_ignores_anchor_ppm2() -> None:
"""Flag OFF: _get_asking_sold_ratio(db, 2, anchor_ppm2=90000) hits asking_to_sold_ratios."""
from app.core.config import settings
from app.services.estimator import _get_asking_sold_ratio
_clear_ratio_cache()
db = _db_sequence([_FakeRow(ratio=0.80, basis="per_rooms")])
with patch.object(settings, "tier_aware_ratio_enabled", False):
ratio, basis = _get_asking_sold_ratio(db, 2, anchor_ppm2=90_000.0)
assert ratio == 0.80
assert basis == "per_rooms"
# Only one DB query (old path, no tier tables touched)
assert db.execute.call_count == 1
# The query must target asking_to_sold_ratios (old table)
sql_str = str(db.execute.call_args_list[0].args[0])
assert "asking_to_sold_ratios" in sql_str
assert "tiered" not in sql_str
def test_flag_off_no_anchor_ppm2_still_works() -> None:
"""Flag OFF with no anchor_ppm2 (positional call) still works as before."""
from app.core.config import settings
from app.services.estimator import _get_asking_sold_ratio
_clear_ratio_cache()
db = _db_sequence([_FakeRow(ratio=0.82, basis="per_rooms")])
with patch.object(settings, "tier_aware_ratio_enabled", False):
ratio, basis = _get_asking_sold_ratio(db, 1)
assert ratio == 0.82
assert basis == "per_rooms"
def test_flag_off_cache_key_is_legacy_not_none() -> None:
"""Flag OFF must cache under (bucket, '_legacy'), NOT (bucket, None).
This prevents key collision with the flag-ON Step-3 'all' fallback which also
caches under (bucket, None). If the process flips tier_aware_ratio_enabled ON
without restart, the '_legacy' entry is not accidentally served as the 'all' row.
"""
from app.core.config import settings
from app.services import estimator
from app.services.estimator import _get_asking_sold_ratio
_clear_ratio_cache()
bucket = min(max(2, 0), 4)
db = _db_sequence([_FakeRow(ratio=0.77, basis="per_rooms")])
with patch.object(settings, "tier_aware_ratio_enabled", False):
_get_asking_sold_ratio(db, 2)
# Flag-OFF entry must be under "_legacy", not None.
assert (bucket, "_legacy") in estimator._asking_sold_ratio_cache
assert (bucket, None) not in estimator._asking_sold_ratio_cache
# ── 2. Flag ON, anchor in low band -> shrunk tier ratio ──────────────────────
def test_flag_on_low_band_returns_shrunk_tier_ratio() -> None:
"""Flag ON, anchor < t33 -> tier='low', shrunk = tier*w + all*(1-w)."""
from app.core.config import settings
from app.services.estimator import _get_asking_sold_ratio
_clear_ratio_cache()
# DB calls sequence:
# 1. bounds (t33=130000, t66=165000)
# 2. tier row (ratio=0.91, n_deals=200)
# 3. 'all' row (ratio=0.80)
db = _db_sequence(
[
_FakeRow(t33=130_000, t66=165_000), # bounds
_FakeRow(ratio=0.91, n_deals=200), # tier row
_FakeRow(ratio=0.80), # 'all' row
]
)
k = 150
w = 200 / (200 + k)
expected_ratio = 0.91 * w + 0.80 * (1.0 - w)
with (
patch.object(settings, "tier_aware_ratio_enabled", True),
patch.object(settings, "tier_ratio_shrink_k", k),
):
ratio, basis = _get_asking_sold_ratio(db, 2, anchor_ppm2=90_000.0)
assert ratio is not None
assert abs(ratio - expected_ratio) < 1e-9
assert basis is not None
assert basis.startswith("per_rooms_tier")
assert "low" in basis
# ── 3. Flag ON, no tier row -> fallback to 'all' ─────────────────────────────
def test_flag_on_no_tier_row_falls_back_to_all() -> None:
"""Flag ON, anchor in mid band, but no tier row for (bucket,mid) -> 'all' fallback."""
from app.core.config import settings
from app.services.estimator import _get_asking_sold_ratio
_clear_ratio_cache()
# DB calls:
# 1. bounds (t33=130000, t66=165000)
# 2. tier row -> None (cell not populated)
# 3. 'all' row (ratio=0.80) -- used as both shrink-target AND fallback
# Since tier_row is None, we skip shrink and fall through to step 3.
db = _db_sequence(
[
_FakeRow(t33=130_000, t66=165_000), # bounds
None, # tier row missing
_FakeRow(ratio=0.80), # 'all' row (shrink target query)
_FakeRow(ratio=0.80), # 'all' row (fallback step 3)
]
)
with (
patch.object(settings, "tier_aware_ratio_enabled", True),
patch.object(settings, "tier_ratio_shrink_k", 150),
):
ratio, basis = _get_asking_sold_ratio(db, 2, anchor_ppm2=145_000.0)
assert ratio == 0.80
assert basis == "per_rooms_all"
# ── 4. Flag ON, no 'all' row -> global fallback ───────────────────────────────
def test_flag_on_no_all_row_falls_back_to_global() -> None:
"""Flag ON, bounds present, tier+all rows missing -> global -1 fallback."""
from app.core.config import settings
from app.services.estimator import _get_asking_sold_ratio
_clear_ratio_cache()
db = _db_sequence(
[
_FakeRow(t33=130_000, t66=165_000), # bounds
None, # tier row -> None
None, # 'all' shrink-target -> None
None, # step-3 'all' fallback -> None
_FakeRow(ratio=0.79), # step-4 global (-1, all)
]
)
with patch.object(settings, "tier_aware_ratio_enabled", True):
ratio, basis = _get_asking_sold_ratio(db, 2, anchor_ppm2=145_000.0)
assert ratio == 0.79
assert basis == "global_all"
# ── 5. Flag ON, no tables / empty -> (None, None), no raise ──────────────────
def test_flag_on_no_tables_returns_none_none() -> None:
"""Flag ON, db.execute raises (tables not migrated) -> (None, None), no raise."""
from app.core.config import settings
from app.services.estimator import _get_asking_sold_ratio
_clear_ratio_cache()
db = MagicMock()
db.execute.side_effect = RuntimeError("relation asking_to_sold_tier_bounds does not exist")
with patch.object(settings, "tier_aware_ratio_enabled", True):
result = _get_asking_sold_ratio(db, 2, anchor_ppm2=90_000.0)
assert result == (None, None)
# rollback must be called on error
db.rollback.assert_called()
# ── 6. Flag ON, no bounds row -> step-3 'all' fallback (skip tier entirely) ──
def test_flag_on_no_bounds_row_falls_back_to_all() -> None:
"""Flag ON, bounds row None -> skip tier, go straight to 'all' step 3."""
from app.core.config import settings
from app.services.estimator import _get_asking_sold_ratio
_clear_ratio_cache()
db = _db_sequence(
[
None, # bounds -> None (bucket has no bound data)
_FakeRow(ratio=0.80), # step-3 'all' row
]
)
with patch.object(settings, "tier_aware_ratio_enabled", True):
ratio, basis = _get_asking_sold_ratio(db, 2, anchor_ppm2=90_000.0)
assert ratio == 0.80
assert basis == "per_rooms_all"
# ── 7. Cache: (bucket, tier) key memoises per tier ───────────────────────────
def test_tier_cache_key_is_bucket_tier_tuple() -> None:
"""Second call with same (bucket, anchor in same tier) -> served from cache after bounds.
Flow for second call:
- cache_key starts as (bucket, None) -- miss
- fetches bounds (1 DB query) -> determines tier_key="low"
- cache_key = (bucket, "low") -- HIT (populated by first call)
- returns cached ratio without further DB queries
Total DB queries: first call (3) + second call (1 bounds) = 4.
"""
from app.core.config import settings
from app.services.estimator import _get_asking_sold_ratio
_clear_ratio_cache()
db = _db_sequence(
[
# First call: bounds + tier_row + all_row
_FakeRow(t33=130_000, t66=165_000),
_FakeRow(ratio=0.91, n_deals=200),
_FakeRow(ratio=0.80),
# Second call: bounds only (tier cache hit after that)
_FakeRow(t33=130_000, t66=165_000),
]
)
with (
patch.object(settings, "tier_aware_ratio_enabled", True),
patch.object(settings, "tier_ratio_shrink_k", 150),
):
first = _get_asking_sold_ratio(db, 2, anchor_ppm2=90_000.0)
second = _get_asking_sold_ratio(db, 2, anchor_ppm2=95_000.0) # same 'low' tier
# Both calls return the same result (same tier).
assert first == second
# First call: 3 queries (bounds + tier_row + all_row).
# Second call: 1 query (bounds) + cache hit on (bucket, "low").
assert db.execute.call_count == 4

80
tradein-mvp/backend/tests/test_asking_to_sold_ratio.py
View file

@ -260,21 +260,39 @@ class _FakeMappingResult:
class _FakeDB:
"""Minimal stand-in for a SQLAlchemy Session — records execute() calls.
The third execute (counters query) returns the seeded counter row; the first two
(DELETE, INSERT) return an empty result.
Execute call order (#928 extended task):
1. DELETE asking_to_sold_ratios
2. REDERIVE INSERT asking_to_sold_ratios
3. COUNTERS SELECT (returns counters_row)
4. DELETE asking_to_sold_ratios_tiered (tiered rows)
5. DELETE asking_to_sold_tier_bounds (tier bounds)
6. REDERIVE_BOUNDS INSERT
7. REDERIVE_TIERED INSERT
8. TIERED_COUNTERS SELECT (returns tiered_counters_row)
"""
def __init__(self, counters_row: dict[str, Any]) -> None:
def __init__(
self,
counters_row: dict[str, Any],
tiered_counters_row: dict[str, Any] | None = None,
) -> None:
self._counters_row = counters_row
self._tiered_counters_row = tiered_counters_row or {
"tiered_rows_written": 0,
"tier_bounds_written": 0,
}
self.executed: list[Any] = []
self.committed = False
self.rolled_back = False
def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeMappingResult:
self.executed.append((stmt, params))
# 3rd call is the counters SELECT.
# 3rd call is the counters SELECT (asking_to_sold_ratios).
if len(self.executed) == 3:
return _FakeMappingResult(self._counters_row)
# 8th call is the tiered counters SELECT.
if len(self.executed) == 8:
return _FakeMappingResult(self._tiered_counters_row)
return _FakeMappingResult(None)
def commit(self) -> None:
@ -285,7 +303,11 @@ class _FakeDB:
def test_counter_logic_with_fake_db(monkeypatch: pytest.MonkeyPatch) -> None:
"""recompute_asking_to_sold_ratios maps the counters row into its return + marks done."""
"""recompute_asking_to_sold_ratios maps the counters row into its return + marks done.
#928: counters now include tiered_rows_written + tier_bounds_written (5 keys total).
FakeDB has 8 execute() calls (old: 3, tiered: 5).
"""
marked: dict[str, Any] = {}
monkeypatch.setattr(
ratio_mod.runs_mod,
@ -294,15 +316,27 @@ def test_counter_logic_with_fake_db(monkeypatch: pytest.MonkeyPatch) -> None:
)
monkeypatch.setattr(ratio_mod.runs_mod, "mark_failed", lambda *a, **k: None)
db = _FakeDB(counters_row={"rows_written": 4, "per_rooms_rows": 3, "used_global_fallback": 1})
db = _FakeDB(
counters_row={"rows_written": 4, "per_rooms_rows": 3, "used_global_fallback": 1},
tiered_counters_row={"tiered_rows_written": 15, "tier_bounds_written": 5},
)
out = ratio_mod.recompute_asking_to_sold_ratios(db, run_id=99) # type: ignore[arg-type]
assert out == {"rows_written": 4, "per_rooms_rows": 3, "used_global_fallback": 1}
expected = {
"rows_written": 4,
"per_rooms_rows": 3,
"used_global_fallback": 1,
"tiered_rows_written": 15,
"tier_bounds_written": 5,
}
assert out == expected
assert db.committed is True
# Three statements: DELETE, re-derive INSERT, counters SELECT.
assert len(db.executed) == 3
# Eight statements: DELETE(old) + REDERIVE(old) + COUNTERS(old)
# + DELETE_TIERED_ROWS + DELETE_TIER_BOUNDS + REDERIVE_BOUNDS + REDERIVE_TIERED
# + TIERED_COUNTERS (#928 extension).
assert len(db.executed) == 8
assert marked["run_id"] == 99
assert marked["counters"] == {"rows_written": 4, "per_rooms_rows": 3, "used_global_fallback": 1}
assert marked["counters"] == expected
def test_counter_logic_failure_path_marks_failed(monkeypatch: pytest.MonkeyPatch) -> None:
@ -324,3 +358,29 @@ def test_counter_logic_failure_path_marks_failed(monkeypatch: pytest.MonkeyPatch
ratio_mod.recompute_asking_to_sold_ratios(db, run_id=7) # type: ignore[arg-type]
assert db.rolled_back is True
assert failed["run_id"] == 7
# ── Migration 098 band / settings-default consistency ────────────────────────
def test_migration_098_band_matches_settings_default() -> None:
"""Migration 098 seed hardcodes ppm² band [30000, 1_200_000].
These literals MUST stay in sync with:
- app/tasks/asking_to_sold_ratio._PPM2_MIN (lower bound)
- app.core.config.Settings.asking_ratio_ppm2_max default (upper bound)
If either default changes, the one-time migration seed diverges from the
daily-refresh output — this test is the regression guard.
"""
from app.core.config import Settings
from app.tasks.asking_to_sold_ratio import _PPM2_MIN
# Lower bound — matches the 30000 literal in migration 098.
assert (
_PPM2_MIN == 30_000
), f"_PPM2_MIN changed ({_PPM2_MIN}); update migration 098 seed literals to match"
# Upper bound — matches the 1200000 literal in migration 098.
assert (
Settings().asking_ratio_ppm2_max == 1_200_000
), "asking_ratio_ppm2_max default changed; update migration 098 seed literals to match"