set50-system/backend/app/auto_credit.py

"""Auto Credit alternative factor — scrape real Thai vehicle sales from Trading Economics.

Source: https://tradingeconomics.com/thailand/total-vehicle-sales
Server-rendered HTML (per `har-derived-api-client` guidance, no JSON API — the
HAR capture shows only ads/analytics). Three tables carry the data:

  Table 1 (announcements): "New Car Sales YoY" rows — latest YoY% per month.
  Table 2 (related indicators): Auto Exports, Vehicle Production, Passenger
      Car Sales current values.
  Table 3 (main series): Total Vehicle Sales `[latest, prev, high, low,
      range, units, freq, seasonality]`.

We parse these so the Auto Credit factor can use vehicle-sales momentum (YoY%)
and the current total, which is the real, non-mock input the user asked for.
"""

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/total-vehicle-sales"


class AutoCreditError(Exception):
    """Raised when the Trading Economics page cannot be fetched or parsed."""


@dataclass(frozen=True)
class AutoCreditSnapshot:
    total_vehicle_sales: Optional[float] = None       # latest units
    prev_vehicle_sales: Optional[float] = None        # prior period units
    new_car_sales_yoy: Optional[float] = None         # latest % YoY
    prev_new_car_sales_yoy: Optional[float] = None
    vehicle_production: Optional[float] = None
    passenger_car_sales: Optional[float] = None
    auto_exports: Optional[float] = None
    as_of: str = ""
    source: str = "tradingeconomics"

    def to_dict(self) -> dict:
        return {
            "source": self.source,
            "as_of": self.as_of,
            "total_vehicle_sales": self.total_vehicle_sales,
            "prev_vehicle_sales": self.prev_vehicle_sales,
            "new_car_sales_yoy": self.new_car_sales_yoy,
            "prev_new_car_sales_yoy": self.prev_new_car_sales_yoy,
            "vehicle_production": self.vehicle_production,
            "passenger_car_sales": self.passenger_car_sales,
            "auto_exports": self.auto_exports,
        }


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 AutoCreditError(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)
    ]


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_auto_credit_html(html_text: str) -> AutoCreditSnapshot:
    """Parse the Trading Economics Thailand total-vehicle-sales page."""
    tables = _tables(html_text)
    if not tables:
        raise AutoCreditError("no tables found in Trading Economics page")

    total = prev = None
    prod = pass_sales = exports = None
    car_yoy = prev_yoy = None

    for table in tables:
        rows = re.findall(r"<tr[^>]*>(.*?)</tr>", table, re.S)
        for row_html in rows:
            cells = _cells(row_html)
            if not cells:
                continue
            joined = " | ".join(cells)
            # Main series row: ['', '61244.00', '57765.00', '157529','5338','1980-2026','Units',...]
            # Detect it as the row with 8+ cells where cell[6] == 'Units'.
            if len(cells) >= 7 and cells[6].strip().lower() == "units" and "Monthly" in joined:
                total = _to_float(cells[1])
                prev = _to_float(cells[2])
            # Related indicators table (label, value, prev, unit, period)
            elif cells[0].strip().lower() == "vehicle production":
                prod = _to_float(cells[1])
            elif cells[0].strip().lower() == "passenger car sales":
                pass_sales = _to_float(cells[1])
            elif cells[0].strip().lower() == "auto exports":
                exports = _to_float(cells[1])
            # New Car Sales YoY announcements: [date,time,'New Car Sales YoY','May','10.60%',...]
            elif cells[0].lower().startswith("202") and "New Car Sales YoY" in joined:
                if len(cells) >= 5:
                    pct = _to_float(cells[4])
                    if pct is None:
                        continue  # pending row (e.g. current period not reported yet)
                    if car_yoy is None:
                        car_yoy = pct
                    else:
                        prev_yoy = car_yoy
                        car_yoy = pct

    if total is None and car_yoy is None and prod is None:
        raise AutoCreditError("no auto-credit values found in Trading Economics page")

    return AutoCreditSnapshot(
        total_vehicle_sales=total,
        prev_vehicle_sales=prev,
        new_car_sales_yoy=car_yoy,
        prev_new_car_sales_yoy=prev_yoy,
        vehicle_production=prod,
        passenger_car_sales=pass_sales,
        auto_exports=exports,
    )


def fetch_auto_credit(timeout: float = 30.0) -> AutoCreditSnapshot:
    html_text = _fetch(timeout=timeout)
    return parse_auto_credit_html(html_text)