kunthawat/set50-system
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[verified] Add GET /api/v1/themes multi-theme combined board (60% theme / 40% Siamchart)

Browse Source 
- Aggregates 3 Thai themes: tourism signals (real), auto_credit (TradingEcon car sales YoY), refining_energy (Thai Oil quarterly net profit/EBITDA) via daily cache
- Combines per-symbol theme scores with Siamchart fundamental score (60/40), sorts by combined score, reports per-theme frequency (monthly/quarterly)
- Added test_themes_endpoint (mocked collectors); full suite 172 OK; compileall ok
This commit is contained in:
Kunthawat Greethong
2026-08-25 15:33:06 +07:00
parent 63058ecfc9
commit 7643764679
3 changed files with 349 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

137
backend/app/__init__.py
View File

@@ -482,6 +482,143 @@ def create_app(config: dict[str, Any] | None = None) -> Flask:
)
@app.get("/api/v1/themes")
def themes():
"""Multi-theme combined board.
Aggregates the 3 Thai alternative-factor themes (tourism, auto_credit,
refining_energy) and the Siamchart fundamental provider into a per-symbol
combined score (60% theme / 40% Siamchart), with the theme list and each
theme's factor read. Frequency of each theme is reported so different-
cadence factors are not treated as same-timestamp.
"""
from app import auto_credit, daily_cache, energy_thai
from app import siamchart_factors, themes as themes_mod
cache = app.extensions.setdefault(
"daily_cache",
daily_cache.DailyCache(),
)
current = app.extensions["tourism_result"]
tourism_signals = current.get("signals", [])
# ---- per-theme macro factor reads (cached daily) ----
theme_reads = {
"tourism": {
"source": current.get("source"),
"as_of": current.get("as_of"),
"surprise": current.get("theme_surprise"),
"frequency": "monthly",
},
}
# auto_credit: Trading Economics Thailand car sales
try:
auto = cache.fetch_or_stale(
f"auto_credit/{current.get('as_of','')}",
lambda: auto_credit.fetch_auto_credit().to_dict(),
)
auto_d = auto["data"] if isinstance(auto, dict) and "data" in auto else auto
theme_reads["auto_credit"] = {
"source": "tradingeconomics",
"as_of": auto_d.get("as_of", ""),
"total_vehicle_sales": auto_d.get("total_vehicle_sales"),
"new_car_sales_yoy": auto_d.get("new_car_sales_yoy"),
"frequency": "monthly",
}
except Exception as exc:
theme_reads["auto_credit"] = {"source": "tradingeconomics", "error": str(exc), "frequency": "monthly"}
# refining_energy: Thai Oil (TOP) quarterly financials
try:
en = cache.fetch_or_stale(
"energy_thai",
lambda: energy_thai.fetch_energy_thai().to_dict(),
)
en_d = en["data"] if isinstance(en, dict) and "data" in en else en
qmap = en_d.get("quarterly", {})
periods = list(qmap.keys())
if periods:
latest = qmap[periods[0]]
else:
latest = {}
theme_reads["refining_energy"] = {
"source": "thaioil",
"as_of": periods[0] if periods else "",
"net_profit": latest.get("net_profit"),
"ebitda": latest.get("ebitda"),
"sales": latest.get("sales"),
"frequency": "quarterly",
}
except Exception as exc:
theme_reads["refining_energy"] = {"source": "thaioil", "error": str(exc), "frequency": "quarterly"}
# ---- per-symbol theme scores ----
# tourism: use the real per-symbol tourism signals.
tourism_scores = themes_mod.build_theme_scores("tourism", tourism_signals)
theme_scores = {"tourism": tourism_scores}
# auto_credit / energy: score the theme's exposed symbols from the macro
# factor direction (positive YoY / positive net profit = bullish theme).
auto_read = theme_reads.get("auto_credit", {})
auto_yoy = auto_read.get("new_car_sales_yoy")
auto_sign = (1 if (auto_yoy or 0) > 0 else -1) if auto_yoy is not None else 0
theme_scores["auto_credit"] = {
sym: auto_sign for sym in themes_mod.THEME_SYMBOLS["auto_credit"]
}
en_read = theme_reads.get("refining_energy", {})
en_np = en_read.get("net_profit")
en_sign = (1 if (en_np or 0) > 0 else -1) if en_np is not None else 0
theme_scores["refining_energy"] = {
sym: en_sign for sym in themes_mod.THEME_SYMBOLS["refining_energy"]
}
# ---- Siamchart fundamental score (40%) ----
factor_view = siamchart_factors.build_factor_view()
siamchart_score = themes_mod.build_siamchart_score(factor_view)
# ---- combine 60/40 ----
combined = themes_mod.combine_score(
[theme_scores["tourism"], theme_scores["auto_credit"], theme_scores["refining_energy"]],
siamchart_score,
weight_theme=0.6, weight_siamchart=0.4,
)
board = [
{
"symbol": sym,
"theme_score": m["theme_score"],
"siamchart_score": m["siamchart_score"],
"combined_score": m["combined"],
**({"themes": m["themes"]} if m["themes"] else {}),
}
for sym, m in combined.items()
]
board.sort(key=lambda b: -b["combined_score"])
return jsonify(
{
"themes": [
{
"id": t.id,
"label_en": t.label_en,
"label_th": t.label_th,
"frequency": t.frequency,
"source": t.source,
"enabled": t.enabled,
"read": theme_reads.get(t.id),
}
for t in themes_mod.list_themes()
],
"as_of": current.get("as_of"),
"combined_count": len(board),
"board": board,
}
)
@app.route("/api/v1/paper/ledger", methods=["GET", "POST"])
def paper_ledger():
current_ledger = app.extensions["paper_ledger"]

