"""Flask application factory for the SET50 alternative-data platform.""" from __future__ import annotations import hmac import json import os import re import secrets import statistics import time from pathlib import Path from typing import Any from flask import Flask, jsonify, request from .bot_tourism import BotTourismSource, TourismSourceError from .event_study import EventStudyError, assess_backtest_readiness from .paper import PaperLedger from .prices import PIT_ARCHIVE_CONTRACT, PriceSnapshotStore, PriceSourceError from .research import ResearchRunError, ResearchRunStore, run_tourism_research from .tourism import compute_tourism_signal from .vintages import VintageStore, VintageStoreError APP_VERSION = "0.5.0" PAPER_AUTH_MODES = {"demo", "token"} LOOPBACK_BIND_HOSTS = {"127.0.0.1", "localhost", "::1"} PAPER_DEMO_WARNING = ( "DEMO MODE: unauthenticated local paper writes; loopback-only and paper-only. No live orders." ) def _load_default_snapshot() -> dict[str, Any]: fixture_path = Path(__file__).resolve().parents[1] / "fixtures" / "tourism_snapshot.json" return json.loads(fixture_path.read_text(encoding="utf-8")) def _signal_summary(result: dict[str, Any]) -> dict[str, int]: signals = result["signals"] return { "total": len(signals), "long": sum(item["side"] == "LONG" for item in signals), "short": sum(item["side"] == "SHORT" for item in signals), "neutral": sum(item["side"] == "NEUTRAL" for item in signals), } def create_app(config: dict[str, Any] | None = None) -> Flask: app = Flask(__name__) data_root = Path(__file__).resolve().parents[1] / "data" app.config.from_mapping( TESTING=False, MODE=os.getenv("APP_MODE", "research"), PAPER_AUTH_MODE=os.getenv("PAPER_AUTH_MODE", "token"), PAPER_BIND_HOST=os.getenv("PAPER_BIND_HOST", os.getenv("HOST", "unknown")), PAPER_WRITE_TOKEN=os.getenv("PAPER_WRITE_TOKEN", ""), PAPER_COOKIE_SECURE=os.getenv("PAPER_COOKIE_SECURE", "0") == "1", PAPER_SESSION_SECONDS=int(os.getenv("PAPER_SESSION_SECONDS", "3600")), PAPER_LEDGER_PATH=Path(os.getenv("PAPER_LEDGER_PATH", str(data_root / "paper" / "ledger.json"))), TOURISM_SOURCE=os.getenv("TOURISM_SOURCE", "fixture"), TOURISM_ADAPTER=None, TOURISM_DATA_ROOT=Path(os.getenv("TOURISM_DATA_ROOT", str(data_root))), TOURISM_RAW_DIR=Path(os.getenv("TOURISM_RAW_DIR", str(data_root / "raw" / "tourism"))), SNAPSHOT_DIR=Path(os.getenv("TOURISM_SNAPSHOT_DIR", str(data_root / "snapshots"))), VINTAGE_STORE=None, PRICE_DATA_ROOT=Path(os.getenv("PRICE_DATA_ROOT", str(data_root / "prices"))), PRICE_STORE=None, RESEARCH_DATA_ROOT=Path(os.getenv("RESEARCH_DATA_ROOT", str(data_root / "research"))), RESEARCH_RUN_STORE=None, SNAPSHOT=_load_default_snapshot(), ) if config: app.config.update(config) paper_auth_mode = str(app.config.get("PAPER_AUTH_MODE", "token")).strip().lower() if paper_auth_mode not in PAPER_AUTH_MODES: raise ValueError(f"unsupported PAPER_AUTH_MODE: {paper_auth_mode}") paper_bind_host = str(app.config.get("PAPER_BIND_HOST", "127.0.0.1")).strip().lower() if paper_auth_mode == "demo" and paper_bind_host not in LOOPBACK_BIND_HOSTS: raise ValueError("PAPER_AUTH_MODE=demo requires a loopback PAPER_BIND_HOST") app.config["PAPER_AUTH_MODE"] = paper_auth_mode app.config["PAPER_BIND_HOST"] = paper_bind_host vintage_store = app.config.get("VINTAGE_STORE") or VintageStore(app.config["TOURISM_DATA_ROOT"]) app.extensions["vintage_store"] = vintage_store price_store = app.config.get("PRICE_STORE") or PriceSnapshotStore(app.config["PRICE_DATA_ROOT"]) app.extensions["price_store"] = price_store research_run_store = app.config.get("RESEARCH_RUN_STORE") or ResearchRunStore(app.config["RESEARCH_DATA_ROOT"]) app.extensions["research_run_store"] = research_run_store source_snapshot = app.config["SNAPSHOT"] source_mode = str(app.config["TOURISM_SOURCE"]).lower() if source_mode == "bot": adapter = app.config.get("TOURISM_ADAPTER") or BotTourismSource(vintage_store=vintage_store) try: source_snapshot = adapter.fetch(exposures=source_snapshot.get("exposures", [])) except TourismSourceError as exc: raise RuntimeError(f"Tourism source startup failed: {exc}") from exc elif source_mode != "fixture": raise ValueError(f"unsupported TOURISM_SOURCE: {source_mode}") result = compute_tourism_signal(source_snapshot) ledger_path = app.config["PAPER_LEDGER_PATH"] if app.config["TESTING"] and (not config or "PAPER_LEDGER_PATH" not in config): ledger_path = None ledger = PaperLedger(ledger_path) paper_sessions: dict[str, float] = {} allowed_symbols = {item["symbol"] for item in result["signals"]} app.extensions["tourism_result"] = result app.extensions["paper_ledger"] = ledger app.extensions["paper_sessions"] = paper_sessions app.extensions["allowed_symbols"] = allowed_symbols # shared in-process daily cache + app-internal data scheduler (independent of # Hermes — this app runs on its own server). from .daily_cache import DailyCache from .scheduler import AppDataScheduler cache = DailyCache() app.extensions["daily_cache"] = cache if not app.config.get("TESTING"): interval_s = int(os.getenv("REFRESH_INTERVAL_SECONDS", "3600")) scheduler = AppDataScheduler(cache, Path(__file__).resolve().parents[1] / "data", interval_seconds=interval_s) scheduler.start() app.extensions["data_scheduler"] = scheduler @app.after_request def add_cors_headers(response): origin = request.headers.get("Origin") allowed_origin = os.getenv("CORS_ORIGIN", "http://localhost:5173") if origin == allowed_origin: response.headers["Access-Control-Allow-Origin"] = origin response.headers["Vary"] = "Origin" response.headers["Access-Control-Allow-Headers"] = "Content-Type" response.headers["Access-Control-Allow-Credentials"] = "true" response.headers["Access-Control-Allow-Methods"] = "GET, POST, OPTIONS" return response def _configured_paper_token() -> str | None: configured = app.config.get("PAPER_WRITE_TOKEN", "") if not isinstance(configured, str) or not configured: return None return configured def _paper_write_authorized() -> tuple[bool, tuple[dict[str, str], int] | None]: if app.config["PAPER_AUTH_MODE"] == "demo": return True, None configured = _configured_paper_token() if not configured: return False, ({"error": "paper writes disabled: PAPER_WRITE_TOKEN is not configured"}, 503) sessions: dict[str, float] = app.extensions["paper_sessions"] now = time.time() expired = [session_id for session_id, expiry in sessions.items() if expiry <= now] for session_id in expired: sessions.pop(session_id, None) session_id = request.cookies.get("paper_session", "") if session_id and session_id in sessions and sessions[session_id] > now: return True, None return False, ({"error": "paper session required"}, 401) def _paper_auth_status(authorized: bool, enabled: bool = True) -> dict[str, Any]: if app.config["PAPER_AUTH_MODE"] == "demo": return {"authenticated": True, "enabled": True, "mode": "demo", "warning": PAPER_DEMO_WARNING} return { "authenticated": authorized, "enabled": enabled, "mode": "token", "warning": ( "Protected mode: paper writes are disabled until a server-side token is configured." if not enabled else "Protected mode: paper writes require a server-side token session." ), } @app.post("/api/v1/auth/paper") def authenticate_paper(): if app.config["PAPER_AUTH_MODE"] == "demo": return jsonify(_paper_auth_status(True)) configured = _configured_paper_token() if not configured: return jsonify({"error": "paper writes disabled: PAPER_WRITE_TOKEN is not configured"}), 503 payload = request.get_json(silent=True) candidate = payload.get("token", "") if isinstance(payload, dict) else "" if not isinstance(candidate, str): candidate = "" if not hmac.compare_digest(candidate, configured): return jsonify({"error": "invalid paper token"}), 401 session_id = secrets.token_urlsafe(32) app.extensions["paper_sessions"][session_id] = time.time() + int(app.config["PAPER_SESSION_SECONDS"]) response = jsonify(_paper_auth_status(True)) response.set_cookie( "paper_session", session_id, max_age=int(app.config["PAPER_SESSION_SECONDS"]), httponly=True, secure=bool(app.config["PAPER_COOKIE_SECURE"]), samesite="Strict", ) return response @app.get("/api/v1/auth/paper") def paper_session_status(): authorized, _ = _paper_write_authorized() enabled = app.config["PAPER_AUTH_MODE"] == "demo" or _configured_paper_token() is not None return jsonify(_paper_auth_status(authorized, enabled=enabled)) def _snapshot_file_for(vintage_id: str) -> Path | None: if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}", vintage_id): return None root = Path(app.config["SNAPSHOT_DIR"]).resolve() candidate = (root / f"{vintage_id}.json").resolve() if candidate.parent != root: return None return candidate def _load_replayed_tourism(vintage_id: str) -> dict[str, Any]: frozen_snapshot = app.extensions["vintage_store"].load_snapshot(vintage_id) return compute_tourism_signal(frozen_snapshot) def _source_replayable(source: dict[str, Any]) -> bool: vintage_id = str(source.get("vintage_id", "")) if not vintage_id or _snapshot_file_for(vintage_id) is None: return False try: _load_replayed_tourism(vintage_id) except (FileNotFoundError, VintageStoreError, OSError, KeyError, TypeError, ValueError): return False return True def _price_health_payload() -> dict[str, Any]: try: price_store = app.extensions["price_store"] snapshots = price_store.list_snapshots() observations = price_store.list_observations() except PriceSourceError as exc: return {"available": False, "status": "error", "error": str(exc), "snapshot_count": 0} if not snapshots: return { "available": False, "status": "missing", "point_in_time": False, "snapshot_count": 0, "observation_count": len(observations), } latest = snapshots[-1] latest_observation = observations[-1] if observations else None try: snapshot = price_store.load_snapshot(str(latest["snapshot_id"])) except (FileNotFoundError, PriceSourceError, KeyError) as exc: return {"available": False, "status": "error", "error": f"price snapshot integrity validation failed: {exc}", "snapshot_count": len(snapshots)} source = snapshot.get("source", {}) return { "available": True, "status": "available", "snapshot_count": len(snapshots), "observation_count": len(observations), "snapshot_id": latest.get("snapshot_id"), "retrieved_at": latest.get("retrieved_at"), "last_seen_at": latest.get("last_seen_at", latest.get("retrieved_at")), "last_observation_at": latest_observation.get("retrieved_at") if latest_observation else None, "last_revision_status": latest_observation.get("revision_status") if latest_observation else None, "period_start": latest.get("period_start"), "period_end": latest.get("period_end"), "quality": source.get("quality"), "point_in_time": source.get("point_in_time"), "archive_contract": source.get("archive_contract"), "provider_release_id": source.get("provider_release_id"), "parser_version": source.get("parser_version"), "symbols": latest.get("symbols", []), } @app.get("/api/v1/prices/health") def prices_health(): return jsonify(_price_health_payload()) @app.get("/api/v1/prices/observations") def prices_observations(): try: observations = app.extensions["price_store"].list_observations() except PriceSourceError as exc: return jsonify({"error": str(exc)}), 422 return jsonify({"count": len(observations), "observations": observations}) @app.post("/api/v1/research/tourism/run") def run_tourism_research_endpoint(): payload = request.get_json(silent=True) or {} if not isinstance(payload, dict): return jsonify({"error": "JSON object required"}), 400 try: report = run_tourism_research( app.extensions["vintage_store"], app.extensions["price_store"], app.extensions["research_run_store"], min_events=payload.get("min_events", 1 if payload.get("mode") == "exploratory" else 12), windows=payload.get("windows", (1, 3, 5, 20)), cost_bps=payload.get("cost_bps", 20.0), execution_lag_sessions=payload.get("execution_lag_sessions", 1), mode=payload.get("mode", "validated"), ) except (ResearchRunError, TypeError, ValueError) as exc: return jsonify({"error": str(exc)}), 400 return jsonify(report) @app.get("/api/v1/research/tourism/latest") def latest_tourism_research(): try: report = app.extensions["research_run_store"].latest() except FileNotFoundError: return jsonify({"error": "no research runs"}), 404 except ResearchRunError as exc: return jsonify({"error": str(exc)}), 422 return jsonify(report) @app.get("/api/v1/data-health") def data_health(): current = app.extensions["tourism_result"] source = current["source"] return jsonify( { "status": current["data_quality"], "source_id": source.get("source_id"), "source_url": source.get("source_url"), "published_at": source.get("published_at"), "retrieved_at": source.get("retrieved_at"), "vintage_id": source.get("vintage_id"), "raw_payload_hash": source.get("raw_payload_hash"), "parser_version": source.get("parser_version"), "available_periods": source.get("available_periods"), "history_points": source.get("history_points"), "replayable": _source_replayable(source), "source_mode": app.config["TOURISM_SOURCE"], } ) @app.get("/api/v1/vintages") def vintages(): as_of = request.args.get("as_of") try: entries = app.extensions["vintage_store"].list_vintages(as_of) except VintageStoreError as exc: return jsonify({"error": str(exc)}), 400 return jsonify({"as_of": as_of, "count": len(entries), "vintages": entries}) @app.get("/api/v1/backtest/tourism") def tourism_backtest_readiness(): raw_min_events = request.args.get("min_events", "12") try: min_events = int(raw_min_events) readiness = assess_backtest_readiness(app.extensions["vintage_store"].list_vintages(), min_events) except (ValueError, TypeError, EventStudyError) as exc: return jsonify({"error": str(exc)}), 400 price_snapshot = _price_health_payload() price_is_pit_archive = ( price_snapshot.get("available") and price_snapshot.get("point_in_time") is True and price_snapshot.get("quality") == "point_in_time_archive" and price_snapshot.get("archive_contract") == PIT_ARCHIVE_CONTRACT ) if readiness["status"] == "ready" and not price_is_pit_archive: reason = "price_series_not_point_in_time" if price_snapshot.get("point_in_time") is not True else "price_archive_contract_missing" readiness = { **readiness, "status": "blocked", "reason": reason, } body = { **readiness, "theme": "tourism", "price_series_required": True, "price_snapshot": price_snapshot, "next_action": "collect independent published vintages before running event study" if readiness["reason"] == "insufficient_vintages" else "provide a verified point-in-time daily price archive contract" if readiness["reason"] in {"price_series_not_point_in_time", "price_archive_contract_missing"} else "run event study", } return jsonify(body), 409 if readiness["status"] == "blocked" else 200 @app.get("/api/v1/replay/tourism") def replay_tourism(): vintage_id = request.args.get("vintage_id", "") if _snapshot_file_for(vintage_id) is None: return jsonify({"error": "invalid vintage_id"}), 400 try: replayed = _load_replayed_tourism(vintage_id) except FileNotFoundError: return jsonify({"error": "vintage snapshot not found"}), 404 except (VintageStoreError, OSError): return jsonify({"error": "vintage snapshot integrity validation failed"}), 422 except (TypeError, ValueError, KeyError) as exc: return jsonify({"error": f"vintage snapshot is invalid: {exc.__class__.__name__}"}), 422 return jsonify({"vintage_id": vintage_id, "result": replayed}) @app.get("/api/v1/health") def health(): return jsonify({"status": "ok", "mode": app.config["MODE"], "version": APP_VERSION}) @app.get("/api/v1/dashboard/summary") def dashboard_summary(): current = app.extensions["tourism_result"] entries = app.extensions["paper_ledger"].entries() return jsonify( { "as_of": current["as_of"], "theme": current["theme"], "strategy_version": current["strategy_version"], "theme_surprise": current["theme_surprise"], "data_health": { "status": current["data_quality"], "source_id": current["source"].get("source_id"), "source_url": current["source"].get("source_url"), "published_at": current["source"].get("published_at"), "retrieved_at": current["source"].get("retrieved_at"), "vintage_id": current["source"].get("vintage_id"), "raw_payload_hash": current["source"].get("raw_payload_hash"), "parser_version": current["source"].get("parser_version"), "available_periods": current["source"].get("available_periods"), "history_points": current["source"].get("history_points"), "source_mode": app.config["TOURISM_SOURCE"], "replayable": _source_replayable(current["source"]), }, "signal_summary": _signal_summary(current), "top_signals": current["signals"][:5], "paper_ledger": {"entries": len(entries), "mode": "paper"}, } ) @app.get("/api/v1/factors/tourism/observations") def tourism_observations(): current = app.extensions["tourism_result"] return jsonify( { "theme": current["theme"], "as_of": current["as_of"], "theme_surprise": current["theme_surprise"], "source": current["source"], "observations": current["observations"], } ) @app.get("/api/v1/signals") def signals(): current = app.extensions["tourism_result"] return jsonify( { "theme": current["theme"], "as_of": current["as_of"], "strategy_version": current["strategy_version"], "signals": current["signals"], } ) @app.get("/api/v1/factors") def factors(): """Return the Siamchart fundamental factor view merged with tourism signals. Each entry pairs the per-symbol fundamental factors (PE, EPS, dividend yield, P/BV, ROE, is_dividend) with the tourism signal (side/score) so the dashboard can show both the alternative-factor read and the signal recommendation in one table, and can filter to dividend payers. """ from app import siamchart_factors factor_view = siamchart_factors.build_factor_view() # signal = derived from the NEW theme-engine combined score (60/40 with # per-symbol quality), NOT the old tourism_result. One source of truth. from app import daily_cache from app.dashboard import RealDashboard cache = app.extensions.setdefault("daily_cache", daily_cache.DailyCache()) current = app.extensions.get("tourism_result") or {} signal_by_symbol = {} try: dash = RealDashboard(current.get("signals", []), cache).build() board = dash.get("board", []) combos = [b.get("combined") for b in board if b.get("combined") is not None] if combos: q1, q3 = statistics.quantiles(combos, n=4)[0], statistics.quantiles(combos, n=4)[2] else: q1 = q3 = 0.0 # market-regime gate (R3): how many themes are in distress. Instead # of a hard binary cliff, use a continuous stress = (negative themes # / total) in [0,1] and shift the LONG bar / SHORT threshold # smoothly with it. In a broad-down market we tighten LONG and pull # more names into SHORT/avoid, so 'best of a falling board' isn't LONG. theme_surprises = [t.get("surprise") for t in dash.get("themes", []) if t.get("surprise") is not None] n_themes = max(len(theme_surprises), 1) n_neg = sum(1 for s in theme_surprises if s < 0) stress = n_neg / n_themes # 0..1 continuous regime gauge # gate offset grows with stress (at stress=1 => +0.15 to go LONG) long_bar = q3 + 0.15 * stress # SHORT threshold widens as stress rises (pull more into avoid) short_bar = q1 - 0.05 - 0.08 * stress fmap = {b.get("symbol"): b for b in board} for row in board: comb = row.get("combined") sym = row.get("symbol") if comb is None: signal_by_symbol[sym] = {"side": None, "score": None} continue fac = fmap.get(sym, {}) # R5 (dividend screen): a name that pays no dividend (or has cut # its yield to a negative/zero level) never goes LONG — dividend # is our core value assumption; literature treats a cut as a # screen-off signal. Downgrade to NEUTRAL/SHORT accordingly. is_div = bool(fac.get("is_dividend")) or (fac.get("dividend_yield") or 0) > 0 if comb >= long_bar: if is_div: side, score = "LONG", round(min(abs(comb) * 3.0, 1.0) * 0.9 + 0.1, 3) else: # high score but no dividend -> strong growth but our # thesis is dividend-anchored; cap at NEUTRAL. side, score = "NEUTRAL", round((comb - q1) / max(q3 - q1, 1e-9), 3) elif comb < short_bar: side, score = "SHORT", round(min(abs(comb) / max(abs(short_bar), 1e-6), 1.0) * 0.5, 3) else: median = combos and statistics.median(combos) or 0.0 side = "SHORT" if (comb < median and stress > 0.5) else "NEUTRAL" score = round(abs(comb) / max(abs(q1), 1e-9) * 0.5, 3) if stress > 0.5 \ else round((comb - q1) / max(q3 - q1, 1e-9), 3) signal_by_symbol[sym] = { "side": side, "score": score, "confidence": "medium", "combined_score": comb, "regime": "risk-off" if stress > 0.4 else "normal", "regime_stress": round(stress, 3), } except Exception: # no dashboard -> fall back to neutral for all pass merged = [] for factor in factor_view.get("factors", []): sig = signal_by_symbol.get(factor["symbol"], {}) merged.append( { **factor, "signal_side": sig.get("side"), "signal_score": sig.get("score"), "signal_confidence": sig.get("confidence"), "combined_score": sig.get("combined_score"), "regime": sig.get("regime"), "signal_target_weight": None, "reason_codes": ([ f"quartile({'LONG' if sig.get('side')=='LONG' else 'SHORT'} 25%), regime={sig.get('regime') or 'normal'}" ] if sig.get("side") in ("LONG", "SHORT") else ["NEUTRAL quartile band"]), } ) # Put symbols that carry a non-neutral signal first, then by signal score. merged.sort( key=lambda f: ( 0 if f.get("signal_side") in ("LONG", "SHORT") else 1, -(abs(f.get("signal_score") or 0)), f["symbol"], ) ) return jsonify( { "available": factor_view.get("available", False), "as_of": factor_view.get("as_of"), "source": factor_view.get("source"), "factor_count": factor_view.get("factor_count", 0), "dividend_count": factor_view.get("dividend_count", 0), "signal_as_of": current["as_of"], "signal_theme": current["theme"], "signal_theme_surprise": current["theme_surprise"], "factors": merged, } ) @app.route("/api/v1/simulation", methods=["POST"]) def simulation(): """Capital-allocation simulation (paper/backtest; never a real order). Body: {capital: float, mode: "backtest"|"forward"}. Uses the SAME canonical combined score as /api/v1/dashboard (via `default_scores`) — not a separate 3-theme recompute — so the "จำลอง" allocation can never disagree with the board on which names rank highest. Prices come from the latest Yahoo snapshot; allocation uses the 50/20/30 dividend buckets. Output is paper/backtest on revised history (non-PIT). """ from app import simulation as sim payload = request.get_json(silent=True) or {} try: capital = float(payload.get("capital", 0)) except (TypeError, ValueError): return jsonify({"error": "capital must be a number"}), 400 mode = payload.get("mode", "backtest") in ("backtest", "forward") and payload.get("mode", "backtest") if capital <= 0: return jsonify({"error": "capital must be > 0"}), 400 try: series = sim.load_price_snapshot() prices = sim.latest_prices(series) except (sim.SimulationError, OSError) as exc: return jsonify({"error": f"price snapshot: {exc}"}), 503 # Single source of truth: the live multi-theme board (combined 60/40 + # quality + momentum + dividend screen), identical to /api/v1/dashboard. from app.dashboard import default_scores try: score_by_symbol = default_scores(None) except Exception: # No collector detail in the response (avoid leaking internal state). return jsonify({"error": "dashboard scores unavailable"}), 503 candidates = [] for sym, meta in score_by_symbol.items(): price = prices.get(sym) if price is None: continue candidates.append( sim.Candidate( symbol=sym, price=price, combined_score=meta.get("combined", 0.0), is_dividend=bool(meta.get("is_dividend")), dividend_yield=float(meta.get("dividend_yield") or 0.0), ) ) try: result = sim.allocate_capital(capital, candidates) except sim.SimulationError as exc: return jsonify({"error": str(exc)}), 400 from app import daily_cache current = app.extensions.get("tourism_result") or {} return jsonify( { "mode": mode, "capital": capital, "as_of": current.get("as_of"), "data_note": "paper/backtest on revised vendor history (non-PIT) — not validated evidence", **result.to_dict(), } ) @app.get("/api/v1/themes") def themes(): """Multi-theme combined board (delegates to the canonical dashboard). Single source: RealDashboard.build() so /api/v1/themes returns the SAME 13-theme set, labels, surprises, and per-symbol combined board as /api/v1/dashboard. Removes the old 3-theme duplicated logic. """ from app.dashboard import RealDashboard, DashboardError from app import daily_cache cache = app.extensions.setdefault("daily_cache", daily_cache.DailyCache()) current = app.extensions.get("tourism_result") or {} try: dash = RealDashboard(current.get("signals", []), cache).build() except DashboardError as exc: return jsonify({"error": str(exc)}), 503 return jsonify({ "themes": dash["themes"], "as_of": dash.get("as_of", ""), "combined_count": len(dash["board"]), "board": dash["board"], "macro": dash.get("macro", {}), }) @app.get("/api/v1/symbols/") 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, momentum=themes_mod._load_momentum(), ) return jsonify(detail) @app.post("/api/v1/backtest") def run_backtest_endpoint(): """Run a real backtest over [start, end] with capital; persist result.""" from app.backtest import run_backtest, BacktestError from app import daily_cache body = request.get_json(silent=True) or {} start = body.get("start") or "2024-06-01" end = body.get("end") or "2026-06-01" capital = float(body.get("capital") or 1_000_000) freq = body.get("freq") or "monthly" try: res = run_backtest(start, end, capital=capital, rebalance_freq=freq) except BacktestError as exc: return jsonify({"error": str(exc)}), 400 runs = app.extensions.setdefault("backtest_runs", []) record = res.to_dict() record["id"] = len(runs) + 1 record["ran_at"] = __import__("datetime").datetime.now( __import__("datetime").timezone.utc).isoformat(timespec="minutes") runs.append(record) return jsonify(record) @app.get("/api/v1/backtest/runs") def backtest_runs(): runs = app.extensions.get("backtest_runs", []) return jsonify({"runs": runs}) @app.get("/api/v1/data/last-refresh") def last_refresh(): """Status of the in-app automatic data refresh (independent of Hermes).""" interval = os.getenv("REFRESH_INTERVAL_SECONDS", "3600") marker_path = Path(__file__).resolve().parents[1] / "data" / "scheduler" / "last_refresh.json" last = None if marker_path.exists(): try: last = json.loads(marker_path.read_text(encoding="utf-8")) except Exception: last = None return jsonify({ "automatic_refresh": True, "interval_seconds": int(interval), "interval_label": f"ทุก {int(interval)//3600} ชั่วโมง" if int(interval) >= 3600 else f"ทุก {int(interval)//60} นาที", "last_refresh": last, }) @app.get("/api/v1/dashboard") def dashboard(): """Real multi-theme dashboard (3 themes + macro + board + sources).""" from app.dashboard import RealDashboard, DashboardError from app import daily_cache cache = app.extensions.setdefault("daily_cache", daily_cache.DailyCache()) current = app.extensions.get("tourism_result") tourism_signals = (current or {}).get("signals", []) try: dash = RealDashboard(tourism_signals, cache).build() except DashboardError as exc: return jsonify({"error": str(exc), "available": False}), 503 return jsonify({"available": True, **dash}) @app.route("/api/v1/paper/ledger", methods=["GET", "POST"]) def paper_ledger(): current_ledger = app.extensions["paper_ledger"] if request.method == "GET": return jsonify({"mode": "paper", "entries": current_ledger.entries()}) authorized, error = _paper_write_authorized() if not authorized: body, status = error return jsonify(body), status payload = request.get_json(silent=True) if not isinstance(payload, dict): return jsonify({"error": "JSON object required"}), 400 try: entry = current_ledger.record(payload, app.extensions["allowed_symbols"]) except ValueError as exc: return jsonify({"error": str(exc)}), 400 return jsonify({"mode": "paper", "entry": entry}), 201 return app