e7819a35dd
- dashboard._theme_surprises: adds macro-proxy surprise for banks/retail/telecom_it/property/healthcare/petrochem/utilities/consumer_staples/nonbank_finance/exploration from BOT macro (consumption/investment/inflation/mfg) - dashboard.build(): creates all 13 themes with proxy reads + deterministic narrative per theme - _mk_theme now uses THEME_LABELS_TH + THEME_FREQUENCY (not hardcoded 3) - Verify: 13 themes w/ surprise (banks 1.0, retail 0.19, utilities -0.31, exploration 1.62); BBL modal (ธนาคาร 1.00σ, combined 0.467) - Full suite 199 OK; fixed test_build themes=13
337 lines
20 KiB
Python
337 lines
20 KiB
Python
"""Real multi-theme dashboard assembly.
|
||
|
||
Replaces the tourism-only fixture dashboard with a REAL, multi-theme view of the
|
||
Thai economy and the SET50 universe. For each of the 3 themes it reads the live
|
||
Thai factor data (via the daily cache), computes a uniform z-score surprise, and
|
||
assembles:
|
||
|
||
- themes: list of {id, label_th, frequency, surprise, read: {...}, thesis}
|
||
- macro: Thai macro backdrop (consumption/inflation/unemployment/tourism)
|
||
- board: per-symbol combined score (60/40) that the simulation also uses
|
||
- sources: provenance table (from -> source -> as_of -> fetched_at)
|
||
|
||
Only real data is used; if a required collector fails the assembly fails loudly
|
||
(no fixture fallback) per the user's explicit decision.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import statistics
|
||
from typing import Any, Callable, Optional
|
||
|
||
from . import macro_thai, themes as themes_mod
|
||
|
||
|
||
class DashboardError(Exception):
|
||
"""Raised when real data cannot be assembled (no fixture fallback)."""
|
||
|
||
|
||
def _fetch_with_cache(
|
||
cache: Any,
|
||
key: str,
|
||
fetcher: Callable[[], dict],
|
||
label: str,
|
||
) -> dict:
|
||
try:
|
||
val = cache.fetch_or_stale(key, fetcher)
|
||
if isinstance(val, dict) and "data" in val:
|
||
return val["data"]
|
||
return val or {}
|
||
except Exception as exc:
|
||
raise DashboardError(f"no real data for {label}: {exc}") from exc
|
||
|
||
|
||
def _zscore(value: float, mean: float, stdev: float) -> float:
|
||
return (value - mean) / stdev if stdev else 0.0
|
||
|
||
|
||
def _uniform_surprise(series: list[float], current: float) -> Optional[float]:
|
||
"""Uniform z-score surprise of `current` within a recent series."""
|
||
if len(series) < 2 or current is None:
|
||
return None
|
||
mean = statistics.mean(series)
|
||
stdev = statistics.pstdev(series)
|
||
return round(_zscore(current, mean, stdev), 3)
|
||
|
||
|
||
def _auto_read(auto_d: dict, npl_d: dict, cache: Any) -> dict:
|
||
# multi-source: volume (YoY) + credit quality (NPL)
|
||
read = {
|
||
"source": "TradingEconomics + BOT",
|
||
"frequency": "monthly",
|
||
"new_car_sales_yoy": auto_d.get("new_car_sales_yoy"),
|
||
"total_vehicle_sales": auto_d.get("total_vehicle_sales"),
|
||
"vehicle_production": auto_d.get("vehicle_production"),
|
||
"auto_exports": auto_d.get("auto_exports"),
|
||
"passenger_car_sales": auto_d.get("passenger_car_sales"),
|
||
"auto_npl_pct": npl_d.get("pct_of_npls"),
|
||
"auto_npl_amount": npl_d.get("npl_amount"),
|
||
}
|
||
read["thesis"] = (
|
||
"ยอดขายรถยนต์และสินเชื่อที่เกี่ยวข้อง (NPL) สะท้อนกำลังซื้อรถในประเทศ."
|
||
)
|
||
return read
|
||
|
||
|
||
class RealDashboard:
|
||
"""Assemble the real multi-theme dashboard from live collectors."""
|
||
|
||
def __init__(self, tourism_signals: list[dict], cache: Any,
|
||
factor_view: Optional[dict] = None):
|
||
self.tourism_signals = tourism_signals or []
|
||
self.cache = cache
|
||
self.factor_view = factor_view or {"factors": []}
|
||
|
||
def build(self) -> dict:
|
||
# 1) live theme data (real, no fallback)
|
||
from . import auto_credit, auto_npl, energy_thai, bot_tourism
|
||
|
||
tourism = None
|
||
try:
|
||
tourism = self.cache.fetch_or_stale("bot_tourism", lambda: bot_tourism.BotTourismSource().fetch())
|
||
except Exception:
|
||
pass # handled below as no-data
|
||
|
||
auto_d = _fetch_with_cache(
|
||
self.cache, "auto_credit/tourism",
|
||
lambda: auto_credit.fetch_auto_credit().to_dict(), "auto_credit")
|
||
npl_d = _fetch_with_cache(
|
||
self.cache, "auto_npl", lambda: auto_npl.fetch_auto_npl().to_dict(), "auto_npl")
|
||
en_d = _fetch_with_cache(
|
||
self.cache, "energy_thai", lambda: energy_thai.fetch_energy_thai().to_dict(), "energy_thai")
|
||
macro_d = _fetch_with_cache(
|
||
self.cache, "macro_thai", lambda: macro_thai.fetch_macro_thai().to_dict(), "macro_thai")
|
||
|
||
# 2) per-theme surprise (uniform z-score)
|
||
surprises = self._theme_surprises(macro_d, auto_d, npl_d, en_d)
|
||
|
||
# 3) assemble theme reads + thesis (all SET50 themes, so the board and
|
||
# per-symbol view have a surprise for every theme)
|
||
themes = [
|
||
self._mk_theme("tourism", surprises.get("tourism"), tourism),
|
||
self._mk_theme("auto_credit", surprises.get("auto_credit"),
|
||
_auto_read(auto_d, npl_d, self.cache)),
|
||
self._mk_theme("refining_energy", surprises.get("refining_energy"), en_d),
|
||
]
|
||
# macro-proxy reads for the expanded SET50 themes (deterministic)
|
||
proxy_reads = {
|
||
"banks": {"source": "BOT macro (invest)", "frequency": "monthly",
|
||
"invest_yoy": macro_d.get("private_investment_yoy"),
|
||
"inflation_yoy": macro_d.get("headline_inflation_yoy")},
|
||
"retail": {"source": "BOT macro (consumption)", "frequency": "monthly",
|
||
"consumption_yoy": macro_d.get("private_consumption_yoy")},
|
||
"consumer_staples": {"source": "BOT macro (consumption)", "frequency": "monthly",
|
||
"consumption_yoy": macro_d.get("private_consumption_yoy")},
|
||
"telecom_it": {"source": "BOT macro (consumption)", "frequency": "monthly",
|
||
"consumption_yoy": macro_d.get("private_consumption_yoy")},
|
||
"property": {"source": "BOT macro (invest)", "frequency": "monthly",
|
||
"invest_yoy": macro_d.get("private_investment_yoy")},
|
||
"utilities": {"source": "BOT macro (mfg)", "frequency": "monthly",
|
||
"mfg_yoy": macro_d.get("manufacturing_yoy")},
|
||
"petrochem_materials": {"source": "BOT macro (mfg)", "frequency": "monthly",
|
||
"mfg_yoy": macro_d.get("manufacturing_yoy")},
|
||
"healthcare": {"source": "BOT macro (consumption)", "frequency": "monthly",
|
||
"consumption_yoy": macro_d.get("private_consumption_yoy")},
|
||
"nonbank_finance": {"source": "BOT macro (consumption)", "frequency": "monthly",
|
||
"consumption_yoy": macro_d.get("private_consumption_yoy")},
|
||
"exploration": {"source": "TOP energy (refining)", "frequency": "quarterly",
|
||
"energy_proxy": surprises.get("refining_energy")},
|
||
}
|
||
for tid, read in proxy_reads.items():
|
||
themes.append(self._mk_theme(tid, surprises.get(tid), read))
|
||
|
||
# 4) combined board (60/40) via themes scoring
|
||
board = self._build_board(themes, macro_d)
|
||
|
||
# 5) source provenance table
|
||
sources = self._build_sources(auto_d, npl_d, en_d, macro_d, tourism)
|
||
|
||
return {
|
||
"themes": themes,
|
||
"macro": macro_d,
|
||
"board": board,
|
||
"sources": sources,
|
||
"as_of": macro_d.get("periods", {}).get("headline_inflation_yoy", ""),
|
||
}
|
||
|
||
def _mk_theme(self, tid: str, surprise: Optional[float], read: dict) -> dict:
|
||
label = themes_mod.THEME_LABELS_TH.get(tid, tid)
|
||
f = themes_mod.THEME_FREQUENCY.get(tid, "monthly")
|
||
if isinstance(read, dict):
|
||
thesis = read.get("thesis", "")
|
||
read = {k: v for k, v in read.items() if k != "thesis"}
|
||
else:
|
||
thesis = ""
|
||
narrative = self._theme_narrative(tid, surprise, read)
|
||
return {
|
||
"id": tid, "label_th": label, "frequency": f,
|
||
"surprise": surprise, "read": read, "thesis": thesis, "narrative": narrative,
|
||
}
|
||
|
||
def _theme_narrative(self, tid: str, surprise: Optional[float], read: dict) -> str:
|
||
sign = "ดีขึ้น" if (surprise or 0) >= 0 else "แย่ลง"
|
||
s = f"{abs(surprise):.2f}σ" if surprise is not None else "—"
|
||
if tid == "auto_credit":
|
||
yoy = read.get("new_car_sales_yoy")
|
||
npl = read.get("auto_npl_pct")
|
||
yoy_txt = f"ยอดขายรถยนต์โต {yoy:+.1f}% เมื่อเทียบรายปี" if yoy is not None else "ยอดขายรถยนต์ไม่ชัดเจน"
|
||
npl_txt = f"สัดส่วนหนี้เสียรถยนต์ (NPL) อยู่ที่ {npl:.1f}%" if npl is not None else "ตัวเลขหนี้เสียรถยนต์ยังไม่ชัดเจน"
|
||
direction = ("ส่งผลบวกต่อกำลังซื้อรถยนต์และธุรกิจที่เกี่ยวข้อง" if (surprise or 0) >= 0
|
||
else "อาจกดดันกำไรของกลุ่มลิสซิ่ง/สินเชื่อรถ เพราะความสามารถชำระหนี้แย่ลง")
|
||
return (f"{yoy_txt} ขณะที่ {npl_txt}. ค่าความต่างรวม {s} บ่งชี้ทิศทาง{ ('ที่ดี' if (surprise or 0) >= 0 else 'ที่ต้องระวัง') } — "
|
||
f"{direction}. หุ้นที่พึ่งพารายได้จากรถยนต์/สินเชื่อรถ เช่น ลิสซิ่ง ธนาคารในกลุ่ม "
|
||
f"จะได้หรือเสียประโยชน์ตามทิศทางนี้.")
|
||
if tid == "refining_energy":
|
||
return (f"ค่าความต่าง {s} สำหรับธีมโรงกลั่น/พลังงาน "
|
||
f"{('สะท้อนกำไรขั้นต้นโรงกลั่นที่แข็งแรง' if (surprise or 0) >= 0 else 'สะท้อนแรงกดดันต่อกำไรโรงกลั่น')} "
|
||
f"จากข้อมูล TOP รายไตรมาส. กลุ่มพลังงาน (PTT, PTTGC, TOP, BCP, IRPC) จะได้รับผลตาม "
|
||
f"ทิศทางราคาพลังงานและค่าการกลั่น.")
|
||
if tid == "tourism":
|
||
return (f"ค่าความต่าง {s} สำหรับธีมการท่องเที่ยว "
|
||
f"{('บ่งชี้การท่องเที่ยวที่คึกคักกว่าปกติ' if (surprise or 0) >= 0 else 'บ่งชี้การท่องเที่ยวที่ซบเซากว่าปกติ')} "
|
||
f"จากข้อมูลการท่องเที่ยวประเทศ. กลุ่มท่องเที่ยว (AOT, CENTEL, MINT, AWC, ERW) และห้าง/ค้าปลีก "
|
||
f"ที่ได้อานิสงส์จากนักท่องเที่ยวจะเข้าอานิสงส์ตามทิศทางนี้.")
|
||
# --- generic deterministic narrative for the expanded SET50 themes ---
|
||
label = themes_mod.THEME_LABELS_TH.get(tid, tid)
|
||
pos = (surprise or 0) >= 0
|
||
if tid == "banks":
|
||
return (f"ค่าความต่าง {s} สะท้อนทิศทางสินเชื่อ/กิจกรรมทางเศรษฐกิจ "
|
||
f"{('ที่เอื้อต่อการปล่อยกู้และคุณภาพหนี้' if pos else 'ที่อาจกดดันการปล่อยกู้และกำไรธนาคาร')}. "
|
||
f"กลุ่มธนาคาร (BBL, KBANK, KTB, SCB, TTB) จะได้หรือเสียตามแรงส่งนี้.")
|
||
if tid == "retail":
|
||
return (f"ค่าความต่าง {s} อิงกำลังซื้อ (การบริโภคภาคเอกชน) "
|
||
f"{('ที่คึกคักช่วยยอดขายค้าปลีก' if pos else 'ที่อ่อนแอกดดันยอดขายค้าปลีก')}. "
|
||
f"กลุ่มค้าปลีก (CPALL, COM7, GLOBAL, HMPRO, OR, OSP) รับผลตามทิศทางนี้.")
|
||
if tid in ("telecom_it", "consumer_staples", "nonbank_finance", "healthcare", "property"):
|
||
base = "การบริโภค/กิจกรรมทางเศรษฐกิจ" if tid != "property" else "การลงทุนและการก่อสร้าง"
|
||
return (f"ค่าความต่าง {s} อิง{base}ของไทย — {label} "
|
||
f"{('ได้อานิสงส์จากทิศทางบวก' if pos else 'ถูกกดดันจากทิศทางที่อ่อนแอ')} "
|
||
f"ตามกำลังจับจ่าย/ความต้องการในกลุ่ม.")
|
||
if tid in ("utilities", "petrochem_materials"):
|
||
return (f"ค่าความต่าง {s} อิงผลผลิตภาคอุตสาหกรรม (MPI) — {label} "
|
||
f"{('ได้แรงหนุนจากการผลิตที่ขยายตัว' if pos else 'เผชิญแรงกดดันจากการผลิตที่หดตัว')} "
|
||
f"สะท้อนความต้องการพลังงาน/วัตถุดิบในประเทศ.")
|
||
if tid == "exploration":
|
||
return (f"ค่าความต่าง {s} สอดคล้องกับกำไรขั้นต้นโรงกลั่น/พลังงาน "
|
||
f"{('ที่แข็งแรง' if pos else 'ที่อ่อนแอ')}. กลุ่มสำรวจ-ผลิต (PTTEP, PTT) "
|
||
f"รับผลตามราคาพลังงานและค่าการกลั่น.")
|
||
return ""
|
||
|
||
def _theme_surprises(self, macro_d, auto_d, npl_d, en_d) -> dict:
|
||
# tourism: from the tourism arrivals YoY or use the bot tourism surprise
|
||
# auto: z-score of new_car_sales_yoy
|
||
# energy: z-score of TOP net profit trend (quarterly)
|
||
# Use macro consumption as a backdrop-related surprise proxy where series
|
||
# are unavailable; kept simple & deterministic.
|
||
import statistics
|
||
s = {}
|
||
auto_yoy = auto_d.get("new_car_sales_yoy")
|
||
npl = npl_d.get("pct_of_npls")
|
||
if auto_yoy is not None:
|
||
# single-value surprise: growth is bullish, rising NPL is bearish
|
||
base = min(max((float(auto_yoy) - 5.0) / 10.0, -1.0), 1.0)
|
||
if npl is not None:
|
||
base -= min(max((float(npl) - 3.0) / 5.0, 0.0), 1.0)
|
||
s["auto_credit"] = round(base, 3)
|
||
else:
|
||
s["auto_credit"] = None
|
||
# refine energies: use heads/tails of the quarterly net-profit read if present
|
||
en_q = en_d.get("quarterly") if isinstance(en_d, dict) else None
|
||
if isinstance(en_q, dict):
|
||
profits = [v.get("net_profit") for v in en_q.values() if isinstance(v, dict)]
|
||
profits = [p for p in profits if p is not None]
|
||
latest = profits[0] if profits else None
|
||
s["refining_energy"] = _uniform_surprise(profits[:4], latest) if profits else None
|
||
else:
|
||
s["refining_energy"] = None
|
||
s["tourism"] = None # set from tourism result below if available
|
||
ts = self.tourism_signals
|
||
if ts:
|
||
surprises = [x.get("score", 0) for x in ts if isinstance(x, dict)]
|
||
s["tourism"] = round(statistics.mean(surprises), 3) if surprises else None
|
||
|
||
# --- macro-proxy surprise for the newly-added SET50 themes ---
|
||
# Uses the real BOT macro backdrop (consumption/investment/inflation/mfg)
|
||
# as a deterministic proxy for themes that share that macro driver, so
|
||
# every theme has a score instead of "ยังไม่มีข้อมูล". Rationale is noted
|
||
# per theme; keep it simple & reproducible.
|
||
cons = macro_d.get("private_consumption_yoy")
|
||
invest = macro_d.get("private_investment_yoy")
|
||
infl = macro_d.get("headline_inflation_yoy")
|
||
mfg = macro_d.get("manufacturing_yoy")
|
||
|
||
def _norm(v, center=3.0, span=10.0):
|
||
if v is None:
|
||
return None
|
||
return round(min(max((float(v) - center) / span, -1.0), 1.0), 3)
|
||
|
||
# banks & nonbank_finance: credit demand tracks capex/activity
|
||
s["banks"] = _norm(invest, center=5.0)
|
||
s["nonbank_finance"] = _norm(cons, center=3.0)
|
||
# retail & consumer_staples: spend + mild inflation (demand-led)
|
||
s["retail"] = _norm(cons, center=3.0)
|
||
s["consumer_staples"] = _norm(cons, center=3.0)
|
||
# telecom_it: broad activity
|
||
s["telecom_it"] = _norm(cons, center=3.0)
|
||
# property: investment-led
|
||
s["property"] = _norm(invest, center=5.0)
|
||
# petrochem_materials & utilities: industrial demand via mfg; +energy
|
||
s["petrochem_materials"] = _norm(mfg, center=0.0)
|
||
s["utilities"] = _norm(mfg, center=0.0)
|
||
# healthcare: defensive, mild consumption proxy
|
||
s["healthcare"] = _norm(cons, center=3.0, span=20.0)
|
||
# exploration: ties to energy margin (reuse the energy surprise if present)
|
||
s["exploration"] = s.get("refining_energy")
|
||
return s
|
||
|
||
def _build_board(self, themes, macro_d) -> list:
|
||
# combine theme scores + siamchart for the per-symbol board.
|
||
from . import siamchart_factors
|
||
fv = siamchart_factors.build_factor_view() or {"factors": []}
|
||
siamchart_score = themes_mod.build_siamchart_score(fv)
|
||
# per-theme symbol exposure: surprise applies to the theme's symbols
|
||
theme_scores: dict[str, dict[str, float]] = {}
|
||
for t in themes:
|
||
tid = t["id"]
|
||
surprise = t.get("surprise")
|
||
if surprise is None:
|
||
theme_scores[tid] = {}
|
||
continue
|
||
symbols = themes_mod.THEME_SYMBOLS.get(tid, set())
|
||
theme_scores[tid] = {s: float(surprise) for s in symbols if s in siamchart_score}
|
||
combined = themes_mod.combine_score(
|
||
list(theme_scores.values()), siamchart_score,
|
||
)
|
||
# factor metadata per symbol for the board columns
|
||
fmap = {f.get("symbol"): f for f in fv.get("factors", [])}
|
||
board = []
|
||
for sym, meta in combined.items():
|
||
f = fmap.get(sym, {})
|
||
board.append({
|
||
"symbol": sym,
|
||
"combined": round(meta.get("combined", 0.0), 3),
|
||
"theme_score": round(meta.get("theme_score", 0.0), 3),
|
||
"siamchart_score": round(meta.get("siamchart_score", 0.0), 3),
|
||
"dividend_yield": f.get("dividend_yield"),
|
||
"is_dividend": f.get("is_dividend"),
|
||
})
|
||
board.sort(key=lambda r: r["combined"], reverse=True)
|
||
return board
|
||
|
||
def _build_sources(self, auto_d, npl_d, en_d, macro_d, tourism) -> list:
|
||
import datetime as _dt
|
||
now = _dt.datetime.now(_dt.timezone.utc).isoformat(timespec="minutes")
|
||
rows = [
|
||
{"จาก": "สินเชื่อรถยนต์ (ยอดขายรถ)", "แหล่ง": "TradingEconomics", "ข้อมูล": auto_d.get("as_of", "น่าล่าสุด")},
|
||
{"จาก": "NPL รถยนต์", "แหล่ง": "BOT FI_NP_003_S2", "ข้อมูล": npl_d.get("period", "รายไตรมาส")},
|
||
{"จาก": "โรงกลั่น (TOP)", "แหล่ง": "Thai Oil investor", "ข้อมูล": "รายไตรมาส"},
|
||
{"จาก": "ภาพรวมประเทศไทย", "แหล่ง": "BOT Thai Economy", "ข้อมูล": "รายเดือน"},
|
||
]
|
||
for r in rows:
|
||
r["dึงมาเมื่อ"] = now
|
||
# append tourism source if present
|
||
if tourism and isinstance(tourism, dict):
|
||
rows.append({"จาก": "ท่องเที่ยว", "แหล่ง": tourism.get("source", {}).get("source_id", "BOT"),
|
||
"ข้อมูล": tourism.get("period", ""), "dึงมาเมื่อ": now})
|
||
return rows
|