[verified] Add BOT regional private-consumption collector (North) for regional tourism/consumption factor
- bot_regional.py: parse BOT BTWS_STAT regional report (reportID per region) -> private consumption index + nondurable index + monthly series - North reportID=954 verified live (idx 100.3, series 6mo); region-keyed for extension to other regions - collect_bot_regional.py CLI -> JSON snapshot - 4 tests; full backend suite 152 OK; compileall ok
This commit is contained in:
Kunthawat Greethong
175
backend/app/bot_regional.py
Normal file
175
backend/app/bot_regional.py
Normal file
@@ -0,0 +1,175 @@
|
||||
"""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; extend as other reportIDs are confirmed.
|
||||
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
|
||||
# whose text carries a Thai month + Buddhist year (256X).
|
||||
if "256" in " ".join(cells) and not re.search(r"\d\.\d", " ".join(cells)):
|
||||
# keep only cells carrying a 4-digit BE year (256X) — 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 "",
|
||||
)
|
||||
31
backend/scripts/collect_bot_regional.py
Normal file
31
backend/scripts/collect_bot_regional.py
Normal file
@@ -0,0 +1,31 @@
|
||||
"""Collect BOT regional private-consumption index for a region to a JSON snapshot."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from app import bot_regional
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--region", default="north",
|
||||
help="BOT region (reportID key), default 'north'")
|
||||
parser.add_argument("--timeout", type=float, default=30.0)
|
||||
args = parser.parse_args()
|
||||
|
||||
snap = bot_regional.fetch_region(args.region, timeout=args.timeout)
|
||||
payload = {
|
||||
"source": "bot",
|
||||
"retrieved_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
|
||||
"factor": "consumption_regional",
|
||||
"data": snap.to_dict(),
|
||||
}
|
||||
print(json.dumps(payload, ensure_ascii=False, sort_keys=True, indent=2))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
59
backend/tests/test_bot_regional.py
Normal file
59
backend/tests/test_bot_regional.py
Normal file
@@ -0,0 +1,59 @@
|
||||
"""Tests for the BOT regional consumption parser."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
|
||||
from app import bot_regional
|
||||
|
||||
|
||||
def _bot_html() -> str:
|
||||
return """
|
||||
<html><head><title>RG_NR_042_S3 ดัชนีและเครื่องชี้การอุปโภคบริโภคภาคเอกชนของภาคเหนือ</title></head>
|
||||
<body>
|
||||
<table>
|
||||
<tr><th>ลำดับ</th><th>รายการ</th><th>มิ.ย. 2569 p</th><th>พ.ค. 2569 r</th><th>เม.ย. 2569 r</th></tr>
|
||||
<tr><td>1</td><td>ดัชนีการอุปโภคบริโภคภาคเอกชน 1/</td><td>100.3</td><td>98.4</td><td>101.2</td></tr>
|
||||
<tr><td>2</td><td>ดัชนีการใช้จ่ายสินค้าไม่คงทน 2/</td><td>98.1</td><td>95.8</td><td>100.7</td></tr>
|
||||
<tr><td>3</td><td>ดัชนีสินค้าอุปโภคบริโภคหมุนเวียนเร็ว</td><td>93.9</td><td>90.2</td><td>96.4</td></tr>
|
||||
</table>
|
||||
</body></html>
|
||||
"""
|
||||
|
||||
|
||||
class BotRegionalParseTest(unittest.TestCase):
|
||||
def test_parses_consumption_index(self) -> None:
|
||||
parsed = bot_regional.parse_bot_regional_html(_bot_html())
|
||||
self.assertEqual(parsed["columns"][0], "มิ.ย. 2569 p")
|
||||
self.assertIn("ดัชนีการอุปโภค", str(parsed["rows"].keys()))
|
||||
# consumption index row value
|
||||
key = next(k for k in parsed["rows"] if "ดัชนีการอุปโภค" in k)
|
||||
self.assertEqual(parsed["rows"][key][0], 100.3)
|
||||
|
||||
def test_fetch_region_builds_snapshot(self) -> None:
|
||||
# Use a fake fetch to avoid network
|
||||
actual_fetch = bot_regional._fetch
|
||||
bot_regional._fetch = lambda rid, timeout=30: _bot_html()
|
||||
try:
|
||||
snap = bot_regional.fetch_region("north")
|
||||
self.assertEqual(snap.region, "north")
|
||||
self.assertEqual(snap.consumption_index, 100.3)
|
||||
self.assertEqual(snap.nondurable_index, 98.1)
|
||||
# series is parallel to months
|
||||
self.assertIsNotNone(snap.months)
|
||||
self.assertIsNotNone(snap.consumption_series)
|
||||
self.assertEqual(len(snap.months), len(snap.consumption_series))
|
||||
finally:
|
||||
bot_regional._fetch = actual_fetch
|
||||
|
||||
def test_unknown_region_raises(self) -> None:
|
||||
with self.assertRaises(bot_regional.BotRegionalError):
|
||||
bot_regional.fetch_region("mars")
|
||||
|
||||
def test_missing_table_raises(self) -> None:
|
||||
with self.assertRaises(bot_regional.BotRegionalError):
|
||||
bot_regional.parse_bot_regional_html("<html>no data</html>")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user