fdfd832fae
- header month-cell detection now matches any 4-digit BE year (25XX) not just 256X - northeast reportID 955 disabled: returns 0.0 BE-2570 placeholder page, no real data yet (documented) - north 954 stays enabled (live idx 100.3); tests + full suite 152 OK
180 lines
7.2 KiB
Python
180 lines
7.2 KiB
Python
"""BOT regional consumption factor — BOT (Bank of Thailand) regional private consumption index.
|
|
|
|
Source: https://app.bot.or.th/BTWS_STAT/statistics/ReportPage.aspx?reportID=<id>
|
|
The Bank of Thailand publishes regional breakdowns of the private consumption
|
|
index per region (North, Northeast, South, Central/Bangkok). Each region has its
|
|
own reportID. Server-rendered HTML table:
|
|
|
|
header row: months (e.g. "มิ.ย. 2569 p พ.ค. 2569 r ..." — p/r are preliminary/revision)
|
|
rows | 1 | ดัชนีการอุปโภคบริโภคภาคเอกชน | 100.3 | 98.4 | ...
|
|
| 2 | ดัชนีการใช้จ่ายสินค้าไม่คงทน | ...
|
|
| 3 | ดัชนีสินค้าอุปโภคบริโภคหมุนเวียนเร็ว | ...
|
|
|
|
We parse the main private-consumption index row per region so the tourism/
|
|
consumption theme can sense regional demand shifts.
|
|
|
|
Region reportIDs (from BOT BTWS_STAT):
|
|
- North: 954
|
|
(others to be confirmed by browsing the BOT statistical menu)
|
|
"""
|
|
|
|
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"
|
|
)
|
|
_BASE_URL = "https://app.bot.or.th/BTWS_STAT/statistics/ReportPage.aspx?reportID="
|
|
|
|
# Default mapping of BOT regions (BTWS_STAT regional consumption reports).
|
|
# Only regions with REAL published values are enabled. north (RG_NR_042, 954)
|
|
# is confirmed live (idx 100.3). northeast (RG_NE_044, 955) currently returns a
|
|
# 0.0 placeholder (BE 2570 forecast page, no real data yet) so it is NOT enabled
|
|
# until BOT publishes real values. south/central reportIDs pending confirmation.
|
|
REGION_REPORT_IDS: dict[str, int] = {
|
|
"north": 954,
|
|
}
|
|
|
|
|
|
class BotRegionalError(Exception):
|
|
"""Raised when a BOT regional report cannot be fetched or parsed."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class BotRegionalSnapshot:
|
|
region: str
|
|
report_id: int
|
|
# Monthly private consumption index: {(year, month_label): value}
|
|
# For simplicity store the latest 6 month values as an ordered list along
|
|
# with their labels.
|
|
consumption_index: Optional[float] = None
|
|
nondurable_index: Optional[float] = None
|
|
months: Optional[list] = None # month labels in order (latest first)
|
|
consumption_series: Optional[list] = None # values parallel to months
|
|
source: str = "bot"
|
|
as_of: str = ""
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"source": self.source,
|
|
"region": self.region,
|
|
"report_id": self.report_id,
|
|
"as_of": self.as_of,
|
|
"consumption_index": self.consumption_index,
|
|
"nondurable_index": self.nondurable_index,
|
|
"months": self.months or [],
|
|
"consumption_series": self.consumption_series or [],
|
|
}
|
|
|
|
|
|
def _fetch(report_id: int, timeout: float = 30.0) -> str:
|
|
url = f"{_BASE_URL}{report_id}"
|
|
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 BotRegionalError(f"failed to fetch BOT report {report_id}: {exc}") from exc
|
|
try:
|
|
return raw.decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
return raw.decode("latin-1", "ignore")
|
|
|
|
|
|
def _clean(event: str) -> str:
|
|
return html.unescape(re.sub(r"<[^>]+>", "", event)).strip()
|
|
|
|
|
|
def parse_bot_regional_html(html_text: str) -> dict:
|
|
"""Parse the BOT regional consumption table into row label -> list of values.
|
|
|
|
Returns {'title': ..., 'columns': [month labels], 'rows': {label: [values]}}.
|
|
"""
|
|
tables = re.findall(r"<table[^>]*>(.*?)</table>", html_text, re.S)
|
|
if not tables:
|
|
raise BotRegionalError("no tables found in BOT page")
|
|
|
|
title = ""
|
|
columns: list[str] = []
|
|
rows: dict[str, list[Optional[float]]] = {}
|
|
|
|
m = re.search(r"<title>(.*?)</title>", html_text, re.S)
|
|
if m:
|
|
title = _clean(m.group(1))
|
|
|
|
for table in tables:
|
|
trs = re.findall(r"<tr[^>]*>(.*?)</tr>", table, re.S)
|
|
for tr in trs:
|
|
cells = [_clean(td) for td in re.findall(r"<t[dh][^>]*>(.*?)</t[dh]>", tr, re.S)]
|
|
cells = [c for c in cells if c]
|
|
if not cells:
|
|
continue
|
|
# header row: month labels (e.g. 'มิ.ย. 2569 p') — keep only cells
|
|
# carrying a 4-digit BE year (25XX, may be 2569/2570/...).
|
|
if re.search(r"25\d\d", " ".join(cells)) and not re.search(r"\d\.\d", " ".join(cells)):
|
|
# keep only cells carrying a 4-digit BE year (25XX) — drops
|
|
# column titles like 'ลำดับ' / 'รายการ'.
|
|
month_cells = [c for c in cells if re.search(r"25\d\d", c)]
|
|
if month_cells:
|
|
columns = month_cells
|
|
continue
|
|
# data row: [no, label, val1, val2, ...] — label is cells[1], values
|
|
# start at cells[2]. Some rows may omit the no/label (merged), in
|
|
# which case label stays from a previous row.
|
|
if len(cells) >= 2:
|
|
# Try to identify (no, label, values...) by checking cells[1]
|
|
label = cells[1].strip()
|
|
val_start = 2
|
|
if not label or label in ("-", "N/A"):
|
|
# fallback: maybe [label, val1, val2...] with no 'no' column
|
|
label = cells[0].strip()
|
|
val_start = 1
|
|
vals = [_to_float(c) for c in cells[val_start:]]
|
|
if any(v is not None for v in vals):
|
|
rows[label or f"row_{len(rows)}"] = vals
|
|
|
|
if not rows:
|
|
raise BotRegionalError("no numeric rows found in BOT regional page")
|
|
return {"title": title, "columns": columns, "rows": rows}
|
|
|
|
|
|
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 fetch_region(region: str = "north", timeout: float = 30.0) -> BotRegionalSnapshot:
|
|
region = region.strip().lower()
|
|
report_id = REGION_REPORT_IDS.get(region)
|
|
if report_id is None:
|
|
raise BotRegionalError(f"no reportID configured for region: {region}")
|
|
html_text = _fetch(report_id, timeout=timeout)
|
|
parsed = parse_bot_regional_html(html_text)
|
|
rows = parsed["rows"]
|
|
cols = parsed["columns"]
|
|
|
|
# The main private consumption index is usually the first numeric row.
|
|
main_key = next((k for k in rows if "ดัชนีการอุปโภค" in k or "ดัชนี" in k), None)
|
|
nondurable_key = next((k for k in rows if "ไม่คงทน" in k), None)
|
|
|
|
return BotRegionalSnapshot(
|
|
region=region,
|
|
report_id=report_id,
|
|
consumption_index=rows[main_key][0] if main_key and rows.get(main_key) else None,
|
|
nondurable_index=rows[nondurable_key][0] if nondurable_key and rows.get(nondurable_key) else None,
|
|
months=cols,
|
|
consumption_series=rows.get(main_key) if main_key else None,
|
|
as_of=cols[0] if cols else "",
|
|
)
|