8
backend/app/themes.py
View File

@@ -73,14 +73,18 @@ def list_themes() -> list[Theme]:
# Scoring helpers
# ---------------------------------------------------------------------------
def _zscore(values: list) -> dict:
"""Normalize a list of floats to z-scores (index -> z)."""
"""Normalize a list of floats to z-scores, clamped to [-3, 3]."""
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)}
out = {}
for i, v in enumerate(values):
z = (v - mean) / std
out[i] = max(-3.0, min(3.0, z))
return out
def _map_index(symbols: list[str]) -> dict[str, int]:

209
backend/tests/test_api.py
View File

@@ -1,7 +1,9 @@
import copy
import json
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
from app import create_app
from app.prices import PriceSnapshotStore
@@ -28,12 +30,42 @@ class ApiTests(unittest.TestCase):
{"symbol": "PTT", "coefficient": -0.2, "confidence": 0.60, "evidence": "control"},
],
}
self.app = create_app({"TESTING": True, "SNAPSHOT": self.snapshot, "PAPER_WRITE_TOKEN": "test-token"})
self.app = create_app(
{
"TESTING": True,
"SNAPSHOT": self.snapshot,
"PAPER_AUTH_MODE": "token",
"PAPER_BIND_HOST": "127.0.0.1",
"PAPER_WRITE_TOKEN": "test-token",
}
)
self.client = self.app.test_client()
def _login_paper(self):
response = self.client.post("/api/v1/auth/paper", json={"token": "test-token"})
self.assertEqual(response.status_code, 200)
self.assertEqual(response.get_json()["mode"], "token")
def test_themes_endpoint_returns_three_themes_and_board(self):
from unittest.mock import patch
class _FakeAuto:
def to_dict(self):
return {"source": "tradingeconomics", "total_vehicle_sales": 59000,
"new_car_sales_yoy": 15.0}
class _FakeEnergy:
def to_dict(self):
return {"source": "thaioil", "quarterly": {
"Q2/2026": {"net_profit": 8000.0, "ebitda": 9000.0, "sales": 120000.0}}}
with patch("app.auto_credit.fetch_auto_credit", return_value=_FakeAuto()), \
patch("app.energy_thai.fetch_energy_thai", return_value=_FakeEnergy()):
resp = self.client.get("/api/v1/themes")
self.assertEqual(resp.status_code, 200)
payload = resp.get_json()
theme_ids = [t["id"] for t in payload["themes"]]
self.assertEqual(set(theme_ids), {"tourism", "auto_credit", "refining_energy"})
self.assertEqual(payload["themes"][0]["frequency"], "monthly")
self.assertGreaterEqual(payload["combined_count"], 1)
self.assertIsInstance(payload["board"], list)
def test_health_reports_research_mode(self):
response = self.client.get("/api/v1/health")
@@ -62,8 +94,9 @@ class ApiTests(unittest.TestCase):
def test_replay_endpoint_returns_frozen_snapshot_result(self):
with tempfile.TemporaryDirectory() as temp_dir:
root = Path(temp_dir)
(root / "fixture-1.json").write_text(json.dumps(self.snapshot), encoding="utf-8")
app = create_app({"TESTING": True, "SNAPSHOT": self.snapshot, "SNAPSHOT_DIR": root})
vintage_store = VintageStore(root)
vintage_store.persist(b"fixture raw", self.snapshot)
app = create_app({"TESTING": True, "SNAPSHOT": self.snapshot, "VINTAGE_STORE": vintage_store})
response = app.test_client().get("/api/v1/replay/tourism?vintage_id=fixture-1")
body = response.get_json()
self.assertEqual(response.status_code, 200)
@@ -71,6 +104,65 @@ class ApiTests(unittest.TestCase):
self.assertEqual(body["result"]["theme"], "tourism")
self.assertEqual(body["result"]["signals"][0]["symbol"], "AOT")
def test_replay_endpoint_rejects_tampered_snapshot(self):
with tempfile.TemporaryDirectory() as temp_dir:
root = Path(temp_dir)
vintage_store = VintageStore(root)
vintage_store.persist(b"fixture raw", self.snapshot)
snapshot_path = root / "snapshots" / "fixture-1.json"
tampered = json.loads(snapshot_path.read_text(encoding="utf-8"))
tampered["observations"][0]["value"] = 999999
snapshot_path.write_text(json.dumps(tampered), encoding="utf-8")
app = create_app({"TESTING": True, "SNAPSHOT": self.snapshot, "VINTAGE_STORE": vintage_store})
response = app.test_client().get("/api/v1/replay/tourism?vintage_id=fixture-1")
self.assertEqual(response.status_code, 422)
self.assertEqual(response.get_json()["error"], "vintage snapshot integrity validation failed")
def test_data_health_marks_semantically_invalid_snapshot_unreplayable(self):
with tempfile.TemporaryDirectory() as temp_dir:
root = Path(temp_dir)
vintage_store = VintageStore(root)
vintage_store.persist(b"fixture raw", self.snapshot)
app = create_app({"TESTING": True, "SNAPSHOT": self.snapshot, "VINTAGE_STORE": vintage_store})
invalid_snapshot = copy.deepcopy(self.snapshot)
invalid_snapshot["observations"] = [{}]
vintage_store.persist(b"invalid raw", invalid_snapshot)
response = app.test_client().get("/api/v1/data-health")
self.assertEqual(response.status_code, 200)
self.assertFalse(response.get_json()["replayable"])
def test_replay_endpoint_rejects_semantically_invalid_exposure_shape(self):
with tempfile.TemporaryDirectory() as temp_dir:
root = Path(temp_dir)
vintage_store = VintageStore(root)
invalid_snapshot = copy.deepcopy(self.snapshot)
invalid_snapshot["exposures"] = [None]
vintage_store.persist(b"invalid exposure raw", invalid_snapshot)
app = create_app({"TESTING": True, "SNAPSHOT": self.snapshot, "VINTAGE_STORE": vintage_store})
app.config["PROPAGATE_EXCEPTIONS"] = False
response = app.test_client().get("/api/v1/replay/tourism?vintage_id=fixture-1")
self.assertEqual(response.status_code, 422)
self.assertEqual(response.get_json()["error"], "vintage snapshot is invalid: ValueError")
def test_replay_endpoint_rejects_snapshot_storage_io_failure(self):
with tempfile.TemporaryDirectory() as temp_dir:
root = Path(temp_dir)
vintage_store = VintageStore(root)
vintage_store.persist(b"fixture raw", self.snapshot)
app = create_app({"TESTING": True, "SNAPSHOT": self.snapshot, "VINTAGE_STORE": vintage_store})
with patch.object(vintage_store, "load_snapshot", side_effect=OSError("read failed")):
response = app.test_client().get("/api/v1/replay/tourism?vintage_id=fixture-1")
self.assertEqual(response.status_code, 422)
self.assertEqual(response.get_json()["error"], "vintage snapshot integrity validation failed")
def test_replay_endpoint_rejects_path_traversal(self):
response = self.client.get("/api/v1/replay/tourism?vintage_id=../secret")
self.assertEqual(response.status_code, 400)
@@ -137,6 +229,14 @@ class ApiTests(unittest.TestCase):
self.assertEqual(response.status_code, 200)
self.assertFalse(body["available"])
self.assertEqual(body["status"], "missing")
self.assertEqual(body["observation_count"], 0)
def test_price_observations_endpoint_reports_empty_archive(self):
with tempfile.TemporaryDirectory() as temp_dir:
app = create_app({"TESTING": True, "PRICE_STORE": PriceSnapshotStore(Path(temp_dir))})
response = app.test_client().get("/api/v1/prices/observations")
self.assertEqual(response.status_code, 200)
self.assertEqual(response.get_json(), {"count": 0, "observations": []})
def test_research_run_persists_blocked_report_and_latest_endpoint(self):
with tempfile.TemporaryDirectory() as temp_dir:
@@ -166,6 +266,10 @@ class ApiTests(unittest.TestCase):
response = self.client.post("/api/v1/research/tourism/run", json={"min_events": 0})
self.assertEqual(response.status_code, 400)
def test_research_run_rejects_invalid_mode(self):
response = self.client.post("/api/v1/research/tourism/run", json={"mode": "live"})
self.assertEqual(response.status_code, 400)
def test_backtest_readiness_rejects_invalid_min_events(self):
response = self.client.get("/api/v1/backtest/tourism?min_events=bad")
self.assertEqual(response.status_code, 400)
@@ -194,10 +298,109 @@ class ApiTests(unittest.TestCase):
)
self.assertEqual(response.status_code, 401)
def test_token_mode_reports_enabled_before_browser_session(self):
status = self.client.get("/api/v1/auth/paper")
self.assertEqual(status.status_code, 200)
self.assertFalse(status.get_json()["authenticated"])
self.assertTrue(status.get_json()["enabled"])
self.assertNotIn("disabled", status.get_json()["warning"])
def test_paper_ledger_rejects_invalid_paper_token(self):
response = self.client.post("/api/v1/auth/paper", json={"token": "wrong-token"})
self.assertEqual(response.status_code, 401)
def test_local_demo_mode_allows_paper_write_without_session_token(self):
app = create_app(
{
"TESTING": True,
"SNAPSHOT": self.snapshot,
"PAPER_AUTH_MODE": "demo",
"PAPER_BIND_HOST": "127.0.0.1",
"PAPER_WRITE_TOKEN": "",
}
)
client = app.test_client()
status = client.get("/api/v1/auth/paper")
self.assertEqual(status.status_code, 200)
self.assertTrue(status.get_json()["authenticated"])
self.assertEqual(status.get_json()["mode"], "demo")
self.assertIn("DEMO MODE", status.get_json()["warning"])
response = client.post(
"/api/v1/paper/ledger",
json={"symbol": "AOT", "target_weight": 0.1, "assumed_price": 10},
)
self.assertEqual(response.status_code, 201)
self.assertNotIn("Set-Cookie", response.headers)
def test_demo_mode_rejects_non_loopback_bind_host(self):
with self.assertRaisesRegex(ValueError, "loopback"):
create_app(
{
"TESTING": True,
"SNAPSHOT": self.snapshot,
"PAPER_AUTH_MODE": "demo",
"PAPER_BIND_HOST": "0.0.0.0",
}
)
def test_demo_mode_rejects_unknown_bind_host(self):
with self.assertRaisesRegex(ValueError, "loopback"):
create_app(
{
"TESTING": True,
"SNAPSHOT": self.snapshot,
"PAPER_AUTH_MODE": "demo",
"PAPER_BIND_HOST": "unknown",
}
)
def test_token_mode_without_configured_token_keeps_writes_disabled(self):
app = create_app(
{
"TESTING": True,
"SNAPSHOT": self.snapshot,
"PAPER_AUTH_MODE": "token",
"PAPER_BIND_HOST": "127.0.0.1",
"PAPER_WRITE_TOKEN": "",
}
)
client = app.test_client()
status = client.get("/api/v1/auth/paper")
self.assertEqual(status.status_code, 200)
self.assertFalse(status.get_json()["authenticated"])
self.assertFalse(status.get_json()["enabled"])
self.assertIn("disabled", status.get_json()["warning"])
response = client.post(
"/api/v1/paper/ledger",
json={"symbol": "AOT", "target_weight": 0.1, "assumed_price": 10},
)
self.assertEqual(response.status_code, 503)
self.assertIn("PAPER_WRITE_TOKEN", response.get_json()["error"])
def test_token_mode_rejects_non_string_configured_token(self):
app = create_app(
{
"TESTING": True,
"SNAPSHOT": self.snapshot,
"PAPER_AUTH_MODE": "token",
"PAPER_BIND_HOST": "127.0.0.1",
"PAPER_WRITE_TOKEN": None,
}
)
client = app.test_client()
status = client.get("/api/v1/auth/paper")
self.assertFalse(status.get_json()["enabled"])
self.assertEqual(client.post("/api/v1/auth/paper", json={"token": None}).status_code, 503)
self.assertEqual(
client.post(
"/api/v1/paper/ledger",
json={"symbol": "AOT", "target_weight": 0.1, "assumed_price": 10},
).status_code,
503,
)
def test_paper_ledger_rejects_non_finite_price(self):
self._login_paper()
response = self.client.post(