From affc29a3bedc5d97a8a4b4d977a6cb2b70a46cc4 Mon Sep 17 00:00:00 2001 From: Kunthawat Greethong Date: Tue, 25 Aug 2026 15:19:39 +0700 Subject: [PATCH] [verified] Add multi-theme registry + combined scoring engine (60% theme / 40% Siamchart) - themes.py: 3-theme registry (tourism monthly, auto_credit monthly, refining_energy quarterly) with Thai labels + frequency; curated SET50 symbol->theme exposure map; z-normalized theme scoring and Siamchart fundamental score; 60/40 combined score (multi-theme mean) - Frequency recorded per theme so consumers don't mix different-cadence factors as same-timestamp - 8 tests; full suite pass --- backend/app/themes.py | 158 +++++++++++++++++++++++++++++++++++ backend/tests/test_themes.py | 72 ++++++++++++++++ 2 files changed, 230 insertions(+) create mode 100644 backend/app/themes.py create mode 100644 backend/tests/test_themes.py diff --git a/backend/app/themes.py b/backend/app/themes.py new file mode 100644 index 0000000..eeb0fe7 --- /dev/null +++ b/backend/app/themes.py @@ -0,0 +1,158 @@ +"""Multi-theme registry, symbol->theme exposure mapping, and combined scoring. + +Aggregates the three Thai alternative-factor themes (tourism, auto_credit, +refining_energy) plus the Siamchart fundamental provider into a per-symbol +combined score, per the user's confirmed weighting: + + combined = 0.6 * theme_score + 0.4 * siamchart_score + +where `theme_score` is the mean of the enabled theme scores that cover a symbol +(so a symbol in several themes averages its theme scores), and `siamchart_score` +is a normalized fundamental score. + +Frequency handling: theme scores carry an explicit `frequency` (daily/monthly/ +quarterly/annual). Consumers must not mix different-frequency factors as if they +were the same timestamp — see `frequency` on each scored theme. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Optional + +# --------------------------------------------------------------------------- +# Theme registry + symbol->theme exposure mapping +# --------------------------------------------------------------------------- + +# Symbols known to belong to each theme. This is a *curated* partial mapping of +# the SET50 universe; symbols not listed here get no direct theme exposure for +# that theme (theme_score contribution = those themes' factor value applied via +# a generic market exposure fallback, see scoring). +THEME_SYMBOLS: dict[str, set[str]] = { + "tourism": { + "AOT", "CENTEL", "MINT", "ERW", "DHOUSE", "AWC", "SNNP", "CPN", "CRC", + "MAJOR", "BEM", "BTS", + }, + "auto_credit": { + "KKP", "TISCO", "TCAP", "THANI", "MTC", "SAWAD", "NSI", "GLAND", + "THG", "ASI", "TGPRO", "AEONTS", "TK", + }, + "refining_energy": { + "PTT", "PTTGC", "TOP", "IRPC", "SPRC", "BCP", "ESSO", "BANPU", "GPSC", + }, +} + +# Theme frequency (data cadence of the underlying factor). Used to keep +# multi-frequency factors from being mixed as same-timestamp. +THEME_FREQUENCY: dict[str, str] = { + "tourism": "monthly", + "auto_credit": "monthly", + "refining_energy": "quarterly", +} + + +@dataclass +class Theme: + id: str + label_en: str + label_th: str + frequency: str + source: str + enabled: bool = True + + +def list_themes() -> list[Theme]: + return [ + Theme("tourism", "Tourism Pulse", "การท่องเที่ยว", "monthly", "BOT tourism", enabled=True), + Theme("auto_credit", "Auto Credit Cycle", "สินเชื่อรถยนต์", "monthly", "TradingEconomics car sales", enabled=True), + Theme("refining_energy", "Refining / Energy", "โรงกลั่น/พลังงาน", "quarterly", "Thai Oil (TOP) financials", enabled=True), + ] + + +# --------------------------------------------------------------------------- +# Scoring helpers +# --------------------------------------------------------------------------- +def _zscore(values: list) -> dict: + """Normalize a list of floats to z-scores (index -> z).""" + n = len(values) + if n == 0: + return {} + mean = sum(values) / n + var = sum((v - mean) ** 2 for v in values) / n + std = var ** 0.5 or 1.0 + return {i: (v - mean) / std for i, v in enumerate(values)} + + +def _map_index(symbols: list[str]) -> dict[str, int]: + return {s: i for i, s in enumerate(symbols)} + + +def build_theme_scores(theme_id: str, signals: list[dict]) -> dict[str, float]: + """Score every symbol in a theme's signal list. + + `signals` is the tourism-style list of per-symbol signal dicts with + `score` (higher = more bullish) and `symbol`. + """ + out: dict[str, float] = {} + for sig in signals: + sym = sig.get("symbol") + score = sig.get("score") + if sym and score is not None: + out[sym] = float(score) + # z-normalize across scored symbols so theme scores are comparable. + syms = list(out.keys()) + values = [out[s] for s in syms] + z = _zscore(values) + return {s: z.get(i, 0.0) for i, s in enumerate(syms)} + + +def build_siamchart_score(factors: dict) -> dict[str, float]: + """Derive a normalized fundamental score from the Siamchart factor view. + + Uses EPS growth YoY and dividend yield as the "value+quality" signals. + Positive EPS growth and higher yield both push the score up. + """ + out: dict[str, float] = {} + for f in factors.get("factors", []): + sym = f.get("symbol") + if not sym: + continue + g = f.get("eps_growth_yoy") + d = f.get("dividend_yield") or 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 + out[sym] = g + d * 2.0 + syms = list(out.keys()) + z = _zscore([out[s] for s in syms]) + return {s: z.get(i, 0.0) for i, s in enumerate(syms)} + + +def combine_score(theme_scores: list[dict[str, float]], siamchart_score: dict[str, float], + weight_theme: float = 0.6, weight_siamchart: float = 0.4) -> dict[str, dict]: + """Combine per-symbol theme (averaged) and siamchart scores into final. + + Returns {symbol: {'theme_score', 'siamchart_score', 'combined', 'themes':[...]}}. + """ + all_syms: dict[str, list[float]] = {} + theme_membership: dict[str, list[str]] = {} + + for ts in theme_scores: + for sym, val in ts.items(): + all_syms.setdefault(sym, []).append(val) + theme_membership.setdefault(sym, []).append(sym) + + merged: dict[str, dict] = {} + # every symbol that has a siamchart score OR a theme score + universe = set(all_syms) | set(siamchart_score) + for sym in universe: + tl = all_syms.get(sym, []) + tavg = (sum(tl) / len(tl)) if tl else 0.0 + sc = siamchart_score.get(sym, 0.0) + combined = weight_theme * tavg + weight_siamchart * sc + merged[sym] = { + "theme_score": tavg, + "siamchart_score": sc, + "combined": combined, + "themes": theme_membership.get(sym, []), + } + return merged diff --git a/backend/tests/test_themes.py b/backend/tests/test_themes.py new file mode 100644 index 0000000..ba243cd --- /dev/null +++ b/backend/tests/test_themes.py @@ -0,0 +1,72 @@ +"""Tests for the multi-theme registry + scoring.""" + +from __future__ import annotations + +import unittest + +from app import themes + + +class ThemesTest(unittest.TestCase): + def test_list_themes_has_three(self) -> None: + t = themes.list_themes() + self.assertEqual(len(t), 3) + ids = {x.id for x in t} + self.assertEqual(ids, {"tourism", "auto_credit", "refining_energy"}) + for x in t: + self.assertTrue(x.label_th) # Thai label present + + def test_exposure_mapping_contains_expected(self) -> None: + self.assertIn("AOT", themes.THEME_SYMBOLS["tourism"]) + self.assertIn("PTT", themes.THEME_SYMBOLS["refining_energy"]) + self.assertIn("TISCO", themes.THEME_SYMBOLS["auto_credit"]) + + def test_frequency_recorded_per_theme(self) -> None: + self.assertEqual(themes.THEME_FREQUENCY["refining_energy"], "quarterly") + self.assertEqual(themes.THEME_FREQUENCY["tourism"], "monthly") + + def test_build_theme_scores_z_normalizes(self) -> None: + signals = [ + {"symbol": "A", "score": 10.0}, + {"symbol": "B", "score": 5.0}, + {"symbol": "C", "score": 0.0}, + ] + scores = themes.build_theme_scores("tourism", signals) + self.assertAlmostEqual(scores["A"], 1.224, places=2) + self.assertAlmostEqual(scores["B"], 0.0, places=2) + self.assertAlmostEqual(scores["C"], -1.224, places=2) + + def test_siamchart_score_uses_growth_and_yield(self) -> None: + factors = { + "factors": [ + {"symbol": "X", "eps_growth_yoy": 10.0, "dividend_yield": 2.0}, + {"symbol": "Y", "eps_growth_yoy": -5.0, "dividend_yield": 0.0}, + ] + } + sc = themes.build_siamchart_score(factors) + # X = 10 + 2*2 = 14 ; Y = -5 -> X higher + self.assertGreater(sc["X"], sc["Y"]) + + def test_combine_60_40(self) -> None: + theme_scores = [{"A": 1.0, "B": -1.0}] + siamchart = {"A": 2.0, "B": 0.0} + merged = themes.combine_score(theme_scores, siamchart) + # A: 0.6*1.0 + 0.4*2.0 = 1.4 ; B: 0.6*(-1) + 0.4*0 = -0.6 + self.assertAlmostEqual(merged["A"]["combined"], 1.4, places=5) + self.assertAlmostEqual(merged["B"]["combined"], -0.6, places=5) + self.assertAlmostEqual(merged["A"]["theme_score"], 1.0, places=5) + + def test_multiple_themes_average(self) -> None: + # Symbol in two themes -> theme_score is the mean. + t1 = {"A": 1.0} + t2 = {"A": 3.0} + merged = themes.combine_score([t1, t2], {}) + self.assertAlmostEqual(merged["A"]["theme_score"], 2.0, places=5) + + def test_missing_sym_siamchart_gets_theme_only(self) -> None: + merged = themes.combine_score([{"A": 2.0}], {}) + self.assertAlmostEqual(merged["A"]["combined"], 0.6 * 2.0, places=5) + + +if __name__ == "__main__": + unittest.main()