set50-system/backend/app/energy_thai.py

"""Thai Energy/Refining factor — Thai Oil (TOP) quarterly financial highlights.

Source: https://investor.thaioilgroup.com/en/financial-information/financial-highlights
(Thai Oil PCL, the largest Thai refinery operator). Server-rendered HTML (the
only real XHR is a breadcrumbs API; the financial tables are in the HTML, so we
scrape the table markup per the `har-derived-api-client` guidance).

Data: quarterly and annual financial tables with rows
  Sales Revenue, EBITDA, Net Profit/(Loss), Basic EPS (Million Baht)
and columns like Q2/2026, Q1/2026, Q4/2025, Q3/2025, Q2/2025 (and annual YYYY).

This is the Thai-specific energy/refining proxy the user chose (replacing the
US EIA crack spread, so the factor reflects a Thai company).
"""

from __future__ import annotations

import html
import re
from dataclasses import dataclass, field
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://investor.thaioilgroup.com/en/financial-information/financial-highlights"


class EnergyThaiError(Exception):
    """Raised when the TOP financial page cannot be fetched or parsed."""


@dataclass(frozen=True)
class EnergyThaiSnapshot:
    # quarterly: {period: {"sales":.., "ebitda":.., "net_profit":.., "basic_eps":..}}
    quarterly: dict = field(default_factory=dict)
    annual: dict = field(default_factory=dict)
    source: str = "thaioil"
    as_of: str = ""

    def to_dict(self) -> dict:
        return {
            "source": self.source,
            "as_of": self.as_of,
            "quarterly": self.quarterly,
            "annual": self.annual,
        }


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 EnergyThaiError(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()
    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_financial_table(table_html: str) -> dict:
    """Parse one 'table--financial' table into {columns: [...], rows: {label: [vals]}}."""
    columns: list[str] = []
    rows: dict[str, list[Optional[float]]] = {}
    trs = re.findall(r"<tr[^>]*>(.*?)</tr>", table_html, re.S)
    for tr in trs:
        cells = _cells(tr)
        cells = [c for c in cells if c]
        if not cells:
            continue
        # header row: first cell empty/th, rest are period labels (Q2/2026, 2025)
        if cells[0] == "" and len(cells) > 1 and not _to_float(cells[1]) is None:
            columns = cells[1:]
            continue
        if re.match(r"^(Q\d/\d{4}|\d{4})$", cells[0]):
            columns = cells
            continue
        # section header like "Operating" (colspan) -> skip
        if len(cells) == 1:
            continue
        # data row: [label, val1, val2, ...]
        label = cells[0]
        # normalize common labels
        norm = label.lower()
        key = None
        if "sales revenue" in norm:
            key = "sales"
        elif norm.startswith("ebitda"):
            key = "ebitda"
        elif "net profit" in norm or "net loss" in norm:
            key = "net_profit"
        elif "basic earnings" in norm or "basic e/l" in norm or "eps" in norm:
            key = "basic_eps"
        if key:
            rows[key] = [_to_float(c) for c in cells[1:]]
    return {"columns": columns, "rows": rows}


def parse_energy_thai_html(html_text: str) -> dict:
    """Parse the TOP financial-highlights page into {'quarterly':..., 'annual':...}."""
    tables = re.findall(
        r'<table[^>]*class="[^"]*table--financial[^"]*"[^>]*>(.*?)</table>',
        html_text, re.S,
    )
    if not tables:
        # fallback: any table containing Qx/YYYY header
        tables = [t for t in re.findall(r"<table[^>]*>(.*?)</table>", html_text, re.S)
                  if re.search(r"Q\d/\d{4}", t)]
    if not tables:
        raise EnergyThaiError("no financial tables found in TOP page")

    q = None
    a = None
    for t in tables:
        parsed = _parse_financial_table(t)
        cols = parsed.get("columns", [])
        # quarterly periods look like 'Q1/2026'; annual like '2025'.
        if any(re.match(r"^Q\d/\d{4}$", c) for c in cols):
            q = parsed
        elif any(re.match(r"^\d{4}$", c) for c in cols):
            a = parsed

    if q is None and a is None:
        raise EnergyThaiError("no recognizable financial periods found in TOP page")
    return {"quarterly": q or {}, "annual": a or {}}


def _to_period_map(parsed: dict) -> dict:
    """Convert {columns, rows} into {period: {metric: value}}."""
    cols = parsed.get("columns", [])
    rows = parsed.get("rows", {})
    out: dict = {}
    for i, col in enumerate(cols):
        out[col] = {}
        for key, vals in rows.items():
            if i < len(vals):
                out[col][key] = vals[i]
    return out


def fetch_energy_thai(timeout: float = 30.0) -> EnergyThaiSnapshot:
    html_text = _fetch(timeout=timeout)
    parsed = parse_energy_thai_html(html_text)
    qmap = _to_period_map(parsed["quarterly"])
    amap = _to_period_map(parsed["annual"])
    # as_of = latest quarterly period (first col)
    periods = list(qmap.keys())
    as_of = periods[0] if periods else ""
    return EnergyThaiSnapshot(quarterly=qmap, annual=amap, as_of=as_of)