[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
2026-08-25 15:19:39 +07:00
parent 3f7fccd25b
commit affc29a3be
2 changed files with 230 additions and 0 deletions
--- a/backend/app/themes.py
+++ 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
--- a/backend/tests/test_themes.py
+++ 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()