set50-system/backend/app/auto_npl.py

"""Auto NPL / credit-quality factor — BOT Gross NPLs by business type.

Source: https://app.bot.or.th/BTWS_STAT/statistics/ReportPage.aspx?reportID=794
(Bank of Thailand, FI_NP_003_S2 — Gross NPLs outstanding classified by business
type, ISIC Rev.4). Server-rendered HTML table.

Table layout (quarterly, 3 metrics per period):
  header2: [ , , 'ยอดคงค้าง NPL', '% ต่อ NPLs', '% ต่อสินเชื่อรวม', (repeat) ]
  row:    [no, sector_label, NPL_amount, pct_of_npls, pct_of_loans, (prev period repeat) ]

This module extracts the **auto loan NPL** (sector 'รถยนต์') — the credit-quality
component that deepens the auto_credit theme beyond just new-car sales.
"""

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://app.bot.or.th/BTWS_STAT/statistics/ReportPage.aspx?reportID=794"
AUTO_LOAN_LABEL = "รถยนต์"


class AutoNplError(Exception):
    """Raised when the BOT NPL page cannot be fetched or parsed."""


@dataclass(frozen=True)
class AutoNplSnapshot:
    # latest period auto-loan NPL (million baht), % of NPLs, % of total loans
    npl_amount: Optional[float] = None
    pct_of_npls: Optional[float] = None   # % of total NPLs
    pct_of_loans: Optional[float] = None  # % of total loans
    period: str = ""
    source: str = "bot"

    def to_dict(self) -> dict:
        return {
            "source": self.source,
            "period": self.period,
            "npl_amount": self.npl_amount,
            "pct_of_npls": self.pct_of_npls,
            "pct_of_loans": self.pct_of_loans,
        }


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 AutoNplError(f"failed to fetch {url}: {exc}") from exc
    try:
        return raw.decode("utf-8")
    except UnicodeDecodeError:
        return raw.decode("latin-1", "ignore")


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 _cells(row_html: str) -> list[str]:
    return [
        html.unescape(re.sub(r"<[^>]+>", "", td)).strip()
        for td in re.findall(r"<t[dh][^>]*>(.*?)</t[dh]>", row_html, re.S)
        if td.strip()
    ]


def parse_auto_npl_html(html_text: str, label: str = AUTO_LOAN_LABEL) -> AutoNplSnapshot:
    """Parse the BOT NPL table; return the auto-loan-sector row (latest period).

    Thin wrapper — the BOT 794 page lists many business sectors, so the actual
    extraction is shared via :func:`_parse_sector`. The ``auto`` variant keeps the
    sector label (default 'รถยนต์') for the auto_credit theme; other themes can
    reuse report 794 with their own sector label (e.g. banks -> financial sector).
    """
    row = _parse_sector(html_text, label)
    if row is None:
        raise AutoNplError(f"no {label!r} NPL row found in BOT NPL page")
    return AutoNplSnapshot(
        npl_amount=row[0], pct_of_npls=row[1], pct_of_loans=row[2], period=row[3],
    )


def _parse_sector(html_text: str, label: str) -> Optional[tuple]:
    """Return (npl, pct_of_npls, pct_of_loans, period) for the given sector label."""
    tables = re.findall(r"<table[^>]*>(.*?)</table>", html_text, re.S)
    period = ""
    npl = pct_n = pct_l = None
    for table in tables:
        trs = re.findall(r"<tr[^>]*>(.*?)</tr>", table, re.S)
        for tr in trs:
            cells = _cells(tr)
            if not cells:
                continue
            # header row carries the period (e.g. Q2/2568); cells[0] may be '' or 'ยอด'
            m = re.match(r"(Q\d/\d{4})", " ".join(cells))
            if m and period == "":
                period = m.group(1)
                continue
            # data row: [no, label, value, pct_npl, pct_loan, ...]
            if len(cells) >= 4 and cells[0].isdigit() and label in cells[1]:
                npl = _to_float(cells[2])
                pct_n = _to_float(cells[3])
                pct_l = _to_float(cells[4]) if len(cells) > 4 else None
                break
    if npl is None and pct_n is None:
        return None
    return (npl, pct_n, pct_l, period)


def fetch_auto_npl(timeout: float = 30.0) -> AutoNplSnapshot:
    return parse_auto_npl_html(_fetch(timeout=timeout))