gendesign/tradein-mvp/backend/app/api/v1/trade_in.py
bot-backend a780e3e66e
All checks were successful
CI Trade-In / changes (pull_request) Successful in 9s
CI / changes (pull_request) Successful in 11s
CI Trade-In / browser-tests (pull_request) Has been skipped
CI Trade-In / frontend-checks (pull_request) Has been skipped
CI / backend-tests (pull_request) Has been skipped
CI / frontend-tests (pull_request) Has been skipped
CI / openapi-codegen-check (pull_request) Has been skipped
CI Trade-In / backend-tests (pull_request) Successful in 5m10s
fix(estimator): ключевать сделки Росреестра area-бакетом, а не комнатностью клиента
`deals.rooms` — не комнатность, а синтетика из площади: import-rosreestr.sh пишет
тот же CASE 30/44/62/85, что `asking_to_sold_ratio.area_bucket`. Прод-замер
2026-09-11: 321 559 из 321 560 сделок удовлетворяют rooms == area_bucket(area_m2),
max(rooms) = 4. Значит предикат `deals.rooms = <РЕАЛЬНЫЕ комнаты клиента>` — это
переодетый фильтр по площади, который противоречит area-полосе ±15% рядом с ним,
как только комнатность клиента нетипична для метража, и НИКОГДА не совпадает у
клиентов с 5+ комнатами.

Замер по 1177 реальным запросам (trade_in_estimates): ключ расходился с
area-бакетом у 359 (30.5%); коридор ДКП пуст у 46.2% из них против 8.2% у
совпадающих. По крупному жилью (≥85 м²): «3 комнаты» — 83.5% пустых коридоров,
«5 комнат» и «6 комнат» — 100%, «4 комнаты» — 5%. Т.е. блок «реальные сделки»
и клампы коридора (cap headline + radius-floor) молча выключались ровно у
крупных лотов.

Прогон тех же 1177 запросов через `_fetch_dkp_corridor` с обоими ключами:
непустых коридоров 809 → 895, пригодных для клампа (n≥10) 567 → 623 (+66, −10),
у 818 клиентов с совпадающей комнатностью выборка не меняется вовсе. Из 66
восстановленных коридоров 7 (5 из них ≥85 м²) обрезали бы headline вниз на
медианных −10.1% — то есть сейчас часть крупных лотов оценивается выше, чем
поддерживают реальные ДКП на той же улице.

Правка — одно и то же во всех четырёх местах, где сделки фильтруются под
клиента: `_fetch_dkp_corridor` (street + city-wide widen), `_fetch_deals`
(радиус) и витрина `/street-deals`.

Бэктест этим НЕ измеряется и в докстринг харнеса добавлена причина (каверза (e)):
у всех 5500 сделок обеих прод-фикстур rooms == area_bucket, т.е. харнес кормит
спайн синтетическим ключом и поэтому по построению не видит расхождения, которое
в проде есть у 30.5% запросов.

Refs #3256
2026-09-12 01:04:05 +05:00

