[verified] Apply R1-R5 (factor formula) + real bank-sector NPL collector

(a) R1-R5 (factor-refinement, grounded in methodology-research.md):
- R1 (PEAD): EPS-growth weight raised 1.0->1.5 in build_siamchart_score / symbol_breakdown (Bernard-Thomas 1990, Livnat-Mendenhall 2006)
- R2 (momentum): 12-1 momentum factor from Yahoo price snapshot (Jegadeesh-Titman 93; lite weight 0.5)
- R3 (regime): binary bear gate -> continuous stress = negative-themes fraction, smooth LONG/SHORT shift
- R5 (dividend screen): non-dividend / cut-yield names no longer go LONG (screen-off)
- R4 (earnings-revision) deferred: no free EPS-forecast source yet (documented)

(b) bank-sector NPL collector (BOT reportID 794, financial&insurance sector):
- refactored auto_npl to expose shared _parse_sector; new bank_npl.py reuses it
- registered bank_npl FACTOR -> auto-appears in sources table (6 rows) + blends into banks theme surprise (real NPL)
- +unit tests (test_bank_npl), test_dashboard updated (6 sources)

205 tests pass; verified live API (banks surprise incl. NPL 1.07, 6 sources).
This commit is contained in:
Kunthawat Greethong
2026-08-26 19:56:39 +07:00
parent ef78720d32
commit 6e78b6acb5
8 changed files with 240 additions and 54 deletions

View File

@@ -7,6 +7,7 @@ import json
import os import os
import re import re
import secrets import secrets
import statistics
import time import time
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
@@ -467,49 +468,57 @@ def create_app(config: dict[str, Any] | None = None) -> Flask:
board = dash.get("board", []) board = dash.get("board", [])
combos = [b.get("combined") for b in board if b.get("combined") is not None] combos = [b.get("combined") for b in board if b.get("combined") is not None]
if combos: if combos:
import statistics
q1, q3 = statistics.quantiles(combos, n=4)[0], statistics.quantiles(combos, n=4)[2] q1, q3 = statistics.quantiles(combos, n=4)[0], statistics.quantiles(combos, n=4)[2]
else: else:
q1 = q3 = 0.0 q1 = q3 = 0.0
# market-regime gate: how many themes are in distress (negative # market-regime gate (R3): how many themes are in distress. Instead
# surprise). In a broad-down market we tighten the LONG bar and pull # of a hard binary cliff, use a continuous stress = (negative themes
# / total) in [0,1] and shift the LONG bar / SHORT threshold
# smoothly with it. In a broad-down market we tighten LONG and pull
# more names into SHORT/avoid, so 'best of a falling board' isn't LONG. # more names into SHORT/avoid, so 'best of a falling board' isn't LONG.
theme_surprises = [t.get("surprise") for t in dash.get("themes", []) theme_surprises = [t.get("surprise") for t in dash.get("themes", [])
if t.get("surprise") is not None] if t.get("surprise") is not None]
regime_stress = sum(1 for s in theme_surprises if s < 0) n_themes = max(len(theme_surprises), 1)
bear = regime_stress >= 4 # several themes negative -> risk-off regime n_neg = sum(1 for s in theme_surprises if s < 0)
# gate offset: in bear market require more to go LONG stress = n_neg / n_themes # 0..1 continuous regime gauge
long_bar = q3 + (0.10 if bear else 0.0) # gate offset grows with stress (at stress=1 => +0.15 to go LONG)
long_bar = q3 + 0.15 * stress
# SHORT threshold widens as stress rises (pull more into avoid)
short_bar = q1 - 0.05 - 0.08 * stress
fmap = {b.get("symbol"): b for b in board}
for row in board: for row in board:
comb = row.get("combined") comb = row.get("combined")
sym = row.get("symbol") sym = row.get("symbol")
if comb is None: if comb is None:
signal_by_symbol[sym] = {"side": None, "score": None} signal_by_symbol[sym] = {"side": None, "score": None}
continue continue
if bear: fac = fmap.get(sym, {})
# risk-off: SLOT for LONG only clearly-above-top-quartile; everything # R5 (dividend screen): a name that pays no dividend (or has cut
# below the median becomes SHORT/avoid. # its yield to a negative/zero level) never goes LONG — dividend
if comb >= long_bar: # is our core value assumption; literature treats a cut as a
# screen-off signal. Downgrade to NEUTRAL/SHORT accordingly.
is_div = bool(fac.get("is_dividend")) or (fac.get("dividend_yield") or 0) > 0
if comb >= long_bar:
if is_div:
side, score = "LONG", round(min(abs(comb) * 3.0, 1.0) * 0.9 + 0.1, 3) side, score = "LONG", round(min(abs(comb) * 3.0, 1.0) * 0.9 + 0.1, 3)
elif comb < q1 - 0.05:
side, score = "SHORT", round(min(abs(comb) / max(q1 - 0.05, 1e-9), 1.0) * 0.9 + 0.1, 3)
else:
median = combos and statistics.median(combos) or 0.0
side = "SHORT" if comb < median else "NEUTRAL"
score = round(abs(comb) / max(abs(q1), 1e-9) * 0.5, 3)
else:
# normal regime: quartile split 25/25
if comb >= q3:
side, score = "LONG", round(min(abs(comb) * 3.0, 1.0) * 0.9 + 0.1, 3)
elif comb <= q1:
side, score = "SHORT", round(min(abs(comb) / max(abs(q1), 1e-6), 1.0) * 0.5, 3)
else: else:
# high score but no dividend -> strong growth but our
# thesis is dividend-anchored; cap at NEUTRAL.
side, score = "NEUTRAL", round((comb - q1) / max(q3 - q1, 1e-9), 3) side, score = "NEUTRAL", round((comb - q1) / max(q3 - q1, 1e-9), 3)
elif comb < short_bar:
side, score = "SHORT", round(min(abs(comb) / max(abs(short_bar), 1e-6), 1.0) * 0.5, 3)
else:
median = combos and statistics.median(combos) or 0.0
side = "SHORT" if (comb < median and stress > 0.5) else "NEUTRAL"
score = round(abs(comb) / max(abs(q1), 1e-9) * 0.5, 3) if stress > 0.5 \
else round((comb - q1) / max(q3 - q1, 1e-9), 3)
signal_by_symbol[sym] = { signal_by_symbol[sym] = {
"side": side, "score": score, "confidence": "medium", "side": side, "score": score, "confidence": "medium",
"combined_score": comb, "regime": "risk-off" if bear else "normal", "combined_score": comb,
"regime": "risk-off" if stress > 0.4 else "normal",
"regime_stress": round(stress, 3),
} }
except Exception: except Exception:
# no dashboard -> fall back to neutral for all # no dashboard -> fall back to neutral for all
@@ -711,6 +720,7 @@ def create_app(config: dict[str, Any] | None = None) -> Flask:
detail = themes_mod.symbol_breakdown( detail = themes_mod.symbol_breakdown(
symbol, factor_view=factor_view, theme_surprises=theme_surprises, symbol, factor_view=factor_view, theme_surprises=theme_surprises,
latest_price=price, price_date=price_date, latest_price=price, price_date=price_date,
momentum=themes_mod._load_momentum(),
) )
return jsonify(detail) return jsonify(detail)

