diff --git a/backend/app/macro_thai.py b/backend/app/macro_thai.py new file mode 100644 index 0000000..fa069af --- /dev/null +++ b/backend/app/macro_thai.py @@ -0,0 +1,132 @@ +"""Thai macro backdrop factor — BOT Thai Economy overview page. + +Source: https://www.bot.or.th/en/thai-economy.html (server-rendered HTML). +Provides a real Thailand macro backdrop requested by the user (data ที่รอบด้าน +ของประเทศ): private consumption %YoY, private investment %YoY, manufacturing +production %YoY, headline/core inflation, unemployment, and tourism arrivals YTD. + +The page renders each indicator as text like: + "Private Consumption Index (%YoY) 4.9% Jun 2026" +so we locate the label and grab the following percent + period. +""" + +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://www.bot.or.th/en/thai-economy.html" + + +class MacroThaiError(Exception): + """Raised when the BOT Thai Economy page cannot be fetched/parsed.""" + + +@dataclass(frozen=True) +class MacroThaiSnapshot: + private_consumption_yoy: Optional[float] = None + private_investment_yoy: Optional[float] = None + manufacturing_yoy: Optional[float] = None + headline_inflation_yoy: Optional[float] = None + core_inflation_yoy: Optional[float] = None + unemployment_pct: Optional[float] = None + tourists_ytd_mn: Optional[float] = None + periods: dict = field(default_factory=dict) + source: str = "bot.thai-economy" + + def to_dict(self) -> dict: + return { + "source": self.source, + "private_consumption_yoy": self.private_consumption_yoy, + "private_investment_yoy": self.private_investment_yoy, + "manufacturing_yoy": self.manufacturing_yoy, + "headline_inflation_yoy": self.headline_inflation_yoy, + "core_inflation_yoy": self.core_inflation_yoy, + "unemployment_pct": self.unemployment_pct, + "tourists_ytd_mn": self.tourists_ytd_mn, + "periods": self.periods, + } + + +# label -> (attribute, expected %-unit) +_LABELS = { + "Private Consumption Index": "private_consumption_yoy", + "Private Investment Index": "private_investment_yoy", + "Manufacturing Production Index": "manufacturing_yoy", + "Headline Inflation": "headline_inflation_yoy", + "Core Inflation": "core_inflation_yoy", + "Unemployment": "unemployment_pct", + "No. of tourists year-to-date": "tourists_ytd_mn", +} + + +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 MacroThaiError(f"failed to fetch {url}: {exc}") from exc + try: + return raw.decode("utf-8") + except UnicodeDecodeError: + return raw.decode("latin-1", "ignore") + + +def _plain(text: str) -> str: + return re.sub(r"\s+", " ", html.unescape(re.sub(r"<[^>]+>", "", text))).strip() + + +def _to_pct(text: str) -> Optional[float]: + m = re.search(r"([-+]?\d+(?:\.\d+)?)\s*%", text) + return float(m.group(1)) if m else None + + +def _has_number(text: str) -> bool: + return bool(re.search(r"\d", text)) + + +def parse_macro_thai_html(html_text: str) -> MacroThaiSnapshot: + text = _plain(html_text) + values: dict = {} + periods: dict = {} + for label, attr in _LABELS.items(): + idx = text.find(label) + if idx < 0: + continue + seg = text[idx : idx + 120] + pct = _to_pct(seg) + if pct is not None: + values[attr] = pct + elif attr == "tourists_ytd_mn": + # tourists YTD is a bare count (million, no %) e.g. "16.2" + m = re.search(r"([-+]?\d+(?:\.\d+)?)", seg) + if m: + values[attr] = float(m.group(1)) + # period: e.g. "Jun 2026" / "Q1/2026" following the number + m = re.search(r"(?:\d+(?:\.\d+)?\s*%?\s*)?([A-Z][a-z]{2}/\d{4}|[A-Z][a-z]{2}\s\d{4}|Q\d/\d{4})", seg) + if m: + periods[attr] = m.group(1) + if not values: + raise MacroThaiError("no recognizable macro indicators found in BOT Thai Economy page") + return MacroThaiSnapshot( + private_consumption_yoy=values.get("private_consumption_yoy"), + private_investment_yoy=values.get("private_investment_yoy"), + manufacturing_yoy=values.get("manufacturing_yoy"), + headline_inflation_yoy=values.get("headline_inflation_yoy"), + core_inflation_yoy=values.get("core_inflation_yoy"), + unemployment_pct=values.get("unemployment_pct"), + tourists_ytd_mn=values.get("tourists_ytd_mn"), + periods=periods, + ) + + +def fetch_macro_thai(timeout: float = 30.0) -> MacroThaiSnapshot: + return parse_macro_thai_html(_fetch(timeout=timeout)) diff --git a/backend/tests/test_macro_thai.py b/backend/tests/test_macro_thai.py new file mode 100644 index 0000000..2539427 --- /dev/null +++ b/backend/tests/test_macro_thai.py @@ -0,0 +1,45 @@ +"""Tests for the Thai macro backdrop collector.""" + +from __future__ import annotations + +import unittest + +from app import macro_thai + + +def _page_html() -> str: + return """ + +
Private Consumption Index (%YoY) 4.9% Jun 2026
+
Private Investment Index (%YoY) 18.1% Jun 2026
+
Manufacturing Production Index (%YoY) -3.1% Jun 2026
+
Headline Inflation (%YoY) 1.95% July 2026
+
Core Inflation (%YoY) 1.34% July 2026
+
Unemployment (%share to Labor force) 0.93% Q1/2026
+
No. of tourists year-to-date (million) 16.2 Jun 2026
+ + """ + + +class MacroThaiParseTest(unittest.TestCase): + def test_parses_all_indicators(self) -> None: + snap = macro_thai.parse_macro_thai_html(_page_html()) + self.assertEqual(snap.private_consumption_yoy, 4.9) + self.assertEqual(snap.private_investment_yoy, 18.1) + self.assertEqual(snap.manufacturing_yoy, -3.1) + self.assertEqual(snap.headline_inflation_yoy, 1.95) + self.assertEqual(snap.unemployment_pct, 0.93) + self.assertEqual(snap.tourists_ytd_mn, 16.2) + + def test_no_data_raises(self) -> None: + with self.assertRaises(macro_thai.MacroThaiError): + macro_thai.parse_macro_thai_html("nothing here") + + def test_to_dict_full(self) -> None: + d = macro_thai.parse_macro_thai_html(_page_html()).to_dict() + self.assertEqual(d["source"], "bot.thai-economy") + self.assertIn("periods", d) + + +if __name__ == "__main__": + unittest.main()