gendesign/tradein-mvp/backend/app/services/yandex_price_history.py

"""Yandex offer price-history capture из gate-ответа.

В отличие от Cian (отдельный detail-fetch priceChanges, см. cian_price_history.py),
Yandex gate-ответ уже несёт `price.previous` + `price.trend`, которые парсятся в
`ScrapedLot.price_previous_rub` / `price_trend`. Otsenka-скрапер НЕ нужен — gate достаточно.

Эта функция вызывается из run_yandex_city_sweep (scraper_kit.orchestration.pipeline,
через инжектируемый EnrichmentJobs.record_yandex_price_history) СРАЗУ ПОСЛЕ
save_listings (те же lots): резолвит
listing_id по (source='yandex', source_id), читает последнюю по change_time точку
offer_price_history и идемпотентно дописывает наблюдение, если цена изменилась.

Idempotent-логика (на каждый yandex-lot):
  - история ПУСТА:
      * price_previous задан И != price_rub → seed previous-точку (change_time = now-1d,
        timestamp синтетический — gate реального времени смены не несёт), затем current-точку;
        НО previous, отличающаяся от текущей цены на порядок и больше, не сеется (#3385);
      * иначе → только current-точку (now).
  - история ЕСТЬ и latest_price != price_rub → одна current-точка (now) — зафиксировали смену.
  - история ЕСТЬ и latest_price == price_rub → skip (цена не менялась).

Per-lot SAVEPOINT: один битый lot не валит весь батч. db.commit() в конце.
psycopg v3 discipline: только CAST(:x AS type), никаких :x::type. Все timestamp'ы
считаются в Python и биндятся как параметры (DB-agnostic, без Postgres-only interval).
"""

from __future__ import annotations

import logging
from datetime import UTC, datetime, timedelta

# #2306: ScrapedLot migrated to scraper_kit (byte-identical field/method surface to
# legacy base-module ScrapedLot, deleted #2397 финальный шаг E — kit единственный
# живой путь). record_yandex_price_history reads lot.* via duck-typing only — no
# isinstance checks — так что любой ScrapedLot-совместимый объект (source, source_id,
# price_rub, price_previous_rub, price_trend) остаётся взаимозаменяем для этой функции.
from scraper_kit.base import ScrapedLot
from sqlalchemy import text
from sqlalchemy.orm import Session

logger = logging.getLogger(__name__)

_SOURCE = "yandex"

# #3385: price.previous, отличающаяся от текущей цены в 9.5 раза и больше (в любую
# сторону), — не рынок, а потерянный разряд или цена в тысячах. Прод 17.09.2026, все
# 8313 посевов yandex: самый большой честный скачок ×2.13, дальше пусто до ×9.74; выше
# порога 11 посевов (×10 — 7, ×175…×1000 — 4), и у всех нижняя цена 0.1–37 тыс/м².
# Посев синтетический (now-1d, реального времени смены gate не несёт), без него
# теряется только фиктивная точка — текущая цена пишется всё равно.
_SEED_IMPLAUSIBLE_RATIO = 9.5

# Резолв listing_id по стабильному (source, source_id) — у Yandex source_id = offer_id.
_RESOLVE_SQL = text(
    """
    SELECT id
    FROM listings
    WHERE source = 'yandex'
      AND source_id = CAST(:source_id AS text)
    LIMIT 1
    """
)

# Последняя зафиксированная цена для листинга (по change_time DESC — индекс oph_listing_time_idx).
_LATEST_SQL = text(
    """
    SELECT price_rub
    FROM offer_price_history
    WHERE listing_id = CAST(:listing_id AS bigint)
      AND source = 'yandex'
    ORDER BY change_time DESC
    LIMIT 1
    """
)

# change_time bound как Python datetime — psycopg v3 адаптирует к timestamptz нативно,
# CAST не нужен (и был бы вреден: SQLite numeric-affinity CAST манглит ISO-строку).
_INSERT_SQL = text(
    """
    INSERT INTO offer_price_history (listing_id, change_time, price_rub, source)
    VALUES (
        CAST(:listing_id AS bigint),
        :change_time,
        CAST(:price_rub AS numeric),
        'yandex'
    )
    """
)


