Files
set50-system/backend/app/macro_thai.py
Kunthawat Greethong 166a885fb9 [verified] Add Thai macro backdrop collector (BOT Thai Economy) — real consumption/inflation/unemployment
- macro_thai.py scrapes bot.or.th/en/thai-economy.html (server-rendered, no auth)
- private consumption +4.9%, private investment +18.1%, mfg -3.1%, headline inflation 1.95%, core 1.34%, unemployment 0.93%, tourists YTD 16.2mn
- 3 tests; live verified; macro backdrop layer (req #6)
2026-08-25 20:27:15 +07:00

133 lines
4.8 KiB
Python

"""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))