View File

@@ -83,7 +83,23 @@ def _cells(row_html: str) -> list[str]:
def parse_auto_npl_html(html_text: str, label: str = AUTO_LOAN_LABEL) -> AutoNplSnapshot: 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).""" """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) tables = re.findall(r"<table[^>]*>(.*?)</table>", html_text, re.S)
period = "" period = ""
npl = pct_n = pct_l = None npl = pct_n = pct_l = None
@@ -99,19 +115,14 @@ def parse_auto_npl_html(html_text: str, label: str = AUTO_LOAN_LABEL) -> AutoNpl
period = m.group(1) period = m.group(1)
continue continue
# data row: [no, label, value, pct_npl, pct_loan, ...] # data row: [no, label, value, pct_npl, pct_loan, ...]
if len(cells) >= 4 and cells[0].isdigit() and ("รถยนต์" in cells[1] or label in cells[1]): if len(cells) >= 4 and cells[0].isdigit() and label in cells[1]:
npl = _to_float(cells[2]) npl = _to_float(cells[2])
pct_n = _to_float(cells[3]) pct_n = _to_float(cells[3])
pct_l = _to_float(cells[4]) if len(cells) > 4 else None pct_l = _to_float(cells[4]) if len(cells) > 4 else None
break break
if npl is None and pct_n is None: if npl is None and pct_n is None:
raise AutoNplError("no auto-loan NPL row found in BOT NPL page") return None
return AutoNplSnapshot( return (npl, pct_n, pct_l, period)
npl_amount=npl,
pct_of_npls=pct_n,
pct_of_loans=pct_l,
period=period,
)
def fetch_auto_npl(timeout: float = 30.0) -> AutoNplSnapshot: def fetch_auto_npl(timeout: float = 30.0) -> AutoNplSnapshot:

