"""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 _load_siamchart_snapshot() -> dict[str, Any]: """Load the latest Siamchart SET50 fundamental snapshot (raw dict with ``rows`` + ``details`` + ``retrieved_at``) or {} if absent/malformed. Used by the PIT scorer: its EPS 5-year series and per-symbol ratios feed the fundamental (40%) dimension. Absent snapshot -> {} (the PIT provider then falls back to the current board for fundamental, flagged partial).""" path = Path(__file__).resolve().parents[1] / "data" / "siamchart" / "set50_master.json" if not path.is_file(): return {} try: data = json.loads(path.read_text(encoding="utf-8")) except (OSError, ValueError): return {} if not isinstance(data, dict): return {} return data def _seed_siamchart_vintage(store) -> None: """Persist the current on-disk siamchart snapshot as the first vintage, if the store is empty. Intentionally idempotent (store.deduplicates by retrieved_at + body).""" try: from .siamchart_vintages import SiamchartVintageStore if isinstance(store, SiamchartVintageStore) and not store.list_ids(): snap = _load_siamchart_snapshot() if snap: store.persist(snap) except Exception: # noqa: BLE001 — seeding must never break the route pass 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 from .dividend_ledger import DividendLedger app.extensions["dividend_ledger"] = DividendLedger( app.config.get("DIVIDEND_LEDGER_PATH") or (data_root / "dividends" / "ledger.json") ) from .backtest_store import BacktestStore app.extensions["backtest_store"] = BacktestStore( app.config.get("BACKTEST_STORE_PATH") or (data_root / "backtest" / "runs.json") ) # 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")) div_cooldown_s = int(os.getenv("DIVIDEND_REFRESH_COOLDOWN_SECONDS", str(6 * 3600))) scheduler = AppDataScheduler(cache, Path(__file__).resolve().parents[1] / "data", interval_seconds=interval_s, dividend_cooldown_seconds=div_cooldown_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/suggestion", methods=["POST"]) def suggestion(): """Capital-allocation suggestion (non-PIT live recommendation; paper). Body: {capital: float}. Uses the SAME canonical combined score as /api/v1/dashboard (via `default_scores`) — not a separate 3-theme recompute — so the "จัดสรรทุน" (suggestion) 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. This is a live recommendation on current data (non-PIT), never a real order. """ 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 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 # Owner's "ทำกำไร" = price likely to rise in 3-6 months -> price-trend # momentum (NOT EPS growth). theme_signal = mean surprise of the themes a # symbol belongs to; buckets 1/2 require theme_signal > 0. try: from app import themes as _th trend = _th.price_trend_score(series) except Exception: trend = {} theme_sig_by_sym: dict[str, float] = {} try: from app.dashboard import RealDashboard from app import daily_cache as _dc _cache = app.extensions.setdefault("daily_cache", _dc.DailyCache()) dash = RealDashboard((app.extensions.get("tourism_result") or {}).get("signals", []), _cache).build() except Exception: dash = {} theme_surprises = {t.get("id"): t.get("surprise") for t in dash.get("themes", [])} board_by_sym = {r.get("symbol"): r for r in dash.get("board", [])} for sym in score_by_symbol: ths = [s for s in (board_by_sym.get(sym, {}).get("themes") or []) if theme_surprises.get(s) is not None] theme_sig_by_sym[sym] = ( (sum(theme_surprises[s] for s in ths) / len(ths)) if ths else 0.0 ) 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), momentum=float(trend.get(sym, 0.0)), theme_signal=float(theme_sig_by_sym.get(sym, 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": "suggestion", "capital": capital, "as_of": current.get("as_of"), "data_note": "suggestion on current board (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") dash = {} 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(), ) # per-source factor-level audit: for each theme this symbol belongs to, # show source -> raw -> normalized -> weight -> contribution so the owner # sees exactly how each source scored and how the weights were applied. fetch_data = dash.get("fetch_data", {}) detail["factor_sources"] = { tid: themes_mod.factor_source_breakdown(fetch_data, tid) for tid in detail.get("themes", []) } # fallback: if the dashboard fetch was empty (degraded), rebuild it once if not fetch_data: try: from app import daily_cache as _dc cache2 = _dc.DailyCache() dash2 = RealDashboard((current or {}).get("signals", []), cache2).build() fetch_data2 = dash2.get("fetch_data", {}) detail["factor_sources"] = { tid: themes_mod.factor_source_breakdown(fetch_data2, tid) for tid in detail.get("themes", []) } except Exception: pass return jsonify(detail) @app.get("/api/v1/backtest/readiness") def backtest_readiness_endpoint(): """Strict PIT backtest readiness + recommended default dates. The backtest must not start before advice is genuinely available. This endpoint reports whether the PIT stores (factor vintages + Siamchart vintage manifest) plus price data cover any usable [start, end] window, and returns the recommended default start/end the UI should prefill. It fails closed (ready=false + missing list) when coverage is absent. """ from pathlib import Path as _Path from .backtest_readiness import evaluate_readiness from .factor_vintages import FactorVintageStore from .siamchart_vintages import SiamchartVintageStore from .simulation import load_price_snapshot, SimulationError data_root = _Path(__file__).resolve().parents[1] / "data" fstore = FactorVintageStore(data_root) sstore = SiamchartVintageStore(data_root) try: series = load_price_snapshot() except SimulationError: series = None start = request.args.get("start") end = request.args.get("end") res = evaluate_readiness( factor_store=fstore, siamchart_store=sstore, price_series=series, start=start, end=end, ) return jsonify(res.to_dict()) @app.post("/api/v1/backtest/run") def backtest_run_endpoint(): """Run a strict, event-driven PIT backtest and persist the result. Fails closed on readiness: if the requested window (or the derived default window) is not fully PIT-covered, returns 400 listing the missing inputs rather than fabricating an invalid run. """ from .backtest_readiness import evaluate_readiness from .backtest_engine import run_event_backtest from .factor_vintages import FactorVintageStore from .siamchart_vintages import SiamchartVintageStore from .simulation import load_price_snapshot from .pit_scorer import PitScoreProvider, make_pit_score_fn from pathlib import Path as _Path body = request.get_json(silent=True) or {} data_root = _Path(__file__).resolve().parents[1] / "data" fstore = FactorVintageStore(data_root) sstore = SiamchartVintageStore(data_root) price_series = load_price_snapshot() # readiness gate (starts below fail closed) start = body.get("start") end = body.get("end") ready = evaluate_readiness( factor_store=fstore, siamchart_store=sstore, price_series=price_series, start=start, end=end, ) if not ready.ready: return jsonify({ "error": "not enough PIT coverage for this backtest window", "missing": ready.missing, "recommended_start": ready.recommended_start, "recommended_end": ready.recommended_end, "coverage": ready.coverage, }), 400 start = start or ready.recommended_start end = end or ready.recommended_end assert start is not None and end is not None # readiness guarantees this capital = float(body.get("capital") or 1_000_000) # seed siamchart vintage baseline + PIT scorer _seed_siamchart_vintage(sstore) provider = PitScoreProvider(fstore, _load_siamchart_snapshot(), siamchart_store=sstore) score_fn = make_pit_score_fn(provider) use_ledger = bool(body.get("use_ledger")) ledger = None if use_ledger: stored = app.extensions.get("dividend_ledger") if stored is not None and stored.symbols(): ledger = stored else: from .dividend_ledger import build_dps_ledger ledger = build_dps_ledger(_load_siamchart_snapshot()) res = run_event_backtest( start=start, end=end, capital=capital, factor_store=fstore, siamchart_store=sstore, dividend_ledger=ledger, price_series=price_series, score_fn=score_fn, ) record = res.to_dict() record["use_ledger"] = use_ledger persisted = app.extensions["backtest_store"].add(record) return jsonify(persisted), 201 @app.get("/api/v1/backtest/run/") def backtest_run_get(run_id: str): store = app.extensions["backtest_store"] rec = store.get(run_id) if rec is None: return jsonify({"error": "run not found"}), 404 return jsonify(rec) @app.get("/api/v1/backtest/run") def backtest_runs_direct(): store = app.extensions["backtest_store"] return jsonify({"runs": store.all()}) @app.get("/api/v1/scheduler/sources") def scheduler_sources(): """Source-health log for the data scheduler (refresh cadence + failures). Lets the frontend show each source's last outcome and copy the failure detail for diagnosis (e.g. a source that changed its page structure is categorized as `structure`, a network blip as `network`). Empty when no scheduler is wired (e.g. TESTING or a read-only host) or no tick yet. """ sched = app.extensions.get("data_scheduler") if sched is None: return jsonify({"sources": []}) limit = int(request.args.get("limit", "200")) return jsonify({"sources": sched._load_source_health(limit)}) @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" use_pit = bool(body.get("use_pit")) use_ledger = bool(body.get("use_ledger")) try: ledger = None if use_ledger: from .dividend_ledger import build_dps_ledger # prefer the persisted dated ledger (real ex-date history, if # populated via /api/v1/dividends/update); otherwise fall back # to DPS estimates built from the current snapshot. stored = app.extensions.get("dividend_ledger") if stored is not None and stored.symbols(): ledger = stored else: ledger = build_dps_ledger(_load_siamchart_snapshot()) if use_pit: from pathlib import Path as _Path from .factor_vintages import FactorVintageStore from .siamchart_vintages import SiamchartVintageStore from .pit_scorer import PitScoreProvider, make_pit_score_fn froot = _Path(__file__).resolve().parents[1] / "data" store = FactorVintageStore(froot) sstore = SiamchartVintageStore(froot) # seed the first vintage from the current snapshot so the store # has a known baseline (no-op if already stored). _seed_siamchart_vintage(sstore) provider = PitScoreProvider(store, _load_siamchart_snapshot(), siamchart_store=sstore) score_fn = make_pit_score_fn(provider) res = run_backtest(start, end, capital=capital, rebalance_freq=freq, score_fn=score_fn, dividend_ledger=ledger) else: res = run_backtest(start, end, capital=capital, rebalance_freq=freq, dividend_ledger=ledger) 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.post("/api/v1/dividends/update") def dividends_update(): """Fetch real dated dividend history for every snapshot symbol and persist a dated dividend ledger. Uses ``siamchart.fetch_dividend_history`` for each symbol (ex-date + per-share DPS). Populates the on-disk ledger so ``use_ledger`` backtests credit REAL dated cash flows (dividend_method=dated_ledger) instead of DPS estimates. Returns per-symbol row counts. """ from .siamchart import fetch_dividend_history from .dividend_ledger import populate_dated_dividends from .siamchart_factors import build_factor_view ledger = app.extensions["dividend_ledger"] fv = build_factor_view() symbols = [f["symbol"] for f in fv.get("factors", [])] counts = populate_dated_dividends(ledger, symbols, fetch_dividend_history) ledger.save() total = sum(counts.values()) return jsonify({ "updated_symbols": len([s for s, n in counts.items() if n > 0]), "symbols": len(symbols), "total_payments": total, "counts": counts, "note": "dated_ledger now used by use_ledger backtests", }) @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.get("/api/v1/learning/momentum") def learning_momentum(): """P4 factor-weight learning report for the 12-1 momentum factor. Runs a strictly point-in-time IC analysis over the price history: does 12-1 momentum at month t predict 3m forward returns across the cross-section? Applies a split-sample holdout validation gate — reports mean/train/holdout IC, t-stat, n periods, `validated`, why not (gate_notes), and a suggested weight. The learned weight is NEVER auto-applied: `validated=false` keeps the weight unchanged. Honest: 503 when no price snapshot exists. Macro/demographic factors are not yet attributable (no vintages). """ from app import weight_learning as wl from app import simulation as sim start = request.args.get("start") or "2024-06-01" end = request.args.get("end") or "2026-06-01" try: series = sim.load_price_snapshot() except (sim.SimulationError, OSError) as exc: return jsonify({"error": f"price snapshot: {exc}"}), 503 symbols = sorted(series.keys()) try: learning = wl.learn_momentum_gated(series, symbols, start, end) except (wl.WeightLearningError, ValueError) as exc: return jsonify({"error": str(exc)}), 400 return jsonify({"factor": learning.to_dict(), "window": {"start": start, "end": end}}) @app.get("/api/v1/learning/factors") def learning_factors(): """P4 dataset-readiness: how many historical points each factor has. Factors with >= `min_points` historical observations are learnable (they accumulate automatically each scheduler run, starting now). Macro/demographic factors start at 0 and become learnable over time. """ from app import factors as factors_mod from app.factor_history import FactorHistory hist_dir = app.config.get("FACTOR_HISTORY_DIR") or ( Path(__file__).resolve().parents[1] / "data" / "factor_history") fh = FactorHistory(hist_dir) try: min_points = max(int(request.args.get("min_points", "12")), 0) except (TypeError, ValueError): return jsonify({"error": "min_points must be a non-negative integer"}), 400 rows = [] for fkey, fact in factors_mod.FACTORS.items(): series = fh.series(fkey) rows.append({ "factor_key": fkey, "name_th": fact.get("name_th", fkey), "source": fact.get("source"), "frequency": fact.get("frequency"), "n_points": len(series), "learnable": len(series) >= min_points, "last_value": series[-1]["value"] if series else None, "last_as_of": series[-1].get("as_of", "") if series else "", }) rows.sort(key=lambda r: (-r["n_points"], r["factor_key"])) return jsonify({"min_points": min_points, "factors": rows}) @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