fcc0da9c8d
- add te_thailand collector (TradingEconomics) -> 8 factors: interest rate, business loan growth, consumer credit, household debt/GDP, retail sales YoY, consumer confidence, residential property prices, business confidence; feed banks/retail/consumer_staples/nonbank_finance/property/telecom/healthcare - add thai_trade collector (TradingEconomics external sector) -> exports/ imports/current-account factors (concurrent in-tree work, verified green) - fix sign inversion: theme weights were negative on sign:-1 factors (NPL, inflation, unemployment) so higher NPL/inflation RAISED scores; direction now lives only in factor sign, theme weights positive (regression-locked) - tests: te_thailand parse+direction, value-key resolution contract, dashboard 8-sources, scheduler vintage counts; suite 362 OK
126 lines
4.3 KiB
Python
126 lines
4.3 KiB
Python
"""Thailand external-sector / trade factor — TradingEconomics Thailand.
|
|
|
|
Source: https://tradingeconomics.com/thailand/current-account
|
|
(server-rendered HTML, same infobox/infrastructure as the auto_credit source —
|
|
the "related indicators" table lists the current value, previous value, unit
|
|
and reporting period for Thailand's external account).
|
|
|
|
Verified live: the current-account page's related-indicators table exposes
|
|
(Jul 2026) Exports 34,781.10, Imports 38,399.60, Balance of Trade -3,611.00
|
|
USD mn and (Jun 2026) Current Account -3,473.85 USD mn. Replaces the stale
|
|
BOT report-60 path (which only carried data through 2011) with a CURRENT,
|
|
free, scrapeable source per the plan's "do not ship a wrong/stale series" gate.
|
|
"""
|
|
|
|
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"
|
|
)
|
|
_URL = "https://tradingeconomics.com/thailand/current-account"
|
|
|
|
|
|
class ThaiTradeError(Exception):
|
|
"""Raised when the TradingEconomics Thailand page cannot be fetched/parsed."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ThaiTradeSnapshot:
|
|
current_account_usdm: Optional[float] = None # USD million
|
|
trade_balance_usdm: Optional[float] = None # USD million
|
|
exports_usdm: Optional[float] = None # USD million
|
|
imports_usdm: Optional[float] = None # USD million
|
|
as_of: str = ""
|
|
source: str = "tradingeconomics.thai-trade"
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"source": self.source,
|
|
"as_of": self.as_of,
|
|
"current_account_usdm": self.current_account_usdm,
|
|
"trade_balance_usdm": self.trade_balance_usdm,
|
|
"exports_usdm": self.exports_usdm,
|
|
"imports_usdm": self.imports_usdm,
|
|
}
|
|
|
|
|
|
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 ThaiTradeError(f"failed to fetch {url}: {exc}") from exc
|
|
try:
|
|
return raw.decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
return raw.decode("latin-1", "ignore")
|
|
|
|
|
|
def _tables(html_text: str) -> list[str]:
|
|
return re.findall(r"<table[^>]*>(.*?)</table>", html_text, re.S)
|
|
|
|
|
|
def _cells(row_html: str) -> list[str]:
|
|
return [
|
|
html.unescape(re.sub(r"<[^>]+>", "", td)).strip()
|
|
for td in re.findall(r"<td[^>]*>(.*?)</td>", row_html, re.S)
|
|
if td.strip()
|
|
]
|
|
|
|
|
|
def _to_float(text: str) -> Optional[float]:
|
|
text = text.replace(",", "").strip().replace("%", "")
|
|
if not text or text in ("-", "N/A", "…"):
|
|
return None
|
|
try:
|
|
return float(text)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def parse_thai_trade_html(html_text: str) -> ThaiTradeSnapshot:
|
|
"""Parse the TradingEconomics Thailand current-account related table."""
|
|
current = trade = exports = imports = None
|
|
as_of = ""
|
|
|
|
for table in _tables(html_text):
|
|
rows = re.findall(r"<tr[^>]*>(.*?)</tr>", table, re.S)
|
|
for row_html in rows:
|
|
cells = _cells(row_html)
|
|
if not cells:
|
|
continue
|
|
label = cells[0].strip()
|
|
if label == "Current Account":
|
|
current = _to_float(cells[1])
|
|
if len(cells) > 4:
|
|
as_of = cells[4]
|
|
elif label == "Balance of Trade":
|
|
trade = _to_float(cells[1])
|
|
elif label == "Exports" and exports is None:
|
|
exports = _to_float(cells[1])
|
|
elif label == "Imports" and imports is None:
|
|
imports = _to_float(cells[1])
|
|
|
|
if current is None and exports is None and imports is None:
|
|
raise ThaiTradeError("no Thai trade series found in TradingEconomics page")
|
|
|
|
return ThaiTradeSnapshot(
|
|
current_account_usdm=current,
|
|
trade_balance_usdm=trade,
|
|
exports_usdm=exports,
|
|
imports_usdm=imports,
|
|
as_of=as_of,
|
|
)
|
|
|
|
|
|
def fetch_thai_trade(timeout: float = 30.0) -> ThaiTradeSnapshot:
|
|
return parse_thai_trade_html(_fetch(timeout=timeout))
|