diff --git "a/backend/main.py" "b/backend/main.py" --- "a/backend/main.py" +++ "b/backend/main.py" @@ -82,6 +82,7 @@ import yfinance as yf from backend.api_models import SwitchRequest, WatchlistRequest from backend.cache_utils import PersistentCache, TTLCache from backend.catalog_utils import build_symbols_catalog, filter_symbols_catalog +from backend.frontend_assets import register_frontend_assets from backend.market_utils import MARKET_SESSIONS as MARKET_SESSIONS_DATA from backend.market_utils import market_status_now as compute_market_status_now from backend.observability_utils import ( @@ -95,6 +96,26 @@ from backend.runtime_utils import ( parse_cors_origins, resolve_runtime_paths, ) +from backend.kronos_adapter import ( + KRONOS_AVAILABLE, + KRONOS_IMPORT_ERROR, + KronosForecaster, +) +from backend.forecasting import ( + DEFAULT_FORECAST_MODEL_SELECTION, + FORECAST_MODEL_ORDER, + FORECAST_RULE_DOCUMENT, + build_query_model_selection, + forecast_model_signature, +) +from backend.forecasting.providers import ( + CHRONOS_AVAILABLE, + CHRONOS_IMPORT_ERROR, + ChronosForecaster as ChronosProviderForecaster, + TIMESFM_AVAILABLE, + TIMESFM_IMPORT_ERROR, + TimesFMForecaster as TimesFMProviderForecaster, +) from backend.security_utils import ( RATE_LIMIT_WHITELIST, admin_only as require_admin_token, @@ -107,6 +128,7 @@ from backend.startup_utils import ( build_source_selftest_urls, clear_stale_ip_limits, run_source_selftest, + warmup_forecaster, ) from backend.symbol_utils import ( assemble_market_peer_payload, @@ -117,7 +139,7 @@ from contextlib import asynccontextmanager from fastapi import FastAPI, HTTPException, Query, WebSocket, WebSocketDisconnect, Request from fastapi.websockets import WebSocketState from fastapi.middleware.cors import CORSMiddleware -from fastapi.staticfiles import StaticFiles +from fastapi.responses import JSONResponse from pydantic import BaseModel try: @@ -129,6 +151,27 @@ except Exception as exc: DEFAULT_ADMIN_TOKEN = "aiforecast_v6_default_secret" + +def _read_bool_env(*keys: str, default: bool) -> bool: + """Read the first defined boolean-like environment variable.""" + false_values = {"0", "false", "no", "off"} + true_values = {"1", "true", "yes", "on"} + + for key in keys: + raw_value = os.getenv(key) + if raw_value is None: + continue + + normalized = raw_value.strip().lower() + if not normalized: + continue + if normalized in true_values: + return True + if normalized in false_values: + return False + + return default + # ────────────────────────────────────────────────────────────────────────────── # Architecture v5: Configuration Management (C-1) # ────────────────────────────────────────────────────────────────────────────── @@ -149,7 +192,21 @@ class Settings(BaseModel): # App Config host: str = os.getenv("HOST", "0.0.0.0") port: int = int(os.getenv("PORT", 8000)) - preload_timesfm: bool = os.getenv("PRELOAD_TIMESFM", "True").lower() == "true" + preload_timesfm: bool = _read_bool_env( + "PRELOAD_TIMESFM", + "TIMESFM_PRELOAD", + default=True, + ) + preload_kronos: bool = _read_bool_env( + "PRELOAD_KRONOS", + "KRONOS_PRELOAD", + default=True, + ) + preload_chronos: bool = _read_bool_env( + "PRELOAD_CHRONOS", + "CHRONOS_PRELOAD", + default=True, + ) cache_version: str = os.getenv("CACHE_VERSION", "v12") cors_allow_origins_raw: str = os.getenv("CORS_ALLOW_ORIGINS", "*") @@ -397,7 +454,7 @@ logging.basicConfig( level=logging.INFO, format="%(asctime)s | %(levelname)s | %(name)s | %(message)s", ) -logger = logging.getLogger("ai-trading-chart") +logger = logging.getLogger("ai-forecast") # Avoid leaking query-string API keys from httpx/httpcore INFO logs. logging.getLogger("httpx").setLevel(logging.WARNING) logging.getLogger("httpcore").setLevel(logging.WARNING) @@ -416,24 +473,24 @@ if IS_FROZEN: else: logger.info("Running in DEV mode. PROJECT_ROOT: %s", PROJECT_ROOT) -# TimesFM 2.5 — foundation model for OHLC4 forecasting -TIMESFM_AVAILABLE = False -TIMESFM_IMPORT_ERROR: Optional[str] = None +if TIMESFM_AVAILABLE: + logger.info("TimesFM library imported successfully") +elif TIMESFM_IMPORT_ERROR: + logger.warning("TimesFM unavailable: %s", TIMESFM_IMPORT_ERROR) -if TORCH_IMPORT_ERROR: - TIMESFM_IMPORT_ERROR = TORCH_IMPORT_ERROR - logger.warning("TimesFM disabled because torch is unavailable: %s", TORCH_IMPORT_ERROR) -else: - try: - import timesfm # pip install timesfm[torch] - TIMESFM_AVAILABLE = True - logger.info("TimesFM library imported successfully") - except Exception as _tfm_ex: - TIMESFM_IMPORT_ERROR = str(_tfm_ex) - logger.error("TimesFM import error: %s", _tfm_ex) - logger.warning("TimesFM not installed — forecasting disabled. Run: pip install timesfm[torch]") - -PRELOAD_TIMESFM = os.getenv("TIMESFM_PRELOAD", "1").strip().lower() not in {"0", "false", "no"} +if KRONOS_AVAILABLE: + logger.info("Kronos library imported successfully") +elif KRONOS_IMPORT_ERROR: + logger.warning("Kronos unavailable: %s", KRONOS_IMPORT_ERROR) + +if CHRONOS_AVAILABLE: + logger.info("Chronos library imported successfully") +elif CHRONOS_IMPORT_ERROR: + logger.warning("Chronos unavailable: %s", CHRONOS_IMPORT_ERROR) + +PRELOAD_TIMESFM = settings.preload_timesfm +PRELOAD_KRONOS = settings.preload_kronos +PRELOAD_CHRONOS = settings.preload_chronos STARTUP_STATE: Dict[str, Any] = { "timesfm": { "available": TIMESFM_AVAILABLE, @@ -444,6 +501,25 @@ STARTUP_STATE: Dict[str, Any] = { "last_error": TIMESFM_IMPORT_ERROR, "model": "google/timesfm-2.5-200m-pytorch", }, + "kronos": { + "available": KRONOS_AVAILABLE, + "preload_enabled": PRELOAD_KRONOS, + "warming": False, + "loaded": False, + "device": "not_loaded", + "last_error": KRONOS_IMPORT_ERROR, + "model": os.getenv("KRONOS_MODEL_HF_ID", "NeoQuasar/Kronos-base"), + "tokenizer": "NeoQuasar/Kronos-Tokenizer-base", + }, + "chronos": { + "available": CHRONOS_AVAILABLE, + "preload_enabled": PRELOAD_CHRONOS, + "warming": False, + "loaded": False, + "device": "not_loaded", + "last_error": CHRONOS_IMPORT_ERROR, + "model": os.getenv("CHRONOS_MODEL_HF_ID", ChronosProviderForecaster.MODEL_HF_ID), + }, "sources": {}, } @@ -594,6 +670,49 @@ CATEGORY_SOURCE_PRIORITY: Dict[str, List[str]] = { } # Fallback for unknown categories DEFAULT_SOURCE_PRIORITY: List[str] = ["binance", "yfinance", "twelvedata", "finnhub"] +SYMBOL_SOURCE_PRIORITY_OVERRIDES: Dict[str, List[str]] = { + "XAUUSD": ["binance", "twelvedata", "yfinance", "finnhub"], + "XAGUSD": ["twelvedata", "yfinance", "binance", "finnhub"], + "XPTUSD": ["twelvedata", "yfinance", "binance", "finnhub"], + "XPDUSD": ["twelvedata", "yfinance", "binance", "finnhub"], +} +DEFAULT_FORECAST_MODEL_CONTEXT = 256 +FORECAST_MODEL_CONTEXT_BY_INTERVAL: Dict[str, int] = { + "1m": 256, + "5m": 256, + "15m": 256, + "1h": 256, + "4h": 256, + "1d": 256, + "1w": 192, +} +FORECAST_CONTEXT_CANDIDATES_BY_INTERVAL: Dict[str, Tuple[int, ...]] = { + "1m": (128, 192, 256), + "5m": (96, 128, 256), + "15m": (96, 128, 256), + "1h": (96, 128, 256), + "4h": (64, 96, 128, 256), + "1d": (48, 64, 128, 256), + "1w": (48, 64, 128, 192), +} +KRONOS_CONTEXT_CANDIDATES_BY_INTERVAL: Dict[str, Tuple[int, ...]] = { + "1m": (96, 128, 192, 256), + "5m": (96, 128, 192, 256), + "15m": (96, 128, 192, 256), + "1h": (96, 128, 192, 256), + "4h": (64, 96, 128, 192, 256), + "1d": (64, 128, 256, 384), + "1w": (64, 128, 256, 384), +} +KRONOS_CONTEXT_CANDIDATES_BY_INTERVAL_CPU: Dict[str, Tuple[int, ...]] = { + "1m": (192,), + "5m": (192,), + "15m": (192,), + "1h": (192,), + "4h": (256,), + "1d": (384,), + "1w": (384,), +} CLIP_DEFAULT = 3.0 INDICATOR_MIN_CONTEXT = 240 @@ -604,6 +723,61 @@ WATCHLIST_TICKER_CONCURRENCY = 12 MARKET_PEER_TICKER_CONCURRENCY = 8 +def _resolve_forecast_context_window(interval: str, available_history: int) -> int: + target_context = FORECAST_MODEL_CONTEXT_BY_INTERVAL.get( + interval, + DEFAULT_FORECAST_MODEL_CONTEXT, + ) + return max(1, min(available_history, target_context)) + + +def _resolve_forecast_context_candidates(interval: str, available_history: int) -> List[int]: + context_cap = _resolve_forecast_context_window(interval, available_history) + raw_candidates = list( + FORECAST_CONTEXT_CANDIDATES_BY_INTERVAL.get(interval, (context_cap,)) + ) + raw_candidates.append(context_cap) + + candidates: List[int] = [] + for candidate in raw_candidates: + bounded = max(32, min(available_history, int(candidate))) + if bounded not in candidates: + candidates.append(bounded) + return sorted(candidates) + + +def _resolve_kronos_context_candidates(interval: str, available_history: int) -> List[int]: + context_cap = _resolve_model_context_cap("kronos", interval, available_history) + cpu_only = torch is None or not torch.cuda.is_available() + raw_candidates = list(KRONOS_CONTEXT_CANDIDATES_BY_INTERVAL.get(interval, (context_cap,))) + if cpu_only: + raw_candidates.extend(KRONOS_CONTEXT_CANDIDATES_BY_INTERVAL_CPU.get(interval, ())) + raw_candidates.append(context_cap) + + candidates: List[int] = [] + for candidate in raw_candidates: + bounded = max(32, min(context_cap, int(candidate))) + if bounded not in candidates: + candidates.append(bounded) + return sorted(candidates) + + +def _resolve_model_context_candidates(model_key: str, interval: str, available_history: int) -> List[int]: + if model_key == "kronos": + return _resolve_kronos_context_candidates(interval, available_history) + return _resolve_forecast_context_candidates(interval, available_history) + + +def _resolve_model_context_cap(model_key: str, interval: str, available_history: int) -> int: + if model_key == "kronos": + kronos_cap = 256 + return max(1, min(available_history, kronos_cap, KronosForecaster.MAX_CONTEXT)) + if model_key == "chronos": + chronos_cap = ChronosProviderForecaster.MAX_CONTEXT + return max(1, min(available_history, chronos_cap)) + return _resolve_forecast_context_window(interval, available_history) + + # ────────────────────────────────────────────────────────────────────────────── # Rate Limiter (token-bucket, per source) # ────────────────────────────────────────────────────────────────────────────── @@ -2018,7 +2192,10 @@ async def _build_synthetic_symbol_history( def _get_source_priority(symbol: str, interval: Optional[str] = None) -> List[str]: cfg = SYMBOLS[symbol] - priority = CATEGORY_SOURCE_PRIORITY.get(cfg.category, DEFAULT_SOURCE_PRIORITY) + priority = SYMBOL_SOURCE_PRIORITY_OVERRIDES.get( + symbol, + CATEGORY_SOURCE_PRIORITY.get(cfg.category, DEFAULT_SOURCE_PRIORITY), + ) if settings.is_hf and cfg.category == "Crypto": priority = ( HF_CRYPTO_SOURCE_PRIORITY_DAILY @@ -2094,7 +2271,7 @@ async def fetch_historical( """ Fetch OHLCV data with fallback and caching. v6.2: Fetch only the context each caller actually needs while keeping - AI routes supplied with deeper history for indicator and Kronos quality. + AI routes supplied with deeper history for indicator and forecast quality. """ prefix = _cache_prefix(symbol, interval) key = f"hist_{prefix}" # BUG-P1-03: No limit in key to increase cache hits @@ -3108,7 +3285,12 @@ def _blend_forecasts( clip_hi = 3.0 if last_close_magnitude < -4 else 2.0 scale = float(np.clip(scale, clip_lo, clip_hi)) if abs(scale - 1.0) > 0.3: - logger.warning("[Kronos] High scale correction: %.4f (first_model=%.8f, last=%.8f)", scale, first_model_est, last_close) + logger.warning( + "[TimesFM] High scale correction: %.4f (first_model=%.8f, last=%.8f)", + scale, + first_model_est, + last_close, + ) model_p10 = raw_p10 * scale model_p50 = raw_p50 * scale @@ -3777,6 +3959,570 @@ def _forecast_path_metrics(p50_path: np.ndarray, last_close: float) -> Dict[str, } +def _forecast_path_responsiveness(p50_path: np.ndarray, last_close: float) -> Dict[str, float]: + """Quantify how much the forecast path actually moves relative to the anchor.""" + if len(p50_path) == 0 or abs(last_close) <= 1e-8: + return { + "range_pct": 0.0, + "step_abs_mean_pct": 0.0, + "step_abs_max_pct": 0.0, + } + + ret_path = ((p50_path / last_close) - 1.0) * 100.0 + diffs = np.diff(ret_path) if len(ret_path) > 1 else np.array([ret_path[-1]], dtype=float) + return { + "range_pct": round(float(np.max(ret_path) - np.min(ret_path)), 4), + "step_abs_mean_pct": round(float(np.mean(np.abs(diffs))), 4), + "step_abs_max_pct": round(float(np.max(np.abs(diffs))), 4), + } + + +def _recent_ohlc4_abs_step_pct(ohlc4: np.ndarray, window: int = 20) -> float: + """Use recent OHLC4 movement as a volatility-aware floor for forecast responsiveness.""" + if len(ohlc4) < 2: + return 0.0 + + base = np.maximum(np.abs(ohlc4[:-1]), 1e-8) + returns = np.diff(ohlc4) / base * 100.0 + recent = returns[-min(window, len(returns)) :] + return round(float(np.mean(np.abs(recent))), 4) if len(recent) else 0.0 + + +def _future_window_path_stats( + ohlc4: np.ndarray, + anchor_index: int, + horizon: int, +) -> Optional[Dict[str, float]]: + """Measure the realized future path that followed a historical anchor point.""" + if anchor_index < 0 or horizon <= 0: + return None + if anchor_index + horizon >= len(ohlc4): + return None + + baseline = float(ohlc4[anchor_index]) + future_path = np.asarray(ohlc4[anchor_index + 1 : anchor_index + 1 + horizon], dtype=float) + responsiveness = _forecast_path_responsiveness(future_path, baseline) + final_return_pct = ( + ((future_path[-1] / max(abs(baseline), 1e-8)) - 1.0) * 100.0 + if len(future_path) + else 0.0 + ) + return { + "range_pct": float(responsiveness["range_pct"]), + "step_abs_mean_pct": float(responsiveness["step_abs_mean_pct"]), + "step_abs_max_pct": float(responsiveness["step_abs_max_pct"]), + "final_return_pct": float(final_return_pct), + } + + +def _rolling_future_path_targets( + ohlc4: np.ndarray, + horizon: int, + max_windows: int = 48, +) -> Optional[Dict[str, float]]: + """Summarize realized future-path amplitudes from recent historical anchors.""" + if len(ohlc4) < horizon + 2: + return None + + last_anchor = len(ohlc4) - horizon - 1 + start_anchor = max(0, last_anchor - max_windows + 1) + stats: List[Dict[str, float]] = [] + for anchor_index in range(start_anchor, last_anchor + 1): + path_stats = _future_window_path_stats(ohlc4, anchor_index, horizon) + if path_stats is not None: + stats.append(path_stats) + + if not stats: + return None + + weights = np.linspace(0.6, 1.0, len(stats), dtype=float) + range_vals = np.array([item["range_pct"] for item in stats], dtype=float) + step_vals = np.array([item["step_abs_mean_pct"] for item in stats], dtype=float) + return { + "window_count": len(stats), + "range_pct": round(float(np.average(range_vals, weights=weights)), 4), + "step_abs_mean_pct": round(float(np.average(step_vals, weights=weights)), 4), + } + + +def _context_regime_signature(ohlc4_window: np.ndarray) -> Dict[str, float]: + """Describe the local market regime of an OHLC4 window using scale-free features.""" + if len(ohlc4_window) < 2: + return { + "net_change_pct": 0.0, + "vol_pct": 0.0, + "abs_step_mean_pct": 0.0, + "range_pct": 0.0, + } + + base = np.maximum(np.abs(ohlc4_window[:-1]), 1e-8) + returns = np.diff(ohlc4_window) / base * 100.0 + first = max(abs(float(ohlc4_window[0])), 1e-8) + return { + "net_change_pct": float(((ohlc4_window[-1] / first) - 1.0) * 100.0), + "vol_pct": float(np.std(returns)) if len(returns) > 1 else 0.0, + "abs_step_mean_pct": float(np.mean(np.abs(returns))) if len(returns) else 0.0, + "range_pct": float(((np.max(ohlc4_window) - np.min(ohlc4_window)) / first) * 100.0), + } + + +def _regime_signature_distance(current: Dict[str, float], candidate: Dict[str, float]) -> float: + """Weighted distance between two market-regime signatures.""" + terms = ( + ("net_change_pct", 0.34, 0.40), + ("vol_pct", 0.28, 0.18), + ("abs_step_mean_pct", 0.22, 0.12), + ("range_pct", 0.16, 0.30), + ) + distance = 0.0 + for key, weight, floor in terms: + lhs = float(current.get(key, 0.0)) + rhs = float(candidate.get(key, 0.0)) + scale = max(abs(lhs), abs(rhs), floor) + distance += weight * (abs(lhs - rhs) / scale) + return float(distance) + + +def _regime_matched_future_targets( + ohlc4: np.ndarray, + context_len: int, + horizon: int, + search_limit: int = 720, + top_k: int = 18, +) -> Optional[Dict[str, float]]: + """Find historical windows with a similar regime and measure their realized future amplitudes.""" + max_anchor = len(ohlc4) - horizon - 1 + if max_anchor < 32: + return None + + signature_len = max(32, min(int(context_len), 96, max_anchor + 1)) + if len(ohlc4) < signature_len + horizon + 1: + return None + + current_signature = _context_regime_signature( + np.asarray(ohlc4[-signature_len:], dtype=float) + ) + min_anchor = signature_len - 1 + start_anchor = max(min_anchor, max_anchor - search_limit + 1) + matches: List[Tuple[float, Dict[str, float]]] = [] + + for anchor_index in range(start_anchor, max_anchor + 1): + context_window = np.asarray( + ohlc4[anchor_index - signature_len + 1 : anchor_index + 1], + dtype=float, + ) + candidate_signature = _context_regime_signature(context_window) + path_stats = _future_window_path_stats(ohlc4, anchor_index, horizon) + if path_stats is None: + continue + distance = _regime_signature_distance(current_signature, candidate_signature) + matches.append((distance, path_stats)) + + if not matches: + return None + + matches.sort(key=lambda item: item[0]) + chosen = matches[: min(top_k, len(matches))] + weights = np.array([1.0 / (item[0] + 0.08) for item in chosen], dtype=float) + range_vals = np.array([item[1]["range_pct"] for item in chosen], dtype=float) + step_vals = np.array([item[1]["step_abs_mean_pct"] for item in chosen], dtype=float) + return { + "match_count": len(chosen), + "signature_len": signature_len, + "range_pct": round(float(np.average(range_vals, weights=weights)), 4), + "step_abs_mean_pct": round(float(np.average(step_vals, weights=weights)), 4), + } + + +def _build_regime_texture_template( + ohlc4: np.ndarray, + context_len: int, + horizon: int, + search_limit: int = 720, + top_k: int = 18, +) -> Optional[Dict[str, Any]]: + """Build a step-return template from regime-matched historical futures.""" + max_anchor = len(ohlc4) - horizon - 1 + if max_anchor < 32: + return None + + signature_len = max(32, min(int(context_len), 96, max_anchor + 1)) + if len(ohlc4) < signature_len + horizon + 1: + return None + + current_signature = _context_regime_signature( + np.asarray(ohlc4[-signature_len:], dtype=float) + ) + min_anchor = signature_len - 1 + start_anchor = max(min_anchor, max_anchor - search_limit + 1) + matches: List[Tuple[float, np.ndarray, Dict[str, float]]] = [] + + for anchor_index in range(start_anchor, max_anchor + 1): + context_window = np.asarray( + ohlc4[anchor_index - signature_len + 1 : anchor_index + 1], + dtype=float, + ) + candidate_signature = _context_regime_signature(context_window) + distance = _regime_signature_distance(current_signature, candidate_signature) + + baseline = float(ohlc4[anchor_index]) + future_path = np.asarray( + ohlc4[anchor_index + 1 : anchor_index + 1 + horizon], + dtype=float, + ) + if len(future_path) != horizon: + continue + + prev_values = np.concatenate(([baseline], future_path[:-1])) + step_returns = ((future_path / np.maximum(np.abs(prev_values), 1e-8)) - 1.0) * 100.0 + path_stats = _forecast_path_responsiveness(future_path, baseline) + matches.append((distance, step_returns, path_stats)) + + if not matches: + return None + + matches.sort(key=lambda item: item[0]) + chosen = matches[: min(top_k, len(matches))] + weights = np.array([1.0 / (distance + 0.08) for distance, _, _ in chosen], dtype=float) + step_matrix = np.vstack([step_returns for _, step_returns, _ in chosen]) + averaged_steps = np.average(step_matrix, axis=0, weights=weights) + + best_distance = float("inf") + best_steps: Optional[np.ndarray] = None + best_score = float("-inf") + for distance, step_returns, path_stats in chosen[: min(6, len(chosen))]: + step_abs_mean_pct = float(np.mean(np.abs(step_returns))) if len(step_returns) else 0.0 + range_pct = float(path_stats.get("range_pct", 0.0)) + texture_score = ( + step_abs_mean_pct * 0.60 + + range_pct * 0.40 + - (distance * 0.12) + ) + if texture_score > best_score: + best_score = texture_score + best_distance = distance + best_steps = step_returns + + if best_steps is None: + best_steps = averaged_steps + best_distance = float(chosen[0][0]) + + blended_steps = (best_steps * 0.72) + (averaged_steps * 0.28) + step_abs_mean_pct = float(np.mean(np.abs(blended_steps))) if len(blended_steps) else 0.0 + simulated_prices = [baseline] + price = baseline + for step_return in blended_steps: + price = max(0.0, price * (1.0 + (float(step_return) / 100.0))) + simulated_prices.append(price) + simulated_path = np.asarray(simulated_prices[1:], dtype=float) + range_pct = _forecast_path_responsiveness(simulated_path, baseline)["range_pct"] if len(simulated_path) else 0.0 + return { + "step_returns_pct": blended_steps, + "step_abs_mean_pct": round(step_abs_mean_pct, 4), + "range_pct": round(float(range_pct), 4), + "match_count": len(chosen), + "signature_len": signature_len, + "best_match_distance": round(best_distance, 4), + } + + +def _derive_target_amplitude_profile( + ohlc4: np.ndarray, + context_len: int, + horizon: int, +) -> Dict[str, Any]: + """Blend recent and regime-matched history into a target amplitude profile.""" + recent_abs_step_pct = _recent_ohlc4_abs_step_pct(ohlc4, window=20) + rolling_targets = _rolling_future_path_targets( + ohlc4, + horizon=horizon, + max_windows=max(24, min(72, horizon * 6)), + ) + regime_targets = _regime_matched_future_targets( + ohlc4, + context_len=context_len, + horizon=horizon, + ) + + step_terms: List[Tuple[float, float]] = [] + if recent_abs_step_pct > 0: + step_terms.append((recent_abs_step_pct, 0.24)) + if rolling_targets is not None: + step_terms.append((float(rolling_targets["step_abs_mean_pct"]), 0.31)) + if regime_targets is not None: + step_terms.append((float(regime_targets["step_abs_mean_pct"]), 0.45)) + + if step_terms: + target_step_abs_mean_pct = sum(value * weight for value, weight in step_terms) / sum( + weight for _, weight in step_terms + ) + else: + target_step_abs_mean_pct = 0.0 + + range_terms: List[Tuple[float, float]] = [] + if rolling_targets is not None: + range_terms.append((float(rolling_targets["range_pct"]), 0.42)) + if regime_targets is not None: + range_terms.append((float(regime_targets["range_pct"]), 0.58)) + if not range_terms and target_step_abs_mean_pct > 0: + fallback_range = target_step_abs_mean_pct * max(2.0, min(math.sqrt(max(horizon, 1)) * 1.6, 4.5)) + range_terms.append((fallback_range, 1.0)) + + target_range_pct = sum(value * weight for value, weight in range_terms) / sum( + weight for _, weight in range_terms + ) if range_terms else 0.0 + + return { + "target_step_abs_mean_pct": round(float(target_step_abs_mean_pct), 4), + "target_range_pct": round(float(target_range_pct), 4), + "recent_abs_step_pct": round(float(recent_abs_step_pct), 4), + "rolling_targets": rolling_targets, + "regime_targets": regime_targets, + } + + +def _apply_forecast_amplitude_calibration( + raw_bundle: Dict[str, Any], + last_ohlc4: float, + target_profile: Dict[str, Any], +) -> Tuple[Dict[str, Any], Dict[str, Any]]: + """Scale raw TimesFM OHLC4 paths so their step/range amplitude stays realistic.""" + raw_p10 = np.asarray(raw_bundle.get("p10", []), dtype=float) + raw_p50 = np.asarray(raw_bundle.get("p50", []), dtype=float) + raw_p90 = np.asarray(raw_bundle.get("p90", []), dtype=float) + raw_responsiveness = _forecast_path_responsiveness(raw_p50, last_ohlc4) + + raw_step = max(float(raw_responsiveness["step_abs_mean_pct"]), 1e-6) + raw_range = max(float(raw_responsiveness["range_pct"]), 1e-6) + target_step = max(float(target_profile.get("target_step_abs_mean_pct") or 0.0), 0.0) + target_range = max(float(target_profile.get("target_range_pct") or 0.0), 0.0) + + ratios: List[Tuple[float, float]] = [] + if target_step > 0: + ratios.append((target_step / raw_step, 0.62)) + if target_range > 0: + ratios.append((target_range / raw_range, 0.38)) + + if ratios: + log_scale = sum(weight * math.log(max(ratio, 1e-6)) for ratio, weight in ratios) / sum( + weight for _, weight in ratios + ) + scale = math.exp(log_scale) + else: + scale = 1.0 + + scale = float(np.clip(scale, 0.82, 2.35)) + if raw_step >= target_step * 0.92 and raw_range >= target_range * 0.88: + scale = float(np.clip(scale, 0.9, 1.25)) + + calibrated_p10 = np.maximum(0.0, last_ohlc4 + ((raw_p10 - last_ohlc4) * scale)) + calibrated_p50 = np.maximum(0.0, last_ohlc4 + ((raw_p50 - last_ohlc4) * scale)) + calibrated_p90 = np.maximum(0.0, last_ohlc4 + ((raw_p90 - last_ohlc4) * scale)) + + calibrated_bundle = _build_raw_ohlc4_bundle( + { + "p10": calibrated_p10, + "p50": calibrated_p50, + "p90": calibrated_p90, + }, + last_ohlc4, + ) + calibrated_bundle["mode"] = "timesfm_ohlc4_vol_calibrated" + + calibration_meta = { + "enabled": True, + "version": "volatility_regime_v1", + "scale": round(scale, 4), + "raw_path": raw_responsiveness, + "calibrated_path": _forecast_path_responsiveness(calibrated_p50, last_ohlc4), + "targets": target_profile, + } + return calibrated_bundle, calibration_meta + + +def _apply_forecast_path_texture( + base_bundle: Dict[str, Any], + last_ohlc4: float, + target_profile: Dict[str, Any], + texture_template: Optional[Dict[str, Any]], +) -> Tuple[Dict[str, Any], Dict[str, Any]]: + """Inject regime-matched step texture so each future point is updated sequentially.""" + base_p10 = np.asarray(base_bundle.get("p10", []), dtype=float) + base_p50 = np.asarray(base_bundle.get("p50", []), dtype=float) + base_p90 = np.asarray(base_bundle.get("p90", []), dtype=float) + base_path = _forecast_path_responsiveness(base_p50, last_ohlc4) + + texture_meta: Dict[str, Any] = { + "enabled": True, + "version": "historical_step_texture_v2", + "applied": False, + "blend_alpha": 0.0, + "base_path": base_path, + "textured_path": base_path, + "template": { + "match_count": 0, + "signature_len": 0, + "step_abs_mean_pct": 0.0, + "range_pct": 0.0, + }, + } + if texture_template is None: + return base_bundle, texture_meta + + template_steps = np.asarray(texture_template.get("step_returns_pct", []), dtype=float) + if len(template_steps) != len(base_p50): + return base_bundle, texture_meta + + template_step = max(float(texture_template.get("step_abs_mean_pct") or 0.0), 1e-6) + template_range = max(float(texture_template.get("range_pct") or 0.0), 1e-6) + target_step = max(float(target_profile.get("target_step_abs_mean_pct") or 0.0), 0.0) + target_range = max(float(target_profile.get("target_range_pct") or 0.0), 0.0) + base_step = float(base_path["step_abs_mean_pct"]) + base_range = float(base_path["range_pct"]) + base_monotonicity = float(base_bundle.get("path_metrics", {}).get("monotonicity", 50.0)) + + step_gap = max((target_step * 0.94) - base_step, 0.0) + range_gap = max((target_range * 0.78) - base_range, 0.0) + if base_monotonicity >= 88.0: + step_gap = max(step_gap, target_step * 0.16) + range_gap = max(range_gap, target_range * 0.10) + + if step_gap <= 0.0 and range_gap <= 0.0: + texture_meta["template"] = { + "match_count": int(texture_template.get("match_count") or 0), + "signature_len": int(texture_template.get("signature_len") or 0), + "step_abs_mean_pct": round(float(texture_template.get("step_abs_mean_pct") or 0.0), 4), + "range_pct": round(float(texture_template.get("range_pct") or 0.0), 4), + } + return base_bundle, texture_meta + + ratios: List[float] = [] + if step_gap > 0.0: + ratios.append(step_gap / template_step) + if range_gap > 0.0: + ratios.append(range_gap / template_range) + blend_alpha = float(np.clip(np.median(ratios) if ratios else 0.0, 0.0, 0.82)) + + def _count_direction_flips(series: np.ndarray) -> int: + if len(series) < 2: + return 0 + signs = [int(np.sign(delta)) for delta in series if abs(delta) >= 0.015] + return sum(1 for prev, curr in zip(signs[:-1], signs[1:]) if prev != curr) + + template_flips = _count_direction_flips(template_steps) + if base_monotonicity >= 92.0 and template_flips >= 1: + blend_alpha = max(blend_alpha, 0.28) + + def _step_returns_from_path(path: np.ndarray, anchor_price: float) -> np.ndarray: + prev_values = np.concatenate(([anchor_price], path[:-1])) + return ((path / np.maximum(np.abs(prev_values), 1e-8)) - 1.0) * 100.0 + + def _rebuild_from_step_returns(path: np.ndarray, step_returns_pct: np.ndarray) -> np.ndarray: + rebuilt: List[float] = [] + prev_price = float(last_ohlc4) + for raw_price, textured_step in zip(path, step_returns_pct): + raw_step = ((float(raw_price) / max(abs(prev_price), 1e-8)) - 1.0) * 100.0 + raw_factor = max(1e-6, 1.0 + (raw_step / 100.0)) + target_factor = max(1e-6, 1.0 + (float(textured_step) / 100.0)) + factor_ratio = target_factor / raw_factor + next_price = max(0.0, float(raw_price) * factor_ratio) + rebuilt.append(next_price) + prev_price = next_price + return np.asarray(rebuilt, dtype=float) + + base_steps_p50 = _step_returns_from_path(base_p50, last_ohlc4) + textured_steps_p50 = (base_steps_p50 * (1.0 - blend_alpha)) + (template_steps * blend_alpha) + + if base_monotonicity >= 92.0 and template_flips >= 1: + textured_flips = _count_direction_flips(textured_steps_p50) + while textured_flips < 1 and blend_alpha < 0.92: + blend_alpha = min(blend_alpha * 1.16, 0.92) + textured_steps_p50 = (base_steps_p50 * (1.0 - blend_alpha)) + (template_steps * blend_alpha) + textured_flips = _count_direction_flips(textured_steps_p50) + + textured_p10 = _rebuild_from_step_returns(base_p10, textured_steps_p50) + textured_p50 = _rebuild_from_step_returns(base_p50, textured_steps_p50) + textured_p90 = _rebuild_from_step_returns(base_p90, textured_steps_p50) + textured_bundle = _build_raw_ohlc4_bundle( + { + "p10": textured_p10, + "p50": textured_p50, + "p90": textured_p90, + }, + last_ohlc4, + ) + textured_path = _forecast_path_responsiveness( + np.asarray(textured_bundle.get("p50", []), dtype=float), + last_ohlc4, + ) + + if target_step > 0.0 and float(textured_path["step_abs_mean_pct"]) > target_step * 1.28: + shrink = (target_step * 1.28) / max(float(textured_path["step_abs_mean_pct"]), 1e-6) + blend_alpha *= shrink + textured_steps_p50 = (base_steps_p50 * (1.0 - blend_alpha)) + (template_steps * blend_alpha) + textured_p10 = _rebuild_from_step_returns(base_p10, textured_steps_p50) + textured_p50 = _rebuild_from_step_returns(base_p50, textured_steps_p50) + textured_p90 = _rebuild_from_step_returns(base_p90, textured_steps_p50) + textured_bundle = _build_raw_ohlc4_bundle( + { + "p10": textured_p10, + "p50": textured_p50, + "p90": textured_p90, + }, + last_ohlc4, + ) + textured_path = _forecast_path_responsiveness( + np.asarray(textured_bundle.get("p50", []), dtype=float), + last_ohlc4, + ) + + textured_bundle["mode"] = "timesfm_ohlc4_textured" + texture_meta.update( + { + "applied": blend_alpha > 0.0, + "blend_alpha": round(blend_alpha, 4), + "textured_path": textured_path, + "template": { + "match_count": int(texture_template.get("match_count") or 0), + "signature_len": int(texture_template.get("signature_len") or 0), + "step_abs_mean_pct": round(float(texture_template.get("step_abs_mean_pct") or 0.0), 4), + "range_pct": round(float(texture_template.get("range_pct") or 0.0), 4), + }, + } + ) + return textured_bundle, texture_meta + + +def _score_forecast_context_candidate( + analysis_bundle: Dict[str, Any], + last_ohlc4: float, + recent_abs_step_pct: float, +) -> Dict[str, float]: + """Prefer contexts that stay confident without collapsing into a near-flat path.""" + p50_path = np.asarray(analysis_bundle.get("p50", []), dtype=float) + responsiveness = _forecast_path_responsiveness(p50_path, last_ohlc4) + step_abs_mean_pct = float(responsiveness["step_abs_mean_pct"]) + range_pct = float(responsiveness["range_pct"]) + step_floor = max(0.015, recent_abs_step_pct * 0.18) + range_floor = max(0.05, recent_abs_step_pct * 0.70) + step_bonus = min(step_abs_mean_pct / step_floor, 1.0) * 14.0 + range_bonus = min(range_pct / range_floor, 1.0) * 10.0 + flat_penalty = ( + 6.0 + if step_abs_mean_pct < (step_floor * 0.55) and range_pct < (range_floor * 0.55) + else 0.0 + ) + confidence = float(analysis_bundle.get("confidence", 0.0)) + score = confidence + step_bonus + range_bonus - flat_penalty + return { + "score": round(score, 4), + "confidence": round(confidence, 4), + "step_floor": round(step_floor, 4), + "range_floor": round(range_floor, 4), + **responsiveness, + } + + def _calc_vote_gauge(buy: int, sell: int, neutral: int) -> float: """Equal-weight vote gauge for oscillator and MA signals.""" buy_f = float(max(0, buy)) @@ -3863,6 +4609,7 @@ def _calc_ai_forecast_score( indicators: dict, horizon: int, interval: str, + model_reference_price: Optional[float] = None, ) -> dict: """MODULE 3: AI score from path quality, trend alignment, magnitude, and certainty.""" trend = indicators.get("trend", {}) @@ -3880,11 +4627,13 @@ def _calc_ai_forecast_score( if not len(p90_path): p90_path = np.array([last_close], dtype=float) + series_reference_price = float(model_reference_price or last_close) path_len = len(p50_path) - path_metrics = _forecast_path_metrics(p50_path, last_close) - forecast_ret_path = ((p50_path / max(last_close, 1e-8)) - 1.0) * 100.0 + path_metrics = _forecast_path_metrics(p50_path, series_reference_price) + forecast_ret_path = ((p50_path / max(series_reference_price, 1e-8)) - 1.0) * 100.0 final_ret_pct = path_metrics["final_return_pct"] weighted_ret_pct = path_metrics["weighted_return_pct"] + market_return_pct = _pct(float(p50_path[-1]), last_close) if len(p50_path) else 0.0 directional_edge_pct = (weighted_ret_pct * 0.60) + (final_ret_pct * 0.40) direction_norm = math.tanh(directional_edge_pct / max(atr_pct * 1.35, 0.35)) final_sign = 0 if abs(final_ret_pct) < 0.05 else (1 if final_ret_pct > 0 else -1) @@ -3944,7 +4693,8 @@ def _calc_ai_forecast_score( "gauge": round(gauge, 1), "normalized_score": round(_gauge_to_normalized_score(gauge), 4), "confidence_pct": round(confidence_pct, 1), - "forecast_return_pct": round(final_ret_pct, 2), + "forecast_return_pct": round(market_return_pct, 2), + "model_reference_return_pct": round(final_ret_pct, 2), "weighted_return_pct": round(weighted_ret_pct, 2), "direction": direction_label, "magnitude_vs_atr": round(move_in_atr, 2), @@ -4195,7 +4945,8 @@ def _rebuild_blended_from_forecast_payload(cached_forecast: Optional[Dict[str, A def _build_trade_analysis( symbol: str, interval: str, data: List[Dict[str, Any]], indicators: Dict[str, Any], forecast_rows: List[Dict[str, Any]], confidence: float, source: str, - blended: Optional[Dict[str, Any]] = None + blended: Optional[Dict[str, Any]] = None, + forecast_reference_price: Optional[float] = None, ) -> Dict[str, Any]: """ TradingView-style technical analysis dashboard (v6.1 Rework). @@ -4209,6 +4960,7 @@ def _build_trade_analysis( lows = np.array([float(d["low"]) for d in data], dtype=float) vols = np.array([float(d.get("volume", 0)) for d in data], dtype=float) last_close = closes[-1] + model_reference_price = float(forecast_reference_price or last_close) def _lv(arr): if arr is None: return None @@ -4336,6 +5088,7 @@ def _build_trade_analysis( indicators, len(forecast_rows) or 10, interval, + model_reference_price=model_reference_price, ) # ── 4. Summary ── @@ -4444,11 +5197,19 @@ async def lifespan(app: FastAPI): _start_background_task(_warmup_timesfm(), "timesfm-warmup") elif not TIMESFM_AVAILABLE: STARTUP_STATE["timesfm"]["last_error"] = TIMESFM_IMPORT_ERROR or "TimesFM import failed" + if PRELOAD_KRONOS and KRONOS_AVAILABLE: + _start_background_task(_warmup_kronos(), "kronos-warmup") + elif not KRONOS_AVAILABLE: + STARTUP_STATE["kronos"]["last_error"] = KRONOS_IMPORT_ERROR or "Kronos import failed" + if PRELOAD_CHRONOS and CHRONOS_AVAILABLE: + _start_background_task(_warmup_chronos(), "chronos-warmup") + elif not CHRONOS_AVAILABLE: + STARTUP_STATE["chronos"]["last_error"] = CHRONOS_IMPORT_ERROR or "Chronos import failed" try: yield finally: - logger.info("Shutting down AI Trading Chart API...") + logger.info("Shutting down AI Forecast API...") for task in background_tasks: task.cancel() if background_tasks: @@ -4461,9 +5222,9 @@ async def lifespan(app: FastAPI): # FastAPI App Instance (v6.0) # ────────────────────────────────────────────────────────────────────────────── app = FastAPI( - title="AI Trading Chart API", + title="AI Forecast API", version=APP_VERSION, - description="OHLCV data, hybrid AI forecasts, technical indicators, and real-time WebSocket prices", + description="OHLCV data, Kronos/TimesFM/Chronos forecasts, technical indicators, and real-time WebSocket prices", lifespan=lifespan, ) @@ -4516,6 +5277,88 @@ class TimesFMForecaster: self._predict_lock: Optional[asyncio.Lock] = None self._compiled_horizon: Optional[int] = None # last compiled max_horizon + @staticmethod + def _prepare_feature_frame(df: pd.DataFrame) -> pd.DataFrame: + """Collapse OHLC data into the OHLC4 signal used by TimesFM.""" + ohlc4 = ( + df[["open", "high", "low", "close"]] + .mean(axis=1) + .astype(np.float32) + ) + return pd.DataFrame({"ohlc4": ohlc4}) + + @classmethod + def _extract_ohlc4_series(cls, df: pd.DataFrame) -> np.ndarray: + """Return the float32 OHLC4 series expected by TimesFM.""" + feature_frame = cls._prepare_feature_frame(df) + ohlc4_series = feature_frame["ohlc4"].to_numpy(dtype=np.float32, copy=False) + cls._validate_input_series(ohlc4_series) + return ohlc4_series + + @staticmethod + def _validate_input_series(series: np.ndarray) -> None: + if series.ndim != 1: + raise HTTPException(status_code=422, detail="TimesFM input series must be 1-D") + if len(series) < 32: + raise HTTPException(status_code=422, detail="TimesFM requires at least 32 OHLC4 points") + if not np.isfinite(series).all(): + raise HTTPException(status_code=422, detail="TimesFM input contains non-finite OHLC4 values") + + @classmethod + def _validate_output_tensors( + cls, + point_forecast: np.ndarray, + quantile_forecast: np.ndarray, + horizon: int, + ) -> Tuple[np.ndarray, np.ndarray, np.ndarray, Dict[str, Any]]: + point_arr = np.asarray(point_forecast) + quant_arr = np.asarray(quantile_forecast) + + if point_arr.ndim != 2 or point_arr.shape[0] != 1 or point_arr.shape[1] < horizon: + raise HTTPException( + status_code=500, + detail=f"TimesFM point forecast has invalid shape: {list(point_arr.shape)}", + ) + if ( + quant_arr.ndim != 3 + or quant_arr.shape[0] != 1 + or quant_arr.shape[1] < horizon + or quant_arr.shape[2] <= cls._Q90 + ): + raise HTTPException( + status_code=500, + detail=f"TimesFM quantile forecast has invalid shape: {list(quant_arr.shape)}", + ) + + p10 = quant_arr[0, :horizon, cls._Q10].astype(float) + p50 = point_arr[0, :horizon].astype(float) + p50_from_quantiles = quant_arr[0, :horizon, cls._Q50].astype(float) + p90 = quant_arr[0, :horizon, cls._Q90].astype(float) + + if not (np.isfinite(p10).all() and np.isfinite(p50).all() and np.isfinite(p90).all()): + raise HTTPException(status_code=500, detail="TimesFM output contains non-finite values") + + quantiles_monotonic = bool( + np.all(p10 <= (p50 + 1e-6)) + and np.all(p50 <= (p90 + 1e-6)) + ) + if not quantiles_monotonic: + raise HTTPException(status_code=500, detail="TimesFM returned non-monotonic quantiles") + + median_matches_point_forecast = bool( + np.allclose(p50, p50_from_quantiles, atol=1e-4, rtol=1e-4) + ) + if not median_matches_point_forecast: + logger.warning("[TimesFM] point_forecast differs from q50 quantile output") + + output_validation = { + "point_shape": [int(dim) for dim in point_arr.shape], + "quantile_shape": [int(dim) for dim in quant_arr.shape], + "median_matches_point_forecast": median_matches_point_forecast, + "quantiles_monotonic": quantiles_monotonic, + } + return p10, p50, p90, output_validation + async def _get_lock(self) -> asyncio.Lock: if self._lock is None: self._lock = asyncio.Lock() @@ -4560,8 +5403,11 @@ class TimesFMForecaster: fix_quantile_crossing=True, ) ) - self._compiled_horizon = max_h - logger.info("[TimesFM] Compiled: ctx=%d max_horizon=%d", ctx, max_h) + compiled_config = getattr(self._model, "forecast_config", None) + actual_ctx = int(getattr(compiled_config, "max_context", ctx)) + actual_max_horizon = int(getattr(compiled_config, "max_horizon", max_h)) + self._compiled_horizon = actual_max_horizon + logger.info("[TimesFM] Compiled: ctx=%d max_horizon=%d", actual_ctx, actual_max_horizon) async def _lazy_load(self) -> None: if self._loaded: @@ -4606,15 +5452,10 @@ class TimesFMForecaster: assert self._model is not None try: - # Build single OHLC4 series: (O+H+L+C)/4 - ohlc4 = ( - df[["open", "high", "low", "close"]] - .mean(axis=1) - .astype(np.float32) - .values - ) - context_len = min(len(ohlc4), self.MAX_CONTEXT) - ohlc4_ctx = ohlc4[-context_len:] + # Build single OHLC4 series for the TimesFM univariate input contract. + ohlc4_series = self._extract_ohlc4_series(df) + context_len = min(len(ohlc4_series), self.MAX_CONTEXT) + ohlc4_ctx = ohlc4_series[-context_len:] # Re-compile if horizon exceeds current compiled max_horizon if self._compiled_horizon is None or horizon > self._compiled_horizon: @@ -4638,11 +5479,11 @@ class TimesFMForecaster: if torch is not None and torch.cuda.is_available(): torch.cuda.empty_cache() - # point_forecast: (1, horizon) - # quantile_forecast: (1, horizon, 10) - p10 = quantile_forecast[0, :horizon, self._Q10].astype(float) - p50 = quantile_forecast[0, :horizon, self._Q50].astype(float) - p90 = quantile_forecast[0, :horizon, self._Q90].astype(float) + p10, p50, p90, output_validation = self._validate_output_tensors( + point_forecast=point_forecast, + quantile_forecast=quantile_forecast, + horizon=horizon, + ) return { "p10": p10, @@ -4662,12 +5503,20 @@ class TimesFMForecaster: "adapter_mode": "timesfm_native", "normalization": "timesfm_internal_revin", }, + "input_validation": { + "series_field": "ohlc4", + "dtype": str(ohlc4_ctx.dtype), + "is_1d": True, + "finite": True, + }, "output_semantics": { "forecast_channel": "ohlc4", "forecast_mode": "single_future_ohlc4_line", "quantile_fields": ["p10", "p50", "p90"], "candle_projection": "omitted", + "reference_baseline": "last_ohlc4", }, + "output_validation": output_validation, } except Exception as ex: logger.error("[TimesFM] Forecast failed: %s", ex, exc_info=True) @@ -4676,7 +5525,9 @@ class TimesFMForecaster: -forecaster = TimesFMForecaster() +forecaster = TimesFMProviderForecaster(logger) +kronos_forecaster = KronosForecaster(logger) +chronos_forecaster = ChronosProviderForecaster(logger) @@ -4781,8 +5632,6 @@ async def websocket_price(websocket: WebSocket, symbol: str): # ────────────────────────────────────────────────────────────────────────────── # Architecture v5: Security & Guardrails (Module F) # ────────────────────────────────────────────────────────────────────────────── -from fastapi.responses import FileResponse, HTMLResponse, JSONResponse - IP_LIMITS: Dict[str, List[float]] = defaultdict(list) REQUEST_METRICS = RequestMetricsRegistry() @@ -4899,7 +5748,32 @@ async def _warmup_timesfm() -> None: logger.error("[TimesFM] Warmup failed: %s", ex) +async def _warmup_kronos() -> None: + """Load Kronos in the background so the default AI model is ready quickly.""" + try: + STARTUP_STATE["kronos"]["warming"] = True + await kronos_forecaster._lazy_load() + STARTUP_STATE["kronos"]["warming"] = False + STARTUP_STATE["kronos"]["loaded"] = kronos_forecaster.is_ready + STARTUP_STATE["kronos"]["device"] = kronos_forecaster.device + logger.info("[Kronos] Warmup complete — ready on %s", kronos_forecaster.device) + except Exception as ex: + STARTUP_STATE["kronos"]["warming"] = False + STARTUP_STATE["kronos"]["last_error"] = str(ex) + logger.error("[Kronos] Warmup failed: %s", ex) + + # ── Symbol / Interval listing ───────────────────────────────────────────────── +async def _warmup_chronos() -> None: + """Load Chronos in the background so the third AI model is ready quickly.""" + await warmup_forecaster( + forecaster=chronos_forecaster, + startup_state=STARTUP_STATE["chronos"], + logger=logger, + label="Chronos", + ) + + _SYMBOLS_CACHE: Optional[Dict[str, Any]] = None @app.get("/api/symbols") @@ -5021,6 +5895,9 @@ async def get_analysis( interval: str = Query("1h"), refresh: bool = Query(False), include_snapshot: bool = Query(False), + use_kronos: bool = Query(True), + use_timesfm: bool = Query(True), + use_chronos: bool = Query(True), ) -> Dict[str, Any]: """ A-5: Direct access to the comprehensive Analysis Engine. @@ -5068,13 +5945,27 @@ async def get_analysis( confidence = 50.0 forecast_rows: List[Dict[str, Any]] = [] blended: Optional[Dict[str, Any]] = None + forecast_reference_price = float(data[-1]["close"]) try: - f_prefix = _cache_prefix(symbol, interval) - f_cache = forecast_cache.get(f"forecast_{f_prefix}10") + default_model_selection = _normalize_forecast_model_selection( + use_kronos, + use_timesfm, + use_chronos, + ) + forecast_cache_key = _forecast_cache_key(symbol, interval, 10, default_model_selection) + f_cache = forecast_cache.get(forecast_cache_key) if f_cache: forecast_rows = f_cache.get("forecast") or [] if forecast_rows: - forecast_ret = _pct(float(forecast_rows[-1]["p50"]), float(f_cache.get("last_close") or data[-1]["close"])) + reference_prices = f_cache.get("reference_prices") or {} + forecast_reference_price = float( + reference_prices.get("ohlc4") + or f_cache.get("last_ohlc4") + or reference_prices.get("close") + or f_cache.get("last_close") + or data[-1]["close"] + ) + forecast_ret = _pct(float(forecast_rows[-1]["p50"]), forecast_reference_price) confidence = float(f_cache.get("ensemble", {}).get("confidence") or 50.0) blended = _rebuild_blended_from_forecast_payload(f_cache, float(data[-1]["close"])) except Exception as ex: @@ -5091,6 +5982,7 @@ async def get_analysis( confidence=confidence, source=source, blended=blended, + forecast_reference_price=forecast_reference_price, ) # Inject MTF into analysis @@ -5206,39 +6098,91 @@ async def get_watchlist_tickers(body: WatchlistRequest) -> Dict[str, Any]: } -def _forecast_cache_key(symbol: str, interval: str, horizon: int) -> str: - return f"forecast_{_cache_prefix(symbol, interval)}{horizon}" +def _normalize_forecast_model_selection( + use_kronos: bool, + use_timesfm: bool, + use_chronos: bool = True, +) -> Dict[str, bool]: + return build_query_model_selection( + use_kronos=use_kronos, + use_timesfm=use_timesfm, + use_chronos=use_chronos, + ) + + +def _forecast_model_signature(model_selection: Dict[str, bool]) -> str: + return forecast_model_signature(model_selection) + + +def _forecast_cache_key( + symbol: str, + interval: str, + horizon: int, + model_selection: Dict[str, bool], +) -> str: + model_sig = _forecast_model_signature(model_selection) + return ( + f"forecast_{_cache_prefix(symbol, interval)}{horizon}:{model_sig}:" + f"rules={FORECAST_RULE_DOCUMENT.version}" + ) def _forecast_payload_is_current(cached: Optional[Dict[str, Any]]) -> bool: if not isinstance(cached, dict): return False - if "forecast_candles" in cached: + if "forecast_candles" in cached or "model_selection" not in cached: return False display = cached.get("display") or {} + requested = (cached.get("model_selection") or {}).get("requested") or {} + forecast_models = cached.get("forecast_models") or {} + active_models = (cached.get("model_selection") or {}).get("active") or [] if ( - display.get("mode") != "raw_timesfm_ohlc4_line" - or display.get("output_mode") != "single_future_ohlc4_line" + display.get("output_mode") != "single_future_ohlc4_line" or display.get("channels") != ["ohlc4"] + or display.get("reference_series") != "ohlc4" + or display.get("market_price_field") != "last_close" + or display.get("forecast_reference_field") != "last_ohlc4" + or display.get("visual_anchor_field") != "last_close" + or display.get("combination_mode") != "mean_of_enabled_models" ): return False - model_meta = cached.get("model") or {} - semantics = model_meta.get("input_semantics") or {} - output_semantics = model_meta.get("output_semantics") or {} - return ( - semantics.get("volume_mode") == "omitted" - and semantics.get("amount_mode") == "omitted" - and semantics.get("active_forecast_channels") == ["ohlc4"] - and semantics.get("feature_channels") == ["ohlc4"] - and semantics.get("price_mode") == "ohlc4_single_channel" - and semantics.get("base_signal") == "ohlc4" - and semantics.get("adapter_mode") == "timesfm_native" - and output_semantics.get("forecast_channel") == "ohlc4" - and output_semantics.get("forecast_mode") == "single_future_ohlc4_line" - and output_semantics.get("candle_projection") == "omitted" - ) + if not isinstance(requested, dict): + return False + if not isinstance(forecast_models, dict): + return False + if not isinstance(active_models, list): + return False + + combined_model_meta = cached.get("model") or {} + if not isinstance(combined_model_meta.get("components"), dict): + return False + + for model_key in FORECAST_MODEL_ORDER: + if model_key not in requested: + return False + model_payload = forecast_models.get(model_key) + if model_payload is None: + return False + if not isinstance(model_payload, dict): + return False + if requested.get(model_key): + if not bool(model_payload.get("success")): + return False + model_meta = model_payload.get("model") or {} + semantics = model_meta.get("input_semantics") or {} + output_semantics = model_meta.get("output_semantics") or {} + if ( + semantics.get("feature_channels") != ["ohlc4"] + or semantics.get("active_forecast_channels") != ["ohlc4"] + or output_semantics.get("forecast_channel") != "ohlc4" + or output_semantics.get("forecast_mode") != "single_future_ohlc4_line" + or output_semantics.get("reference_baseline") != "last_ohlc4" + ): + return False + + return True def _load_cached_forecast_response(cache_key: str, interval: str) -> Optional[Dict[str, Any]]: @@ -5273,21 +6217,22 @@ async def _prepare_forecast_response_payload( refresh=refresh, min_context=FORECAST_CONTEXT, ) - if not TIMESFM_AVAILABLE: - last_ohlc4 = ( - float( - np.mean( - [ - float(data_list[-1]["open"]), - float(data_list[-1]["high"]), - float(data_list[-1]["low"]), - float(data_list[-1]["close"]), - ] - ) + last_close = float(data_list[-1]["close"]) if data_list else 0.0 + last_ohlc4 = ( + float( + np.mean( + [ + float(data_list[-1]["open"]), + float(data_list[-1]["high"]), + float(data_list[-1]["low"]), + float(data_list[-1]["close"]), + ] ) - if data_list - else 0.0 ) + if data_list + else 0.0 + ) + if not TIMESFM_AVAILABLE: return { "symbol": symbol, "interval": interval, @@ -5299,10 +6244,19 @@ async def _prepare_forecast_response_payload( "_blended": {"confidence": 0.0, "agreement": False, "scale": 1.0, "model_weight": 0.0, "anchor_weight": 1.0, "model_bias_pct": 0.0}, "source": source, "horizon": horizon, - "last_close": last_ohlc4, + "last_close": last_close, + "last_ohlc4": last_ohlc4, + "reference_prices": { + "close": round(last_close, 6), + "ohlc4": round(last_ohlc4, 6), + "forecast_baseline": "ohlc4", + }, "model": { "name": "offline", "context_length": 0, + "available_history": len(data_list), + "context_strategy": "recent_tail", + "context_cap": _resolve_forecast_context_window(interval, len(data_list)) if data_list else 0, "quantiles": [0.1, 0.5, 0.9], "cache_version": CACHE_VERSION, "sample_count": 0, @@ -5316,12 +6270,20 @@ async def _prepare_forecast_response_payload( "amount_mode": "omitted", "adapter_mode": "timesfm_native", }, + "input_validation": { + "series_field": "ohlc4", + "dtype": "float32", + "is_1d": True, + "finite": True, + }, "output_semantics": { "forecast_channel": "ohlc4", "forecast_mode": "single_future_ohlc4_line", "quantile_fields": ["p10", "p50", "p90"], "candle_projection": "omitted", + "reference_baseline": "last_ohlc4", }, + "output_validation": {}, }, "_cache_origin": cache_origin, } @@ -5338,25 +6300,98 @@ async def _prepare_forecast_response_payload( else: df_hist["amount"] = df_hist["amount"].fillna(0) - # TimesFM handles context internally; pass the full available history - context_len = min(len(df_hist), TimesFMForecaster.MAX_CONTEXT) - df_context = df_hist.tail(context_len).reset_index(drop=True) - - logger.info("[forecast] %s %s | ctx=%d/%d | horizon=%d", symbol, interval, context_len, len(df_hist), horizon) - last_time = int(df_hist["time"].iloc[-1]) step = STEP_SECONDS[interval] + ohlc4_hist = ( + df_hist[["open", "high", "low", "close"]] + .mean(axis=1) + .to_numpy(dtype=float) + ) + last_ohlc4 = float(ohlc4_hist[-1]) + last_close = float(df_hist["close"].iloc[-1]) + preferred_context_len = _resolve_forecast_context_window(interval, len(df_hist)) + context_candidates = _resolve_forecast_context_candidates(interval, len(df_hist)) + context_trials: List[Dict[str, Any]] = [] + + for candidate_len in context_candidates: + df_context = df_hist.tail(candidate_len).reset_index(drop=True) + candidate_output = await forecaster.forecast( + df=df_context[["open", "high", "low", "close", "volume"]], + horizon=horizon, + ) + raw_candidate_bundle = _build_raw_ohlc4_bundle(candidate_output, last_ohlc4) + target_profile = _derive_target_amplitude_profile( + ohlc4=ohlc4_hist, + context_len=candidate_len, + horizon=horizon, + ) + candidate_bundle, calibration_meta = _apply_forecast_amplitude_calibration( + raw_bundle=raw_candidate_bundle, + last_ohlc4=last_ohlc4, + target_profile=target_profile, + ) + candidate_score = _score_forecast_context_candidate( + analysis_bundle=candidate_bundle, + last_ohlc4=last_ohlc4, + recent_abs_step_pct=float(target_profile.get("recent_abs_step_pct") or 0.0), + ) + context_trials.append( + { + "context_length": candidate_len, + "model_output": candidate_output, + "analysis_bundle": candidate_bundle, + "calibration_meta": calibration_meta, + **candidate_score, + "final_return_pct": float( + candidate_bundle.get("path_metrics", {}).get("final_return_pct", 0.0) + ), + "weighted_return_pct": float( + candidate_bundle.get("path_metrics", {}).get("weighted_return_pct", 0.0) + ), + } + ) - model_output = await forecaster.forecast( - df=df_context[["open", "high", "low", "close", "volume"]], + selected_trial = max( + context_trials, + key=lambda trial: ( + float(trial["score"]), + float(trial["confidence"]), + float(trial["step_abs_mean_pct"]), + -int(trial["context_length"]), + ), + ) + context_len = int(selected_trial["context_length"]) + model_output = selected_trial["model_output"] + analysis_bundle = selected_trial["analysis_bundle"] + texture_template = _build_regime_texture_template( + ohlc4=ohlc4_hist, + context_len=context_len, horizon=horizon, ) + analysis_bundle, texture_meta = _apply_forecast_path_texture( + base_bundle=analysis_bundle, + last_ohlc4=last_ohlc4, + target_profile=selected_trial["calibration_meta"]["targets"], + texture_template=texture_template, + ) + selected_trial["analysis_bundle"] = analysis_bundle + selected_trial["texture_meta"] = texture_meta - last_ohlc4 = float(df_hist[["open", "high", "low", "close"]].mean(axis=1).iloc[-1]) - last_close = last_ohlc4 - analysis_bundle = _build_raw_close_bundle(model_output, last_ohlc4) logger.info( - "[forecast] raw-ohlc4-line | %s %s | confidence=%.1f agreement=%s", + "[forecast] %s %s | ctx=%d/%d | horizon=%d | candidates=%s | recent_abs_step=%.4f | selected_score=%.2f | calibration_scale=%.2f | texture_blend=%.2f", + symbol, + interval, + context_len, + len(df_hist), + horizon, + context_candidates, + float(selected_trial["calibration_meta"]["targets"].get("recent_abs_step_pct") or 0.0), + float(selected_trial["score"]), + float(selected_trial["calibration_meta"].get("scale") or 1.0), + float(texture_meta.get("blend_alpha") or 0.0), + ) + logger.info( + "[forecast] textured-ohlc4-line | %s %s | confidence=%.1f agreement=%s", symbol, interval, analysis_bundle["confidence"], @@ -5380,40 +6415,78 @@ async def _prepare_forecast_response_payload( "source": source, "horizon": horizon, "last_close": last_close, + "last_ohlc4": last_ohlc4, + "reference_prices": { + "close": round(last_close, 6), + "ohlc4": round(last_ohlc4, 6), + "forecast_baseline": "ohlc4", + }, "forecast_rows": forecast_rows, "from_persistent_cache": False, "model": { - "name": model_output.get("model_name", TimesFMForecaster.MODEL_HF_ID), + "name": model_output.get("model_name", TimesFMProviderForecaster.MODEL_HF_ID), "context_length": int(model_output.get("context_length", context_len)), + "available_history": len(df_hist), + "context_strategy": "recent_tail", + "context_cap": preferred_context_len, + "context_candidates": context_candidates, + "postprocess": { + "amplitude_calibration": selected_trial["calibration_meta"], + "path_texture": texture_meta, + }, + "context_selector": { + "recent_abs_step_pct": float( + selected_trial["calibration_meta"]["targets"].get("recent_abs_step_pct") or 0.0 + ), + "selected_context_length": context_len, + "selected_score": round(float(selected_trial["score"]), 4), + "candidates": [ + { + "context_length": int(trial["context_length"]), + "score": round(float(trial["score"]), 4), + "confidence": round(float(trial["confidence"]), 4), + "range_pct": round(float(trial["range_pct"]), 4), + "step_abs_mean_pct": round(float(trial["step_abs_mean_pct"]), 4), + "step_abs_max_pct": round(float(trial["step_abs_max_pct"]), 4), + "final_return_pct": round(float(trial["final_return_pct"]), 4), + "weighted_return_pct": round(float(trial["weighted_return_pct"]), 4), + "calibration_scale": round(float(trial["calibration_meta"].get("scale") or 1.0), 4), + } + for trial in context_trials + ], + }, "quantiles": [0.1, 0.5, 0.9], "cache_version": CACHE_VERSION, "sample_count": 0, # TimesFM is deterministic; no sampling needed "input_semantics": model_output.get("input_semantics", {}), + "input_validation": model_output.get("input_validation", {}), "output_semantics": model_output.get("output_semantics", {}), + "output_validation": model_output.get("output_validation", {}), }, "indicators_snapshot": indicators, "_data_list": data_list, "_analysis_bundle": analysis_bundle, + "_forecast_reference_price": last_ohlc4, "_model_output": model_output, "_cache_origin": cache_origin, "ai_runtime": { "mode": "local_only", - "model": str(model_output.get("model_name", TimesFMForecaster.MODEL_HF_ID)), + "model": str(model_output.get("model_name", TimesFMProviderForecaster.MODEL_HF_ID)), "device": forecaster.device, }, } -def _build_raw_close_bundle( +def _build_raw_ohlc4_bundle( model_output: Dict[str, Any], - last_close: float, + last_ohlc4: float, ) -> Dict[str, Any]: - """Build the close-path bundle directly from raw Kronos output.""" + """Build the OHLC4-path bundle directly from raw TimesFM output.""" raw_p10 = np.array(model_output["p10"], dtype=float) raw_p50 = np.array(model_output["p50"], dtype=float) raw_p90 = np.array(model_output["p90"], dtype=float) - path_metrics = _forecast_path_metrics(raw_p50, last_close) + path_metrics = _forecast_path_metrics(raw_p50, last_ohlc4) avg_band_pct = float( np.mean((raw_p90 - raw_p10) / np.maximum(np.abs(raw_p50), 1e-8)) * 100.0 ) if len(raw_p50) else 0.0 @@ -5464,6 +6537,10 @@ async def _finalize_forecast_response_payload(payload: Dict[str, Any]) -> Dict[s "output_mode": "single_future_ohlc4_line", "uncertainty_source": "timesfm_quantile_head", "uses_anchor_blending": False, + "reference_series": "ohlc4", + "market_price_field": "last_close", + "forecast_reference_field": "last_ohlc4", + "visual_anchor_field": "last_close", }, "ai_runtime": payload.get("ai_runtime", {"mode": "local_only", "model": "offline"}), } @@ -5480,6 +6557,7 @@ async def _finalize_forecast_response_payload(payload: Dict[str, Any]) -> Dict[s confidence=float(analysis_bundle.get("confidence", 50.0)), source=payload["source"], blended=analysis_bundle, + forecast_reference_price=float(payload.get("_forecast_reference_price") or payload["last_close"]), ) response = { @@ -5488,6 +6566,8 @@ async def _finalize_forecast_response_payload(payload: Dict[str, Any]) -> Dict[s "source": payload["source"], "horizon": payload["horizon"], "last_close": payload["last_close"], + "last_ohlc4": payload.get("last_ohlc4", payload["last_close"]), + "reference_prices": payload.get("reference_prices", {}), "forecast": payload["forecast_rows"], "from_persistent_cache": payload.get("from_persistent_cache", False), "model": payload["model"], @@ -5497,6 +6577,10 @@ async def _finalize_forecast_response_payload(payload: Dict[str, Any]) -> Dict[s "output_mode": "single_future_ohlc4_line", "uncertainty_source": "timesfm_quantile_head", "uses_anchor_blending": False, + "reference_series": "ohlc4", + "market_price_field": "last_close", + "forecast_reference_field": "last_ohlc4", + "visual_anchor_field": "last_close", }, "ensemble": { "mode": "raw_timesfm_ohlc4", @@ -5527,13 +6611,604 @@ async def _finalize_forecast_response_payload(payload: Dict[str, Any]) -> Dict[s return make_json_compatible(response) +def _is_forecast_model_available(model_key: str) -> bool: + return { + "kronos": KRONOS_AVAILABLE, + "timesfm": TIMESFM_AVAILABLE, + "chronos": CHRONOS_AVAILABLE, + }.get(model_key, False) + + +def _build_requested_model_stub(model_key: str, enabled: bool, available: bool, error: Optional[str] = None) -> Dict[str, Any]: + return { + "enabled": enabled, + "available": available, + "success": False, + "skipped": not enabled, + "error": error, + "forecast": [], + "model": {}, + "ensemble": {}, + "model_diagnostics": {}, + "ai_runtime": { + "mode": "local_only", + "model": model_key, + "device": "not_loaded", + }, + } + + +async def _run_forecast_model_pipeline( + model_key: str, + symbol: str, + interval: str, + horizon: int, + df_hist: pd.DataFrame, + ohlc4_hist: np.ndarray, + last_ohlc4: float, + last_time: int, + step_seconds: int, +) -> Dict[str, Any]: + if model_key == "kronos": + if not KRONOS_AVAILABLE: + return _build_requested_model_stub( + model_key=model_key, + enabled=True, + available=False, + error=KRONOS_IMPORT_ERROR or "Kronos unavailable", + ) + runner = kronos_forecaster + display_mode = "raw_kronos_ohlc4_line" + elif model_key == "timesfm": + if not TIMESFM_AVAILABLE: + return _build_requested_model_stub( + model_key=model_key, + enabled=True, + available=False, + error=TIMESFM_IMPORT_ERROR or "TimesFM unavailable", + ) + runner = forecaster + display_mode = "raw_timesfm_ohlc4_line" + elif model_key == "chronos": + if not CHRONOS_AVAILABLE: + return _build_requested_model_stub( + model_key=model_key, + enabled=True, + available=False, + error=CHRONOS_IMPORT_ERROR or "Chronos unavailable", + ) + runner = chronos_forecaster + display_mode = "raw_chronos_ohlc4_line" + else: + return _build_requested_model_stub( + model_key=model_key, + enabled=True, + available=False, + error=f"Unsupported AI model: {model_key}", + ) + + context_cap = _resolve_model_context_cap(model_key, interval, len(df_hist)) + context_candidates = _resolve_model_context_candidates(model_key, interval, len(df_hist)) + context_trials: List[Dict[str, Any]] = [] + + for candidate_len in context_candidates: + df_context = df_hist.tail(candidate_len).reset_index(drop=True) + if model_key == "kronos": + candidate_output = await runner.forecast( + df=df_context, + horizon=horizon, + interval=interval, + step_seconds=step_seconds, + symbol=symbol, + ) + else: + candidate_output = await runner.forecast( + df=df_context, + horizon=horizon, + interval=interval, + step_seconds=step_seconds, + symbol=symbol, + ) + + raw_candidate_bundle = _build_raw_ohlc4_bundle(candidate_output, last_ohlc4) + raw_candidate_bundle["mode"] = f"raw_{model_key}_ohlc4" + target_profile = _derive_target_amplitude_profile( + ohlc4=ohlc4_hist, + context_len=candidate_len, + horizon=horizon, + ) + candidate_bundle, calibration_meta = _apply_forecast_amplitude_calibration( + raw_bundle=raw_candidate_bundle, + last_ohlc4=last_ohlc4, + target_profile=target_profile, + ) + candidate_score = _score_forecast_context_candidate( + analysis_bundle=candidate_bundle, + last_ohlc4=last_ohlc4, + recent_abs_step_pct=float(target_profile.get("recent_abs_step_pct") or 0.0), + ) + context_trials.append( + { + "context_length": candidate_len, + "model_output": candidate_output, + "analysis_bundle": candidate_bundle, + "calibration_meta": calibration_meta, + **candidate_score, + "final_return_pct": float(candidate_bundle.get("path_metrics", {}).get("final_return_pct", 0.0)), + "weighted_return_pct": float(candidate_bundle.get("path_metrics", {}).get("weighted_return_pct", 0.0)), + } + ) + + selected_trial = max( + context_trials, + key=lambda trial: ( + float(trial["score"]), + float(trial["confidence"]), + float(trial["step_abs_mean_pct"]), + -int(trial["context_length"]), + ), + ) + context_len = int(selected_trial["context_length"]) + model_output = selected_trial["model_output"] + analysis_bundle = selected_trial["analysis_bundle"] + texture_template = _build_regime_texture_template( + ohlc4=ohlc4_hist, + context_len=context_len, + horizon=horizon, + ) + analysis_bundle, texture_meta = _apply_forecast_path_texture( + base_bundle=analysis_bundle, + last_ohlc4=last_ohlc4, + target_profile=selected_trial["calibration_meta"]["targets"], + texture_template=texture_template, + ) + analysis_bundle["mode"] = f"{model_key}_ohlc4_textured" + + logger.info( + "[forecast:%s] %s %s | ctx=%d/%d | horizon=%d | candidates=%s | selected_score=%.2f | calibration_scale=%.2f | texture_blend=%.2f", + model_key, + symbol, + interval, + context_len, + len(df_hist), + horizon, + context_candidates, + float(selected_trial["score"]), + float(selected_trial["calibration_meta"].get("scale") or 1.0), + float(texture_meta.get("blend_alpha") or 0.0), + ) + + forecast_rows: List[Dict[str, Any]] = [ + { + "time": last_time, + "p10": last_ohlc4, + "p50": last_ohlc4, + "p90": last_ohlc4, + "is_actual": True, + } + ] + for step_index in range(horizon): + forecast_rows.append( + { + "time": int(last_time + step_seconds * (step_index + 1)), + "p10": round(float(analysis_bundle["p10"][step_index]), 6), + "p50": round(float(analysis_bundle["p50"][step_index]), 6), + "p90": round(float(analysis_bundle["p90"][step_index]), 6), + } + ) + + model_meta = { + "model_key": model_key, + "name": str(model_output.get("model_name", model_key)), + "tokenizer_name": model_output.get("tokenizer_name"), + "context_length": int(model_output.get("context_length", context_len)), + "available_history": len(df_hist), + "context_strategy": "recent_tail", + "context_cap": context_cap, + "context_candidates": context_candidates, + "postprocess": { + "amplitude_calibration": selected_trial["calibration_meta"], + "path_texture": texture_meta, + }, + "context_selector": { + "recent_abs_step_pct": float( + selected_trial["calibration_meta"]["targets"].get("recent_abs_step_pct") or 0.0 + ), + "selected_context_length": context_len, + "selected_score": round(float(selected_trial["score"]), 4), + "candidates": [ + { + "context_length": int(trial["context_length"]), + "score": round(float(trial["score"]), 4), + "confidence": round(float(trial["confidence"]), 4), + "range_pct": round(float(trial["range_pct"]), 4), + "step_abs_mean_pct": round(float(trial["step_abs_mean_pct"]), 4), + "step_abs_max_pct": round(float(trial["step_abs_max_pct"]), 4), + "final_return_pct": round(float(trial["final_return_pct"]), 4), + "weighted_return_pct": round(float(trial["weighted_return_pct"]), 4), + "calibration_scale": round(float(trial["calibration_meta"].get("scale") or 1.0), 4), + } + for trial in context_trials + ], + }, + "quantiles": [0.1, 0.5, 0.9], + "cache_version": CACHE_VERSION, + "sample_count": int(model_output.get("sample_count") or 0), + "input_semantics": model_output.get("input_semantics", {}), + "input_validation": model_output.get("input_validation", {}), + "output_semantics": model_output.get("output_semantics", {}), + "output_validation": model_output.get("output_validation", {}), + } + ensemble_meta = { + "mode": analysis_bundle["mode"], + "model_weight": analysis_bundle["model_weight"], + "anchor_weight": analysis_bundle["anchor_weight"], + "trend_agreement": analysis_bundle["agreement"], + "confidence": analysis_bundle["confidence"], + "model_bias_pct": analysis_bundle["model_bias_pct"], + "alignment_scale": analysis_bundle["scale"], + "used_for_display": True, + } + diagnostics = { + "raw_last_p10": round(float(model_output["p10"][-1]), 6), + "raw_last_p50": round(float(model_output["p50"][-1]), 6), + "raw_last_p90": round(float(model_output["p90"][-1]), 6), + "display_last_p50": round(float(analysis_bundle["p50"][-1]), 6), + } + + return { + "enabled": True, + "available": True, + "success": True, + "skipped": False, + "error": None, + "forecast": forecast_rows, + "model": model_meta, + "ensemble": ensemble_meta, + "model_diagnostics": diagnostics, + "ai_runtime": { + "mode": "local_only", + "model": str(model_meta["name"]), + "device": runner.device, + }, + "_analysis_bundle": analysis_bundle, + "_display_mode": display_mode, + } + + +def _combine_model_analysis_bundles( + model_payloads: Dict[str, Dict[str, Any]], + active_model_keys: List[str], + last_ohlc4: float, +) -> Dict[str, Any]: + if len(active_model_keys) == 1: + return dict(model_payloads[active_model_keys[0]]["_analysis_bundle"]) + + bundles = [model_payloads[model_key]["_analysis_bundle"] for model_key in active_model_keys] + combined_p10 = np.mean([np.array(bundle["p10"], dtype=float) for bundle in bundles], axis=0) + combined_p50 = np.mean([np.array(bundle["p50"], dtype=float) for bundle in bundles], axis=0) + combined_p90 = np.mean([np.array(bundle["p90"], dtype=float) for bundle in bundles], axis=0) + final_signs = [] + for bundle in bundles: + final_return = float(bundle.get("path_metrics", {}).get("final_return_pct") or 0.0) + final_signs.append(0 if abs(final_return) < 0.05 else (1 if final_return > 0 else -1)) + + agreement = len({sign for sign in final_signs if sign != 0}) <= 1 + return { + "p10": combined_p10, + "p50": combined_p50, + "p90": combined_p90, + "model_weight": 1.0, + "anchor_weight": 0.0, + "agreement": agreement, + "scale": float(np.mean([float(bundle.get("scale", 1.0) or 1.0) for bundle in bundles])), + "confidence": round(float(np.mean([float(bundle.get("confidence", 50.0) or 50.0) for bundle in bundles])), 2), + "model_bias_pct": 0.0, + "path_metrics": _forecast_path_metrics(combined_p50, last_ohlc4), + "mode": "mean_of_enabled_models", + } + + +def _build_combined_forecast_rows( + analysis_bundle: Dict[str, Any], + last_time: int, + step_seconds: int, + horizon: int, + last_ohlc4: float, +) -> List[Dict[str, Any]]: + forecast_rows: List[Dict[str, Any]] = [ + { + "time": last_time, + "p10": round(float(last_ohlc4), 6), + "p50": round(float(last_ohlc4), 6), + "p90": round(float(last_ohlc4), 6), + "is_actual": True, + } + ] + for step_index in range(horizon): + forecast_rows.append( + { + "time": int(last_time + step_seconds * (step_index + 1)), + "p10": round(float(analysis_bundle["p10"][step_index]), 6), + "p50": round(float(analysis_bundle["p50"][step_index]), 6), + "p90": round(float(analysis_bundle["p90"][step_index]), 6), + } + ) + return forecast_rows + + +def _average_ai_scores(ai_scores: List[Dict[str, Any]], interval: str) -> Dict[str, Any]: + gauge = float(np.mean([float(score.get("gauge", 50.0) or 50.0) for score in ai_scores])) + certainty = float(np.mean([float(score.get("certainty", 0.0) or 0.0) for score in ai_scores])) + confidence_pct = float(np.mean([float(score.get("confidence_pct", 0.0) or 0.0) for score in ai_scores])) + path_consistency = float(np.mean([float(score.get("path_consistency", 50.0) or 50.0) for score in ai_scores])) + monotonicity = float(np.mean([float(score.get("monotonicity", 50.0) or 50.0) for score in ai_scores])) + max_adverse_excursion_pct = float(np.mean([float(score.get("max_adverse_excursion_pct", 0.0) or 0.0) for score in ai_scores])) + forecast_return_pct = float(np.mean([float(score.get("forecast_return_pct", 0.0) or 0.0) for score in ai_scores])) + weighted_return_pct = float(np.mean([float(score.get("weighted_return_pct", 0.0) or 0.0) for score in ai_scores])) + direction_label = "bullish" if gauge >= 58 else "bearish" if gauge <= 42 else "neutral" + return { + "gauge": round(gauge, 1), + "normalized_score": round(_gauge_to_normalized_score(gauge), 4), + "certainty": round(certainty, 1), + "confidence_pct": round(confidence_pct, 1), + "path_consistency": round(path_consistency, 1), + "monotonicity": round(monotonicity, 1), + "max_adverse_excursion_pct": round(max_adverse_excursion_pct, 2), + "forecast_return_pct": round(forecast_return_pct, 2), + "weighted_return_pct": round(weighted_return_pct, 2), + "direction_label": direction_label, + "signal": _gauge_to_signal(gauge, interval), + "formula": "mean(enabled_models)", + } + + +async def _build_multi_model_forecast_response( + symbol: str, + interval: str, + horizon: int, + refresh: bool, + cache_origin: str, + model_selection: Dict[str, bool], +) -> Dict[str, Any]: + data_list, source, indicators = await get_indicators_cached( + symbol, + interval, + FORECAST_CONTEXT, + refresh=refresh, + min_context=FORECAST_CONTEXT, + ) + if len(data_list) < 40: + raise HTTPException(422, "Insufficient historical data for forecasting") + + df_hist = pd.DataFrame(data_list) + df_hist["timestamps"] = pd.to_datetime(df_hist["time"], unit="s", utc=True) + + if "amount" not in df_hist.columns or df_hist["amount"].isna().all() or df_hist["amount"].sum() == 0: + typical = (df_hist["high"] + df_hist["low"] + df_hist["close"]) / 3 + df_hist["amount"] = (df_hist["volume"] * typical).fillna(0) + else: + df_hist["amount"] = df_hist["amount"].fillna(0) + + last_time = int(df_hist["time"].iloc[-1]) + step_seconds = STEP_SECONDS[interval] + last_close = float(df_hist["close"].iloc[-1]) + ohlc4_hist = df_hist[["open", "high", "low", "close"]].mean(axis=1).to_numpy(dtype=float) + last_ohlc4 = float(ohlc4_hist[-1]) + + requested_models = { + model_key: bool(model_selection.get(model_key, False)) + for model_key in FORECAST_MODEL_ORDER + } + if not any(requested_models.values()): + raise HTTPException(status_code=400, detail="At least one AI model must be enabled") + + model_payloads: Dict[str, Dict[str, Any]] = {} + active_model_keys: List[str] = [] + for model_key in FORECAST_MODEL_ORDER: + if not requested_models[model_key]: + model_payloads[model_key] = _build_requested_model_stub( + model_key=model_key, + enabled=False, + available=_is_forecast_model_available(model_key), + ) + continue + + model_payload = await _run_forecast_model_pipeline( + model_key=model_key, + symbol=symbol, + interval=interval, + horizon=horizon, + df_hist=df_hist, + ohlc4_hist=ohlc4_hist, + last_ohlc4=last_ohlc4, + last_time=last_time, + step_seconds=step_seconds, + ) + model_payloads[model_key] = model_payload + if model_payload.get("success"): + active_model_keys.append(model_key) + + if not active_model_keys: + error_messages = [ + f"{model_key}: {model_payloads[model_key].get('error')}" + for model_key in FORECAST_MODEL_ORDER + if requested_models[model_key] + ] + raise HTTPException(status_code=503, detail="; ".join(error_messages) or "No enabled AI model produced a forecast") + + combined_bundle = _combine_model_analysis_bundles(model_payloads, active_model_keys, last_ohlc4) + combined_forecast_rows = _build_combined_forecast_rows( + analysis_bundle=combined_bundle, + last_time=last_time, + step_seconds=step_seconds, + horizon=horizon, + last_ohlc4=last_ohlc4, + ) + + model_analyses: Dict[str, Dict[str, Any]] = {} + ai_scores: List[Dict[str, Any]] = [] + for model_key in active_model_keys: + model_payload = model_payloads[model_key] + model_analysis = await asyncio.to_thread( + _build_trade_analysis, + symbol=symbol, + interval=interval, + data=data_list, + indicators=indicators, + forecast_rows=model_payload["forecast"], + confidence=float(model_payload["ensemble"].get("confidence", 50.0)), + source=source, + blended=model_payload["_analysis_bundle"], + forecast_reference_price=last_ohlc4, + ) + model_analyses[model_key] = model_analysis + ai_scores.append(model_analysis["ai_gauge"]) + model_payload["analysis"] = model_analysis + + analysis = await asyncio.to_thread( + _build_trade_analysis, + symbol=symbol, + interval=interval, + data=data_list, + indicators=indicators, + forecast_rows=combined_forecast_rows, + confidence=float(combined_bundle.get("confidence", 50.0)), + source=source, + blended=combined_bundle, + forecast_reference_price=last_ohlc4, + ) + combined_ai_score = _average_ai_scores(ai_scores, interval) + technical_score = analysis["technicals"] + summary = _calc_summary_score_v2(technical_score, combined_ai_score, interval) + analysis["ai_gauge"] = combined_ai_score + analysis["summary"] = summary + analysis["dashboard"] = _build_dashboard_payload(last_close, combined_forecast_rows, technical_score, combined_ai_score, summary) + analysis["ai_models"] = { + "requested": requested_models, + "active": active_model_keys, + "formula": "mean(enabled_models)", + "models": { + model_key: { + "gauge": round(float(model_analyses[model_key]["ai_gauge"]["gauge"]), 1), + "signal": model_analyses[model_key]["ai_gauge"]["signal"], + "confidence_pct": round(float(model_analyses[model_key]["ai_gauge"].get("confidence_pct", 0.0)), 1), + "certainty": round(float(model_analyses[model_key]["ai_gauge"].get("certainty", 0.0)), 1), + } + for model_key in active_model_keys + }, + } + + selected_display_mode = model_payloads[active_model_keys[0]].get("_display_mode", "multi_model_ohlc4_line") + response = { + "symbol": symbol, + "interval": interval, + "source": source, + "horizon": horizon, + "last_close": last_close, + "last_ohlc4": last_ohlc4, + "reference_prices": { + "close": round(last_close, 6), + "ohlc4": round(last_ohlc4, 6), + "forecast_baseline": "ohlc4", + }, + "forecast": combined_forecast_rows, + "forecast_models": { + model_key: { + key: value + for key, value in model_payloads[model_key].items() + if not key.startswith("_") + } + for model_key in FORECAST_MODEL_ORDER + }, + "model_selection": { + "requested": requested_models, + "active": active_model_keys, + "defaults": DEFAULT_FORECAST_MODEL_SELECTION, + "shared_horizon": horizon, + "independent_forecasts": True, + "independent_analysis": True, + "combination_scope": "top_level_combined_view_and_gauge_only", + "combination_mode": "mean_of_enabled_models", + }, + "forecast_contract": { + "input_series": "ohlc4", + "input_timeframe": interval, + "recommended_context_length": FORECAST_RULE_DOCUMENT.recommended_context_length, + "default_line_width": FORECAST_RULE_DOCUMENT.default_line_width, + "shared_horizon": horizon, + "output_points": [f"T+{step_index}" for step_index in range(1, horizon + 1)], + "reference_anchor": "T0.last_ohlc4", + "independent_models": True, + }, + "rules": FORECAST_RULE_DOCUMENT.to_metadata(), + "from_persistent_cache": False, + "model": { + "name": model_payloads[active_model_keys[0]]["model"]["name"] if len(active_model_keys) == 1 else "mean_of_enabled_models", + "active_models": active_model_keys, + "components": { + model_key: model_payloads[model_key]["model"] + for model_key in FORECAST_MODEL_ORDER + if model_payloads[model_key].get("success") + }, + "cache_version": CACHE_VERSION, + }, + "display": { + "mode": selected_display_mode if len(active_model_keys) == 1 else "multi_model_ohlc4_line", + "channels": ["ohlc4"], + "output_mode": "single_future_ohlc4_line", + "uncertainty_source": "model_quantiles", + "uses_anchor_blending": False, + "reference_series": "ohlc4", + "market_price_field": "last_close", + "forecast_reference_field": "last_ohlc4", + "visual_anchor_field": "last_close", + "render_models": active_model_keys, + "combination_mode": "mean_of_enabled_models", + }, + "ensemble": { + "mode": combined_bundle["mode"], + "model_weight": combined_bundle["model_weight"], + "anchor_weight": combined_bundle["anchor_weight"], + "trend_agreement": combined_bundle["agreement"], + "confidence": combined_bundle["confidence"], + "model_bias_pct": combined_bundle["model_bias_pct"], + "alignment_scale": combined_bundle["scale"], + "used_for_display": True, + }, + "model_diagnostics": { + model_key: model_payloads[model_key].get("model_diagnostics", {}) + for model_key in FORECAST_MODEL_ORDER + }, + "indicators_snapshot": indicators, + "analysis": analysis, + "generated_at": int(time.time()), + "cache": { + "origin": cache_origin, + "refresh_requested": refresh, + }, + "ai_runtime": { + "mode": "local_only", + "active_models": active_model_keys, + "devices": { + model_key: model_payloads[model_key].get("ai_runtime", {}).get("device", "not_loaded") + for model_key in FORECAST_MODEL_ORDER + }, + }, + } + return make_json_compatible(response) + + # ── Forecast ────────────────────────────────────────────────────────────────── @app.get("/api/forecast/{symbol}") async def get_forecast( symbol: str, interval: str = Query("1h"), horizon: int = Query(10, ge=5, le=300), - refresh: bool = Query(False) + refresh: bool = Query(False), + use_kronos: bool = Query(True), + use_timesfm: bool = Query(True), + use_chronos: bool = Query(True), ) -> Dict[str, Any]: symbol = _get_canonical_symbol(symbol) if symbol not in SYMBOLS: @@ -5541,7 +7216,12 @@ async def get_forecast( if interval not in SUPPORTED_INTERVALS: raise HTTPException(400, f"Unsupported interval: {interval}") - cache_key = _forecast_cache_key(symbol, interval, horizon) + model_selection = _normalize_forecast_model_selection( + use_kronos, + use_timesfm, + use_chronos, + ) + cache_key = _forecast_cache_key(symbol, interval, horizon, model_selection) cache_origin = "live" if not refresh: @@ -5558,14 +7238,14 @@ async def get_forecast( return await inflight_task async def _build_forecast_response() -> Dict[str, Any]: - payload = await _prepare_forecast_response_payload( + response = await _build_multi_model_forecast_response( symbol=symbol, interval=interval, horizon=horizon, refresh=refresh, cache_origin=cache_origin, + model_selection=model_selection, ) - response = await _finalize_forecast_response_payload(payload) forecast_cache.set(cache_key, response, ttl_seconds=forecast_ttl(interval)) persistent_cache.set(cache_key, response, ttl=forecast_ttl(interval) * 4) return response @@ -5669,11 +7349,30 @@ async def cache_stats(request: Request) -> Dict[str, Any]: async def health_check() -> Dict[str, Any]: STARTUP_STATE["timesfm"]["loaded"] = forecaster.is_ready STARTUP_STATE["timesfm"]["device"] = forecaster.device + STARTUP_STATE["kronos"]["loaded"] = kronos_forecaster.is_ready + STARTUP_STATE["kronos"]["device"] = kronos_forecaster.device + STARTUP_STATE["chronos"]["loaded"] = chronos_forecaster.is_ready + STARTUP_STATE["chronos"]["device"] = chronos_forecaster.device + any_model_available = bool(TIMESFM_AVAILABLE or KRONOS_AVAILABLE or CHRONOS_AVAILABLE) + ready_models = [ + model_key + for model_key, loaded in { + "kronos": STARTUP_STATE["kronos"]["loaded"], + "timesfm": STARTUP_STATE["timesfm"]["loaded"], + "chronos": STARTUP_STATE["chronos"]["loaded"], + }.items() + if loaded + ] return { - "status": "online" if TIMESFM_AVAILABLE else "degraded", + "status": "online" if any_model_available else "degraded", "version": APP_VERSION, - "model_ready": forecaster.is_ready, - "device": forecaster.device, + "model_ready": bool(ready_models), + "ready_models": ready_models, + "device": { + "kronos": kronos_forecaster.device, + "timesfm": forecaster.device, + "chronos": chronos_forecaster.device, + }, "symbols_count": len(SYMBOLS), "timestamp": datetime.now(timezone.utc).isoformat(), "cache_version": CACHE_VERSION, @@ -5681,7 +7380,9 @@ async def health_check() -> Dict[str, Any]: "cors_allow_origins": CORS_ALLOW_ORIGINS, "admin_auth_configured": ADMIN_TOKEN != DEFAULT_ADMIN_TOKEN, "request_metrics": REQUEST_METRICS.snapshot(), + "kronos": STARTUP_STATE["kronos"], "timesfm": STARTUP_STATE["timesfm"], + "chronos": STARTUP_STATE["chronos"], "startup_checks": STARTUP_STATE["sources"], } @@ -5698,6 +7399,15 @@ async def get_ai_rules() -> Dict[str, Any]: return ai_rule_registry.snapshot() +@app.get("/api/forecasting/rules") +async def get_forecasting_rules() -> Dict[str, Any]: + return { + "rules": FORECAST_RULE_DOCUMENT.to_metadata(), + "content": FORECAST_RULE_DOCUMENT.content, + "models": list(FORECAST_MODEL_ORDER), + } + + @app.get("/api/metrics") async def get_metrics(request: Request): """Export Prometheus-ready metrics (latencies, cache hits, CB states).""" @@ -5724,7 +7434,9 @@ async def get_metrics(request: Request): }, "circuit_breakers": cb_stats, "sources": STARTUP_STATE["sources"], + "kronos_status": STARTUP_STATE["kronos"]["loaded"], "timesfm_status": STARTUP_STATE["timesfm"]["loaded"], + "chronos_status": STARTUP_STATE["chronos"]["loaded"], } @@ -5794,86 +7506,10 @@ async def get_volume_profile( # ── Static Frontend ─────────────────────────────────────────────────────────── -FRONTEND_PATH = os.path.join(PROJECT_ROOT, "frontend") -if os.path.exists(FRONTEND_PATH): - INDEX_PATH = os.path.join(FRONTEND_PATH, "index.html") - AIBG_PATH = os.path.join(FRONTEND_PATH, "AIBG.png") - FAVICON_PATH = os.path.join(FRONTEND_PATH, "favicon.svg") - WORKSPACE_JS_PATH = os.path.join(FRONTEND_PATH, "workspace.js") - WORKSPACE_CSS_PATH = os.path.join(FRONTEND_PATH, "workspace.css") - - def _frontend_asset_headers() -> Dict[str, str]: - return { - "Cache-Control": "no-store, no-cache, must-revalidate, max-age=0", - "Pragma": "no-cache", - "Expires": "0", - } - - def _frontend_asset_version() -> str: - asset_paths = [INDEX_PATH, WORKSPACE_JS_PATH, WORKSPACE_CSS_PATH] - version_parts: List[str] = [APP_VERSION, CACHE_VERSION] - for asset_path in asset_paths: - if os.path.exists(asset_path): - version_parts.append(str(int(os.path.getmtime(asset_path)))) - return "-".join(version_parts) - - @app.get("/", include_in_schema=False) - @app.get("/index.html", include_in_schema=False) - async def serve_frontend_index() -> HTMLResponse: - if not os.path.exists(INDEX_PATH): - raise HTTPException(status_code=404, detail="Frontend index not found") - - html = Path(INDEX_PATH).read_text(encoding="utf-8") - html = html.replace("__FRONTEND_ASSET_VERSION__", _frontend_asset_version()) - headers = _frontend_asset_headers() - return HTMLResponse(content=html, headers=headers) - - @app.get("/workspace.js", include_in_schema=False) - async def serve_workspace_js() -> FileResponse: - if not os.path.exists(WORKSPACE_JS_PATH): - raise HTTPException(status_code=404, detail="Workspace JS asset not found") - - return FileResponse( - WORKSPACE_JS_PATH, - media_type="application/javascript", - headers=_frontend_asset_headers(), - ) - - @app.get("/workspace.css", include_in_schema=False) - async def serve_workspace_css() -> FileResponse: - if not os.path.exists(WORKSPACE_CSS_PATH): - raise HTTPException(status_code=404, detail="Workspace CSS asset not found") - - return FileResponse( - WORKSPACE_CSS_PATH, - media_type="text/css", - headers=_frontend_asset_headers(), - ) - - @app.get("/AIBG.png", include_in_schema=False) - async def serve_aibg() -> FileResponse: - if not os.path.exists(AIBG_PATH): - raise HTTPException(status_code=404, detail="AIBG asset not found") - - return FileResponse( - AIBG_PATH, - media_type="image/png", - headers=_frontend_asset_headers(), - ) - - @app.get("/favicon.svg", include_in_schema=False) - @app.get("/favicon.ico", include_in_schema=False) - async def serve_favicon() -> FileResponse: - if not os.path.exists(FAVICON_PATH): - raise HTTPException(status_code=404, detail="Favicon not found") - - return FileResponse( - FAVICON_PATH, - media_type="image/svg+xml", - headers=_frontend_asset_headers(), - ) - - app.mount("/", StaticFiles(directory=FRONTEND_PATH, html=True), name="frontend") - logger.info("Mounted frontend: %s", FRONTEND_PATH) -else: - logger.warning("Frontend path not found: %s", FRONTEND_PATH) +register_frontend_assets( + app, + project_root=PROJECT_ROOT, + app_version=APP_VERSION, + cache_version=CACHE_VERSION, + logger=logger, +)