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

fix(tradein/location): заменить сломанный location-coef на калиброванный location index

Browse source 
coef = 0.95 + score/100*0.10 не был связан с ценой и не участвовал в расчёте
estimator'а вовсе (0 упоминаний), но интерфейс рисовал «база -> результат»,
обещая влияние на цену. Замеры на боевой БД: по 1500 адресам ЕКБ 67% попадают
в -1%..+1%, весь город укладывается в размах 1.10x; медиана руб/м2 по бакетам
коэффициента плоская и немонотонная (бакет -4% дороже бакета +3%). Для
сравнения, расстояние до центра даёт монотонный градиент с размахом 2.70x
(93 677 -> 249 686 руб/м2 по 31 тыс. лотов).

Новый показатель — отклонение медианы руб/м2 сопоставимых активных листингов
в радиусе от медианы по городу (percentile_cont, лестница радиусов до набора
выборки). Честная деградация вне ЕКБ и при малой выборке вместо молчаливого
вырождения в 0.95. В цену по-прежнему не идёт — аналоги берутся из того же
района, локация в базовой цене уже учтена.

Заодно две причины потери POI:
1. Overpass-фильтр метро ловил только station=subway, без railway=station+subway=yes.
2. v_tradein_osm_poi_ekb резала всё, что не редактировали в OSM 2 года. Проверено
   на проде: из 5133 точек доходило 2787 (-46%), причём смещённо — парки -84%,
   больницы -80%, магазины -83%, остановки 0%. Из 9 станций метро терялись 4,
   включая Площадь 1905 года. Станция не перестаёт существовать оттого, что её
   тег два года не трогали; Site Finder использует эту дату как мягкий сигнал
   уверенности, а не как фильтр существования.
This commit is contained in:
bot-backend 2026-07-26 23:09:43 +03:00
parent d56103219a
commit e7272183d4
11 changed files with 1329 additions and 812 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

135
backend/app/services/site_finder/poi_loader.py
View file

@ -19,60 +19,110 @@ logger = logging.getLogger(__name__)
OVERPASS_URL = "https://overpass-api.de/api/interpreter" OVERPASS_URL = "https://overpass-api.de/api/interpreter"
EKB_BBOX = (56.7, 60.5, 56.95, 60.75) # (south, west, north, east) EKB_BBOX = (56.7, 60.5, 56.95, 60.75) # (south, west, north, east)
# Маппинг OSM-тег → нормализованная category # Маппинг набора OSM-тегов (все теги в кортеже должны совпасть — AND) → нормализованная
OSM_CATEGORIES: dict[tuple[str, str], str] = { # category. Каждая запись — один per-category Overpass-запрос (см. _build_overpass_query);
# несколько записей с ОДИНАКОВЫМ значением category (как у metro_stop ниже) — это "ИЛИ" на
# уровне отдельных HTTP-запросов: элемент, подходящий под любую из альтернативных схем
# разметки, попадёт в категорию.
OSM_CATEGORIES: dict[tuple[tuple[str, str], ...], str] = {
# amenity tags — школы расширены (school/college/university) # amenity tags — школы расширены (school/college/university)
("amenity", "school"): "school", (("amenity", "school"),): "school",
("amenity", "college"): "school", (("amenity", "college"),): "school",
("amenity", "university"): "school", (("amenity", "university"),): "school",
("amenity", "kindergarten"): "kindergarten", (("amenity", "kindergarten"),): "kindergarten",
("amenity", "pharmacy"): "pharmacy", (("amenity", "pharmacy"),): "pharmacy",
("amenity", "hospital"): "hospital", (("amenity", "hospital"),): "hospital",
("amenity", "clinic"): "hospital", (("amenity", "clinic"),): "hospital",
# shop tags — supermarket расширен # shop tags — supermarket расширен
("shop", "mall"): "shop_mall", (("shop", "mall"),): "shop_mall",
("shop", "supermarket"): "shop_supermarket", (("shop", "supermarket"),): "shop_supermarket",
("shop", "hypermarket"): "shop_supermarket", (("shop", "hypermarket"),): "shop_supermarket",
("shop", "convenience"): "shop_small", (("shop", "convenience"),): "shop_small",
("shop", "bakery"): "shop_small", (("shop", "bakery"),): "shop_small",
# leisure # leisure
("leisure", "park"): "park", (("leisure", "park"),): "park",
# transit # transit
("railway", "tram_stop"): "tram_stop", (("railway", "tram_stop"),): "tram_stop",
("highway", "bus_stop"): "bus_stop", (("highway", "bus_stop"),): "bus_stop",
# метро (одна линия в ЕКБ, но добавляем для полноты) # Метро ЕКБ (9 станций, одна линия). Fix (location-index rework): фильтр раньше ловил
("station", "subway"): "metro_stop", # ТОЛЬКО station=subway и подтягивал лишь 5/9 станций — часть станций в OSM размечена
# без ключа "station" вовсе, комбинацией railway=station + subway=yes (альтернативная,
# но распространённая схема разметки метро). Обе схемы — отдельными записями ниже, чтобы
# не терять станции, размеченные любой из них.
(("station", "subway"),): "metro_stop",
(("railway", "station"), ("subway", "yes")): "metro_stop",
} }
def _build_overpass_query_single(key: str, value: str) -> str: def _build_overpass_query(tag_filters: tuple[tuple[str, str], ...]) -> str:
"""Запрос для одной пары tag → нормированной категории. """Запрос для ОДНОЙ комбинации tag=value (обычно один тег, иногда несколько — все AND).
Раньше делали один большой запрос на все 14 категорий — Overpass возвращал Раньше делали один большой запрос на все 14 категорий — Overpass возвращал
504 Gateway Timeout (запрос слишком тяжёлый). Сплит на per-category даёт 504 Gateway Timeout (запрос слишком тяжёлый). Сплит на per-category даёт
14 быстрых запросов вместо одного 60+ секундного. быстрые запросы вместо одного 60+ секундного.
""" """
south, west, north, east = EKB_BBOX south, west, north, east = EKB_BBOX
bbox = f"({south},{west},{north},{east})" bbox = f"({south},{west},{north},{east})"
return ( filt = "".join(f'["{k}"="{v}"]' for k, v in tag_filters)
f"[out:json][timeout:30];" return f"[out:json][timeout:30];(node{filt}{bbox};way{filt}{bbox};);out center meta;"
f'(node["{key}"="{value}"]{bbox};way["{key}"="{value}"]{bbox};);'
f"out center meta;"
)
def _classify(tags: dict[str, str]) -> str | None: def _classify(tags: dict[str, str]) -> str | None:
"""Определить category из OSM-тегов. None если не соответствует ни одной.""" """Определить category из OSM-тегов. None если не соответствует ни одной."""
for (k, v), cat in OSM_CATEGORIES.items(): for tag_filters, cat in OSM_CATEGORIES.items():
if tags.get(k) == v: if all(tags.get(k) == v for k, v in tag_filters):
return cat return cat
return None return None
def _tag_filters_desc(tag_filters: tuple[tuple[str, str], ...]) -> str:
return ",".join(f"{k}={v}" for k, v in tag_filters)
async def _fetch_category(
client: httpx.AsyncClient, tag_filters: tuple[tuple[str, str], ...], category: str
) -> list[dict]:
"""Один per-category Overpass-запрос с ОДНИМ повтором при транзиентной ошибке.
Fix (location-index rework, "не потерялись крупные категории"): раньше единственная
неудача (таймаут / 504) на всю неделю обнуляла категорию целиком (следующая попытка —
только на следующем weekly run). Один retry с паузой снимает большую часть транзиентных
сбоев без риска зациклиться (Overpass rate-limit — max 2 concurrent, поэтому не более
2 попыток на категорию).
"""
tag_desc = _tag_filters_desc(tag_filters)
query = _build_overpass_query(tag_filters)
for attempt in (1, 2):
try:
r = await client.post(OVERPASS_URL, data={"data": query})
r.raise_for_status()
elements: list[dict] = r.json().get("elements", [])
logger.info(
"Overpass: %s (%s) → %d [attempt %d]", tag_desc, category, len(elements), attempt
)
# Привязываем category именно к тому per-category запросу, под который
# элемент реально пришёл. Элемент с двумя целевыми тегами (например
# amenity=pharmacy + shop=supermarket) приходит дважды — каждая копия
# несёт свою category. Иначе _classify по dict-порядку молча терял бы
# вторую категорию при UPSERT по UNIQUE(osm_type, osm_id, category). См. #1372.
for el in elements:
el["_gd_category"] = category
return elements
except Exception as e:
if attempt == 1:
logger.warning("Overpass failed for %s (attempt 1, retrying): %s", tag_desc, e)
await asyncio.sleep(3.0)
continue
logger.warning(
"Overpass failed for %s after retry — category skipped this run: %s", tag_desc, e
)
return []
async def fetch_overpass() -> list[dict]: async def fetch_overpass() -> list[dict]:
"""Запросить Overpass API per category, вернуть combined список elements. """Запросить Overpass API per category, вернуть combined список elements.
Делаем 14 отдельных запросов вместо одного гигантского — большой запрос Делаем отдельные запросы вместо одного гигантского — большой запрос
отдаёт 504 Gateway Timeout. Между запросами sleep 1с (Overpass usage отдаёт 504 Gateway Timeout. Между запросами sleep 1с (Overpass usage
policy: max 2 concurrent, лучше 1 req/s). policy: max 2 concurrent, лучше 1 req/s).
@ -85,27 +135,14 @@ async def fetch_overpass() -> list[dict]:
} }
all_elements: list[dict] = [] all_elements: list[dict] = []
async with httpx.AsyncClient(timeout=60, headers=headers) as client: async with httpx.AsyncClient(timeout=60, headers=headers) as client:
for (key, value), category in OSM_CATEGORIES.items(): for tag_filters, category in OSM_CATEGORIES.items():
query = _build_overpass_query_single(key, value) elements = await _fetch_category(client, tag_filters, category)
try: all_elements.extend(elements)
r = await client.post(OVERPASS_URL, data={"data": query})
r.raise_for_status()
elements: list[dict] = r.json().get("elements", [])
logger.info("Overpass: %s=%s (%s) → %d", key, value, category, len(elements))
# Привязываем category именно к тому per-category запросу, под который
# элемент реально пришёл. Элемент с двумя целевыми тегами (например
# amenity=pharmacy + shop=supermarket) приходит дважды — каждая копия
# несёт свою category. Иначе _classify по dict-порядку молча терял бы
# вторую категорию при UPSERT по UNIQUE(osm_type, osm_id, category). См. #1372.
for el in elements:
el["_gd_category"] = category
all_elements.extend(elements)
except Exception as e:
# Не падаем на одной категории — логируем и продолжаем
logger.warning("Overpass failed for %s=%s: %s", key, value, e)
await asyncio.sleep(1.0) await asyncio.sleep(1.0)
logger.info( logger.info(
"Overpass: total %d elements across %d categories", len(all_elements), len(OSM_CATEGORIES) "Overpass: total %d elements across %d category-queries",
len(all_elements),
len(OSM_CATEGORIES),
) )
return all_elements return all_elements

48
data/sql/188_tradein_osm_poi_view_relax_freshness.sql Normal file
View file