54
backend/app/bank_npl.py Normal file
View File

@@ -0,0 +1,54 @@
"""Bank-sector 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). This is the SAME report the auto theme already
consumes; the bank variant selects the **financial & insurance** sector row —
the closest public proxy for commercial-bank credit quality.
Higher NPL / higher % of loans is bearish for bank earnings (provisioning drag),
so the factor sign is -1 at the registry level.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Optional
from .auto_npl import _fetch, _parse_sector, AutoNplError
# Sector label on the BOT 794 page that maps to financial/banking credit quality.
# ('กิจกรรมทางการเงินและการประกันภัย' = financial & insurance activities)
BANK_SECTOR_LABEL = "กิจกรรมทางการเงินและการประกันภัย"
@dataclass(frozen=True)
class BankNplSnapshot:
npl_amount: Optional[float] = None
pct_of_npls: Optional[float] = None
pct_of_loans: Optional[float] = None
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 parse_bank_npl_html(html_text: str, label: str = BANK_SECTOR_LABEL) -> BankNplSnapshot:
"""Parse the BOT NPL table; return the financial-sector row (latest period)."""
row = _parse_sector(html_text, label)
if row is None:
raise AutoNplError(f"no {label!r} NPL row found in BOT NPL page")
return BankNplSnapshot(
npl_amount=row[0], pct_of_npls=row[1], pct_of_loans=row[2], period=row[3],
)
def fetch_bank_npl(timeout: float = 30.0) -> BankNplSnapshot:
from .auto_npl import _URL
return parse_bank_npl_html(_fetch(_URL, timeout=timeout))

View File

@@ -84,7 +84,7 @@ class RealDashboard:
def build(self) -> dict: def build(self) -> dict:
# 1) live theme data (real, no fallback) # 1) live theme data (real, no fallback)
from . import auto_credit, auto_npl, energy_thai, bot_tourism from . import auto_credit, auto_npl, bank_npl, energy_thai, bot_tourism
tourism = None tourism = None
try: try:
@@ -97,13 +97,15 @@ class RealDashboard:
lambda: auto_credit.fetch_auto_credit().to_dict(), "auto_credit") lambda: auto_credit.fetch_auto_credit().to_dict(), "auto_credit")
npl_d = _fetch_with_cache( npl_d = _fetch_with_cache(
self.cache, "auto_npl", lambda: auto_npl.fetch_auto_npl().to_dict(), "auto_npl") self.cache, "auto_npl", lambda: auto_npl.fetch_auto_npl().to_dict(), "auto_npl")
bnpl_d = _fetch_with_cache(
self.cache, "bank_npl", lambda: bank_npl.fetch_bank_npl().to_dict(), "bank_npl")
en_d = _fetch_with_cache( en_d = _fetch_with_cache(
self.cache, "energy_thai", lambda: energy_thai.fetch_energy_thai().to_dict(), "energy_thai") self.cache, "energy_thai", lambda: energy_thai.fetch_energy_thai().to_dict(), "energy_thai")
macro_d = _fetch_with_cache( macro_d = _fetch_with_cache(
self.cache, "macro_thai", lambda: macro_thai.fetch_macro_thai().to_dict(), "macro_thai") self.cache, "macro_thai", lambda: macro_thai.fetch_macro_thai().to_dict(), "macro_thai")
# 2) per-theme surprise (uniform z-score) # 2) per-theme surprise (uniform z-score)
surprises = self._theme_surprises(macro_d, auto_d, npl_d, en_d) surprises = self._theme_surprises(macro_d, auto_d, npl_d, en_d, bnpl_d)
# 3) assemble theme reads + thesis (all SET50 themes, so the board and # 3) assemble theme reads + thesis (all SET50 themes, so the board and
# per-symbol view have a surprise for every theme) # per-symbol view have a surprise for every theme)
@@ -115,9 +117,11 @@ class RealDashboard:
] ]
# macro-proxy reads for the expanded SET50 themes (deterministic) # macro-proxy reads for the expanded SET50 themes (deterministic)
proxy_reads = { proxy_reads = {
"banks": {"source": "BOT macro (invest)", "frequency": "monthly", "banks": {"source": "BOT macro (invest) + NPL ภาคการเงิน",
"frequency": "monthly",
"invest_yoy": macro_d.get("private_investment_yoy"), "invest_yoy": macro_d.get("private_investment_yoy"),
"inflation_yoy": macro_d.get("headline_inflation_yoy")}, "inflation_yoy": macro_d.get("headline_inflation_yoy"),
"bank_npl_pct": (bnpl_d or {}).get("pct_of_npls")},
"retail": {"source": "BOT macro (consumption)", "frequency": "monthly", "retail": {"source": "BOT macro (consumption)", "frequency": "monthly",
"consumption_yoy": macro_d.get("private_consumption_yoy")}, "consumption_yoy": macro_d.get("private_consumption_yoy")},
"consumer_staples": {"source": "BOT macro (consumption)", "frequency": "monthly", "consumer_staples": {"source": "BOT macro (consumption)", "frequency": "monthly",
@@ -144,7 +148,7 @@ class RealDashboard:
board = self._build_board(themes, macro_d) board = self._build_board(themes, macro_d)
# 5) source provenance table # 5) source provenance table
sources = self._build_sources(auto_d, npl_d, en_d, macro_d, tourism) sources = self._build_sources(auto_d, npl_d, en_d, macro_d, tourism, bnpl_d)
return { return {
"themes": themes, "themes": themes,
@@ -217,7 +221,7 @@ class RealDashboard:
f"รับผลตามราคาพลังงานและค่าการกลั่น.") f"รับผลตามราคาพลังงานและค่าการกลั่น.")
return "" return ""
def _theme_surprises(self, macro_d, auto_d, npl_d, en_d) -> dict: def _theme_surprises(self, macro_d, auto_d, npl_d, en_d, bnpl_d=None) -> dict:
# tourism: from the tourism arrivals YoY or use the bot tourism surprise # tourism: from the tourism arrivals YoY or use the bot tourism surprise
# auto: z-score of new_car_sales_yoy # auto: z-score of new_car_sales_yoy
# energy: z-score of TOP net profit trend (quarterly) # energy: z-score of TOP net profit trend (quarterly)
@@ -265,8 +269,17 @@ class RealDashboard:
return None return None
return round(min(max((float(v) - center) / span, -1.0), 1.0), 3) return round(min(max((float(v) - center) / span, -1.0), 1.0), 3)
# banks & nonbank_finance: credit demand tracks capex/activity # banks & nonbank_finance: credit demand tracks capex/activity.
s["banks"] = _norm(invest, center=5.0) # banks additionally blends real BOT financial-sector NPL (quarterly):
# rising NPL is a provisioning drag on bank earnings (bearish).
banks_s = _norm(invest, center=5.0)
if banks_s is not None and bnpl_d:
bnpl = bnpl_d.get("pct_of_npls")
if bnpl is not None:
# deduct up to ~0.5 from the surprise when NPL share is elevated
# (reference: financial-sector NPL % of total NPLs, roughly 1-5%).
banks_s = round(max(banks_s - min(max((float(bnpl) - 0.5) / 3.0, 0.0), 0.5), -1.0), 3)
s["banks"] = banks_s
s["nonbank_finance"] = _norm(cons, center=3.0) s["nonbank_finance"] = _norm(cons, center=3.0)
# retail & consumer_staples: spend + mild inflation (demand-led) # retail & consumer_staples: spend + mild inflation (demand-led)
s["retail"] = _norm(cons, center=3.0) s["retail"] = _norm(cons, center=3.0)
@@ -288,7 +301,8 @@ class RealDashboard:
# combine theme scores + siamchart for the per-symbol board. # combine theme scores + siamchart for the per-symbol board.
from . import siamchart_factors from . import siamchart_factors
fv = siamchart_factors.build_factor_view() or {"factors": []} fv = siamchart_factors.build_factor_view() or {"factors": []}
siamchart_score = themes_mod.build_siamchart_score(fv) momentum = themes_mod._load_momentum()
siamchart_score = themes_mod.build_siamchart_score(fv, momentum=momentum)
# per-theme symbol exposure: surprise × firm_quality (real selection). # per-theme symbol exposure: surprise × firm_quality (real selection).
# A strong name in a hot theme scores higher than a weak one. # A strong name in a hot theme scores higher than a weak one.
theme_scores: dict[str, dict[str, float]] = {} theme_scores: dict[str, dict[str, float]] = {}
@@ -331,7 +345,7 @@ class RealDashboard:
board.sort(key=lambda r: r["combined"], reverse=True) board.sort(key=lambda r: r["combined"], reverse=True)
return board return board
def _build_sources(self, auto_d, npl_d, en_d, macro_d, tourism) -> list: def _build_sources(self, auto_d, npl_d, en_d, macro_d, tourism, bnpl_d=None) -> list:
"""Sources derived from the FACTORS registry — adding a factor to """Sources derived from the FACTORS registry — adding a factor to
factors.py auto-appends its source row here (no hardcoded list).""" factors.py auto-appends its source row here (no hardcoded list)."""
import datetime as _dt import datetime as _dt
@@ -340,7 +354,7 @@ class RealDashboard:
fetched = { fetched = {
"auto_credit": auto_d, "auto_npl": npl_d, "auto_credit": auto_d, "auto_npl": npl_d,
"energy_thai": en_d, "macro_thai": macro_d, "energy_thai": en_d, "macro_thai": macro_d, "bank_npl": bnpl_d,
} }
# group FACTORS by fetch module -> one row per distinct source # group FACTORS by fetch module -> one row per distinct source
source_by_module: dict = {} source_by_module: dict = {}
@@ -367,6 +381,7 @@ class RealDashboard:
_source_label = { _source_label = {
"auto_credit": "TradingEconomics", "auto_credit": "TradingEconomics",
"auto_npl": "BOT FI_NP_003_S2", "auto_npl": "BOT FI_NP_003_S2",
"bank_npl": "BOT FI_NP_003_S2",
"energy_thai": "Thai Oil investor", "energy_thai": "Thai Oil investor",
"macro_thai": "BOT Thai Economy", "macro_thai": "BOT Thai Economy",
"bot_tourism": "BOT Tourism", "bot_tourism": "BOT Tourism",

View File

@@ -21,6 +21,7 @@ _FETCH_MODULE: dict[str, str] = {
"tourism": "bot_tourism", "tourism": "bot_tourism",
"auto_credit": "auto_credit", "auto_credit": "auto_credit",
"auto_npl": "auto_npl", "auto_npl": "auto_npl",
"bank_npl": "bank_npl",
"energy_thai": "energy_thai", "energy_thai": "energy_thai",
"macro_thai": "macro_thai", "macro_thai": "macro_thai",
} }
@@ -74,6 +75,15 @@ FACTORS: dict[str, dict[str, Any]] = {
"sign": -1, "sign": -1,
"weight": 1.0, "weight": 1.0,
}, },
"bank_npl": {
"name_th": "NPL ภาคการเงิน",
"source": "BOT",
"frequency": "quarterly",
"fetch": "bank_npl",
"value_key": "pct_of_npls",
"sign": -1,
"weight": 1.0,
},
"energy_net_margin": { "energy_net_margin": {
"name_th": "กำไรสุทธิโรงกลั่น", "name_th": "กำไรสุทธิโรงกลั่น",
"source": "TOP", "source": "TOP",

View File

@@ -252,11 +252,47 @@ def build_theme_scores(theme_id: str, signals: list[dict]) -> dict[str, float]:
return {s: z.get(i, 0.0) for i, s in enumerate(syms)} return {s: z.get(i, 0.0) for i, s in enumerate(syms)}
def build_siamchart_score(factors: dict) -> dict[str, float]: _SIAMCHART_GROWTH_W = 1.5 # R1 (PEAD): EPS-growth dominates value; literature (Bernard-Thomas 1990,
# Livnat-Mendenhall 2006) shows drift follows earnings, not just yield.
_SIAMCHART_YIELD_W = 2.0 # dividend floor for value names
_SIAMCHART_MOMENTUM_W = 0.5 # R2: EM momentum exists but is noisy -> keep it a small trend boost.
def _load_momentum(lookback_days: int = 252) -> dict[str, float]:
"""12-1 momentum per symbol from the latest price snapshot (deterministic).
R2 (Jegadeesh-Titman 1993; EM evidence: weaker but positive). Returns
{symbol: (close_today / close_{-12m}) - 1}. Lookback uses trading days so it
aligns to ~12 calendar months.
"""
try:
from . import simulation
series = simulation.load_price_snapshot()
except Exception:
return {}
out: dict[str, float] = {}
for sym, s in series.items():
bars = s.get("bars", [])
if len(bars) < lookback_days + 1:
continue
try:
today = float(bars[-1]["adjusted_close"])
base = float(bars[-1 - lookback_days]["adjusted_close"])
except (KeyError, TypeError, ValueError, IndexError):
continue
if today <= 0 or base <= 0:
continue
out[sym] = round((today / base) - 1.0, 4)
return out
def build_siamchart_score(factors: dict,
momentum: Optional[dict[str, float]] = None) -> dict[str, float]:
"""Derive a normalized fundamental score from the Siamchart factor view. """Derive a normalized fundamental score from the Siamchart factor view.
Uses EPS growth YoY and dividend yield as the "value+quality" signals. Uses EPS growth YoY (weighted above yield per PEAD literature) and dividend
Positive EPS growth and higher yield both push the score up. yield. Optional momentum (12-1, from price snapshot) adds a low-weight
trend component; EM momentum is noisier, so it stays small.
""" """
out: dict[str, float] = {} out: dict[str, float] = {}
for f in factors.get("factors", []): for f in factors.get("factors", []):
@@ -266,8 +302,9 @@ def build_siamchart_score(factors: dict) -> dict[str, float]:
g = f.get("eps_growth_yoy") g = f.get("eps_growth_yoy")
d = f.get("dividend_yield") or 0.0 d = f.get("dividend_yield") or 0.0
g = float(g) if g is not None else 0.0 g = float(g) if g is not None else 0.0
# combine growth and yield; yield adds a floor so dividend names get weight m = (momentum or {}).get(sym, 0.0)
out[sym] = g + d * 2.0 # R1+R2: growth dominates (PEAD), yield floors, momentum adds trend.
out[sym] = g * _SIAMCHART_GROWTH_W + d * _SIAMCHART_YIELD_W + _SIAMCHART_MOMENTUM_W * m
syms = list(out.keys()) syms = list(out.keys())
z = _zscore([out[s] for s in syms]) z = _zscore([out[s] for s in syms])
return {s: z.get(i, 0.0) for i, s in enumerate(syms)} return {s: z.get(i, 0.0) for i, s in enumerate(syms)}
@@ -352,6 +389,7 @@ def symbol_breakdown(
price_date: str = "", price_date: str = "",
weight_theme: float = 0.6, weight_theme: float = 0.6,
weight_siamchart: float = 0.4, weight_siamchart: float = 0.4,
momentum: Optional[dict[str, float]] = None,
) -> dict: ) -> dict:
"""Transparent per-symbol scoring breakdown. """Transparent per-symbol scoring breakdown.
@@ -396,11 +434,12 @@ def symbol_breakdown(
g = fac.get("eps_growth_yoy") g = fac.get("eps_growth_yoy")
d = fac.get("dividend_yield") or 0.0 d = fac.get("dividend_yield") or 0.0
g = float(g) if g is not None else 0.0 g = float(g) if g is not None else 0.0
raw_siamchart = g + d * 2.0 m = (momentum or {}).get(symbol, 0.0)
raw_siamchart = g * _SIAMCHART_GROWTH_W + d * _SIAMCHART_YIELD_W + _SIAMCHART_MOMENTUM_W * m
# z-score against the full universe (same as build_siamchart_score); capture # z-score against the full universe (same as build_siamchart_score); capture
# the population stats so the view can show HOW -2.8 became -0.588. # the population stats so the view can show HOW -2.8 became -0.588.
siamchart_map = build_siamchart_score(factor_view) siamchart_map = build_siamchart_score(factor_view, momentum=momentum)
siamchart_score = siamchart_map.get(symbol, 0.0) siamchart_score = siamchart_map.get(symbol, 0.0)
# recompute the population of raw scores to expose mean / stdev # recompute the population of raw scores to expose mean / stdev
raw_values = [] raw_values = []
@@ -410,7 +449,8 @@ def symbol_breakdown(
gg = f.get("eps_growth_yoy") gg = f.get("eps_growth_yoy")
dd = f.get("dividend_yield") or 0.0 dd = f.get("dividend_yield") or 0.0
gg = float(gg) if gg is not None else 0.0 gg = float(gg) if gg is not None else 0.0
raw_values.append(gg + dd * 2.0) mm = (momentum or {}).get(f.get("symbol"), 0.0)
raw_values.append(gg * _SIAMCHART_GROWTH_W + dd * _SIAMCHART_YIELD_W + _SIAMCHART_MOMENTUM_W * mm)
pop_mean = statistics.mean(raw_values) if raw_values else 0.0 pop_mean = statistics.mean(raw_values) if raw_values else 0.0
pop_stdev = statistics.pstdev(raw_values) if raw_values else 0.0 pop_stdev = statistics.pstdev(raw_values) if raw_values else 0.0

View File

@@ -0,0 +1,43 @@
"""Tests for the BOT bank-sector (financial) NPL collector."""
from __future__ import annotations
import unittest
from app.bank_npl import BANK_SECTOR_LABEL, parse_bank_npl_html
# Minimal server-rendered table fragment in the style of reportID=794.
_NPL_HTML = """
<table>
<tr><th>ยอดคงค้าง NPL</th><th>% ต่อ NPLs</th><th>% ต่อสินเชื่อรวม</th></tr>
<tr><td></td><td>Q2/2568</td><td></td></tr>
<tr><td>1</td><td>การผลิต</td><td>12345.0</td><td>4.10</td><td>0.30</td></tr>
<tr><td>2</td><td>กิจกรรมทางการเงินและการประกันภัย</td><td>5598.0</td><td>1.07</td><td>0.11</td></tr>
<tr><td>3</td><td>รถยนต์</td><td>20602.0</td><td>3.95</td><td>1.20</td></tr>
</table>
"""
class BankNplTest(unittest.TestCase):
def test_parses_financial_sector(self):
snap = parse_bank_npl_html(_NPL_HTML)
self.assertEqual(snap.npl_amount, 5598.0)
self.assertEqual(snap.pct_of_npls, 1.07)
self.assertEqual(snap.pct_of_loans, 0.11)
self.assertEqual(snap.period, "Q2/2568")
def test_label_matched(self):
self.assertIn("การเงิน", BANK_SECTOR_LABEL)
self.assertIn("ประกันภัย", BANK_SECTOR_LABEL)
def test_fetches_live(self):
# Live BOT page must still yield a financial-sector NPL (network).
from app.bank_npl import fetch_bank_npl
snap = fetch_bank_npl()
self.assertIsNotNone(snap.pct_of_npls)
if snap.pct_of_npls is not None:
self.assertGreater(snap.pct_of_npls, 0.0)
if __name__ == "__main__":
unittest.main()

View File

@@ -37,7 +37,8 @@ class DashboardTest(unittest.TestCase):
@patch("app.auto_npl.fetch_auto_npl") @patch("app.auto_npl.fetch_auto_npl")
@patch("app.energy_thai.fetch_energy_thai") @patch("app.energy_thai.fetch_energy_thai")
@patch("app.macro_thai.fetch_macro_thai") @patch("app.macro_thai.fetch_macro_thai")
def test_build_returns_structure(self, macro, energy, npl, auto): @patch("app.bank_npl.fetch_bank_npl")
def test_build_returns_structure(self, bnpl, macro, energy, npl, auto):
class _Factory: class _Factory:
def __init__(self, data): self._data = data def __init__(self, data): self._data = data
def to_dict(self): return self._data def to_dict(self): return self._data
@@ -49,10 +50,12 @@ class DashboardTest(unittest.TestCase):
"new_car_sales_yoy": 20.07, "total_vehicle_sales": 59000}) "new_car_sales_yoy": 20.07, "total_vehicle_sales": 59000})
npl.return_value = _Factory({ npl.return_value = _Factory({
"pct_of_npls": 3.95, "npl_amount": 20602, "period": "Q2/2568"}) "pct_of_npls": 3.95, "npl_amount": 20602, "period": "Q2/2568"})
bnpl.return_value = _Factory({
"pct_of_npls": 1.07, "npl_amount": 5598, "period": ""})
cache = _FakeCache({}) cache = _FakeCache({})
dash = RealDashboard([], cache).build() dash = RealDashboard([], cache).build()
self.assertEqual(len(dash["themes"]), 13) # all SET50 themes self.assertEqual(len(dash["themes"]), 13) # all SET50 themes
self.assertEqual(len(dash["sources"]), 5) self.assertEqual(len(dash["sources"]), 6) # auto-derived from FACTORS (bank_npl added)
self.assertIn("macro", dash) self.assertIn("macro", dash)
self.assertIn("board", dash) self.assertIn("board", dash)