a7d010173f
#6: аналоги сортируются по relevance — расстояние + близость года постройки + совпадение типа дома (соразмерные квартиры из встречи Птицы). #7: медиана корректируется коэффициентом по состоянию ремонта (требует ремонта 0.92 ... евроремонт 1.08).
519 lines
21 KiB
Python
519 lines
21 KiB
Python
"""Trade-In Estimator — реальное SQL aggregation поверх listings + deals.
|
||
|
||
Заменяет старый _mock_estimate() из api/v1/trade_in.py.
|
||
|
||
Алгоритм:
|
||
1. Geocode address → (lat, lon)
|
||
2. SELECT listings с фильтрами:
|
||
- PostGIS ST_DWithin (geom, point, 800m) — радиус поиска
|
||
- rooms = target_rooms (точное совпадение)
|
||
- area_m2 BETWEEN target × 0.85 AND target × 1.15
|
||
- scraped_at > NOW() - 14 days (свежие)
|
||
- is_active = true
|
||
3. Tukey outlier filter (1.5 × IQR rule)
|
||
4. Median / Q1 / Q3 / count → confidence
|
||
5. То же для deals (period = 12 mo).
|
||
6. Сохранить в trade_in_estimates + вернуть AggregatedEstimate
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
from datetime import UTC, datetime, timedelta
|
||
from typing import Any
|
||
from uuid import UUID, uuid4
|
||
|
||
from sqlalchemy import text
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.schemas.trade_in import AggregatedEstimate, AnalogLot, TradeInEstimateInput
|
||
from app.services.geocoder import GeocodeResult, geocode
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
# ── Constants ────────────────────────────────────────────────────────────────
|
||
DEFAULT_RADIUS_M = 1000 # ПО ВСТРЕЧЕ ПТИЦЫ: «локация не дальше 800-1000 м»
|
||
FALLBACK_RADIUS_M = 2000
|
||
AREA_TOLERANCE = 0.15 # ±15% площади
|
||
LISTINGS_FRESH_DAYS = 14 # объявления не старше 14 дней
|
||
DEALS_PERIOD_MONTHS = 12 # сделки за последний год
|
||
|
||
# Поправочные коэффициенты на состояние ремонта. Аналоги в выборке — микс
|
||
# состояний (≈ "стандартный/косметический"), коэффициент сдвигает медиану под
|
||
# конкретный ремонт целевой квартиры. Встреча Птицы: ремонт влияет на цену.
|
||
_REPAIR_COEF: dict[str, float] = {
|
||
"needs_repair": 0.92, # требует ремонта — ниже рынка
|
||
"standard": 0.98,
|
||
"good": 1.03,
|
||
"excellent": 1.08, # евроремонт — выше рынка
|
||
}
|
||
_REPAIR_LABEL: dict[str | None, str] = {
|
||
"needs_repair": "требует ремонта",
|
||
"standard": "стандартный ремонт",
|
||
"good": "хороший ремонт",
|
||
"excellent": "евроремонт",
|
||
}
|
||
|
||
|
||
def _repair_coefficient(repair_state: str | None) -> float:
|
||
"""Множитель к медиане по состоянию ремонта. None → 1.0 (без поправки)."""
|
||
if not repair_state:
|
||
return 1.0
|
||
return _REPAIR_COEF.get(repair_state, 1.0)
|
||
|
||
|
||
# ── Public ───────────────────────────────────────────────────────────────────
|
||
async def estimate_quality(
|
||
payload: TradeInEstimateInput, db: Session
|
||
) -> AggregatedEstimate:
|
||
"""Главная функция — оценка квартиры по реальным данным.
|
||
|
||
Returns:
|
||
AggregatedEstimate с estimate_id, медианой, диапазоном, аналогами, сделками.
|
||
"""
|
||
# 1. Geocode
|
||
geo: GeocodeResult | None = None
|
||
if payload.address:
|
||
geo = await geocode(payload.address, db)
|
||
|
||
if geo is None:
|
||
# Без координат не можем искать через PostGIS. Возвращаем low confidence.
|
||
logger.warning("geocode failed for %s — returning low-confidence estimate", payload.address)
|
||
return _empty_estimate(payload, reason="address_not_geocoded")
|
||
|
||
# 2. Three-tier fallback:
|
||
# a) 800m + ±15% area
|
||
# b) 2km + ±15% area (fallback_used = True)
|
||
# c) 2km + ±25% area (fallback_used = True, area_widened = True)
|
||
listings, fallback_used = _fetch_analogs(
|
||
db, lat=geo.lat, lon=geo.lon, rooms=payload.rooms, area=payload.area_m2,
|
||
radius_m=DEFAULT_RADIUS_M,
|
||
year_built=payload.year_built, house_type=payload.house_type,
|
||
)
|
||
area_widened = False
|
||
|
||
if len(listings) < 5:
|
||
listings_wide, _ = _fetch_analogs(
|
||
db, lat=geo.lat, lon=geo.lon, rooms=payload.rooms, area=payload.area_m2,
|
||
radius_m=FALLBACK_RADIUS_M,
|
||
year_built=payload.year_built, house_type=payload.house_type,
|
||
)
|
||
if len(listings_wide) > len(listings):
|
||
listings = listings_wide
|
||
fallback_used = True
|
||
|
||
# Tier C: если даже на 2км мало — расширяем area tolerance до ±25%
|
||
# (актуально для отдалённых районов / новостроек с нестандартной планировкой)
|
||
if len(listings) < 3:
|
||
listings_widearea, _ = _fetch_analogs(
|
||
db, lat=geo.lat, lon=geo.lon, rooms=payload.rooms, area=payload.area_m2,
|
||
radius_m=FALLBACK_RADIUS_M, area_tolerance=0.25,
|
||
year_built=payload.year_built, house_type=payload.house_type,
|
||
)
|
||
if len(listings_widearea) > len(listings):
|
||
listings = listings_widearea
|
||
fallback_used = True
|
||
area_widened = True
|
||
|
||
# 3. Outlier filter
|
||
listings_clean = _filter_outliers(listings)
|
||
|
||
# 4. Aggregation
|
||
if listings_clean:
|
||
prices_ppm2 = sorted(lot["price_per_m2"] for lot in listings_clean if lot["price_per_m2"])
|
||
median_ppm2 = _percentile(prices_ppm2, 0.5)
|
||
q1_ppm2 = _percentile(prices_ppm2, 0.25)
|
||
q3_ppm2 = _percentile(prices_ppm2, 0.75)
|
||
median_price = int(median_ppm2 * payload.area_m2)
|
||
range_low = int(q1_ppm2 * payload.area_m2)
|
||
range_high = int(q3_ppm2 * payload.area_m2)
|
||
n_analogs = len(listings_clean)
|
||
else:
|
||
median_ppm2 = 0
|
||
median_price = 0
|
||
range_low = 0
|
||
range_high = 0
|
||
n_analogs = 0
|
||
|
||
# 4b. Поправка на состояние ремонта (встреча Птицы: ремонт влияет на цену).
|
||
# Аналоги — микс состояний; коэффициент сдвигает оценку под ремонт клиента.
|
||
repair_coef = _repair_coefficient(payload.repair_state)
|
||
repair_note = ""
|
||
if listings_clean and repair_coef != 1.0:
|
||
median_price = int(median_price * repair_coef)
|
||
range_low = int(range_low * repair_coef)
|
||
range_high = int(range_high * repair_coef)
|
||
median_ppm2 = median_ppm2 * repair_coef
|
||
pct = int(round((repair_coef - 1.0) * 100))
|
||
repair_note = (
|
||
f" Цена скорректирована на состояние ремонта "
|
||
f"({_REPAIR_LABEL.get(payload.repair_state, '')} {pct:+d}%)."
|
||
)
|
||
|
||
confidence, explanation = _compute_confidence(
|
||
n_analogs, median_ppm2, q1_ppm2 if listings_clean else 0,
|
||
q3_ppm2 if listings_clean else 0, fallback_used, area_widened,
|
||
)
|
||
explanation = (explanation or "") + repair_note
|
||
|
||
# 5. Deals — фактические сделки за период
|
||
deals = _fetch_deals(
|
||
db, lat=geo.lat, lon=geo.lon, rooms=payload.rooms, area=payload.area_m2,
|
||
radius_m=DEFAULT_RADIUS_M,
|
||
)
|
||
|
||
# 6. Сохраняем в trade_in_estimates
|
||
estimate_id = uuid4()
|
||
now = datetime.now(tz=UTC)
|
||
expires_at = now + timedelta(hours=24)
|
||
|
||
analogs_lots = [_listing_to_analog(lot) for lot in listings_clean[:10]]
|
||
deals_lots = [_deal_to_analog(d) for d in deals[:10]]
|
||
|
||
sources_used_pre = sorted({lot.get("source") for lot in listings_clean if lot.get("source")})
|
||
freshness_pre = _compute_freshness_minutes(listings_clean)
|
||
db.execute(
|
||
text(
|
||
"""
|
||
INSERT INTO trade_in_estimates (
|
||
id, address, lat, lon,
|
||
area_m2, rooms, floor, total_floors,
|
||
year_built, house_type, repair_state, has_balcony,
|
||
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
|
||
) VALUES (
|
||
CAST(:id AS uuid),
|
||
:address, :lat, :lon,
|
||
:area, :rooms, :floor, :total_floors,
|
||
:year_built, :house_type, :repair_state, :has_balcony,
|
||
:median_price, :range_low, :range_high, :median_ppm2,
|
||
:confidence, :explanation, :n_analogs,
|
||
CAST(:analogs_json AS jsonb),
|
||
CAST(:deals_json AS jsonb),
|
||
CAST(:sources_json AS jsonb),
|
||
:freshness,
|
||
:expires_at
|
||
)
|
||
"""
|
||
),
|
||
{
|
||
"id": str(estimate_id),
|
||
"address": geo.full_address,
|
||
"lat": geo.lat,
|
||
"lon": geo.lon,
|
||
"area": payload.area_m2,
|
||
"rooms": payload.rooms,
|
||
"floor": payload.floor,
|
||
"total_floors": payload.total_floors,
|
||
"year_built": payload.year_built,
|
||
"house_type": payload.house_type,
|
||
"repair_state": payload.repair_state,
|
||
"has_balcony": payload.has_balcony,
|
||
"median_price": median_price,
|
||
"range_low": range_low,
|
||
"range_high": range_high,
|
||
"median_ppm2": int(median_ppm2),
|
||
"confidence": confidence,
|
||
"explanation": explanation,
|
||
"n_analogs": n_analogs,
|
||
"analogs_json": json.dumps(
|
||
[a.model_dump(mode="json") for a in analogs_lots], ensure_ascii=False
|
||
),
|
||
"deals_json": json.dumps(
|
||
[a.model_dump(mode="json") for a in deals_lots], ensure_ascii=False
|
||
),
|
||
"sources_json": json.dumps(sources_used_pre, ensure_ascii=False),
|
||
"freshness": freshness_pre,
|
||
"expires_at": expires_at,
|
||
},
|
||
)
|
||
db.commit()
|
||
|
||
logger.info(
|
||
"estimate: id=%s addr=%s rooms=%d area=%.1f → median=%d (n=%d, conf=%s)",
|
||
estimate_id,
|
||
geo.full_address[:60],
|
||
payload.rooms,
|
||
payload.area_m2,
|
||
median_price,
|
||
n_analogs,
|
||
confidence,
|
||
)
|
||
|
||
sources_used = sorted({lot.source for lot in analogs_lots if lot.source})
|
||
freshness_min = _compute_freshness_minutes(listings_clean)
|
||
|
||
return AggregatedEstimate(
|
||
estimate_id=estimate_id,
|
||
median_price_rub=median_price,
|
||
range_low_rub=range_low,
|
||
range_high_rub=range_high,
|
||
median_price_per_m2=int(median_ppm2),
|
||
confidence=confidence,
|
||
confidence_explanation=explanation,
|
||
n_analogs=n_analogs,
|
||
period_months=DEALS_PERIOD_MONTHS,
|
||
analogs=analogs_lots,
|
||
actual_deals=deals_lots,
|
||
expires_at=expires_at,
|
||
target_address=geo.full_address,
|
||
target_lat=geo.lat,
|
||
target_lon=geo.lon,
|
||
sources_used=sources_used,
|
||
data_freshness_minutes=freshness_min,
|
||
)
|
||
|
||
|
||
def _compute_freshness_minutes(lots: list[dict[str, Any]]) -> int | None:
|
||
"""Минут с последнего парсинга — для UI «обновлено N мин назад»."""
|
||
if not lots:
|
||
return None
|
||
from datetime import datetime as _dt
|
||
|
||
now = _dt.now(tz=UTC)
|
||
scraped = [lot.get("scraped_at") or lot.get("listing_date") for lot in lots]
|
||
scraped_dt: list[datetime] = []
|
||
for s in scraped:
|
||
if s is None:
|
||
continue
|
||
# listings rows из mappings — scraped_at это datetime, не date
|
||
if hasattr(s, "tzinfo"):
|
||
scraped_dt.append(s if s.tzinfo else s.replace(tzinfo=UTC))
|
||
if not scraped_dt:
|
||
return None
|
||
return int((now - max(scraped_dt)).total_seconds() / 60)
|
||
|
||
|
||
# ── Internals ────────────────────────────────────────────────────────────────
|
||
def _fetch_analogs(
|
||
db: Session, *, lat: float, lon: float, rooms: int, area: float, radius_m: int,
|
||
area_tolerance: float = AREA_TOLERANCE,
|
||
year_built: int | None = None, house_type: str | None = None,
|
||
) -> tuple[list[dict[str, Any]], bool]:
|
||
"""SELECT аналогов с PostGIS distance + house-match relevance.
|
||
|
||
House-match (встреча Птицы — «соразмерные квартиры»): сортировка не просто
|
||
по расстоянию, а по relevance-скору, где учитывается близость года постройки
|
||
и совпадение типа дома. Так аналог «рядом + та же эпоха дома» побеждает
|
||
аналог «чуть ближе, но дом на 30 лет старше».
|
||
|
||
Returns:
|
||
(list_of_listings_as_dicts, fallback_radius_used_flag)
|
||
"""
|
||
rows = db.execute(
|
||
text(
|
||
"""
|
||
SELECT
|
||
source, source_url, address, lat, lon,
|
||
rooms, area_m2, floor, total_floors,
|
||
price_rub, price_per_m2,
|
||
listing_date, days_on_market, photo_urls,
|
||
scraped_at,
|
||
ST_Distance(geom::geography, ST_MakePoint(:lon, :lat)::geography) AS distance_m
|
||
FROM listings
|
||
WHERE ST_DWithin(geom::geography, ST_MakePoint(:lon, :lat)::geography, :radius)
|
||
AND rooms = :rooms
|
||
AND area_m2 BETWEEN :area_min AND :area_max
|
||
AND is_active = true
|
||
AND scraped_at > NOW() - (:fresh_days || ' days')::interval
|
||
AND price_rub > 0
|
||
ORDER BY (
|
||
distance_m / 1000.0
|
||
+ CASE WHEN :target_year IS NOT NULL AND year_built IS NOT NULL
|
||
THEN abs(year_built - :target_year) / 12.0 ELSE 0 END
|
||
+ CASE WHEN :target_house_type IS NOT NULL AND house_type IS NOT NULL
|
||
AND house_type <> :target_house_type
|
||
THEN 1.5 ELSE 0 END
|
||
)
|
||
LIMIT 50
|
||
"""
|
||
),
|
||
{
|
||
"lat": lat,
|
||
"lon": lon,
|
||
"radius": radius_m,
|
||
"rooms": rooms,
|
||
"area_min": area * (1 - area_tolerance),
|
||
"area_max": area * (1 + area_tolerance),
|
||
"fresh_days": LISTINGS_FRESH_DAYS,
|
||
"target_year": year_built,
|
||
"target_house_type": house_type,
|
||
},
|
||
).mappings().all()
|
||
|
||
return [dict(r) for r in rows], radius_m > DEFAULT_RADIUS_M
|
||
|
||
|
||
def _fetch_deals(
|
||
db: Session, *, lat: float, lon: float, rooms: int, area: float, radius_m: int
|
||
) -> list[dict[str, Any]]:
|
||
rows = db.execute(
|
||
text(
|
||
"""
|
||
SELECT
|
||
source, address, lat, lon,
|
||
rooms, area_m2, floor, total_floors,
|
||
price_rub, price_per_m2,
|
||
deal_date, days_on_market,
|
||
ST_Distance(geom::geography, ST_MakePoint(:lon, :lat)::geography) AS distance_m
|
||
FROM deals
|
||
WHERE ST_DWithin(geom::geography, ST_MakePoint(:lon, :lat)::geography, :radius)
|
||
AND rooms = :rooms
|
||
AND area_m2 BETWEEN :area_min AND :area_max
|
||
AND deal_date > NOW() - (:months || ' months')::interval
|
||
AND price_rub > 0
|
||
ORDER BY deal_date DESC
|
||
LIMIT 30
|
||
"""
|
||
),
|
||
{
|
||
"lat": lat,
|
||
"lon": lon,
|
||
"radius": radius_m,
|
||
"rooms": rooms,
|
||
"area_min": area * (1 - AREA_TOLERANCE),
|
||
"area_max": area * (1 + AREA_TOLERANCE),
|
||
"months": DEALS_PERIOD_MONTHS,
|
||
},
|
||
).mappings().all()
|
||
|
||
return [dict(r) for r in rows]
|
||
|
||
|
||
def _filter_outliers(lots: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
"""Tukey IQR rule: исключаем точки вне [Q1 - 1.5×IQR, Q3 + 1.5×IQR]."""
|
||
if len(lots) < 5:
|
||
return lots # на маленькой выборке нечего фильтровать
|
||
|
||
prices = sorted(lot["price_per_m2"] for lot in lots if lot.get("price_per_m2"))
|
||
if len(prices) < 4:
|
||
return lots
|
||
|
||
q1 = _percentile(prices, 0.25)
|
||
q3 = _percentile(prices, 0.75)
|
||
iqr = q3 - q1
|
||
low = q1 - 1.5 * iqr
|
||
high = q3 + 1.5 * iqr
|
||
|
||
clean = [lot for lot in lots if low <= lot.get("price_per_m2", 0) <= high]
|
||
if len(clean) < len(lots):
|
||
logger.info("outlier filter: %d → %d (Q1=%d Q3=%d)", len(lots), len(clean), q1, q3)
|
||
return clean
|
||
|
||
|
||
def _percentile(sorted_values: list[float], p: float) -> float:
|
||
"""Linear interpolation percentile (не округляем — оставляем float)."""
|
||
if not sorted_values:
|
||
return 0.0
|
||
if len(sorted_values) == 1:
|
||
return float(sorted_values[0])
|
||
n = len(sorted_values)
|
||
rank = p * (n - 1)
|
||
lo = int(rank)
|
||
hi = min(lo + 1, n - 1)
|
||
frac = rank - lo
|
||
return sorted_values[lo] + (sorted_values[hi] - sorted_values[lo]) * frac
|
||
|
||
|
||
def _compute_confidence(
|
||
n_analogs: int,
|
||
median_ppm2: float,
|
||
q1: float,
|
||
q3: float,
|
||
fallback_radius_used: bool,
|
||
area_widened: bool = False,
|
||
) -> tuple[str, str]:
|
||
"""Confidence + explanation string.
|
||
|
||
high — n≥10 AND IQR/median < 0.15
|
||
medium — n≥5 OR IQR/median < 0.25
|
||
low — иначе
|
||
"""
|
||
if median_ppm2 == 0:
|
||
return "low", "Не найдено аналогов — попробуйте уточнить адрес или расширить параметры."
|
||
|
||
iqr = q3 - q1
|
||
iqr_pct = iqr / median_ppm2 if median_ppm2 > 0 else 1.0
|
||
notes = []
|
||
if fallback_radius_used:
|
||
notes.append("расширили радиус до 2 км")
|
||
if area_widened:
|
||
notes.append("расширили допуск по площади до ±25%")
|
||
fallback_note = f" ({', '.join(notes)} из-за нехватки данных)" if notes else ""
|
||
|
||
if n_analogs >= 10 and iqr_pct < 0.15:
|
||
return (
|
||
"high",
|
||
f"Найдено {n_analogs} аналогов, разброс цены ±{int(iqr_pct * 100 / 2)}% от медианы{fallback_note}.",
|
||
)
|
||
# medium только если есть достаточно точек ИЛИ узкий разброс при ≥3 точках
|
||
if n_analogs >= 5 or (n_analogs >= 3 and iqr_pct < 0.25):
|
||
return (
|
||
"medium",
|
||
f"Найдено {n_analogs} аналогов, разброс цены ±{int(iqr_pct * 100 / 2)}% от медианы{fallback_note}.",
|
||
)
|
||
return (
|
||
"low",
|
||
f"Только {n_analogs} аналог{'а' if 2 <= n_analogs <= 4 else 'ов' if n_analogs != 1 else ''}, "
|
||
f"разброс ±{int(iqr_pct * 100 / 2)}% — рекомендуется ручная проверка{fallback_note}.",
|
||
)
|
||
|
||
|
||
def _listing_to_analog(row: dict[str, Any]) -> AnalogLot:
|
||
return AnalogLot(
|
||
address=row.get("address") or "",
|
||
area_m2=float(row.get("area_m2") or 0),
|
||
rooms=int(row.get("rooms") or 0),
|
||
floor=row.get("floor"),
|
||
total_floors=row.get("total_floors"),
|
||
price_rub=int(row["price_rub"]),
|
||
price_per_m2=int(row.get("price_per_m2") or 0),
|
||
listing_date=row.get("listing_date"),
|
||
days_on_market=row.get("days_on_market"),
|
||
photo_url=(row.get("photo_urls") or [None])[0] if isinstance(row.get("photo_urls"), list) else None,
|
||
source=row.get("source"),
|
||
source_url=row.get("source_url"),
|
||
distance_m=int(row["distance_m"]) if row.get("distance_m") is not None else None,
|
||
)
|
||
|
||
|
||
def _deal_to_analog(row: dict[str, Any]) -> AnalogLot:
|
||
"""deals не имеют photo_url — упрощённо."""
|
||
return AnalogLot(
|
||
address=row.get("address") or "",
|
||
area_m2=float(row.get("area_m2") or 0),
|
||
rooms=int(row.get("rooms") or 0),
|
||
floor=row.get("floor"),
|
||
total_floors=row.get("total_floors"),
|
||
price_rub=int(row["price_rub"]),
|
||
price_per_m2=int(row.get("price_per_m2") or 0),
|
||
listing_date=row.get("deal_date"),
|
||
days_on_market=row.get("days_on_market"),
|
||
photo_url=None,
|
||
source=row.get("source"),
|
||
source_url=None, # rosreestr сделки без публичной ссылки
|
||
distance_m=int(row["distance_m"]) if row.get("distance_m") is not None else None,
|
||
)
|
||
|
||
|
||
def _empty_estimate(payload: TradeInEstimateInput, *, reason: str) -> AggregatedEstimate:
|
||
"""Fallback когда нет данных для оценки."""
|
||
now = datetime.now(tz=UTC)
|
||
return AggregatedEstimate(
|
||
estimate_id=uuid4(),
|
||
median_price_rub=0,
|
||
range_low_rub=0,
|
||
range_high_rub=0,
|
||
median_price_per_m2=0,
|
||
confidence="low",
|
||
n_analogs=0,
|
||
period_months=DEALS_PERIOD_MONTHS,
|
||
analogs=[],
|
||
actual_deals=[],
|
||
expires_at=now + timedelta(hours=24),
|
||
)
|