@ -0,0 +1,48 @@
-- 188_tradein_osm_poi_view_relax_freshness.sql
-- Fix for 185_tradein_osm_poi_view.sql: v_tradein_osm_poi_ekb enforced a HARD 2-year
-- OSM last-edit-date filter that silently dropped legitimate, stable infrastructure.
--
-- CONTEXT (trade-in location-index rework, replaces the broken location-coef):
-- Audit of the trade-in POI mirror (osm_poi_ekb_local, fed by this view via the FDW
-- bridge) found only 5 of 9 EKB metro stations and just 2787 POI total reaching
-- tradein-mvp, despite osm_poi_ekb (this table, Site Finder's own registry) having more.
--
-- Root cause: this view's WHERE clause dropped any POI whose OSM `last_osm_edit_date` is
-- older than 2 years. A subway station node, once correctly mapped, is essentially never
-- re-edited in OSM — "stale last edit" here means "nobody touched this tag in years",
-- NOT "this station stopped existing". The same logic applies to schools/hospitals/parks:
-- physically permanent infrastructure that simply isn't re-edited often.
--
-- The "2-year freshness" requirement itself (see 82_osm_poi_ekb.sql, "требование
-- Максима") was intended as a SOFT confidence signal, not a hard existence filter — Site
-- Finder itself (backend/app/api/v1/parcels.py, "POI freshness" confidence subscore) only
-- uses last_osm_edit_date to DERATE a confidence score; every POI stays in the result set
-- regardless of staleness. This view diverged into a hard filter when the FDW bridge was
-- built (185) — this migration fixes that divergence, matching Site Finder's own intent.
--
-- WHAT: CREATE OR REPLACE VIEW, same 4-column shape as 185 (category, name, lat, lon) — no
-- WHERE clause. tradein-mvp's FDW foreign table (tradein-mvp/backend/data/sql/
-- 168_fdw_osm_poi_ekb.sql) is untouched — same column list, so no FDW-side change needed.
--
-- Idempotent: CREATE OR REPLACE VIEW. GRANT re-applied (idempotent, matches 185).
BEGIN;
CREATE OR REPLACE VIEW v_tradein_osm_poi_ekb AS
SELECT
category,
name,
lat,
lon
FROM osm_poi_ekb;
GRANT SELECT ON v_tradein_osm_poi_ekb TO tradein_fdw_reader;
COMMENT ON VIEW v_tradein_osm_poi_ekb IS
'FDW source for tradein-mvp (postgres_fdw) location-index/nearby-POI list (replaces the '
'broken location-coef, #2045). No freshness filter — last_osm_edit_date is a soft '
'confidence signal only (see Site Finder parcels.py), not evidence a POI stopped '
'existing. Fixes 185_tradein_osm_poi_view.sql hard 2-year WHERE filter that silently '
'dropped 4/9 EKB metro stations + other stable infrastructure from the trade-in mirror.';
COMMIT;

90
tradein-mvp/backend/app/api/v1/trade_in.py
View file

@ -28,8 +28,8 @@ from app.schemas.trade_in import (
HouseAnalyticsResponse, HouseAnalyticsResponse,
HouseInfoForEstimate, HouseInfoForEstimate,
IMVBenchmarkResponse, IMVBenchmarkResponse,
LocationCoefFactorOut, LocationIndexResponse,
LocationCoefResponse, NearbyPoiOut,
PhotoMeta, PhotoMeta,
PlacementHistoryEntry, PlacementHistoryEntry,
PriceHistoryYearPoint, PriceHistoryYearPoint,
@ -1567,39 +1567,43 @@ def get_estimate_imv_benchmark(
) )
# ── Location-coef POI scoring (#2045 BE-3, LocationDrawer) ──────────────────── # ── Location index (issue TBD, замена сломанного location-coef #2045) ────────
@router.get("/location-coef", response_model=LocationCoefResponse) @router.get("/location-index", response_model=LocationIndexResponse)
def get_location_coef( def get_location_index(
estimate_id: UUID, estimate_id: UUID,
db: Annotated[Session, Depends(get_db)], db: Annotated[Session, Depends(get_db)],
x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None, x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
radius_m: int | None = None, radius_m: int | None = None,
) -> LocationCoefResponse: ) -> LocationIndexResponse:
"""Location-coefficient POI-скоринг для оценки (#2045 BE-3, LocationDrawer). """Location index для оценки (замена сломанного location-coef, LocationDrawer).
Резолвит lat/lon/median_price оценки, считает coef через Резолвит lat/lon оценки, считает индекс через
app.services.location_coef.compute_location_coef — straight-line POI weighted score app.services.location_index.compute_location_index: % отклонения медианы ₽/м²
(портировано из Site Finder poi_score.py) поверх локального зеркала osm_poi_ekb_local, сопоставимых активных листингов в радиусе точки от медианы ₽/м² по всему Екатеринбургу
обновляемого scheduler'ом (source=osm_poi_ekb_refresh). result_price_rub = round( (percentile_cont(0.5) — устойчиво к выбросам). НЕ участвует в цене — estimator.py про
base_price_rub * coef). этот показатель не знает (аналоги уже несут локацию в базовой цене).
404 — оценки нет / IDOR (тот же _assert_estimate_access_by_id, что и у соседних 404 — оценки нет / IDOR (тот же _assert_estimate_access_by_id, что и у соседних
derived-роутов). radius_m опционален (None → DEFAULT_RADIUS_M=1200м, подобран для derived-роутов). radius_m опционален (None → адаптивная лестница радиусов
МКД, НЕ Ptica-дефолт 2000м для участков); явное значение клэмпится в [500, 3000]. RADIUS_LADDER_M, расширяется пока выборка не наберёт MIN_SAMPLE_SIZE); явное значение
клэмпится в [500, 3000] и используется РОВНО как задано (без расширения).
Graceful fallback (НЕ 500, НЕ сфабрикованные факторы): Честная деградация (НЕ 500, НЕ сфабрикованные значения) — см. LocationIndexResponse:
- osm_poi_ekb_local пуста/не отрефрешена на этом окружении → coef=1.0, factors=[], - status="out_of_coverage" — у оценки нет lat/lon, ИЛИ точка вне гео-охвата продукта
geo_source="unavailable" (см. compute_location_coef). (Екатеринбург).
- у оценки нет lat/lon (легаси/geo-fallback не сработал на POST) → тот же fallback. - status="insufficient_data" — даже на максимальном радиусе сопоставимых активных
листингов меньше порога.
- poi_status="unavailable" — osm_poi_ekb_local пуста/не отрефрешена (независимо от
status выше — «что рядом» и числовой индекс деградируют раздельно).
""" """
_assert_estimate_access_by_id(db, estimate_id, x_authenticated_user) _assert_estimate_access_by_id(db, estimate_id, x_authenticated_user)
row = db.execute( row = db.execute(
text( text(
""" """
SELECT lat, lon, median_price SELECT lat, lon
FROM trade_in_estimates FROM trade_in_estimates
WHERE id = CAST(:id AS uuid) WHERE id = CAST(:id AS uuid)
""" """
@ -1609,34 +1613,38 @@ def get_location_coef(
if row is None: if row is None:
raise HTTPException(status_code=404, detail="estimate not found") raise HTTPException(status_code=404, detail="estimate not found")
base_price_rub = int(row.median_price or 0)
if row.lat is None or row.lon is None: if row.lat is None or row.lon is None:
logger.info("location_coef: estimate=%s has no lat/lon — unavailable fallback", estimate_id) logger.info(
return LocationCoefResponse( "location_index: estimate=%s has no lat/lon — out_of_coverage fallback", estimate_id
coef=1.0, )
factors=[], return LocationIndexResponse(
geo_source="unavailable", status="out_of_coverage",
base_price_rub=base_price_rub, location_index_pct=None,
result_price_rub=base_price_rub, local_median_price_per_m2=None,
city_median_price_per_m2=None,
sample_size=0,
radius_m=radius_m or 0,
nearby_poi=[],
poi_status="unavailable",
) )
from app.services.location_coef import DEFAULT_RADIUS_M, compute_location_coef from app.services.location_index import compute_location_index
resolved_radius = DEFAULT_RADIUS_M if radius_m is None else max(500, min(radius_m, 3000)) resolved_radius = None if radius_m is None else max(500, min(radius_m, 3000))
result = compute_location_coef(db, float(row.lat), float(row.lon), radius_m=resolved_radius) result = compute_location_index(db, float(row.lat), float(row.lon), radius_m=resolved_radius)
return LocationCoefResponse( return LocationIndexResponse(
coef=result.coef, status=result.status,
factors=[ location_index_pct=result.location_index_pct,
LocationCoefFactorOut( local_median_price_per_m2=result.local_median_price_per_m2,
poi_type=f.poi_type, name=f.name, distance_m=f.distance_m, weight=f.weight city_median_price_per_m2=result.city_median_price_per_m2,
) sample_size=result.sample_size,
for f in result.factors radius_m=result.radius_m,
nearby_poi=[
NearbyPoiOut(poi_type=p.poi_type, name=p.name, distance_m=p.distance_m)
for p in result.nearby_poi
], ],
geo_source=result.geo_source, poi_status=result.poi_status,
base_price_rub=base_price_rub,
result_price_rub=round(base_price_rub * result.coef),
) )

49
tradein-mvp/backend/app/schemas/trade_in.py
View file

@ -598,29 +598,48 @@ class QuotaStatus(BaseModel):
unlimited: bool # True для admin / kopylov / без заголовка unlimited: bool # True для admin / kopylov / без заголовка
class LocationCoefFactorOut(BaseModel): class NearbyPoiOut(BaseModel):
"""Один POI-фактор в ответе GET /api/v1/trade-in/location-coef (#2045 BE-3).""" """Один пункт «что рядом» в ответе GET /api/v1/trade-in/location-index.
Качественная справка (школа 185 м, остановка 93 м) — НЕ участвует в location_index_pct.
"""
poi_type: str # категория POI (school/kindergarten/metro_stop/... — те же значения, poi_type: str # категория POI (school/kindergarten/metro_stop/... — те же значения,
# что в osm_poi_ekb на стороне gendesign) # что в osm_poi_ekb на стороне gendesign)
name: str | None name: str | None
distance_m: float distance_m: float
weight: float
class LocationCoefResponse(BaseModel): class LocationIndexResponse(BaseModel):
"""Ответ GET /api/v1/trade-in/location-coef (#2045 BE-3, LocationDrawer). """Ответ GET /api/v1/trade-in/location-index — замена сломанного location-coef.
coef — MVP-эвристика (НЕ откалибрована на реальных ценовых дельтах, см. ИСТОРИЯ: старый `location-coef` (`coef = 0.95 + score/100*0.10`, range [0.95,1.05],
app/services/location_coef.py::_score_to_coef), диапазон [0.95, 1.05]. `result_price_rub = round(base_price_rub * coef)`) не был откалиброван на ценах — 67% из
result_price_rub = round(base_price_rub * coef). 1500 адресов ЕКБ попадали в ±1%, а бакеты coef НЕ монотонны относительно медианы ₽/м² по
4000 активным лотам (дороже — не значит выше coef). Полностью заменён.
geo_source="unavailable" — osm_poi_ekb_local пуста/не отрефрешена на этом окружении location_index_pct — % отклонения медианы ₽/м² сопоставимых активных листингов в радиусе
(graceful fallback: coef=1.0, factors=[], НЕ 500 и НЕ сфабрикованные факторы). точки от медианы ₽/м² по всему Екатеринбургу (см. app/services/location_index.py). НЕ
зажат искусственно — диапазон реальный. НЕ участвует в цене (estimator.py про него не
знает: аналоги уже несут локацию в базовой цене, повторное умножение — double-count).
status:
- "ok" — location_index_pct/local_median_price_per_m2 надёжны.
- "out_of_coverage" — точка вне гео-охвата продукта (Екатеринбург). Все числовые
поля None — честный прочерк на фронте, НЕ 0%.
- "insufficient_data" — даже на максимальном радиусе сопоставимых активных листингов
меньше порога (см. MIN_SAMPLE_SIZE). Числовые поля None, но sample_size/radius_m
показывают, что реально нашлось.
poi_status — независимый статус для nearby_poi: "ok" | "unavailable" (osm_poi_ekb_local
ещё не отрефрешена на этом окружении — пустой список, НЕ сфабрикованные точки).
""" """
coef: float status: str
factors: list[LocationCoefFactorOut] location_index_pct: float | None
geo_source: str local_median_price_per_m2: int | None
base_price_rub: int city_median_price_per_m2: int | None
result_price_rub: int sample_size: int
radius_m: int
nearby_poi: list[NearbyPoiOut]
poi_status: str

224
tradein-mvp/backend/app/services/location_coef.py
View file

@ -1,224 +0,0 @@
"""Location-coefficient POI scoring for trade-in estimates (#2045 BE-3, LocationDrawer).
Ported straight-line formula from Site Finder (ПТИЦА)
`backend/app/services/site_finder/poi_score.py::compute_poi_weighted_top7`:
weight = (1 / (distance_m + 100)) * CATEGORY_WEIGHTS[category]
Reads from the LOCAL mirror table `osm_poi_ekb_local` (populated by
`app/tasks/osm_poi_ekb_refresh.py` from the `gendesign_osm_poi_ekb` FDW — see migrations
168-170). We deliberately do NOT query the FDW directly per-request: the same per-row cost
measured for the analogous cadastral-buildings FDW (~1.16s/row without a geom index on the
remote) would make a synchronous endpoint unusable.
Scope (MVP, #2045 BE-3):
- Straight-line distance only. The ORS routing-decay mode from Site Finder
(`compute_poi_routing_decay`) is NOT ported — no ORS infrastructure in trade-in, out of
MVP scope.
- Radius tuned for apartments (1000-1500m), NOT Site Finder's 2000m default for land parcels.
- The score→coef mapping (`_score_to_coef`) is a NEW MVP heuristic, not present in Site
Finder (there POI score is a ranking metric, not a price multiplier) — see its docstring.
"""
from __future__ import annotations
import logging
from typing import Any
from pydantic import BaseModel
from sqlalchemy import text
logger = logging.getLogger(__name__)
# Веса по категории — скопированы as-is из Site Finder CATEGORY_WEIGHTS
# (backend/app/services/site_finder/poi_score.py), чтобы ranking POI был согласован
# между продуктами.
CATEGORY_WEIGHTS: dict[str, float] = {
"metro_stop": 6.0,
"school": 5.0,
"kindergarten": 4.5,
"hospital": 4.0,
"shop_mall": 4.0,
"shop_supermarket": 3.5,
"bus_stop": 4.5,
"park": 3.5,
"pharmacy": 2.5,
"tram_stop": 2.0,
"shop_small": 2.0,
"default": 1.0,
}
# Радиус подобран для КВАРТИР (МКД), а не для участков (Ptica default 2000м) —
# пешая доступность в пределах квартала/микрорайона.
DEFAULT_RADIUS_M = 1200
DEFAULT_TOP_N = 7
# Теоретический максимум суммы весов top-7 POI.
#
# ИСТОРИЯ (audit R2 #7a): раньше максимум считался при d=0 (все top-7 категорий прямо у
# порога) — недостижимо на практике. Аудит показал, что даже сильная центральная локация в
# Екб (транспорт/садик/супермаркет рядом, школа/метро/ТЦ уже заметно дальше — типичный
# профиль квартала, реалистичные дистанции по категориям ~80-700м) даёт raw_sum лишь
# ~0.08-0.095, т.е. score ~25-30/100 при старой нормировке (/100 при d=0) → coef ~0.975-0.98.
# Сильная локация систематически читалась как "немного снижает стоимость" — знаменатель
# сжимал реалистичные scores в нижнюю треть шкалы 0-100.
#
# ФИКС: нормируем не на d=0, а на РЕФЕРЕНСНОЕ расстояние _REF_DISTANCE_M — top-7 весов,
# как если бы каждая категория была на этом расстоянии. REF=100м ровно удваивает
# знаменатель (distance_m + 100): 100+100=200 вместо 0+100=100 → теоретический максимум
# вдвое ниже прежнего → реалистичные scores вдвое выше: сильный центр ~25-30 → ~50-60
# (coef ~1.00-1.01), с запасом выше 1.0 для исключительно плотных локаций (все top-7
# категорий <150м). Слабые/разреженные локации остаются далеко ниже 1.0; пустая
# osm_poi_ekb_local (рефреш не запускался) по-прежнему даёт coef=1.0 через отдельный
# graceful-fallback путь в compute_location_coef, эту нормировку не трогающий.
#
# ВНИМАНИЕ: REF_DIST — калибровочная константа для MVP-эвристики (см. _score_to_coef),
# НЕ откалибрована на реальных ценовых дельтах — только на распределении реалистичных POI-
# профилей, чтобы диапазон [0.95, 1.05] соответствовал интуиции "сильный центр ≈ 1.0".
#
# Top-7 категорий по убыванию веса: 6.0+5.0+4.5+4.5+4.0+4.0+3.5 = 31.5 (тот же набор, что у Ptica).
_TOP7_WEIGHT_SUM: float = sum(sorted(CATEGORY_WEIGHTS.values(), reverse=True)[:7])
_REF_DISTANCE_M: float = 100.0
_MAX_STRAIGHT_SCORE: float = _TOP7_WEIGHT_SUM / (
_REF_DISTANCE_M + 100.0
) # ≈ 0.1575 (было 0.315 при d=0)
# coef диапазон ±5% вокруг 1.0 — heuristic v1, НЕ откалибровано на реальных ценовых дельтах
# (в отличие от Ptica, где poi_weighted_score — ранжирующая метрика, не ценовой множитель).
_COEF_BASE = 0.95
_COEF_SPREAD = 0.10
def _category_weight(category: str | None) -> float:
"""Вернуть вес категории. Если не знаем — default."""
return CATEGORY_WEIGHTS.get(category or "default", CATEGORY_WEIGHTS["default"])
class LocationCoefFactor(BaseModel):
"""Один POI-фактор в ответе location-coef."""
poi_type: str
name: str | None
distance_m: float
weight: float
class LocationCoefResult(BaseModel):
"""Результат compute_location_coef — потребляется эндпоинтом location-coef."""
coef: float
factors: list[LocationCoefFactor]
geo_source: str # "osm_poi_ekb" (норма) | "unavailable" (mirror пуста/не отрефрешена)
def _score_to_coef(poi_weighted_score: float) -> float:
"""MVP-эвристика score(0..100) → ценовой коэффициент.
coef = 0.95 + (score/100) * 0.10 → диапазон [0.95, 1.05].
ВНИМАНИЕ: это НЕ откалиброванная на реальных ценовых дельтах формула — первая рабочая
эвристика для MVP location-coef. Site Finder использует ту же POI-модель как ранжирующую
метрику (poi_weighted_score), а не как прямой ценовой множитель; здесь смысл другой,
поэтому маппинг введён отдельно и явно помечен как heuristic v1.
"""
return round(_COEF_BASE + (poi_weighted_score / 100.0) * _COEF_SPREAD, 4)
_NEAREST_POI_SQL = text(
"""
SELECT
p.name,
p.category,
CAST(
ST_Distance(
p.geom::geography,
ST_SetSRID(ST_MakePoint(:lon, :lat), 4326)::geography
) AS double precision
) AS distance_m
FROM osm_poi_ekb_local p
WHERE p.geom IS NOT NULL
AND ST_DWithin(
p.geom::geography,
ST_SetSRID(ST_MakePoint(:lon, :lat), 4326)::geography,
CAST(:radius_m AS double precision)
)
ORDER BY distance_m ASC
LIMIT :limit
"""
)
def compute_location_coef(
db: Any,
lat: float,
lon: float,
radius_m: int = DEFAULT_RADIUS_M,
top_n: int = DEFAULT_TOP_N,
) -> LocationCoefResult:
"""Посчитать location-coef для координат (lat, lon) по POI из osm_poi_ekb_local.
Graceful fallback: если osm_poi_ekb_local пуста (рефреш ещё не запускался на этом
окружении) — возвращает coef=1.0, factors=[], geo_source="unavailable" вместо 500 или
сфабрикованных факторов. Отсутствие POI В РАДИУСЕ у непустой таблицы — это легитимный
результат (coef=0.95, factors=[], geo_source="osm_poi_ekb"), не fallback.
Args:
db: SQLAlchemy Session.
lat: широта целевой квартиры.
lon: долгота целевой квартиры.
radius_m: радиус поиска в метрах (default 1200 — подобран для МКД, не для участков).
top_n: количество POI, учитываемых в score (default 7).
"""
total = db.execute(text("SELECT count(*) FROM osm_poi_ekb_local")).scalar() or 0
if total == 0:
logger.warning(
"location_coef: osm_poi_ekb_local is empty (refresh job not yet run on this "
"environment) — returning unavailable fallback, no fabricated factors"
)
return LocationCoefResult(coef=1.0, factors=[], geo_source="unavailable")
rows = (
db.execute(
_NEAREST_POI_SQL,
{"lat": lat, "lon": lon, "radius_m": radius_m, "limit": top_n * 10},
)
.mappings()
.all()
)
scored: list[tuple[float, LocationCoefFactor]] = []
for row in rows:
distance_m = float(row["distance_m"])
category = row["category"] or "default"
weight = (1.0 / (distance_m + 100.0)) * _category_weight(category)
scored.append(
(
weight,
LocationCoefFactor(
poi_type=category,
name=row["name"],
distance_m=round(distance_m, 1),
weight=round(weight, 6),
),
)
)
scored.sort(key=lambda pair: pair[0], reverse=True)
top_factors = [factor for _weight, factor in scored[:top_n]]
raw_sum = sum(factor.weight for factor in top_factors)
poi_weighted_score = min(100.0, (raw_sum / _MAX_STRAIGHT_SCORE) * 100.0)
coef = _score_to_coef(poi_weighted_score)
logger.debug(
"location_coef: lat=%.5f lon=%.5f radius=%dm poi_found=%d top=%d " "score=%.1f coef=%.4f",
lat,
lon,
radius_m,
len(rows),
len(top_factors),
poi_weighted_score,
coef,
)
return LocationCoefResult(coef=coef, factors=top_factors, geo_source="osm_poi_ekb")

436
tradein-mvp/backend/app/services/location_index.py Normal file
View file

@ -0,0 +1,436 @@
"""Location index for trade-in estimates — replaces the broken `location_coef` (LocationDrawer).
ИСТОРИЯ / ПОЧЕМУ ПЕРЕПИСАНО:
Старый `location_coef.py` считал `coef = 0.95 + (poi_weighted_score/100) * 0.10` — диапазон
жёстко зажат в [0.95, 1.05], без какой-либо калибровки на реальных ценах. Аудит на боевой БД
(1500 адресов ЕКБ + 4000 активных лотов) показал:
- 67% адресов попадали в −1%…+1%, у ~25% coef был РОВНО 1.0 (score=50) — почти
неинформативно, весь город умещался в −4%…+5%;
- связи с ценой не было вообще: медиана ₽/м² по бакетам coef плоская и НЕ монотонна
(бакет −4% дороже бакета +3%).
Для сравнения, расстояние до центра ЕКБ на 31 тыс. лотов даёт чистый монотонный градиент
(0-2км 249 686 ₽/м² → 12-13км 93 677 ₽/м², разброс 2.7×) — сигнал в данных есть, просто
POI-score его не улавливал (POI ranking ≠ цена).
НОВЫЙ ПОКАЗАТЕЛЬ (location index):
location_index_pct = (медиана ₽/м² сопоставимых активных листингов в радиусе точки −
медиана ₽/м² по всему ЕКБ) / медиана по ЕКБ * 100
Самообновляем (те же `listings`, что уже скрейпятся под estimator), интерпретируем напрямую
("район на N% дороже/дешевле среднего по городу"), устойчив к выбросам (percentile_cont(0.5) —
медиана самой природой игнорирует единичные экстремумы, в отличие от mean/min/max), и НЕ зажат
искусственно — если район реально на 40% дороже, так и покажет.
ЧЕСТНАЯ ДЕГРАДАЦИЯ (см. LocationIndexResult.status):
- "out_of_coverage" — точка вне гео-охвата продукта (bbox Екатеринбурга). НЕ 0%, НЕ
fallback-число — прочерк на фронте.
- "insufficient_data" — даже на максимальном радиусе выборки < MIN_SAMPLE_SIZE сопоставимых
активных листингов. Тоже прочерк, а не шум по 3 объявлениям.
- "ok" — index надёжен.
В ЦЕНУ НЕ ИДЁТ: estimator.py про этот модуль не знает и не должен знать — аналоги уже берутся
из того же района (локация учтена в базовой цене через сам подбор сопоставимых объектов),
повторное умножение на локационный индекс было бы двойным учётом одного и того же эффекта.
POI («что рядом» — школа/метро/остановка) сохранены как ОТДЕЛЬНАЯ качественная справка
(`nearby_poi`, ранжирование как раньше в location_coef.py), но больше не участвуют в числовом
показателе.
"""
from __future__ import annotations
import logging
from typing import Any
from pydantic import BaseModel
from sqlalchemy import text
logger = logging.getLogger(__name__)
# ── Гео-охват продукта: только Екатеринбург ──────────────────────────────────
# Тот же bbox, что EKB_BBOX в backend/app/services/site_finder/poi_loader.py (main
# gendesign backend, Overpass-загрузчик osm_poi_ekb) и что использовался при аудите
# (1500 адресов / 4000 активных лотов / 2787 POI, все — "по Екатеринбургу"). tradein-mvp —
# отдельный деплой/venv от backend/, поэтому константа продублирована, не импортирована;
# при изменении bbox в одном месте — проверить и второе (комментарий в обе стороны).
_EKB_BBOX_SOUTH = 56.70
_EKB_BBOX_WEST = 60.50
_EKB_BBOX_NORTH = 56.95
_EKB_BBOX_EAST = 60.75
def _in_ekb_bbox(lat: float, lon: float) -> bool:
"""True если точка внутри гео-охвата продукта (Екатеринбург)."""
return _EKB_BBOX_SOUTH <= lat <= _EKB_BBOX_NORTH and _EKB_BBOX_WEST <= lon <= _EKB_BBOX_EAST
# ── Калибровочные константы (радиус / минимальная выборка) ──────────────────
#
# Плотность-прикидка для обоснования порядка величины (НЕ подтверждено живым запросом к
# прод-БД в этом изменении — см. PR description "непроверенное"): ЕКБ-аудит насчитал ~4000
# активных лотов в bbox площадью ~ 27.8км (0.25° широты) × 15.3км (0.25° долготы на широте
# 56.8°) ≈ 425 км² → плотность ~9.4 лота/км². Круг радиусом 800м имеет площадь ~2.01 км² →
# ожидаемо ~19 лотов при равномерной плотности — близко к MIN_SAMPLE_SIZE=20, т.е. стартовый
# радиус разумен для "средней" точки. Плотность в городе крайне неравномерна (центр много
# гуще окраин) — поэтому лестница радиусов расширяется, а не фиксированный радиус.
RADIUS_LADDER_M: tuple[int, ...] = (800, 1500, 2500)
# Ниже этого числа сопоставимых активных листингов медиана — шум, не показатель.
# Порог не откалиброван статистически (например через доверительный интервал медианы) —
# первая рабочая оценка для MVP. TODO: перепроверить на реальном распределении выборок по
# районам ЕКБ (см. "непроверенное" в отчёте задачи).
MIN_SAMPLE_SIZE = 20
# Санитарные (НЕ бизнес-калибровочные) границы ₽/м² — отсекают заведомо битые скрейп-строки
# (парсинг ошибся на порядок и т.п.), не сужают реальный рынок ЕКБ (там диапазон примерно
# 40-400 тыс₽/м², с большим запасом по краям).
_PRICE_PER_M2_SANITY_MIN = 20_000
_PRICE_PER_M2_SANITY_MAX = 1_000_000
DEFAULT_POI_RADIUS_M = 1200 # как в старом location_coef.py — подобран для МКД
DEFAULT_POI_TOP_N = 7
# Веса по категории POI — те же, что были в location_coef.py (ranking "что рядом",
# больше НЕ конвертируются в число, влияющее на индекс).
CATEGORY_WEIGHTS: dict[str, float] = {
"metro_stop": 6.0,
"school": 5.0,
"kindergarten": 4.5,
"hospital": 4.0,
"shop_mall": 4.0,
"shop_supermarket": 3.5,
"bus_stop": 4.5,
"park": 3.5,
"pharmacy": 2.5,
"tram_stop": 2.0,
"shop_small": 2.0,
"default": 1.0,
}
def _category_weight(category: str | None) -> float:
"""Вернуть вес категории. Если не знаем — default."""
return CATEGORY_WEIGHTS.get(category or "default", CATEGORY_WEIGHTS["default"])
class NearbyPoi(BaseModel):
"""Один пункт «что рядом» — качественная справка, НЕ участвует в location_index_pct."""
poi_type: str
name: str | None
distance_m: float
class LocationIndexResult(BaseModel):
"""Результат compute_location_index — потребляется эндпоинтом location-index."""
status: str # "ok" | "out_of_coverage" | "insufficient_data"
location_index_pct: float | None
local_median_price_per_m2: int | None
city_median_price_per_m2: int | None
sample_size: int
radius_m: int
nearby_poi: list[NearbyPoi]
poi_status: str # "ok" | "unavailable" (osm_poi_ekb_local пуста/не отрефрешена)
def _pct_deviation(local_median_ppm2: float, city_median_ppm2: float) -> float:
"""% отклонения локальной медианы от городской.
Округление до 1 знака — не создаёт ложной точности (исходные данные — шумные скрейп-цены).
"""
if city_median_ppm2 <= 0:
# Защита от деления на ноль при вырожденной городской выборке — не должно
# случаться в проде (там ~4000 активных лотов), только в пустой dev-БД.
return 0.0
return round((local_median_ppm2 - city_median_ppm2) / city_median_ppm2 * 100.0, 1)
# ── SQL: медиана ₽/м² сопоставимых активных листингов ────────────────────────
#
# percentile_cont(0.5) — тот же идиом, что уже используется в estimator.py для медианных
# ₽/м² трендов (_fetch_price_trend) — устойчив к выбросам В ОТЛИЧИЕ от AVG/min/max: единичный
# аномально дорогой/дешёвый лот не сдвигает медиану заметно.
#
# geo_precision IS DISTINCT FROM 'city' — тот же фильтр, что в estimator.py (#769 Part E):
# исключает листинги с геокодом до центра города (city-centroid fallback без номера дома),
# которые иначе "подмешивались" бы в любой радиус вокруг центра.
#
# price_per_m2 BETWEEN sanity-границы — не бизнес-калибровка, а защита от битых строк
# (см. _PRICE_PER_M2_SANITY_MIN/MAX выше).
#
# bbox-фильтр (lat/lon) — сопоставимые листинги считаются ТОЛЬКО по Екатеринбургу, даже если
# сам продукт уже скрейпит соседние города области (city-sweep): географию location_index
# явно ограничил владелец продукта.
_MEDIAN_PPM2_LOCAL_SQL = text(
"""
SELECT
CAST(percentile_cont(0.5) WITHIN GROUP (ORDER BY price_per_m2) AS double precision)
AS median_ppm2,
count(*) AS n
FROM listings
WHERE is_active = true
AND price_per_m2 IS NOT NULL
AND price_per_m2 BETWEEN CAST(:price_min AS integer) AND CAST(:price_max AS integer)
AND (geo_precision IS DISTINCT FROM 'city')
AND lat BETWEEN CAST(:bbox_south AS double precision)
AND CAST(:bbox_north AS double precision)
AND lon BETWEEN CAST(:bbox_west AS double precision)
AND CAST(:bbox_east AS double precision)
AND ST_DWithin(
geom::geography,
ST_SetSRID(ST_MakePoint(:lon, :lat), 4326)::geography,
CAST(:radius_m AS double precision)
)
"""
)
_MEDIAN_PPM2_CITYWIDE_SQL = text(
"""
SELECT
CAST(percentile_cont(0.5) WITHIN GROUP (ORDER BY price_per_m2) AS double precision)
AS median_ppm2,
count(*) AS n
FROM listings
WHERE is_active = true
AND price_per_m2 IS NOT NULL
AND price_per_m2 BETWEEN CAST(:price_min AS integer) AND CAST(:price_max AS integer)
AND (geo_precision IS DISTINCT FROM 'city')
AND lat BETWEEN CAST(:bbox_south AS double precision)
AND CAST(:bbox_north AS double precision)
AND lon BETWEEN CAST(:bbox_west AS double precision)
AND CAST(:bbox_east AS double precision)
"""
)
_NEAREST_POI_SQL = text(
"""
SELECT
p.name,
p.category,
CAST(
ST_Distance(
p.geom::geography,
ST_SetSRID(ST_MakePoint(:lon, :lat), 4326)::geography
) AS double precision
) AS distance_m
FROM osm_poi_ekb_local p
WHERE p.geom IS NOT NULL
AND ST_DWithin(
p.geom::geography,
ST_SetSRID(ST_MakePoint(:lon, :lat), 4326)::geography,
CAST(:radius_m AS double precision)
)
ORDER BY distance_m ASC
LIMIT :limit
"""
)
def _local_median_ppm2(db: Any, lat: float, lon: float, radius_m: int) -> tuple[float | None, int]:
row = (
db.execute(
_MEDIAN_PPM2_LOCAL_SQL,
{
"lat": lat,
"lon": lon,
"radius_m": radius_m,
"price_min": _PRICE_PER_M2_SANITY_MIN,
"price_max": _PRICE_PER_M2_SANITY_MAX,
"bbox_south": _EKB_BBOX_SOUTH,
"bbox_north": _EKB_BBOX_NORTH,
"bbox_west": _EKB_BBOX_WEST,
"bbox_east": _EKB_BBOX_EAST,
},
)
.mappings()
.first()
)
if row is None:
return None, 0
median = row["median_ppm2"]
return (float(median) if median is not None else None), int(row["n"] or 0)
def _citywide_median_ppm2(db: Any) -> tuple[float | None, int]:
row = (
db.execute(
_MEDIAN_PPM2_CITYWIDE_SQL,
{
"price_min": _PRICE_PER_M2_SANITY_MIN,
"price_max": _PRICE_PER_M2_SANITY_MAX,
"bbox_south": _EKB_BBOX_SOUTH,
"bbox_north": _EKB_BBOX_NORTH,
"bbox_west": _EKB_BBOX_WEST,
"bbox_east": _EKB_BBOX_EAST,
},
)
.mappings()
.first()
)
if row is None:
return None, 0
median = row["median_ppm2"]
return (float(median) if median is not None else None), int(row["n"] or 0)
def _fetch_nearby_poi(
db: Any, lat: float, lon: float, radius_m: int, top_n: int
) -> tuple[list[NearbyPoi], str]:
"""Top-N POI поблизости — качественная справка «что рядом», не числовой показатель.
Graceful fallback: osm_poi_ekb_local пуста (рефреш ещё не запускался на этом окружении)
→ ([], "unavailable") вместо 500 или сфабрикованного списка.
"""
total = db.execute(text("SELECT count(*) FROM osm_poi_ekb_local")).scalar() or 0
if total == 0:
logger.warning(
"location_index: osm_poi_ekb_local is empty (refresh job not yet run on this "
"environment) — nearby_poi unavailable, no fabricated factors"
)
return [], "unavailable"
rows = (
db.execute(
_NEAREST_POI_SQL,
{"lat": lat, "lon": lon, "radius_m": radius_m, "limit": top_n * 10},
)
.mappings()
.all()
)
ranked: list[tuple[float, NearbyPoi]] = []
for row in rows:
distance_m = float(row["distance_m"])
category = row["category"] or "default"
weight = (1.0 / (distance_m + 100.0)) * _category_weight(category)
ranked.append(
(
weight,
NearbyPoi(poi_type=category, name=row["name"], distance_m=round(distance_m, 1)),
)
)
ranked.sort(key=lambda pair: pair[0], reverse=True)
return [poi for _weight, poi in ranked[:top_n]], "ok"
def compute_location_index(
db: Any,
lat: float,
lon: float,
*,
radius_m: int | None = None,
poi_radius_m: int = DEFAULT_POI_RADIUS_M,
poi_top_n: int = DEFAULT_POI_TOP_N,
) -> LocationIndexResult:
"""Посчитать location index для координат (lat, lon).
location_index_pct = (медиана ₽/м² сопоставимых активных листингов в радиусе точки −
медиана ₽/м² по всему ЕКБ) / медиана по ЕКБ * 100. Радиус — лестница RADIUS_LADDER_M
(расширяется, пока выборка не наберёт MIN_SAMPLE_SIZE), если явный radius_m не передан
(тогда используется РОВНО он, без расширения — для отладки/тестов).
Args:
db: SQLAlchemy Session.
lat: широта целевой точки.
lon: долгота целевой точки.
radius_m: явный радиус в метрах — если задан, лестница не используется.
poi_radius_m: радиус для качественного списка «что рядом» (независим от индекса).
poi_top_n: сколько POI показать в «что рядом».
Returns:
LocationIndexResult со status:
- "out_of_coverage" — точка вне bbox Екатеринбурга, ничего не считаем.
- "insufficient_data" — даже на максимальном радиусе сопоставимых листингов
меньше MIN_SAMPLE_SIZE (или городская выборка-эталон сама вырождена).
- "ok" — location_index_pct надёжен.
"""
if not _in_ekb_bbox(lat, lon):
logger.info(
"location_index: lat=%.5f lon=%.5f outside EKB coverage bbox — out_of_coverage",
lat,
lon,
)
return LocationIndexResult(
status="out_of_coverage",
location_index_pct=None,
local_median_price_per_m2=None,
city_median_price_per_m2=None,
sample_size=0,
radius_m=radius_m or RADIUS_LADDER_M[0],
nearby_poi=[],
poi_status="unavailable",
)
nearby_poi, poi_status = _fetch_nearby_poi(db, lat, lon, poi_radius_m, poi_top_n)
city_median, city_n = _citywide_median_ppm2(db)
if city_median is None or city_n < MIN_SAMPLE_SIZE:
logger.warning(
"location_index: citywide reference sample too small (n=%d) — insufficient_data",
city_n,
)
return LocationIndexResult(
status="insufficient_data",
location_index_pct=None,
local_median_price_per_m2=None,
city_median_price_per_m2=(round(city_median) if city_median is not None else None),
sample_size=city_n,
radius_m=radius_m or RADIUS_LADDER_M[-1],
nearby_poi=nearby_poi,
poi_status=poi_status,
)
radii = [radius_m] if radius_m is not None else list(RADIUS_LADDER_M)
local_median: float | None = None
sample_size = 0
used_radius = radii[-1]
for r in radii:
local_median, sample_size = _local_median_ppm2(db, lat, lon, r)
used_radius = r
if sample_size >= MIN_SAMPLE_SIZE:
break
if local_median is None or sample_size < MIN_SAMPLE_SIZE:
logger.info(
"location_index: lat=%.5f lon=%.5f sample=%d < MIN_SAMPLE_SIZE=%d up to "
"radius=%dm — insufficient_data",
lat,
lon,
sample_size,
MIN_SAMPLE_SIZE,
used_radius,
)
return LocationIndexResult(
status="insufficient_data",
location_index_pct=None,
local_median_price_per_m2=None,
city_median_price_per_m2=round(city_median),
sample_size=sample_size,
radius_m=used_radius,
nearby_poi=nearby_poi,
poi_status=poi_status,
)
pct = _pct_deviation(local_median, city_median)
logger.debug(
"location_index: lat=%.5f lon=%.5f radius=%dm n=%d local=%d city=%d pct=%.1f",
lat,
lon,
used_radius,
sample_size,
round(local_median),
round(city_median),
pct,
)
return LocationIndexResult(
status="ok",
location_index_pct=pct,
local_median_price_per_m2=round(local_median),
city_median_price_per_m2=round(city_median),
sample_size=sample_size,
radius_m=used_radius,
nearby_poi=nearby_poi,
poi_status=poi_status,
)

8
tradein-mvp/backend/app/tasks/osm_poi_ekb_refresh.py
View file

@ -1,4 +1,4 @@
"""OSM POI local-mirror refresh (#2045 BE-3, LocationDrawer location-coef). """OSM POI local-mirror refresh (#2045 BE-3, LocationDrawer location-index).
Populates `osm_poi_ekb_local` (empty at deploy, migration 169) via a single bulk scan of the Populates `osm_poi_ekb_local` (empty at deploy, migration 169) via a single bulk scan of the
`gendesign_osm_poi_ekb` FDW foreign table (migration 168 — live view of gendesign's `gendesign_osm_poi_ekb` FDW foreign table (migration 168 — live view of gendesign's
@ -8,9 +8,11 @@ of #2045, already merged + deployed on gendesign).
WHY a local mirror (perf fact measured for the analogous gendesign_cad_buildings FDW — see WHY a local mirror (perf fact measured for the analogous gendesign_cad_buildings FDW — see
`app/tasks/cadastral_geo_match.py`): a per-request FDW nearest-POI query pays a per-row FDW `app/tasks/cadastral_geo_match.py`): a per-request FDW nearest-POI query pays a per-row FDW
round-trip (~1.16s/row without a geom index on the remote table) — UNUSABLE for a synchronous round-trip (~1.16s/row without a geom index on the remote table) — UNUSABLE for a synchronous
endpoint (`GET /api/v1/trade-in/location-coef`). We materialize the FDW once (single bulk endpoint (`GET /api/v1/trade-in/location-index`). We materialize the FDW once (single bulk
scan) into `osm_poi_ekb_local` with a real Point geom + GIST index, then scan) into `osm_poi_ekb_local` with a real Point geom + GIST index, then
`app/services/location_coef.py` runs fast LOCAL ST_DWithin/ST_Distance queries per estimate. `app/services/location_index.py` runs fast LOCAL ST_DWithin/ST_Distance queries per estimate
(nearby-POI qualitative list only — the numeric index itself comes from `listings`, not POI;
see that module's docstring for the location-coef → location-index rewrite history).
Scheduler source='osm_poi_ekb_refresh' (daily — OSM POI data changes rarely). Pure internal Scheduler source='osm_poi_ekb_refresh' (daily — OSM POI data changes rarely). Pure internal
DB op — one FDW read + local TRUNCATE+INSERT, no HTTP/anti-bot. DB op — one FDW read + local TRUNCATE+INSERT, no HTTP/anti-bot.

229
tradein-mvp/backend/tests/services/test_location_coef.py
View file

@ -1,229 +0,0 @@
"""Unit tests for app.services.location_coef (#2045 BE-3, LocationDrawer).
No live Postgres needed — DB is a minimal fake returning queued results (mirrors the
convention in tests/tasks/test_cadastral_geo_match.py). Covers:
- pure functions: _category_weight, _score_to_coef, normalization constants
- compute_location_coef: weighted top-N scoring, empty-mirror graceful fallback,
no-POI-in-radius (legit zero-score, NOT "unavailable")
"""
from __future__ import annotations
import os
from typing import Any
# psycopg v3 driver required; stub DATABASE_URL before any app import (settings needs a DSN).
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
from app.services import location_coef as lc
# ── Pure functions ──────────────────────────────────────────────────────────
def test_category_weight_known_categories() -> None:
assert lc._category_weight("metro_stop") == 6.0
assert lc._category_weight("school") == 5.0
assert lc._category_weight("kindergarten") == 4.5
assert lc._category_weight("hospital") == 4.0
assert lc._category_weight("shop_mall") == 4.0
assert lc._category_weight("shop_supermarket") == 3.5
assert lc._category_weight("bus_stop") == 4.5
assert lc._category_weight("park") == 3.5
assert lc._category_weight("pharmacy") == 2.5
assert lc._category_weight("tram_stop") == 2.0
assert lc._category_weight("shop_small") == 2.0
def test_category_weight_unknown_and_none_fall_back_to_default() -> None:
assert lc._category_weight("unknown_category") == 1.0
assert lc._category_weight(None) == 1.0
def test_top7_weight_sum_matches_ptica() -> None:
"""Same category set as Site Finder → identical top-7 weight-sum constant (31.5)."""
assert lc._TOP7_WEIGHT_SUM == 31.5
def test_max_straight_score_normalized_at_reference_distance_not_zero() -> None:
"""Audit R2 #7a: denominator uses _REF_DISTANCE_M (100m), not an unreachable d=0 max.
Old d=0 normalization was _TOP7_WEIGHT_SUM / 100 == 0.315 — the theoretical max with all
top-7 categories AT the doorstep. That compressed realistic scores (~25-30/100 for even a
strong central address) into the bottom third of the range. The new normalization halves
the max (denominator distance+100 doubles from 100 to 200 at REF=100m), doubling realistic
scores instead.
"""
assert lc._REF_DISTANCE_M == 100.0
assert abs(lc._MAX_STRAIGHT_SCORE - 0.1575) < 1e-9
def test_score_to_coef_bounds() -> None:
assert lc._score_to_coef(0.0) == 0.95
assert lc._score_to_coef(100.0) == 1.05
def test_score_to_coef_midpoint() -> None:
assert lc._score_to_coef(50.0) == 1.0
def test_score_to_coef_is_monotonic() -> None:
scores = [0.0, 10.0, 25.0, 50.0, 75.0, 90.0, 100.0]
coefs = [lc._score_to_coef(s) for s in scores]
assert coefs == sorted(coefs)
# ── compute_location_coef with a fake DB ─────────────────────────────────────
class _FakeResult:
def __init__(self, *, scalar_value: Any = None, mapping_rows: list[dict] | None = None):
self._scalar_value = scalar_value
self._mapping_rows = mapping_rows or []
def scalar(self) -> Any:
return self._scalar_value
def mappings(self) -> Any:
class _Mappings:
def __init__(self, rows: list[dict]) -> None:
self._rows = rows
def all(self) -> list[dict]:
return self._rows
return _Mappings(self._mapping_rows)
class _FakeDB:
"""Minimal Session stand-in: execute() returns queued results in order."""
def __init__(self, results: list[_FakeResult]) -> None:
self._results = list(results)
self.executed: list[Any] = []
def execute(self, clause: Any, params: dict | None = None) -> _FakeResult:
self.executed.append((clause, params))
return self._results.pop(0)
def test_compute_location_coef_empty_mirror_returns_unavailable() -> None:
"""osm_poi_ekb_local not yet refreshed (count=0) → unavailable, no fabricated factors."""
db = _FakeDB([_FakeResult(scalar_value=0)])
result = lc.compute_location_coef(db, lat=56.84, lon=60.6)
assert result.coef == 1.0
assert result.factors == []
assert result.geo_source == "unavailable"
# Only the count probe ran — no nearest-POI query issued against an empty mirror.
assert len(db.executed) == 1
def test_compute_location_coef_no_poi_in_radius_is_legit_zero_score() -> None:
"""Mirror populated (count>0) but nothing within radius → coef floor, NOT unavailable."""
db = _FakeDB(
[
_FakeResult(scalar_value=500), # mirror has rows elsewhere
_FakeResult(mapping_rows=[]), # nothing near this point
]
)
result = lc.compute_location_coef(db, lat=56.84, lon=60.6)
assert result.factors == []
assert result.geo_source == "osm_poi_ekb"
assert result.coef == lc._score_to_coef(0.0) == 0.95
def test_compute_location_coef_weights_and_ranks_top_n() -> None:
"""Nearer + higher-weight-category POI ranks above farther/lower-weight ones."""
rows = [
{"name": "Школа №1", "category": "school", "distance_m": 300.0},
{"name": "ТЦ Мега", "category": "shop_mall", "distance_m": 900.0},
{"name": "Метро Ботаническая", "category": "metro_stop", "distance_m": 150.0},
{"name": "Аптека", "category": "pharmacy", "distance_m": 50.0},
]
db = _FakeDB([_FakeResult(scalar_value=1000), _FakeResult(mapping_rows=rows)])
result = lc.compute_location_coef(db, lat=56.84, lon=60.6, top_n=7)
assert result.geo_source == "osm_poi_ekb"
assert len(result.factors) == 4
# metro_stop (weight 6.0) at 150m beats school (5.0) at 300m despite being closer only
# marginally — sanity check the ranking is weight-driven, not distance-only.
assert result.factors[0].poi_type == "metro_stop"
# Weights strictly descending (sorted DESC by weight before slicing to top_n).
weights = [f.weight for f in result.factors]
assert weights == sorted(weights, reverse=True)
# coef must land inside the documented [0.95, 1.05] MVP range.
assert 0.95 <= result.coef <= 1.05
def test_compute_location_coef_strong_central_fixture_lands_near_one() -> None:
"""Audit R2 #7a: a realistic strong central-EKB POI profile now reads as ~1.0, not ~0.98.
Fixture mirrors a real strong central address audit: near transit/kindergarten/pharmacy,
but school/metro/mall noticeably farther (a typical quarter profile, NOT everything at
the doorstep). Under the OLD d=0 normalization this fixture scores ~28/100 → coef ~0.978
(verified separately against the pre-fix formula). After re-centering on _REF_DISTANCE_M
it should land close to 1.0 (within the documented ±0.01 calibration target).
"""
rows = [
{"name": "Остановка", "category": "bus_stop", "distance_m": 80.0},
{"name": "Аптека", "category": "pharmacy", "distance_m": 120.0},
{"name": "Супермаркет", "category": "shop_supermarket", "distance_m": 180.0},
{"name": "Детсад №5", "category": "kindergarten", "distance_m": 220.0},
{"name": "Школа №10", "category": "school", "distance_m": 350.0},
{"name": "Метро Геологическая", "category": "metro_stop", "distance_m": 500.0},
{"name": "ТЦ Гринвич", "category": "shop_mall", "distance_m": 700.0},
]
db = _FakeDB([_FakeResult(scalar_value=len(rows)), _FakeResult(mapping_rows=rows)])
result = lc.compute_location_coef(db, lat=56.838, lon=60.605, top_n=7)
assert result.geo_source == "osm_poi_ekb"
assert len(result.factors) == 7
assert abs(result.coef - 1.0) <= 0.01
def test_compute_location_coef_weak_fixture_stays_well_below_one() -> None:
"""A sparse/poor location (only far, low-weight POI) must still stay well below 1.0."""
rows = [
{"name": "Магазинчик", "category": "shop_small", "distance_m": 950.0},
{"name": "Прочее", "category": "some_unknown_tag", "distance_m": 1100.0},
]
db = _FakeDB([_FakeResult(scalar_value=len(rows)), _FakeResult(mapping_rows=rows)])
result = lc.compute_location_coef(db, lat=56.838, lon=60.605, top_n=7)
assert result.geo_source == "osm_poi_ekb"
assert result.coef < 0.98
def test_compute_location_coef_limits_to_top_n() -> None:
"""More than top_n candidates → only top_n factors surface in the response."""
rows = [
{"name": f"POI {i}", "category": "shop_small", "distance_m": float(100 + i * 10)}
for i in range(20)
]
db = _FakeDB([_FakeResult(scalar_value=20), _FakeResult(mapping_rows=rows)])
result = lc.compute_location_coef(db, lat=56.84, lon=60.6, top_n=7)
assert len(result.factors) == 7
def test_compute_location_coef_unknown_category_uses_default_weight() -> None:
rows = [{"name": "Неизвестный POI", "category": "some_new_osm_tag", "distance_m": 200.0}]
db = _FakeDB([_FakeResult(scalar_value=1), _FakeResult(mapping_rows=rows)])
result = lc.compute_location_coef(db, lat=56.84, lon=60.6)
assert len(result.factors) == 1
expected_weight = (1.0 / (200.0 + 100.0)) * lc.CATEGORY_WEIGHTS["default"]
assert abs(result.factors[0].weight - round(expected_weight, 6)) < 1e-9
def test_compute_location_coef_passes_radius_param() -> None:
"""radius_m is forwarded as a bound param (psycopg v3 CAST discipline, no :p::type)."""
db = _FakeDB([_FakeResult(scalar_value=1), _FakeResult(mapping_rows=[])])
lc.compute_location_coef(db, lat=56.84, lon=60.6, radius_m=1500)
_clause, params = db.executed[1]
assert params is not None
assert params["radius_m"] == 1500
def test_no_psycopg_v3_colon_colon_cast() -> None:
"""psycopg v3: never :param::type — must use CAST(:param AS type)."""
import re
assert not re.search(r":\w+::", str(lc._NEAREST_POI_SQL.text))

346
tradein-mvp/backend/tests/services/test_location_index.py Normal file
View file

@ -0,0 +1,346 @@
"""Unit tests for app.services.location_index (replaces test_location_coef.py).
No live Postgres needed — DB is a minimal fake returning queued results (mirrors the
convention in tests/tasks/test_cadastral_geo_match.py / the deleted test_location_coef.py).
Covers:
- pure functions: _category_weight, _in_ekb_bbox, _pct_deviation (incl. monotonicity)
- _fetch_nearby_poi: qualitative POI ranking (unchanged behaviour from the old module)
- compute_location_index: out-of-coverage degradation, insufficient-sample degradation
(citywide AND local), radius-ladder expansion, explicit radius_m override, happy path
- SQL discipline: psycopg v3 CAST, percentile_cont (not naive AVG/MIN/MAX) for outlier
robustness
"""
from __future__ import annotations
import os
from typing import Any
# psycopg v3 driver required; stub DATABASE_URL before any app import (settings needs a DSN).
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
from app.services import location_index as lc
# A point well inside the EKB coverage bbox (city centre, Ploshchad 1905 goda area).
_LAT_IN_EKB = 56.838
_LON_IN_EKB = 60.605
# ── Pure functions ──────────────────────────────────────────────────────────
def test_category_weight_known_categories() -> None:
assert lc._category_weight("metro_stop") == 6.0
assert lc._category_weight("school") == 5.0
assert lc._category_weight("kindergarten") == 4.5
assert lc._category_weight("hospital") == 4.0
assert lc._category_weight("shop_mall") == 4.0
assert lc._category_weight("shop_supermarket") == 3.5
assert lc._category_weight("bus_stop") == 4.5
assert lc._category_weight("park") == 3.5
assert lc._category_weight("pharmacy") == 2.5
assert lc._category_weight("tram_stop") == 2.0
assert lc._category_weight("shop_small") == 2.0
def test_category_weight_unknown_and_none_fall_back_to_default() -> None:
assert lc._category_weight("unknown_category") == 1.0
assert lc._category_weight(None) == 1.0
def test_in_ekb_bbox_center_is_inside() -> None:
assert lc._in_ekb_bbox(_LAT_IN_EKB, _LON_IN_EKB) is True
def test_in_ekb_bbox_bounds_are_inclusive() -> None:
assert lc._in_ekb_bbox(56.70, 60.50) is True
assert lc._in_ekb_bbox(56.95, 60.75) is True
def test_in_ekb_bbox_outside_is_rejected() -> None:
# Nizhny Tagil — same oblast (region_code=66), well outside the EKB product bbox.
assert lc._in_ekb_bbox(57.910, 59.970) is False
# Just past each edge of the bbox.
assert lc._in_ekb_bbox(56.69, 60.60) is False
assert lc._in_ekb_bbox(56.96, 60.60) is False
assert lc._in_ekb_bbox(56.80, 60.49) is False
assert lc._in_ekb_bbox(56.80, 60.76) is False
def test_pct_deviation_above_and_below_city_median() -> None:
assert lc._pct_deviation(165_000.0, 150_000.0) == 10.0
assert lc._pct_deviation(135_000.0, 150_000.0) == -10.0
assert lc._pct_deviation(150_000.0, 150_000.0) == 0.0
def test_pct_deviation_not_artificially_clamped() -> None:
"""Owner requirement: a genuinely +40% district must read as +40%, not clamped."""
assert lc._pct_deviation(210_000.0, 150_000.0) == 40.0
def test_pct_deviation_guards_zero_division() -> None:
assert lc._pct_deviation(100_000.0, 0.0) == 0.0
def test_pct_deviation_is_monotonic_in_local_median() -> None:
"""Индекс строго монотонен по локальной медиане при фиксированной городской — в отличие
от старого coef (немонотонные бакеты на реальных данных, см. модуль docstring).
Значения ниже — медианы ₽/м² по дистанционным бакетам от центра ЕКБ, измеренные на 31
тыс. лотов (аудит владельца продукта), отсортированные по возрастанию. Индекс,
построенный на этих же локальных медианах, обязан сохранить порядок.
"""
city_median = 155_000.0
local_medians_ascending = [
93_677.0,
136_729.0,
150_063.0,
159_382.0,
159_486.0,
191_682.0,
249_686.0,
]
pct_values = [lc._pct_deviation(m, city_median) for m in local_medians_ascending]
assert pct_values == sorted(pct_values)
# ── SQL discipline ────────────────────────────────────────────────────────────
def test_no_psycopg_v3_colon_colon_cast() -> None:
"""psycopg v3: never :param::type — must use CAST(:param AS type)."""
import re
for sql in (
lc._MEDIAN_PPM2_LOCAL_SQL,
lc._MEDIAN_PPM2_CITYWIDE_SQL,
lc._NEAREST_POI_SQL,
):
assert not re.search(r":\w+::", str(sql.text))
def test_median_queries_use_percentile_not_naive_minmax() -> None:
"""Outlier robustness requirement: percentile_cont(0.5) (median), not AVG/MIN/MAX."""
for sql in (lc._MEDIAN_PPM2_LOCAL_SQL, lc._MEDIAN_PPM2_CITYWIDE_SQL):
sql_text = str(sql.text).lower()
assert "percentile_cont(0.5)" in sql_text
assert "avg(" not in sql_text
assert "min(" not in sql_text
assert "max(" not in sql_text
def test_median_queries_exclude_city_centroid_and_bound_bbox() -> None:
"""Comparable-selection quality control (owner requirement #1): city-centroid geocodes
excluded (mirrors estimator.py #769 Part E), sample bounded to the EKB bbox."""
for sql in (lc._MEDIAN_PPM2_LOCAL_SQL, lc._MEDIAN_PPM2_CITYWIDE_SQL):
sql_text = str(sql.text)
assert "geo_precision IS DISTINCT FROM 'city'" in sql_text
assert "bbox_south" in sql_text and "bbox_north" in sql_text
assert "bbox_west" in sql_text and "bbox_east" in sql_text
# ── _fetch_nearby_poi (qualitative "что рядом" list) ─────────────────────────
class _FakeResult:
def __init__(
self,
*,
scalar_value: Any = None,
mapping_rows: list[dict] | None = None,
mapping_one: dict | None = None,
):
self._scalar_value = scalar_value
self._mapping_rows = mapping_rows or []
self._mapping_one = mapping_one
def scalar(self) -> Any:
return self._scalar_value
def mappings(self) -> Any:
outer = self
class _Mappings:
def all(self) -> list[dict]:
return outer._mapping_rows
def first(self) -> dict | None:
return outer._mapping_one
return _Mappings()
class _FakeDB:
"""Minimal Session stand-in: execute() returns queued results in order."""
def __init__(self, results: list[_FakeResult]) -> None:
self._results = list(results)
self.executed: list[Any] = []
def execute(self, clause: Any, params: dict | None = None) -> _FakeResult:
self.executed.append((clause, params))
return self._results.pop(0)
def test_fetch_nearby_poi_empty_mirror_returns_unavailable() -> None:
db = _FakeDB([_FakeResult(scalar_value=0)])
poi, status = lc._fetch_nearby_poi(db, _LAT_IN_EKB, _LON_IN_EKB, lc.DEFAULT_POI_RADIUS_M, 7)
assert poi == []
assert status == "unavailable"
assert len(db.executed) == 1 # only the count probe ran
def test_fetch_nearby_poi_no_poi_in_radius_is_legit_ok() -> None:
db = _FakeDB([_FakeResult(scalar_value=500), _FakeResult(mapping_rows=[])])
poi, status = lc._fetch_nearby_poi(db, _LAT_IN_EKB, _LON_IN_EKB, lc.DEFAULT_POI_RADIUS_M, 7)
assert poi == []
assert status == "ok"
def test_fetch_nearby_poi_ranks_by_weight_not_distance_only() -> None:
rows = [
{"name": "Школа №1", "category": "school", "distance_m": 300.0},
{"name": "ТЦ Мега", "category": "shop_mall", "distance_m": 900.0},
{"name": "Метро Ботаническая", "category": "metro_stop", "distance_m": 150.0},
{"name": "Аптека", "category": "pharmacy", "distance_m": 50.0},
]
db = _FakeDB([_FakeResult(scalar_value=1000), _FakeResult(mapping_rows=rows)])
poi, status = lc._fetch_nearby_poi(db, _LAT_IN_EKB, _LON_IN_EKB, 1200, 7)
assert status == "ok"
assert len(poi) == 4
# metro_stop (weight 6.0) at 150m outranks school (5.0) at 300m — weight-driven, not
# distance-only ranking.
assert poi[0].poi_type == "metro_stop"
def test_fetch_nearby_poi_limits_to_top_n() -> None:
rows = [
{"name": f"POI {i}", "category": "shop_small", "distance_m": float(100 + i * 10)}
for i in range(20)
]
db = _FakeDB([_FakeResult(scalar_value=20), _FakeResult(mapping_rows=rows)])
poi, _status = lc._fetch_nearby_poi(db, _LAT_IN_EKB, _LON_IN_EKB, 1200, 7)
assert len(poi) == 7
# ── compute_location_index: degradation + ladder logic ───────────────────────
def test_compute_location_index_out_of_coverage_skips_all_db_calls() -> None:
"""Owner requirement #2: point outside EKB → honest 'no data', never a fallback number.
Also a perf/honesty check: no DB round-trip at all for an out-of-scope point.
"""
db = _FakeDB([])
result = lc.compute_location_index(db, lat=57.910, lon=59.970) # Nizhny Tagil
assert result.status == "out_of_coverage"
assert result.location_index_pct is None
assert result.local_median_price_per_m2 is None
assert result.city_median_price_per_m2 is None
assert result.sample_size == 0
assert result.nearby_poi == []
assert result.poi_status == "unavailable"
assert db.executed == []
def test_compute_location_index_citywide_sample_too_small_short_circuits() -> None:
"""Degenerate citywide reference (e.g. empty dev DB) → insufficient_data without ever
issuing a local-radius query (nothing to compare against anyway)."""
db = _FakeDB(
[
_FakeResult(scalar_value=0), # POI mirror empty
_FakeResult(mapping_one={"median_ppm2": None, "n": 3}), # citywide: n < MIN
]
)
result = lc.compute_location_index(db, lat=_LAT_IN_EKB, lon=_LON_IN_EKB)
assert result.status == "insufficient_data"
assert result.location_index_pct is None
assert result.sample_size == 3
assert len(db.executed) == 2 # poi-count + citywide only — no radius-ladder queries
def test_compute_location_index_first_radius_rung_sufficient() -> None:
db = _FakeDB(
[
_FakeResult(scalar_value=0), # poi mirror empty
_FakeResult(mapping_one={"median_ppm2": 150_000.0, "n": 4000}), # citywide
_FakeResult(mapping_one={"median_ppm2": 165_000.0, "n": 25}), # radius[0]=800
]
)
result = lc.compute_location_index(db, lat=_LAT_IN_EKB, lon=_LON_IN_EKB)
assert result.status == "ok"
assert result.radius_m == lc.RADIUS_LADDER_M[0]
assert result.sample_size == 25
assert result.local_median_price_per_m2 == 165_000
assert result.city_median_price_per_m2 == 150_000
assert result.location_index_pct == 10.0
assert len(db.executed) == 3 # ladder stopped at rung 1 — no further radius queries
def test_compute_location_index_expands_ladder_when_first_rung_insufficient() -> None:
db = _FakeDB(
[
_FakeResult(scalar_value=0),
_FakeResult(mapping_one={"median_ppm2": 150_000.0, "n": 4000}),
_FakeResult(mapping_one={"median_ppm2": 200_000.0, "n": 10}), # 800m: too few
_FakeResult(mapping_one={"median_ppm2": 180_000.0, "n": 30}), # 1500m: enough
]
)
result = lc.compute_location_index(db, lat=_LAT_IN_EKB, lon=_LON_IN_EKB)
assert result.status == "ok"
assert result.radius_m == lc.RADIUS_LADDER_M[1]
assert result.sample_size == 30
assert len(db.executed) == 4
def test_compute_location_index_insufficient_even_at_max_radius() -> None:
db = _FakeDB(
[
_FakeResult(scalar_value=0),
_FakeResult(mapping_one={"median_ppm2": 150_000.0, "n": 4000}),
_FakeResult(mapping_one={"median_ppm2": 200_000.0, "n": 5}), # 800m
_FakeResult(mapping_one={"median_ppm2": 195_000.0, "n": 12}), # 1500m
_FakeResult(mapping_one={"median_ppm2": 190_000.0, "n": 15}), # 2500m — still < 20
]
)
result = lc.compute_location_index(db, lat=_LAT_IN_EKB, lon=_LON_IN_EKB)
assert result.status == "insufficient_data"
assert result.location_index_pct is None
assert result.local_median_price_per_m2 is None
assert result.city_median_price_per_m2 == 150_000
assert result.radius_m == lc.RADIUS_LADDER_M[-1]
assert result.sample_size == 15 # honest: shows how close it got, not just "no data"
assert len(db.executed) == 5 # exhausted the full ladder
def test_compute_location_index_explicit_radius_skips_ladder() -> None:
"""An explicit radius_m must be used AS-IS — no adaptive expansion (caller-controlled)."""
db = _FakeDB(
[
_FakeResult(scalar_value=0),
_FakeResult(mapping_one={"median_ppm2": 150_000.0, "n": 4000}),
_FakeResult(mapping_one={"median_ppm2": 160_000.0, "n": 50}), # single query only
]
)
result = lc.compute_location_index(db, lat=_LAT_IN_EKB, lon=_LON_IN_EKB, radius_m=1000)
assert result.status == "ok"
assert result.radius_m == 1000
assert len(db.executed) == 3 # exactly one radius query, no ladder rungs tried
nearest_call_params = db.executed[2][1]
assert nearest_call_params["radius_m"] == 1000
def test_compute_location_index_poi_unavailable_does_not_block_index() -> None:
"""poi_status and status degrade INDEPENDENTLY — an empty POI mirror must not prevent a
perfectly computable price-based index."""
db = _FakeDB(
[
_FakeResult(scalar_value=0), # poi mirror empty
_FakeResult(mapping_one={"median_ppm2": 150_000.0, "n": 4000}),
_FakeResult(mapping_one={"median_ppm2": 172_500.0, "n": 40}),
]
)
result = lc.compute_location_index(db, lat=_LAT_IN_EKB, lon=_LON_IN_EKB)
assert result.status == "ok"
assert result.poi_status == "unavailable"
assert result.nearby_poi == []
assert result.location_index_pct == 15.0

251
tradein-mvp/backend/tests/test_location_coef_endpoint.py
View file

@ -1,251 +0,0 @@
"""Tests for GET /api/v1/trade-in/location-coef (#2045 BE-3, LocationDrawer).
Mirrors the IDOR + mocked-DB conventions of test_estimate_idor.py / test_street_deals_endpoint.py
(no live Postgres needed). Covers:
- IDOR guard (owner / other pilot 404 / admin / unauthenticated 401)
- graceful fallback: osm_poi_ekb_local empty/not-refreshed, estimate has no lat/lon
- happy path: base_price_rub * coef → result_price_rub, factors surfaced
- radius_m query-param clamping
"""
from __future__ import annotations
import os
import sys
from types import SimpleNamespace
from unittest.mock import MagicMock
# psycopg v3 driver required; stub DATABASE_URL before any app import.
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
# WeasyPrint requires GTK — not present in CI/Windows. Stub before any app import
# (trade_in.py imports generate_trade_in_pdf at module load).
_wp_mock = MagicMock()
sys.modules.setdefault("weasyprint", _wp_mock)
sys.modules.setdefault("weasyprint.CSS", _wp_mock)
sys.modules.setdefault("weasyprint.HTML", _wp_mock)
import pytest # noqa: E402
from fastapi import FastAPI # noqa: E402
from fastapi.testclient import TestClient # noqa: E402
_ESTIMATE_ID = "22222222-2222-2222-2222-222222222222"
@pytest.fixture(autouse=True)
def _restore_get_role():
"""Restore app.core.auth.get_role after each test (mirror test_estimate_idor)."""
from app.core import auth as auth_mod
original = auth_mod.get_role
yield
auth_mod.get_role = original
@pytest.fixture()
def trade_in_app() -> FastAPI:
"""Minimal FastAPI app mounting only the trade-in router."""
from app.api.v1 import trade_in as trade_in_module
application = FastAPI()
application.include_router(trade_in_module.router, prefix="/api/v1/trade-in")
return application
def _client_with(app: FastAPI, db_mock: MagicMock, role: str | None) -> TestClient:
"""Override get_db with *db_mock*; patch get_role to return *role* (or raise KeyError)."""
from app.core.db import get_db
def _override_db():
yield db_mock
app.dependency_overrides[get_db] = _override_db
auth_mod = sys.modules["app.core.auth"]
if role is None:
def _raise_keyerror(_u: str):
raise KeyError(_u)
auth_mod.get_role = _raise_keyerror # type: ignore[assignment]
else:
auth_mod.get_role = lambda _u: role # type: ignore[assignment]
return TestClient(app)
def _fetchone_result(row: object) -> MagicMock:
r = MagicMock()
r.fetchone.return_value = row
return r
def _scalar_result(value: object) -> MagicMock:
r = MagicMock()
r.scalar.return_value = value
return r
def _mapping_result(rows: list[dict]) -> MagicMock:
r = MagicMock()
r.mappings.return_value.all.return_value = rows
return r
def _db_with(*results: MagicMock) -> MagicMock:
db = MagicMock()
db.execute.side_effect = list(results)
return db
# ── IDOR guard ────────────────────────────────────────────────────────────────
def test_location_coef_other_pilot_gets_404(trade_in_app: FastAPI) -> None:
"""Non-owner pilot must NOT read someone else's location-coef → 404."""
db = _db_with(_fetchone_result(SimpleNamespace(created_by="victim")))
client = _client_with(trade_in_app, db, role="pilot")
resp = client.get(
"/api/v1/trade-in/location-coef",
params={"estimate_id": _ESTIMATE_ID},
headers={"X-Authenticated-User": "attacker"},
)
assert resp.status_code == 404
def test_location_coef_requires_authenticated_user(trade_in_app: FastAPI) -> None:
"""No X-Authenticated-User header → 401 (guard query already ran, header check fails)."""
db = _db_with(_fetchone_result(SimpleNamespace(created_by="kopylov")))
client = _client_with(trade_in_app, db, role="pilot")
resp = client.get(
"/api/v1/trade-in/location-coef",
params={"estimate_id": _ESTIMATE_ID},
)
assert resp.status_code == 401
def test_location_coef_unknown_estimate_returns_404(trade_in_app: FastAPI) -> None:
db = _db_with(_fetchone_result(None))
client = _client_with(trade_in_app, db, role="pilot")
resp = client.get(
"/api/v1/trade-in/location-coef",
params={"estimate_id": _ESTIMATE_ID},
headers={"X-Authenticated-User": "kopylov"},
)
assert resp.status_code == 404
def test_location_coef_admin_can_read_any(trade_in_app: FastAPI) -> None:
"""Admin reads any estimate's location-coef regardless of owner → 200."""
db = _db_with(
_fetchone_result(SimpleNamespace(created_by="someone_else")), # guard
_fetchone_result(SimpleNamespace(lat=None, lon=None, median_price=5_000_000)), # target
)
client = _client_with(trade_in_app, db, role="admin")
resp = client.get(
"/api/v1/trade-in/location-coef",
params={"estimate_id": _ESTIMATE_ID},
headers={"X-Authenticated-User": "admin"},
)
assert resp.status_code == 200
# ── Graceful fallback ───────────────────────────────────────────────────────
def test_location_coef_no_lat_lon_returns_unavailable(trade_in_app: FastAPI) -> None:
"""Estimate without lat/lon → unavailable fallback, no osm_poi_ekb_local query at all."""
db = _db_with(
_fetchone_result(SimpleNamespace(created_by="kopylov")), # guard
_fetchone_result(SimpleNamespace(lat=None, lon=None, median_price=5_000_000)), # target
)
client = _client_with(trade_in_app, db, role="pilot")
resp = client.get(
"/api/v1/trade-in/location-coef",
params={"estimate_id": _ESTIMATE_ID},
headers={"X-Authenticated-User": "kopylov"},
)
assert resp.status_code == 200
body = resp.json()
assert body["geo_source"] == "unavailable"
assert body["coef"] == 1.0
assert body["factors"] == []
assert body["base_price_rub"] == 5_000_000
assert body["result_price_rub"] == 5_000_000
# Short-circuits before compute_location_coef: only guard + target queries ran.
assert db.execute.call_count == 2
def test_location_coef_empty_mirror_returns_unavailable(trade_in_app: FastAPI) -> None:
"""osm_poi_ekb_local not yet refreshed (count=0) → unavailable, no fabricated factors."""
db = _db_with(
_fetchone_result(SimpleNamespace(created_by="kopylov")), # guard
_fetchone_result(SimpleNamespace(lat=56.84, lon=60.6, median_price=5_000_000)), # target
_scalar_result(0), # osm_poi_ekb_local count
)
client = _client_with(trade_in_app, db, role="pilot")
resp = client.get(
"/api/v1/trade-in/location-coef",
params={"estimate_id": _ESTIMATE_ID},
headers={"X-Authenticated-User": "kopylov"},
)
assert resp.status_code == 200
body = resp.json()
assert body["geo_source"] == "unavailable"
assert body["coef"] == 1.0
assert body["factors"] == []
assert body["result_price_rub"] == body["base_price_rub"]
# ── Happy path ────────────────────────────────────────────────────────────────
def test_location_coef_happy_path_computes_result_price(trade_in_app: FastAPI) -> None:
"""POI found within radius → coef applied to base_price_rub, factors surfaced."""
poi_rows = [
{"name": "Школа №1", "category": "school", "distance_m": 300.0},
{"name": "Метро Ботаническая", "category": "metro_stop", "distance_m": 150.0},
]
db = _db_with(
_fetchone_result(SimpleNamespace(created_by="kopylov")), # guard
_fetchone_result(SimpleNamespace(lat=56.84, lon=60.6, median_price=5_000_000)), # target
_scalar_result(1000), # osm_poi_ekb_local count
_mapping_result(poi_rows), # nearest POI
)
client = _client_with(trade_in_app, db, role="pilot")
resp = client.get(
"/api/v1/trade-in/location-coef",
params={"estimate_id": _ESTIMATE_ID},
headers={"X-Authenticated-User": "kopylov"},
)
assert resp.status_code == 200
body = resp.json()
assert body["geo_source"] == "osm_poi_ekb"
assert body["base_price_rub"] == 5_000_000
assert 0.95 <= body["coef"] <= 1.05
assert body["result_price_rub"] == round(5_000_000 * body["coef"])
assert len(body["factors"]) == 2
poi_types = {f["poi_type"] for f in body["factors"]}
assert poi_types == {"school", "metro_stop"}
def test_location_coef_radius_m_clamped_to_bounds(trade_in_app: FastAPI) -> None:
"""Explicit radius_m outside [500, 3000] is clamped before hitting the DB query."""
db = _db_with(
_fetchone_result(SimpleNamespace(created_by="kopylov")), # guard
_fetchone_result(SimpleNamespace(lat=56.84, lon=60.6, median_price=5_000_000)), # target
_scalar_result(10), # osm_poi_ekb_local count
_mapping_result([]), # nearest POI query
)
client = _client_with(trade_in_app, db, role="pilot")
resp = client.get(
"/api/v1/trade-in/location-coef",
params={"estimate_id": _ESTIMATE_ID, "radius_m": 50},
headers={"X-Authenticated-User": "kopylov"},
)
assert resp.status_code == 200
# 4th execute() call is the nearest-POI query — inspect its bound radius_m param.
nearest_call = db.execute.call_args_list[3]
params = (
nearest_call.args[1] if len(nearest_call.args) > 1 else nearest_call.kwargs.get("params")
)
assert params["radius_m"] == 500

325
tradein-mvp/backend/tests/test_location_index_endpoint.py Normal file
View file

@ -0,0 +1,325 @@
"""Tests for GET /api/v1/trade-in/location-index (replaces test_location_coef_endpoint.py).
Mirrors the IDOR + mocked-DB conventions of test_estimate_idor.py / the deleted
test_location_coef_endpoint.py (no live Postgres needed). Covers:
- IDOR guard (owner / other pilot 404 / admin / unauthenticated 401)
- honest degradation: no lat/lon, point outside EKB coverage, insufficient comparable
sample, empty osm_poi_ekb_local mirror (independent of the index status)
- happy path: status="ok", location_index_pct computed, nearby_poi surfaced
- radius_m query-param clamping
"""
from __future__ import annotations
import os
import sys
from types import SimpleNamespace
from unittest.mock import MagicMock
# psycopg v3 driver required; stub DATABASE_URL before any app import.
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
# WeasyPrint requires GTK — not present in CI/Windows. Stub before any app import
# (trade_in.py imports generate_trade_in_pdf at module load).
_wp_mock = MagicMock()
sys.modules.setdefault("weasyprint", _wp_mock)
sys.modules.setdefault("weasyprint.CSS", _wp_mock)
sys.modules.setdefault("weasyprint.HTML", _wp_mock)
import pytest # noqa: E402
from fastapi import FastAPI # noqa: E402
from fastapi.testclient import TestClient # noqa: E402
_ESTIMATE_ID = "22222222-2222-2222-2222-222222222222"
_LAT_IN_EKB = 56.838
_LON_IN_EKB = 60.605
@pytest.fixture(autouse=True)
def _restore_get_role():
"""Restore app.core.auth.get_role after each test (mirror test_estimate_idor)."""
from app.core import auth as auth_mod
original = auth_mod.get_role
yield
auth_mod.get_role = original
@pytest.fixture()
def trade_in_app() -> FastAPI:
"""Minimal FastAPI app mounting only the trade-in router."""
from app.api.v1 import trade_in as trade_in_module
application = FastAPI()
application.include_router(trade_in_module.router, prefix="/api/v1/trade-in")
return application
def _client_with(app: FastAPI, db_mock: MagicMock, role: str | None) -> TestClient:
"""Override get_db with *db_mock*; patch get_role to return *role* (or raise KeyError)."""
from app.core.db import get_db
def _override_db():
yield db_mock
app.dependency_overrides[get_db] = _override_db
auth_mod = sys.modules["app.core.auth"]
if role is None:
def _raise_keyerror(_u: str):
raise KeyError(_u)
auth_mod.get_role = _raise_keyerror # type: ignore[assignment]
else:
auth_mod.get_role = lambda _u: role # type: ignore[assignment]
return TestClient(app)
def _fetchone_result(row: object) -> MagicMock:
r = MagicMock()
r.fetchone.return_value = row
return r
def _scalar_result(value: object) -> MagicMock:
r = MagicMock()
r.scalar.return_value = value
return r
def _mapping_all_result(rows: list[dict]) -> MagicMock:
r = MagicMock()
r.mappings.return_value.all.return_value = rows
return r
def _mapping_one_result(row: dict | None) -> MagicMock:
r = MagicMock()
r.mappings.return_value.first.return_value = row
return r
def _db_with(*results: MagicMock) -> MagicMock:
db = MagicMock()
db.execute.side_effect = list(results)
return db
# ── IDOR guard ────────────────────────────────────────────────────────────────
def test_location_index_other_pilot_gets_404(trade_in_app: FastAPI) -> None:
"""Non-owner pilot must NOT read someone else's location-index → 404."""
db = _db_with(_fetchone_result(SimpleNamespace(created_by="victim")))
client = _client_with(trade_in_app, db, role="pilot")
resp = client.get(
"/api/v1/trade-in/location-index",
params={"estimate_id": _ESTIMATE_ID},
headers={"X-Authenticated-User": "attacker"},
)
assert resp.status_code == 404
def test_location_index_requires_authenticated_user(trade_in_app: FastAPI) -> None:
"""No X-Authenticated-User header → 401 (guard query already ran, header check fails)."""
db = _db_with(_fetchone_result(SimpleNamespace(created_by="kopylov")))
client = _client_with(trade_in_app, db, role="pilot")
resp = client.get(
"/api/v1/trade-in/location-index",
params={"estimate_id": _ESTIMATE_ID},
)
assert resp.status_code == 401
def test_location_index_unknown_estimate_returns_404(trade_in_app: FastAPI) -> None:
db = _db_with(_fetchone_result(None))
client = _client_with(trade_in_app, db, role="pilot")
resp = client.get(
"/api/v1/trade-in/location-index",
params={"estimate_id": _ESTIMATE_ID},
headers={"X-Authenticated-User": "kopylov"},
)
assert resp.status_code == 404
def test_location_index_admin_can_read_any(trade_in_app: FastAPI) -> None:
"""Admin reads any estimate's location-index regardless of owner → 200."""
db = _db_with(
_fetchone_result(SimpleNamespace(created_by="someone_else")), # guard
_fetchone_result(SimpleNamespace(lat=None, lon=None)), # target — no lat/lon
)
client = _client_with(trade_in_app, db, role="admin")
resp = client.get(
"/api/v1/trade-in/location-index",
params={"estimate_id": _ESTIMATE_ID},
headers={"X-Authenticated-User": "admin"},
)
assert resp.status_code == 200
# ── Honest degradation ────────────────────────────────────────────────────────
def test_location_index_no_lat_lon_returns_out_of_coverage(trade_in_app: FastAPI) -> None:
"""Estimate without lat/lon → out_of_coverage, no compute_location_index DB call at all."""
db = _db_with(
_fetchone_result(SimpleNamespace(created_by="kopylov")), # guard
_fetchone_result(SimpleNamespace(lat=None, lon=None)), # target
)
client = _client_with(trade_in_app, db, role="pilot")
resp = client.get(
"/api/v1/trade-in/location-index",
params={"estimate_id": _ESTIMATE_ID},
headers={"X-Authenticated-User": "kopylov"},
)
assert resp.status_code == 200
body = resp.json()
assert body["status"] == "out_of_coverage"
assert body["location_index_pct"] is None
assert body["nearby_poi"] == []
assert body["poi_status"] == "unavailable"
# Short-circuits before compute_location_index: only guard + target queries ran.
assert db.execute.call_count == 2
def test_location_index_point_outside_ekb_bbox_returns_out_of_coverage(
trade_in_app: FastAPI,
) -> None:
"""Estimate has lat/lon, but outside the EKB coverage bbox (e.g. Nizhny Tagil)."""
db = _db_with(
_fetchone_result(SimpleNamespace(created_by="kopylov")), # guard
_fetchone_result(SimpleNamespace(lat=57.910, lon=59.970)), # target — Nizhny Tagil
)
client = _client_with(trade_in_app, db, role="pilot")
resp = client.get(
"/api/v1/trade-in/location-index",
params={"estimate_id": _ESTIMATE_ID},
headers={"X-Authenticated-User": "kopylov"},
)
assert resp.status_code == 200
body = resp.json()
assert body["status"] == "out_of_coverage"
assert body["location_index_pct"] is None
# compute_location_index short-circuits — no further DB calls beyond guard + target.
assert db.execute.call_count == 2
def test_location_index_insufficient_sample_returns_no_fabricated_number(
trade_in_app: FastAPI,
) -> None:
"""Comparable sample stays below MIN_SAMPLE_SIZE even at max radius → insufficient_data,
never a noisy number computed from a handful of listings."""
db = _db_with(
_fetchone_result(SimpleNamespace(created_by="kopylov")), # guard
_fetchone_result(SimpleNamespace(lat=_LAT_IN_EKB, lon=_LON_IN_EKB)), # target
_scalar_result(0), # osm_poi_ekb_local count
_mapping_one_result({"median_ppm2": 150_000.0, "n": 4000}), # citywide
_mapping_one_result({"median_ppm2": 200_000.0, "n": 5}), # 800m
_mapping_one_result({"median_ppm2": 195_000.0, "n": 12}), # 1500m
_mapping_one_result({"median_ppm2": 190_000.0, "n": 15}), # 2500m — still short
)
client = _client_with(trade_in_app, db, role="pilot")
resp = client.get(
"/api/v1/trade-in/location-index",
params={"estimate_id": _ESTIMATE_ID},
headers={"X-Authenticated-User": "kopylov"},
)
assert resp.status_code == 200
body = resp.json()
assert body["status"] == "insufficient_data"
assert body["location_index_pct"] is None
assert body["local_median_price_per_m2"] is None
assert body["sample_size"] == 15
assert body["city_median_price_per_m2"] == 150_000
def test_location_index_poi_mirror_empty_does_not_block_index(trade_in_app: FastAPI) -> None:
"""poi_status="unavailable" is independent of status="ok" — an un-refreshed POI mirror
must not prevent a perfectly computable price-based index."""
db = _db_with(
_fetchone_result(SimpleNamespace(created_by="kopylov")), # guard
_fetchone_result(SimpleNamespace(lat=_LAT_IN_EKB, lon=_LON_IN_EKB)), # target
_scalar_result(0), # osm_poi_ekb_local count == 0
_mapping_one_result({"median_ppm2": 150_000.0, "n": 4000}), # citywide
_mapping_one_result({"median_ppm2": 172_500.0, "n": 40}), # 800m — sufficient
)
client = _client_with(trade_in_app, db, role="pilot")
resp = client.get(
"/api/v1/trade-in/location-index",
params={"estimate_id": _ESTIMATE_ID},
headers={"X-Authenticated-User": "kopylov"},
)
assert resp.status_code == 200
body = resp.json()
assert body["status"] == "ok"
assert body["location_index_pct"] == 15.0
assert body["poi_status"] == "unavailable"
assert body["nearby_poi"] == []
# ── Happy path ────────────────────────────────────────────────────────────────
def test_location_index_happy_path(trade_in_app: FastAPI) -> None:
"""Comparable sample found → index computed, nearby POI surfaced as qualitative info."""
poi_rows = [
{"name": "Школа №1", "category": "school", "distance_m": 300.0},
{"name": "Метро Ботаническая", "category": "metro_stop", "distance_m": 150.0},
]
db = _db_with(
_fetchone_result(SimpleNamespace(created_by="kopylov")), # guard
_fetchone_result(SimpleNamespace(lat=_LAT_IN_EKB, lon=_LON_IN_EKB)), # target
_scalar_result(1000), # osm_poi_ekb_local count
_mapping_all_result(poi_rows), # nearest POI
_mapping_one_result({"median_ppm2": 150_000.0, "n": 4000}), # citywide
_mapping_one_result({"median_ppm2": 165_000.0, "n": 25}), # 800m — sufficient
)
client = _client_with(trade_in_app, db, role="pilot")
resp = client.get(
"/api/v1/trade-in/location-index",
params={"estimate_id": _ESTIMATE_ID},
headers={"X-Authenticated-User": "kopylov"},
)
assert resp.status_code == 200
body = resp.json()
assert body["status"] == "ok"
assert body["poi_status"] == "ok"
assert body["location_index_pct"] == 10.0
assert body["local_median_price_per_m2"] == 165_000
assert body["city_median_price_per_m2"] == 150_000
assert body["sample_size"] == 25
assert body["radius_m"] == 800
assert len(body["nearby_poi"]) == 2
poi_types = {f["poi_type"] for f in body["nearby_poi"]}
assert poi_types == {"school", "metro_stop"}
# New contract must NOT resurrect the misleading price-multiplier fields.
assert "coef" not in body
assert "base_price_rub" not in body
assert "result_price_rub" not in body
def test_location_index_radius_m_clamped_to_bounds(trade_in_app: FastAPI) -> None:
"""Explicit radius_m outside [500, 3000] is clamped before hitting the DB query."""
db = _db_with(
_fetchone_result(SimpleNamespace(created_by="kopylov")), # guard
_fetchone_result(SimpleNamespace(lat=_LAT_IN_EKB, lon=_LON_IN_EKB)), # target
_scalar_result(0), # osm_poi_ekb_local count
_mapping_one_result({"median_ppm2": 150_000.0, "n": 4000}), # citywide
_mapping_one_result({"median_ppm2": 160_000.0, "n": 50}), # explicit radius query
)
client = _client_with(trade_in_app, db, role="pilot")
resp = client.get(
"/api/v1/trade-in/location-index",
params={"estimate_id": _ESTIMATE_ID, "radius_m": 50},
headers={"X-Authenticated-User": "kopylov"},
)
assert resp.status_code == 200
body = resp.json()
assert body["radius_m"] == 500
# 5th execute() call (index 4) is the explicit-radius local-median query.
nearest_call = db.execute.call_args_list[4]
params = (
nearest_call.args[1] if len(nearest_call.args) > 1 else nearest_call.kwargs.get("params")
)
assert params["radius_m"] == 500