def _insert_point(
    db: Session,
    *,
    listing_id: int,
    price_rub: float,
    change_time: datetime,
) -> None:
    db.execute(
        _INSERT_SQL,
        {
            "listing_id": listing_id,
            "change_time": change_time,
            "price_rub": price_rub,
        },
    )


def _implausible_seed(previous: float, current: float) -> bool:
    """previous и current разнятся на порядок и больше (или одна из цен не > 0)."""
    low, high = sorted((float(previous), float(current)))
    return low <= 0 or high / low >= _SEED_IMPLAUSIBLE_RATIO


def record_yandex_price_history(db: Session, lots: list[ScrapedLot]) -> int:
    """Дописать offer_price_history по yandex-lots из gate price.previous/trend.

    Args:
        db: SQLAlchemy session (caller-owned; commit делается здесь в конце).
        lots: те же ScrapedLot, что были переданы в save_listings. Не-yandex и
            lot'ы без source_id / без резолвящегося листинга — пропускаются.

    Returns:
        Число фактически вставленных строк offer_price_history.
    """
    if not lots:
        return 0

    now = datetime.now(UTC)
    prev_ts = now - timedelta(days=1)

    inserted = 0
    skipped = 0
    seed_rejected = 0
    errors = 0

    for lot in lots:
        if lot.source != _SOURCE:
            continue
        if not lot.source_id:
            skipped += 1
            continue

        try:
            with db.begin_nested():
                row = db.execute(
                    _RESOLVE_SQL,
                    {"source_id": lot.source_id},
                ).fetchone()
                if row is None:
                    skipped += 1
                    continue
                listing_id = int(row[0])

                latest = db.execute(
                    _LATEST_SQL,
                    {"listing_id": listing_id},
                ).fetchone()

                # Сдвиг разряда у yandex ловится ТОЛЬКО на посеве (#3385). Ветка «история
                # есть» на проде не записала ни одной строки (из 28 558 строк этого
                # писателя на 17.09): смену listings.price_rub раньше неё пишет триггер
                # record_listing_price_change (save_listings идёт первым), и сюда
                # приходит latest == lot.price_rub. Все 7 скачков ×10 в yandex-истории
                # — пары «посев из price.previous → текущая цена», ошибочна нижняя.
                if latest is None:
                    # История пуста — seed (+ опционально previous-точка).
                    prev = lot.price_previous_rub
                    if (
                        prev is not None
                        and prev != lot.price_rub
                        and _implausible_seed(prev, lot.price_rub)
                    ):
                        seed_rejected += 1
                        logger.warning(
                            "yandex_price_history: price.previous=%s не сеется — на порядок "
                            "от текущей %s (source_id=%s)",
                            prev,
                            lot.price_rub,
                            lot.source_id,
                        )
                    elif prev is not None and prev != lot.price_rub:
                        _insert_point(
                            db,
                            listing_id=listing_id,
                            price_rub=prev,
                            change_time=prev_ts,
                        )
                        inserted += 1
                    _insert_point(
                        db,
                        listing_id=listing_id,
                        price_rub=lot.price_rub,
                        change_time=now,
                    )
                    inserted += 1
                else:
                    latest_price = float(latest[0])
                    if latest_price != float(lot.price_rub):
                        _insert_point(
                            db,
                            listing_id=listing_id,
                            price_rub=lot.price_rub,
                            change_time=now,
                        )
                        inserted += 1
                    else:
                        skipped += 1
        except Exception as exc:
            # Per-lot SAVEPOINT откатился — продолжаем батч, не роняем остальные lot'ы.
            errors += 1
            logger.warning(
                "yandex_price_history: lot failed source_id=%s: %s",
                lot.source_id,
                exc,
            )

    db.commit()
    logger.info(
        "yandex_price_history: lots=%d inserted=%d skipped=%d seed_rejected=%d errors=%d",
        len(lots),
        inserted,
        skipped,
        seed_rejected,
        errors,
    )
    return inserted