"""Trade-In Estimator — mock endpoint (TI-1). MOCK implementation: returns realistic ЕКБ price bands by rooms/floor/repair. TODO TI-1b: заменить _mock_estimate() на реальный SQL aggregation из objective_lots + rosreestr_deals после OBJ-1/2 merge. """ from __future__ import annotations import json import logging import random from datetime import UTC, datetime, timedelta from typing import Annotated from uuid import UUID, uuid4 from fastapi import APIRouter, Depends from sqlalchemy import text from sqlalchemy.orm import Session from app.core.db import get_db from app.schemas.trade_in import AggregatedEstimate, AnalogLot, TradeInEstimateInput logger = logging.getLogger(__name__) router = APIRouter() # ЕКБ-адреса для фейковых аналогов (реальные улицы центра) _EKB_STREETS = [ "ул. Малышева", "ул. Куйбышева", "ул. 8 Марта", "ул. Белинского", "пр. Ленина", "ул. Толмачёва", "ул. Радищева", "ул. Мамина-Сибиряка", "ул. Луначарского", "ул. Первомайская", ] # Базовые ценовые диапазоны по комнатности (ЕКБ, 2026) _PRICE_BANDS: dict[int, dict[str, int | float | str]] = { 0: { # студия ~25 м² "median": 6_500_000, "low": 5_800_000, "high": 7_500_000, "ppm2": 260_000, "ref_area": 25.0, }, 1: { # 1к ~40 м² "median": 9_000_000, "low": 8_000_000, "high": 10_500_000, "ppm2": 225_000, "ref_area": 40.0, }, 2: { # 2к ~60 м² "median": 12_500_000, "low": 11_000_000, "high": 14_000_000, "ppm2": 208_000, "ref_area": 60.0, }, 3: { # 3к ~80 м² "median": 17_000_000, "low": 15_000_000, "high": 19_000_000, "ppm2": 213_000, "ref_area": 80.0, }, } def _floor_factor(floor: int, total_floors: int) -> float: """±5% поправка за этаж: 1й и последний этаж снижают цену.""" if floor == 1: return 0.95 if floor == total_floors: return 0.97 return 1.0 def _repair_factor(repair_state: str | None) -> float: """±10% поправка за состояние отделки.""" factors = { "needs_repair": 0.90, "standard": 1.00, "good": 1.05, "excellent": 1.10, } return factors.get(repair_state or "standard", 1.0) def _confidence(rooms: int) -> str: if 1 <= rooms <= 3: return "high" return "medium" def _gen_analogs( rooms: int, area_m2: float, base_ppm2: int, n: int, *, is_listing: bool, ) -> list[AnalogLot]: """Генерирует список фейковых аналогов (объявления или сделки).""" rng = random.Random(42 + rooms + n) result: list[AnalogLot] = [] today = datetime.now(tz=UTC).date() for i in range(n): street = _EKB_STREETS[i % len(_EKB_STREETS)] building_no = rng.randint(1, 120) apt_no = rng.randint(1, 300) addr = f"г. Екатеринбург, {street}, {building_no}, кв. {apt_no}" area_jitter = area_m2 * rng.uniform(0.85, 1.15) ppm2_jitter = int(base_ppm2 * rng.uniform(0.90, 1.10)) price = int(area_jitter * ppm2_jitter) floor_val = rng.randint(2, 16) total_fl = rng.randint(floor_val, 20) if is_listing: dom = rng.randint(5, 120) listing_dt = today - timedelta(days=dom) else: dom = rng.randint(10, 60) listing_dt = today - timedelta(days=rng.randint(30, 365)) result.append( AnalogLot( address=addr, area_m2=round(area_jitter, 1), rooms=rooms if rooms > 0 else 0, floor=floor_val, total_floors=total_fl, price_rub=price, price_per_m2=ppm2_jitter, listing_date=listing_dt, days_on_market=dom, photo_url=None, ) ) return result def _mock_estimate(payload: TradeInEstimateInput) -> AggregatedEstimate: """Возвращает mock-оценку на основе диапазонов ЕКБ 2026. Логика: - Берём базовый band по rooms (0-3+). - Масштабируем на фактическую площадь относительно референсной. - Применяем поправку за этаж (±5%) и отделку (±10%). - Генерируем 7-10 аналогов (листинги) и 3-5 actual_deals. """ rooms_key = min(payload.rooms, 3) # 4к+ → диапазон 3к band = _PRICE_BANDS[rooms_key] # Масштаб по площади ref_area: float = band["ref_area"] # type: ignore[assignment] area_scale = payload.area_m2 / ref_area ff = _floor_factor(payload.floor, payload.total_floors) rf = _repair_factor(payload.repair_state) combined = area_scale * ff * rf median = int(band["median"] * combined) # type: ignore[operator] low = int(band["low"] * combined) # type: ignore[operator] high = int(band["high"] * combined) # type: ignore[operator] ppm2 = int(band["ppm2"] * ff * rf) # type: ignore[operator] n_analogs = random.randint(7, 10) n_deals = random.randint(3, 5) analogs = _gen_analogs(rooms_key, payload.area_m2, ppm2, n_analogs, is_listing=True) actual_deals = _gen_analogs( rooms_key, payload.area_m2, int(ppm2 * 0.93), n_deals, is_listing=False ) now = datetime.now(tz=UTC) return AggregatedEstimate( estimate_id=uuid4(), median_price_rub=median, range_low_rub=low, range_high_rub=high, median_price_per_m2=ppm2, confidence=_confidence(payload.rooms), n_analogs=n_analogs + n_deals, period_months=24, analogs=analogs, actual_deals=actual_deals, expires_at=now + timedelta(hours=24), ) @router.post("/estimate", response_model=AggregatedEstimate) def estimate( payload: TradeInEstimateInput, db: Annotated[Session, Depends(get_db)], ) -> AggregatedEstimate: """MOCK реализация оценки квартиры для Trade-In. TODO TI-1b: заменить на реальный SQL aggregation из objective_lots после OBJ-1/2 merge (issue #314). """ result = _mock_estimate(payload) 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, ) db.execute( text( """ INSERT INTO trade_in_estimates ( id, address, 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, n_analogs, analogs, actual_deals, expires_at ) VALUES ( CAST(:id AS uuid), :address, :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, :n_analogs, CAST(:analogs AS jsonb), CAST(:actual_deals AS jsonb), :expires_at ) """ ), { "id": str(result.estimate_id), "address": payload.address, "area_m2": 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": result.median_price_rub, "range_low": result.range_low_rub, "range_high": result.range_high_rub, "median_price_per_m2": result.median_price_per_m2, "confidence": result.confidence, "n_analogs": result.n_analogs, "analogs": analogs_json, "actual_deals": deals_json, "expires_at": result.expires_at, }, ) db.commit() logger.info( "trade_in estimate saved id=%s rooms=%d area=%.1f confidence=%s", result.estimate_id, payload.rooms, payload.area_m2, result.confidence, ) return result @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 истёк. """ from fastapi import HTTPException row = db.execute( text( """ SELECT id, median_price, range_low, range_high, median_price_per_m2, confidence, n_analogs, analogs, actual_deals, expires_at, address, area_m2, rooms 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") analogs = [AnalogLot(**a) for a in (row.analogs or [])] actual_deals = [AnalogLot(**a) for a in (row.actual_deals or [])] 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, n_analogs=row.n_analogs, period_months=24, analogs=analogs, actual_deals=actual_deals, expires_at=row.expires_at, )