[verified] Add Thai Energy/Refining collector (TOP quarterly financials) replacing EIA US crack spread

- energy_thai.py: scrape Thai Oil (TOP) investor financial-highlights -> quarterly + annual EBITDA/Net Profit/Sales (Million Baht), largest Thai refinery
- Thai-specific factor per user (energy must reflect Thai companies, not US EIA proxy); Krungsri was projection-only, TOP gives real quarterly actuals
- Frequency: quarterly (documented in research note)
- 4 tests; full backend suite 156 OK; compileall ok; static scan clean

backend/app/energy_thai.py

@@ -0,0 +1,170 @@
+"""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)

backend/scripts/collect_energy_thai.py

@@ -0,0 +1,29 @@
+"""Collect the Thai Energy/Refining factor (TOP quarterly financials) to JSON snapshot."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from datetime import datetime, timezone
+
+from app import energy_thai
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--timeout", type=float, default=30.0)
+    args = parser.parse_args()
+
+    snap = energy_thai.fetch_energy_thai(timeout=args.timeout)
+    payload = {
+        "source": "thaioil",
+        "retrieved_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
+        "factor": "energy_thai",
+        "data": snap.to_dict(),
+    }
+    print(json.dumps(payload, ensure_ascii=False, sort_keys=True, indent=2))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

backend/tests/test_energy_thai.py

@@ -0,0 +1,68 @@
+"""Tests for the Thai Energy/Refining (TOP financial highlights) parser."""
+
+from __future__ import annotations
+
+import unittest
+
+from app import energy_thai
+
+
+def _top_html() -> str:
+    return """
+    <html><body>
+    <table class="table table--financial">
+      <thead><tr><th class="persist"></th><th>2025</th><th>2024</th><th>2023</th></tr></thead>
+      <tbody>
+        <tr class="table__header"><td colspan="4">Operating</td></tr>
+        <tr><td>Sales Revenue</td><td>394,336</td><td>455,857</td><td>459,402</td></tr>
+        <tr><td>EBITDA</td><td>17,619</td><td>22,026</td><td>35,453</td></tr>
+        <tr><td>Net Profit / (Loss)</td><td>14,584</td><td>9,959</td><td>19,443</td></tr>
+        <tr><td>Basic Earnings / (Loss) (Baht / Share)</td><td>6.53</td><td>4.46</td><td>8.70</td></tr>
+      </tbody>
+    </table>
+    <table class="table table--financial">
+      <thead><tr><th class="persist"></th><th>Q2/2026</th><th>Q1/2026</th><th>Q4/2025</th></tr></thead>
+      <tbody>
+        <tr class="table__header"><td colspan="4">Operating</td></tr>
+        <tr><td>Sales Revenue</td><td>129,709</td><td>114,809</td><td>108,931</td></tr>
+        <tr><td>EBITDA</td><td>8,915</td><td>31,641</td><td>5,981</td></tr>
+        <tr><td>Net Profit / (Loss)</td><td>8,284</td><td>19,481</td><td>2,458</td></tr>
+        <tr><td>Basic Earnings / (Loss) (Baht / Share)</td><td>3.71</td><td>8.41</td><td>1.10</td></tr>
+      </tbody>
+    </table>
+    </body></html>
+    """
+
+
+class EnergyThaiParseTest(unittest.TestCase):
+    def test_parses_quarterly_and_annual(self) -> None:
+        parsed = energy_thai.parse_energy_thai_html(_top_html())
+        self.assertIn("Q2/2026", parsed["quarterly"]["columns"])
+        self.assertIn("2025", parsed["annual"]["columns"])
+        self.assertEqual(parsed["quarterly"]["rows"]["net_profit"][0], 8284.0)
+        self.assertEqual(parsed["annual"]["rows"]["ebitda"][0], 17619.0)
+
+    def test_period_map(self) -> None:
+        parsed = energy_thai.parse_energy_thai_html(_top_html())
+        qmap = energy_thai._to_period_map(parsed["quarterly"])
+        self.assertEqual(qmap["Q2/2026"]["net_profit"], 8284.0)
+        self.assertEqual(qmap["Q2/2026"]["sales"], 129709.0)
+        amap = energy_thai._to_period_map(parsed["annual"])
+        self.assertEqual(amap["2025"]["ebitda"], 17619.0)
+
+    def test_no_tables_raises(self) -> None:
+        with self.assertRaises(energy_thai.EnergyThaiError):
+            energy_thai.parse_energy_thai_html("<html>no data</html>")
+
+    def test_negative_net_profit_parses(self) -> None:
+        html_text = """
+        <table class="table table--financial">
+          <tr><th></th><th>Q1/2026</th></tr>
+          <tr><td>Net Profit / (Loss)</td><td>-5,380</td></tr>
+        </table>"""
+        parsed = energy_thai.parse_energy_thai_html(html_text)
+        self.assertEqual(parsed["quarterly"]["rows"]["net_profit"][0], -5380.0)
+
+
+if __name__ == "__main__":
+    unittest.main()

docs/alternative-factor-source-research.md

@@ -14,7 +14,8 @@ The US EIA 3-2-1 crack spread is a **US** proxy; the user wants **Thai** refiner
 - **TOP (Thai Oil)** analyst-meeting PDFs (`top.listedcompany.com`) — GRM/GIM per quarter, $/bbl.
 - **EPPO / MOPH** — Thai retail fuel prices (more frequent, but that's fuel price, not GRM).
 
-**Decision state:** Thai GRM quarterly is the direction; exact source (Krungsri aggregate vs TOP) pending user confirmation.
+**Decision (final):** Use **Thai Oil (TOP) quarterly financial highlights** (`investor.thaioilgroup.com/en/financial-information/financial-highlights`) — real quarterly EBITDA/Net Profit/Sales of the largest Thai refinery (Million Baht), scraped from server-rendered HTML. Implemented in `backend/app/energy_thai.py`. This is Thai-specific and replaces both the US EIA crack spread and the Krungsri outlook projection. Frequency: **quarterly**. (Krungsri Research remains a qualitative backdrop; EIA is a US proxy — both superseded by TOP for the factor.)
+