[verified] Add per-symbol analysis breakdown endpoint + per-theme narrative

- themes.symbol_breakdown(): transparent scoring derivation (theme_score, siamchart_score components, combined = 0.6*theme + 0.4*siamchart npolut)
- GET /api/v1/symbols/<symbol>: themes + theme surprise contributions + fundamentals + price + weights (ข้อ 7)
- dashboard.py _theme_narrative(): long-form Thai explanation of each theme's analysis outcome + implication for its stocks (ข้อ 5)
- 2 tests; full suite OK; live verified (AOT: combined 0.107 = 0.6*0.571 + 0.4*(-0.588))
This commit is contained in:
Kunthawat Greethong
2026-08-25 21:18:20 +07:00
parent bfa9b08af6
commit 33a4662cd4
4 changed files with 165 additions and 1 deletions

View File

@@ -739,6 +739,42 @@ def create_app(config: dict[str, Any] | None = None) -> Flask:
) )
@app.get("/api/v1/symbols/<symbol>")
def symbol_detail(symbol: str):
"""Transparent per-symbol analysis breakdown (themes -> weights -> combined)."""
from app import themes as themes_mod
from app import siamchart_factors, simulation
symbol = symbol.upper()
factor_view = siamchart_factors.build_factor_view()
if symbol not in {f.get("symbol") for f in factor_view.get("factors", [])}:
return jsonify({"error": f"unknown symbol {symbol}", "symbol": symbol}), 404
# per-theme surprise from the real dashboard
from app.dashboard import RealDashboard
from app import daily_cache
cache = app.extensions.setdefault("daily_cache", daily_cache.DailyCache())
current = app.extensions.get("tourism_result")
try:
dash = RealDashboard((current or {}).get("signals", []), cache).build()
theme_surprises = {t["id"]: t.get("surprise") for t in dash.get("themes", [])}
except Exception:
theme_surprises = {}
# latest price from the Yahoo snapshot
price = None
price_date = ""
try:
series = simulation.load_price_snapshot()
bars = series.get(symbol, {}).get("bars", [])
if bars:
price = float(bars[-1]["adjusted_close"])
price_date = bars[-1].get("date", "")
except Exception:
pass
detail = themes_mod.symbol_breakdown(
symbol, factor_view=factor_view, theme_surprises=theme_surprises,
latest_price=price, price_date=price_date,
)
return jsonify(detail)
@app.get("/api/v1/data/last-refresh") @app.get("/api/v1/data/last-refresh")
def last_refresh(): def last_refresh():
"""Status of the in-app automatic data refresh (independent of Hermes).""" """Status of the in-app automatic data refresh (independent of Hermes)."""

View File

@@ -139,11 +139,37 @@ class RealDashboard:
read = {k: v for k, v in read.items() if k != "thesis"} read = {k: v for k, v in read.items() if k != "thesis"}
else: else:
thesis = "" thesis = ""
narrative = self._theme_narrative(tid, surprise, read)
return { return {
"id": tid, "label_th": label, "frequency": freq, "id": tid, "label_th": label, "frequency": freq,
"surprise": surprise, "read": read, "thesis": thesis, "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"ที่ได้อานิสงส์จากนักท่องเที่ยวจะเข้าอานิสงส์ตามทิศทางนี้.")
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) -> 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

View File

@@ -160,3 +160,72 @@ def combine_score(theme_scores: list[dict[str, float]], siamchart_score: dict[st
"themes": theme_membership.get(sym, []), "themes": theme_membership.get(sym, []),
} }
return merged return merged
def symbol_breakdown(
symbol: str,
*,
factor_view: dict,
theme_surprises: dict[str, float],
latest_price: Optional[float] = None,
price_date: str = "",
weight_theme: float = 0.6,
weight_siamchart: float = 0.4,
) -> dict:
"""Transparent per-symbol scoring breakdown.
Shows exactly how `combined` was derived:
theme_score = mean of the theme surprise scores covering this symbol
siamchart_score = z-scored (EPS growth + dividend_yield*2)
combined = weight_theme*theme_score + weight_siamchart*siamchart_score
Returns a dict suitable for the /api/v1/symbols/<symbol> view. Deterministic
and reuses the same formula as `combine_score` so the board and the detail
always agree.
"""
# the themes this symbol belongs to (from the curated exposure map)
member_themes = [tid for tid, syms in THEME_SYMBOLS.items() if symbol in syms]
theme_lines = []
theme_values = []
for tid in member_themes:
s = theme_surprises.get(tid)
if s is not None:
theme_values.append(float(s))
theme_lines.append({"theme": tid, "surprise": round(float(s), 3)})
theme_score = (sum(theme_values) / len(theme_values)) if theme_values else 0.0
# find the factor row for this symbol
fac = next((f for f in factor_view.get("factors", []) if f.get("symbol") == symbol), {})
g = fac.get("eps_growth_yoy")
d = fac.get("dividend_yield") or 0.0
g = float(g) if g is not None else 0.0
raw_siamchart = g + d * 2.0
# z-score against the full universe (same as build_siamchart_score)
siamchart_score = build_siamchart_score(factor_view).get(symbol, 0.0)
combined = round(weight_theme * theme_score + weight_siamchart * siamchart_score, 3)
return {
"symbol": symbol,
"company_name": fac.get("company_name", ""),
"themes": member_themes,
"theme_contributions": theme_lines,
"theme_score": round(theme_score, 3),
"siamchart_score": round(siamchart_score, 3),
"siamchart_components": {
"eps_growth_yoy": fac.get("eps_growth_yoy"),
"dividend_yield": fac.get("dividend_yield"),
"raw_growth_plus_yield_2x": round(raw_siamchart, 3),
},
"combined_score": combined,
"weights": {"theme": weight_theme, "siamchart": weight_siamchart},
"fundamentals": {
"pe": fac.get("pe"),
"eps": fac.get("eps"),
"pbv": fac.get("pbv"),
"roe": fac.get("roe"),
"dps": fac.get("dps"),
"is_dividend": fac.get("is_dividend"),
},
"price": {"latest": latest_price, "date": price_date},
}

View File

@@ -70,3 +70,36 @@ class ThemesTest(unittest.TestCase):
if __name__ == "__main__": if __name__ == "__main__":
unittest.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)
self.assertEqual(d["themes"], ["refining_energy"]) # BANPU in energy map
self.assertEqual(d["theme_contributions"], []) # but no surprise set -> 0