fdfb74ca88
После C-5 fix (PR #487) Avito coords либо NULL либо реальные. Estimator исключал source='avito' из radius search потому что раньше там был jitter (±0.005°) от 5 anchor cron'ов. Сейчас: - 3580 Avito + 641 Yandex + 106 N1 + 38 Cian listings без coords - estimator._fetch_analogs ловил только cian/yandex/n1 = ~30% эффективной базы Bundled fix: 1. app/tasks/geocode_missing.py — batch geocoder (Nominatim 1/s, dedup по address) 2. POST /admin/scrape/geocode-missing-listings + GET status — manual trigger 3. estimator.py: убран AND source <> 'avito' — после backfill Avito включён в radius Result: после backfill +4166 Avito listings в radius search = +40% эффективной data. Confidence "high" будет в ~70% случаев вместо ~30%. Tests: 11 новых tests/tasks/test_geocode_missing.py — all pass.
1072 lines
43 KiB
Python
1072 lines
43 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, 1000m) — радиус поиска
|
||
- source ≠ avito (у Avito фейковые anchor-jitter координаты — не гео-аналог)
|
||
- 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 hashlib
|
||
import json
|
||
import logging
|
||
from datetime import UTC, datetime, timedelta
|
||
from typing import Any
|
||
from uuid import 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
|
||
from app.services.house_metadata import get_house_metadata
|
||
from app.services.scrapers.avito_imv import (
|
||
IMVAddressNotFoundError,
|
||
IMVEvaluation,
|
||
compute_imv_cache_key,
|
||
evaluate_via_imv,
|
||
save_imv_evaluation,
|
||
)
|
||
from app.services.scrapers.cian_valuation import (
|
||
CianValuationResult,
|
||
estimate_via_cian_valuation,
|
||
)
|
||
from app.services.scrapers.yandex_valuation import (
|
||
YandexValuationResult,
|
||
YandexValuationScraper,
|
||
)
|
||
|
||
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 # сделки за последний год
|
||
|
||
# Поправочные коэффициенты на состояние ремонта. Аналоги в выборке — микс
|
||
# состояний (≈ "стандартный/косметический"), коэффициент сдвигает медиану под
|
||
# конкретный ремонт целевой квартиры. Встреча Птицы: ремонт влияет на цену.
|
||
_IMV_HOUSE_TYPE_MAP: dict[str | None, str | None] = {
|
||
"panel": "panel",
|
||
"brick": "brick",
|
||
"monolith": "monolith",
|
||
"monolith_brick": "monolith_brick",
|
||
"monolithic": "monolith",
|
||
"block": "block",
|
||
"wood": "wood",
|
||
None: None,
|
||
}
|
||
_IMV_REPAIR_MAP: dict[str | None, str | None] = {
|
||
"needs_repair": "required",
|
||
"standard": "cosmetic",
|
||
"good": "euro",
|
||
"excellent": "designer",
|
||
None: None,
|
||
}
|
||
|
||
_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)
|
||
|
||
|
||
# ── Avito IMV cache lookup (Stage 3) ────────────────────────────────────────
|
||
IMV_CACHE_TTL_HOURS = 24
|
||
|
||
YANDEX_VALUATION_CACHE_TTL_HOURS = 24
|
||
YANDEX_VALUATION_DEFAULT_CATEGORY = "APARTMENT"
|
||
YANDEX_VALUATION_DEFAULT_TYPE = "SELL"
|
||
|
||
|
||
async def _get_or_fetch_imv_cached(
|
||
db: Session,
|
||
*,
|
||
address: str,
|
||
rooms: int,
|
||
area_m2: float,
|
||
floor: int,
|
||
floor_at_home: int,
|
||
house_type: str,
|
||
renovation_type: str,
|
||
has_balcony: bool,
|
||
has_loggia: bool,
|
||
estimate_id_for_link: Any = None,
|
||
) -> IMVEvaluation | None:
|
||
"""Cached IMV lookup. TTL 24h по cache_key (sha256 of address + params).
|
||
|
||
1. compute cache_key
|
||
2. SELECT из avito_imv_evaluations WHERE cache_key = :ck AND fetched_at > NOW() - 24h
|
||
3. Если hit → возвращаем reconstructed IMVEvaluation
|
||
4. Cache miss → call evaluate_via_imv, save_imv_evaluation, return
|
||
|
||
Graceful: на любой error возвращаем None (estimator продолжает без IMV).
|
||
"""
|
||
try:
|
||
cache_key = compute_imv_cache_key(
|
||
address, rooms, area_m2, floor, floor_at_home,
|
||
house_type, renovation_type, has_balcony, has_loggia,
|
||
)
|
||
|
||
existing = db.execute(
|
||
text(
|
||
"""
|
||
SELECT id, cache_key, address, rooms, area_m2, floor, floor_at_home,
|
||
house_type, renovation_type, has_balcony, has_loggia,
|
||
lat, lon, geo_hash, avito_address_id, avito_location_id,
|
||
avito_metro_id, avito_district_id,
|
||
recommended_price, lower_price, higher_price, market_count,
|
||
raw_response, fetched_at
|
||
FROM avito_imv_evaluations
|
||
WHERE cache_key = :ck
|
||
AND fetched_at > NOW() - (:ttl_hours || ' hours')::interval
|
||
ORDER BY fetched_at DESC
|
||
LIMIT 1
|
||
"""
|
||
),
|
||
{"ck": cache_key, "ttl_hours": IMV_CACHE_TTL_HOURS},
|
||
).mappings().first()
|
||
|
||
if existing is not None:
|
||
logger.info(
|
||
"imv: cache HIT key=%s recommended=%d",
|
||
cache_key[:8], existing["recommended_price"],
|
||
)
|
||
from app.services.scrapers.avito_imv import IMVGeo
|
||
return IMVEvaluation(
|
||
cache_key=existing["cache_key"],
|
||
address=existing["address"],
|
||
rooms=existing["rooms"],
|
||
area_m2=float(existing["area_m2"]),
|
||
floor=existing["floor"],
|
||
floor_at_home=existing["floor_at_home"],
|
||
house_type=existing["house_type"],
|
||
renovation_type=existing["renovation_type"],
|
||
has_balcony=existing["has_balcony"],
|
||
has_loggia=existing["has_loggia"],
|
||
geo=IMVGeo(
|
||
geo_hash=existing["geo_hash"] or "",
|
||
lat=existing["lat"],
|
||
lon=existing["lon"],
|
||
avito_address_id=existing["avito_address_id"],
|
||
avito_location_id=existing["avito_location_id"],
|
||
avito_metro_id=existing["avito_metro_id"],
|
||
avito_district_id=existing["avito_district_id"],
|
||
),
|
||
recommended_price=existing["recommended_price"],
|
||
lower_price=existing["lower_price"],
|
||
higher_price=existing["higher_price"],
|
||
market_count=existing["market_count"],
|
||
raw_response=existing.get("raw_response"),
|
||
)
|
||
|
||
# Cache miss — fresh fetch
|
||
logger.info("imv: cache MISS key=%s — fetching fresh", cache_key[:8])
|
||
result = await evaluate_via_imv(
|
||
address=address, rooms=rooms, area_m2=area_m2,
|
||
floor=floor, floor_at_home=floor_at_home,
|
||
house_type=house_type, renovation_type=renovation_type,
|
||
has_balcony=has_balcony, has_loggia=has_loggia,
|
||
)
|
||
save_imv_evaluation(db, result, estimate_id=estimate_id_for_link)
|
||
logger.info(
|
||
"imv: fresh recommended=%d range=(%d, %d) count=%d",
|
||
result.recommended_price, result.lower_price, result.higher_price,
|
||
result.market_count or 0,
|
||
)
|
||
return result
|
||
|
||
except IMVAddressNotFoundError as e:
|
||
logger.warning("imv: address not found in Avito geocoder: %s", e)
|
||
return None
|
||
except Exception as e:
|
||
logger.warning("imv: fetch failed — estimator продолжает без IMV: %s", e)
|
||
return None
|
||
|
||
|
||
# ── Yandex Valuation cache lookup (Stage 8) ─────────────────────────────────
|
||
|
||
def _yandex_valuation_cache_key(
|
||
address: str, offer_category: str, offer_type: str
|
||
) -> str:
|
||
"""SHA256 cache key for Yandex Valuation lookups."""
|
||
payload = f"{address}|{offer_category}|{offer_type}"
|
||
return hashlib.sha256(payload.encode("utf-8")).hexdigest()
|
||
|
||
|
||
async def _get_or_fetch_yandex_valuation_cached(
|
||
db: Session,
|
||
*,
|
||
address: str,
|
||
offer_category: str = YANDEX_VALUATION_DEFAULT_CATEGORY,
|
||
offer_type: str = YANDEX_VALUATION_DEFAULT_TYPE,
|
||
) -> YandexValuationResult | None:
|
||
"""Cached Yandex Valuation lookup. TTL 24h via external_valuations table.
|
||
|
||
Returns None on any error / cache miss + fetch failure — caller continues
|
||
without Yandex enrichment (graceful degradation).
|
||
"""
|
||
cache_key = _yandex_valuation_cache_key(address, offer_category, offer_type)
|
||
|
||
# Cache lookup
|
||
try:
|
||
cached = db.execute(
|
||
text(
|
||
"""
|
||
SELECT raw_payload, fetched_at
|
||
FROM external_valuations
|
||
WHERE source = 'yandex_valuation'
|
||
AND cache_key = :ck
|
||
AND expires_at > NOW()
|
||
ORDER BY fetched_at DESC
|
||
LIMIT 1
|
||
"""
|
||
),
|
||
{"ck": cache_key},
|
||
).mappings().first()
|
||
except Exception as e:
|
||
logger.warning("yandex_valuation: cache lookup failed: %s", e)
|
||
cached = None
|
||
|
||
if cached is not None and cached.get("raw_payload"):
|
||
try:
|
||
payload_dict = (
|
||
cached["raw_payload"]
|
||
if isinstance(cached["raw_payload"], dict)
|
||
else json.loads(cached["raw_payload"])
|
||
)
|
||
logger.info(
|
||
"yandex_valuation: cache HIT key=%s items=%d",
|
||
cache_key[:8],
|
||
len(payload_dict.get("history_items", [])),
|
||
)
|
||
return YandexValuationResult.model_validate(payload_dict)
|
||
except Exception as e:
|
||
logger.warning("yandex_valuation: cache deserialize failed — refetching: %s", e)
|
||
|
||
# Fresh fetch
|
||
try:
|
||
async with YandexValuationScraper() as scraper:
|
||
result = await scraper.fetch_house_history(
|
||
address=address,
|
||
offer_category=offer_category,
|
||
offer_type=offer_type,
|
||
)
|
||
except Exception as e:
|
||
logger.warning(
|
||
"yandex_valuation: fetch failed — estimator продолжает без Yandex: %s", e
|
||
)
|
||
return None
|
||
|
||
if result is None:
|
||
logger.info("yandex_valuation: empty result for address=%s", address[:60])
|
||
return None
|
||
|
||
# Save to cache (UPSERT on (source, cache_key))
|
||
try:
|
||
db.execute(
|
||
text(
|
||
"""
|
||
INSERT INTO external_valuations (
|
||
source, cache_key, address,
|
||
raw_payload,
|
||
fetched_at, expires_at
|
||
) VALUES (
|
||
'yandex_valuation', :ck, :addr,
|
||
CAST(:payload AS jsonb),
|
||
NOW(), NOW() + (:ttl_hours || ' hours')::interval
|
||
)
|
||
ON CONFLICT (source, cache_key) DO UPDATE
|
||
SET raw_payload = EXCLUDED.raw_payload,
|
||
fetched_at = NOW(),
|
||
expires_at = NOW() + (:ttl_hours || ' hours')::interval
|
||
"""
|
||
),
|
||
{
|
||
"ck": cache_key,
|
||
"addr": address,
|
||
"payload": json.dumps(result.model_dump(mode="json"), ensure_ascii=False),
|
||
"ttl_hours": YANDEX_VALUATION_CACHE_TTL_HOURS,
|
||
},
|
||
)
|
||
db.commit()
|
||
logger.info(
|
||
"yandex_valuation: fresh fetch saved key=%s items=%d",
|
||
cache_key[:8],
|
||
len(result.history_items),
|
||
)
|
||
except Exception as e:
|
||
logger.warning("yandex_valuation: cache save failed (continuing): %s", e)
|
||
db.rollback()
|
||
|
||
return result
|
||
|
||
|
||
def _save_yandex_history_items(
|
||
db: Session,
|
||
result: YandexValuationResult,
|
||
) -> int:
|
||
"""Persist history items to house_placement_history. Returns saved count.
|
||
|
||
house_id stays NULL — estimator doesn't compute target_house_id yet.
|
||
Idempotent via UNIQUE (source, ext_item_id); we synthesize ext_item_id from
|
||
(address|date|area|floor) hash since Yandex history items don't carry an
|
||
explicit ID.
|
||
"""
|
||
saved = 0
|
||
for item in result.history_items:
|
||
# Synthesize stable ext_item_id (no native ID in valuation page)
|
||
ext_seed = (
|
||
f"{result.address}|{item.publish_date}|{item.area_m2}|{item.floor}|"
|
||
f"{item.start_price}|{item.last_price}"
|
||
)
|
||
ext_item_id = hashlib.sha256(ext_seed.encode("utf-8")).hexdigest()[:32]
|
||
try:
|
||
db.execute(
|
||
text(
|
||
"""
|
||
INSERT INTO house_placement_history (
|
||
source, ext_item_id,
|
||
rooms, area_m2, floor,
|
||
start_price, start_price_date,
|
||
last_price, last_price_date,
|
||
exposure_days,
|
||
raw_payload
|
||
) VALUES (
|
||
'yandex_valuation', :ext_id,
|
||
:rooms, :area, :floor,
|
||
:start_price, :publish_date,
|
||
:last_price, :publish_date,
|
||
:exposure,
|
||
CAST(:raw AS jsonb)
|
||
)
|
||
ON CONFLICT (source, ext_item_id) DO NOTHING
|
||
"""
|
||
),
|
||
{
|
||
"ext_id": ext_item_id,
|
||
"rooms": item.rooms,
|
||
"area": item.area_m2,
|
||
"floor": item.floor,
|
||
"start_price": item.start_price,
|
||
"last_price": item.last_price,
|
||
"publish_date": item.publish_date,
|
||
"exposure": item.exposure_days,
|
||
"raw": json.dumps(item.model_dump(mode="json"), ensure_ascii=False),
|
||
},
|
||
)
|
||
saved += 1
|
||
except Exception as e:
|
||
logger.warning("yandex_valuation: failed to save history item: %s", e)
|
||
db.rollback()
|
||
continue
|
||
if saved:
|
||
db.commit()
|
||
return saved
|
||
|
||
|
||
# ── 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, db, reason="address_not_geocoded")
|
||
|
||
# 2. #392: обогащаем год / тип дома из картографии (OSM Overpass), если
|
||
# пользователь их не указал — это улучшает house-match аналогов (#6).
|
||
# Best-effort: при недоступности OSM target_* остаются None.
|
||
target_year = payload.year_built
|
||
target_house_type = payload.house_type
|
||
if target_year is None or target_house_type is None:
|
||
house_meta = await get_house_metadata(geo.lat, geo.lon, db)
|
||
if house_meta is not None:
|
||
if target_year is None:
|
||
target_year = house_meta.year_built
|
||
if target_house_type is None:
|
||
target_house_type = house_meta.house_type
|
||
|
||
# 3. Three-tier fallback:
|
||
# a) 1km + ±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=target_year, house_type=target_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=target_year, house_type=target_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=target_year, house_type=target_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
|
||
|
||
# ── Stage 3: Avito IMV evaluation as 5-th source (on-demand cached) ──
|
||
imv_eval: IMVEvaluation | None = None
|
||
imv_house_type = _IMV_HOUSE_TYPE_MAP.get(target_house_type)
|
||
imv_renovation = _IMV_REPAIR_MAP.get(payload.repair_state)
|
||
# IMV требует: address, rooms, area, floor, floor_at_home, house_type, renovation_type.
|
||
# Если payload не содержит required fields — skip IMV (graceful).
|
||
if (
|
||
geo is not None
|
||
and geo.full_address
|
||
and payload.rooms is not None
|
||
and payload.area_m2
|
||
and payload.floor is not None
|
||
and payload.total_floors is not None
|
||
and imv_house_type is not None
|
||
and imv_renovation is not None
|
||
):
|
||
imv_eval = await _get_or_fetch_imv_cached(
|
||
db,
|
||
address=geo.full_address,
|
||
rooms=payload.rooms,
|
||
area_m2=payload.area_m2,
|
||
floor=payload.floor,
|
||
floor_at_home=payload.total_floors,
|
||
house_type=imv_house_type,
|
||
renovation_type=imv_renovation,
|
||
has_balcony=bool(payload.has_balcony),
|
||
has_loggia=False, # payload не разделяет балкон/лоджия → дефолт False
|
||
)
|
||
|
||
# Include IMV в sources_used если получили
|
||
sources_used_pre = sorted({lot.get("source") for lot in listings_clean if lot.get("source")})
|
||
if imv_eval is not None:
|
||
sources_used_pre = sorted(set(sources_used_pre) | {"avito_imv"})
|
||
|
||
# ── Stage 8: Yandex Valuation as on-demand source (anonymous, cached 24h) ──
|
||
yandex_val: YandexValuationResult | None = None
|
||
if geo is not None and geo.full_address:
|
||
yandex_val = await _get_or_fetch_yandex_valuation_cached(
|
||
db, address=geo.full_address,
|
||
)
|
||
if yandex_val is not None:
|
||
sources_used_pre = sorted(set(sources_used_pre) | {"yandex_valuation"})
|
||
saved_hist = _save_yandex_history_items(db, yandex_val)
|
||
logger.info(
|
||
"yandex_valuation: history items processed=%d saved=%d"
|
||
" (house_id=NULL — matching deferred)",
|
||
len(yandex_val.history_items), saved_hist,
|
||
)
|
||
|
||
# ── Stage 9: Cian Valuation as 7th source (on-demand, 24h cached, graceful if no cookies) ──
|
||
cian_val: CianValuationResult | None = None
|
||
if (
|
||
geo is not None
|
||
and geo.full_address
|
||
and payload.rooms is not None
|
||
and payload.area_m2
|
||
and payload.floor is not None
|
||
and payload.total_floors is not None
|
||
):
|
||
try:
|
||
cian_val = await estimate_via_cian_valuation(
|
||
db,
|
||
address=geo.full_address,
|
||
total_area=payload.area_m2,
|
||
rooms_count=payload.rooms,
|
||
floor=payload.floor,
|
||
total_floors=payload.total_floors,
|
||
repair_type="cosmetic",
|
||
deal_type="sale",
|
||
use_cache=True,
|
||
)
|
||
if cian_val is not None and cian_val.sale_price_rub:
|
||
sources_used_pre = sorted(set(sources_used_pre) | {"cian_valuation"})
|
||
logger.info(
|
||
"cian_valuation: price=%s accuracy=%s house_id=%s",
|
||
cian_val.sale_price_rub,
|
||
cian_val.sale_accuracy,
|
||
cian_val.external_house_id,
|
||
)
|
||
except Exception as exc:
|
||
logger.warning("cian_valuation: lookup failed (graceful): %s", exc)
|
||
|
||
# 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]]
|
||
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,
|
||
ownership_type, has_mortgage, client_name, client_phone,
|
||
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,
|
||
:ownership_type, :has_mortgage, :client_name, :client_phone,
|
||
: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": target_year,
|
||
"house_type": target_house_type,
|
||
"repair_state": payload.repair_state,
|
||
"has_balcony": payload.has_balcony,
|
||
"ownership_type": payload.ownership_type,
|
||
"has_mortgage": payload.has_mortgage,
|
||
"client_name": payload.client_name,
|
||
"client_phone": payload.client_phone,
|
||
"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()
|
||
|
||
# Link saved IMV evaluation к этому estimate_id (для analytics joining)
|
||
if imv_eval is not None:
|
||
try:
|
||
db.execute(
|
||
text(
|
||
"""
|
||
UPDATE avito_imv_evaluations
|
||
SET estimate_id = CAST(:estimate_id AS uuid)
|
||
WHERE cache_key = :cache_key
|
||
AND (estimate_id IS NULL OR estimate_id = CAST(:estimate_id AS uuid))
|
||
"""
|
||
),
|
||
{"estimate_id": str(estimate_id), "cache_key": imv_eval.cache_key},
|
||
)
|
||
db.commit()
|
||
except Exception as e:
|
||
logger.warning("imv: failed to link estimate_id to evaluation: %s", e)
|
||
|
||
logger.info(
|
||
"estimate: id=%s addr=%s rooms=%d area=%.1f → median=%d (n=%d, conf=%s)%s%s",
|
||
estimate_id,
|
||
geo.full_address[:60],
|
||
payload.rooms,
|
||
payload.area_m2,
|
||
median_price,
|
||
n_analogs,
|
||
confidence,
|
||
f" imv={imv_eval.recommended_price}" if imv_eval else "",
|
||
f" cian={cian_val.sale_price_rub}" if cian_val and cian_val.sale_price_rub else "",
|
||
)
|
||
|
||
sources_used = sorted({lot.source for lot in analogs_lots if lot.source})
|
||
if imv_eval is not None:
|
||
sources_used = sorted(set(sources_used) | {"avito_imv"})
|
||
if yandex_val is not None:
|
||
sources_used = sorted(set(sources_used) | {"yandex_valuation"})
|
||
if cian_val is not None and cian_val.sale_price_rub:
|
||
sources_used = sorted(set(sources_used) | {"cian_valuation"})
|
||
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,
|
||
est_days_on_market=_estimate_days_on_market(listings_clean, deals),
|
||
area_m2=payload.area_m2,
|
||
rooms=payload.rooms,
|
||
floor=payload.floor,
|
||
total_floors=payload.total_floors,
|
||
year_built=target_year,
|
||
house_type=target_house_type,
|
||
repair_state=payload.repair_state,
|
||
has_balcony=payload.has_balcony,
|
||
)
|
||
|
||
|
||
def _estimate_days_on_market(
|
||
listings: list[dict[str, Any]], deals: list[dict[str, Any]]
|
||
) -> int | None:
|
||
"""Прогноз срока продажи — медиана days_on_market по аналогам/сделкам.
|
||
|
||
Возвращает None если ни у одного аналога нет данных о сроке экспозиции
|
||
(наши парсеры не всегда его отдают — честно показываем «нет данных»).
|
||
"""
|
||
values = [
|
||
int(lot["days_on_market"])
|
||
for lot in (*listings, *deals)
|
||
if lot.get("days_on_market") and int(lot["days_on_market"]) > 0
|
||
]
|
||
if len(values) < 3:
|
||
return None
|
||
values.sort()
|
||
return values[len(values) // 2]
|
||
|
||
|
||
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
|
||
-- 2026-05-23: Avito coords теперь real (PR #487 убрал jitter после
|
||
-- C-5 audit). Listings с NULL coords отфильтруются через ST_DWithin
|
||
-- (geom IS NULL → не matches). geocode-missing-listings backfill
|
||
-- подтягивает координаты для address-only Avito листингов.
|
||
ORDER BY (
|
||
-- distance_m — это SELECT-алиас. В ORDER BY-ВЫРАЖЕНИИ (не голым
|
||
-- термом) PostgreSQL трактует имя как входную колонку listings,
|
||
-- которой нет → "column distance_m does not exist". Инлайним ST_Distance.
|
||
ST_Distance(geom::geography, ST_MakePoint(:lon, :lat)::geography) / 1000.0
|
||
-- CAST обязателен: target_year / target_house_type приходят NULL
|
||
-- без типа → PostgreSQL "could not determine data type of parameter"
|
||
-- (AmbiguousParameter). Явный тип снимает неоднозначность.
|
||
+ CASE WHEN CAST(:target_year AS integer) IS NOT NULL AND year_built IS NOT NULL
|
||
THEN abs(year_built - CAST(:target_year AS integer)) / 12.0 ELSE 0 END
|
||
+ CASE WHEN CAST(:target_house_type AS text) IS NOT NULL AND house_type IS NOT NULL
|
||
AND house_type <> CAST(:target_house_type AS text)
|
||
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, db: Session, *, reason: str
|
||
) -> AggregatedEstimate:
|
||
"""Fallback когда нет данных для оценки.
|
||
|
||
Сохраняет запись в БД (confidence='low', пустые analogs/deals), чтобы GET /estimate/{id}
|
||
не возвращал 404. C-4 security audit.
|
||
"""
|
||
estimate_id = uuid4()
|
||
now = datetime.now(tz=UTC)
|
||
expires_at = now + timedelta(hours=24)
|
||
|
||
db.execute(
|
||
text(
|
||
"""
|
||
INSERT INTO trade_in_estimates (
|
||
id, address,
|
||
area_m2, rooms, floor, total_floors,
|
||
year_built, house_type, repair_state, has_balcony,
|
||
ownership_type, has_mortgage, client_name, client_phone,
|
||
median_price, range_low, range_high, median_price_per_m2,
|
||
confidence, confidence_explanation, n_analogs,
|
||
analogs, actual_deals,
|
||
sources_used,
|
||
expires_at
|
||
) VALUES (
|
||
CAST(:id AS uuid), :address,
|
||
:area, :rooms, :floor, :total_floors,
|
||
:year_built, :house_type, :repair_state, :has_balcony,
|
||
:ownership_type, :has_mortgage, :client_name, :client_phone,
|
||
0, 0, 0, 0,
|
||
'low', :explanation, 0,
|
||
'[]'::jsonb, '[]'::jsonb,
|
||
'[]'::jsonb,
|
||
:expires_at
|
||
)
|
||
"""
|
||
),
|
||
{
|
||
"id": str(estimate_id),
|
||
"address": payload.address,
|
||
"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,
|
||
"ownership_type": payload.ownership_type,
|
||
"has_mortgage": payload.has_mortgage,
|
||
"client_name": payload.client_name,
|
||
"client_phone": payload.client_phone,
|
||
"explanation": reason,
|
||
"expires_at": expires_at,
|
||
},
|
||
)
|
||
db.commit()
|
||
logger.info(
|
||
"empty_estimate: id=%s reason=%s addr=%s", estimate_id, reason, payload.address[:60]
|
||
)
|
||
|
||
return AggregatedEstimate(
|
||
estimate_id=estimate_id,
|
||
median_price_rub=0,
|
||
range_low_rub=0,
|
||
range_high_rub=0,
|
||
median_price_per_m2=0,
|
||
confidence="low",
|
||
confidence_explanation=reason,
|
||
n_analogs=0,
|
||
period_months=DEALS_PERIOD_MONTHS,
|
||
analogs=[],
|
||
actual_deals=[],
|
||
expires_at=expires_at,
|
||
)
|