gendesign/tradein-mvp/backend/app/api/v1/trade_in.py
Light1YT 35bd0238ef
All checks were successful
Deploy Trade-In / changes (push) Successful in 6s
Deploy Trade-In / build-frontend (push) Has been skipped
Deploy Trade-In / build-backend (push) Successful in 42s
Deploy Trade-In / deploy (push) Successful in 34s
feat(tradein): estimator additive expected_sold price (#648 S3) (#661)
2026-05-29 13:37:33 +00:00

1402 lines
53 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 logging
from datetime import UTC, date, datetime, timedelta
from typing import Annotated, Any
from uuid import UUID
from fastapi import APIRouter, Depends, File, Header, HTTPException, Response, UploadFile
from sqlalchemy import text
from sqlalchemy.orm import Session
from app.core.db import get_db
from app.schemas.trade_in import (
AggregatedEstimate,
AnalogLot,
CianPriceChangeStats,
HouseAnalyticsKpi,
HouseAnalyticsResponse,
HouseInfoForEstimate,
IMVBenchmarkResponse,
PhotoMeta,
PlacementHistoryEntry,
PriceHistoryYearPoint,
QuotaStatus,
RecentSoldEntry,
SalesListingPair,
SalesVsListingsResponse,
SellTimeBucket,
SellTimeSensitivityResponse,
StreetDealsResponse,
TradeInEstimateInput,
)
from app.services import account_quota
from app.services.exporters.trade_in_pdf import generate_trade_in_pdf
from app.services.image_sanitizer import ImageSanitizationError, sanitize_image
logger = logging.getLogger(__name__)
router = APIRouter()
@router.post("/estimate", response_model=AggregatedEstimate)
async def estimate(
payload: TradeInEstimateInput,
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).
"""
account_quota.check_and_raise(db, x_authenticated_user)
from app.services.estimator import estimate_quality
result = await estimate_quality(payload, db)
account_quota.increment(db, x_authenticated_user)
return result
@router.get("/quota", response_model=QuotaStatus)
def get_quota(
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.
Без заголовка (dev-режим без Caddy) — unlimited True, used 0.
"""
status = account_quota.get_status(db, x_authenticated_user)
return QuotaStatus(**status)
@router.get("/estimate/{estimate_id}", response_model=AggregatedEstimate)
def get_estimate(
estimate_id: UUID,
db: Annotated[Session, Depends(get_db)],
) -> AggregatedEstimate:
"""Получить сохранённую оценку по UUID (для генерации PDF).
Возвращает 404 если оценка не найдена или TTL истёк.
"""
row = db.execute(
text(
"""
SELECT id, median_price, range_low, range_high, median_price_per_m2,
confidence, confidence_explanation, n_analogs,
analogs, actual_deals, sources_used, data_freshness_minutes,
expires_at, 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
FROM trade_in_estimates
WHERE id = CAST(:id AS uuid)
AND expires_at > NOW()
"""
),
{"id": str(estimate_id)},
).fetchone()
if row is None:
raise HTTPException(status_code=404, detail="estimate not found or 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 [])]
# ВАЖНО: возвращаем ПОЛНЫЙ набор полей. Раньше эндпоинт отдавал огрызок
# без 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,
period_months=12,
analogs=analogs,
actual_deals=actual_deals,
expires_at=row.expires_at,
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 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 []),
)
@router.get("/estimate/{estimate_id}/pdf")
def estimate_pdf(
estimate_id: UUID,
db: Annotated[Session, Depends(get_db)],
brand: str | None = None,
) -> Response:
"""Скачать 4-страничный PDF-отчёт для оценки trade-in.
Возвращает 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,
analogs, actual_deals, sources_used, data_freshness_minutes,
expires_at,
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
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.expires_at.replace(tzinfo=UTC) < datetime.now(tz=UTC):
raise HTTPException(status_code=410, detail="estimate expired (24h TTL)")
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,
period_months=24,
analogs=analogs,
actual_deals=actual_deals,
expires_at=row.expires_at,
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 []),
)
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.services.brand import get_brand as _resolve_brand
brand_obj = _resolve_brand(brand, 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()],
) -> PhotoMeta:
"""Загрузить фото квартиры к оценке (#394). Хранение в estimate_photos (bytea)."""
estimate = db.execute(
text("SELECT 1 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")
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})"
)
content = await file.read()
if not content:
raise HTTPException(status_code=400, detail="empty file")
if len(content) > _MAX_PHOTO_BYTES:
raise HTTPException(status_code=413, detail="file too large (max 10 MB)")
# 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)],
) -> list[PhotoMeta]:
"""Список фото оценки — метаданные, без содержимого (#394)."""
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)],
) -> Response:
"""Отдать содержимое фото — image bytes (#394)."""
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,
) -> list[dict[str, object]]:
"""История оценок (#399) — последние N записей trade_in_estimates."""
rows = db.execute(
text(
"""
SELECT id, address, rooms, area_m2, median_price,
confidence, n_analogs, created_at
FROM trade_in_estimates
ORDER BY created_at DESC
LIMIT :limit
"""
),
{"limit": min(max(limit, 1), 200)},
).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) — для страницы «Кэш»."""
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 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
"""
)
).mappings().fetchone()
return dict(row) 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)],
) -> 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.
"""
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)],
) -> list[PlacementHistoryEntry]:
"""Историческая продажная активность по дому(ам) target estimate.
Возвращает rows из house_placement_history для всех houses связанных с
target адресом. Сортировано по last_price_date DESC.
"""
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)],
) -> 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.
"""
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 to 300m if too few hist rows
radius_used = 0
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)],
) -> 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-аналогов или нет истории.
"""
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)],
) -> SellTimeSensitivityResponse:
"""Срок продажи в зависимости от цены к медиане дома/района.
4 бакета: -5% / медиана (±3%) / +5% / +10%. Median exposure_days + p25/p75.
Filter last_price > start_price * 0.7 — отбрасываем подозрительно
заниженные лоты (выбросы, ошибки парсинга).
"""
# 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 to 300m if too few rows (same threshold as house-analytics)
radius_used = 0
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)
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=r["n_lots"] if r else 0,
)
)
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)],
) -> 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)
"""
# Сначала пытаемся найти 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,
)
# ── 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, extract_street_name
now = datetime.now(tz=UTC)
period_from: date = (now - timedelta(days=period_months * 30)).date()
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)
rows = db.execute(
text(
"""
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 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 + "%",
"rooms": rooms,
"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) ──────────────
@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, 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,
pairs=[],
)
street_name = extract_street_name(address)
if not street_name:
logger.warning("sales-vs-listings: cannot extract street from %r", address)
return _empty()
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)
)
"""
),
{
"street_pattern": "%" + street_name + "%",
"area_m2": area_m2,
"rooms": rooms,
"window_days": window_days,
"area_tolerance": area_tolerance,
"period_months": period_months,
},
).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
),
)
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
)
logger.info(
"sales-vs-listings: street=%r deals=%d with_listings=%d linkage=%.1f%% median_disc=%s",
street_name, total_deals, deals_with_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,
pairs=pairs,
)