set50-system/backend/app/siamchart_factors.py

"""Merge Siamchart fundamental snapshot with tourism signals for the dashboard.

This module loads the locally-collected Siamchart SET50 fundamental snapshot
(``backend/data/siamchart/set50_master.json``) and exposes a per-symbol factor
view that the dashboard can render: PE, EPS (latest + YoY growth), dividend
yield, P/BV, ROE, plus a ``is_dividend`` flag (dividend yield > 0) so the UI can
filter to dividend-paying names.

It is deliberately read-only and pure: it never fetches (that is the collector's
job) and only reads whatever snapshot path is currently on disk. If the snapshot
is missing it returns ``available=False`` so the UI can say so instead of crash.
"""

from __future__ import annotations

import json
from pathlib import Path
from typing import Any, Optional

_DEFAULT_SNAPSHOT = Path(__file__).resolve().parents[1] / "data" / "siamchart" / "set50_master.json"


def _as_float(value: Any) -> Optional[float]:
    if value is None or value == "":
        return None
    try:
        return float(str(value).replace(",", ""))
    except (ValueError, TypeError):
        return None


def _load_snapshot(snapshot_path: Optional[Path] = None) -> dict[str, Any]:
    path = snapshot_path or _DEFAULT_SNAPSHOT
    if not path.exists():
        return {"available": False}
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
        data["available"] = True
        data["_source_path"] = str(path)
        return data
    except (OSError, ValueError):
        return {"available": False}


def build_factor_view(
    snapshot_path: Optional[Path] = None,
) -> dict[str, Any]:
    """Build the per-symbol factor view from the Siamchart snapshot.

    Returns a dict shaped for the dashboard:
        {
          "available": bool,
          "as_of": str | None,
          "source": str,
          "factors": [ {symbol, pe, eps, eps_growth_yoy, dividend_yield,
                        pbv, roe, is_dividend, ...} , ... ],
        }
    """
    snapshot = _load_snapshot(snapshot_path)
    if not snapshot.get("available"):
        return {
            "available": False,
            "as_of": None,
            "source": "siamchart",
            "factors": [],
        }

    details = snapshot.get("details", {})
    rows = snapshot.get("rows", [])

    factors: list[dict[str, Any]] = []
    for row in rows:
        symbol = row.get("symbol")
        if not symbol:
            continue
        detail = details.get(symbol, {})
        ratios = detail.get("ratios", {})
        eps_series = row.get("eps", {})
        eps_yoy_series = row.get("eps_yoy", {})

        # Latest EPS = the most recent year we have — the series is keyed by
        # ascending year (1..5), so we want the LAST non-None value, not the
        # first (which would be the oldest year).
        sorted_eps_values = [eps_series[k] for k in sorted(eps_series) if eps_series.get(k) is not None]
        eps_latest = _as_float(sorted_eps_values[-1]) if sorted_eps_values else None
        # EPS YoY: store_real_data does not embed the web's YoY column (it's
        # computed client-side), so derive the growth of the latest period vs the
        # prior period from the EPS series when both are available.
        eps_growth = None
        if len(sorted_eps_values) >= 2 and sorted_eps_values[-2] not in (None, 0):
            eps_growth = round((sorted_eps_values[-1] - sorted_eps_values[-2]) / abs(sorted_eps_values[-2]) * 100, 2)
        # If the snapshot did carry an explicit YoY (a future source may), prefer it.
        explicit = _as_float(next((v for k, v in sorted(eps_yoy_series.items()) if v is not None), None)) \
            if eps_yoy_series else None
        if explicit is not None:
            eps_growth = explicit

        dividend_yield = _as_float(ratios.get("Yield %") or ratios.get("Yield"))
        pe = _as_float(ratios.get("PE") or ratios.get("P/E") or row.get("pe"))
        pbv = _as_float(ratios.get("P/BV"))
        roe = _as_float(ratios.get("ROE%") or ratios.get("ROAE %") or ratios.get("ROAE%"))
        dps = _as_float(ratios.get("DPS"))

        factors.append(
            {
                "symbol": symbol,
                "company_name": detail.get("symbol") or detail.get("full_name"),
                "pe": pe,
                "eps": eps_latest,
                "eps_growth_yoy": eps_growth,
                "dividend_yield": dividend_yield,
                "dps": dps,
                "pbv": pbv,
                "roe": roe,
                "is_dividend": bool(dividend_yield and dividend_yield > 0),
            }
        )

    factors.sort(key=lambda f: (f["symbol"]))
    return {
        "available": True,
        "as_of": snapshot.get("retrieved_at"),
        "source": "siamchart",
        "factor_count": len(factors),
        "dividend_count": sum(1 for f in factors if f["is_dividend"]),
        "factors": factors,
    }