d87a1ada39
A. Cross-theme comparability: - compute_theme_surprises now weight-normalizes by total |weight| (weighted average), so every theme surprise on same [-1,1] scale regardless of factor count/weight (retail 0.189->0.145; auto_credit 1.0->0.64). B. Historical factor store (enables learning macro/demographic factors): - New factor_history.py: append-only per-factor JSONL, dedupes unchanged values, rejects non-finite, records every FACTORS value each scheduler run. - scheduler.py: jobs carry fetch_module; refresh_all records factor history (non-fatal); added bank_npl job. - GET /api/v1/learning/factors?min_points= reports n_points/learnable per factor so users see when P4 learning unlocks (validated query parsing). - weight_learning: generic learn_factor_series() aggregator (momentum reuses). Independent review deleg_5dd358e3 passed=true (empty security/logic arrays); its two robustness suggestions applied (finite guard in record(), clean 400 on bad min_points). 234 tests pass; Vite build passes.
186 lines
8.4 KiB
Python
186 lines
8.4 KiB
Python
"""Tests for the multi-theme registry + scoring."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from app import themes
|
|
|
|
|
|
class ThemesTest(unittest.TestCase):
|
|
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_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()
|
|
|
|
|
|
class SymbolBreakdownTest(unittest.TestCase):
|
|
def test_breakdown_shows_components(self):
|
|
factor_view = {
|
|
"factors": [
|
|
{"symbol": "AOT", "eps_growth_yoy": 10.0, "dividend_yield": 2.0,
|
|
"pe": 20.0, "eps": 5.0, "pbv": 2.0, "roe": 15.0, "is_dividend": True,
|
|
"company_name": "Airports"},
|
|
]
|
|
}
|
|
theme_surprises = {"tourism": 0.57, "auto_credit": 0.81, "refining_energy": 1.62}
|
|
from app import themes
|
|
d = themes.symbol_breakdown("AOT", factor_view=factor_view, theme_surprises=theme_surprises,
|
|
latest_price=67.0, price_date="2026-08-21")
|
|
self.assertEqual(d["symbol"], "AOT")
|
|
self.assertEqual(d["themes"], ["tourism"]) # AOT in tourism map
|
|
self.assertEqual(d["theme_contributions"][0]["surprise"], 0.57)
|
|
self.assertIn("siamchart_components", d)
|
|
self.assertEqual(d["weights"], {"theme": 0.6, "siamchart": 0.4})
|
|
self.assertIsInstance(d["combined_score"], float)
|
|
self.assertEqual(d["price"]["latest"], 67.0)
|
|
self.assertIn("fundamentals", d)
|
|
|
|
def test_breakdown_symbol_without_theme(self):
|
|
factor_view = {"factors": [{"symbol": "BANPU", "eps_growth_yoy": -2.0,
|
|
"dividend_yield": 0.0, "is_dividend": False,
|
|
"company_name": "BANPU"}]}
|
|
from app import themes
|
|
d = themes.symbol_breakdown("BANPU", factor_view=factor_view, theme_surprises={})
|
|
self.assertEqual(d["theme_score"], 0.0)
|
|
# BANPU is in energy/petrochem/utilities maps (full SET50 coverage)
|
|
self.assertTrue(set(d["themes"]) >= {"refining_energy", "petrochem_materials", "utilities"})
|
|
# no surprises set -> every contribution has surprise=None and theme_score 0
|
|
self.assertTrue(all(c["surprise"] is None for c in d["theme_contributions"]))
|
|
|
|
|
|
class QualitySelectionTest(unittest.TestCase):
|
|
def test_quality_differentiates_strong_vs_weak_in_theme(self):
|
|
from app import themes
|
|
fv = {"factors": [
|
|
{"symbol": "BBL", "roe": 12.0, "eps_growth_yoy": 8.0},
|
|
{"symbol": "KBANK", "roe": 10.0, "eps_growth_yoy": 5.0},
|
|
{"symbol": "KTB", "roe": 9.0, "eps_growth_yoy": 3.0},
|
|
{"symbol": "SCB", "roe": 8.0, "eps_growth_yoy": 2.0},
|
|
{"symbol": "TTB", "roe": 5.0, "eps_growth_yoy": -2.0},
|
|
]}
|
|
q_bbl = themes.quality_within_theme("BBL", "banks", fv)
|
|
q_ttb = themes.quality_within_theme("TTB", "banks", fv)
|
|
self.assertGreater(q_bbl, q_ttb) # strong bank outscores weak one
|
|
|
|
def test_breakdown_has_quality_and_theme_score_per_theme(self):
|
|
from app import themes
|
|
fv = {"factors": [
|
|
{"symbol": "BBL", "roe": 12.0, "eps_growth_yoy": 8.0},
|
|
{"symbol": "KTB", "roe": 9.0, "eps_growth_yoy": 3.0},
|
|
]}
|
|
d = themes.symbol_breakdown("BBL", factor_view=fv, theme_surprises={"banks": 1.0})
|
|
contrib = next(c for c in d["theme_contributions"] if c["theme"] == "banks")
|
|
self.assertIn("quality", contrib)
|
|
self.assertIn("theme_score", contrib)
|
|
self.assertAlmostEqual(contrib["theme_score"], contrib["surprise"] * contrib["quality"], places=3)
|
|
|
|
|
|
class RegistryDrivenSurpriseTest(unittest.TestCase):
|
|
"""P0-B: the declarative FACTORS/THEMES registry is now the single source
|
|
of truth for theme surprises — editing a weight genuinely changes output."""
|
|
|
|
@staticmethod
|
|
def _fetched(**macro):
|
|
# Build a minimal `fetched` dict (fetch-module -> collector dict).
|
|
return {
|
|
"macro_thai": {
|
|
"private_consumption_yoy": macro.get("cons", 4.9),
|
|
"private_investment_yoy": macro.get("invest", 18.1),
|
|
"headline_inflation_yoy": macro.get("infl", 1.95),
|
|
"manufacturing_yoy": macro.get("mfg", -3.1),
|
|
"tourists_ytd_mn": macro.get("tour", 16.2),
|
|
},
|
|
"auto_credit": {
|
|
"new_car_sales_yoy": 20.07, "vehicle_production": 117383.0,
|
|
"auto_exports": 81526.0,
|
|
},
|
|
"auto_npl": {"pct_of_npls": 3.95},
|
|
"energy_thai": {
|
|
"quarterly": {"Q1/2026": {"net_profit": 19481.0, "sales": 114809.0}},
|
|
},
|
|
}
|
|
|
|
def test_retail_driven_by_registry_weights(self):
|
|
from app import themes
|
|
s = themes.compute_theme_surprises(self._fetched())
|
|
# registry retail = weighted average over |weights| of
|
|
# consumption*1.0 + inflation*(-0.3):
|
|
# cons=4.9 -> (4.9-3)/10=0.19 (w=1.0) ; inflation sign -1 ->
|
|
# -(1.95-2)/10=+0.005 (w=-0.3) -> weighted = 0.19 - 0.0015 = 0.1885
|
|
# surprise = 0.1885 / (1.0 + 0.3) = 0.145
|
|
self.assertIsNotNone(s["retail"])
|
|
self.assertAlmostEqual(s["retail"], 0.145, places=3)
|
|
|
|
def test_changing_factor_weight_changes_output(self):
|
|
"""The defining property of P0-B: the registry is not decorative."""
|
|
from unittest.mock import patch
|
|
from app import themes
|
|
base = themes.compute_theme_surprises(self._fetched())["retail"]
|
|
# Double consumption's weight in the retail theme -> surprise must rise.
|
|
original = themes.THEMES["retail"]["factors"]
|
|
try:
|
|
themes.THEMES["retail"]["factors"] = [
|
|
{"key": "macro_consumption", "weight": 2.0},
|
|
{"key": "macro_inflation", "weight": -0.3},
|
|
]
|
|
changed = themes.compute_theme_surprises(self._fetched())["retail"]
|
|
finally:
|
|
themes.THEMES["retail"]["factors"] = original
|
|
self.assertGreater(changed, base)
|
|
|
|
def test_tourism_override_wins(self):
|
|
from app import themes
|
|
s = themes.compute_theme_surprises(self._fetched(), tourism_surprise=0.123)
|
|
self.assertEqual(s["tourism"], 0.123)
|
|
# Without override, tourism falls back to registry factors (not None).
|
|
s2 = themes.compute_theme_surprises(self._fetched())
|
|
self.assertIsNotNone(s2["tourism"])
|
|
|
|
def test_normalize_rejects_non_finite(self):
|
|
from app import factors
|
|
self.assertIsNone(factors.normalize(float("nan")))
|
|
self.assertIsNone(factors.normalize(float("inf")))
|
|
self.assertIsNone(factors.normalize(None))
|
|
# finite value still normalizes
|
|
self.assertEqual(factors.normalize(15.0, sign=1, center=5.0, span=10.0), 1.0)
|