3020 lines
155 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""Trade-In Estimator — endpoints (TI-2 PDF, photos, history).
Реальная оценка делается через app.services.estimator.estimate_quality().
"""
from __future__ import annotations
import asyncio
import calendar
import json
import logging
import math
from datetime import UTC, date, datetime, timedelta
from typing import Annotated, Any, Literal
from uuid import UUID
from fastapi import APIRouter, Depends, File, Header, HTTPException, Request, Response, UploadFile
from sqlalchemy import text
from sqlalchemy.orm import Session
from app.core.anon_session import get_or_create_anon_session_id
from app.core.config import settings
from app.core.db import get_db
from app.core.ratelimit import SlidingWindowLimiter, _client_ip
from app.schemas.trade_in import (
AggregatedEstimate,
AnalogLot,
AvitoImvSummary,
CianPriceChangeStats,
CoverageProbeInput,
CoverageProbeResponse,
DkpCorridor,
HouseAnalyticsKpi,
HouseAnalyticsResponse,
HouseInfoForEstimate,
IMVBenchmarkResponse,
LocationIndexResponse,
NearbyPoiOut,
PhotoMeta,
PlacementHistoryEntry,
PriceHistoryYearPoint,
PriceTrendPoint,
QuotaStatus,
RecentSoldEntry,
SalesListingPair,
SalesVsListingsResponse,
SellTimeBucket,
SellTimeSensitivityResponse,
StreetDealsResponse,
TradeInEstimateInput,
)
from app.services import account_quota
from app.services import regions as regions_mod
from app.services.exporters.trade_in_pdf import generate_trade_in_pdf
from app.services.image_sanitizer import ImageSanitizationError, sanitize_image
from app.services.user_events import schedule_event
logger = logging.getLogger(__name__)
router = APIRouter()
# ── B2C anti-abuse этап 2 (#b2c-antiabuse-2) ────────────────────────────────
# Отдельный, куда более строгий лимит частоты specifically на POST /estimate —
# см. settings.estimate_rate_limit/_window_s (app/core/config.py) для обоснования
# значений. Тот же паттерн, что _send_limiter в app/api/v1/support.py: singleton
# SlidingWindowLimiter поверх общего RateLimitMiddleware (app/main.py), который
# уже применяется КО ВСЕМ /api/* путям, но с щедрым порогом, рассчитанным на
# дешёвые запросы — один /estimate запускает цепочку внешних вызовов, суммарно
# занимающую десятки секунд (см. estimator._with_budget budgets).
_estimate_limiter = SlidingWindowLimiter(
limit=settings.estimate_rate_limit, window_s=settings.estimate_rate_limit_window_s
)
# ── Потолок одновременных оценок (#3082) ─────────────────────────────────────
#
# Рейт-лимит выше меряет ЧАСТОТУ (300/60с вправе стартовать в одну секунду), а
# квота — счётная и помесячная: ни один из них не ограничивает ПАРАЛЛЕЛИЗМ.
# Оценка 0.82.4с держит соединение общего пула SQLAlchemy (дефолт 5+10) и
# внешние тиры; пила одновременных оценок выедает пул и тормозит весь /api/v1/*.
# Образец — public/mera.py::_suggest_slots (4 слота на секундное автодополнение).
#
# 4 слота: вместе с 4 слотами suggest — 8 одновременно удерживаемых соединений
# из 15 возможных, остаток пула остаётся прочим ручкам. Ожидание слота 5с ≈ две
# длительности оценки: если за это время слот не освободился, очередь глубока и
# честный ответ — быстрый 429 с Retry-After, а не растущая очередь (очередь под
# нагрузкой — те же занятые соединения плюс таймаут у клиента; mera.py:117-127).
#
# Семафор живёт в памяти процесса — при переходе на несколько воркеров uvicorn
# (#3083) фактический лимит умножится на число воркеров; задачи согласовывать.
_ESTIMATE_CONCURRENCY = 4
_ESTIMATE_SLOT_WAIT_S = 5.0
_estimate_slots = asyncio.Semaphore(_ESTIMATE_CONCURRENCY)
def _resolve_quota_identity(
request: Request,
response: Response,
x_authenticated_user: str | None,
) -> tuple[str | None, int]:
"""Резолвит (quota_key, default_limit) для account_quota.* — #b2c-antiabuse-2.
- X-Authenticated-User присутствует → (username, MONTHLY_LIMIT) — существующий
pilot/admin-флоу БЕЗ изменений (персональные override в
account_quota_overrides применяются как раньше через account_quota.user_limit).
- Заголовка нет И settings.quota_dev_fail_open=True (явный dev-флаг локальной
разработки без Caddy) → (None, MONTHLY_LIMIT) — account_quota трактует None
как unlimited. Флаг по умолчанию ВЫКЛЮЧЕН — это НЕ дефолтный прод-путь.
- Заголовка нет И флаг не задан (default, прод-путь для анонимов — продукт
открывается наружу) → анонимный ключ на основе подписанной session-cookie
(app.core.anon_session) + client IP, default_limit =
settings.anon_estimate_quota_limit (гораздо строже пилот-лимита). Честно:
смена IP или чистка cookie обходит этот лимит — цель поднять стоимость
злоупотребления, а не сделать его невозможным (тот же принцип, что и во
всей account_quota-схеме, #747).
"""
if x_authenticated_user:
return x_authenticated_user, account_quota.MONTHLY_LIMIT
if settings.quota_dev_fail_open:
return None, account_quota.MONTHLY_LIMIT
session_id = get_or_create_anon_session_id(request, response)
anon_key = f"anon:{session_id}:{_client_ip(request)}"
return anon_key, settings.anon_estimate_quota_limit
# PR-D1: единственное определение «оценка читаема» — раньше SQL-фильтр (404,
# ниже в get_estimate) и Python-проверка (410, в estimate_pdf) уже разошлись
# по коду ответа; третий потребитель (`/r/<token>`, PR-9) разошёлся бы
# неизбежно без унификации. `retain_until > NOW()` при NULL даёт NULL → false
# в SQL — для всех существующих строк (retain_until IS NULL) поведение не
# меняется вообще. Не копировать это выражение по месту — только через
# константу/хелпер ниже. Payments retention, PR #2754.
ESTIMATE_READABLE_SQL = "(expires_at > NOW() OR retain_until > NOW())"
def estimate_readable(expires_at: datetime, retain_until: datetime | None) -> bool:
"""Python-зеркало ESTIMATE_READABLE_SQL — та же дизъюнкция, без похода в БД.
tzinfo-нормализация повторяет прежнюю Python-проверку (estimate_pdf) —
`.replace(tzinfo=UTC)`, не переизобретается.
"""
now = datetime.now(tz=UTC)
if expires_at.replace(tzinfo=UTC) > now:
return True
return retain_until is not None and retain_until.replace(tzinfo=UTC) > now
def _assert_estimate_access(created_by: str | None, x_authenticated_user: str | None) -> None:
"""IDOR guard (#690): только владелец оценки или admin могут её читать.
Зеркало скоупинга /history (#656). 401 — нет заголовка X-Authenticated-User
(Caddy basic_auth обязателен); 403 — юзер отсутствует в roles.yaml; 404 —
pilot читает чужую оценку (скрываем существование, не подтверждаем чужой id).
"""
if not x_authenticated_user:
raise HTTPException(
status_code=401,
detail="no authenticated user (valid session required)",
)
from app.core.auth import get_role
try:
role = get_role(x_authenticated_user)
except KeyError:
logger.warning(
"user %r authenticated via Caddy but missing from roles.yaml",
x_authenticated_user,
)
raise HTTPException(status_code=403, detail="user not in roles config") from None
if role == "admin":
return
if created_by is None or created_by != x_authenticated_user:
raise HTTPException(status_code=404, detail="estimate not found")
def _assert_estimate_access_by_id(
db: Session, estimate_id: UUID, x_authenticated_user: str | None
) -> None:
"""IDOR guard для derived-роутов, не читающих саму оценку (#690).
Тянет created_by оценки и применяет _assert_estimate_access. 404 если оценки
нет (как и owner-mismatch — существование не подтверждаем).
"""
row = db.execute(
text("SELECT created_by FROM trade_in_estimates WHERE id = CAST(:id AS uuid)"),
{"id": str(estimate_id)},
).fetchone()
if row is None:
raise HTTPException(status_code=404, detail="estimate not found")
_assert_estimate_access(row.created_by, x_authenticated_user)
def _resolve_target_house_id(
db: Session, address: str | None, lat: float | None, lon: float | None
) -> int | None:
"""Резолвит house_id целевого дома по адресу (geo fallback) — для GET-rehydrate.
Тот же паттерн, что в placement-history / house-analytics: нормализованный
short_address → houses, иначе tight 100м geo-fallback. Берём первый матч
(детерминированно — для price_trend/avito_imv достаточно одного дома).
Best-effort: None при отсутствии адреса/координат/матча (graceful).
"""
if address:
row = db.execute(
text(
"""
SELECT id FROM houses
WHERE short_address = tradein_normalize_short_addr(:addr)
OR tradein_normalize_short_addr(address) = tradein_normalize_short_addr(:addr)
LIMIT 1
"""
),
{"addr": address},
).fetchone()
if row is not None:
return row.id
if lat is not None and lon is not None:
row = db.execute(
text(
"""
SELECT id FROM houses
WHERE geom IS NOT NULL
AND ST_DWithin(
geom::geography, ST_MakePoint(:lon, :lat)::geography, 100
)
ORDER BY ST_Distance(geom::geography, ST_MakePoint(:lon, :lat)::geography)
LIMIT 1
"""
),
{"lat": lat, "lon": lon},
).fetchone()
if row is not None:
return row.id
return None
# ── Revival на GET /estimate/{id} (incident 2026-08-10) ─────────────────────
# Заказчик открыл сохранённую ссылку (?id=...) и увидел «НЕДОСТАТОЧНО ДАННЫХ»:
# запись создана ДО фикса оценщика (#oblast-E/#oblast-F, PR #2823/#2825) и
# лежит в БД мёртвой (median_price<=0/NULL), хотя тот же адрес/параметры
# сейчас честно считаются. get_estimate() ниже пытается пересчитать такую
# строку ОДИН раз (throttled) через тот же estimate_quality(), что и POST
# /estimate, и пишет результат В ТУ ЖЕ строку (id/ссылка не меняются). Живую
# строку (median_price>0) этот путь не трогает вообще — сохранённая клиенту
# цена неприкосновенна.
def _precision_to_qc_geo(precision: str | None) -> int | None:
"""Best-effort обратное отображение к estimator._qc_geo_to_precision.
AggregatedEstimate наружу отдаёт только бакетированный address_precision
(house/street/approximate), не сырой dadata.qc_geo (0..5) — тот остаётся
приватным для estimate_quality(). При revival нам нужно записать ЧТО-ТО в
колонку dadata_qc_geo, чтобы будущие (уже НЕ revival, обычные) GET той же
теперь-живой строки не откатили address_precision в None. Бакеты 2..5
(settlement/city/region/unknown) неразличимы ПОСЛЕ _qc_geo_to_precision —
2 репрезентативно для всех: тот же helper на чтении схлопывает их обратно
в тот же "approximate", наблюдаемое поведение не меняется.
"""
if precision == "house":
return 0
if precision == "street":
return 1
if precision == "approximate":
return 2
return None
def _payload_from_dead_row(row: Any) -> TradeInEstimateInput:
"""Восстанавливает вход оценки из мёртвой сохранённой строки для revival.
Только поля, реально персистящиеся в trade_in_estimates при создании
(address/lat/lon/area_m2/rooms/floor/total_floors/year_built/house_type/
repair_state/has_balcony) — CRM-only поля (ownership_type/has_mortgage)
на расчёт не влияют и не нужны здесь. radius_m НИКОГДА не персистится
(payload.radius_m живёт только в рамках одного POST-запроса, ни главный
INSERT, ни _empty_estimate его не пишут) — None здесь даёт тот же
default-каскад (DEFAULT_RADIUS_M/FALLBACK_RADIUS_M), что у подавляющего
большинства сохранённых строк (явный радиус выбирает меньшинство).
consent=None + require_consent=False у вызывающего — revival не новое
согласие физлица, а служебный recompute уже существующей записи.
"""
return TradeInEstimateInput(
address=row.address,
area_m2=float(row.area_m2),
rooms=row.rooms,
floor=row.floor,
total_floors=row.total_floors,
year_built=row.year_built,
house_type=row.house_type,
repair_state=row.repair_state,
has_balcony=row.has_balcony,
lat=row.lat,
lon=row.lon,
radius_m=None,
consent=None,
)
async def _try_revive_dead_estimate(
db: Session, estimate_id: UUID, row: Any
) -> AggregatedEstimate | None:
"""Пытается пересчитать «мёртвую» (median_price<=0/NULL) строку на месте.
Возвращает свежий AggregatedEstimate (estimate_id ПОДМЕНЁН на исходный —
id/ссылка не меняются) при успехе; None если: (а) throttle ещё не истёк /
заявку уже забрал параллельный запрос — anti-storm через атомарный
conditional `UPDATE ... RETURNING` ниже (тот же паттерн, что
account_quota.increment, #747): WHERE перепроверяет и «мертва ли строка
сейчас», и «давно ли последняя попытка» НЕПОСРЕДСТВЕННО в БД, а не по
значению, прочитанному раньше в Python — TOCTOU-гонка между двумя
параллельными GET невозможна, проигравший просто не дублирует работу;
(б) пересчёт сам дал 0 (по-прежнему недостаточно данных); (в) пересчёт
упал с исключением (сеть/геокод/что угодно). Во всех трёх случаях caller
обязан отдать сохранённую (по-прежнему мёртвую) строку как раньше — НЕ 500.
"""
claim = db.execute(
text(
"""
UPDATE trade_in_estimates
SET revival_attempted_at = NOW()
WHERE id = CAST(:id AS uuid)
AND (median_price <= 0 OR median_price IS NULL)
AND (
revival_attempted_at IS NULL
OR revival_attempted_at
< NOW() - make_interval(mins => CAST(:throttle AS integer))
)
RETURNING id
"""
),
{"id": str(estimate_id), "throttle": settings.trade_in_revival_throttle_minutes},
).fetchone()
db.commit()
if claim is None:
logger.info("estimate revival throttled/lost race: id=%s", estimate_id)
return None
from app.services.estimator import estimate_quality
try:
payload = _payload_from_dead_row(row)
result = await estimate_quality(
payload,
db,
created_by=row.created_by,
client_ip=None,
require_consent=False,
)
except Exception:
logger.exception("estimate revival failed: id=%s address=%r", estimate_id, row.address)
return None
temp_id = result.estimate_id
if result.median_price_rub <= 0:
logger.info("estimate revival still insufficient data: id=%s", estimate_id)
db.execute(
text("DELETE FROM trade_in_estimates WHERE id = CAST(:id AS uuid)"),
{"id": str(temp_id)},
)
db.commit()
return None
# estimate_quality() persists under a BRAND NEW uuid (temp_id) — it has no
# notion of "recompute this existing row". Copy the computed OUTPUT fields
# into the ORIGINAL row (id/link contract), then drop the throwaway one.
# INPUT snapshot (address/area/rooms/...) is untouched — it did not change,
# only the outputs were recomputed.
# #incident-2026-08-11: created_at is DELIBERATELY excluded from this SET —
# it is the client's original request date (printed in /history and in
# AggregatedEstimate.created_at, see app/schemas/trade_in.py:317-318), NOT
# a recompute output. It previously got clobbered with the throwaway temp
# row's created_at (=NOW() at recompute time), which also silently
# re-sorted the row to the top of `GET /history ORDER BY created_at DESC`.
# revival_completed_at (migration 256) is the audit trail for "when did a
# revival LAST successfully rewrite this row" — distinct from
# revival_attempted_at (255), which is stamped on every claim regardless
# of outcome (throttle loss / recompute failure included).
db.execute(
text(
"""
UPDATE avito_imv_evaluations
SET estimate_id = CAST(:orig AS uuid)
WHERE estimate_id = CAST(:temp AS uuid)
"""
),
{"orig": str(estimate_id), "temp": str(temp_id)},
)
db.execute(
text(
"""
UPDATE trade_in_estimates SET
median_price = :median_price,
range_low = :range_low,
range_high = :range_high,
median_price_per_m2 = :median_ppm2,
confidence = :confidence,
confidence_explanation = :explanation,
n_analogs = :n_analogs,
analogs = CAST(:analogs_json AS jsonb),
actual_deals = CAST(:deals_json AS jsonb),
sources_used = CAST(:sources_json AS jsonb),
data_freshness_minutes = :freshness,
canonical_address = :canonical_address,
house_cadnum = :house_cadnum,
house_fias_id = :house_fias_id,
dadata_qc_geo = :dadata_qc_geo,
dadata_metro = CAST(:dadata_metro_json AS jsonb),
expected_sold_price = :expected_sold_price,
expected_sold_range_low = :expected_sold_range_low,
expected_sold_range_high = :expected_sold_range_high,
expected_sold_per_m2 = :expected_sold_per_m2,
asking_to_sold_ratio = :asking_to_sold_ratio,
ratio_basis = :ratio_basis,
relaxations = CAST(:relaxations_json AS jsonb),
reliability = :reliability,
revival_completed_at = NOW()
WHERE id = CAST(:id AS uuid)
"""
),
{
"id": str(estimate_id),
"median_price": result.median_price_rub,
"range_low": result.range_low_rub,
"range_high": result.range_high_rub,
"median_ppm2": result.median_price_per_m2,
"confidence": result.confidence,
"explanation": result.confidence_explanation,
"n_analogs": result.n_analogs,
"analogs_json": json.dumps(
[a.model_dump(mode="json") for a in result.analogs], ensure_ascii=False
),
"deals_json": json.dumps(
[a.model_dump(mode="json") for a in result.actual_deals], ensure_ascii=False
),
"sources_json": json.dumps(result.sources_used, ensure_ascii=False),
"freshness": result.data_freshness_minutes,
"canonical_address": result.canonical_address,
"house_cadnum": result.house_cadnum,
"house_fias_id": result.house_fias_id,
"dadata_qc_geo": _precision_to_qc_geo(result.address_precision),
"dadata_metro_json": json.dumps(result.metro_nearest, ensure_ascii=False),
"expected_sold_price": result.expected_sold_price_rub,
"expected_sold_range_low": result.expected_sold_range_low_rub,
"expected_sold_range_high": result.expected_sold_range_high_rub,
"expected_sold_per_m2": result.expected_sold_per_m2,
"asking_to_sold_ratio": result.asking_to_sold_ratio,
"ratio_basis": result.ratio_basis,
"relaxations_json": json.dumps(result.relaxations, ensure_ascii=False),
"reliability": result.reliability,
},
)
db.execute(
text("DELETE FROM trade_in_estimates WHERE id = CAST(:id AS uuid)"),
{"id": str(temp_id)},
)
db.commit()
logger.info(
"estimate revived: id=%s median=%d n=%d confidence=%s reliability=%s",
estimate_id,
result.median_price_rub,
result.n_analogs,
result.confidence,
result.reliability,
)
# created_at on the returned object must mirror the DB row (untouched
# original request date, NOT the temp row's NOW()) — see UPDATE above.
return result.model_copy(update={"estimate_id": estimate_id, "created_at": row.created_at})
@router.post("/estimate", response_model=AggregatedEstimate)
async def estimate(
payload: TradeInEstimateInput,
request: Request,
response: Response,
db: Annotated[Session, Depends(get_db)],
x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
) -> AggregatedEstimate:
"""Реальная оценка через SQL aggregation поверх listings + deals.
1. Geocode address → lat/lon
2. PostGIS ST_DWithin радиус 800м (или 2км fallback)
3. Tukey IQR outlier filter
4. Median + Q1 + Q3 + confidence с explanation
Применяется лимит 15 успешных оценок в месяц на аккаунт (кроме admin/kopylov);
анонимные запросы (без X-Authenticated-User) — свой, гораздо более строгий
лимит на anon-сессию+IP (#b2c-antiabuse-2).
"""
# #b2c-antiabuse-2 п.4: отдельный жёсткий лимит частоты на дорогой публичный
# путь — самая дешёвая проверка первой, до квоты и до дорогой цепочки внешних
# вызовов. Ключ user:/ip: — тот же принцип, что общий RateLimitMiddleware
# (app/main.py); НЕ anon-сессия квоты (rate limit — про network-identity и
# burst-защиту capacity сервера, а не про месячный business-лимит).
_rl_key = (
f"user:{x_authenticated_user}" if x_authenticated_user else f"ip:{_client_ip(request)}"
)
_retry_after = _estimate_limiter.check(_rl_key)
if _retry_after is not None:
raise HTTPException(
status_code=429,
detail="Слишком много запросов на оценку. Попробуйте через несколько минут.",
headers={"Retry-After": str(int(_retry_after) + 1)},
)
quota_key, quota_default_limit = _resolve_quota_identity(
request, response, x_authenticated_user
)
account_quota.check_and_raise(db, quota_key, default_limit=quota_default_limit)
from app.services.estimator import estimate_quality
# #3082: слот одновременности берём ПОСЛЕ дешёвых отказов (рейт-лимит, квота)
# и ДО дорогой цепочки внешних вызовов; release — в finally ниже.
try:
await asyncio.wait_for(_estimate_slots.acquire(), timeout=_ESTIMATE_SLOT_WAIT_S)
except TimeoutError:
raise HTTPException(
status_code=429,
detail="Сервис оценки сейчас занят. Попробуйте ещё раз через несколько секунд.",
headers={"Retry-After": "5"},
) from None
# #654: ранее любое исключение estimate_quality всплывало необработанным и
# маскировалось апстрим-прокси (Caddy) как непрозрачный 502. Ловим, логируем
# через logger.exception (→ GlitchTip/Sentry получает stack trace) и отдаём
# явный 503 — так любая БУДУЩАЯ реальная ошибка становится видимой, а не
# «глотается» шлюзом. HTTPException пробрасываем как есть (это не сбой).
# created_by (#656) прокидываем в estimate_quality для скоупа /history — ТОЛЬКО
# реальный account username (НЕ anon-ключ квоты): у анонимов нет "аккаунта",
# по которому имеет смысл скоупить /history.
# ЭТАП 4 B2C (152-ФЗ): require_consent=True только когда нет
# X-Authenticated-User — сегодня rbac_guard (app/core/rbac.py) уже требует
# этот заголовок на любом non-public пути, так что эта ветка пока
# недостижима в проде (анонимный /estimate ещё не открыт другими частями
# ЭТАП 4/B2C работ) — гейт готов ЗАРАНЕЕ, на момент открытия анонимного
# доступа. client_ip — proof-of-consent (estimate_quality персистит его
# на trade_in_estimates только когда require_consent=True; B2B-пилоты
# остаются NULL, см. estimator.py::_estimate_consent_persist_fields).
try:
result = await estimate_quality(
payload,
db,
created_by=x_authenticated_user,
client_ip=_client_ip(request),
require_consent=x_authenticated_user is None,
)
except HTTPException:
raise
except Exception:
logger.exception("estimate failed for address=%r", payload.address)
raise HTTPException(
status_code=503,
detail="estimate temporarily unavailable — try again shortly",
) from None
finally:
# #3082: слот возвращаем сразу после дорогой части — инкремент квоты и
# сериализация ответа ниже дёшевы и слот держать не должны.
_estimate_slots.release()
# #747: атомарно-условный инкремент — источник истины по лимиту. check_and_raise
# выше остаётся быстрым pre-check (429 до дорогой оценки), но финальное решение
# тут: при гонке двух /estimate на used=lim-1 второй получит False.
# Не списываем квоту за пустой результат (нерезолвящийся адрес и т.п.) — иначе
# платный слот сгорает за insufficient_data=True (median=0, n_analogs=0) с HTTP 200.
if not result.insufficient_data and not account_quota.increment(
db, quota_key, default_limit=quota_default_limit
):
# Аутентифицированный путь — байт-в-байт прежнее сообщение (может не
# отражать персональный override, это pre-existing поведение, вне
# scope этого фикса). Анонимный путь — динамический текст с ПРАВИЛЬНЫМ
# anon-лимитом (#b2c-antiabuse-2), а не захардкоженным MONTHLY_LIMIT.
detail = (
account_quota.LIMIT_EXHAUSTED_MESSAGE
if x_authenticated_user
else account_quota.limit_exhausted_message(quota_default_limit)
)
raise HTTPException(status_code=429, detail=detail)
# Feature 2/3 foundation: "что искали" — обогащённая estimate_request-запись
# в user_events (адрес/площадь/комнаты + estimate_id для join с trade_in_estimates).
# schedule_event сам никогда не raises — сбой аудита не должен ронять ответ.
schedule_event(
event_type="estimate_request",
username=x_authenticated_user or "",
ip=_client_ip(request),
user_agent=request.headers.get("user-agent"),
path=str(request.url.path),
method="POST",
estimate_id=str(result.estimate_id),
payload={
"address": payload.address,
"area_m2": payload.area_m2,
"rooms": payload.rooms,
},
)
return result
@router.get("/quota", response_model=QuotaStatus)
def get_quota(
request: Request,
response: Response,
db: Annotated[Session, Depends(get_db)],
x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
) -> QuotaStatus:
"""Статус квоты оценок для текущего аккаунта (или анонимной сессии).
Возвращает limit / used / remaining / unlimited для X-Authenticated-User.
Без заголовка (прод, публичный путь) — статус СВОЕЙ анонимной квоты
(anon-session cookie + IP), а НЕ безлимитный, если явно не включён
settings.quota_dev_fail_open (#b2c-antiabuse-2, dev без Caddy).
"""
quota_key, quota_default_limit = _resolve_quota_identity(
request, response, x_authenticated_user
)
status = account_quota.get_status(db, quota_key, default_limit=quota_default_limit)
return QuotaStatus(**status)
@router.get("/estimate/{estimate_id}", response_model=AggregatedEstimate)
def get_estimate(
estimate_id: UUID,
db: Annotated[Session, Depends(get_db)],
x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
) -> AggregatedEstimate:
"""Получить сохранённую оценку по UUID (для генерации PDF).
Возвращает 404 если оценка не найдена или TTL истёк.
"""
return load_estimate(db, estimate_id, x_authenticated_user=x_authenticated_user)
def load_estimate(
db: Session,
estimate_id: UUID,
*,
x_authenticated_user: str | None,
capability_granted: bool = False,
) -> AggregatedEstimate:
"""Тело GET /estimate/{id} без FastAPI-обвязки — чтобы у ВТОРОГО права
доступа был тот же самый загрузчик, а не его копия.
`capability_granted=True` — вызывающая сторона уже доказала право доступа
ДРУГИМ способом, чем `X-Authenticated-User` + roles.yaml: capability-ссылка
`/r/<token>` (app/api/v1/payments.py) отдаёт оплаченный отчёт анониму,
у которого идентичности нет и не будет — там правом является сам
непредсказуемый токен, сверенный по `payment_entitlements`.
Флаг — ИМЕННО параметр обычной функции, а не поле запроса: у route-хендлера
`get_estimate` выше его нет, поэтому подобрать его снаружи (query/заголовком)
невозможно — включить его может только код в этом процессе. Обратное
(добавить параметр в сам хендлер с default=False) сделало бы обход IDOR-
гварда #690 доступным любому клиенту через `?capability_granted=true`.
"""
row = db.execute(
text(
f"""
SELECT id, median_price, range_low, range_high, median_price_per_m2,
confidence, confidence_explanation, n_analogs,
market_percentile,
analogs, actual_deals, sources_used, data_freshness_minutes,
expires_at, retain_until, address, lat, lon,
area_m2, rooms, floor, total_floors,
year_built, house_type, repair_state, has_balcony,
canonical_address, house_cadnum, house_fias_id,
dadata_qc_geo, dadata_metro,
expected_sold_price, expected_sold_range_low,
expected_sold_range_high, expected_sold_per_m2,
asking_to_sold_ratio, ratio_basis, created_by, created_at,
relaxations, reliability
FROM trade_in_estimates
WHERE id = CAST(:id AS uuid)
AND {ESTIMATE_READABLE_SQL}
"""
),
{"id": str(estimate_id)},
).fetchone()
if row is None:
raise HTTPException(status_code=404, detail="estimate not found or expired")
if not capability_granted:
_assert_estimate_access(row.created_by, x_authenticated_user)
# #incident-2026-08-10: строка «мертва» (median_price<=0/NULL) — посчитана
# ДО фикса оценщика (#oblast-E/#oblast-F, PR #2823/#2825). Пробуем
# пересчитать её на месте (throttled, race-safe — см. докстринг
# _try_revive_dead_estimate) через тот же путь, что и POST /estimate.
# Живую строку (median_price>0) не трогаем вообще. asyncio.run() — sync↔
# async мост (тот же паттерн, что app/scheduler_main.py): get_estimate
# остаётся `def` (Starlette гоняет его в threadpool, как сейчас), поэтому
# ОСТАЛЬНЫЕ синхронные db.execute() ниже по функции не переезжают на event
# loop — только сама попытка revival временно занимает свой поток на время
# await estimate_quality(). Любая ошибка расчёта — не 500: revived is None,
# и функция просто продолжает как раньше, отдавая сохранённую строку.
if row.median_price is None or row.median_price <= 0:
revived = asyncio.run(_try_revive_dead_estimate(db, estimate_id, row))
if revived is not None:
return revived
from app.services.estimator import (
_canonical_sources,
_cv_from_ppm2,
_fetch_dkp_corridor,
_fetch_house_imv_anchor,
_fetch_price_trend,
_qc_geo_to_precision,
_resolve_target_city,
_source_counts,
rehydrate_search_radius_m,
)
analogs = [AnalogLot(**a) for a in (row.analogs or [])]
actual_deals = [AnalogLot(**a) for a in (row.actual_deals or [])]
# #2632: search_radius_m колонкой не персистится — восстанавливаем его из
# того, что персистится (подпись каскада «радиус расширен до N м», иначе
# размах сохранённых аналогов). Без этого GET отдавал null, фронт падал на
# превью-радиус 1 км и рисовал круг, за которым лежат его же пины (прод
# 2026-08-11: 10 из 10 аналогов вне круга, самый дальний — 4381 м).
persisted_relaxations = list(getattr(row, "relaxations", None) or [])
search_radius_m = rehydrate_search_radius_m(
persisted_relaxations, [a.distance_m for a in analogs]
)
# #2043 (BE-1): CV / счётчики источников на rehydrate — best-effort из
# сохранённых analogs (top-N, усечённо: полная выборка не персистится). На
# свежей оценке (POST) считаются по полной выборке; здесь — по тому, что есть
# в строке. created_at берём из колонки (persisted).
cv = _cv_from_ppm2([a.price_per_m2 for a in analogs])
source_counts = _source_counts([a.source for a in analogs])
# #2087 (M1): sources_used рехайдрейтим тем же helper'ом, что и POST — из ТЕХ ЖЕ
# persisted analogs (листинговая часть) оценочных флагов, вычитанных из
# persisted-колонки sources_used (avito_imv/yandex_valuation/cian_valuation).
# Так один estimate_id → идентичный sources_used на POST/GET/reload, а
# source_counts.keys() ⊆ sources_used (обе части — из одного набора analogs).
# Чинит рассинхром и на СТАРЫХ строках (где колонка хранит радиусный набор +
# quarter_index): листинговая часть пересобирается из analogs, служебный шум
# отбрасывается фильтром helper'а.
sources_used = _canonical_sources((a.source for a in analogs), (row.sources_used or []))
# #696: POST-only производные поля (price_trend / avito_imv / dkp_corridor /
# last_scraped_at) на GET-rehydrate ранее были null → на shared-link / PDF /
# ?id= restore пропадали графики тренда, IMV-якорь, коридор ДКП и метка
# свежести. Пересчитываем их здесь из персистированного состояния (house_id
# ререзолвится по адресу/гео; last_scraped_at реконструируется из
# created_at data_freshness_minutes — оба поля персистятся). Всё best-effort:
# None при отсутствии данных (без регрессий по сравнению с прежним поведением).
area_f = float(row.area_m2) if row.area_m2 is not None else None
target_house_id = _resolve_target_house_id(db, row.address, row.lat, row.lon)
price_trend_raw = _fetch_price_trend(db, target_house_id=target_house_id)
price_trend = (
[PriceTrendPoint(month=p["month"], ppm2=p["ppm2"]) for p in price_trend_raw]
if price_trend_raw
else None
)
# (oblast C2): city-scope корридора на GET-rehydrate. NB: row.address здесь =
# geo.full_address (персистится на estimate-time), который РОНЯЕТ город для
# не-ЕКБ («Ленина, 1») → target_city=None → corridor unscoped/None для не-ЕКБ
# (KNOWN LIMITATION: POST-path чинит через payload.address; GET получит паритет
# когда raw payload.address начнёт персиститься — follow-up). Для ЕКБ ок; reorder
# ниже безвреден (оба source city-stripped для не-ЕКБ, оба екб для ЕКБ).
target_city = _resolve_target_city(row.address) or _resolve_target_city(row.canonical_address)
# #3051 PR-A: тот же region_code-скоуп, что и POST /estimate — гарантирует
# «регион не резолвится → DEFAULT_REGION_CODE (66)», не NULL (NULL в SQL
# обнулил бы фильтр). row.lat/row.lon персистятся с estimate-time.
target_region = (
regions_mod.region_for_point(row.lat, row.lon)
if row.lat is not None and row.lon is not None
else None
)
target_region_code = target_region.code if target_region else regions_mod.DEFAULT_REGION_CODE
dkp_raw = _fetch_dkp_corridor(
db,
address=row.address,
rooms=row.rooms,
area=area_f,
city=target_city,
region_code=target_region_code,
)
dkp_corridor = DkpCorridor(**dkp_raw) if dkp_raw else None
imv_raw = _fetch_house_imv_anchor(
db, target_house_id=target_house_id, rooms=row.rooms, area=area_f
)
avito_imv = (
AvitoImvSummary(
recommended_price=int(imv_raw["recommended_price"]),
lower_price=int(imv_raw["lower_price"]) if imv_raw.get("lower_price") else None,
higher_price=int(imv_raw["higher_price"]) if imv_raw.get("higher_price") else None,
# #3323: `is not None` (0 — самый тонкий рынок, не «неизвестно») + thin_market
# считаем тем же порогом, что POST-путь в estimator, иначе одна и та же
# оценка при переоткрытии по ссылке / в PDF теряла флаг тонкого рынка.
market_count=(
int(imv_raw["market_count"]) if imv_raw.get("market_count") is not None else None
),
thin_market=(
imv_raw.get("market_count") is not None
and int(imv_raw["market_count"]) < settings.avito_imv_thin_market_threshold
),
)
if imv_raw is not None and imv_raw.get("recommended_price")
else None
)
last_scraped_at = (
row.created_at - timedelta(minutes=row.data_freshness_minutes)
if row.created_at is not None and row.data_freshness_minutes is not None
else None
)
# ВАЖНО: возвращаем ПОЛНЫЙ набор полей. Раньше эндпоинт отдавал огрызок
# без sources_used / confidence_explanation / координат — и при открытии
# оценки по ссылке (?id=) карточка источников пустела до «0/7».
# DaData-обогащёнка (canonical/cadnum/fias/precision/metro) тоже
# rehydrate'ится из row — иначе бейдж точности + метро не показывались
# на shared-link reopen / в PDF.
return AggregatedEstimate(
estimate_id=row.id,
median_price_rub=row.median_price,
range_low_rub=row.range_low,
range_high_rub=row.range_high,
median_price_per_m2=row.median_price_per_m2,
confidence=row.confidence,
confidence_explanation=row.confidence_explanation,
n_analogs=row.n_analogs,
# #2899: getattr — тот же defensive-идиом, что у relaxations/reliability
# ниже: строка без колонки (старый in-memory double, любая выборка до
# миграции 267) деградирует в None — «позиции не знаем», — а не роняет
# ответ AttributeError'ом.
market_percentile=getattr(row, "market_percentile", None),
period_months=12,
analogs=analogs,
actual_deals=actual_deals,
expires_at=row.expires_at,
retain_until=row.retain_until,
target_address=row.address,
target_lat=row.lat,
target_lon=row.lon,
sources_used=sources_used,
data_freshness_minutes=row.data_freshness_minutes,
# #696 — POST-only производные поля, пересчитанные на rehydrate (см. выше).
last_scraped_at=last_scraped_at,
price_trend=price_trend,
avito_imv=avito_imv,
dkp_corridor=dkp_corridor,
# #648 Stage 3 — sold-correction columns rehydrated for shared-link reopen.
# (numeric ratio → float for the Pydantic field.)
expected_sold_price_rub=row.expected_sold_price,
expected_sold_range_low_rub=row.expected_sold_range_low,
expected_sold_range_high_rub=row.expected_sold_range_high,
expected_sold_per_m2=row.expected_sold_per_m2,
asking_to_sold_ratio=(
float(row.asking_to_sold_ratio) if row.asking_to_sold_ratio is not None else None
),
ratio_basis=row.ratio_basis,
area_m2=row.area_m2,
rooms=row.rooms,
floor=row.floor,
total_floors=row.total_floors,
year_built=row.year_built,
house_type=row.house_type,
repair_state=row.repair_state,
has_balcony=row.has_balcony,
canonical_address=row.canonical_address,
house_cadnum=row.house_cadnum,
house_fias_id=row.house_fias_id,
address_precision=_qc_geo_to_precision(row.dadata_qc_geo),
metro_nearest=(row.dadata_metro or []),
# #2043 (BE-1): достоверность выборки — CV, счётчики источников, дата создания.
cv=cv,
source_counts=source_counts,
created_at=row.created_at,
# PR #2823 open follow-up (fixed incident 2026-08-10, migration 255):
# relaxations/reliability теперь персистятся — GET-rehydrate больше не
# теряет красный баннер «точность снижена» при открытии по ссылке.
# getattr defensive: старые in-memory test doubles / любая строка без
# этих колонок (не должно случаться после миграции) деградируют в
# дефолт схемы (ok / []), а не падают AttributeError.
relaxations=persisted_relaxations,
reliability=getattr(row, "reliability", None) or "ok",
# #2632: фактический радиус подбора (реконструкция выше). requested_radius_m
# осознанно НЕ заполняем — payload.radius_m не персистится, и подставить
# сюда дефолт значило бы выдать догадку за то, что просил пользователь.
search_radius_m=search_radius_m,
)
@router.get("/estimate/{estimate_id}/pdf")
def estimate_pdf(
estimate_id: UUID,
db: Annotated[Session, Depends(get_db)],
x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
) -> Response:
"""Скачать 4-страничный PDF-отчёт для оценки trade-in.
Бренд PDF определяется по владельцу оценки (estimate.created_by → brand slug),
а НЕ из ?brand= query param (#7 brand-spoofing fix).
Возвращает application/pdf attachment.
404 — оценка не найдена.
410 — оценка просрочена (TTL 24ч).
"""
row = db.execute(
text(
"""
SELECT id, median_price, range_low, range_high, median_price_per_m2,
confidence, confidence_explanation, n_analogs,
market_percentile,
analogs, actual_deals, sources_used, data_freshness_minutes,
expires_at, retain_until,
address, lat, lon, area_m2, rooms, floor, total_floors,
year_built, house_type, repair_state, has_balcony,
canonical_address, house_cadnum, house_fias_id,
dadata_qc_geo, dadata_metro,
expected_sold_price, expected_sold_range_low,
expected_sold_range_high, expected_sold_per_m2,
asking_to_sold_ratio, ratio_basis, created_by,
relaxations, reliability
FROM trade_in_estimates
WHERE id = CAST(:id AS uuid)
"""
),
{"id": str(estimate_id)},
).fetchone()
if row is None:
raise HTTPException(status_code=404, detail="estimate not found")
_assert_estimate_access(row.created_by, x_authenticated_user)
# PR-D1: тот же гейт, что в get_estimate (см. ESTIMATE_READABLE_SQL) — раньше
# здесь была независимая Python-проверка expires_at, разошедшаяся с SQL-
# фильтром GET-ручки. "estimate expired (24h TTL)" убрано из текста: при
# годовом retain_until упоминание 24ч в ответе API стало бы ложью.
if not estimate_readable(row.expires_at, row.retain_until):
raise HTTPException(status_code=410, detail="estimate expired")
from app.services.estimator import _qc_geo_to_precision
analogs = [AnalogLot(**a) for a in (row.analogs or [])]
actual_deals = [AnalogLot(**a) for a in (row.actual_deals or [])]
estimate = AggregatedEstimate(
estimate_id=row.id,
median_price_rub=row.median_price,
range_low_rub=row.range_low,
range_high_rub=row.range_high,
median_price_per_m2=row.median_price_per_m2,
confidence=row.confidence,
confidence_explanation=row.confidence_explanation,
n_analogs=row.n_analogs,
# #2899: getattr — тот же defensive-идиом, что у relaxations/reliability
# ниже: строка без колонки (старый in-memory double, любая выборка до
# миграции 267) деградирует в None — «позиции не знаем», — а не роняет
# ответ AttributeError'ом.
market_percentile=getattr(row, "market_percentile", None),
# #1351: окно сделок — 12 мес (estimator.DEALS_PERIOD_MONTHS), как в POST
# /estimate и GET /estimate/{id}. Раньше PDF-ветка хардкодила 24 →
# экспортёр рисовал ложный ~2-летний диапазон в клиентском документе.
period_months=12,
analogs=analogs,
actual_deals=actual_deals,
expires_at=row.expires_at,
retain_until=row.retain_until,
target_address=row.address,
target_lat=row.lat,
target_lon=row.lon,
sources_used=row.sources_used or [],
data_freshness_minutes=row.data_freshness_minutes,
# #648 Stage 3 — sold-correction columns rehydrated so the PDF carries them.
expected_sold_price_rub=row.expected_sold_price,
expected_sold_range_low_rub=row.expected_sold_range_low,
expected_sold_range_high_rub=row.expected_sold_range_high,
expected_sold_per_m2=row.expected_sold_per_m2,
asking_to_sold_ratio=(
float(row.asking_to_sold_ratio) if row.asking_to_sold_ratio is not None else None
),
ratio_basis=row.ratio_basis,
canonical_address=row.canonical_address,
house_cadnum=row.house_cadnum,
house_fias_id=row.house_fias_id,
address_precision=_qc_geo_to_precision(row.dadata_qc_geo),
metro_nearest=(row.dadata_metro or []),
# migration 255 — та же сноска «точность снижена», что и на JSON GET,
# теперь и в PDF-регенерации сохранённой оценки (см. get_estimate).
relaxations=list(getattr(row, "relaxations", None) or []),
reliability=getattr(row, "reliability", None) or "ok",
)
input_snapshot = {
"address": row.address,
"area_m2": row.area_m2,
"rooms": row.rooms,
"floor": row.floor,
"total_floors": row.total_floors,
"year_built": row.year_built,
"house_type": row.house_type,
"repair_state": row.repair_state,
"has_balcony": row.has_balcony,
}
from app.core.auth import get_brand_for_user as _brand_for_user
from app.services.brand import get_brand as _resolve_brand
# #7 brand-spoofing fix: бренд определяется по владельцу оценки,
# а не по ?brand= query param (который был удалён из сигнатуры).
owner_brand_slug = _brand_for_user(row.created_by) if row.created_by else None
brand_obj = _resolve_brand(owner_brand_slug, db)
pdf_bytes = generate_trade_in_pdf(estimate, input_snapshot, brand=brand_obj)
filename = f"trade-in-{brand_obj.slug}-{estimate_id}.pdf"
logger.info(
"PDF generated estimate_id=%s brand=%s size=%d",
estimate_id,
brand_obj.slug,
len(pdf_bytes),
)
return Response(
content=pdf_bytes,
media_type="application/pdf",
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
)
# ── Фото квартиры (#394) ─────────────────────────────────────────────────────
_MAX_PHOTO_BYTES = 10 * 1024 * 1024 # 10 МБ на фото
_MAX_PHOTOS_PER_ESTIMATE = 12
_ALLOWED_IMAGE_TYPES = {"image/jpeg", "image/png", "image/webp", "image/heic"}
@router.post("/estimate/{estimate_id}/photos", response_model=PhotoMeta)
async def upload_photo(
estimate_id: UUID,
db: Annotated[Session, Depends(get_db)],
file: Annotated[UploadFile, File()],
x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
) -> PhotoMeta:
"""Загрузить фото квартиры к оценке (#394). Хранение в estimate_photos (bytea)."""
estimate = db.execute(
text("SELECT created_by FROM trade_in_estimates WHERE id = CAST(:id AS uuid)"),
{"id": str(estimate_id)},
).fetchone()
if estimate is None:
raise HTTPException(status_code=404, detail="estimate not found")
_assert_estimate_access(estimate.created_by, x_authenticated_user)
ctype = (file.content_type or "").lower()
if ctype not in _ALLOWED_IMAGE_TYPES:
raise HTTPException(
status_code=415, detail=f"unsupported content-type: {ctype or 'unknown'}"
)
count = db.execute(
text("SELECT count(*) FROM estimate_photos WHERE estimate_id = CAST(:id AS uuid)"),
{"id": str(estimate_id)},
).scalar_one()
if count >= _MAX_PHOTOS_PER_ESTIMATE:
raise HTTPException(
status_code=409, detail=f"photo limit reached ({_MAX_PHOTOS_PER_ESTIMATE})"
)
# #2233: читаем тело чанками с жёстким капом, а не await file.read() целиком —
# иначе multi-GB аплоад буферизуется в RAM и OOM-killed backend (mem_limit 768m, #2214).
# 413 бросается СРАЗУ при превышении, остаток тела запроса НЕ читается.
chunks: list[bytes] = []
total = 0
while chunk := await file.read(64 * 1024):
total += len(chunk)
if total > _MAX_PHOTO_BYTES:
raise HTTPException(status_code=413, detail="file too large (max 10 MB)")
chunks.append(chunk)
content = b"".join(chunks)
if not content:
raise HTTPException(status_code=400, detail="empty file")
# Sanitize: re-encode through Pillow to drop EXIF, kill polyglot payloads,
# cap dimensions. Closes finding #6 from 2026-05-24 audit.
try:
sanitized_bytes, sanitized_ctype = sanitize_image(content)
except ImageSanitizationError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
row = (
db.execute(
text(
"""
INSERT INTO estimate_photos
(estimate_id, filename, content_type, content, size_bytes)
VALUES (CAST(:eid AS uuid), :fn, :ct, :content, :sz)
RETURNING id, filename, content_type, size_bytes, uploaded_at
"""
),
{
"eid": str(estimate_id),
"fn": file.filename,
"ct": sanitized_ctype,
"content": sanitized_bytes,
"sz": len(sanitized_bytes),
},
)
.mappings()
.fetchone()
)
db.commit()
logger.info(
"photo uploaded: estimate=%s photo=%s orig_size=%d sanitized_size=%d ctype=%s",
estimate_id,
row["id"],
len(content),
len(sanitized_bytes),
sanitized_ctype,
)
return PhotoMeta(**row)
@router.get("/estimate/{estimate_id}/photos", response_model=list[PhotoMeta])
def list_photos(
estimate_id: UUID,
db: Annotated[Session, Depends(get_db)],
x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
) -> list[PhotoMeta]:
"""Список фото оценки — метаданные, без содержимого (#394)."""
_assert_estimate_access_by_id(db, estimate_id, x_authenticated_user)
rows = (
db.execute(
text(
"""
SELECT id, filename, content_type, size_bytes, uploaded_at
FROM estimate_photos
WHERE estimate_id = CAST(:id AS uuid)
ORDER BY uploaded_at
"""
),
{"id": str(estimate_id)},
)
.mappings()
.all()
)
return [PhotoMeta(**r) for r in rows]
@router.get("/estimate/{estimate_id}/photos/{photo_id}")
def get_photo(
estimate_id: UUID,
photo_id: UUID,
db: Annotated[Session, Depends(get_db)],
x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
) -> Response:
"""Отдать содержимое фото — image bytes (#394)."""
_assert_estimate_access_by_id(db, estimate_id, x_authenticated_user)
row = db.execute(
text(
"""
SELECT content, content_type
FROM estimate_photos
WHERE id = CAST(:pid AS uuid) AND estimate_id = CAST(:eid AS uuid)
"""
),
{"pid": str(photo_id), "eid": str(estimate_id)},
).fetchone()
if row is None:
raise HTTPException(status_code=404, detail="photo not found")
return Response(
content=bytes(row.content),
media_type=row.content_type,
headers={"Cache-Control": "private, max-age=3600"},
)
# ── История и кэш (#399) ─────────────────────────────────────────────────────
@router.get("/history")
def estimate_history(
db: Annotated[Session, Depends(get_db)],
limit: int = 50,
account: str | None = None,
x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
) -> list[dict[str, object]]:
"""История оценок (#399) — последние N записей trade_in_estimates.
Скоупинг (#656 — закрывает cross-pilot data-leak): non-admin видит ТОЛЬКО
свои оценки (created_by = X-Authenticated-User); legacy NULL-строки без
владельца не попадают. Admin видит все строки, либо фильтрует по ?account=<user>.
401 если заголовок отсутствует (mirror /me — Caddy basic_auth обязателен).
#2043 (BE-1): в проекцию добавлен sources_used (jsonb-массив источников) —
разлочивает колонку «ИСТОЧНИКОВ N/7» в CacheView (len(sources_used) на строку).
#2417: добавлены floor/total_floors/year_built/house_type/repair_state/
has_balcony — уже персистятся в trade_in_estimates при создании оценки
(см. TradeInEstimateInput / AggregatedEstimate), но раньше не выбирались
здесь. Nullable — legacy строки / незаполненные поля отдаются как None.
"""
if not x_authenticated_user:
raise HTTPException(
status_code=401,
detail="no authenticated user (valid session required)",
)
from app.core.auth import get_role
try:
role = get_role(x_authenticated_user)
except KeyError:
logger.warning(
"user %r authenticated via Caddy but missing from roles.yaml",
x_authenticated_user,
)
raise HTTPException(status_code=403, detail="user not in roles config") from None
params: dict[str, object] = {"limit": min(max(limit, 1), 200)}
where = ""
if role == "admin":
# Admin: все строки, либо фильтр по конкретному аккаунту.
if account:
where = "WHERE created_by = :account"
params["account"] = account
else:
# Non-admin: жёстко скоупим на свои строки; ?account игнорируется.
where = "WHERE created_by = :owner"
params["owner"] = x_authenticated_user
rows = (
db.execute(
text(
f"""
SELECT id, address, rooms, area_m2, median_price,
confidence, n_analogs, created_at,
COALESCE(sources_used, '[]'::jsonb) AS sources_used,
floor, total_floors, year_built,
house_type, repair_state, has_balcony
FROM trade_in_estimates
{where}
ORDER BY created_at DESC
LIMIT :limit
"""
),
params,
)
.mappings()
.all()
)
return [dict(r) for r in rows]
@router.get("/cache-stats")
def cache_stats(db: Annotated[Session, Depends(get_db)]) -> dict[str, object]:
"""Состояние данных и кэшей (#399) — для страницы «Кэш».
#2043 (BE-1) KPI-расширение для CacheView:
- avg_median_price — средняя headline-медиана по НЕпустым оценкам
(median_price > 0, чтобы insufficient-data нули не занижали среднее).
NULL если непустых оценок нет.
- repeat_address_pct — доля строк-оценок, чей address встречается ≥2 раз
(индикатор потенциала кэш-хитов «повторный адрес»). Считается по
trade_in_estimates с непустым address; NULL при отсутствии адресов.
NB: это честный best-effort по persisted оценкам, а не hit-rate реального
кэша (отдельного счётчика попаданий не ведём).
#2660: listings_active сам по себе врал — «активно» на проде не означает
«живо» (деактиватор протухших покрывает не все источники). Рядом отдаём
listings_active_stale — сколько из них не виделись listings_stale_days
(= LISTINGS_FRESH_DAYS эстиматора; прод 2026-08-05: 37 900 активных при
20 935 не виденных 14+ дней). Счётчик не прячем, а разделяем.
"""
from app.services.estimator import LISTINGS_FRESH_DAYS
row = (
db.execute(
text(
"""
SELECT
(SELECT count(*) FROM geocode_cache) AS geocode_cache,
(SELECT count(*) FROM geocode_cache WHERE expires_at > NOW())
AS geocode_cache_fresh,
(SELECT count(*) FROM listings WHERE is_active) AS listings_active,
(SELECT count(*) FROM listings
WHERE is_active
AND last_seen_at <= NOW() - (:fresh_days || ' days')::interval)
AS listings_active_stale,
(SELECT max(scraped_at) FROM listings) AS listings_last_scraped,
(SELECT count(*) FROM deals) AS deals,
(SELECT count(*) FROM gendesign_cad_buildings) AS cad_buildings,
(SELECT count(*) FROM house_metadata) AS house_metadata,
(SELECT count(*) FROM trade_in_estimates) AS estimates_total,
(SELECT round(avg(median_price))
FROM trade_in_estimates WHERE median_price > 0) AS avg_median_price,
(SELECT round(
100.0 * count(*) FILTER (WHERE addr_count > 1)
/ NULLIF(count(*), 0), 1)
FROM (
SELECT count(*) OVER (PARTITION BY address) AS addr_count
FROM trade_in_estimates
WHERE address IS NOT NULL AND address <> ''
) t) AS repeat_address_pct
"""
),
{"fresh_days": LISTINGS_FRESH_DAYS},
)
.mappings()
.fetchone()
)
# Порог отдаём рядом с числом — чтобы UI подписывал «не виделись N дней»,
# а не заводил второе определение свежести у себя.
return (dict(row) | {"listings_stale_days": LISTINGS_FRESH_DAYS}) if row else {}
# ── Stage 4a: house info + IMV benchmark для UI ───────────────────────────────
_HOUSE_SELECT_COLS = """
h.id AS house_id, h.source, h.ext_house_id, h.address, h.short_address,
h.lat, h.lon, h.year_built, h.total_floors, h.house_type,
h.passenger_elevators, h.cargo_elevators,
h.has_concierge, h.closed_yard, h.has_playground, h.parking_type,
h.developer_name, h.rating, h.reviews_count,
COALESCE(h.raw_characteristics, '[]'::jsonb) AS raw_characteristics
"""
@router.get("/estimate/{estimate_id}/houses", response_model=list[HouseInfoForEstimate])
def get_estimate_houses(
estimate_id: UUID,
db: Annotated[Session, Depends(get_db)],
x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
) -> list[HouseInfoForEstimate]:
"""House(s) информация для estimate.
Логика (двойной поиск, union-deduplicate):
1. Прямое совпадение по нормализованному адресу (tradein_normalize_short_addr).
2. Geo-nearby — ST_DWithin 500м, любой source (avito/derived/etc.).
Возвращаем прямой матч + nearby (dedup by id), up to ~6 домов.
Пустой список если нет matches.
"""
_assert_estimate_access_by_id(db, estimate_id, x_authenticated_user)
target = db.execute(
text(
"""
SELECT lat, lon, address FROM trade_in_estimates
WHERE id = CAST(:id AS uuid)
"""
),
{"id": str(estimate_id)},
).fetchone()
if target is None:
raise HTTPException(status_code=404, detail="estimate not found")
# Path 1: прямой матч по нормализованному адресу
direct: list[Any] = []
if target.address:
direct = list(
db.execute(
text(
f"""
SELECT DISTINCT {_HOUSE_SELECT_COLS},
0 AS distance_m
FROM houses h
WHERE h.short_address = tradein_normalize_short_addr(:addr)
OR tradein_normalize_short_addr(h.address)
= tradein_normalize_short_addr(:addr)
LIMIT 1
"""
),
{"addr": target.address},
)
.mappings()
.all()
)
# Path 2: geo-nearby (any source, 500м radius)
nearby: list[Any] = []
if target.lat is not None and target.lon is not None:
nearby = list(
db.execute(
text(
f"""
SELECT DISTINCT {_HOUSE_SELECT_COLS},
ST_Distance(
h.geom::geography,
ST_MakePoint(:lon, :lat)::geography
)::int AS distance_m
FROM houses h
WHERE h.geom IS NOT NULL
AND ST_DWithin(
h.geom::geography,
ST_MakePoint(:lon, :lat)::geography,
500
)
ORDER BY distance_m
LIMIT 5
"""
),
{"lat": target.lat, "lon": target.lon},
)
.mappings()
.all()
)
# Merge: direct первым, затем nearby (dedup by house_id)
seen_ids = {r["house_id"] for r in direct}
merged = list(direct) + [r for r in nearby if r["house_id"] not in seen_ids]
return [
HouseInfoForEstimate(**{k: v for k, v in row.items() if k != "distance_m"})
for row in merged
]
@router.get("/estimate/{estimate_id}/placement-history", response_model=list[PlacementHistoryEntry])
def get_estimate_placement_history(
estimate_id: UUID,
db: Annotated[Session, Depends(get_db)],
x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
) -> list[PlacementHistoryEntry]:
"""Историческая продажная активность по дому(ам) target estimate.
Возвращает rows из house_placement_history для всех houses связанных с
target адресом. Сортировано по last_price_date DESC.
"""
_assert_estimate_access_by_id(db, estimate_id, x_authenticated_user)
target = db.execute(
text(
"""
SELECT lat, lon, address FROM trade_in_estimates
WHERE id = CAST(:id AS uuid)
"""
),
{"id": str(estimate_id)},
).fetchone()
if target is None:
raise HTTPException(status_code=404, detail="estimate not found")
# Поиск house_ids по нормализованному адресу
house_ids: list[int] = []
if target.address:
rows = db.execute(
text(
"""
SELECT id FROM houses
WHERE short_address = tradein_normalize_short_addr(:addr)
OR tradein_normalize_short_addr(address) = tradein_normalize_short_addr(:addr)
"""
),
{"addr": target.address},
).all()
house_ids = [r.id for r in rows]
if not house_ids and target.lat is not None and target.lon is not None:
# Geo fallback: 100м (tight radius чтобы не смешать соседние дома)
rows = db.execute(
text(
"""
SELECT id FROM houses
WHERE geom IS NOT NULL
AND ST_DWithin(
geom::geography,
ST_MakePoint(:lon, :lat)::geography,
100
)
LIMIT 3
"""
),
{"lat": target.lat, "lon": target.lon},
).all()
house_ids = [r.id for r in rows]
if not house_ids:
return []
history = (
db.execute(
text(
"""
SELECT id, source, house_id, ext_item_id, title, rooms, area_m2,
floor, total_floors, start_price, start_price_date,
last_price, last_price_date, removed_date, exposure_days,
notes
FROM house_placement_history
WHERE house_id = ANY(:house_ids)
ORDER BY COALESCE(last_price_date, start_price_date) DESC NULLS LAST
LIMIT 50
"""
),
{"house_ids": house_ids},
)
.mappings()
.all()
)
return [PlacementHistoryEntry(**dict(r)) for r in history]
@router.get("/estimate/{estimate_id}/house-analytics", response_model=HouseAnalyticsResponse)
def get_estimate_house_analytics(
estimate_id: UUID,
db: Annotated[Session, Depends(get_db)],
x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
radius_m: int | None = None,
) -> HouseAnalyticsResponse:
"""House-level analytics from house_placement_history backfill.
Resolves target house(s) — если в самом доме <8 hist rows — расширяем поиск до 300м.
Возвращает: price-history by year (median ₽/м²), recent sold (12mo), KPI.
#2044 (BE-2): optional query-param radius_m (1005000) явно задаёт радиус
расширения выборки домов. None → текущая авто-логика (расширяем до 300 м
только если в самом доме <8 записей). radius_m возвращается в ответе.
"""
_assert_estimate_access_by_id(db, estimate_id, x_authenticated_user)
target = db.execute(
text("SELECT lat, lon, address FROM trade_in_estimates WHERE id = CAST(:id AS uuid)"),
{"id": str(estimate_id)},
).fetchone()
if target is None:
raise HTTPException(status_code=404, detail="estimate not found")
# 1. Resolve target house_ids (same as placement-history endpoint)
house_ids: list[int] = []
if target.address:
rows = db.execute(
text(
"SELECT id FROM houses WHERE short_address = tradein_normalize_short_addr(:addr) "
"OR tradein_normalize_short_addr(address) = tradein_normalize_short_addr(:addr)"
),
{"addr": target.address},
).all()
house_ids = [r.id for r in rows]
if not house_ids and target.lat is not None and target.lon is not None:
rows = db.execute(
text(
"SELECT id FROM houses WHERE geom IS NOT NULL AND ST_DWithin("
"geom::geography, ST_MakePoint(:lon, :lat)::geography, 100) LIMIT 3"
),
{"lat": target.lat, "lon": target.lon},
).all()
house_ids = [r.id for r in rows]
# 2. Expand search radius. #2044 (BE-2): explicit radius_m query-param
# overrides the auto 0→300 heuristic. None → byte-identical current
# behaviour (расширяем до 300 м только при <8 in-house rows).
radius_used = 0
if radius_m is not None:
expand_radius = max(100, min(radius_m, 5000))
if target.lat is not None and target.lon is not None:
rows = db.execute(
text(
"SELECT id FROM houses WHERE geom IS NOT NULL AND ST_DWithin("
"geom::geography, ST_MakePoint(:lon, :lat)::geography, :radius) LIMIT 30"
),
{"lat": target.lat, "lon": target.lon, "radius": expand_radius},
).all()
house_ids = sorted(set(house_ids) | {r.id for r in rows})
radius_used = expand_radius
else:
n_in_house = 0
if house_ids:
n_in_house = (
db.execute(
text("SELECT COUNT(*) FROM house_placement_history WHERE house_id = ANY(:ids)"),
{"ids": house_ids},
).scalar()
or 0
)
if n_in_house < 8 and target.lat is not None and target.lon is not None:
rows = db.execute(
text(
"SELECT id FROM houses WHERE geom IS NOT NULL AND ST_DWithin("
"geom::geography, ST_MakePoint(:lon, :lat)::geography, 300) LIMIT 30"
),
{"lat": target.lat, "lon": target.lon},
).all()
house_ids = sorted(set(house_ids) | {r.id for r in rows})
radius_used = 300
if not house_ids:
return HouseAnalyticsResponse(
house_ids=[],
radius_m=0,
price_history=[],
recent_sold=[],
kpi=HouseAnalyticsKpi(
total_lots=0,
sold_count=0,
sold_rate_pct=0.0,
median_exposure_days=None,
median_bargain_pct=None,
),
)
# 3. Price history by year × source (median ₽/м²)
price_history_rows = (
db.execute(
text(
"""
SELECT
EXTRACT(YEAR FROM COALESCE(last_price_date, start_price_date))::int AS year,
source,
COUNT(*) AS n_lots,
percentile_cont(0.5) WITHIN GROUP (ORDER BY last_price / NULLIF(area_m2, 0))::int
AS median_price_per_m2,
percentile_cont(0.5) WITHIN GROUP (ORDER BY last_price)::int AS median_price_rub
FROM house_placement_history
WHERE house_id = ANY(:ids)
AND last_price IS NOT NULL AND last_price > 100000
AND area_m2 IS NOT NULL AND area_m2 > 10
AND COALESCE(last_price_date, start_price_date) IS NOT NULL
GROUP BY year, source
ORDER BY year ASC, source ASC
"""
),
{"ids": house_ids},
)
.mappings()
.all()
)
# 4. Recent sold (12 months, with removed_date)
recent_sold_rows = (
db.execute(
text(
"""
SELECT id, source, rooms, area_m2, floor, start_price, last_price,
removed_date, exposure_days,
CASE WHEN start_price > 0 AND last_price IS NOT NULL
THEN ROUND((start_price - last_price)::numeric / start_price * 100, 1)
ELSE NULL END AS discount_pct
FROM house_placement_history
WHERE house_id = ANY(:ids)
AND removed_date IS NOT NULL
AND removed_date > (NOW() - INTERVAL '12 months')::date
ORDER BY removed_date DESC
LIMIT 20
"""
),
{"ids": house_ids},
)
.mappings()
.all()
)
# 5. KPI aggregate
kpi_row = (
db.execute(
text(
"""
SELECT
COUNT(*) AS total_lots,
COUNT(*) FILTER (WHERE removed_date IS NOT NULL) AS sold_count,
CASE WHEN COUNT(*) > 0
THEN ROUND(
COUNT(*) FILTER (WHERE removed_date IS NOT NULL)::numeric
/ COUNT(*) * 100,
1
)
ELSE 0 END AS sold_rate_pct,
percentile_cont(0.5) WITHIN GROUP (ORDER BY exposure_days)
FILTER (WHERE exposure_days IS NOT NULL) AS median_exposure_days,
ROUND(
percentile_cont(0.5) WITHIN GROUP (
ORDER BY (start_price - last_price)::numeric / NULLIF(start_price, 0) * 100
) FILTER (
WHERE start_price > 0
AND last_price IS NOT NULL
AND last_price != start_price
)::numeric,
1
) AS median_bargain_pct
FROM house_placement_history
WHERE house_id = ANY(:ids)
"""
),
{"ids": house_ids},
)
.mappings()
.first()
)
return HouseAnalyticsResponse(
house_ids=house_ids,
radius_m=radius_used,
price_history=[PriceHistoryYearPoint(**dict(r)) for r in price_history_rows],
recent_sold=[RecentSoldEntry(**dict(r)) for r in recent_sold_rows],
kpi=HouseAnalyticsKpi(
total_lots=kpi_row["total_lots"] or 0 if kpi_row else 0,
sold_count=kpi_row["sold_count"] or 0 if kpi_row else 0,
sold_rate_pct=float(kpi_row["sold_rate_pct"] or 0) if kpi_row else 0.0,
median_exposure_days=(
int(kpi_row["median_exposure_days"])
if kpi_row and kpi_row["median_exposure_days"] is not None
else None
),
median_bargain_pct=(
float(kpi_row["median_bargain_pct"])
if kpi_row and kpi_row["median_bargain_pct"] is not None
else None
),
),
)
@router.get(
"/estimate/{estimate_id}/cian-price-changes",
response_model=list[CianPriceChangeStats],
)
def get_estimate_cian_price_changes(
estimate_id: UUID,
db: Annotated[Session, Depends(get_db)],
x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
) -> list[CianPriceChangeStats]:
"""История изменений цены для Cian-аналогов из estimate.
Для каждого cian-аналога в estimate.analogs:
- extract cian_id из URL (/sale/flat/<id>/)
- JOIN listings (source='cian', source_id IN cian_ids)
- JOIN offer_price_history per listing_id
- Aggregate: n_changes, last_change_time, last_diff_percent, total_change_pct
Возвращает только аналоги с хотя бы одним изменением цены.
Пустой список если нет cian-аналогов или нет истории.
"""
_assert_estimate_access_by_id(db, estimate_id, x_authenticated_user)
rows = (
db.execute(
text(
"""
WITH cian_analogs AS (
SELECT DISTINCT
substring(a->>'source_url' from '/sale/flat/(\\d+)/') AS cian_id
FROM trade_in_estimates, jsonb_array_elements(analogs) a
WHERE id = CAST(:eid AS uuid)
AND a->>'source' = 'cian'
AND a->>'source_url' IS NOT NULL
AND substring(a->>'source_url' from '/sale/flat/(\\d+)/') IS NOT NULL
),
listings_resolved AS (
SELECT l.id AS listing_id, l.source_id AS cian_id,
l.price_rub::int AS current_price
FROM listings l
JOIN cian_analogs c ON l.source_id = c.cian_id
WHERE l.source = 'cian'
),
changes_agg AS (
SELECT
oph.listing_id,
COUNT(*) AS n_changes,
MAX(oph.change_time) AS last_change_time,
(array_agg(oph.diff_percent ORDER BY oph.change_time DESC))[1]
AS last_diff_percent,
(
SELECT oph2.price_rub::int
FROM offer_price_history oph2
WHERE oph2.listing_id = oph.listing_id
ORDER BY oph2.change_time ASC
LIMIT 1
) AS first_seen_price
FROM offer_price_history oph
WHERE oph.listing_id IN (SELECT listing_id FROM listings_resolved)
GROUP BY oph.listing_id
)
SELECT
lr.cian_id,
lr.listing_id,
ca.n_changes::int,
ca.last_change_time,
ca.last_diff_percent::float AS last_diff_percent,
ca.first_seen_price,
lr.current_price,
CASE
WHEN ca.first_seen_price IS NOT NULL AND ca.first_seen_price > 0
THEN ROUND(
(lr.current_price - ca.first_seen_price)::numeric
/ ca.first_seen_price * 100,
1
)::float
ELSE NULL
END AS total_change_pct
FROM listings_resolved lr
JOIN changes_agg ca ON ca.listing_id = lr.listing_id
WHERE ca.n_changes > 0
ORDER BY ca.last_change_time DESC
"""
),
{"eid": str(estimate_id)},
)
.mappings()
.all()
)
return [CianPriceChangeStats(**dict(r)) for r in rows]
@router.get(
"/estimate/{estimate_id}/sell-time-sensitivity",
response_model=SellTimeSensitivityResponse,
)
def get_estimate_sell_time_sensitivity(
estimate_id: UUID,
db: Annotated[Session, Depends(get_db)],
x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
radius_m: int | None = None,
) -> SellTimeSensitivityResponse:
"""Срок продажи в зависимости от цены к медиане дома/района.
4 бакета: -5% / медиана (±3%) / +5% / +10%. Median exposure_days + p25/p75.
Filter last_price > start_price * 0.7 — отбрасываем подозрительно
заниженные лоты (выбросы, ошибки парсинга).
#2044 (BE-2): optional query-param radius_m (1005000) явно задаёт радиус
расширения выборки домов. None → текущая авто-логика (300 м при <8 записях).
"""
_assert_estimate_access_by_id(db, estimate_id, x_authenticated_user)
# 1. Resolve house_ids (same logic as house-analytics)
target = db.execute(
text("SELECT lat, lon, address FROM trade_in_estimates WHERE id = CAST(:id AS uuid)"),
{"id": str(estimate_id)},
).fetchone()
if target is None:
raise HTTPException(status_code=404, detail="estimate not found")
house_ids: list[int] = []
if target.address:
rows = db.execute(
text(
"SELECT id FROM houses WHERE short_address = tradein_normalize_short_addr(:addr) "
"OR tradein_normalize_short_addr(address) = tradein_normalize_short_addr(:addr)"
),
{"addr": target.address},
).all()
house_ids = [r.id for r in rows]
if not house_ids and target.lat is not None and target.lon is not None:
rows = db.execute(
text(
"SELECT id FROM houses WHERE geom IS NOT NULL AND ST_DWithin("
"geom::geography, ST_MakePoint(:lon, :lat)::geography, 100) LIMIT 3"
),
{"lat": target.lat, "lon": target.lon},
).all()
house_ids = [r.id for r in rows]
# Expand search radius (same threshold as house-analytics). #2044 (BE-2):
# explicit radius_m overrides the auto 0→300 heuristic; None → byte-identical.
radius_used = 0
if radius_m is not None:
expand_radius = max(100, min(radius_m, 5000))
if target.lat is not None and target.lon is not None:
rows = db.execute(
text(
"SELECT id FROM houses WHERE geom IS NOT NULL AND ST_DWithin("
"geom::geography, ST_MakePoint(:lon, :lat)::geography, :radius) LIMIT 30"
),
{"lat": target.lat, "lon": target.lon, "radius": expand_radius},
).all()
house_ids = sorted(set(house_ids) | {r.id for r in rows})
radius_used = expand_radius
else:
n_in_house = 0
if house_ids:
n_in_house = (
db.execute(
text("SELECT COUNT(*) FROM house_placement_history WHERE house_id = ANY(:ids)"),
{"ids": house_ids},
).scalar()
or 0
)
if n_in_house < 8 and target.lat is not None and target.lon is not None:
rows = db.execute(
text(
"SELECT id FROM houses WHERE geom IS NOT NULL AND ST_DWithin("
"geom::geography, ST_MakePoint(:lon, :lat)::geography, 300) LIMIT 30"
),
{"lat": target.lat, "lon": target.lon},
).all()
house_ids = sorted(set(house_ids) | {r.id for r in rows})
radius_used = 300
if not house_ids:
return SellTimeSensitivityResponse(
house_ids=[],
radius_m=0,
target_median_price_per_m2=None,
buckets=[],
)
# 2. Compute benchmark median ₽/м² for last 2 years
target_median = db.execute(
text(
"""
SELECT percentile_cont(0.5) WITHIN GROUP (
ORDER BY last_price / NULLIF(area_m2, 0)
)::int AS median_ppm2
FROM house_placement_history
WHERE house_id = ANY(:ids)
AND last_price IS NOT NULL AND last_price > 100000
AND area_m2 > 10
AND COALESCE(last_price_date, start_price_date) > (NOW() - INTERVAL '2 years')::date
AND (start_price = 0 OR last_price > start_price * 0.7)
"""
),
{"ids": house_ids},
).scalar()
# 3. Per-year median (для расчёта premium per lot); используем CTE для bucket-расчёта
bucket_rows = (
db.execute(
text(
"""
WITH year_medians AS (
SELECT
EXTRACT(YEAR FROM COALESCE(last_price_date, start_price_date))::int AS year,
percentile_cont(0.5) WITHIN GROUP (
ORDER BY last_price / NULLIF(area_m2, 0)
) AS median_ppm2
FROM house_placement_history
WHERE house_id = ANY(:ids)
AND last_price IS NOT NULL AND area_m2 > 10
AND (start_price = 0 OR last_price > start_price * 0.7)
GROUP BY year
),
lots_with_premium AS (
SELECT
hph.exposure_days,
CASE
WHEN ym.median_ppm2 IS NULL OR ym.median_ppm2 = 0 THEN NULL
ELSE ((hph.last_price / NULLIF(hph.area_m2, 0)) - ym.median_ppm2)
/ ym.median_ppm2 * 100
END AS premium_pct
FROM house_placement_history hph
JOIN year_medians ym ON ym.year = EXTRACT(YEAR FROM
COALESCE(hph.last_price_date, hph.start_price_date))::int
WHERE hph.house_id = ANY(:ids)
AND hph.removed_date IS NOT NULL
AND hph.exposure_days IS NOT NULL
AND hph.area_m2 > 10
AND (hph.start_price = 0 OR hph.last_price > hph.start_price * 0.7)
),
bucketed AS (
SELECT
CASE
WHEN premium_pct BETWEEN -10 AND -3 THEN 'cheap'
WHEN premium_pct BETWEEN -3 AND 3 THEN 'median'
WHEN premium_pct BETWEEN 3 AND 8 THEN 'plus5'
WHEN premium_pct BETWEEN 8 AND 15 THEN 'plus10'
ELSE NULL
END AS bucket,
exposure_days
FROM lots_with_premium
WHERE premium_pct IS NOT NULL
)
SELECT
bucket,
COUNT(*) AS n_lots,
percentile_cont(0.5) WITHIN GROUP (ORDER BY exposure_days)::int
AS median_exposure_days,
percentile_cont(0.25) WITHIN GROUP (ORDER BY exposure_days)::int AS p25_days,
percentile_cont(0.75) WITHIN GROUP (ORDER BY exposure_days)::int AS p75_days
FROM bucketed
WHERE bucket IS NOT NULL
GROUP BY bucket
"""
),
{"ids": house_ids},
)
.mappings()
.all()
)
# 4. Build buckets — гарантируем все 4 даже если данных нет в bucket
bucket_map = {r["bucket"]: dict(r) for r in bucket_rows}
bucket_definitions = [
("cheap", -5.0),
("median", 0.0),
("plus5", 5.0),
("plus10", 10.0),
]
buckets: list[SellTimeBucket] = []
for label, pct in bucket_definitions:
r = bucket_map.get(label)
n_lots = r["n_lots"] if r else 0
buckets.append(
SellTimeBucket(
price_premium_label=label,
price_premium_pct=pct,
median_exposure_days=r["median_exposure_days"] if r else None,
p25_days=r["p25_days"] if r else None,
p75_days=r["p75_days"] if r else None,
n_lots=n_lots,
# #1995: малая выборка → median/p25/p75 шумные (наблюдалась
# немонотонность между бакетами, напр. +10% быстрее +5%). Честный
# флаг вместо тихого шума — фронт решает, как показать.
insufficient_data=n_lots < settings.sell_time_sensitivity_min_n_lots,
)
)
return SellTimeSensitivityResponse(
house_ids=house_ids,
radius_m=radius_used,
target_median_price_per_m2=int(target_median) if target_median else None,
buckets=buckets,
)
@router.get("/estimate/{estimate_id}/imv-benchmark", response_model=IMVBenchmarkResponse)
def get_estimate_imv_benchmark(
estimate_id: UUID,
db: Annotated[Session, Depends(get_db)],
x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
) -> IMVBenchmarkResponse:
"""Avito IMV benchmark для estimate (для UI badge «наша 6.4М · Avito 6.29М»).
Источники lookup:
1. avito_imv_evaluations WHERE estimate_id = :id (если linked в estimator)
2. Если не linked — fallback: most recent IMV для same address (TTL 24h)
"""
_assert_estimate_access_by_id(db, estimate_id, x_authenticated_user)
# Сначала пытаемся найти directly linked
row = db.execute(
text(
"""
SELECT cache_key, recommended_price, lower_price, higher_price,
market_count, fetched_at
FROM avito_imv_evaluations
WHERE estimate_id = CAST(:id AS uuid)
ORDER BY fetched_at DESC
LIMIT 1
"""
),
{"id": str(estimate_id)},
).fetchone()
if row is None:
# Fallback: same address за 24h (на случай если link не успел)
est = db.execute(
text(
"""
SELECT address FROM trade_in_estimates
WHERE id = CAST(:id AS uuid)
"""
),
{"id": str(estimate_id)},
).fetchone()
if est is None:
raise HTTPException(status_code=404, detail="estimate not found")
if est.address:
row = db.execute(
text(
"""
SELECT cache_key, recommended_price, lower_price, higher_price,
market_count, fetched_at
FROM avito_imv_evaluations
WHERE address = :address
AND fetched_at > NOW() - INTERVAL '24 hours'
ORDER BY fetched_at DESC
LIMIT 1
"""
),
{"address": est.address},
).fetchone()
if row is None:
return IMVBenchmarkResponse(available=False)
# Get our_median_price для compare
our = db.execute(
text(
"""
SELECT median_price FROM trade_in_estimates
WHERE id = CAST(:id AS uuid)
"""
),
{"id": str(estimate_id)},
).fetchone()
our_median = our.median_price if our else None
diff_pct = None
if our_median and row.recommended_price:
diff_pct = round((our_median - row.recommended_price) / row.recommended_price * 100, 1)
return IMVBenchmarkResponse(
available=True,
cache_key=row.cache_key,
recommended_price=row.recommended_price,
lower_price=row.lower_price,
higher_price=row.higher_price,
market_count=row.market_count,
fetched_at=row.fetched_at,
our_median_price=our_median,
diff_pct=diff_pct,
)
# ── Location index (issue TBD, замена сломанного location-coef #2045) ────────
@router.get("/location-index", response_model=LocationIndexResponse)
def get_location_index(
estimate_id: UUID,
db: Annotated[Session, Depends(get_db)],
x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
radius_m: int | None = None,
) -> LocationIndexResponse:
"""Location index для оценки (замена сломанного location-coef, LocationDrawer).
Резолвит lat/lon оценки, считает индекс через
app.services.location_index.compute_location_index: % отклонения медианы ₽/м²
сопоставимых активных листингов в радиусе точки от медианы ₽/м² по всему Екатеринбургу
(percentile_cont(0.5) — устойчиво к выбросам). НЕ участвует в цене — estimator.py про
этот показатель не знает (аналоги уже несут локацию в базовой цене).
404 — оценки нет / IDOR (тот же _assert_estimate_access_by_id, что и у соседних
derived-роутов). radius_m опционален (None → адаптивная лестница радиусов
RADIUS_LADDER_M, расширяется пока выборка не наберёт MIN_SAMPLE_SIZE); явное значение
клэмпится в [500, 3000] и используется РОВНО как задано (без расширения).
Честная деградация (НЕ 500, НЕ сфабрикованные значения) — см. LocationIndexResponse:
- status="out_of_coverage"у оценки нет lat/lon, ИЛИ точка вне гео-охвата продукта
(Екатеринбург).
- status="insufficient_data" — даже на максимальном радиусе сопоставимых активных
листингов меньше порога.
- poi_status="unavailable" — osm_poi_ekb_local пуста/не отрефрешена (независимо от
status выше — «что рядом» и числовой индекс деградируют раздельно).
"""
_assert_estimate_access_by_id(db, estimate_id, x_authenticated_user)
row = db.execute(
text(
"""
SELECT lat, lon
FROM trade_in_estimates
WHERE id = CAST(:id AS uuid)
"""
),
{"id": str(estimate_id)},
).fetchone()
if row is None:
raise HTTPException(status_code=404, detail="estimate not found")
if row.lat is None or row.lon is None:
logger.info(
"location_index: estimate=%s has no lat/lon — out_of_coverage fallback", estimate_id
)
return LocationIndexResponse(
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 0,
nearby_poi=[],
poi_status="unavailable",
)
from app.services.location_index import compute_location_index
resolved_radius = None if radius_m is None else max(500, min(radius_m, 3000))
result = compute_location_index(db, float(row.lat), float(row.lon), radius_m=resolved_radius)
return LocationIndexResponse(
status=result.status,
location_index_pct=result.location_index_pct,
local_median_price_per_m2=result.local_median_price_per_m2,
city_median_price_per_m2=result.city_median_price_per_m2,
sample_size=result.sample_size,
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
],
poi_status=result.poi_status,
)
# ── Street-level deals (rosreestr open dataset) ───────────────────────────────
@router.get("/street-deals", response_model=StreetDealsResponse)
def get_street_deals(
address: str,
area_m2: float,
rooms: int,
db: Annotated[Session, Depends(get_db)] = None, # type: ignore[assignment]
period_months: int = 12,
area_tolerance: float = 0.15,
) -> StreetDealsResponse:
"""ДКП-сделки Росреестра по улице целевого адреса.
Open dataset Росреестра агрегирует адреса до улицы (без номера дома).
Поэтому это per-street view, не per-house. Фильтр по rooms + area
сужает выборку до квартир-аналогов.
После PR-A (#549) таблица deals содержит только ДКП (ДДУ-первичка отфильтрована
в import-rosreestr.sh).
"""
from app.services.estimator import (
_deal_to_analog,
_percentile,
_resolve_target_city,
extract_street_name,
)
from app.tasks.asking_to_sold_ratio import area_bucket
now = datetime.now(tz=UTC)
# #1381: отображаемое окно должно совпадать с SQL-фильтром ниже, который
# использует календарный interval PostgreSQL (NOW() - N months), а не
# фиксированные 30-дневные месяцы. Считаем cutoff календарной арифметикой.
_from_month0 = (now.year * 12 + (now.month - 1)) - period_months
_from_year, _from_month_idx = divmod(_from_month0, 12)
_from_month = _from_month_idx + 1
# Клампим день для коротких месяцев (как делает PostgreSQL interval).
_from_day = min(now.day, calendar.monthrange(_from_year, _from_month)[1])
period_from: date = date(_from_year, _from_month, _from_day)
period_to: date = now.date()
def _empty() -> StreetDealsResponse:
return StreetDealsResponse(
street=None,
period_from=period_from,
period_to=period_to,
count=0,
median_price_rub=0,
median_price_per_m2=0,
range_low_rub=0,
range_high_rub=0,
deals=[],
)
street_name = extract_street_name(address)
if not street_name:
logger.warning("street-deals: could not extract street from address=%r", address)
return _empty()
area_min = area_m2 * (1.0 - area_tolerance)
area_max = area_m2 * (1.0 + area_tolerance)
# #C1 city-scope (п.3, консистентно с estimator._fetch_dkp_corridor): резолвим
# город целевого адреса через _resolve_target_city (словарь ~30 городов обл.66
# вкл. ЕКБ + sweep-города) и фильтруем сделки по этому городу — одноимённые улицы
# др. городов не контаминируют витрину. None (адрес вне словаря) → фильтр не
# применяется. `city_filter` — литерал (не user-input), значение идёт bind-параметром.
target_city = _resolve_target_city(address)
city_filter = "AND LOWER(city) = CAST(:target_city AS text)" if target_city else ""
rows = (
db.execute(
text(
f"""
SELECT address, area_m2, rooms, floor, total_floors,
price_rub, price_per_m2, deal_date, source
FROM deals
WHERE source = 'rosreestr'
AND address ILIKE :street_pattern
AND address ~* :street_regex
{city_filter}
AND rooms = CAST(:rooms AS integer)
AND area_m2 BETWEEN :area_min AND :area_max
AND deal_date > NOW() - (CAST(:period_months AS integer) || ' months')::interval
AND price_rub > 0
ORDER BY deal_date DESC
"""
),
{
"street_pattern": "%" + street_name + "%",
"street_regex": r"\m" + street_name + r"\M",
"target_city": target_city.lower() if target_city else None,
# #3256: deals.rooms — синтетика из площади (тот же CASE 30/44/62/85,
# что area_bucket), а не комнатность. Сравнение с РЕАЛЬНЫМИ комнатами
# клиента опустошает витрину «сделки на улице» ровно там, где площадь
# нетипична для комнатности (и ВСЕГДА при 5+ комнатах). Ключуем тем же
# area-бакетом — см. развёрнутый комментарий в estimator._fetch_dkp_corridor.
"rooms": area_bucket(area_m2),
"area_min": area_min,
"area_max": area_max,
"period_months": period_months,
},
)
.mappings()
.all()
)
if not rows:
logger.info(
"street-deals: no rows found street=%r rooms=%d area=%.1f±%.0f%%",
street_name,
rooms,
area_m2,
area_tolerance * 100,
)
return StreetDealsResponse(
street=street_name,
period_from=period_from,
period_to=period_to,
count=0,
median_price_rub=0,
median_price_per_m2=0,
range_low_rub=0,
range_high_rub=0,
deals=[],
)
count = len(rows)
prices_rub = sorted(float(r["price_rub"]) for r in rows)
prices_ppm2 = sorted(float(r["price_per_m2"]) for r in rows if r["price_per_m2"])
median_ppm2 = _percentile(prices_ppm2, 0.5) if prices_ppm2 else 0.0
median_price_rub = (
int(median_ppm2 * area_m2) if median_ppm2 else int(_percentile(prices_rub, 0.5))
)
range_low_rub = int(prices_rub[0])
range_high_rub = int(prices_rub[-1])
top10 = [_deal_to_analog(dict(r)) for r in rows[:10]]
logger.info(
"street-deals: street=%r rooms=%d area=%.1f count=%d median_ppm2=%.0f",
street_name,
rooms,
area_m2,
count,
median_ppm2,
)
return StreetDealsResponse(
street=street_name,
period_from=period_from,
period_to=period_to,
count=count,
median_price_rub=median_price_rub,
median_price_per_m2=int(median_ppm2),
range_low_rub=range_low_rub,
range_high_rub=range_high_rub,
deals=top10,
)
# ── Sales vs Listings (PR K — Foundation Phase 1 of issue #564) ──────────────
# #2666 гейт правдоподобия на «медианный торг». Пейринг ДКП↔объявление идёт по
# УЛИЦЕ без номера дома (data_quality="street_only", ADR #721): на длинной улице
# сделка и объявление могут стоять в разных домах и разных ценовых классах, и
# тогда discount_pct — не торг, а разница между двумя чужими друг другу лотами.
# Гард #2660 (миграция 211) убрал предвзятые пары «вторичка ↔ новостройка» и тем
# самым сделал остаток артефактов ВИДНЫМ: по `%Космонавтов%` 2-комн. медиана
# уехала с 11.9% на +36.4%, т.е. пользователю написали бы «продали на 36%
# дороже, чем просили». Здесь не чиним пейринг (это ADR-уровень), а перестаём
# показывать число, которому нельзя верить.
#
# Пороги подобраны по проду 2026-08-05 (симуляция эндпоинта на 238 РЕАЛЬНЫХ
# пользовательских запросах из trade_in_estimates — тот же address/area/rooms,
# что уходил в виджет; 128 из них дали хотя бы одну пару):
#
# MIN_PAIRS = 10 — бутстрап по 12 «плотным» группам (n ≥ 60 пар): из полной
# выборки берём подвыборку размера k и смотрим, насколько медиана подвыборки
# отклоняется от полной. p90 |отклонения|: k=5 → 18.8 п.п., k=10 → 12.0,
# k=15 → 9.9, k=20 → 8.2. Кривая ломается ровно на 10 (5→10 даёт 6.8 п.п.
# шума, 10→15 уже только 2.1, а каждые +5 к порогу стоят ещё ~8-10% улиц).
# Совпадает с уже принятым в продукте порогом малой выборки
# settings.sell_time_sensitivity_min_n_lots = 10.
#
# SANE_MIN/MAX = [60%, +20%] — асимметричны намеренно, у сторон разная природа:
# ВЕРХ. В наблюдаемом распределении 128 групп положительный хвост РАЗОРВАН:
# +11.1, +10.8, +16.9 — и дальше пусто до +33.7, +34.2, +34.6, +39.0, +52.5,
# +70.2, +81.5, +103.1. Отсечка +20% попадает в пустой промежуток, т.е. режет
# отдельный кластер, а не край континуума. Сверху её подпирает рынок: ни один
# городской бакет asking_to_sold_ratios не даёт плюса вообще (max ratio 0.9132
# = 8.7% торга), так что «продали на +20% дороже ask» уже вдвое дальше любого
# рыночно объяснимого плюса.
# НИЗ. Разрыва нет — минус идёт сплошняком от 5% до 87%, и это ожидаемо:
# у большого отрицательного торга есть механизм (занижение цены в ДКП), в
# отличие от большого плюса. Поэтому граница грубая, «заведомо не рынок»:
# худший городской бакет (студии, ratio 0.7623) = 23.8%, 60% в 2.5 раза
# глубже. Режет 6 групп из 128 (87 … 64).
#
# Цена гейта на проде: из 128 групп с парами число сохраняют 64 (50%), 59 (46%)
# теряют его по «мало пар» и ещё 5 (4%) — по диапазону. Виджет при этом остаётся:
# сделки, медиана ₽/м², диапазон и сами пары считаются мимо гейта, гаснет ровно
# строка «медианный торг», и вместо неё уходит median_discount_explanation.
#
# MIN_DISTINCT_LISTINGS = 2 (#2672) — ПАРЫ НЕ ЯВЛЯЮТСЯ НЕЗАВИСИМЫМИ НАБЛЮДЕНИЯМИ,
# и MIN_PAIRS этого не видит. DISTINCT ON подбирает по объявлению на сделку, но
# ОДНО объявление переиспользуется на многих сделках улицы: у показываемых групп
# медиана — 18 сделок на одно различное объявление. До этого порога из 64
# показываемых чисел 22 (34%) стояли на ОДНОМ объявлении (худший живой кейс —
# `Белинского` 1-комн.: 50 пар, 1 объявление, 50.6%), 50 (78%) — меньше чем на
# трёх. «50 пар» там означало не 50 наблюдений рынка, а 50 сделок, поделённых на
# ОДНУ цену предложения: число говорило о том, чем эта конкретная квартира
# отличалась от типичной сделки, а не о торге на улице.
#
# Почему именно 2, и почему порог здесь обоснован ИНАЧЕ, чем MIN_PAIRS. Разброс
# со стороны объявлений мерили джекнайфом (выкинуть одно объявление, 45 групп,
# 118 повторов): p50 3.6, p90 18.8, max 80.4 п.п. — тот же порядок, что и шум
# при 5 парах, который при выборе MIN_PAIRS сочли неприемлемым. Но на группах с
# ОДНИМ объявлением ни джекнайф, ни кластерный бутстрап не дают числа вообще:
# выкидывать нечего, пересэмплировать нечего, отклонение тождественно 0.
# Их «нулевая ошибка» — не малая ошибка, а отсутствие измерения, и агрегат по
# всем 64 группам от их добавления УЛУЧШАЛСЯ (кластер-бутстрап p90 16.0 → 11.5),
# т.е. метрика становилась тем зеленее, чем больше в ней неизмеримого. Поэтому
# 2 — не статистический выбор, а граница выразимости: ниже неё нет выборки, о
# разбросе которой можно спрашивать, и показывать число = фабриковать точность.
# Выше 2 порог уже статистический, и данные (прод 2026-08-06, те же 128 групп)
# говорят, что он должен быть выше — но ценой почти всей витрины:
# объявлений ≥ 2 → 42 группы (33%), джекнайф p90 17.4;
# объявлений ≥ 3 → 14 групп (11%), p90 10.9 (планка MIN_PAIRS — 12.0);
# объявлений ≥ 4 → 7 групп ( 5%), p90 5.5.
# Порог 3 попадал бы в принятую планку шума, но оставляет 11% витрины и всё
# равно не делает число защищаемым (ошибка со стороны СДЕЛОК никуда не делась и
# складывается с ней). Выбирать между «9% покрытия» и «выключить строку» —
# решение владельца, не гейта; здесь снимается ровно то, что не является
# наблюдением рынка в принципе. Понижать MIN_PAIRS в компенсацию нельзя:
# вернувшиеся группы стоят на тех же одном-двух объявлениях (ложная точность).
#
# SANE_MIN ужесточён 60% → 35% (#2672). Исходное подозрение «60% режет живой
# рынок» проверено и ОПРОВЕРГНУТО: до 60% проходило всё, законный механизм
# большого минуса (занижение цены в ДКП) сохранён целиком. Ошибка была в другую
# сторону — граница пропускала неправдоподобный отрицательный хвост: 26 из 64
# показываемых чисел (41%) лежали ниже 23.7%, худшего объяснимого рынком
# бакета (asking_to_sold_ratios: студии, ratio 0.7634, 1 519 сделок; ни один
# бакет не глубже), самое глубокое показываемое — 58.5%. Мы гасили «+34%» и
# показывали «58.5%», полученный из ТОГО ЖЕ артефакта пейринга. Асимметрия
# работала против пользователя: абсурдный плюс сам себя опровергает («продали
# дороже, чем просили» — виджету просто не поверят), абсурдный минус выглядит
# правдоподобно и подталкивает продавца к выводу, что его улица торгуется за
# полцены. 35% ≈ в 1.5 раза глубже худшего рыночного бакета (запас на занижение
# в ДКП сохранён) и попадает в разрыв наблюдаемого распределения 37.6 → 33.9.
# Живой кейс из ревью: Серов, Ленина 163, 2-комн., 21 пара → 46.5% показывался.
#
# ПОШТУЧНЫЙ discount_pct В СТРОКАХ ТАБЛИЦЫ (#2672 п.3). Гейт гасил сводное число,
# а таблица под ним продолжала показывать проценты, посчитанные из ТЕХ ЖЕ пар:
# на живом Космонавтове (2-комн., медиана 37.6% погашена) шесть из первых
# двенадцати строк — от +42% до +77%, и все против одной и той же цены
# предложения. Масштаб на проде 2026-08-07 (427 реальных запросов из
# trade_in_estimates, 135 групп с парами): медиана погашена у 102 групп, и в
# них видно 3 678 строк с процентом — 72.8% всех показываемых процентов.
#
# Гасим строку там, и только там, где причина — свойство САМОЙ ПАРЫ:
# а) объявлений < MIN_DISTINCT_LISTINGS — тогда столбец «разница» это
# столбец цены сделки, поделённый на одну и ту же константу: он не даёт
# ни одного наблюдения сверх уже показанных цен, но выглядит как N торгов;
# б) медиана вне санитарного диапазона — по определению медианы это
# утверждение О СТРОКАХ: половина из них ещё дальше от рынка, чем она.
# «Мало пар» строку НЕ гасит: это свойство ВЫБОРКИ, про отдельную пару оно
# ничего не говорит, а микрокопия «пар всего 4, поэтому процент в строке не
# показываем» была бы ложной причиной. Цена этого исключения — 3 группы / 20
# строк на проде, где медианы нет, а проценты в строках есть.
# Флаги (а)/(б) считаются НЕЗАВИСИМО от порядка веток гейта: порядок «мало пар
# → одно объявление → диапазон» прячет вторую причину за первой, и на проде 55
# групп гаснут как «мало пар», хотя стоят ещё и на ОДНОМ объявлении. По ветке
# гейта строки гасились бы не там, где надо.
# Цена на проде: из 5 054 строк с процентом гаснет 3 658 (72.4%), остаётся
# 1 396. Само число «медианный торг» этой правкой НЕ меняется — 33 группы из
# 135 и до, и после (замер обеих версий модуля в одном процессе на ОДНИХ И ТЕХ
# ЖЕ живых парах). Обе цены — сделки и объявления — в строке остаются:
# убирается не данные, а наша подпись «торг» под их разностью.
#
# ПОТОЛОК ГЕЙТА (знать до следующей правки — здесь НЕ чинится):
# 1. Пейринг по УЛИЦЕ, а не по дому — корень всего перечисленного (ADR #721).
# Гейт по различным объявлениям честный промежуточный шаг, а не решение:
# он убирает числа, которые не являются наблюдением, но оставшиеся всё ещё
# сравнивают сделку в одном доме с объявлением в другом.
# 2. В группах, ПРОШЕДШИХ гейт, поштучные проценты остаются как есть — включая
# 426 строк из 1 396 (31%), лежащих вне того же диапазона [35%, +20%], по
# которому мы гасим медиану. Отдельного порога для ОДНОЙ пары у нас нет:
# диапазон калиброван на медианах групп, а у одной сделки законный разброс
# шире (занижение цены в ДКП — механизм поштучный, не медианный). Считать
# его = вводить некалиброванный порог, чего #2672 прямо предостерегает.
SALES_VS_LISTINGS_MIN_PAIRS = 10
SALES_VS_LISTINGS_MIN_DISTINCT_LISTINGS = 2
SALES_VS_LISTINGS_SANE_DISCOUNT_MIN_PCT = -35.0
SALES_VS_LISTINGS_SANE_DISCOUNT_MAX_PCT = 20.0
@router.get("/sales-vs-listings", response_model=SalesVsListingsResponse)
def get_sales_vs_listings(
address: str,
area_m2: float,
rooms: int,
db: Annotated[Session, Depends(get_db)] = None, # type: ignore[assignment]
window_days: int = 180,
area_tolerance: float = 0.15,
period_months: int = 24,
) -> SalesVsListingsResponse:
"""Pairs (ДКП-сделка, listing) для улицы целевого адреса (PR K / #564).
Для каждой ДКП-сделки Росреестра в окне `period_months` пытаемся найти
matching listing на той же улице с такими же rooms / близкой area_m2 /
listing_date в окне [deal_date - window_days, deal_date + 30d grace].
Возвращаем LEFT JOIN: сделки без listing match сохраняются (listing_* = None),
чтобы вычислить linkage_rate.
discount_pct = (deal_price - listing_price) / listing_price * 100.
Отрицательный = продали дешевле asking → reasoned discount от торга.
Per-street view: Росреестр open dataset агрегирует адреса до улицы.
"""
from app.services.estimator import _percentile, _resolve_target_city, extract_street_name
def _empty(reason_street: str | None = None) -> SalesVsListingsResponse:
return SalesVsListingsResponse(
street=reason_street,
period_months=period_months,
window_days=window_days,
area_tolerance=area_tolerance,
total_deals=0,
deals_with_listings=0,
linkage_rate_pct=0.0,
median_discount_pct=None,
data_quality="no_data", # #721 ADR v3: нет сделок / улица не извлеклась
pairs=[],
)
street_name = extract_street_name(address)
if not street_name:
logger.warning("sales-vs-listings: cannot extract street from %r", address)
return _empty()
# #2583 H4 city-scope (зеркало /street-deals #C1, trade_in.py:1717): без него
# street_pattern матчит одноимённые улицы ЛЮБОГО города обл.66 на ОБЕИХ сторонах
# JOIN (deals.address / listings.address хранят "<Город>, <Улица>") — прод-аудит
# показал 49% явно чужого города + 50% NULL-city listings для проверенных стритов,
# медианный discount_pct уезжал в -60%+ на смеси рынков. target_city резолвится тем
# же словарём (~30 городов обл.66), что и street-deals; None (адрес вне словаря,
# известная H1) → фильтр не применяется на TVF-стороне (см. миграцию 205).
target_city = _resolve_target_city(address)
rows = (
db.execute(
text(
"""
SELECT
deal_id, deal_date, deal_price_rub, deal_price_per_m2,
deal_area_m2, deal_rooms, deal_floor, deal_address,
listing_id, listing_source, listing_source_url,
listing_date, listing_price_rub, listing_price_per_m2,
listing_area_m2, days_listing_to_deal, discount_pct
FROM street_sales_vs_listings(
CAST(:street_pattern AS text),
CAST(:area_m2 AS numeric),
CAST(:rooms AS integer),
CAST(:window_days AS integer),
CAST(:area_tolerance AS numeric),
CAST(:period_months AS integer),
CAST(:target_city AS text)
)
"""
),
{
"street_pattern": "%" + street_name + "%",
"area_m2": area_m2,
"rooms": rooms,
"window_days": window_days,
"area_tolerance": area_tolerance,
"period_months": period_months,
"target_city": target_city,
},
)
.mappings()
.all()
)
if not rows:
logger.info(
"sales-vs-listings: no deals street=%r rooms=%d area=%.1f period_months=%d",
street_name,
rooms,
area_m2,
period_months,
)
return _empty(reason_street=street_name)
pairs = [
SalesListingPair(
deal_id=r["deal_id"],
deal_date=r["deal_date"],
deal_price_rub=int(r["deal_price_rub"]),
deal_price_per_m2=int(r["deal_price_per_m2"] or 0),
deal_area_m2=float(r["deal_area_m2"]),
deal_rooms=int(r["deal_rooms"]),
deal_floor=r["deal_floor"],
deal_address=r["deal_address"],
listing_id=r["listing_id"],
listing_source=r["listing_source"],
listing_source_url=r["listing_source_url"],
listing_date=r["listing_date"],
listing_price_rub=(
int(r["listing_price_rub"]) if r["listing_price_rub"] is not None else None
),
listing_price_per_m2=(
int(r["listing_price_per_m2"]) if r["listing_price_per_m2"] is not None else None
),
listing_area_m2=(
float(r["listing_area_m2"]) if r["listing_area_m2"] is not None else None
),
days_listing_to_deal=r["days_listing_to_deal"],
discount_pct=(float(r["discount_pct"]) if r["discount_pct"] is not None else None),
# #1995: street_sales_vs_listings() фильтрует ТОЛЬКО source='rosreestr'
# (067_v_street_sales_vs_listings.sql) → все pairs сейчас квартальной
# precision (deal_date = period_start_date). Честная маркировка, не баг.
deal_date_precision="quarter",
)
for r in rows
]
total_deals = len(pairs)
deals_with_listings = sum(1 for p in pairs if p.listing_id is not None)
linkage_rate_pct = round(deals_with_listings / total_deals * 100, 1) if total_deals else 0.0
discounts = sorted(p.discount_pct for p in pairs if p.discount_pct is not None)
median_discount = round(_percentile(discounts, 0.5), 2) if discounts else None
# #2672: сколько РАЗЛИЧНЫХ объявлений стоит за этими парами. len(discounts)
# считает сделки, а не наблюдения рынка — одно объявление попадает в пару
# к десяткам сделок улицы (см. шапку секции).
n_distinct_listings = len(
{p.listing_id for p in pairs if p.discount_pct is not None and p.listing_id is not None}
)
# #2672 п.3: те же две проверки, но применённые к КАЖДОЙ СТРОКЕ таблицы, а не
# к сводному числу (обоснование — в шапке секции, блок «ПОШТУЧНЫЙ ПРОЦЕНТ»).
# Считаются ДО гейта, потому что гейт обнуляет median_discount, и порядок его
# веток (мало пар → одно объявление → диапазон) прячет вторую причину за
# первой: на проде 55 групп гасятся как «мало пар», хотя стоят ещё и на ОДНОМ
# объявлении. Для строк важна причина, а не то, какая ветка сработала раньше.
pairs_stand_on_one_listing = n_distinct_listings < SALES_VS_LISTINGS_MIN_DISTINCT_LISTINGS
median_is_implausible = median_discount is not None and not (
SALES_VS_LISTINGS_SANE_DISCOUNT_MIN_PCT
<= median_discount
<= SALES_VS_LISTINGS_SANE_DISCOUNT_MAX_PCT
)
# #2666 гейт правдоподобия (обоснование порогов — в шапке секции). Число либо
# отдаётся, либо гасится с объяснением ПОЧЕМУ — молча пустое поле пользователь
# прочитает как поломку, а не как честность.
median_discount_explanation: str | None = None
if median_discount is not None:
if len(discounts) < SALES_VS_LISTINGS_MIN_PAIRS:
# Формулировка — ФАКТ про выборку, а не обещание надёжности выше
# порога: 10 пар тоже не гарантия (см. «ПОТОЛОК ГЕЙТА» выше —
# пары псевдореплики), обещать «от 10 надёжно» мы не вправе.
median_discount_explanation = (
f"Медианный торг не показываем: пар «сделка ↔ объявление» всего "
f"{len(discounts)} — на такой выборке медиана гуляет на десятки "
f"процентных пунктов."
)
elif pairs_stand_on_one_listing:
# Числа стоят В КОНЦЕ клауз намеренно: «различных объявлений всего 1»
# грамматично при любом значении, «на 1 различных объявлений» — нет.
median_discount_explanation = (
f"Медианный торг не показываем: сделок {len(discounts)}, а разных "
f"объявлений для сравнения всего {n_distinct_listings} — такой процент "
f"говорит о цене одной конкретной квартиры, а не о торге на улице."
)
elif median_is_implausible:
# Типографский минус (U+2212) — как в fmtDiscount на фронте.
shown = f"{median_discount:+.1f}".replace("-", "")
# Про «пары строятся по улице, а не по дому» здесь НЕ пишем: ровно
# следующим блоком это говорит street_only-дисклеймер (карточка) /
# хвост note (v2-mappers). Проверено скриншотом — две формулировки
# подряд читались как стена текста.
median_discount_explanation = (
f"Медианный торг не показываем: расчёт дал неправдоподобное значение "
f"({shown}%) — такого торга на рынке не бывает."
)
if median_discount_explanation is not None:
logger.info(
"sales-vs-listings: median_discount gated street=%r rooms=%d "
"n_pairs=%d distinct_listings=%d value=%+.2f%%",
street_name,
rooms,
len(discounts),
n_distinct_listings,
median_discount,
)
median_discount = None
# #2672 п.3: под погашенной медианой строки таблицы продолжали показывать
# проценты из ТЕХ ЖЕ пар (живой кейс — Космонавтов: +76%, +73%, +63% против
# одной и той же цены предложения). Гасим их там, и только там, где причина —
# свойство самой пары; «мало пар» свойство ВЫБОРКИ, про отдельную строку оно
# ничего не говорит, поэтому одну строку не трогает (обоснование и цена —
# в шапке секции). Обе цены остаются в строке: мы убираем не данные, а нашу
# подпись «торг» под разностью, которой не можем ручаться.
if discounts and (pairs_stand_on_one_listing or median_is_implausible):
if pairs_stand_on_one_listing:
# Оба числа названы совместно с фразой медианы: там «сделок N», здесь
# «одна и та же цена» — читателю видно и сколько строк, и на скольких
# объявлениях они стоят.
row_explanation = (
"Проценты по каждой сделке тоже не показываем: все они считаются "
"против одной и той же цены объявления."
)
else:
# Медиана вне диапазона — это утверждение О СТРОКАХ: по определению
# медианы половина из них лежит по дальнюю сторону от неё, т.е. тоже
# вне рыночного диапазона. Значение здесь НЕ повторяем: в ветке
# диапазона оно уже названо предыдущим предложением (вышло бы дважды
# в одном абзаце), а в ветке «мало пар» мы его намеренно не
# показываем — и печатать его в пояснении было бы отказом на словах.
row_explanation = (
"Проценты по каждой сделке тоже не показываем: половина из них — "
"за пределами того, как торгуется рынок."
)
for pair in pairs:
pair.discount_pct = None
median_discount_explanation = (
f"{median_discount_explanation} {row_explanation}"
if median_discount_explanation
else row_explanation
)
logger.info(
"sales-vs-listings: per-row discount_pct gated street=%r rooms=%d rows=%d "
"distinct_listings=%d reason=%s",
street_name,
rooms,
len(discounts),
n_distinct_listings,
"one_listing" if pairs_stand_on_one_listing else "implausible_median",
)
logger.info(
"sales-vs-listings: street=%r deals=%d with_listings=%d distinct_listings=%d "
"linkage=%.1f%% median_disc=%s",
street_name,
total_deals,
deals_with_listings,
n_distinct_listings,
linkage_rate_pct,
f"{median_discount:+.2f}%" if median_discount is not None else "n/a",
)
return SalesVsListingsResponse(
street=street_name,
period_months=period_months,
window_days=window_days,
area_tolerance=area_tolerance,
total_deals=total_deals,
deals_with_listings=deals_with_listings,
linkage_rate_pct=linkage_rate_pct,
median_discount_pct=median_discount,
median_discount_explanation=median_discount_explanation,
# street_sales_vs_listings матчит по УЛИЦЕ (не по дому, #721 ADR) →
# даже при deals_with_listings>0 это street-level, не house. house_linked НЕ emit'им.
data_quality="street_only" if total_deals > 0 else "no_data",
pairs=pairs,
)
# ── Coverage probe (#2894) — бесплатный шаг лэндинга, ЦЕНЫ НЕТ ─────────────────
# До оплаты человек видит, СКОЛЬКО похожих квартир продаётся рядом и КАК БЫСТРО
# они уходят — ни одной рублёвой цифры (см. CoverageProbeResponse docstring).
# Один SQL, ноль внешних вызовов, ноль записей — ручка дешёвая специально: её
# планируется открыть анонимам отдельной задачей (#2895, со своим consent-
# гейтом). RBAC здесь НЕ трогаем — путь остаётся закрытым (не в _PUBLIC_PATHS).
# строго 1000м по ТЗ #2894 (НЕ DEFAULT_RADIUS_M эстиматора — тот допускает fallback до 2000)
COVERAGE_RADIUS_M = 1000
COVERAGE_AREA_TOLERANCE = 0.15 # ±15% площади
COVERAGE_FRESH_DAYS = 14 # объявления не старше 14 дней (тот же канон, что LISTINGS_FRESH_DAYS)
# MAJOR-2 (независимый ревью #2894): days_on_market на проде заполнена практически
# только у yandex (avito/cian/domklik — 0 заполнено) — возраст известен у меньшинства
# когорты, и на тонких когортах "медиана" считалась по 1-2 объявлениям. Ниже порога
# n_with_age медиану не отдаём (null) — не продуктовое решение, а честность при
# заведомо шумной статистике по единичным точкам.
COVERAGE_MIN_AGE_SAMPLES = 5
# 15% свежих yandex-строк имеют days_on_market > 365 (максимум 4261) — это почти
# наверняка мёртвое/забытое объявление, которое никто не снял с публикации, а не
# сигнал о реальном времени экспозиции рынка. Отбрасываем как выброс из медианы.
COVERAGE_MAX_AGE_DAYS = 365
# Списки городов и пороги — константа РЯДОМ С РУЧКОЙ (issue #2894 требование), не в БД.
#
# ⚠️ Эти списки обязаны совпадать с `OBLAST_CITIES`
# (frontend/src/lib/city-registry.ts) — тем, что человек видит в дропдауне.
# Расхождение поймано на проде 16.08.2026: Серов предлагался к выбору, но
# отсутствовал здесь, и житель Серова получал «этот адрес вне области, по
# которой мы собираем данные» — про город В ТОЙ ЖЕ области, который мы ему сами
# и предложили. Сверка теперь автоматическая, см.
# tests/test_public_mera_api.py::test_offered_cities_match_coverage_cities.
COVERAGE_GREEN_CITIES = ("Екатеринбург", "Верхняя Пышма", "Берёзовский", "Среднеуральск")
# Серов добавлен 16.08.2026: в жёлтый тир, а не в зелёный — в радиусе 15 км от
# центра 363 активных объявления (все свежие), это на порядок меньше городов
# вокруг Екатеринбурга, но заведомо не ноль.
COVERAGE_YELLOW_CITIES = ("Нижний Тагил", "Каменск-Уральский", "Первоуральск", "Ревда", "Серов")
COVERAGE_GREEN_MIN_N = 8
COVERAGE_YELLOW_MIN_N = 12
def _fold_city(name: str) -> str:
"""ёЁ→еЕ + casefold — та же normalization-идиома, что для адресов (см. #1774)."""
return name.strip().translate(str.maketrans("ёЁ", "ее")).casefold()
_COVERAGE_CITY_THRESHOLDS: dict[str, tuple[str, int]] = {
**{_fold_city(c): (c, COVERAGE_GREEN_MIN_N) for c in COVERAGE_GREEN_CITIES},
**{_fold_city(c): (c, COVERAGE_YELLOW_MIN_N) for c in COVERAGE_YELLOW_CITIES},
}
# Повторная проверка ручки #2894 (2026-08): город раньше резолвился модой
# `listings.city` найденной когорты — оказалось, что `listings.city` это город
# СВИП-контекста скрейпера (миграция 196 — колонка заполняется тем городом,
# который скрейпер обходил, не геокодом самого объявления). Замер на проде:
# в радиусе 1000 м вокруг Берёзовского 90/90 строк имеют city='Екатеринбург';
# вокруг Ревды 74/74 — city='Первоуральск'. Следствие: продавец в Берёзовском
# видел на лэндинге «Екатеринбург», а сами COVERAGE_GREEN/YELLOW_CITIES для
# городов-спутников были НЕДОСТИЖИМЫ (в БД нет ни одной строки с их city).
# Фикс — детерминированный резолв по координатам ЗАПРОСА (никакого участия
# клиента, никакой моды когорты): ближайший центроид города из списка ниже,
# если он в пределах COVERAGE_CITY_MATCH_RADIUS_KM.
#
# Координаты — константа РЯДОМ С РУЧКОЙ, не таблица в БД: единственный
# существующий кандидат на "готовый реестр городов" — это
# frontend/src/lib/city-registry.ts (OBLAST_CITIES) и backend
# geocoder.py::SVERDLOVSK_OBLAST_CITIES — оба хранят ТОЛЬКО текстовые лейблы
# (city_hint для геокодера), без координат. Заводить миграцию + таблицу ради
# статичного справочника из 8 географических центров населённых пунктов —
# оверинжиниринг; координаты (WGS84, общедоступные центры НП) живут здесь же,
# рядом с порогами, которые они резолвят.
COVERAGE_CITY_MATCH_RADIUS_KM = 25.0 # дальше — город не определён (not_covered)
_CITY_CENTROIDS_DEG: dict[str, tuple[float, float]] = {
"Екатеринбург": (56.8389, 60.6057),
"Верхняя Пышма": (56.9789, 60.5636),
"Берёзовский": (56.9096, 60.8034),
"Среднеуральск": (56.9848, 60.4759),
"Нижний Тагил": (57.9099, 59.9819),
"Каменск-Уральский": (56.4110, 61.9243),
"Первоуральск": (56.9083, 59.9483),
"Ревда": (56.7986, 59.9298),
"Серов": (59.6047, 60.5772),
}
def _haversine_km(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
"""Расстояние по большому кругу (км), радиус Земли 6371 км."""
r_earth_km = 6371.0
phi1, phi2 = math.radians(lat1), math.radians(lat2)
dphi = math.radians(lat2 - lat1)
dlambda = math.radians(lon2 - lon1)
a = math.sin(dphi / 2) ** 2 + math.cos(phi1) * math.cos(phi2) * math.sin(dlambda / 2) ** 2
return 2 * r_earth_km * math.asin(math.sqrt(a))
def _resolve_coverage_city(lat: float, lon: float) -> tuple[str, int, bool]:
"""Резолвит (display_city, threshold, is_supported) для пробы покрытия — ПО КООРДИНАТАМ.
Город = ближайший центроид из `_CITY_CENTROIDS_DEG`, если расстояние до него
< `COVERAGE_CITY_MATCH_RADIUS_KM`; иначе город не определён. Детерминированно
и без участия клиента — см. комментарий над `_CITY_CENTROIDS_DEG` про то,
почему `listings.city` (мода когорты) и `city_hint` (клиентский вход) сюда
больше НЕ допускаются в качестве источника истины.
"""
nearest_city: str | None = None
nearest_km = math.inf
for city, (clat, clon) in _CITY_CENTROIDS_DEG.items():
distance_km = _haversine_km(lat, lon, clat, clon)
if distance_km < nearest_km:
nearest_km = distance_km
nearest_city = city
if nearest_city is None or nearest_km > COVERAGE_CITY_MATCH_RADIUS_KM:
return "", 0, False
display, threshold = _COVERAGE_CITY_THRESHOLDS[_fold_city(nearest_city)]
return display, threshold, True
@router.post("/coverage", response_model=CoverageProbeResponse)
def coverage_probe(
payload: CoverageProbeInput,
db: Annotated[Session, Depends(get_db)],
) -> CoverageProbeResponse:
"""Бесплатная проба покрытия (issue #2894) — сколько похожих квартир рядом.
Когорта — тот же дедуп/cap-канон, что radius-тиры в estimator._fetch_analogs
(rn_dup по (source, source_id), rn_addr cap по адресу, реюз тех же
приватных helper'ов эстиматора — импорт локальный, как и в остальных
ручках этого файла, чтобы не тащить тяжёлый app.services.estimator
в module-level import graph): ST_DWithin 1000м, rooms точное совпадение,
area ±15%, scraped_at не старше 14 дней, is_active.
MAJOR-1 fix (независимый ревью #2894): когорта пробы обязана быть
ПОДМНОЖЕСТВОМ когорты платного эстиматора, не шире её — иначе проба честно
отвечает "ok" там, где платный расчёт увидит 0. Три предиката ниже — тот же
канон, что estimator._COMMON_WHERE (app/services/estimator.py:5441/5460) и
inline-копия Tier W (estimator.py:5910/5916/5932, radius-тир, откуда реально
берутся аналоги на 1000 м): guard новостроек, geo_precision != 'city'
(#769 Part E — city-centroid листинги без реального адреса), price_rub > 0.
В ответе НЕТ ни одной цены — см. CoverageProbeResponse docstring.
MAJOR-2 (независимый ревью #2894): days_on_market на проде фактически
заполнена только у ОДНОГО источника (yandex) — это ограничение данных, а
не продуктовое решение. n_with_age в ответе честно считает, по скольким
объявлениям взята медиана; ниже COVERAGE_MIN_AGE_SAMPLES — null (см. поле
в ответе). Значения > COVERAGE_MAX_AGE_DAYS (почти наверняка мёртвое
объявление) в расчёт медианы не берутся.
#oblast (2026-08): house_placement_history.exposure_days — реальная (не
цензурированная) экспозиция history-строк — НЕ используется здесь: это
house-level архив (join по house_id, не привязан к текущей radius/rooms/
area когорте один-в-один), а не активные листинги в подобранном радиусе;
сведение двух разных когорт усложнило бы «один дешёвый SQL» без выигрыша
в честности (у нас и так честное имя поля — age активного объявления, не
срок продажи). См. openQuestions PR #2894 при ревью.
Повторная проверка ручки (2026-08): город больше НЕ берётся из моды
`listings.city` найденной когорты и НЕ зависит от `payload.city_hint` —
оба источника ненадёжны (см. комментарий над `_CITY_CENTROIDS_DEG`).
Город резолвится детерминированно по `payload.lat/lon` через
`_resolve_coverage_city` — `city_hint` в payload остаётся только
информационным полем (см. `CoverageProbeInput.city_hint`), на результат
не влияет.
"""
from app.services.estimator import _RN_DUP_WINDOW, MAX_ANALOGS_PER_ADDRESS
area_min = payload.area_m2 * (1 - COVERAGE_AREA_TOLERANCE)
area_max = payload.area_m2 * (1 + COVERAGE_AREA_TOLERANCE)
row = (
db.execute(
text(
f"""
WITH base AS (
SELECT
days_on_market,
row_number() OVER (
PARTITION BY address ORDER BY scraped_at DESC
) AS rn_addr,
{_RN_DUP_WINDOW}
FROM listings
WHERE is_active = true
AND rooms = :rooms
AND area_m2 BETWEEN :area_min AND :area_max
AND scraped_at > NOW() - (:fresh_days || ' days')::interval
AND ST_DWithin(
geom::geography, ST_MakePoint(:lon, :lat)::geography, :radius
)
-- MAJOR-1: sync с estimator._COMMON_WHERE (5441) / Tier W (5916) —
AND price_rub > 0
-- MAJOR-1: sync с estimator._COMMON_WHERE (5460) / Tier W (5932) —
-- guard новостроек, NULL = legacy вторичка до м.011
AND (listing_segment IS NULL OR listing_segment = 'vtorichka')
-- MAJOR-1: sync с estimator Tier W (5910/5945-5948, #769 Part E) —
-- исключает city-centroid листинги без реального адреса;
-- IS DISTINCT FROM пропускает NULL (неизвестная точность)
AND (geo_precision IS DISTINCT FROM 'city')
)
SELECT
count(*) AS n_listings,
count(*) FILTER (
WHERE days_on_market IS NOT NULL
AND days_on_market <= :max_age_days
) AS n_with_age,
percentile_cont(0.5) WITHIN GROUP (ORDER BY days_on_market)
FILTER (
WHERE days_on_market IS NOT NULL
AND days_on_market <= :max_age_days
) AS median_age_days
FROM base
WHERE rn_addr <= :max_per_addr
AND rn_dup = 1
"""
),
{
"rooms": payload.rooms,
"area_min": area_min,
"area_max": area_max,
"fresh_days": COVERAGE_FRESH_DAYS,
"lat": payload.lat,
"lon": payload.lon,
"radius": COVERAGE_RADIUS_M,
"max_per_addr": MAX_ANALOGS_PER_ADDRESS,
"max_age_days": COVERAGE_MAX_AGE_DAYS,
},
)
.mappings()
.fetchone()
)
n_listings = int(row["n_listings"]) if row else 0
n_with_age = int(row["n_with_age"]) if row and row["n_with_age"] is not None else 0
median_age = (
round(row["median_age_days"])
if row is not None
and row["median_age_days"] is not None
and n_with_age >= COVERAGE_MIN_AGE_SAMPLES
else None
)
city, threshold, supported = _resolve_coverage_city(payload.lat, payload.lon)
if not supported or n_listings == 0:
status: Literal["ok", "thin", "not_covered"] = "not_covered"
# Nit-fix (повторная проверка #2894): threshold неприменим при
# not_covered — см. CoverageProbeResponse.threshold docstring. Раньше
# поддерживаемый (по координатам) город с пустой когортой отдавал
# реальный порог (8/12) вместе с not_covered — противоречило докстрингу.
threshold = 0
elif n_listings >= threshold:
status = "ok"
else:
status = "thin"
logger.info(
"coverage probe rooms=%d area=%.1f city=%r status=%s n=%d n_with_age=%d",
payload.rooms,
payload.area_m2,
city,
status,
n_listings,
n_with_age,
)
return CoverageProbeResponse(
status=status,
n_listings=n_listings,
median_listing_age_days=median_age,
n_with_age=n_with_age,
radius_m=COVERAGE_RADIUS_M,
city=city,
threshold=threshold,
)