set50-system/backend/app/refining_energy.py

"""Refining / Energy alternative factor — EIA Wholesale Spot Petroleum Prices.

Source: https://www.eia.gov/todayinenergy/prices.php (US Energy Information
Administration, open public data). The page is server-rendered HTML (no hidden
JSON API needed) and is **not** behind a bot challenge, so `urllib` fetch works
directly — unlike RBN Energy which sits behind a Cloudflare challenge (403 via
urllib).

Table: "Wholesale Spot Petroleum Prices, <date> Close"
  Product / Area / Price / PercentChange*
  Crude Oil ($/barrel)      WTI 87.21 -2.8 / Brent 96.92 +3.1 / Louisian. Light 90.71 -2.7
  Gasoline (RBOB) ($/gal)   NY 3.42 / Gulf 3.52 / LA 3.52
  3:2:1 Crack Spread ($/bbl) Gulf Coast (LLS) 71.10 +5.7
  ... (heating oil, diesel, natural gas)

The 3:2:1 crack spread and crude benchmarks are the refining-margin proxy the
energy theme scores on.
"""

from __future__ import annotations

import html
import re
from dataclasses import dataclass
from typing import Optional
from urllib.request import Request, urlopen

_USER_AGENT = (
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36"
)
_URL = "https://www.eia.gov/todayinenergy/prices.php"


class RefiningError(Exception):
    """Raised when the EIA page cannot be fetched or parsed."""


@dataclass(frozen=True)
class RefiningSnapshot:
    crack_spread: Optional[float] = None        # 3:2:1 crack spread $/bbl
    crack_spread_change: Optional[float] = None  # +5.7
    wti_crude: Optional[float] = None
    brent_crude: Optional[float] = None
    gasoline_gulf: Optional[float] = None
    as_of: str = ""
    source: str = "eia"

    def to_dict(self) -> dict:
        return {
            "source": self.source,
            "as_of": self.as_of,
            "crack_spread": self.crack_spread,
            "crack_spread_change": self.crack_spread_change,
            "wti_crude": self.wti_crude,
            "brent_crude": self.brent_crude,
            "gasoline_gulf": self.gasoline_gulf,
        }


def _fetch(url: str = _URL, timeout: float = 30.0) -> str:
    req = Request(url, headers={"User-Agent": _USER_AGENT, "Accept": "text/html"})
    try:
        with urlopen(req, timeout=timeout) as resp:
            raw = resp.read()
    except Exception as exc:
        raise RefiningError(f"failed to fetch {url}: {exc}") from exc
    try:
        return raw.decode("utf-8")
    except UnicodeDecodeError:
        return raw.decode("latin-1", "ignore")


def _to_float(text: str) -> Optional[float]:
    text = text.replace(",", "").strip().replace("+", "")
    if not text or text in ("-", "N/A"):
        return None
    try:
        return float(text)
    except ValueError:
        return None


def _cells(row_html: str) -> list[str]:
    return [
        html.unescape(re.sub(r"<[^>]+>", "", td)).strip()
        for td in re.findall(r"<t[dh][^>]*>(.*?)</t[dh]>", row_html, re.S)
    ]


def parse_refining_html(html_text: str) -> RefiningSnapshot:
    """Parse the EIA Today in Energy daily prices page."""
    tables = re.findall(r"<table[^>]*>(.*?)</table>", html_text, re.S)
    if not tables:
        raise RefiningError("no tables found in EIA prices page")

    crack = crack_chg = None
    wti = brent = gasoline_gulf = None
    current_product = ""

    for table in tables:
        rows = re.findall(r"<tr[^>]*>(.*?)</tr>", table, re.S)
        for row_html in rows:
            cells = _cells(row_html)
            text = " ".join(cells)
            low = text.lower()
            # Track the product section from its header row (cells[0] non-empty,
            # e.g. "Gasoline (RBOB) ($/gallon)" / "Heating Oil ($/gallon)").
            if cells and cells[0].strip():
                c0 = cells[0].lower()
                # Only update product section from a product header row (which
                # carries a unit like "($/barrel)" / "($/gallon)") — Area-only
                # sub-rows like 'Gulf Coast' / 'Brent' must not reset it.
                if "($/" in c0 or "($ /" in c0:
                    if "gasoline" in c0 or "rbbob" in c0:
                        current_product = "gasoline"
                    elif "heating oil" in c0:
                        current_product = "heating_oil"
                    elif "low-sulfur diesel" in c0 or "diesel" in c0:
                        current_product = "diesel"
                    else:
                        current_product = c0
            if len(cells) >= 2:
                area = cells[1].strip().lower()
                c0 = cells[0].strip().lower() if cells and cells[0].strip() else ""
                # When the row's Area label sits in cells[0] (Product column not
                # repeated, e.g. ['Brent','96.92','+3.1']), price is cells[1];
                # when it sits in cells[1] (e.g. ['','Brent','96.92','+3.1']),
                # price is cells[2].
                if c0 == "brent" or area == "brent":
                    brent = (_to_float(cells[1]) if c0 == "brent" else (_to_float(cells[2]) if len(cells) > 2 else None))
                elif c0 == "wti" or area == "wti":
                    wti = (_to_float(cells[1]) if c0 == "wti" else (_to_float(cells[2]) if len(cells) > 2 else None))
                elif current_product == "gasoline" and (c0 == "gulf coast" or "gulf coast" in area):
                    gasoline_gulf = (_to_float(cells[1]) if c0 == "gulf coast" else (_to_float(cells[2]) if len(cells) > 2 else None))
            if "3:2:1 crack spread" in low:
                # row: '3:2:1 Crack Spread ($/barrel)', 'Gulf Coast (LLS)', '71.10', '+5.7'
                for k in range(len(cells)):
                    if "gulf coast" in cells[k].lower():
                        crack = _to_float(cells[k + 1]) if k + 1 < len(cells) else None
                        crack_chg = _to_float(cells[k + 2]) if k + 2 < len(cells) else None
                        break
                if crack is None and len(cells) >= 3:
                    crack = _to_float(cells[-2])
                    crack_chg = _to_float(cells[-1])

    if crack is None and wti is None and brent is None:
        raise RefiningError("no refining/energy values found in EIA prices page")

    return RefiningSnapshot(
        crack_spread=crack,
        crack_spread_change=crack_chg,
        wti_crude=wti,
        brent_crude=brent,
        gasoline_gulf=gasoline_gulf,
    )


def fetch_refining(timeout: float = 30.0) -> RefiningSnapshot:
    html_text = _fetch(timeout=timeout)
    return parse_refining_html(html_text)