diff --git a/backend/app/__init__.py b/backend/app/__init__.py index c013db1..c25c949 100644 --- a/backend/app/__init__.py +++ b/backend/app/__init__.py @@ -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"] diff --git a/backend/app/themes.py b/backend/app/themes.py index eeb0fe7..5b7c5ba 100644 --- a/backend/app/themes.py +++ b/backend/app/themes.py @@ -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]: diff --git a/backend/tests/test_api.py b/backend/tests/test_api.py index f642ef0..1bacdc0 100644 --- a/backend/tests/test_api.py +++ b/backend/tests/test_api.py @@ -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(