ems/backend/services/price_importer.py

"""OTE CZ DAM spot price import (15min slots, shared market table)."""

from __future__ import annotations

import json
import logging
from datetime import date, datetime, timedelta, timezone
from typing import Any
from zoneinfo import ZoneInfo

import httpx

from app.config import get_settings

logger = logging.getLogger(__name__)

MARKET_SOURCE = "OTE_CZ"


async def import_ote_prices(
    site_id: int,
    db,
    target_date: date | None = None,
) -> tuple[int, str, float]:
    """
    Stáhne DAM ceny OTE pro zvolený den (nebo „zítřek“ v TZ lokality), uloží 96 slotů (15 min).

    Schéma DB: ``ems.market_interval_price`` má PK ``(market_source, interval_start)``;
    ceny v ``buy_raw_price_czk_kwh`` / ``sell_raw_price_czk_kwh`` (pro OTE stejné).

    Returns:
        ``(počet_slotů, datum_YMD, první_cena_kč_kwh)``. Počet 96 při úspěchu, -1 při chybě.
        První cena je cena prvního 15min slotu dne; při chybě 0.0.
        Datum je prázdný řetězec jen pokud site neexistuje nebo je neplatná timezone.
    """
    row = await db.fetchrow(
        "SELECT timezone FROM ems.site WHERE id = $1",
        site_id,
    )
    if row is None:
        logger.error("import_ote_prices: site id=%s nenalezen", site_id)
        return -1, "", 0.0

    tz_name: str = row["timezone"] or "Europe/Prague"
    try:
        site_tz = ZoneInfo(tz_name)
    except Exception as e:
        logger.error("import_ote_prices: neplatná timezone %r: %s", tz_name, e)
        return -1, "", 0.0

    if target_date is not None:
        target_day = target_date
    else:
        now_local = datetime.now(site_tz)
        target_day = (now_local + timedelta(days=1)).date()
    date_str = target_day.isoformat()

    cet = ZoneInfo("Europe/Prague")
    now_cet = datetime.now(cet)
    tomorrow_cet = (now_cet + timedelta(days=1)).date()
    if target_day == tomorrow_cet:
        cutoff = now_cet.replace(hour=13, minute=30, second=0, microsecond=0)
        if now_cet < cutoff:
            logger.warning(
                "OTE prices for tomorrow may not be available yet (before 13:30 CET)"
            )

    settings = get_settings()
    base_url = settings.ote_api_url.rstrip("/")
    url = f"{base_url}?date={date_str}"
    eur_czk = float(settings.eur_czk_rate)

    try:
        async with httpx.AsyncClient(timeout=30.0) as client:
            resp = await client.get(url)
            resp.raise_for_status()
            body = resp.json()
    except httpx.TimeoutException:
        logger.warning("import_ote_prices: timeout při GET %s", url)
        return -1, date_str, 0.0
    except httpx.HTTPStatusError as e:
        logger.warning(
            "import_ote_prices: HTTP %s při GET %s: %s",
            e.response.status_code,
            url,
            e.response.text[:500],
        )
        return -1, date_str, 0.0
    except httpx.HTTPError as e:
        logger.warning("import_ote_prices: HTTP chyba při GET %s: %s", url, e)
        return -1, date_str, 0.0
    except Exception as e:
        logger.warning("import_ote_prices: neočekávaná chyba při stahování: %s", e)
        return -1, date_str, 0.0

    hourly_eur_mwh: dict[int, float] | None = None
    try:
        points: list[dict[str, Any]] = body["data"]["dataLine"][0]["point"]
        hourly_eur_mwh = {}
        for p in points:
            x = int(p["x"])
            y = float(p["y"])
            hourly_eur_mwh[x] = y
    except (KeyError, TypeError, ValueError, IndexError):
        snippet = json.dumps(body, ensure_ascii=False)[:500]
        logger.error("import_ote_prices: neočekádaná struktura OTE, začátek: %s", snippet)
        return -1, date_str, 0.0

    if len(hourly_eur_mwh) != 24 or set(hourly_eur_mwh.keys()) != set(range(1, 25)):
        logger.error(
            "import_ote_prices: očekáváno 24 bodů x=1..24, dostáno klíče %s",
            sorted(hourly_eur_mwh.keys()),
        )
        return -1, date_str, 0.0

    slots: list[tuple[datetime, datetime, float]] = []
    for h in range(24):
        x = h + 1
        eur_mwh = hourly_eur_mwh[x]
        price_czk_kwh = eur_mwh * eur_czk / 1000.0
        for minute in (0, 15, 30, 45):
            interval_start_local = datetime(
                target_day.year,
                target_day.month,
                target_day.day,
                h,
                minute,
                tzinfo=site_tz,
            )
            interval_start_utc = interval_start_local.astimezone(timezone.utc)
            interval_end_utc = interval_start_utc + timedelta(minutes=15)
            slots.append((interval_start_utc, interval_end_utc, price_czk_kwh))

    for interval_start_utc, interval_end_utc, price in slots:
        await db.execute(
            """
            INSERT INTO ems.market_interval_price (
                market_source,
                interval_start,
                interval_end,
                buy_raw_price_czk_kwh,
                sell_raw_price_czk_kwh,
                currency,
                imported_at
            )
            VALUES ($1, $2, $3, $4, $5, 'CZK', now())
            ON CONFLICT (market_source, interval_start)
            DO UPDATE SET
                interval_end = EXCLUDED.interval_end,
                buy_raw_price_czk_kwh = EXCLUDED.buy_raw_price_czk_kwh,
                sell_raw_price_czk_kwh = EXCLUDED.sell_raw_price_czk_kwh,
                imported_at = now()
            """,
            MARKET_SOURCE,
            interval_start_utc,
            interval_end_utc,
            price,
            price,
        )

    first_price = float(slots[0][2]) if slots else 0.0
    return len(slots), date_str, first_price


if __name__ == "__main__":
    import asyncio
    import os

    import asyncpg
    from dotenv import load_dotenv

    load_dotenv()

    async def test():
        conn = await asyncpg.connect(os.getenv("DATABASE_URL"))
        n, d, fp = await import_ote_prices(1, conn)
        print(f"Uloženo {n} slotů pro {d}, první cena {fp}")
        await conn.close()

    asyncio.run(test())