""" AI Trading Chart — Backend API v4.0 ===================================== FastAPI backend serving OHLCV data, AI forecasts, technical indicators, real-time WebSocket prices, and a comprehensive symbol registry. ═══════════════════════════════════════════════════════════════════════════════ FREE DATA SOURCES (confirmed, no paid key required) ═══════════════════════════════════════════════════════════════════════════════ ┌─────────────────┬─────────────────────────────────────┬──────────────────┐ │ Source │ Best For │ Limit (Free) │ ├─────────────────┼─────────────────────────────────────┼──────────────────┤ │ Binance API │ Crypto – fastest, deepest history │ 1200 req/min │ │ Bybit API │ Crypto – backup to Binance │ 120 req/min │ │ CoinGecko API │ Crypto OHLCV + market data │ 30 req/min │ │ yfinance │ Stocks/ETF/Indices/Commodities/Forex│ Unlimited* │ │ Finnhub │ Forex/Stock/Crypto candles │ 60 req/min │ │ Twelvedata │ Forex/Stocks/Crypto │ 800 credits/day │ │ Alpha Vantage │ Forex/Stocks (fallback) │ 25 req/day │ │ FRED (St.Louis) │ Macro indicators (DXY, rates...) │ Unlimited │ └─────────────────┴─────────────────────────────────────┴──────────────────┘ * yfinance scrapes Yahoo Finance — no official rate limit but use responsibly. Source priority per asset class: Crypto : binance → bybit → coingecko → yfinance → finnhub Forex : twelvedata → finnhub → yfinance Stocks : yfinance → finnhub → twelvedata Commodities: yfinance → twelvedata → finnhub Indices : yfinance → twelvedata → finnhub VN Stocks : yfinance (.VN suffix) ═══════════════════════════════════════════════════════════════════════════════ Changelog v4.0 vs v3.0 ═══════════════════════════════════════════════════════════════════════════════ [FEAT] 120+ symbols across 10 categories (Kim loại, Năng lượng, Nông sản, Nguyên liệu, Crypto, Cặp tiền, Chỉ số, Cổ phiếu Mỹ, Cổ phiếu VN, Trái phiếu/Lãi suất) [FEAT] Bybit API — free crypto fallback after Binance [FEAT] CoinGecko API — free crypto OHLCV + 24h stats [FEAT] Per-category SOURCE_PRIORITY — each asset class uses its best source [FEAT] GET /api/indicators/{symbol} — RSI, MACD, Bollinger Bands, EMA, ATR [FEAT] GET /api/ticker/{symbol} — real-time last price + 24h stats [FEAT] GET /api/market-status — which markets are open right now [FEAT] GET /api/search?q= — fuzzy symbol search [FEAT] WebSocket /ws/price/{symbol} — real-time price stream [FEAT] GET /api/crypto/market — top-N crypto by market cap (CoinGecko) [FEAT] Global rate-limiter with per-source token buckets [FEAT] Exponential backoff with jitter on 429 / network errors [FIX] All v3.0 fixes retained (bias correction, dedup, 4h alignment…) [IMPR] SymbolConfig extended: category_en, coingecko_id, bybit mapping [IMPR] _normalize_ohlcv: gap detection — warns on suspiciously long candle gaps [IMPR] Startup self-test pings all active sources and logs reachability """ from __future__ import annotations import asyncio import hmac import logging import math import os import hashlib from dotenv import load_dotenv # D-1: Load environment variables from .env file (v6.0) load_dotenv() os.environ["HF_HUB_DISABLE_SYMLINKS_WARNING"] = "1" os.environ["HF_HUB_DISABLE_TELEMETRY"] = "1" import re import sys import time from collections import defaultdict from dataclasses import asdict, dataclass, field from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, List, Optional, Set, Tuple import httpx import numpy as np import pandas as pd 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 ( RequestMetricsRegistry, elapsed_ms, make_request_id, start_timer, ) from backend.runtime_utils import ( make_json_compatible, 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, normalize_watchlist_symbols, rate_limit_guard as check_rate_limit, validate_cache_target, validate_symbol_and_interval, ) 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, build_market_peer_candidates, search_symbols_catalog, ) 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.responses import JSONResponse from pydantic import BaseModel try: import torch TORCH_IMPORT_ERROR: Optional[str] = None except Exception as exc: torch = None # type: ignore[assignment] TORCH_IMPORT_ERROR = str(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) # ────────────────────────────────────────────────────────────────────────────── class Settings(BaseModel): # D-1: Security Hardening (v6.0) - Enforce environment variables twelvedata_api_key: Optional[str] = os.getenv("TWELVEDATA_API_KEY") finnhub_api_key: Optional[str] = os.getenv("FINNHUB_API_KEY") binance_api_key: Optional[str] = os.getenv("BINANCE_API_KEY") binance_api_secret: Optional[str] = os.getenv("BINANCE_API_SECRET") bybit_api_key: Optional[str] = os.getenv("BYBIT_API_KEY") bybit_api_secret: Optional[str] = os.getenv("BYBIT_API_SECRET") alphavantage_api_key: Optional[str] = os.getenv("ALPHAVANTAGE_API_KEY") admin_token: str = os.getenv("ADMIN_TOKEN", DEFAULT_ADMIN_TOKEN) # Environment detection is_hf: bool = os.getenv("SPACE_ID") is not None # App Config host: str = os.getenv("HOST", "0.0.0.0") port: int = int(os.getenv("PORT", 8000)) 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", "*") # WebSocket Config ws_heartbeat_interval: int = 20 # Circuit Breaker Config cb_failure_threshold: int = 5 cb_recovery_timeout: int = 60 settings = Settings() DEFAULT_AI_RULE_PATHS = [ r"D:\Python\Rules\core-behavior.mdc", r"D:\Python\Rules\thinking-reasoning.mdc", r"D:\Python\Rules\tool-usage.mdc", r"D:\Python\Rules\agentic-coding.mdc", r"D:\Python\Rules\code-quality.mdc", r"D:\Python\Rules\output-format.mdc", ] @dataclass class AIRuleDocument: path: str name: str description: str always_apply: bool body: str sections: List[str] modified_at: float frontmatter: Dict[str, str] = field(default_factory=dict) class AIRuleRegistry: def __init__(self, rule_paths: List[str]): self.rule_paths = [str(Path(p)) for p in rule_paths] self.documents: List[AIRuleDocument] = [] self.missing_paths: List[str] = [] self.load_errors: Dict[str, str] = {} self.combined_instruction = "" self.version = "uninitialized" self._mtimes: Dict[str, float] = {} self.refresh() def _parse_frontmatter(self, raw_text: str) -> Tuple[Dict[str, str], str]: lines = raw_text.splitlines() if len(lines) < 3 or lines[0].strip() != "---": return {}, raw_text.strip() meta: Dict[str, str] = {} closing_index = None for index in range(1, len(lines)): if lines[index].strip() == "---": closing_index = index break if ":" not in lines[index]: continue key, value = lines[index].split(":", 1) meta[key.strip()] = value.strip().strip('"').strip("'") if closing_index is None: return {}, raw_text.strip() body = "\n".join(lines[closing_index + 1:]).strip() return meta, body def _extract_sections(self, body: str) -> List[str]: sections: List[str] = [] for line in body.splitlines(): stripped = line.strip() if stripped.startswith("## "): sections.append(stripped[3:].strip()) return sections def _load_document(self, path: str) -> AIRuleDocument: rule_path = Path(path) raw_text = rule_path.read_text(encoding="utf-8") meta, body = self._parse_frontmatter(raw_text) return AIRuleDocument( path=str(rule_path), name=rule_path.stem, description=meta.get("description", ""), always_apply=meta.get("alwaysApply", "").lower() == "true", body=body, sections=self._extract_sections(body), modified_at=rule_path.stat().st_mtime, frontmatter=meta, ) def _compile_instruction(self, documents: List[AIRuleDocument]) -> str: blocks = [ "You must follow the runtime policy below for every response.", "Treat these rules as active operating constraints, not optional guidance.", ] for doc in documents: header = doc.description or doc.name blocks.append(f"[{doc.name}] {header}") blocks.append(doc.body.strip()) return "\n\n".join(blocks).strip() def refresh(self) -> None: documents: List[AIRuleDocument] = [] missing_paths: List[str] = [] load_errors: Dict[str, str] = {} mtimes: Dict[str, float] = {} for path in self.rule_paths: rule_path = Path(path) if not rule_path.exists(): missing_paths.append(str(rule_path)) continue try: doc = self._load_document(str(rule_path)) documents.append(doc) mtimes[str(rule_path)] = doc.modified_at except Exception as exc: load_errors[str(rule_path)] = str(exc) documents.sort(key=lambda doc: self.rule_paths.index(doc.path)) self.documents = documents self.missing_paths = missing_paths self.load_errors = load_errors self._mtimes = mtimes digest_source = "||".join( f"{doc.path}:{doc.modified_at}:{hashlib.sha256(doc.body.encode('utf-8')).hexdigest()}" for doc in documents ) self.version = hashlib.sha256(digest_source.encode("utf-8")).hexdigest()[:12] if digest_source else "empty" self.combined_instruction = self._compile_instruction(documents) def refresh_if_needed(self) -> None: for path in self.rule_paths: rule_path = Path(path) current_mtime = rule_path.stat().st_mtime if rule_path.exists() else -1.0 if self._mtimes.get(str(rule_path), -2.0) != current_mtime: self.refresh() return def build_system_instruction(self, task_contract: str) -> str: self.refresh_if_needed() parts = [self.combined_instruction] if self.combined_instruction else [] if task_contract.strip(): parts.append(task_contract.strip()) return "\n\n".join(parts).strip() def snapshot(self) -> Dict[str, Any]: self.refresh_if_needed() return { "version": self.version, "loaded_count": len(self.documents), "missing_count": len(self.missing_paths), "error_count": len(self.load_errors), "missing_paths": self.missing_paths, "load_errors": self.load_errors, "documents": [ { "name": doc.name, "path": doc.path, "description": doc.description, "always_apply": doc.always_apply, "sections": doc.sections, "modified_at": doc.modified_at, } for doc in self.documents ], "compiled_instruction_preview": self.combined_instruction[:2000], } ai_rule_registry = AIRuleRegistry(DEFAULT_AI_RULE_PATHS) # ────────────────────────────────────────────────────────────────────────────── # Architecture v5: Global HTTP Client Pool (C-2) # ────────────────────────────────────────────────────────────────────────────── class GlobalHTTPClient: """Singleton HTTP client for optimized connection pooling (C-2).""" _client: Optional[httpx.AsyncClient] = None @classmethod async def get_client(cls) -> httpx.AsyncClient: if cls._client is None or cls._client.is_closed: cls._client = httpx.AsyncClient( timeout=httpx.Timeout(15.0, connect=5.0), limits=httpx.Limits(max_connections=100, max_keepalive_connections=20) ) return cls._client @classmethod async def close(cls): if cls._client and not cls._client.is_closed: await cls._client.aclose() # ────────────────────────────────────────────────────────────────────────────── # Architecture v5: Circuit Breaker Pattern (C-3) # ────────────────────────────────────────────────────────────────────────────── class CircuitBreaker: """Protects against failing data sources (C-3).""" def __init__(self, name: str, threshold: int = 5, timeout: int = 60): self.name = name self.threshold = threshold self.timeout = timeout self.failures = 0 self.last_failure_time = 0.0 self.state = "CLOSED" # CLOSED, OPEN, HALF-OPEN def allow_request(self) -> bool: if self.state == "OPEN": if time.time() - self.last_failure_time > self.timeout: self.state = "HALF-OPEN" return True return False return True def record_success(self): self.failures = 0 self.state = "CLOSED" def record_failure(self): self.failures += 1 self.last_failure_time = time.time() if self.failures >= self.threshold: self.state = "OPEN" logger.error("[CB] %s is now OPEN - circuit broken", self.name) def open_for(self, timeout: Optional[int] = None) -> None: if timeout is not None: self.timeout = max(self.timeout, timeout) self.failures = max(self.failures, self.threshold) self.last_failure_time = time.time() self.state = "OPEN" logger.error("[CB] %s is OPEN for %ss", self.name, self.timeout) # Instance per source source_breakers: Dict[str, CircuitBreaker] = { s: CircuitBreaker(s, settings.cb_failure_threshold, settings.cb_recovery_timeout) for s in ["binance", "bybit", "coingecko", "twelvedata", "finnhub", "yfinance", "alphavantage"] } # ────────────────────────────────────────────────────────────────────────────── # Logging # ────────────────────────────────────────────────────────────────────────────── logging.basicConfig( level=logging.INFO, format="%(asctime)s | %(levelname)s | %(name)s | %(message)s", ) 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) APP_VERSION = "6.1.0" # ─── PyInstaller / Frozen detection ───────────────────────────────────────── IS_FROZEN = getattr(sys, 'frozen', False) BUNDLE_DIR = getattr(sys, '_MEIPASS', os.path.dirname(os.path.abspath(__file__))) RUNTIME_PATHS = resolve_runtime_paths(__file__, IS_FROZEN, BUNDLE_DIR) CURRENT_DIR = RUNTIME_PATHS.current_dir PROJECT_ROOT = RUNTIME_PATHS.project_root if IS_FROZEN: logger.info("Running in FROZEN mode. BUNDLE_DIR: %s", BUNDLE_DIR) else: logger.info("Running in DEV mode. PROJECT_ROOT: %s", PROJECT_ROOT) if TIMESFM_AVAILABLE: logger.info("TimesFM library imported successfully") elif TIMESFM_IMPORT_ERROR: logger.warning("TimesFM unavailable: %s", TIMESFM_IMPORT_ERROR) 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, "preload_enabled": PRELOAD_TIMESFM, "warming": False, "loaded": False, "device": "not_loaded", "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": {}, } # ────────────────────────────────────────────────────────────────────────────── # Architecture v5: Structured Logging & Persistence (C-4, C-6) # ────────────────────────────────────────────────────────────────────────────── import json class StructuredLogger: @staticmethod def info(msg: str, **kwargs): payload = {"level": "INFO", "msg": msg, "ts": datetime.now().isoformat()} payload.update(kwargs) logger.info(msg) # Keeping standard log for dev, but payload is ready for JSON persistent_cache = PersistentCache( db_path=os.path.join(PROJECT_ROOT, "data", "aiforecast.db"), cache_version_getter=lambda: settings.cache_version, logger=logger, ) # B-12: Global Configuration Instances CACHE_VERSION = settings.cache_version ADMIN_TOKEN = settings.admin_token CORS_ALLOW_ORIGINS = parse_cors_origins(settings.cors_allow_origins_raw) CORS_ALLOW_CREDENTIALS = CORS_ALLOW_ORIGINS != ["*"] # ── API Key Pool (round-robin rotation for multi-pane) ──────────────────────── class APIKeyPool: """Round-robin key rotation to distribute API calls across multiple keys.""" def __init__(self, primary_key, pool_csv, name=""): self._name = name self._keys = [] self._index = 0 self._exhausted = set() if pool_csv: self._keys = [k.strip() for k in pool_csv.split(",") if k.strip()] if not self._keys and primary_key: self._keys = [primary_key] logger.info("[APIKeyPool] %s: %d keys loaded", name, len(self._keys)) @property def primary(self): return self._keys[0] if self._keys else None def next_key(self): if not self._keys: return None available = [k for k in self._keys if k not in self._exhausted] if not available: self._exhausted.clear() available = self._keys key = available[self._index % len(available)] self._index += 1 return key def mark_exhausted(self, key): self._exhausted.add(key) remaining = len(self._keys) - len(self._exhausted) logger.warning("[APIKeyPool] %s: key ...%s exhausted (%d remaining)", self._name, key[-8:], remaining) def reset(self): self._exhausted.clear() self._index = 0 @property def pool_size(self): return len(self._keys) @property def available_count(self): return len(self._keys) - len(self._exhausted) twelve_keys = [os.getenv(f"TWELVEDATA_API_KEY_{i}") for i in range(1, 9)] twelvedata_pool = APIKeyPool( settings.twelvedata_api_key, ",".join([k for k in twelve_keys if k]), name="TwelveData", ) finnhub_keys = [os.getenv(f"FINNHUB_API_KEY_{i}") for i in range(1, 9)] finnhub_pool = APIKeyPool( settings.finnhub_api_key, ",".join([k for k in finnhub_keys if k]), name="Finnhub", ) # Backward compat: keep single-key globals pointing to primary TWELVEDATA_API_KEY = twelvedata_pool.primary FINNHUB_API_KEY = finnhub_pool.primary # Binance, Bybit, CoinGecko, yfinance, FRED — no key or optional key required # ────────────────────────────────────────────────────────────────────────────── # Constants # ────────────────────────────────────────────────────────────────────────────── SUPPORTED_INTERVALS: frozenset = frozenset({"1m", "5m", "15m", "1h", "4h", "1d", "1w"}) INTERVAL_ORDER: List[str] = ["1m", "5m", "15m", "1h", "4h", "1d", "1w"] TWELVE_INTERVAL_MAP: Dict[str, str] = { "1m": "1min", "5m": "5min", "15m": "15min", "1h": "1h", "4h": "4h", "1d": "1day", "1w": "1week", } FINNHUB_RESOLUTION_MAP: Dict[str, str] = { "1m": "1", "5m": "5", "15m": "15", "1h": "60","4h": "240","1d": "D", "1w": "W", } YF_INTERVAL_MAP: Dict[str, str] = { "1m": "1m", "5m": "5m", "15m": "15m", "1h": "60m","4h": "60m","1d": "1d", "1w": "1wk", } YF_PERIOD_MAP: Dict[str, str] = { "1m": "7d", "5m": "60d","15m": "60d", "1h": "730d","4h": "60d","1d": "max","1w": "max", } BINANCE_INTERVAL_MAP: Dict[str, str] = { "1m": "1m", "5m": "5m", "15m": "15m", "1h": "1h", "4h": "4h","1d": "1d", "1w": "1w", } BYBIT_INTERVAL_MAP: Dict[str, str] = { "1m": "1", "5m": "5", "15m": "15", "1h": "60", "4h": "240","1d": "D", "1w": "W", } STEP_SECONDS: Dict[str, int] = { "1m": 60, "5m": 300, "15m": 900, "1h": 3600, "4h": 14400,"1d": 86400,"1w": 604800, } # Source priority by asset category (first available mapping wins) CATEGORY_SOURCE_PRIORITY: Dict[str, List[str]] = { "Crypto": ["binance", "bybit", "coingecko", "yfinance", "finnhub"], "Cặp tiền": ["twelvedata", "finnhub", "yfinance"], "Real Strength": ["yfinance", "twelvedata", "finnhub"], "Kim loại": ["binance", "twelvedata", "yfinance", "finnhub"], "Năng lượng": ["twelvedata", "yfinance", "finnhub"], "Nông sản": ["twelvedata", "yfinance"], "Nguyên liệu CN": ["twelvedata", "yfinance"], "Chỉ số": ["binance", "twelvedata", "yfinance", "finnhub"], "Cổ phiếu Mỹ": ["binance", "twelvedata", "finnhub", "yfinance"], "Cổ phiếu VN": ["yfinance"], "Trái phiếu": ["twelvedata", "yfinance"], "ETF": ["binance", "twelvedata", "yfinance", "finnhub"], } # 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 ANALYSIS_CONTEXT = 750 FORECAST_CONTEXT = 1500 HIGHER_TIMEFRAME_CONTEXT = 240 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) # ────────────────────────────────────────────────────────────────────────────── class TokenBucket: """Simple async token-bucket rate limiter.""" def __init__(self, rate: float, capacity: int) -> None: self._rate = rate # tokens per second self._capacity = capacity self._tokens = float(capacity) self._last = time.monotonic() async def acquire(self, tokens: int = 1) -> None: while True: now = time.monotonic() elapsed = now - self._last self._tokens = min(self._capacity, self._tokens + elapsed * self._rate) self._last = now if self._tokens >= tokens: self._tokens -= tokens return await asyncio.sleep(0.1) # Per-source buckets (conservative — stays well within free limits) _TWELVEDATA_POOL_SIZE = max(1, twelvedata_pool.pool_size) _TWELVEDATA_RATE = min(1.0, 0.1 * _TWELVEDATA_POOL_SIZE) _TWELVEDATA_CAPACITY = max(2, min(8, _TWELVEDATA_POOL_SIZE)) _rate_limiters: Dict[str, TokenBucket] = { "binance": TokenBucket(rate=10.0, capacity=20), "bybit": TokenBucket(rate=1.5, capacity=5), "coingecko": TokenBucket(rate=0.4, capacity=3), "twelvedata": TokenBucket(rate=_TWELVEDATA_RATE, capacity=_TWELVEDATA_CAPACITY), "finnhub": TokenBucket(rate=1.0, capacity=5), "yfinance": TokenBucket(rate=5.0, capacity=10), "alphavantage":TokenBucket(rate=0.02, capacity=1), # 25/day } async def _rate_limit(source: str) -> None: bucket = _rate_limiters.get(source) if bucket: await bucket.acquire() async def _retry(coro_fn, retries: int = 3, base_delay: float = 1.0): """Exponential backoff with jitter for transient failures.""" for attempt in range(retries): try: return await coro_fn() except (httpx.ReadTimeout, httpx.ConnectTimeout, httpx.RemoteProtocolError, httpx.ConnectError, httpx.PoolTimeout): if attempt == retries - 1: raise delay = base_delay * (2 ** attempt) + (0.1 * attempt) await asyncio.sleep(delay) # ────────────────────────────────────────────────────────────────────────────── # Symbol Registry # ────────────────────────────────────────────────────────────────────────────── @dataclass class SymbolConfig: symbol: str label: str label_en: str category: str mappings: Dict[str, str] coingecko_id: Optional[str] = None # for CoinGecko OHLCV bybit_category: str = "linear" # "linear" or "spot" binance_type: str = "spot" # "spot" or "futures" description: str = "" @dataclass(frozen=True) class SyntheticComponentSpec: name: str mode: str left_symbol: str right_symbol: Optional[str] = None weight: float = 1.0 enabled: bool = True @dataclass(frozen=True) class SyntheticSymbolConfig: symbol: str scale: float alpha: float wick_shrink: float components: Tuple[SyntheticComponentSpec, ...] # ─── Helper to build entry quickly ──────────────────────────────────────────── def _s(sym: str, label: str, label_en: str, cat: str, mappings: Dict[str, str], cg_id: str = None, desc: str = "", bybit_cat: str = "linear", bin_type: str = "spot") -> SymbolConfig: return SymbolConfig(sym, label, label_en, cat, mappings, cg_id, bybit_cat, bin_type, desc) SYMBOLS: Dict[str, SymbolConfig] = { # ══════════════════════════════════════════════════════════════════════ # 1. KIM LOẠI (Metals) # ══════════════════════════════════════════════════════════════════════ "XAUUSD": _s("XAUUSD","Vàng (Gold)","Gold","Kim loại",{"binance":"PAXGUSDT","twelvedata":"XAU/USD","yfinance":"GC=F"}), "XAGUSD": _s("XAGUSD","Bạc (Silver)","Silver","Kim loại",{"twelvedata":"XAG/USD","yfinance":"SI=F"}), "XPTUSD": _s("XPTUSD","Bạch kim (Platinum)","Platinum","Kim loại",{"twelvedata":"XPT/USD","yfinance":"PL=F"}), "XPDUSD": _s("XPDUSD","Palladium","Palladium","Kim loại",{"twelvedata":"XPD/USD","yfinance":"PA=F"}), "HGUSD": _s("HGUSD","Đồng (Copper)","Copper","Kim loại",{"twelvedata":"COPPER","yfinance":"HG=F"}), "ALUSD": _s("ALUSD","Nhôm (Aluminum)","Aluminum","Kim loại",{"yfinance":"AL=F"}), "ZNUSD": _s("ZNUSD","Kẽm (Zinc)","Zinc","Kim loại",{"yfinance":"ZN=F"}), "NIUSD": _s("NIUSD","Niken (Nickel)","Nickel","Kim loại",{"yfinance":"NI=F"}), "PBUSD": _s("PBUSD","Chì (Lead)","Lead","Kim loại",{"yfinance":"LED=F"}), "SNUSD": _s("SNUSD","Thiếc (Tin)","Tin","Kim loại",{"yfinance":"SN=F"}), # ══════════════════════════════════════════════════════════════════════ # 2. NĂNG LƯỢNG (Energy) # ══════════════════════════════════════════════════════════════════════ "USOIL": _s("USOIL","Dầu thô WTI","WTI Oil","Năng lượng",{"twelvedata":"WTI","yfinance":"CL=F"}), "UKOIL": _s("UKOIL","Dầu Brent","Brent Oil","Năng lượng",{"twelvedata":"BRENT","yfinance":"BZ=F"}), "NGAS": _s("NGAS","Khí tự nhiên","Natural Gas","Năng lượng",{"twelvedata":"NATGAS","yfinance":"NG=F"}), "GASO": _s("GASO","Xăng RBOB","Gasoline","Năng lượng",{"yfinance":"RB=F"}), "HEAT": _s("HEAT","Dầu sưởi","Heating Oil","Năng lượng",{"yfinance":"HO=F"}), "COAL": _s("COAL","Than (Coal)","Coal","Năng lượng",{"yfinance":"MTF=F"}), # ══════════════════════════════════════════════════════════════════════ # 3. NGUYÊN LIỆU CÔNG NGHIỆP (Industrial Materials) # ══════════════════════════════════════════════════════════════════════ "COFFEE": _s("COFFEE","Cà phê Arabica","Arabica Coffee","Nguyên liệu CN",{"twelvedata":"COFFEE","yfinance":"KC=F"}), "ROBUSTA":_s("ROBUSTA","Cà phê Robusta","Robusta Coffee","Nguyên liệu CN",{"yfinance":"RC=F"}), "COCOA": _s("COCOA","Ca cao (Cocoa)","Cocoa","Nguyên liệu CN",{"yfinance":"CC=F"}), "SUGAR11":_s("SUGAR11","Đường 11","Sugar No.11","Nguyên liệu CN",{"twelvedata":"SUGAR","yfinance":"SB=F"}), "LSUGAR": _s("LSUGAR","Đường trắng","White Sugar","Nguyên liệu CN",{"yfinance":"LSU.L"}), "PALMOIL":_s("PALMOIL","Dầu cọ thô","Palm Oil","Nguyên liệu CN",{"yfinance":"FCPO=F"}), "COTTON": _s("COTTON","Bông (Cotton)","Cotton","Nguyên liệu CN",{"twelvedata":"COTTON","yfinance":"CT=F"}), "RUBBER": _s("RUBBER","Cao su RSS3","Rubber","Nguyên liệu CN",{"yfinance":"JR=F"}), "ORANGE": _s("ORANGE","Nước cam","Orange Juice","Nguyên liệu CN",{"yfinance":"OJ=F"}), # ══════════════════════════════════════════════════════════════════════ # 4. NÔNG SẢN (Agriculture) # ══════════════════════════════════════════════════════════════════════ "CORN": _s("CORN","Ngô (Corn)","Corn","Nông sản",{"twelvedata":"CORN","yfinance":"ZC=F"}), "WHEAT": _s("WHEAT","Lúa mì (Wheat)","Wheat","Nông sản",{"twelvedata":"WHEAT","yfinance":"ZW=F"}), "SOY": _s("SOY","Đậu tương","Soybeans","Nông sản",{"twelvedata":"SOYBEAN","yfinance":"ZS=F"}), "SOYOIL": _s("SOYOIL","Dầu đậu tương","Soybean Oil","Nông sản",{"yfinance":"ZL=F"}), "SOYMEAL":_s("SOYMEAL","Khô đậu tương","Soybean Meal","Nông sản",{"yfinance":"ZM=F"}), "RICE": _s("RICE","Gạo (Rice)","Rice","Nông sản",{"yfinance":"ZR=F"}), "OATS": _s("OATS","Yến mạch (Oats)","Oats","Nông sản",{"yfinance":"ZO=F"}), # ══════════════════════════════════════════════════════════════════════ # 5. CRYPTO (Top 50+) # ══════════════════════════════════════════════════════════════════════ "BTCUSD": _s("BTCUSD","Bitcoin (BTC)","Bitcoin","Crypto",{"binance":"BTCUSDT","bybit":"BTCUSDT","coingecko":"bitcoin"}), "ETHUSD": _s("ETHUSD","Ethereum (ETH)","Ethereum","Crypto",{"binance":"ETHUSDT","bybit":"ETHUSDT","coingecko":"ethereum"}), "BNBUSD": _s("BNBUSD","BNB (BNB)","BNB","Crypto",{"binance":"BNBUSDT","bybit":"BNBUSDT","coingecko":"binancecoin"}), "SOLUSD": _s("SOLUSD","Solana (SOL)","Solana","Crypto",{"binance":"SOLUSDT","bybit":"SOLUSDT","coingecko":"solana"}), "XRPUSD": _s("XRPUSD","Ripple (XRP)","XRP","Crypto",{"binance":"XRPUSDT","bybit":"XRPUSDT","coingecko":"ripple"}), "ADAUSD": _s("ADAUSD","Cardano (ADA)","Cardano","Crypto",{"binance":"ADAUSDT","bybit":"ADAUSDT","coingecko":"cardano"}), "AVAXUSD": _s("AVAXUSD","Avalanche (AVAX)","Avalanche","Crypto",{"binance":"AVAXUSDT","bybit":"AVAXUSDT","coingecko":"avalanche-2"}), "DOTUSD": _s("DOTUSD","Polkadot (DOT)","Polkadot","Crypto",{"binance":"DOTUSDT","bybit":"DOTUSDT","coingecko":"polkadot"}), "MATICUSD": _s("MATICUSD","Polygon (MATIC)","Polygon","Crypto",{"binance":"MATICUSDT","bybit":"MATICUSDT","coingecko":"matic-network"}), "LINKUSD": _s("LINKUSD","Chainlink (LINK)","Chainlink","Crypto",{"binance":"LINKUSDT","bybit":"LINKUSDT","coingecko":"chainlink"}), "LTCUSD": _s("LTCUSD","Litecoin (LTC)","Litecoin","Crypto",{"binance":"LTCUSDT","bybit":"LTCUSDT","coingecko":"litecoin"}), "UNIUSD": _s("UNIUSD","Uniswap (UNI)","Uniswap","Crypto",{"binance":"UNIUSDT","bybit":"UNIUSDT","coingecko":"uniswap"}), "ATOMUSD": _s("ATOMUSD","Cosmos (ATOM)","Cosmos","Crypto",{"binance":"ATOMUSDT"}), "NEARUSD": _s("NEARUSD","NEAR Protocol","NEAR","Crypto",{"binance":"NEARUSDT"}), "APTUSD": _s("APTUSD","Aptos (APT)","Aptos","Crypto",{"binance":"APTUSDT"}), "SUIUSD": _s("SUIUSD","Sui (SUI)","Sui","Crypto",{"binance":"SUIUSDT"}), "ARBUSD": _s("ARBUSD","Arbitrum (ARB)","Arbitrum","Crypto",{"binance":"ARBUSDT"}), "OPUSD": _s("OPUSD","Optimism (OP)","Optimism","Crypto",{"binance":"OPUSDT"}), "TONUSD": _s("TONUSD","Toncoin (TON)","Toncoin","Crypto",{"binance":"TONUSDT"}), "TRXUSD": _s("TRXUSD","Tron (TRX)","Tron","Crypto",{"binance":"TRXUSDT"}), "DOGEUSD":_s("DOGEUSD","Dogecoin (DOGE)","Dogecoin","Crypto",{"binance":"DOGEUSDT"}), "SHIBUSD":_s("SHIBUSD","Shiba Inu (SHIB)","Shiba Inu","Crypto",{"binance":"SHIBUSDT"}), "PEPEUSD":_s("PEPEUSD","PEPE Coin","PEPE","Crypto",{"binance":"PEPEUSDT"}), "FLOKIUSD":_s("FLOKIUSD","FLOKI Inu","FLOKI","Crypto",{"binance":"FLOKIUSDT"}), "BONKUSD":_s("BONKUSD","BONK Coin","BONK","Crypto",{"binance":"BONKUSDT"}), "WIFUSD": _s("WIFUSD","dogwifhat (WIF)","WIF","Crypto",{"binance":"WIFUSDT"}), "RNDRUSD":_s("RNDRUSD","Render (RNDR)","Render","Crypto",{"binance":"RNDRUSDT"}), "FETUSD": _s("FETUSD","Fetch.ai (FET)","Fetch.ai","Crypto",{"binance":"FETUSDT"}), "STXUSD": _s("STXUSD","Stacks (STX)","Stacks","Crypto",{"binance":"STXUSDT"}), "FILUSD": _s("FILUSD","Filecoin (FIL)","Filecoin","Crypto",{"binance":"FILUSDT"}), "VETUSD": _s("VETUSD","VeChain (VET)","VeChain","Crypto",{"binance":"VETUSDT"}), "ICPUSD": _s("ICPUSD","Internet Computer","ICP","Crypto",{"binance":"ICPUSDT"}), # ══════════════════════════════════════════════════════════════════════ # 6. CẶP TIỀN (Forex) # ══════════════════════════════════════════════════════════════════════ "DXY": _s("DXY","Chỉ số USD (DXY)","USD Index","Real Strength",{"twelvedata":"DXY","yfinance":"DX-Y.NYB"}), "USDX": _s("USDX","Sức mạnh USD (USDx)","USD Strength (USDx)","Real Strength",{"synthetic":"USDX"}, desc="Synthetic USD strength index built from weighted USD crosses."), "EURX": _s("EURX","Sức mạnh EUR (EURx)","EUR Strength (EURx)","Real Strength",{"synthetic":"EURX"}, desc="Synthetic EUR strength index built from weighted EUR crosses."), "GBPX": _s("GBPX","Sức mạnh GBP (GBPx)","GBP Strength (GBPx)","Real Strength",{"synthetic":"GBPX"}, desc="Synthetic GBP strength index built from weighted GBP crosses."), "CHFX": _s("CHFX","Sức mạnh CHF (CHFx)","CHF Strength (CHFx)","Real Strength",{"synthetic":"CHFX"}, desc="Synthetic CHF strength index built from weighted CHF crosses."), "JPYX": _s("JPYX","Sức mạnh JPY (JPYx)","JPY Strength (JPYx)","Real Strength",{"synthetic":"JPYX"}, desc="Synthetic JPY strength index built from weighted JPY crosses."), "CADX": _s("CADX","Sức mạnh CAD (CADx)","CAD Strength (CADx)","Real Strength",{"synthetic":"CADX"}, desc="Synthetic CAD strength index built from weighted CAD crosses."), "AUDX": _s("AUDX","Sức mạnh AUD (AUDx)","AUD Strength (AUDx)","Real Strength",{"synthetic":"AUDX"}, desc="Synthetic AUD strength index built from weighted AUD crosses."), "NZDX": _s("NZDX","Sức mạnh NZD (NZDx)","NZD Strength (NZDx)","Real Strength",{"synthetic":"NZDX"}, desc="Synthetic NZD strength index built from weighted NZD crosses."), "EURUSD": _s("EURUSD","EUR/USD","EUR/USD","Cặp tiền",{"twelvedata":"EUR/USD","yfinance":"EURUSD=X"}), "GBPUSD": _s("GBPUSD","GBP/USD","GBP/USD","Cặp tiền",{"twelvedata":"GBP/USD","yfinance":"GBPUSD=X"}), "USDJPY": _s("USDJPY","USD/JPY","USD/JPY","Cặp tiền",{"twelvedata":"USD/JPY","yfinance":"JPY=X"}), "USDCHF": _s("USDCHF","USD/CHF","USD/CHF","Cặp tiền",{"twelvedata":"USD/CHF","yfinance":"CHF=X"}), "AUDUSD": _s("AUDUSD","AUD/USD","AUD/USD","Cặp tiền",{"twelvedata":"AUD/USD","yfinance":"AUDUSD=X"}), "USDCAD": _s("USDCAD","USD/CAD","USD/CAD","Cặp tiền",{"twelvedata":"USD/CAD","yfinance":"CAD=X"}), "NZDUSD": _s("NZDUSD","NZD/USD","NZD/USD","Cặp tiền",{"twelvedata":"NZD/USD","yfinance":"NZDUSD=X"}), "EURGBP": _s("EURGBP","EUR/GBP","EUR/GBP","Cặp tiền",{"twelvedata":"EUR/GBP","yfinance":"EURGBP=X"}), "EURJPY": _s("EURJPY","EUR/JPY","EUR/JPY","Cặp tiền",{"twelvedata":"EUR/JPY","yfinance":"EURJPY=X"}), "EURCHF": _s("EURCHF","EUR/CHF","EUR/CHF","Cặp tiền",{"twelvedata":"EUR/CHF","yfinance":"EURCHF=X"}), "EURCAD": _s("EURCAD","EUR/CAD","EUR/CAD","Cặp tiền",{"twelvedata":"EUR/CAD","yfinance":"EURCAD=X"}), "EURAUD": _s("EURAUD","EUR/AUD","EUR/AUD","Cặp tiền",{"twelvedata":"EUR/AUD","yfinance":"EURAUD=X"}), "EURNZD": _s("EURNZD","EUR/NZD","EUR/NZD","Cặp tiền",{"twelvedata":"EUR/NZD","yfinance":"EURNZD=X"}), "GBPJPY": _s("GBPJPY","GBP/JPY","GBP/JPY","Cặp tiền",{"twelvedata":"GBP/JPY","yfinance":"GBPJPY=X"}), "GBPCHF": _s("GBPCHF","GBP/CHF","GBP/CHF","Cặp tiền",{"twelvedata":"GBP/CHF","yfinance":"GBPCHF=X"}), "GBPCAD": _s("GBPCAD","GBP/CAD","GBP/CAD","Cặp tiền",{"twelvedata":"GBP/CAD","yfinance":"GBPCAD=X"}), "GBPAUD": _s("GBPAUD","GBP/AUD","GBP/AUD","Cặp tiền",{"twelvedata":"GBP/AUD","yfinance":"GBPAUD=X"}), "GBPNZD": _s("GBPNZD","GBP/NZD","GBP/NZD","Cặp tiền",{"twelvedata":"GBP/NZD","yfinance":"GBPNZD=X"}), "CHFJPY": _s("CHFJPY","CHF/JPY","CHF/JPY","Cặp tiền",{"twelvedata":"CHF/JPY","yfinance":"CHFJPY=X"}), "CADCHF": _s("CADCHF","CAD/CHF","CAD/CHF","Cặp tiền",{"twelvedata":"CAD/CHF","yfinance":"CADCHF=X"}), "AUDCHF": _s("AUDCHF","AUD/CHF","AUD/CHF","Cặp tiền",{"twelvedata":"AUD/CHF","yfinance":"AUDCHF=X"}), "NZDCHF": _s("NZDCHF","NZD/CHF","NZD/CHF","Cặp tiền",{"twelvedata":"NZD/CHF","yfinance":"NZDCHF=X"}), "CADJPY": _s("CADJPY","CAD/JPY","CAD/JPY","Cặp tiền",{"twelvedata":"CAD/JPY","yfinance":"CADJPY=X"}), "AUDJPY": _s("AUDJPY","AUD/JPY","AUD/JPY","Cặp tiền",{"twelvedata":"AUD/JPY","yfinance":"AUDJPY=X"}), "NZDJPY": _s("NZDJPY","NZD/JPY","NZD/JPY","Cặp tiền",{"twelvedata":"NZD/JPY","yfinance":"NZDJPY=X"}), "AUDCAD": _s("AUDCAD","AUD/CAD","AUD/CAD","Cặp tiền",{"twelvedata":"AUD/CAD","yfinance":"AUDCAD=X"}), "NZDCAD": _s("NZDCAD","NZD/CAD","NZD/CAD","Cặp tiền",{"twelvedata":"NZD/CAD","yfinance":"NZDCAD=X"}), "AUDNZD": _s("AUDNZD","AUD/NZD","AUD/NZD","Cặp tiền",{"twelvedata":"AUD/NZD","yfinance":"AUDNZD=X"}), "USDVND": _s("USDVND","USD/VND","USD/VND","Cặp tiền",{"yfinance":"VND=X"}), "USDCNH": _s("USDCNH","USD/CNH","USD/CNH","Cặp tiền",{"twelvedata":"USD/CNH"}), "USDHKD": _s("USDHKD","USD/HKD","USD/HKD","Cặp tiền",{"twelvedata":"USD/HKD"}), "USDSGD": _s("USDSGD","USD/SGD","USD/SGD","Cặp tiền",{"twelvedata":"USD/SGD"}), # ══════════════════════════════════════════════════════════════════════ # 7. CHỈ SỐ THẾ GIỚI (Global Indices) # ══════════════════════════════════════════════════════════════════════ "SP500": _s("SP500","S&P 500","S&P 500","Chỉ số",{"yfinance":"^GSPC","twelvedata":"SPX"}, bin_type="futures"), "NASDAQ100":_s("NASDAQ100","Nasdaq 100","Nasdaq 100","Chỉ số",{"yfinance":"^NDX","twelvedata":"NDX"}, bin_type="futures"), "DOW30": _s("DOW30","Dow Jones 30","Dow Jones","Chỉ số",{"yfinance":"^DJI"}), "RUSSELL2000":_s("RUSSELL2000","Russell 2000","Russell 2000","Chỉ số",{"yfinance":"^RUT"}), "VIX": _s("VIX","Chỉ số sợ hãi (VIX)","VIX Index","Chỉ số",{"yfinance":"^VIX"}), "UK100": _s("UK100","FTSE 100 (UK)","FTSE 100","Chỉ số",{"yfinance":"^FTSE"}), "DAX40": _s("DAX40","DAX 40 (Đức)","DAX 40","Chỉ số",{"yfinance":"^GDAXI"}), "CAC40": _s("CAC40","CAC 40 (Pháp)","CAC 40","Chỉ số",{"yfinance":"^FCHI"}), "EU50": _s("EU50","Euro Stoxx 50","Euro Stoxx 50","Chỉ số",{"yfinance":"^STOXX50E"}), "NIKKEI225":_s("NIKKEI225","Nikkei 225 (Nhật)","Nikkei 225","Chỉ số",{"yfinance":"^N225"}), "HSI": _s("HSI","Hang Seng (HK)","Hang Seng","Chỉ số",{"yfinance":"^HSI"}), "VNINDEX": _s("VNINDEX","VN-Index","VN-Index","Chỉ số",{"yfinance":"^VNINDEX"}), "HNXINDEX": _s("HNXINDEX","HNX-Index","HNX-Index","Chỉ số",{"yfinance":"^HNX"}), # ══════════════════════════════════════════════════════════════════════ # 8. CỔ PHIẾU MỸ (US Stocks) # ══════════════════════════════════════════════════════════════════════ "AAPL": _s("AAPL","Apple Inc.","Apple","Cổ phiếu Mỹ",{"yfinance":"AAPL"}), "MSFT": _s("MSFT","Microsoft Corp.","Microsoft","Cổ phiếu Mỹ",{"yfinance":"MSFT"}), "GOOGL": _s("GOOGL","Alphabet (Google)","Google","Cổ phiếu Mỹ",{"yfinance":"GOOGL"}), "AMZN": _s("AMZN","Amazon.com","Amazon","Cổ phiếu Mỹ",{"yfinance":"AMZN"}), "NVDA": _s("NVDA","Nvidia Corp.","Nvidia","Cổ phiếu Mỹ",{"yfinance":"NVDA"}), "META": _s("META","Meta Platforms","Meta","Cổ phiếu Mỹ",{"yfinance":"META"}), "TSLA": _s("TSLA","Tesla Inc.","Tesla","Cổ phiếu Mỹ",{"yfinance":"TSLA"}), "AMD": _s("AMD","AMD","AMD","Cổ phiếu Mỹ",{"yfinance":"AMD"}), "INTC": _s("INTC","Intel Corp.","Intel","Cổ phiếu Mỹ",{"yfinance":"INTC"}), "TSM": _s("TSM","TSMC","TSMC","Cổ phiếu Mỹ",{"yfinance":"TSM"}), "AVGO": _s("AVGO","Broadcom","Broadcom","Cổ phiếu Mỹ",{"yfinance":"AVGO"}), "NFLX": _s("NFLX","Netflix","Netflix","Cổ phiếu Mỹ",{"yfinance":"NFLX"}), "COIN": _s("COIN","Coinbase","Coinbase","Cổ phiếu Mỹ",{"yfinance":"COIN"}), "MSTR": _s("MSTR","MicroStrategy","MicroStrategy","Cổ phiếu Mỹ",{"yfinance":"MSTR"}), "JPM": _s("JPM","JPMorgan Chase","JPMorgan","Cổ phiếu Mỹ",{"yfinance":"JPM"}), "GS": _s("GS","Goldman Sachs","Goldman Sachs","Cổ phiếu Mỹ",{"yfinance":"GS"}), "V": _s("V","Visa Inc.","Visa","Cổ phiếu Mỹ",{"yfinance":"V"}), "MA": _s("MA","Mastercard","Mastercard","Cổ phiếu Mỹ",{"yfinance":"MA"}), "WMT": _s("WMT","Walmart","Walmart","Cổ phiếu Mỹ",{"yfinance":"WMT"}), "XOM": _s("XOM","ExxonMobil","ExxonMobil","Cổ phiếu Mỹ",{"yfinance":"XOM"}), # ══════════════════════════════════════════════════════════════════════ # 9. CỔ PHIẾU VIỆT NAM (Vietnam Stocks) # ══════════════════════════════════════════════════════════════════════ "VCB": _s("VCB","Vietcombank (VCB)","Vietcombank","Cổ phiếu VN",{"yfinance":"VCB.VN"}), "FPT": _s("FPT","FPT Corp (FPT)","FPT","Cổ phiếu VN",{"yfinance":"FPT.VN"}), "VIC": _s("VIC","Vingroup (VIC)","Vingroup","Cổ phiếu VN",{"yfinance":"VIC.VN"}), "VHM": _s("VHM","Vinhomes (VHM)","Vinhomes","Cổ phiếu VN",{"yfinance":"VHM.VN"}), "HPG": _s("HPG","Hòa Phát (HPG)","Hoa Phat Steel","Cổ phiếu VN",{"yfinance":"HPG.VN"}), "SSI": _s("SSI","Chứng khoán SSI","SSI","Cổ phiếu VN",{"yfinance":"SSI.VN"}), "VND": _s("VND","VNDirect (VND)","VNDirect","Cổ phiếu VN",{"yfinance":"VND.VN"}), "MWG": _s("MWG","Thế Giới Di Động","MWG","Cổ phiếu VN",{"yfinance":"MWG.VN"}), "VNM": _s("VNM","Vinamilk (VNM)","Vinamilk","Cổ phiếu VN",{"yfinance":"VNM.VN"}), "MSN": _s("MSN","Masan Group","Masan","Cổ phiếu VN",{"yfinance":"MSN.VN"}), "GAS": _s("GAS","PV GAS","PV GAS","Cổ phiếu VN",{"yfinance":"GAS.VN"}), "BID": _s("BID","BIDV Bank","BIDV","Cổ phiếu VN",{"yfinance":"BID.VN"}), "CTG": _s("CTG","VietinBank","VietinBank","Cổ phiếu VN",{"yfinance":"CTG.VN"}), "TCB": _s("TCB","Techcombank","Techcombank","Cổ phiếu VN",{"yfinance":"TCB.VN"}), "MBB": _s("MBB","MB Bank","MB Bank","Cổ phiếu VN",{"yfinance":"MBB.VN"}), "STB": _s("STB","Sacombank","Sacombank","Cổ phiếu VN",{"yfinance":"STB.VN"}), "ACB": _s("ACB","ACB Bank","ACB","Cổ phiếu VN",{"yfinance":"ACB.VN"}), "HDB": _s("HDB","HDBank","HDBank","Cổ phiếu VN",{"yfinance":"HDB.VN"}), "VPB": _s("VPB","VPBank","VPBank","Cổ phiếu VN",{"yfinance":"VPB.VN"}), "DGC": _s("DGC","Hóa chất Đức Giang","DGC","Cổ phiếu VN",{"yfinance":"DGC.VN"}), # ══════════════════════════════════════════════════════════════════════ # 10. TRÁI PHIẾU & ETF # ══════════════════════════════════════════════════════════════════════ "US10Y": _s("US10Y","Trái phiếu Mỹ 10Y","US 10Y Treasury","Trái phiếu",{"yfinance":"^TNX"}), "US02Y": _s("US02Y","Trái phiếu Mỹ 2Y","US 2Y Treasury","Trái phiếu",{"yfinance":"^IRX"}), "SPY": _s("SPY","SPDR S&P 500 ETF","SPY","ETF",{"yfinance":"SPY"}), "QQQ": _s("QQQ","Invesco QQQ Trust","QQQ","ETF",{"yfinance":"QQQ"}), "DIA": _s("DIA","Dow Jones ETF","DIA","ETF",{"yfinance":"DIA"}), } CRYPTO_QUOTE_SUFFIXES: Tuple[str, ...] = ("USDT", "USDC", "USD", "BTC", "ETH") HF_CRYPTO_SOURCE_PRIORITY_DAILY: List[str] = [ "yfinance", "twelvedata", "finnhub", "coingecko", "binance", "bybit", ] HF_CRYPTO_SOURCE_PRIORITY_INTRADAY: List[str] = [ "twelvedata", "yfinance", "finnhub", "coingecko", "binance", "bybit", ] def _split_crypto_symbol(symbol: str) -> Tuple[Optional[str], Optional[str]]: upper_symbol = symbol.upper() for quote in CRYPTO_QUOTE_SUFFIXES: if upper_symbol.endswith(quote) and len(upper_symbol) > len(quote): return upper_symbol[: -len(quote)], quote return None, None def _get_dynamic_crypto_mappings(symbol: str) -> Dict[str, str]: cfg = SYMBOLS[symbol] if cfg.category != "Crypto": return {} base, quote = _split_crypto_symbol(cfg.symbol) if not base or not quote: return {} derived: Dict[str, str] = {} normalized_quote = "USD" if quote == "USDT" else quote binance_symbol = cfg.mappings.get("binance") if "twelvedata" not in cfg.mappings: derived["twelvedata"] = f"{base}/{normalized_quote}" if normalized_quote == "USD" and "yfinance" not in cfg.mappings: derived["yfinance"] = f"{base}-USD" if "finnhub" not in cfg.mappings: if binance_symbol: derived["finnhub"] = f"BINANCE:{binance_symbol}" elif quote in {"USD", "USDT", "USDC"}: finnhub_quote = "USDT" if quote == "USD" else quote derived["finnhub"] = f"BINANCE:{base}{finnhub_quote}" return derived def _get_symbol_mapping(symbol: str, source: str) -> Optional[str]: cfg = SYMBOLS[symbol] direct_mapping = cfg.mappings.get(source) if direct_mapping: return direct_mapping return _get_dynamic_crypto_mappings(symbol).get(source) def _has_source_mapping(symbol: str, source: str) -> bool: return _get_symbol_mapping(symbol, source) is not None # ────────────────────────────────────────────────────────────────────────────── # TTL Cache (unchanged from v3, with improved stats) # ────────────────────────────────────────────────────────────────────────────── SYNTHETIC_SYMBOLS: Dict[str, SyntheticSymbolConfig] = { "USDX": SyntheticSymbolConfig( symbol="USDX", scale=43.0, alpha=1.0, wick_shrink=0.8, components=( SyntheticComponentSpec(name="EURUSD", mode="inverse", left_symbol="EURUSD"), SyntheticComponentSpec(name="GBPUSD", mode="inverse", left_symbol="GBPUSD"), SyntheticComponentSpec(name="USDCHF", mode="direct", left_symbol="USDCHF"), SyntheticComponentSpec(name="USDJPY", mode="direct", left_symbol="USDJPY"), SyntheticComponentSpec(name="USDCAD", mode="direct", left_symbol="USDCAD"), SyntheticComponentSpec(name="AUDUSD", mode="inverse", left_symbol="AUDUSD"), SyntheticComponentSpec(name="NZDUSD", mode="inverse", left_symbol="NZDUSD"), ), ), "EURX": SyntheticSymbolConfig( symbol="EURX", scale=43.0, alpha=1.0, wick_shrink=0.6, components=( SyntheticComponentSpec(name="EURUSD", mode="direct", left_symbol="EURUSD"), SyntheticComponentSpec(name="EURGBP", mode="direct", left_symbol="EURGBP"), SyntheticComponentSpec(name="EURCHF", mode="product", left_symbol="EURUSD", right_symbol="USDCHF"), SyntheticComponentSpec(name="EURJPY", mode="direct", left_symbol="EURJPY"), SyntheticComponentSpec(name="EURCAD", mode="product", left_symbol="EURUSD", right_symbol="USDCAD"), SyntheticComponentSpec(name="EURAUD", mode="ratio", left_symbol="EURUSD", right_symbol="AUDUSD"), SyntheticComponentSpec(name="EURNZD", mode="ratio", left_symbol="EURUSD", right_symbol="NZDUSD"), ), ), "GBPX": SyntheticSymbolConfig( symbol="GBPX", scale=43.0, alpha=1.0, wick_shrink=0.6, components=( SyntheticComponentSpec(name="GBPUSD", mode="direct", left_symbol="GBPUSD"), SyntheticComponentSpec(name="EURGBP", mode="inverse", left_symbol="EURGBP"), SyntheticComponentSpec(name="GBPCHF", mode="product", left_symbol="GBPUSD", right_symbol="USDCHF"), SyntheticComponentSpec(name="GBPJPY", mode="direct", left_symbol="GBPJPY"), SyntheticComponentSpec(name="GBPCAD", mode="product", left_symbol="GBPUSD", right_symbol="USDCAD"), SyntheticComponentSpec(name="GBPAUD", mode="ratio", left_symbol="GBPUSD", right_symbol="AUDUSD"), SyntheticComponentSpec(name="GBPNZD", mode="ratio", left_symbol="GBPUSD", right_symbol="NZDUSD"), ), ), "CHFX": SyntheticSymbolConfig( symbol="CHFX", scale=43.0, alpha=1.0, wick_shrink=0.8, components=( SyntheticComponentSpec(name="USDCHF", mode="inverse", left_symbol="USDCHF"), SyntheticComponentSpec(name="EURCHF", mode="inverse", left_symbol="EURCHF"), SyntheticComponentSpec(name="GBPCHF", mode="inverse", left_symbol="GBPCHF"), SyntheticComponentSpec(name="CHFJPY", mode="direct", left_symbol="CHFJPY"), SyntheticComponentSpec(name="CADCHF", mode="inverse", left_symbol="CADCHF"), SyntheticComponentSpec(name="AUDCHF", mode="inverse", left_symbol="AUDCHF"), SyntheticComponentSpec(name="NZDCHF", mode="inverse", left_symbol="NZDCHF"), ), ), "JPYX": SyntheticSymbolConfig( symbol="JPYX", scale=43.0, alpha=1.0, wick_shrink=1.0, components=( SyntheticComponentSpec(name="USDJPY", mode="inverse", left_symbol="USDJPY"), SyntheticComponentSpec(name="EURJPY", mode="inverse", left_symbol="EURJPY"), SyntheticComponentSpec(name="GBPJPY", mode="inverse", left_symbol="GBPJPY"), SyntheticComponentSpec(name="CHFJPY", mode="inverse", left_symbol="CHFJPY"), SyntheticComponentSpec(name="CADJPY", mode="inverse", left_symbol="CADJPY"), SyntheticComponentSpec(name="AUDJPY", mode="inverse", left_symbol="AUDJPY"), SyntheticComponentSpec(name="NZDJPY", mode="inverse", left_symbol="NZDJPY"), ), ), "CADX": SyntheticSymbolConfig( symbol="CADX", scale=44.0, alpha=1.0, wick_shrink=0.8, components=( SyntheticComponentSpec(name="USDCAD", mode="inverse", left_symbol="USDCAD"), SyntheticComponentSpec(name="EURCAD", mode="inverse", left_symbol="EURCAD"), SyntheticComponentSpec(name="GBPCAD", mode="inverse", left_symbol="GBPCAD"), SyntheticComponentSpec(name="CADCHF", mode="direct", left_symbol="CADCHF"), SyntheticComponentSpec(name="CADJPY", mode="direct", left_symbol="CADJPY"), SyntheticComponentSpec(name="AUDCAD", mode="inverse", left_symbol="AUDCAD"), SyntheticComponentSpec(name="NZDCAD", mode="inverse", left_symbol="NZDCAD"), ), ), "AUDX": SyntheticSymbolConfig( symbol="AUDX", scale=44.0, alpha=1.0, wick_shrink=0.6, components=( SyntheticComponentSpec(name="AUDUSD", mode="direct", left_symbol="AUDUSD"), SyntheticComponentSpec(name="EURAUD", mode="inverse", left_symbol="EURAUD"), SyntheticComponentSpec(name="GBPAUD", mode="inverse", left_symbol="GBPAUD"), SyntheticComponentSpec(name="AUDCHF", mode="direct", left_symbol="AUDCHF"), SyntheticComponentSpec(name="AUDJPY", mode="direct", left_symbol="AUDJPY"), SyntheticComponentSpec(name="AUDCAD", mode="direct", left_symbol="AUDCAD"), SyntheticComponentSpec(name="AUDNZD", mode="direct", left_symbol="AUDNZD"), ), ), "NZDX": SyntheticSymbolConfig( symbol="NZDX", scale=44.0, alpha=1.0, wick_shrink=0.8, components=( SyntheticComponentSpec(name="NZDUSD", mode="direct", left_symbol="NZDUSD"), SyntheticComponentSpec(name="EURNZD", mode="inverse", left_symbol="EURNZD"), SyntheticComponentSpec(name="GBPNZD", mode="inverse", left_symbol="GBPNZD"), SyntheticComponentSpec(name="NZDCHF", mode="direct", left_symbol="NZDCHF"), SyntheticComponentSpec(name="NZDJPY", mode="direct", left_symbol="NZDJPY"), SyntheticComponentSpec(name="NZDCAD", mode="direct", left_symbol="NZDCAD"), SyntheticComponentSpec(name="AUDNZD", mode="inverse", left_symbol="AUDNZD"), ), ), } def _get_canonical_symbol(sym: str) -> str: """Try to find the registry ID for a given symbol or alias.""" s = sym.upper() if s in SYMBOLS: return s # Try common suffixes for suffix in ["USD", "USDT"]: if f"{s}{suffix}" in SYMBOLS: return f"{s}{suffix}" # Try prefix removal if s.startswith("BINANCE:") and s[8:] in SYMBOLS: return s[8:] # Search in mappings for reg_id in SYMBOLS: for source_name in ("binance", "bybit", "coingecko", "twelvedata", "yfinance", "finnhub"): mapping = _get_symbol_mapping(reg_id, source_name) if mapping and s == mapping.upper(): return reg_id return s historical_cache = TTLCache() forecast_cache = TTLCache() ticker_cache = TTLCache() ai_verdict_cache = TTLCache() indicators_cache = TTLCache() source_history_cache = TTLCache() _watchlist_ticker_semaphore = asyncio.Semaphore(WATCHLIST_TICKER_CONCURRENCY) _market_peer_ticker_semaphore = asyncio.Semaphore(MARKET_PEER_TICKER_CONCURRENCY) # Explicitly clear on startup to ensure fresh v6.1+ format historical_cache.clear() forecast_cache.clear() ticker_cache.clear() ai_verdict_cache.clear() indicators_cache.clear() source_history_cache.clear() def _cache_prefix(symbol: str, interval: str) -> str: return f"{CACHE_VERSION}:{symbol}:{interval}:" def interval_ttl(interval: str) -> int: if interval in {"1m", "5m"}: return 20 if interval == "15m": return 30 if interval in {"1h", "4h"}: return 60 return 900 def forecast_ttl(interval: str) -> int: if interval in {"1m", "5m"}: return 60 if interval in {"15m", "1h"}: return 300 if interval == "4h": return 600 return 1800 def indicators_ttl(interval: str) -> int: if interval in {"1m", "5m"}: return 45 if interval in {"15m", "30m"}: return 120 if interval in {"1h", "4h"}: return 300 return 900 def verdict_ttl(interval: str) -> int: if interval in {"1m", "5m"}: return 45 if interval == "15m": return 90 if interval in {"1h", "4h"}: return 300 return 900 def _analysis_verdict_cache_key( symbol: str, interval: str, analysis: Dict[str, Any], forecast_pct: float, ) -> str: summary = analysis.get("summary", {}) payload = { "symbol": symbol, "interval": interval, "summary": { "bias": summary.get("bias"), "signal": summary.get("signal"), "score": round(float(summary.get("score", 0.0) or 0.0), 2), }, "osc_signal": analysis.get("oscillators", {}).get("signal"), "ma_signal": analysis.get("moving_averages", {}).get("signal"), "mtf": analysis.get("multi_timeframe", {}), "forecast_pct": round(float(forecast_pct), 2), "rules_version": ai_rule_registry.version, } digest = hashlib.sha256( json.dumps(payload, ensure_ascii=False, sort_keys=True).encode("utf-8") ).hexdigest() return f"ai_verdict:{CACHE_VERSION}:{symbol}:{interval}:{digest[:24]}" _HISTORICAL_INFLIGHT: Dict[str, "asyncio.Task[Tuple[List[Dict[str, Any]], str]]"] = {} _INDICATORS_INFLIGHT: Dict[str, "asyncio.Task[Dict[str, Any]]"] = {} _FORECAST_INFLIGHT: Dict[str, "asyncio.Task[Dict[str, Any]]"] = {} _TICKER_INFLIGHT: Dict[str, "asyncio.Task[Dict[str, Any]]"] = {} _SOURCE_HISTORY_INFLIGHT: Dict[str, "asyncio.Task[List[Dict[str, Any]]]"] = {} # ────────────────────────────────────────────────────────────────────────────── # Data Helpers # ────────────────────────────────────────────────────────────────────────────── def _parse_timestamp(ts: Any) -> int: if isinstance(ts, (int, float, np.integer, np.floating)): return int(ts // 1000) if ts > 10_000_000_000 else int(ts) if isinstance(ts, str): return int(pd.to_datetime(ts, utc=True).timestamp()) if isinstance(ts, datetime): if ts.tzinfo is None: ts = ts.replace(tzinfo=timezone.utc) return int(ts.timestamp()) raise ValueError(f"Unsupported timestamp type: {type(ts)}") def _normalize_ohlcv(records: List[Dict[str, Any]], interval: str = "1h") -> List[Dict[str, Any]]: """Parse, validate (zombie-candle filter), deduplicate (timestamp-only).""" if interval in {"1w", "1d"}: threshold = 0.005 elif interval in {"4h", "1h"}: threshold = 0.001 else: threshold = 0.0001 # B-1: Dynamic adaptive zombie-candle threshold (BUG-07 Fix) # Reduce threshold further for specific low-volatility asset classes symbol_cfg = None if records and "symbol" in records[0]: # Not always present in raw rows pass normalized: List[Dict[str, Any]] = [] for row in records: try: t = _parse_timestamp(row["time"]) o = float(row["open"]) h = float(row["high"]) l = float(row["low"]) c = float(row["close"]) v = float(row.get("volume") or 0.0) if any(math.isnan(x) or math.isinf(x) for x in [o, h, l, c, v]): continue # Adaptive threshold: permit smaller ranges for shorter timeframes # and specifically for assets where o > 0. # 1e-9 is effectively "almost zero" to catch true dead candles while preserving JPY/Bonds. if o > 0 and (h - l) / o < 1e-9: continue normalized.append({"time": t, "open": o, "high": h, "low": l, "close": c, "volume": v}) except Exception: continue normalized.sort(key=lambda x: x["time"]) seen: set = set() deduped: List[Dict[str, Any]] = [] for r in normalized: if r["time"] in seen: continue seen.add(r["time"]) deduped.append(r) return deduped # ────────────────────────────────────────────────────────────────────────────── # Data Sources # ────────────────────────────────────────────────────────────────────────────── async def fetch_binance(symbol: str, interval: str, limit: int) -> List[Dict[str, Any]]: await _rate_limit("binance") cfg = SYMBOLS[symbol] endpoint_symbol = _get_symbol_mapping(symbol, "binance") if not endpoint_symbol: raise RuntimeError("No Binance mapping for this symbol") # Define request details endpoint = "/fapi/v1/klines" if cfg.binance_type == "futures" else "/api/v3/klines" params = { "symbol": endpoint_symbol, "interval": BINANCE_INTERVAL_MAP.get(interval, "1h"), "limit": min(max(limit, 1), 1000), } # B-2: Endpoint Rotation for HF/Cloud environments endpoints_spot = ["https://api.binance.com", "https://api1.binance.com", "https://api2.binance.com", "https://api3.binance.com", "https://data-api.binance.com"] endpoints_fapi = ["https://fapi.binance.com"] # fapi usually more restricted, but try first selected_endpoints = endpoints_fapi if cfg.binance_type == "futures" else endpoints_spot # If on HF, we know api.binance.com is likely blocked, so we can try data-api or alternates faster if settings.is_hf and cfg.binance_type == "spot": # Move data-api to the front for HF spot selected_endpoints = ["https://data-api.binance.com", "https://api1.binance.com", "https://api2.binance.com", "https://api3.binance.com", "https://api.binance.com"] last_error = None for base_url in selected_endpoints: try: cb = source_breakers["binance"] if not cb.allow_request(): continue # Try next endpoint or fall through async def _do_fetch(): client = await GlobalHTTPClient.get_client() resp = await client.get(f"{base_url}{endpoint}", params=params, timeout=10.0) if resp.status_code == 451: logger.warning("[Binance] Endpoint %s blocked (451). Trying next...", base_url) raise RuntimeError("IP Blocked") if resp.status_code == 429: raise HTTPException(status_code=429, detail="Binance rate limit") resp.raise_for_status() cb.record_success() return resp.json() payload = await _retry(_do_fetch) # If successful, parse and return parsed = [ {"time": int(k[0])//1000, "open": k[1], "high": k[2], "low": k[3], "close": k[4], "volume": k[5]} for k in payload ] return _normalize_ohlcv(parsed, interval)[-limit:] except Exception as ex: last_error = ex logger.error("[Binance] Failed with %s: %s", base_url, ex) continue raise last_error or HTTPException(status_code=503, detail="Binance all endpoints failed") async def fetch_bybit(symbol: str, interval: str, limit: int) -> List[Dict[str, Any]]: """Bybit V5 kline endpoint — free, no API key.""" await _rate_limit("bybit") endpoint_symbol = _get_symbol_mapping(symbol, "bybit") if not endpoint_symbol: raise RuntimeError("No Bybit mapping for this symbol") bybit_cat = SYMBOLS[symbol].bybit_category params = { "category": bybit_cat, "symbol": endpoint_symbol, "interval": BYBIT_INTERVAL_MAP.get(interval, "60"), "limit": min(max(limit, 1), 1000), # Bybit V5 supports up to 1000 } logger.info("[Bybit] %s %s (cat=%s)", symbol, interval, bybit_cat) # B-3: Endpoint Rotation for Bybit bybit_endpoints = ["https://api.bybit.com", "https://api.bytick.com", "https://api.bybit.nl"] if settings.is_hf: # Prefer bytick on HF bybit_endpoints = ["https://api.bytick.com", "https://api.bybit.com", "https://api.bybit.nl"] last_error = None for base_url in bybit_endpoints: try: cb = source_breakers["bybit"] if not cb.allow_request(): continue async def _do_fetch(): client = await GlobalHTTPClient.get_client() url = f"{base_url}/v5/market/kline" headers = {} query_params = params.copy() if settings.bybit_api_key and settings.bybit_api_secret: timestamp = str(int(time.time() * 1000)) recv_window = "5000" sorted_params = "&".join([f"{k}={v}" for k, v in sorted(query_params.items())]) raw_str = timestamp + settings.bybit_api_key + recv_window + sorted_params signature = hmac.new(settings.bybit_api_secret.encode('utf-8'), raw_str.encode('utf-8'), hashlib.sha256).hexdigest() headers = { 'X-BAPI-API-KEY': settings.bybit_api_key, 'X-BAPI-TIMESTAMP': timestamp, 'X-BAPI-SIGN-TYPE': '2', 'X-BAPI-RECV-WINDOW': recv_window, 'X-BAPI-SIGN': signature } resp = await client.get(url, params=query_params, headers=headers, timeout=10.0) if resp.status_code == 403: logger.warning("[Bybit] Endpoint %s forbidden (403). Trying next...", base_url) raise RuntimeError("IP Blocked") resp.raise_for_status() cb.record_success() return resp.json() data = await _retry(_do_fetch) if data.get("retCode", -1) != 0: raise RuntimeError(f"Bybit Error: {data}") rows = data.get("result", {}).get("list", []) parsed = [ {"time": int(r[0])//1000, "open": r[1], "high": r[2], "low": r[3], "close": r[4], "volume": r[5]} for r in rows ] parsed.reverse() return _normalize_ohlcv(parsed, interval)[-limit:] except Exception as ex: last_error = ex logger.error("[Bybit] Failed with %s: %s", base_url, ex) continue raise last_error or HTTPException(status_code=503, detail="Bybit all endpoints failed") async def fetch_coingecko(symbol: str, interval: str, limit: int) -> List[Dict[str, Any]]: """ CoinGecko OHLCV — free, no API key (30 req/min). Days mapping: CoinGecko auto-selects granularity based on days requested. 1-2 days → hourly candles 3-90 days → daily candles (use for 1h/4h fallback) 90+ days → weekly candles """ await _rate_limit("coingecko") cg_id = _get_symbol_mapping(symbol, "coingecko") or SYMBOLS[symbol].coingecko_id if not cg_id: raise RuntimeError("No CoinGecko ID for this symbol") # BUG-03: skip sub-hourly crypto if using coingecko (not supported by free OHLC) if interval in {"1m", "5m", "15m"}: raise RuntimeError(f"CoinGecko does not support {interval} OHLC granularity") # Map interval to days days_map = {"1h": 14, "4h": 90, "1d": 365, "1w": 730} days = days_map.get(interval, 14) logger.info("[CoinGecko] %s %s (days=%d)", symbol, interval, days) async def _fetch(): cb = source_breakers["coingecko"] if not cb.allow_request(): raise HTTPException(status_code=503, detail="CoinGecko circuit is OPEN") try: client = await GlobalHTTPClient.get_client() resp = await client.get( f"https://api.coingecko.com/api/v3/coins/{cg_id}/ohlc", params={"vs_currency": "usd", "days": days}, ) if resp.status_code == 429: cb.record_failure() raise HTTPException(status_code=429, detail="CoinGecko rate limit") if resp.status_code >= 500: cb.record_failure() raise HTTPException(status_code=resp.status_code, detail="CoinGecko server error") cb.record_success() return resp.json() except Exception as ex: cb.record_failure() raise ex rows = await _retry(_fetch) # CoinGecko OHLC: [timestamp_ms, open, high, low, close] parsed = [ {"time": int(r[0])//1000, "open": r[1], "high": r[2], "low": r[3], "close": r[4], "volume": 0.0} for r in rows ] return _normalize_ohlcv(parsed, interval)[-limit:] async def fetch_twelvedata(symbol: str, interval: str, limit: int) -> List[Dict[str, Any]]: await _rate_limit("twelvedata") endpoint_symbol = _get_symbol_mapping(symbol, "twelvedata") if not endpoint_symbol: raise RuntimeError("No TwelveData mapping for this symbol") params = { "symbol": endpoint_symbol, "interval": TWELVE_INTERVAL_MAP[interval], "outputsize": min(max(limit, 30), 5000), "apikey": twelvedata_pool.next_key(), "format": "JSON", } logger.info("[TwelveData] %s %s", symbol, interval) async def _fetch(): cb = source_breakers["twelvedata"] if not cb.allow_request(): raise HTTPException(status_code=503, detail="TwelveData circuit is OPEN") try: client = await GlobalHTTPClient.get_client() resp = await client.get("https://api.twelvedata.com/time_series", params=params) if resp.status_code == 429: cb.record_failure() twelvedata_pool.mark_exhausted(params["apikey"]) cb.open_for(timeout=1800) raise RuntimeError("TwelveData rate limit") if resp.status_code >= 500: cb.record_failure() raise HTTPException(status_code=resp.status_code, detail="TwelveData server error") cb.record_success() return resp.json() except Exception as ex: cb.record_failure() raise ex payload = await _retry(_fetch) if payload.get("code") == 429: cb.open_for(timeout=1800) raise RuntimeError(f"TwelveData error: {payload}") if "code" in payload and payload.get("code") == 400: raise RuntimeError(f"TwelveData error: {payload}") values = payload.get("values", []) parsed = [ {"time": v.get("datetime"), "open": v.get("open"), "high": v.get("high"), "low": v.get("low"), "close": v.get("close"), "volume": v.get("volume", 0)} for v in values ] parsed.reverse() return _normalize_ohlcv(parsed, interval)[-limit:] async def fetch_finnhub(symbol: str, interval: str, limit: int) -> List[Dict[str, Any]]: finnhub_symbol = _get_symbol_mapping(symbol, "finnhub") if not finnhub_symbol: return [] cb = source_breakers.get("finnhub") if cb and not cb.allow_request(): return [] await _rate_limit("finnhub") client = await GlobalHTTPClient.get_client() async def _call(): r = await client.get( "https://finnhub.io/api/v1/stock/candle", params={ "symbol": finnhub_symbol, "resolution": FINNHUB_RESOLUTION_MAP.get(interval, "D"), "count": limit, "token": finnhub_pool.next_key(), }, timeout=10, ) r.raise_for_status() return r.json() try: d = await _retry(_call) if d.get("s") != "ok": if cb: cb.record_success() return [] if cb: cb.record_success() return [ {"time": t, "open": o, "high": h, "low": l, "close": c, "volume": v} for t, o, h, l, c, v in zip(d["t"], d["o"], d["h"], d["l"], d["c"], d["v"]) ] except Exception as ex: if cb: cb.record_failure() logger.warning("[finnhub] %s failed: %s", symbol, ex) return [] def _resample_4h(df: pd.DataFrame) -> pd.DataFrame: if df.empty: return df out = df.resample("4h", label="left", closed="left").agg( {"Open": "first", "High": "max", "Low": "min", "Close": "last", "Volume": "sum"} ) return out.dropna(subset=["Open", "High", "Low", "Close"]) async def fetch_yfinance(symbol: str, interval: str, limit: int) -> List[Dict[str, Any]]: await _rate_limit("yfinance") ticker = _get_symbol_mapping(symbol, "yfinance") if not ticker: raise RuntimeError("No yfinance mapping for this symbol") yf_interval = YF_INTERVAL_MAP[interval] period = YF_PERIOD_MAP[interval] logger.info("[yfinance] %s %s", symbol, interval) def _download() -> pd.DataFrame: # B-5: Robust lookback (v6.0) - ensure EMA200 context p = "2y" if interval == "1d" else "60d" if "m" in interval else "1y" return yf.download(tickers=ticker, interval=yf_interval, period=p, progress=False, auto_adjust=False, threads=False) df = await asyncio.to_thread(_download) if df is None or df.empty: raise RuntimeError("yfinance: empty dataframe") if isinstance(df.columns, pd.MultiIndex): df.columns = df.columns.get_level_values(0) df.index = ( df.index.tz_localize("UTC") if df.index.tz is None else df.index.tz_convert("UTC") ) if interval == "4h": df = _resample_4h(df) parsed = [ {"time": int(idx.timestamp()), "open": row.get("Open"), "high": row.get("High"), "low": row.get("Low"), "close": row.get("Close"), "volume": row.get("Volume", 0)} for idx, row in df.iterrows() ] return _normalize_ohlcv(parsed, interval)[-limit:] async def fetch_alphavantage(symbol: str, interval: str, limit: int) -> List[Dict[str, Any]]: """Alpha Vantage - primarily for Macro/Historical. Needs API key.""" if not settings.alphavantage_api_key: raise ValueError("Alpha Vantage API key missing") await _rate_limit("alphavantage") endpoint_symbol = SYMBOLS[symbol].mappings.get("alphavantage", symbol) # AV has specific functions for daily/intraday func = "TIME_SERIES_INTRADAY" if interval in ["1m", "5m", "15m", "60min"] else "TIME_SERIES_DAILY" url = f"https://www.alphavantage.co/query?function={func}&symbol={endpoint_symbol}&apikey={settings.alphavantage_api_key}&outputsize=full" if "INTRADAY" in func: url += f"&interval={interval if interval != '1h' else '60min'}" async with httpx.AsyncClient(timeout=15) as client: resp = await client.get(url) data = resp.json() # Parse logic based on AV's weird nested JSON key = next((k for k in data.keys() if "Time Series" in k), None) if not key: raise RuntimeError(f"Alpha Vantage: {data.get('Note', data.get('Information', 'Unknown error'))}") series = data[key] parsed = [] for ts, vals in series.items(): parsed.append({ "time": int(datetime.strptime(ts, "%Y-%m-%d %H:%M:%S" if " " in ts else "%Y-%m-%d").replace(tzinfo=timezone.utc).timestamp()), "open": float(vals["1. open"]), "high": float(vals["2. high"]), "low": float(vals["3. low"]), "close": float(vals["4. close"]), "volume": float(vals.get("5. volume", 0)) }) parsed.sort(key=lambda x: x["time"]) return _normalize_ohlcv(parsed, interval)[-limit:] async def _fetch_historical_from_source( source: str, symbol: str, interval: str, limit: int, ) -> List[Dict[str, Any]]: if source == "binance": return await fetch_binance(symbol, interval, limit) if source == "bybit": return await fetch_bybit(symbol, interval, limit) if source == "coingecko": return await fetch_coingecko(symbol, interval, limit) if source == "twelvedata": return await fetch_twelvedata(symbol, interval, limit) if source == "finnhub": return await fetch_finnhub(symbol, interval, limit) if source == "yfinance": return await fetch_yfinance(symbol, interval, limit) if source == "alphavantage": return await fetch_alphavantage(symbol, interval, limit) raise ValueError(f"Unsupported source fetcher: {source}") async def _fetch_historical_from_source_cached( source: str, symbol: str, interval: str, limit: int, ) -> List[Dict[str, Any]]: """ Deduplicate repeated component fetches used by synthetic symbols. This keeps Real Strength baskets from re-downloading the same FX cross multiple times across EURX/GBPX/CHFX/... within one refresh window. """ cache_key = f"source_hist:{CACHE_VERSION}:{source}:{symbol}:{interval}:{limit}" cached = source_history_cache.get(cache_key) if cached is not None: return cached inflight_task = _SOURCE_HISTORY_INFLIGHT.get(cache_key) if inflight_task is not None: return await inflight_task task = asyncio.create_task( _fetch_historical_from_source(source, symbol, interval, limit), name=f"source_hist:{source}:{symbol}:{interval}:{limit}", ) _SOURCE_HISTORY_INFLIGHT[cache_key] = task try: data = await task source_history_cache.set(cache_key, data, ttl_seconds=interval_ttl(interval)) return data finally: if _SOURCE_HISTORY_INFLIGHT.get(cache_key) is task: _SOURCE_HISTORY_INFLIGHT.pop(cache_key, None) def _is_synthetic_symbol(symbol: str) -> bool: return symbol in SYNTHETIC_SYMBOLS def _extract_candle_value(candle: Dict[str, Any], field_name: str) -> float: return float(candle[field_name]) def _combine_component_candles( spec: SyntheticComponentSpec, left_candle: Dict[str, Any], right_candle: Optional[Dict[str, Any]] = None, use_body_only_extrema: bool = False, ) -> Dict[str, float]: if spec.mode == "direct": open_value = _extract_candle_value(left_candle, "open") close_value = _extract_candle_value(left_candle, "close") if use_body_only_extrema: return { "open": open_value, "high": max(open_value, close_value), "low": min(open_value, close_value), "close": close_value, } return { "open": open_value, "high": _extract_candle_value(left_candle, "high"), "low": _extract_candle_value(left_candle, "low"), "close": close_value, } if spec.mode == "inverse": lo = _extract_candle_value(left_candle, "open") lh = _extract_candle_value(left_candle, "high") ll = _extract_candle_value(left_candle, "low") lc = _extract_candle_value(left_candle, "close") if min(lo, lh, ll, lc) <= 0: raise ValueError(f"Synthetic component {spec.name} has non-positive inverse values") open_value = 1.0 / lo close_value = 1.0 / lc if use_body_only_extrema: return { "open": open_value, "high": max(open_value, close_value), "low": min(open_value, close_value), "close": close_value, } return { "open": open_value, "high": 1.0 / ll, "low": 1.0 / lh, "close": close_value, } if right_candle is None: raise ValueError(f"Synthetic component {spec.name} requires a right candle") lo = _extract_candle_value(left_candle, "open") lh = _extract_candle_value(left_candle, "high") ll = _extract_candle_value(left_candle, "low") lc = _extract_candle_value(left_candle, "close") ro = _extract_candle_value(right_candle, "open") rh = _extract_candle_value(right_candle, "high") rl = _extract_candle_value(right_candle, "low") rc = _extract_candle_value(right_candle, "close") if spec.mode == "product": open_value = lo * ro close_value = lc * rc if use_body_only_extrema: return { "open": open_value, "high": max(open_value, close_value), "low": min(open_value, close_value), "close": close_value, } return { "open": open_value, "high": lh * rh, "low": ll * rl, "close": close_value, } if spec.mode == "ratio": if min(ro, rh, rl, rc) <= 0: raise ValueError(f"Synthetic component {spec.name} has non-positive divisor values") open_value = lo / ro close_value = lc / rc if use_body_only_extrema: return { "open": open_value, "high": max(open_value, close_value), "low": min(open_value, close_value), "close": close_value, } return { "open": open_value, "high": lh / rl, "low": ll / rh, "close": close_value, } raise ValueError(f"Unsupported synthetic component mode: {spec.mode}") def _weighted_geometric_mean( values: List[Tuple[float, float]], scale: float, alpha: float, ) -> float: sum_ln = 0.0 sum_weight = 0.0 for value, weight in values: if value <= 0 or weight <= 0: continue sum_ln += weight * math.log(value) sum_weight += weight if sum_weight <= 0: raise ValueError("Synthetic symbol received no valid positive component values") return scale * math.exp(alpha * (sum_ln / sum_weight)) def _build_synthetic_ohlc( component_values: Dict[str, Dict[str, float]], config: SyntheticSymbolConfig, ) -> Dict[str, float]: open_value = _weighted_geometric_mean( [(component_values[spec.name]["open"], spec.weight) for spec in config.components if spec.enabled], config.scale, config.alpha, ) close_value = _weighted_geometric_mean( [(component_values[spec.name]["close"], spec.weight) for spec in config.components if spec.enabled], config.scale, config.alpha, ) high_raw = _weighted_geometric_mean( [(component_values[spec.name]["high"], spec.weight) for spec in config.components if spec.enabled], config.scale, config.alpha, ) low_raw = _weighted_geometric_mean( [(component_values[spec.name]["low"], spec.weight) for spec in config.components if spec.enabled], config.scale, config.alpha, ) midpoint = (open_value + close_value) / 2.0 high_value = midpoint + (high_raw - midpoint) * config.wick_shrink low_value = midpoint + (low_raw - midpoint) * config.wick_shrink return { "open": open_value, "high": max(high_value, open_value, close_value), "low": min(low_value, open_value, close_value), "close": close_value, } async def _build_synthetic_symbol_history( symbol: str, interval: str, fetch_limit: int, cache_key: str, ) -> Tuple[List[Dict[str, Any]], str]: config = SYNTHETIC_SYMBOLS[symbol] headroom = max(50, fetch_limit // 2) component_limit = min(2000, fetch_limit + headroom) required_symbols = { spec.left_symbol for spec in config.components if spec.enabled } | { spec.right_symbol for spec in config.components if spec.enabled and spec.right_symbol } minimum_required = min(20, fetch_limit) preferred_sources = ( ["yfinance", "twelvedata", "finnhub"] if interval in {"1d", "1w"} else ["twelvedata", "finnhub", "yfinance"] ) component_rows: Dict[str, List[Dict[str, Any]]] = {} component_sources: Dict[str, str] = {} common_times: List[int] = [] last_error_messages: List[str] = [] use_body_only_extrema = interval not in {"1d", "1w"} async def _load_component_rows( source_name: str, component_symbol: str, ) -> Tuple[str, Optional[List[Dict[str, Any]]], Optional[str]]: if source_name not in SYMBOLS[component_symbol].mappings: return component_symbol, None, f"{component_symbol}: missing {source_name} mapping" try: rows = await _fetch_historical_from_source_cached( source_name, component_symbol, interval, component_limit, ) return component_symbol, rows, None except Exception as exc: return component_symbol, None, f"{component_symbol}: {exc}" for source in preferred_sources: current_rows: Dict[str, List[Dict[str, Any]]] = {} current_sources: Dict[str, str] = {} source_errors: List[str] = [] component_results = await asyncio.gather( *[ _load_component_rows(source, component_symbol) for component_symbol in sorted(required_symbols) ] ) for component_symbol, rows, error_message in component_results: if error_message: source_errors.append(error_message) continue if rows is None: source_errors.append(f"{component_symbol}: empty rows") continue current_rows[component_symbol] = rows current_sources[component_symbol] = source if source_errors: last_error_messages = source_errors continue time_sets = [ {int(row["time"]) for row in rows} for rows in current_rows.values() if rows ] current_common_times = sorted(set.intersection(*time_sets)) if time_sets else [] if len(current_common_times) < minimum_required: last_error_messages = [ f"{source}: aligned={len(current_common_times)} required>={minimum_required}" ] continue component_rows = current_rows component_sources = current_sources common_times = current_common_times break if len(common_times) < minimum_required: raise HTTPException( status_code=502, detail={ "message": f"Insufficient aligned candles to build synthetic symbol {symbol}", "errors": last_error_messages, }, ) candles_by_symbol = { component_symbol: {int(row["time"]): row for row in rows} for component_symbol, rows in component_rows.items() } synthetic_rows: List[Dict[str, Any]] = [] for timestamp in common_times[-fetch_limit:]: component_values: Dict[str, Dict[str, float]] = {} for spec in config.components: if not spec.enabled: continue left_candle = candles_by_symbol[spec.left_symbol][timestamp] right_candle = ( candles_by_symbol[spec.right_symbol][timestamp] if spec.right_symbol else None ) component_values[spec.name] = _combine_component_candles( spec, left_candle, right_candle, use_body_only_extrema=use_body_only_extrema, ) synthetic_ohlc = _build_synthetic_ohlc(component_values, config) synthetic_rows.append( { "time": timestamp, "open": round(float(synthetic_ohlc["open"]), 8), "high": round(float(synthetic_ohlc["high"]), 8), "low": round(float(synthetic_ohlc["low"]), 8), "close": round(float(synthetic_ohlc["close"]), 8), "volume": 0.0, } ) source = "synthetic:" + ",".join(sorted(set(component_sources.values()))) historical_cache.set(cache_key, (synthetic_rows, source), ttl_seconds=interval_ttl(interval)) return synthetic_rows, source def _get_source_priority(symbol: str, interval: Optional[str] = None) -> List[str]: cfg = SYMBOLS[symbol] 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 if interval in {"1d", "1w"} else HF_CRYPTO_SOURCE_PRIORITY_INTRADAY ) return [source_name for source_name in priority if _has_source_mapping(symbol, source_name)] async def _run_historical_fetch( symbol: str, interval: str, fetch_limit: int, cache_key: str, ) -> Tuple[List[Dict[str, Any]], str]: if _is_synthetic_symbol(symbol): return await _build_synthetic_symbol_history(symbol, interval, fetch_limit, cache_key) priority = _get_source_priority(symbol, interval) errors: List[str] = [] for source in priority: try: if source == "binance": data = await fetch_binance(symbol, interval, fetch_limit) elif source == "bybit": data = await fetch_bybit(symbol, interval, fetch_limit) elif source == "coingecko": data = await fetch_coingecko(symbol, interval, fetch_limit) elif source == "twelvedata": data = await fetch_twelvedata(symbol, interval, fetch_limit) elif source == "finnhub": data = await fetch_finnhub(symbol, interval, fetch_limit) elif source == "yfinance": data = await fetch_yfinance(symbol, interval, fetch_limit) elif source == "alphavantage": data = await fetch_alphavantage(symbol, interval, fetch_limit) else: continue if len(data) >= 20: historical_cache.set(cache_key, (data, source), ttl_seconds=interval_ttl(interval)) return data, source errors.append(f"{source}: insufficient ({len(data)} candles)") except HTTPException as ex: errors.append(f"{source}: HTTP {ex.status_code} {ex.detail}") logger.warning( "[fetch_historical] %s/%s %s HTTP %s: %s", symbol, interval, source, ex.status_code, ex.detail, ) continue except Exception as ex: errors.append(f"{source}: {ex}") logger.warning("[fetch_historical] %s/%s %s: %s", symbol, interval, source, ex) raise HTTPException( status_code=502, detail={"message": f"All sources failed: {symbol}/{interval}", "errors": errors}, ) async def fetch_historical( symbol: str, interval: str, limit: int, refresh: bool = False, min_context: Optional[int] = None, ) -> Tuple[List[Dict[str, Any]], str]: """ 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 forecast quality. """ prefix = _cache_prefix(symbol, interval) key = f"hist_{prefix}" # BUG-P1-03: No limit in key to increase cache hits required_context = max(limit, min_context or limit) if refresh: historical_cache.delete(key) else: cached = historical_cache.get(key) if cached is not None: try: # v6.1: Cache stores (data, source) data_cached, source_cached = cached if len(data_cached) >= required_context: return data_cached[-limit:], source_cached except (ValueError, TypeError): # Handle old cache format gracefully historical_cache.delete(key) fetch_limit = required_context inflight_key = f"{key}:refresh={int(refresh)}:fetch={fetch_limit}" task = None if refresh else _HISTORICAL_INFLIGHT.get(inflight_key) if task is None: task = asyncio.create_task( _run_historical_fetch(symbol, interval, fetch_limit, key), name=f"hist:{symbol}:{interval}:{fetch_limit}", ) _HISTORICAL_INFLIGHT[inflight_key] = task try: data, source = await task return data[-limit:], source finally: if _HISTORICAL_INFLIGHT.get(inflight_key) is task: _HISTORICAL_INFLIGHT.pop(inflight_key, None) # ────────────────────────────────────────────────────────────────────────────── # Real-time Ticker (last price + 24h stats) # ────────────────────────────────────────────────────────────────────────────── async def fetch_ticker(symbol: str, interval: Optional[str] = None) -> Dict[str, Any]: interval_key = interval or "default" cache_key = f"ticker:{symbol}:{interval_key}" cached = ticker_cache.get(cache_key) if cached: return cached inflight_key = cache_key inflight_task = _TICKER_INFLIGHT.get(inflight_key) if inflight_task is not None: return await inflight_task async def _run() -> Dict[str, Any]: if _is_synthetic_symbol(symbol): synthetic_interval = interval or "1d" rows, source = await fetch_historical(symbol, synthetic_interval, 2, min_context=2) current_row = rows[-1] previous_row = rows[-2] if len(rows) > 1 else current_row current_price = float(current_row["close"]) previous_close = float(previous_row["close"]) high_24h = float(current_row["high"]) low_24h = float(current_row["low"]) change_value = current_price - previous_close result = { "symbol": symbol, "price": current_price, "change": change_value, "change_pct": ((change_value / previous_close) * 100.0) if previous_close else 0.0, "high_24h": high_24h, "low_24h": low_24h, "volume_24h": 0.0, "source": source, "timestamp": int(time.time()), } ticker_cache.set(cache_key, result, ttl_seconds=10) return result cfg = SYMBOLS[symbol] priority = _get_source_priority(symbol, interval=interval) client = await GlobalHTTPClient.get_client() for source in priority: try: if source == "binance": binance_symbol = _get_symbol_mapping(symbol, "binance") if not binance_symbol: continue await _rate_limit("binance") base_url = "https://fapi.binance.com" if cfg.binance_type == "futures" else "https://api.binance.com" endpoint = "/fapi/v1/ticker/24hr" if cfg.binance_type == "futures" else "/api/v3/ticker/24hr" r = await client.get( f"{base_url}{endpoint}", params={"symbol": binance_symbol}, timeout=10.0, ) d = r.json() res = { "price": float(d.get("lastPrice") or d.get("price") or 0), "change": float(d.get("priceChange", 0)), "change_pct": float(d.get("priceChangePercent", 0)), "high_24h": float(d.get("highPrice", 0)), "low_24h": float(d.get("lowPrice", 0)), "volume_24h": float(d.get("volume", 0)), "source": "binance", } elif source == "twelvedata": twelvedata_symbol = _get_symbol_mapping(symbol, "twelvedata") if not twelvedata_symbol: continue await _rate_limit("twelvedata") r = await client.get( "https://api.twelvedata.com/quote", params={"symbol": twelvedata_symbol, "apikey": twelvedata_pool.next_key()}, timeout=10.0, ) d = r.json() if "close" not in d and "price" not in d: continue p = float(d.get("price") or d.get("close") or 0) pc = float(d.get("previous_close") or p) res = { "price": p, "change": p - pc, "change_pct": ((p - pc) / pc * 100) if pc != 0 else 0, "high_24h": float(d.get("high") or p), "low_24h": float(d.get("low") or p), "volume_24h": float(d.get("volume") or 0), "source": "twelvedata", } elif source == "bybit": bybit_symbol = _get_symbol_mapping(symbol, "bybit") if not bybit_symbol: continue await _rate_limit("bybit") url = "https://api.bybit.com/v5/market/tickers" query_params = {"category": cfg.bybit_category, "symbol": bybit_symbol} headers = {} if settings.bybit_api_key and settings.bybit_api_secret: timestamp = str(int(time.time() * 1000)) recv_window = "5000" sorted_params = "&".join([f"{k}={v}" for k, v in sorted(query_params.items())]) raw_str = timestamp + settings.bybit_api_key + recv_window + sorted_params signature = hmac.new( settings.bybit_api_secret.encode("utf-8"), raw_str.encode("utf-8"), hashlib.sha256, ).hexdigest() headers = { "X-BAPI-API-KEY": settings.bybit_api_key, "X-BAPI-TIMESTAMP": timestamp, "X-BAPI-SIGN-TYPE": "2", "X-BAPI-RECV-WINDOW": recv_window, "X-BAPI-SIGN": signature, } r = await client.get(url, params=query_params, headers=headers, timeout=10.0) d = r.json() if d.get("retCode") != 0 or not d.get("result", {}).get("list"): continue t = d["result"]["list"][0] res = { "price": float(t["lastPrice"]), "change": float(t["price24hPcnt"]) * float(t["lastPrice"]) / 100, "change_pct": float(t["price24hPcnt"]) * 100, "high_24h": float(t["highPrice24h"]), "low_24h": float(t["lowPrice24h"]), "volume_24h": float(t["volume24h"]), "source": "bybit", } elif source == "yfinance": yfinance_symbol = _get_symbol_mapping(symbol, "yfinance") if not yfinance_symbol: continue def _yf_info(): ticker = yf.Ticker(yfinance_symbol) history = ticker.history(period="5d") if history.empty: return None last_price = float(history["Close"].iloc[-1]) prev_close = float(history["Close"].iloc[-2]) if len(history) > 1 else last_price return { "price": last_price, "change": last_price - prev_close, "change_pct": ((last_price - prev_close) / prev_close * 100) if prev_close != 0 else 0, "high_24h": float(history["High"].iloc[-1]), "low_24h": float(history["Low"].iloc[-1]), "volume_24h": float(history["Volume"].iloc[-1]), "source": "yfinance", } res = await asyncio.to_thread(_yf_info) if not res: continue else: continue res.update({"symbol": symbol, "timestamp": int(time.time())}) ttl = 5 if cfg.category == "Crypto" else (10 if cfg.category in ("Cặp tiền", "Chỉ số", "Real Strength") else 30) ticker_cache.set(cache_key, res, ttl_seconds=ttl) return res except Exception as ex: logger.debug("[ticker] %s/%s failed: %s", symbol, source, ex) continue raise HTTPException(status_code=502, detail=f"Ticker failed for {symbol} after trying {priority}") task = asyncio.create_task(_run(), name=f"ticker:{symbol}:{interval_key}") _TICKER_INFLIGHT[inflight_key] = task try: return await task finally: if _TICKER_INFLIGHT.get(inflight_key) is task: _TICKER_INFLIGHT.pop(inflight_key, None) # ────────────────────────────────────────────────────────────────────────────── # Technical Indicators # ────────────────────────────────────────────────────────────────────────────── # D-2: High-Performance Vectorized Indicators (NumPy) def _ema(arr: np.ndarray, period: int) -> np.ndarray: """Vectorized EMA using NumPy (replaces loops).""" if len(arr) == 0: return np.array([], dtype=float) alpha = 2.0 / (period + 1.0) # Use pandas ewm for robust vectorized calculation (v6.0) return pd.Series(arr).ewm(alpha=alpha, adjust=False).mean().values def _rsi(close: np.ndarray, period: int = 14) -> np.ndarray: """Vectorized RSI using NumPy/Pandas.""" delta = np.diff(close) gain = np.where(delta > 0, delta, 0.0) loss = np.where(delta < 0, -delta, 0.0) avg_gain = pd.Series(gain).ewm(alpha=1.0/period, adjust=False).mean() avg_loss = pd.Series(loss).ewm(alpha=1.0/period, adjust=False).mean() rs = avg_gain / avg_loss.replace(0, np.nan) rsi = 100 - (100 / (1 + rs)) rsi = rsi.where(avg_loss > 0, 100.0) rsi = rsi.where(avg_gain > 0, 0.0) rsi = rsi.where(~((avg_gain == 0) & (avg_loss == 0)), 50.0) # Prepend NaN to match original array length return np.concatenate([[np.nan], rsi.values]) def _bollinger(close: np.ndarray, period=20, k=2.0) -> Tuple[np.ndarray, np.ndarray, np.ndarray]: """Vectorized Bollinger Bands.""" s = pd.Series(close) mid = s.rolling(window=period).mean() std = s.rolling(window=period).std(ddof=0) return (mid + k*std).values, mid.values, (mid - k*std).values def _macd(close: np.ndarray, fast=12, slow=26, signal=9 ) -> Tuple[np.ndarray, np.ndarray, np.ndarray]: ema_fast = _ema(close, fast) ema_slow = _ema(close, slow) macd_line = ema_fast - ema_slow sig_line = _ema(np.where(np.isnan(macd_line), 0, macd_line), signal) histogram = macd_line - sig_line return macd_line, sig_line, histogram def _atr(high: np.ndarray, low: np.ndarray, close: np.ndarray, period=14) -> np.ndarray: """Vectorized ATR.""" tr = np.maximum(high[1:] - low[1:], np.maximum(np.abs(high[1:] - close[:-1]), np.abs(low[1:] - close[:-1]))) tr = np.concatenate([[np.nan], tr]) return pd.Series(tr).ewm(alpha=1.0/period, adjust=False).mean().values def _stoch_rsi(close: np.ndarray, rsi_period=14, stoch_period=14, smooth_k=3, smooth_d=3) -> Tuple[np.ndarray, np.ndarray]: """Vectorized Stochastic RSI.""" rsi_vals = pd.Series(_rsi(close, rsi_period)) roll_min = rsi_vals.rolling(window=stoch_period).min() roll_max = rsi_vals.rolling(window=stoch_period).max() k = 100 * (rsi_vals - roll_min) / (roll_max - roll_min).replace(0, np.inf) k_smooth = k.rolling(window=smooth_k).mean() d_smooth = k_smooth.rolling(window=smooth_d).mean() return k_smooth.values, d_smooth.values def _stoch_kd( high: np.ndarray, low: np.ndarray, close: np.ndarray, k_period: int = 14, smooth_k: int = 3, d_period: int = 3, ) -> Tuple[np.ndarray, np.ndarray]: """Standard stochastic oscillator computed from price high/low/close.""" high_s = pd.Series(high) low_s = pd.Series(low) close_s = pd.Series(close) hh = high_s.rolling(window=k_period).max() ll = low_s.rolling(window=k_period).min() raw_k = 100.0 * (close_s - ll) / (hh - ll).replace(0, np.nan) k = raw_k.rolling(window=smooth_k).mean() d = k.rolling(window=d_period).mean() return k.values, d.values def _sma(arr: np.ndarray, period: int) -> np.ndarray: """Vectorized Simple Moving Average.""" if len(arr) == 0: return np.array([], dtype=float) return pd.Series(arr).rolling(window=period).mean().values def _cci(high: np.ndarray, low: np.ndarray, close: np.ndarray, period: int = 20) -> np.ndarray: """Vectorized Commodity Channel Index.""" tp = (high + low + close) / 3.0 tp_series = pd.Series(tp) sma = tp_series.rolling(window=period).mean() # Optimized MAD calculation (Vectorized) mad = tp_series.rolling(window=period).apply(lambda x: np.abs(x - x.mean()).mean(), raw=True) return ((tp_series - sma) / (0.015 * mad.replace(0, np.inf))).values def _adx(high: np.ndarray, low: np.ndarray, close: np.ndarray, period: int = 14) -> tuple: """Vectorized Average Directional Index (v6.0).""" plus_dm = np.where((high[1:] - high[:-1] > low[:-1] - low[1:]) & (high[1:] - high[:-1] > 0), high[1:] - high[:-1], 0.0) minus_dm = np.where((low[:-1] - low[1:] > high[1:] - high[:-1]) & (low[:-1] - low[1:] > 0), low[:-1] - low[1:], 0.0) tr = np.maximum(high[1:] - low[1:], np.maximum(np.abs(high[1:] - close[:-1]), np.abs(low[1:] - close[:-1]))) # Pad first index plus_dm = np.concatenate([[0.0], plus_dm]) minus_dm = np.concatenate([[0.0], minus_dm]) tr = np.concatenate([[0.0], tr]) tr_sum = pd.Series(tr).ewm(alpha=1.0/period, adjust=False).mean() plus_di = 100 * pd.Series(plus_dm).ewm(alpha=1.0/period, adjust=False).mean() / tr_sum.replace(0, np.inf) minus_di = 100 * pd.Series(minus_dm).ewm(alpha=1.0/period, adjust=False).mean() / tr_sum.replace(0, np.inf) dx = 100 * np.abs(plus_di - minus_di) / (plus_di + minus_di).replace(0, np.inf) adx = dx.ewm(alpha=1.0/period, adjust=False).mean() return adx.values, plus_di.values, minus_di.values def _awesome_oscillator(high: np.ndarray, low: np.ndarray) -> np.ndarray: """Awesome Oscillator = SMA(5, median) - SMA(34, median).""" median = (high + low) / 2.0 sma5 = _sma(median, 5) sma34 = _sma(median, 34) return sma5 - sma34 def _momentum(close: np.ndarray, period: int = 10) -> np.ndarray: """Vectorized Momentum Index centered at 100.""" base = pd.Series(close).shift(period) return (pd.Series(close) / base.replace(0, np.nan) * 100.0).values def _roc(close: np.ndarray, period: int = 12) -> np.ndarray: """Rate of Change in percentage.""" base = pd.Series(close).shift(period) return (pd.Series(close) / base.replace(0, np.nan) - 1.0).mul(100.0).values def _trix(close: np.ndarray, period: int = 18) -> np.ndarray: """Triple EMA oscillator in percentage.""" ema1 = pd.Series(_ema(close, period)) ema2 = ema1.ewm(span=period, adjust=False).mean() ema3 = ema2.ewm(span=period, adjust=False).mean() return ema3.pct_change().mul(100.0).values def _ppo(close: np.ndarray, fast: int = 12, slow: int = 26) -> np.ndarray: """Percentage Price Oscillator.""" ema_fast = _ema(close, fast) ema_slow = _ema(close, slow) return ((ema_fast - ema_slow) / np.where(np.abs(ema_slow) < 1e-8, np.nan, ema_slow) * 100.0) def _cmo(close: np.ndarray, period: int = 14) -> np.ndarray: """Chande Momentum Oscillator.""" delta = pd.Series(close).diff() up = delta.clip(lower=0.0).rolling(period).sum() down = (-delta.clip(upper=0.0)).rolling(period).sum() return ((up - down) / (up + down).replace(0, np.nan) * 100.0).values def _dpo(close: np.ndarray, period: int = 20) -> np.ndarray: """Detrended Price Oscillator.""" offset = int(period / 2) + 1 sma = pd.Series(close).rolling(period).mean() return (pd.Series(close) - sma.shift(offset)).values def _aroon_oscillator(high: np.ndarray, low: np.ndarray, period: int = 25) -> np.ndarray: """Aroon Oscillator = Aroon Up - Aroon Down.""" hs = pd.Series(high) ls = pd.Series(low) aroon_up = hs.rolling(period).apply(lambda x: ((period - 1 - (len(x) - 1 - int(np.argmax(x)))) / (period - 1)) * 100.0, raw=True) aroon_down = ls.rolling(period).apply(lambda x: ((period - 1 - (len(x) - 1 - int(np.argmin(x)))) / (period - 1)) * 100.0, raw=True) return (aroon_up - aroon_down).values def _tsi(close: np.ndarray, long_period: int = 25, short_period: int = 13) -> np.ndarray: """True Strength Index.""" delta = pd.Series(close).diff() abs_delta = delta.abs() ema1 = delta.ewm(span=long_period, adjust=False).mean() ema2 = ema1.ewm(span=short_period, adjust=False).mean() abs_ema1 = abs_delta.ewm(span=long_period, adjust=False).mean() abs_ema2 = abs_ema1.ewm(span=short_period, adjust=False).mean() return (ema2 / abs_ema2.replace(0, np.nan) * 100.0).values def _demarker(high: np.ndarray, low: np.ndarray, period: int = 14) -> np.ndarray: """DeMarker oscillator in range 0..100.""" high_delta = pd.Series(high).diff() low_delta = -pd.Series(low).diff() demax = high_delta.clip(lower=0.0) demin = low_delta.clip(lower=0.0) demax_avg = demax.rolling(period).sum() demin_avg = demin.rolling(period).sum() denom = (demax_avg + demin_avg).replace(0, np.nan) return (demax_avg / denom * 100.0).values def _force_index(close: np.ndarray, volume: np.ndarray, period: int = 13) -> np.ndarray: """Force Index smoothed with an EMA.""" raw = pd.Series(close).diff() * pd.Series(volume) return raw.ewm(span=period, adjust=False).mean().values def _williams_r(high: np.ndarray, low: np.ndarray, close: np.ndarray, period: int = 14) -> np.ndarray: """Vectorized Williams %R.""" s = pd.Series(close) hh = pd.Series(high).rolling(window=period).max() ll = pd.Series(low).rolling(window=period).min() wr = -100 * (hh - s) / (hh - ll).replace(0, np.inf) return wr.values def _bull_bear_power(high: np.ndarray, low: np.ndarray, close: np.ndarray, period: int = 13) -> np.ndarray: """Vectorized Bull Bear Power (v6.0).""" ema_val = _ema(close, period) return (high - ema_val) + (low - ema_val) def _ultimate_oscillator(high: np.ndarray, low: np.ndarray, close: np.ndarray, p1: int = 7, p2: int = 14, p3: int = 28) -> np.ndarray: """Vectorized Ultimate Oscillator (v6.0).""" n = len(close) if n < p3 + 1: return np.full(n, np.nan) prev_close = pd.Series(close).shift(1) tl = np.minimum(low, prev_close.values) th = np.maximum(high, prev_close.values) bp = pd.Series(close - tl) tr = pd.Series(th - tl) def _avg(p): return bp.rolling(p).sum() / tr.rolling(p).sum().replace(0, np.inf) avg1 = _avg(p1) avg2 = _avg(p2) avg3 = _avg(p3) uo = 100 * (4 * avg1 + 2 * avg2 + avg3) / 7.0 return uo.values def _ichimoku_base(high: np.ndarray, low: np.ndarray, period: int = 26) -> np.ndarray: """Vectorized Ichimoku Base Line.""" hh = pd.Series(high).rolling(window=period).max() ll = pd.Series(low).rolling(window=period).min() return ((hh + ll) / 2.0).values def _ichimoku_cloud(high: np.ndarray, low: np.ndarray) -> Dict[str, np.ndarray]: """Basic Ichimoku lines for cloud-position scoring.""" hs = pd.Series(high) ls = pd.Series(low) tenkan = ((hs.rolling(9).max() + ls.rolling(9).min()) / 2.0).values kijun = ((hs.rolling(26).max() + ls.rolling(26).min()) / 2.0).values span_a = ((pd.Series(tenkan) + pd.Series(kijun)) / 2.0).values span_b = ((hs.rolling(52).max() + ls.rolling(52).min()) / 2.0).values return { "tenkan": tenkan, "kijun": kijun, "span_a": span_a, "span_b": span_b, } def _vwma(close: np.ndarray, volume: np.ndarray, period: int = 20) -> np.ndarray: """Vectorized Volume Weighted Moving Average.""" cv = pd.Series(close * volume) v = pd.Series(volume) return (cv.rolling(period).sum() / v.rolling(period).sum().replace(0, np.inf)).values def _wma(arr: np.ndarray, period: int) -> np.ndarray: """Weighted moving average used by Hull MA.""" if len(arr) == 0: return np.array([], dtype=float) weights = np.arange(1, period + 1, dtype=float) return pd.Series(arr).rolling(window=period).apply( lambda x: float(np.dot(x, weights) / weights.sum()), raw=True, ).values def _hull_ma(close: np.ndarray, period: int = 9) -> np.ndarray: """Hull Moving Average.""" half = max(period // 2, 1) sqrt_p = max(int(math.sqrt(period)), 1) wma_half = _wma(close, half) wma_full = _wma(close, period) diff = 2 * wma_half - wma_full hull = _wma(np.where(np.isnan(diff), close, diff), sqrt_p) return hull def _calc_pivot_points(high_val: float, low_val: float, close_val: float ) -> Dict[str, Any]: """Calculate pivot points using 5 methods: Classic, Fibonacci, Camarilla, Woodie, DM.""" p_classic = (high_val + low_val + close_val) / 3.0 r1_c = 2 * p_classic - low_val s1_c = 2 * p_classic - high_val r2_c = p_classic + (high_val - low_val) s2_c = p_classic - (high_val - low_val) r3_c = high_val + 2 * (p_classic - low_val) s3_c = low_val - 2 * (high_val - p_classic) # Fibonacci diff = high_val - low_val r1_f = p_classic + 0.382 * diff r2_f = p_classic + 0.618 * diff r3_f = p_classic + 1.0 * diff s1_f = p_classic - 0.382 * diff s2_f = p_classic - 0.618 * diff s3_f = p_classic - 1.0 * diff # Camarilla r1_cam = close_val + diff * 1.1 / 12 r2_cam = close_val + diff * 1.1 / 6 r3_cam = close_val + diff * 1.1 / 4 s1_cam = close_val - diff * 1.1 / 12 s2_cam = close_val - diff * 1.1 / 6 s3_cam = close_val - diff * 1.1 / 4 # Woodie p_w = (high_val + low_val + 2 * close_val) / 4.0 r1_w = 2 * p_w - low_val s1_w = 2 * p_w - high_val r2_w = p_w + diff s2_w = p_w - diff r3_w = high_val + 2 * (p_w - low_val) s3_w = low_val - 2 * (high_val - p_w) # DM (Demark) if close_val < p_classic: x = high_val + 2 * low_val + close_val elif close_val > p_classic: x = 2 * high_val + low_val + close_val else: x = high_val + low_val + 2 * close_val p_dm = x / 4.0 r1_dm = x / 2.0 - low_val s1_dm = x / 2.0 - high_val def _r(v): return round(float(v), 2) levels = [] for lbl, cl, fb, cm, wd, dm_r, dm_s in [ ("R3", r3_c, r3_f, r3_cam, r3_w, None, None), ("R2", r2_c, r2_f, r2_cam, r2_w, None, None), ("R1", r1_c, r1_f, r1_cam, r1_w, r1_dm, None), ("P", p_classic, p_classic, p_classic, p_w, p_dm, None), ("S1", s1_c, s1_f, s1_cam, s1_w, None, s1_dm), ("S2", s2_c, s2_f, s2_cam, s2_w, None, None), ("S3", s3_c, s3_f, s3_cam, s3_w, None, None), ]: row = { "level": lbl, "classic": _r(cl), "fibonacci": _r(fb), "camarilla": _r(cm), "woodie": _r(wd), } if lbl == "R1": row["dm"] = _r(r1_dm) elif lbl == "P": row["dm"] = _r(p_dm) elif lbl == "S1": row["dm"] = _r(s1_dm) else: row["dm"] = None levels.append(row) return {"data": levels} # ───────────────────────────────────────────────────────────────────────────── # TradingView-style action classification helpers # ───────────────────────────────────────────────────────────────────────────── def _osc_action(name: str, value: float, **kw) -> str: """Classify oscillator value as 'Mua' / 'Bán' / 'Trung lập'.""" if value is None or math.isnan(value): return "Trung lập" atr = max(float(kw.get("atr", 0.0) or 0.0), 1e-8) def _bounded_action( current: float, min_value: float, max_value: float, *, bullish_high: bool = True, ) -> str: span = max(max_value - min_value, 1e-8) ratio = _clamp((current - min_value) / span, 0.0, 1.0) if not bullish_high: ratio = 1.0 - ratio if ratio > (2.0 / 3.0): return "Mua" if ratio < (1.0 / 3.0): return "Bán" return "Trung lập" if name == "rsi": return _bounded_action(value, 0.0, 100.0) if name in {"stoch", "stoch_rsi", "ultimate", "demarker"}: return _bounded_action(value, 0.0, 100.0) if name == "williams": return _bounded_action(value, -100.0, 0.0, bullish_high=True) if name == "cci": return "Mua" if value > 100 else "Bán" if value < -100 else "Trung lập" if name == "momentum": return "Mua" if value > 100 else "Bán" if value < 100 else "Trung lập" if name == "roc": return "Mua" if value > 1.0 else "Bán" if value < -1.0 else "Trung lập" if name == "trix": return "Mua" if value > 0.0 else "Bán" if value < 0.0 else "Trung lập" if name == "ppo": return "Mua" if value > 0.35 else "Bán" if value < -0.35 else "Trung lập" if name == "cmo": return "Mua" if value > 33.0 else "Bán" if value < -33.0 else "Trung lập" if name == "dpo": return "Mua" if value > 0.0 else "Bán" if value < 0.0 else "Trung lập" if name == "aroon": return "Mua" if value > 33.0 else "Bán" if value < -33.0 else "Trung lập" if name == "tsi": return "Mua" if value > 25.0 else "Bán" if value < -25.0 else "Trung lập" if name == "adx": plus_di = kw.get("plus_di", 0) minus_di = kw.get("minus_di", 0) if value < 20: return "Trung lập" return "Mua" if plus_di > minus_di else "Bán" if name == "ao": prev = float(kw.get("prev", value) or value) if value > 0 and value > prev: return "Mua" if value < 0 and value < prev: return "Bán" return "Trung lập" if name == "macd": signal = kw.get("signal", 0) return "Mua" if value > signal else "Bán" if value < signal else "Trung lập" if name == "bbp": bbp_norm = float(kw.get("bbp_norm", value / atr)) return "Mua" if bbp_norm > 0.5 else "Bán" if bbp_norm < -0.5 else "Trung lập" if name == "force_index": scale = max(float(kw.get("scale", 1.0) or 1.0), 1e-8) norm = value / scale return "Mua" if norm > 0.25 else "Bán" if norm < -0.25 else "Trung lập" return "Trung lập" def _ema_pair_action(fast_val: Optional[float], slow_val: Optional[float]) -> str: """EMA chain classification based on relative ordering only.""" if fast_val is None or slow_val is None: return "Trung lập" if math.isnan(fast_val) or math.isnan(slow_val): return "Trung lập" if fast_val > slow_val: return "Mua" if fast_val < slow_val: return "Bán" return "Trung lập" def _ema_pair_threshold_pct(pair_key: str) -> float: thresholds = { "ema_1_5": 0.08, "ema_5_10": 0.10, "ema_10_20": 0.14, "ema_20_50": 0.22, "ema_50_100": 0.35, "ema_100_200": 0.55, } return thresholds.get(pair_key, 0.18) def _ema_pair_score(fast_val: Optional[float], slow_val: Optional[float], pair_key: str) -> float: if fast_val is None or slow_val is None: return 0.0 if math.isnan(fast_val) or math.isnan(slow_val): return 0.0 threshold = max(_ema_pair_threshold_pct(pair_key), 1e-4) spread_pct = _pct(fast_val, slow_val) return _clamp(math.tanh(spread_pct / threshold), -1.0, 1.0) def _osc_signal_score(name: str, value: float, **kw) -> float: """Continuous oscillator score in [-1, 1] for weighting strength internally.""" if value is None or math.isnan(value): return 0.0 scale = max(float(kw.get("scale", 1.0) or 1.0), 1e-8) def _bounded_score( current: float, min_value: float, max_value: float, *, bullish_high: bool = True, ) -> float: span = max(max_value - min_value, 1e-8) ratio = _clamp((current - min_value) / span, 0.0, 1.0) score = (ratio * 2.0) - 1.0 return _clamp(score if bullish_high else -score, -1.0, 1.0) if name == "rsi": return _bounded_score(value, 0.0, 100.0) if name in {"stoch", "stoch_rsi", "ultimate", "demarker"}: return _bounded_score(value, 0.0, 100.0) if name == "cci": return _clamp(value / 200.0, -1.0, 1.0) if name == "adx": plus_di = float(kw.get("plus_di", 0.0) or 0.0) minus_di = float(kw.get("minus_di", 0.0) or 0.0) strength = _clamp((value - 18.0) / 22.0, 0.0, 1.0) direction = math.tanh((plus_di - minus_di) / max(plus_di + minus_di, 10.0) * 3.0) return _clamp(direction * strength, -1.0, 1.0) if name == "ao": return _clamp(math.tanh(value / scale), -1.0, 1.0) if name == "momentum": return _clamp((value - 100.0) / 6.0, -1.0, 1.0) if name == "bbp": bbp_norm = float(kw.get("bbp_norm", value / max(float(kw.get("atr", scale) or scale), 1e-8))) return _clamp(math.tanh(bbp_norm / 1.0), -1.0, 1.0) if name == "dpo": return _clamp(math.tanh(value / scale), -1.0, 1.0) if name == "macd": signal = float(kw.get("signal", 0.0) or 0.0) return _clamp(math.tanh((value - signal) / scale), -1.0, 1.0) if name == "williams": return _bounded_score(value, -100.0, 0.0) if name == "roc": return _clamp(math.tanh(value / 3.0), -1.0, 1.0) if name == "trix": return _clamp(math.tanh(value / 0.35), -1.0, 1.0) if name == "ppo": return _clamp(math.tanh(value / 0.8), -1.0, 1.0) if name == "cmo": return _clamp(value / 50.0, -1.0, 1.0) if name == "aroon": return _clamp(value / 60.0, -1.0, 1.0) if name == "tsi": return _clamp(value / 25.0, -1.0, 1.0) if name == "force_index": return _clamp(math.tanh(value / scale), -1.0, 1.0) return 0.0 def compute_indicators(data: List[Dict[str, Any]]) -> Dict[str, Any]: """Compute a full suite of technical indicators on OHLCV data.""" if len(data) < 30: return {"error": "Insufficient data (need ≥30 candles)"} closes = np.array([d["close"] for d in data], dtype=float) highs = np.array([d["high"] for d in data], dtype=float) lows = np.array([d["low"] for d in data], dtype=float) vols = np.array([d["volume"]for d in data], dtype=float) times = [d["time"] for d in data] def _last(arr): v = arr[-1] return None if math.isnan(v) else round(float(v), 8) def _series(arr, n: int = 50): out = [] window = arr[-min(n, len(arr)):] offset = len(arr) - len(window) for i, v in enumerate(window): idx = offset + i out.append({"time": times[idx], "value": None if math.isnan(v) else round(float(v), 8)}) return out ema9 = _ema(closes, 9) ema21 = _ema(closes, 21) ema50 = _ema(closes, 50) ema200 = _ema(closes, 200) rsi14 = _rsi(closes, 14) macd_l, macd_s, macd_h = _macd(closes) bb_u, bb_m, bb_l = _bollinger(closes) atr14 = _atr(highs, lows, closes, 14) stoch_k, stoch_d = _stoch_rsi(closes) roc12 = _roc(closes, 12) trix18 = _trix(closes, 18) ppo12 = _ppo(closes, 12, 26) cmo14 = _cmo(closes, 14) dpo20 = _dpo(closes, 20) aroon25 = _aroon_oscillator(highs, lows, 25) tsi25 = _tsi(closes, 25, 13) # Volume SMA 20 (Vectorized v6.0) vol_sma = _sma(vols, 20) last_close = closes[-1] last_atr = _last(atr14) or 0 return { "ema": { "ema9": _last(ema9), "ema21": _last(ema21), "ema50": _last(ema50), "ema200": _last(ema200), }, "rsi": { "value": _last(rsi14), "signal": ( "overbought" if (_last(rsi14) or 50) > 70 else "oversold" if (_last(rsi14) or 50) < 30 else "neutral" ), }, "macd": { "macd": _last(macd_l), "signal": _last(macd_s), "histogram": _last(macd_h), "cross": ( "bullish" if (_last(macd_h) or 0) > 0 else "bearish" if (_last(macd_h) or 0) < 0 else "neutral" ), }, "bollinger": { "upper": _last(bb_u), "middle": _last(bb_m), "lower": _last(bb_l), "bandwidth": round(((_last(bb_u) or 0) - (_last(bb_l) or 0)) / ((_last(bb_m) or 1)), 6), }, "atr": { "value": _last(atr14), "pct": round(last_atr / last_close * 100, 4) if last_close else None, }, "stoch_rsi": { "k": _last(stoch_k), "d": _last(stoch_d), "signal": ( "overbought" if (_last(stoch_k) or 50) > 80 else "oversold" if (_last(stoch_k) or 50) < 20 else "neutral" ), }, "roc": { "value": _last(roc12), "signal": "bullish" if (_last(roc12) or 0) > 1.0 else "bearish" if (_last(roc12) or 0) < -1.0 else "neutral", }, "trix": { "value": _last(trix18), "signal": "bullish" if (_last(trix18) or 0) > 0 else "bearish" if (_last(trix18) or 0) < 0 else "neutral", }, "ppo": { "value": _last(ppo12), "signal": "bullish" if (_last(ppo12) or 0) > 0.35 else "bearish" if (_last(ppo12) or 0) < -0.35 else "neutral", }, "cmo": { "value": _last(cmo14), "signal": "bullish" if (_last(cmo14) or 0) > 20 else "bearish" if (_last(cmo14) or 0) < -20 else "neutral", }, "dpo": { "value": _last(dpo20), "signal": "bullish" if (_last(dpo20) or 0) > 0 else "bearish" if (_last(dpo20) or 0) < 0 else "neutral", }, "aroon": { "value": _last(aroon25), "signal": "bullish" if (_last(aroon25) or 0) > 25 else "bearish" if (_last(aroon25) or 0) < -25 else "neutral", }, "tsi": { "value": _last(tsi25), "signal": "bullish" if (_last(tsi25) or 0) > 5 else "bearish" if (_last(tsi25) or 0) < -5 else "neutral", }, "volume": { "last": round(float(vols[-1]), 2), "sma20": round(float(vol_sma[-1]), 2), "above_avg": bool(vols[-1] > vol_sma[-1]), }, "trend": { "ema_bullish_stack": bool( all(x is not None for x in [_last(ema9),_last(ema21),_last(ema50)]) and _last(ema9) > _last(ema21) > _last(ema50) # type: ignore ), "above_ema200": bool(_last(ema200) is not None and last_close > (_last(ema200) or 0)), "close": round(float(last_close), 8), "short_momentum_ref": float(closes[max(0, len(closes)-6)]) if len(closes) else float(last_close), }, "short_momentum_ref": float(closes[max(0, len(closes)-6)]) if len(closes) else float(last_close), "series": { "ema9": _series(ema9), "ema21": _series(ema21), "ema50": _series(ema50), "bb_upper": _series(bb_u), "bb_mid": _series(bb_m), "bb_lower": _series(bb_l), }, } # ───────────────────────────────────────────────────────────────────────────── # UTILITIES # ───────────────────────────────────────────────────────────────────────────── def _clamp(v: float, lo: float, hi: float) -> float: return max(lo, min(hi, v)) def _pct(a: float, b: float) -> float: """(a - b) / b * 100, an toàn với b = 0.""" return (a - b) / b * 100.0 if b != 0 else 0.0 def _safe(v: Optional[float], default: float = 0.0) -> float: if v is None or math.isnan(v) or math.isinf(v): return default return float(v) def _round(v: float, d: int = 6) -> float: return round(float(v), d) def _build_anchor_forecast( data: List[Dict[str, Any]], indicators: Dict[str, Any], horizon: int, interval: str, ) -> Dict[str, np.ndarray]: """Fallback statistical forecast to cross-verify AI model stability.""" closes = np.array([float(d["close"]) for d in data], dtype=float) if len(closes) < 3: last_close = closes[-1] flat = np.full(horizon, last_close, dtype=float) return {"p10": flat.copy(), "p50": flat.copy(), "p90": flat.copy()} last_close = float(closes[-1]) ema21 = float(indicators["ema"].get("ema21") or last_close) ema50 = float(indicators["ema"].get("ema50") or last_close) rsi = float(indicators["rsi"].get("value") or 50.0) atr = float(indicators["atr"].get("value") or max(last_close * 0.006, 1.0)) band = float(indicators["bollinger"].get("bandwidth") or 0.02) returns = pd.Series(closes).pct_change().dropna().tail(96) fast_drift = float(returns.tail(min(12, len(returns))).mean()) if len(returns) else 0.0 slow_drift = float(returns.mean()) if len(returns) else 0.0 vol = float(returns.std()) if len(returns) > 1 else 0.0 ema_trend = ((last_close / ema21) - 1.0) * 0.30 + ((ema21 / ema50) - 1.0) * 0.25 mean_revert = -((last_close - ema21) / last_close) * 0.18 if last_close else 0.0 rsi_bias = 0.0018 if rsi < 35 else (-0.0018 if rsi > 65 else 0.0) volatility_drag = -vol * 0.35 step_return = _clamp( fast_drift * 0.55 + slow_drift * 0.20 + ema_trend + mean_revert + rsi_bias + volatility_drag, -0.03, 0.03, ) anchor_p50: List[float] = [] price = last_close for i in range(horizon): decay = max(0.35, 1.0 - (i / max(horizon * 1.6, 1))) price = price * (1.0 + step_return * decay) anchor_p50.append(price) step_scale = math.sqrt(max(STEP_SECONDS[interval], 60) / 86400.0) base_spread = max(atr * 0.70, last_close * max(vol * step_scale * 1.8, band * 0.22, 0.0035)) p50 = np.array(anchor_p50, dtype=float) p10 = np.array([max(0.0, x - base_spread * math.sqrt(i + 1)) for i, x in enumerate(anchor_p50)], dtype=float) p90 = np.array([x + base_spread * math.sqrt(i + 1) for i, x in enumerate(anchor_p50)], dtype=float) return {"p10": p10, "p50": p50, "p90": p90} def _blend_forecasts( last_close: float, model_output: Dict[str, Any], anchor_output: Dict[str, np.ndarray], indicators: Dict[str, Any], ) -> Dict[str, Any]: """Ensemble blending of AI model output and statistical anchor.""" 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) # A-6: Robust scale calculation using median of first 3 model predictions first_model_est = float(np.median(raw_p50[:3])) if len(raw_p50) >= 3 else (float(raw_p50[0]) if len(raw_p50) else last_close) scale = (last_close / first_model_est) if abs(first_model_est) > 1e-8 else 1.0 # Clip scale to [0.5, 2.0] to prevent extreme corrections on low-price coins (BUG-06) last_close_magnitude = math.floor(math.log10(max(abs(last_close), 1e-10))) clip_lo = 0.3 if last_close_magnitude < -4 else 0.5 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( "[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 model_p90 = raw_p90 * scale anchor_p10 = anchor_output["p10"] anchor_p50 = anchor_output["p50"] anchor_p90 = anchor_output["p90"] model_dir = np.sign(model_p50[-1] - last_close) if len(model_p50) else 0.0 anchor_dir = np.sign(anchor_p50[-1] - last_close) if len(anchor_p50) else 0.0 agreement = bool(model_dir == anchor_dir or model_dir == 0 or anchor_dir == 0) # B-2: Advanced weighting sensitive to RSI extremes (v6.0) rsi_val = indicators["rsi"].get("value") or 50.0 rsi_extreme = abs(rsi_val - 50.0) > 20.0 # <30 or >70 band_width_pct = abs(model_p90[-1] - model_p10[-1]) / last_close if last_close else 0.0 # If RSI is extreme, favor the AI model which better handles mean-reversion base_weight = 0.72 if agreement else 0.55 if rsi_extreme: base_weight += 0.08 model_weight = _clamp(base_weight - max(0.0, band_width_pct - 0.08) * 2.0, 0.30, 0.88) anchor_weight = 1.0 - model_weight blend_p10 = model_p10 * model_weight + anchor_p10 * anchor_weight blend_p50 = model_p50 * model_weight + anchor_p50 * anchor_weight blend_p90 = model_p90 * model_weight + anchor_p90 * anchor_weight bias_pct = abs((scale - 1.0) * 100.0) # B-2: Advanced confidence (v6.1 Rework) # We use a simplified version of the new AI scoring logic here to keep blended dict consistent forecast_ret_pct = abs((blend_p50[-1] - last_close) / last_close * 100) if last_close else 0 atr_pct = float(indicators["atr"].get("pct") or 1.0) # Certainty (band width) band_pct = abs(blend_p90[-1] - blend_p10[-1]) / abs(blend_p50[-1]) if abs(blend_p50[-1]) > 1e-8 else 0.1 certainty = math.exp(-band_pct * 3.0) # Confidence: derived from agreement, magnitude (vs ATR), and certainty base_conf = 65.0 if agreement else 45.0 magnitude_bonus = min(20.0, (forecast_ret_pct / max(atr_pct, 0.1)) * 5.0) confidence = (base_conf + magnitude_bonus) * certainty # Penalize scale bias bias_penalty = min(15.0, abs(scale - 1.0) * 30.0) confidence = max(10.0, min(95.0, confidence - bias_penalty)) return { "p10": blend_p10, "p50": blend_p50, "p90": blend_p90, "model_weight": round(model_weight, 4), "anchor_weight": round(anchor_weight, 4), "agreement": agreement, "scale": round(scale, 6), "confidence": round(confidence, 2), "model_bias_pct": round(bias_pct, 4), } # ───────────────────────────────────────────────────────────────────────────── # MODULE: Analysis Engine v2.0 # ───────────────────────────────────────────────────────────────────────────── @dataclass class Signal: name: str value: float weight: float label: str detail: str = "" def _build_signals( indicators: Dict[str, Any], last_close: float, forecast_return_pct: float, confidence: float, interval: str, ) -> List[Signal]: """ Tạo danh sách 18 tín hiệu từ bộ indicators đã tính sẵn. Mỗi tín hiệu trả về giá trị trong [-1, +1]: +1 = bullish mạnh nhất -1 = bearish mạnh nhất 0 = trung tính """ signals: List[Signal] = [] # ── Lấy giá trị ────────────────────────────────────────────────────────── rsi = _safe(indicators["rsi"].get("value"), 50.0) ema9 = _safe(indicators["ema"].get("ema9"), last_close) ema21 = _safe(indicators["ema"].get("ema21"), last_close) ema50 = _safe(indicators["ema"].get("ema50"), last_close) ema200 = _safe(indicators["ema"].get("ema200"), last_close) macd_v = _safe(indicators["macd"].get("macd"), 0.0) macd_sig = _safe(indicators["macd"].get("signal"), 0.0) macd_h = _safe(indicators["macd"].get("histogram"), 0.0) bb_upper = _safe(indicators["bollinger"].get("upper"), last_close * 1.02) bb_mid = _safe(indicators["bollinger"].get("middle"), last_close) bb_lower = _safe(indicators["bollinger"].get("lower"), last_close * 0.98) bb_bw = _safe(indicators["bollinger"].get("bandwidth"), 0.02) atr = _safe(indicators["atr"].get("value"), last_close * 0.01) atr_pct = _safe(indicators["atr"].get("pct"), 1.0) stoch_k = _safe(indicators["stoch_rsi"].get("k"), 50.0) stoch_d = _safe(indicators["stoch_rsi"].get("d"), 50.0) vol_cur = _safe(indicators["volume"].get("last"), 0.0) vol_avg = _safe(indicators["volume"].get("sma20"), 1.0) # ── 1: RSI momentum ── if rsi > 70: rsi_score = _clamp(-((rsi - 70) / 30), -1.0, -0.1) rsi_lbl = f"RSI quá mua ({rsi:.1f})" elif rsi < 30: rsi_score = _clamp((30 - rsi) / 30, 0.1, 1.0) rsi_lbl = f"RSI quá bán ({rsi:.1f})" else: rsi_score = _clamp((rsi - 50) / 20.0, -1.0, 1.0) rsi_lbl = f"RSI trung tính ({rsi:.1f})" signals.append(Signal("rsi_momentum", rsi_score, 1.6, rsi_lbl)) # ── 2: Stochastic RSI ── st_score = _clamp((stoch_k - 50) / 30.0, -1.0, 1.0) kd_cross = 0.4 if stoch_k > stoch_d else -0.4 signals.append(Signal("stoch_rsi", _clamp(st_score + kd_cross * 0.3, -1, 1), 1.2, f"StochRSI K{'>' if kd_cross>0 else '<'}D ({stoch_k:.1f})")) # ── 3: MACD Crossover ── m_score = 1.0 if macd_v > macd_sig else -1.0 signals.append(Signal("macd_cross", m_score, 1.5, f"MACD {'Bullish' if m_score>0 else 'Bearish'}")) # ── 4: EMA Stack (9/21/50) ── ema_bull = (ema9 > ema21 > ema50) ema_bear = (ema9 < ema21 < ema50) e_score = 1.0 if ema_bull else -1.0 if ema_bear else 0.0 signals.append(Signal("ema_stack", e_score, 1.8, "Cấu trúc EMA stack")) # ── 5: Price vs EMA200 ── dist_200 = _pct(last_close, ema200) signals.append(Signal("price_vs_ema200", _clamp(dist_200 / 10.0, -1.0, 1.0), 1.6, f"Giá vs EMA200 ({dist_200:+.2f}%)")) # ── 6: EMA9 Slope ── dist_9 = _pct(last_close, ema9) signals.append(Signal("ema9_slope", _clamp(dist_9 / 1.5, -1.0, 1.0), 1.0, f"Giá vs EMA9 ({dist_9:+.2f}%)")) # ── 7: Bollinger Position ── bb_range = bb_upper - bb_lower bb_pos = (last_close - bb_lower) / bb_range if bb_range > 0 else 0.5 bb_score = 0.0 if bb_pos < 0.1: bb_score = 0.8 elif bb_pos > 0.9: bb_score = -0.8 signals.append(Signal("bollinger_position", bb_score, 1.1, f"Vị trí BB ({bb_pos:.0%})")) # ── 8: BB Squeeze ── signals.append(Signal("bb_squeeze", 0.0, 0.5, "BB Squeeze" if bb_bw < 0.015 else "BB Bình thường")) # ── 9: Volume Confirmation ── vol_ratio = vol_cur / vol_avg if vol_avg > 0 else 1.0 signals.append(Signal("volume_confirm", _clamp((vol_ratio - 1.0) * 0.5, -1, 1), 1.3, f"KLGD ({vol_ratio:.1f}x TB)")) # ── 10: AI Forecast ── signals.append(Signal("ai_forecast", _clamp(forecast_return_pct / 3.0, -1, 1) * (confidence/100.0), 2.0, f"AI {forecast_return_pct:+.2f}%")) # ── 11: EMA Cross (9 vs 21) ── signals.append(Signal("ema_cross_9_21", 1.0 if ema9 > ema21 else -1.0, 0.6, f"EMA9 {'>' if ema9>ema21 else '<'} EMA21")) # ── 12: RSI Extremes ── rsi_ext = 1.0 if rsi < 20 else -1.0 if rsi > 80 else 0.0 signals.append(Signal("rsi_extreme", rsi_ext, 0.7, "RSI cực đại" if rsi_ext !=0 else "RSI ổn định")) # ── 13: MACD Hist Momentum ── signals.append(Signal("macd_hist_momentum", _clamp(macd_h / (atr or 1), -1, 1), 0.9, f"MACD Hist ({macd_h:+.4f})")) # ── 14: ATR Volatility ── signals.append(Signal("atr_volatility", 0.0, 0.3, f"ATR % ({atr_pct:.1f}%)")) # ── 15: Price Momentum (5-candle) ── # BUG-01 Fix: Read from trend or top-level correctly mom_ref = _safe(indicators.get("short_momentum_ref") or indicators.get("trend", {}).get("short_momentum_ref"), last_close) signals.append(Signal("short_momentum", _pct(last_close, mom_ref), 0.8, "Momentum ngắn hạn")) # ── 16: Volatility Expansion ── signals.append(Signal("vol_expansion", 0.5 if vol_ratio > 2.0 else 0.0, 0.6, "Bùng nổ KLGD" if vol_ratio > 2.0 else "KLGD ổn định")) # ── 17: EMA50 Support/Resist ── signals.append(Signal("price_vs_ema50", _clamp(_pct(last_close, ema50) / 2.0, -1, 1), 1.0, f"Giá vs EMA50 ({_pct(last_close, ema50):+.2f}%)")) # ── 18: Trend Alignment ── align = 1.0 if (ema9 > ema200 and forecast_return_pct > 0) else -1.0 if (ema9 < ema200 and forecast_return_pct < 0) else 0.0 signals.append(Signal("trend_alignment", align, 1.2, "Đồng thuận xu hướng")) return signals @dataclass class PriceLevel: price: float type: str strength: int label: str def _calc_price_levels(data: List[Dict[str, Any]], atr: float) -> Dict[str, Any]: """B-15: Proper swing detection for support and resistance (BUG-15).""" if len(data) < 20: return {} closes = np.array([float(d["close"]) for d in data]) highs = np.array([float(d["high"]) for d in data]) lows = np.array([float(d["low"]) for d in data]) # Use previous COMPLETED candle for pivot calculation (BUG-P1-10) if len(highs) >= 2: last_h = float(highs[-2]) last_l = float(lows[-2]) last_c = float(closes[-2]) else: last_h, last_l, last_c = float(highs[-1]), float(lows[-1]), float(closes[-1]) pivots = _calc_pivot_points(last_h, last_l, last_c) # 1. Detect Pivot Highs and Lows (window=3) pivots_h = [] pivots_l = [] for i in range(3, len(highs)-3): if highs[i] == max(highs[i-3:i+4]): pivots_h.append(highs[i]) if lows[i] == min(lows[i-3:i+4]): pivots_l.append(lows[i]) # Fallback if no pivots if not pivots_h: pivots_h = [max(highs[-20:])] if not pivots_l: pivots_l = [min(lows[-20:])] # 2. Select nearest support and resistance ns = max([p for p in pivots_l if p < last_c], default=last_c - atr) nr = min([p for p in pivots_h if p > last_c], default=last_c + atr) return { "nearest_support": _round(ns, 6), "nearest_resistance":_round(nr, 6), "key_support": _round(ns - atr, 6), "key_resistance": _round(nr + atr, 6), "level_list": [ {"price": ns, "type": "support", "strength": 3, "label": f"Hỗ trợ {ns:.5g}"}, {"price": nr, "type": "resistance", "strength": 3, "label": f"Kháng cự {nr:.5g}"}, ] } def _classify_regime(indicators: Dict[str, Any], last_close: float, atr_pct: float) -> Tuple[str, str]: """Phân loại trạng thái thị trường dựa trên biến động và xu hướng.""" rsi = _safe(indicators["rsi"].get("value"), 50.0) ema9 = _safe(indicators["ema"].get("ema9"), last_close) ema50 = _safe(indicators["ema"].get("ema50"), last_close) ema200 = _safe(indicators["ema"].get("ema200"), last_close) # Logic 8 trạng thái if atr_pct > 3.5: if rsi > 60: return "volatile_bull", "Tăng trưởng biến động cao" if rsi < 40: return "volatile_bear", "Sụt giảm biến động cao" return "high_chaos", "Thị trường hỗn loạn / Vol cao" if last_close > ema200 and ema9 > ema50: if rsi > 70: return "overextended_bull", "Tăng trưởng quá mức (Overbought)" return "stable_bull", "Xu hướng tăng ổn định" if last_close < ema200 and ema9 < ema50: if rsi < 30: return "capitulation", "Hoảng loạn / Quá bán (Oversold)" return "stable_bear", "Xu hướng giảm ổn định" if abs(_pct(ema9, ema50)) < 0.5: return "tight_range", "Tích lũy biên độ hẹp (Squeeze)" return "sideways", "Thị trường đi ngang (Sideway)" def _calc_momentum(data: List[Dict[str, Any]]) -> Dict[str, Any]: """Tính toán momentum đa khung thời gian.""" if len(data) < 50: return {"short": 0, "mid": 0, "long": 0, "aligned": False} closes = [float(d["close"]) for d in data] last = closes[-1] short = _pct(last, closes[max(0, len(closes)-6)]) mid = _pct(last, closes[max(0, len(closes)-21)]) long = _pct(last, closes[max(0, len(closes)-51)]) aligned = (short > 0 and mid > 0 and long > 0) or (short < 0 and mid < 0 and long < 0) return { "short": round(short, 4), "mid": round(mid, 4), "long": round(long, 4), "short_lbl": f"{'Tăng' if short>0 else 'Giảm'} {abs(short):.2f}%", "aligned": aligned } @dataclass class ConfluenceResult: raw_score: float confluence: float bias: str conviction: str grade: str active_signals: int bull_count: int bear_count: int neutral_count: int def _calc_confluence(signals: List[Signal]) -> ConfluenceResult: tw = sum(s.weight for s in signals) ws = sum(s.value * s.weight for s in signals) raw = ws / tw if tw > 0 else 0.0 conf = _clamp(50.0 + raw * 50.0, 0.0, 100.0) bias = "bullish" if conf >= 60 else "bearish" if conf <= 40 else "neutral" return ConfluenceResult( raw_score=raw, confluence=conf, bias=bias, conviction="strong" if abs(conf-50)>20 else "moderate", grade="A" if abs(conf-50)>25 else "B", active_signals=len(signals), bull_count=sum(1 for s in signals if s.value > 0.2), bear_count=sum(1 for s in signals if s.value < -0.2), neutral_count=sum(1 for s in signals if abs(s.value) <= 0.2) ) @dataclass class TradeSetup: tier: str direction: str entry_low: float entry_high: float entry_mid: float stop_loss: float take_profit1: float take_profit2: float take_profit3: float risk_reward: float risk_pct: float valid: bool notes: str def _build_setups(bias: str, last_close: float, atr: float, levels: Dict[str, Any], projected: float) -> List[TradeSetup]: """Tạo các phương án giao dịch theo 3 cấp độ: Conservative, Standard, Aggressive.""" setups = [] ns = levels.get("nearest_support", last_close - atr) nr = levels.get("nearest_resistance", last_close + atr) ks = levels.get("key_support", ns - atr) kr = levels.get("key_resistance", nr + atr) if bias == "bullish": # 1. Conservative (Thận trọng - Chờ hồi) sl_c = ks - atr * 0.3 tp_c = nr setups.append(TradeSetup( tier="conservative", direction="long", entry_low=ks, entry_high=ns, entry_mid=(ks+ns)/2, stop_loss=sl_c, take_profit1=last_close + (last_close-sl_c), take_profit2=tp_c, take_profit3=tp_c + atr, risk_reward=_round((tp_c - ns) / (ns - sl_c), 2) if ns > sl_c else 0.0, risk_pct=_round((ns - sl_c) / ns * 100, 2), valid=True, notes="Đợi giá kiểm tra lại vùng hỗ trợ cứng trước khi tham gia." )) # 2. Standard (Tiêu chuẩn - Theo xu hướng) sl_s = ns - atr * 0.5 tp_s = projected setups.append(TradeSetup( tier="standard", direction="long", entry_low=ns, entry_high=last_close, entry_mid=(ns+last_close)/2, stop_loss=sl_s, take_profit1=last_close + atr, take_profit2=projected, take_profit3=max(projected, kr), risk_reward=_round((projected - last_close) / (last_close - sl_s), 2) if last_close > sl_s else 0.0, risk_pct=_round((last_close - sl_s) / last_close * 100, 2), valid=True, notes="Giao dịch theo đà tăng hiện tại, dừng lỗ dưới hỗ trợ gần nhất." )) # 3. Aggressive (Săn đuổi - Buy Stop / Breakout) sl_a = last_close - atr * 0.8 tp_a = max(projected, kr) + atr setups.append(TradeSetup( tier="aggressive", direction="long", entry_low=last_close, entry_high=last_close + atr*0.2, entry_mid=last_close+atr*0.1, stop_loss=sl_a, take_profit1=last_close + atr*1.5, take_profit2=tp_a - atr, take_profit3=tp_a, risk_reward=_round((tp_a - last_close) / (last_close - sl_a), 2) if last_close > sl_a else 0.0, risk_pct=_round((last_close - sl_a) / last_close * 100, 2), valid=True, notes="Vào lệnh trực tiếp để bắt kịp sóng tăng mạnh. Rủi ro cao hơn." )) elif bias == "bearish": # 1. Conservative (Thận trọng - Chờ hồi) sl_c = kr + atr * 0.3 tp_c = ns setups.append(TradeSetup( tier="conservative", direction="short", entry_low=nr, entry_high=kr, entry_mid=(nr+kr)/2, stop_loss=sl_c, take_profit1=last_close - (sl_c-last_close), take_profit2=tp_c, take_profit3=tp_c - atr, risk_reward=_round((nr - tp_c) / (sl_c - nr), 2) if sl_c > nr else 0.0, risk_pct=_round((sl_c - nr) / nr * 100, 2), valid=True, notes="Chờ giá hồi lên vùng kháng cự mạnh để tối ưu RR." )) # Standard Short sl_s = nr + atr * 0.5 tp_s = projected setups.append(TradeSetup( tier="standard", direction="short", entry_low=last_close, entry_high=nr, entry_mid=(last_close+nr)/2, stop_loss=sl_s, take_profit1=last_close - atr, take_profit2=projected, take_profit3=min(projected, ks), risk_reward=_round((last_close - projected) / (sl_s - last_close), 2) if sl_s > last_close else 0.0, risk_pct=_round((sl_s - last_close) / last_close * 100, 2), valid=True, notes="Bán theo xu hướng giảm, dừng lỗ trên kháng cự gần nhất." )) return [s for s in setups if s.valid] def _build_scenarios( bias: str, last_close: float, atr: float, levels: Dict[str, Any], forecast_rows: List[Dict[str, Any]], confluence: float, momentum: Dict[str, Any] ) -> List[Dict[str, Any]]: """Phân tích các kịch bản có thể xảy ra.""" fc_p50 = float(forecast_rows[-1]["p50"]) fc_p90 = float(forecast_rows[-1]["p90"]) fc_p10 = float(forecast_rows[-1]["p10"]) # Kịch bản cơ sở (P50) base = { "name": "base", "label": "Kịch bản Cơ sở", "probability": round(confluence, 1), "price_target": _round(fc_p50, 6), "return_pct": _round(_pct(fc_p50, last_close), 2), "catalyst": "Duy trì xu hướng hiện tại", "invalidation": "Phá vỡ vùng EMA21" } # BUG-10: Normalize probabilities to ensure sum = 100% # Using raw scores from confluence and bias bull_raw = (100 - confluence) * 0.4 + (20 if bias == 'bullish' else 0) base_raw = confluence bear_raw = max(5.0, 100.0 - base_raw - bull_raw) # Min 5% for bear in base calc # Normalize total = bull_raw + base_raw + bear_raw bull_prob = (bull_raw / total) * 100 base_prob = (base_raw / total) * 100 bear_prob = 100.0 - bull_prob - base_prob # Exact sum base["probability"] = round(base_prob, 1) bull = { "name": "bull", "label": "Kịch bản Tích cực", "probability": round(bull_prob, 1), "price_target": _round(fc_p90, 6), "return_pct": _round(_pct(fc_p90, last_close), 2), "catalyst": "Đột phá Momentum mạnh", "invalidation": "RSI quay đầu giảm" } bear = { "name": "bear", "label": "Kịch bản Tiêu cực", "probability": round(bear_prob, 1), "price_target": _round(fc_p10, 6), "return_pct": _round(_pct(fc_p10, last_close), 2), "catalyst": "Đảo chiều bất ngờ / Tin xấu", "invalidation": "Hỗ trợ cứng được giữ vững" } return [base, bull, bear] def _build_risk_framework( bias: str, last_close: float, atr: float, atr_pct: float, levels: Dict[str, Any], setups: List[TradeSetup], interval: str ) -> Dict[str, Any]: """Cung cấp hướng dẫn quản trị rủi ro.""" vol_adjusted_risk = _clamp(2.0 - (atr_pct / 5.0), 0.5, 1.5) advice = "Kích thước vị thế tiêu chuẩn (1% tài khoản)." if atr_pct > 4.0: advice = "Thị trường biến động mạnh: Giảm 50% khối lượng lệnh." elif atr_pct < 0.5: advice = "Biến động thấp: Cân nhắc chốt lời ngắn (Scalp)." return { "max_risk_pct": round(vol_adjusted_risk, 2), "position_size_advice": advice, "stop_loss_type": "Volatility-based (ATR)", "invalidation_point": levels.get("key_resistance" if bias=="bearish" else "key_support", last_close) } def _build_reasoning( bias: str, conviction: str, regime_key: str, regime_lbl: str, signals: List[Signal], indicators: Dict[str, Any], momentum: Dict[str, Any], levels: Dict[str, Any], last_close: float, atr_pct: float, interval: str, forecast_ret: float ) -> Tuple[List[str], List[str], List[str]]: """Tự động sinh các lý do, cảnh báo và cơ hội bằng tiếng Việt chuyên sâu.""" reasons, warnings, opportunities = [], [], [] # Logic Lý do if bias == "bullish": reasons.append(f"Xu hướng chủ đạo là TĂNG ({conviction}) trên khung {interval}.") reasons.append(f"Dự báo AI cho thấy tiềm năng tăng trưởng {forecast_ret:+.2f}% trong ngắn hạn.") if indicators["trend"].get("ema_bullish_stack"): reasons.append("Hệ thống EMA đang xếp chồng Bullish mạnh mẽ, xác nhận lực mua áp ảo.") elif bias == "bearish": reasons.append(f"Áp lực GIẢM giá chiếm ưu thế ({conviction}) trên khung {interval}.") reasons.append(f"AI nhận diện tín hiệu suy yếu với mục tiêu giảm về vùng {forecast_ret:+.2f}%.") else: reasons.append(f"Thị trường đang trong trạng thái TÍCH LŨY / ĐI NGANG trên khung {interval}.") reasons.append("Chưa có tín hiệu bứt phá rõ rệt từ các chỉ báo kỹ thuật quan trọng.") if regime_lbl: reasons.append(f"Trạng thái thị trường hiện tại: {regime_lbl}.") # Logic Cảnh báo rsi = _safe(indicators["rsi"].get("value"), 50.0) if rsi > 70: warnings.append("Chỉ số RSI đi vào vùng quá mua (>70), rủi ro đảo chiều kỹ thuật cao.") if atr_pct > 3.0: warnings.append("Biến động thị trường đang ở mức cao (ATR), ưu tiên quản lý vốn chặt chẽ.") # Logic Cơ hội if bias == "bullish": opportunities.append(f"Cơ hội Long khi giá điều chỉnh về vùng hỗ trợ {levels.get('nearest_support')}.") else: opportunities.append(f"Cơ hội Short khi giá hồi phục chạm kháng cự {levels.get('nearest_resistance')}.") return reasons, warnings, opportunities # ── Technical Analysis Weights & Constants (Dashboard Rework v6.1) ──────────── OSC_WEIGHTS = { "rsi": 2.1, "macd": 2.4, "stoch_rsi": 1.1, "stoch": 1.2, "cci": 1.7, "adx": 1.8, "williams": 1.5, "ultimate": 1.1, "bbp": 1.0, "ao": 0.9, "momentum": 1.6, "roc": 1.3, "trix": 1.4, "ppo": 1.4, "cmo": 1.2, "dpo": 0.8, "aroon": 1.3, "tsi": 1.4, "demarker": 1.2, "force_index": 1.4, } MA_WEIGHT_MAP = { "ema_1_5": 0.9, "ema_5_10": 1.1, "ema_10_20": 1.35, "ema_20_50": 1.8, "ema_50_100": 2.2, "ema_100_200": 2.7, } def _extract_osc_key(label: str) -> str: l = label.lower() if "sức mạnh tương đối" in l or ("rsi" in l and "nhanh" not in l): return "rsi" if "macd" in l: return "macd" if "stochastic %k" in l: return "stoch" if "nhanh" in l or "stoch_rsi" in l: return "stoch_rsi" if "kênh hàng hóa" in l or "cci" in l: return "cci" if "định hướng" in l or "adx" in l: return "adx" if "williams" in l: return "williams" if "ultimate" in l: return "ultimate" if "bbp" in l or "sức mạnh giá" in l: return "bbp" if "ao" in l: return "ao" if "xung lượng" in l or "momentum" in l: return "momentum" if "roc" in l: return "roc" if "trix" in l: return "trix" if "ppo" in l: return "ppo" if "cmo" in l: return "cmo" if "dpo" in l: return "dpo" if "aroon" in l: return "aroon" if "tsi" in l: return "tsi" if "demarker" in l: return "demarker" if "force index" in l: return "force_index" return "unknown" def _get_ma_weight(label: str) -> float: l = label.lower() if l in MA_WEIGHT_MAP: return MA_WEIGHT_MAP[l] pair = re.findall(r"\d+", l) if len(pair) >= 2: return MA_WEIGHT_MAP.get(f"ema_{pair[0]}_{pair[1]}", 1.0) return 1.0 def _gauge_to_signal(gauge: float, interval: str = "1h") -> str: """Interval-aware 5-level signal converter.""" thresholds = { "1m": (80, 62, 38, 20), "5m": (80, 62, 38, 20), "15m": (77, 60, 40, 23), "1h": (77, 60, 40, 23), "4h": (75, 58, 42, 25), "1d": (75, 58, 42, 25), "1w": (72, 56, 44, 28), } strong_buy, buy, sell, strong_sell = thresholds.get(interval, thresholds["1h"]) if gauge >= strong_buy: return "Mua mạnh" if gauge >= buy: return "Mua" if gauge > sell: return "Trung lập" if gauge > strong_sell: return "Bán" return "Bán mạnh" def _gauge_to_normalized_score(gauge: float) -> float: """Convert a 0..100 gauge into a -1..1 frontend-friendly scale.""" return _clamp((float(gauge) - 50.0) / 50.0, -1.0, 1.0) def _forecast_path_metrics(p50_path: np.ndarray, last_close: float) -> Dict[str, float]: """Measure forecast quality from the full path, not only the final endpoint.""" if len(p50_path) == 0 or abs(last_close) <= 1e-8: return { "weighted_return_pct": 0.0, "final_return_pct": 0.0, "path_consistency": 50.0, "monotonicity": 50.0, "max_adverse_excursion_pct": 0.0, "mean_step_return_pct": 0.0, } ret_path = ((p50_path / last_close) - 1.0) * 100.0 step_weights = np.linspace(1.0, 0.65, len(ret_path)) weighted_ret_pct = float(np.average(ret_path, weights=step_weights)) final_ret_pct = float(ret_path[-1]) final_sign = 0 if abs(final_ret_pct) < 0.05 else (1 if final_ret_pct > 0 else -1) if len(ret_path) > 1 and final_sign != 0: signed_steps = [ 1.0 if np.sign(curr - prev) == final_sign else 0.0 for prev, curr in zip(ret_path[:-1], ret_path[1:]) if abs(curr - prev) >= 0.02 ] path_consistency = float(sum(signed_steps) / len(signed_steps)) if signed_steps else 0.5 else: path_consistency = 0.5 if len(ret_path) > 1: monotonicity = float(np.mean(np.diff(ret_path) >= 0)) if final_sign >= 0 else float(np.mean(np.diff(ret_path) <= 0)) else: monotonicity = 0.5 if final_sign > 0: adverse = abs(float(np.min(ret_path))) elif final_sign < 0: adverse = abs(float(np.max(ret_path))) else: adverse = max(abs(float(np.min(ret_path))), abs(float(np.max(ret_path)))) mean_step_return_pct = float(np.mean(np.diff(ret_path))) if len(ret_path) > 1 else final_ret_pct return { "weighted_return_pct": round(weighted_ret_pct, 2), "final_return_pct": round(final_ret_pct, 2), "path_consistency": round(path_consistency * 100.0, 1), "monotonicity": round(monotonicity * 100.0, 1), "max_adverse_excursion_pct": round(adverse, 2), "mean_step_return_pct": round(mean_step_return_pct, 3), } 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)) sell_f = float(max(0, sell)) neutral_f = float(max(0, neutral)) denominator = buy_f + sell_f + neutral_f if denominator <= 0: return 50.0 raw_gauge = 50.0 + 50.0 * ((buy_f - sell_f) / denominator) return max(0.0, min(100.0, raw_gauge)) def _calc_osc_score(osc_data: list, interval: str = "1h") -> dict: """MODULE 1: Equal-weight vote scoring for oscillators.""" buy = sell = neutral = 0 buy_weight = sell_weight = neutral_weight = 0.0 for item in osc_data: key = item.get("key") or _extract_osc_key(item["name"]) w = OSC_WEIGHTS.get(key, 1.0) # Robust case-insensitive check act = str(item.get("action", "")).strip().lower() if act == "mua": buy += 1 buy_weight += w elif act == "bán": sell += 1 sell_weight += w else: neutral += 1 neutral_weight += w gauge = _calc_vote_gauge(buy, sell, neutral) normalized = _gauge_to_normalized_score(gauge) return { "gauge": round(gauge, 1), "normalized_score": round(normalized, 4), "signal": _gauge_to_signal(gauge, interval), "buy": buy, "sell": sell, "neutral": neutral, "buy_weight": round(buy_weight, 2), "sell_weight": round(sell_weight, 2), "neutral_weight": round(neutral_weight, 2), } def _calc_ma_score(ma_data: list, closes: np.ndarray, interval: str = "1h") -> dict: """MODULE 2: Equal-weight vote scoring for moving averages.""" buy = sell = neutral = 0 buy_weight = sell_weight = neutral_weight = 0.0 for item in ma_data: w = _get_ma_weight(item.get("key") or item["name"]) act = str(item.get("action", "")).strip().lower() if act == "mua": buy += 1 buy_weight += w elif act == "bán": sell += 1 sell_weight += w else: neutral += 1 neutral_weight += w gauge = _calc_vote_gauge(buy, sell, neutral) normalized = _gauge_to_normalized_score(gauge) ema50 = _ema(closes, 50)[-1] if len(closes) >= 50 else float("nan") ema200 = _ema(closes, 200)[-1] if len(closes) >= 200 else float("nan") return { "gauge": round(gauge, 1), "normalized_score": round(normalized, 4), "signal": _gauge_to_signal(gauge, interval), "buy": buy, "sell": sell, "neutral": neutral, "buy_weight": round(buy_weight, 2), "sell_weight": round(sell_weight, 2), "neutral_weight": round(neutral_weight, 2), "structure_bonus": 0.0, "golden_cross": bool(not any(math.isnan(x) for x in (ema50, ema200)) and ema50 > ema200), "death_cross": bool(not any(math.isnan(x) for x in (ema50, ema200)) and ema50 < ema200), } def _calc_ai_forecast_score( blended: dict, forecast_rows: List[Dict[str, Any]], last_close: float, 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", {}) atr_pct = max(float(indicators.get("atr", {}).get("pct") or 0.0), 0.1) rsi = float(indicators.get("rsi", {}).get("value") or 50.0) p50_path = np.array(blended.get("p50", []), dtype=float) p10_path = np.array(blended.get("p10", []), dtype=float) p90_path = np.array(blended.get("p90", []), dtype=float) if not len(p50_path): p50_path = np.array([last_close], dtype=float) if not len(p10_path): p10_path = np.array([last_close], dtype=float) 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, 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) path_consistency = path_metrics["path_consistency"] / 100.0 monotonicity = path_metrics["monotonicity"] / 100.0 adverse_excursion_pct = path_metrics["max_adverse_excursion_pct"] avg_band_pct = float(np.mean((p90_path - p10_path) / np.maximum(np.abs(p50_path), 1e-8)) * 100.0) band_certainty = math.exp(-avg_band_pct / 4.0) ensemble_conf = max(0.0, min(1.0, float(blended.get("confidence", 50.0)) / 100.0)) scale_penalty = min(0.18, abs(float(blended.get("scale", 1.0)) - 1.0) * 0.35) stability_penalty = min(0.20, adverse_excursion_pct / max(atr_pct * 5.0, 1.0) * 0.12) certainty_score = ( ensemble_conf * 0.45 + band_certainty * 0.35 + path_consistency * 0.12 + monotonicity * 0.08 ) - scale_penalty - stability_penalty certainty_score = max(0.08, min(0.98, certainty_score)) move_in_atr = abs(final_ret_pct) / atr_pct magnitude_score = math.tanh(move_in_atr / 1.6) horizon_decay = max(0.70, 1.0 - max(horizon - 12, 0) * 0.012) magnitude_score *= horizon_decay ema_stack = bool(trend.get("ema_bullish_stack", False)) above_200 = bool(trend.get("above_ema200", False)) trend_alignment = 0.0 if final_sign > 0 and ema_stack and above_200: trend_alignment = 0.10 elif final_sign < 0 and (not ema_stack) and (not above_200): trend_alignment = -0.10 elif final_sign > 0 and not above_200: trend_alignment = -0.07 elif final_sign < 0 and above_200: trend_alignment = 0.07 exhaustion_penalty = 0.0 if final_sign > 0 and rsi >= 74: exhaustion_penalty = min(0.12, (rsi - 74.0) / 100.0) elif final_sign < 0 and rsi <= 26: exhaustion_penalty = -min(0.12, (26.0 - rsi) / 100.0) path_strength = 0.55 + 0.25 * path_consistency + 0.20 * monotonicity effective_strength = direction_norm * path_strength * (0.45 + 0.55 * certainty_score) directional_push = effective_strength * (22.0 + 18.0 * magnitude_score) alignment_push = trend_alignment * 35.0 exhaustion_push = -exhaustion_penalty * 35.0 gauge = max(8.0, min(92.0, 50.0 + directional_push + alignment_push + exhaustion_push)) confidence_pct = 35.0 + certainty_score * 55.0 + min(move_in_atr, 1.5) * 6.0 confidence_pct = max(20.0, min(95.0, confidence_pct)) 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), "confidence_pct": round(confidence_pct, 1), "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), "band_uncertainty_pct": round(avg_band_pct, 2), "certainty": round(certainty_score * 100.0, 1), "path_consistency": round(path_consistency * 100.0, 1), "monotonicity": round(monotonicity * 100.0, 1), "max_adverse_excursion_pct": round(adverse_excursion_pct, 2), "path_metrics": path_metrics, "signal": _gauge_to_signal(gauge, interval) } def _calc_summary_score(osc_score: dict, ma_score: dict, ai_score: dict) -> dict: """MODULE 4: Normalize + Weighted combine with compatibility fields.""" W_OSC = 0.35 W_MA = 0.35 W_AI = 0.30 composite = (osc_score["gauge"] * W_OSC + ma_score["gauge"] * W_MA + ai_score["gauge"] * W_AI) # Confidence multiplier: pull towards neutral if AI certainty is low certainty_factor = ai_score["certainty"] / 100.0 pulled_to_neutral = composite + (50.0 - composite) * (1.0 - certainty_factor) * 0.20 final_gauge = max(5.0, min(95.0, pulled_to_neutral)) dist = abs(final_gauge - 50.0) conviction = "Rất mạnh" if dist >= 25 else "Mạnh" if dist >= 15 else "Trung bình" if dist >= 8 else "Yếu" # Compatibility fields for legacy frontend (Total votes) buy = osc_score["buy"] + ma_score["buy"] sell = osc_score["sell"] + ma_score["sell"] neutral = osc_score["neutral"] + ma_score["neutral"] bias = "neutral" if final_gauge >= 58: bias = "bullish" elif final_gauge <= 42: bias = "bearish" return { "gauge": round(final_gauge, 1), "signal": _gauge_to_signal(final_gauge), "conviction": conviction, "bias": bias, "buy": buy, "sell": sell, "neutral": neutral, "components": { "oscillators": round(osc_score["gauge"], 1), "moving_averages": round(ma_score["gauge"], 1), "ai_forecast": round(ai_score["gauge"], 1) } } def _calc_technical_score_v2(osc_score: dict, ma_score: dict, regime: str = "sideways", interval: str = "1h") -> dict: """Blend oscillators and moving averages with equal credibility per vote.""" buy = osc_score["buy"] + ma_score["buy"] sell = osc_score["sell"] + ma_score["sell"] neutral = osc_score["neutral"] + ma_score["neutral"] final_gauge = _calc_vote_gauge(buy, sell, neutral) osc_delta = osc_score["gauge"] - 50.0 ma_delta = ma_score["gauge"] - 50.0 osc_dir = math.copysign(1.0, osc_delta) if abs(osc_delta) >= 1.0 else 0.0 ma_dir = math.copysign(1.0, ma_delta) if abs(ma_delta) >= 1.0 else 0.0 return { "gauge": round(final_gauge, 1), "normalized_score": round(_gauge_to_normalized_score(final_gauge), 4), "signal": _gauge_to_signal(final_gauge, interval), "buy": buy, "sell": sell, "neutral": neutral, "buy_weight": round(osc_score.get("buy_weight", 0.0) + ma_score.get("buy_weight", 0.0), 2), "sell_weight": round(osc_score.get("sell_weight", 0.0) + ma_score.get("sell_weight", 0.0), 2), "neutral_weight": round(osc_score.get("neutral_weight", 0.0) + ma_score.get("neutral_weight", 0.0), 2), "alignment": osc_dir == ma_dir and osc_dir != 0.0, "components": { "oscillators": round(osc_score["gauge"], 1), "moving_averages": round(ma_score["gauge"], 1), } } def _calc_summary_score_v2(tech_score: dict, ai_score: dict, interval: str = "1h") -> dict: """Final decision score derived from AI certainty weighting.""" tech_gauge = float(tech_score["gauge"]) ai_gauge = float(ai_score["gauge"]) ai_certainty = _clamp(float(ai_score.get("certainty", 50.0)) / 100.0, 0.0, 1.0) tech_weight = 1.0 - ai_certainty ai_weight = ai_certainty final_gauge = max(5.0, min(95.0, (ai_gauge * ai_weight) + (tech_gauge * tech_weight))) tech_delta = tech_gauge - 50.0 ai_delta = ai_gauge - 50.0 tech_dir = 0 if abs(tech_delta) < 2.0 else (1 if tech_delta > 0 else -1) ai_dir = 0 if abs(ai_delta) < 2.0 else (1 if ai_delta > 0 else -1) dist = abs(final_gauge - 50.0) conviction = "R?t m?nh" if dist >= 25 else "M?nh" if dist >= 15 else "Trung b?nh" if dist >= 8 else "Y?u" bias = "neutral" if final_gauge >= 58: bias = "bullish" elif final_gauge <= 42: bias = "bearish" return { "gauge": round(final_gauge, 1), "normalized_score": round(_gauge_to_normalized_score(final_gauge), 4), "signal": _gauge_to_signal(final_gauge, interval), "conviction": conviction, "bias": bias, "agreement": tech_dir != 0 and tech_dir == ai_dir, "buy": tech_score["buy"], "sell": tech_score["sell"], "neutral": tech_score["neutral"], "buy_weight": tech_score.get("buy_weight", 0.0), "sell_weight": tech_score.get("sell_weight", 0.0), "neutral_weight": tech_score.get("neutral_weight", 0.0), "components": { "technical": round(tech_gauge, 1), "oscillators": round(tech_score["components"]["oscillators"], 1), "moving_averages": round(tech_score["components"]["moving_averages"], 1), "ai_forecast": round(ai_gauge, 1), "ai_weight": round(ai_weight, 2), "technical_weight": round(tech_weight, 2), "certainty_pct": round(ai_certainty * 100.0, 1), "formula": "ai_gauge * certainty + technical_gauge * (1 - certainty)", } } def _build_dashboard_payload( last_close: float, forecast_rows: List[Dict[str, Any]], technical_score: Dict[str, Any], ai_score: Dict[str, Any], summary: Dict[str, Any], ) -> Dict[str, Any]: """Single source of truth for hero gauges consumed by the frontend.""" forecast_end = last_close if forecast_rows: forecast_end = float(forecast_rows[-1].get("p50") or last_close) return { "technical": { "gauge": technical_score["gauge"], "normalized_score": technical_score["normalized_score"], "signal": technical_score["signal"], "buy": technical_score["buy"], "sell": technical_score["sell"], "neutral": technical_score["neutral"], "buy_weight": technical_score.get("buy_weight", 0.0), "sell_weight": technical_score.get("sell_weight", 0.0), "neutral_weight": technical_score.get("neutral_weight", 0.0), }, "ai": { "gauge": ai_score["gauge"], "normalized_score": ai_score["normalized_score"], "signal": ai_score["signal"], "forecast_return_pct": ai_score["forecast_return_pct"], "weighted_return_pct": ai_score["weighted_return_pct"], "confidence_pct": ai_score["confidence_pct"], "certainty": ai_score["certainty"], "path_consistency": ai_score["path_consistency"], "monotonicity": ai_score.get("monotonicity", 50.0), "max_adverse_excursion_pct": ai_score.get("max_adverse_excursion_pct", 0.0), "current_price": round(float(last_close), 6), "forecast_price": round(float(forecast_end), 6), }, "summary": { "gauge": summary["gauge"], "normalized_score": summary["normalized_score"], "signal": summary["signal"], "conviction": summary["conviction"], "bias": summary["bias"], "agreement": summary.get("agreement", False), "buy_weight": summary.get("buy_weight", 0.0), "sell_weight": summary.get("sell_weight", 0.0), "neutral_weight": summary.get("neutral_weight", 0.0), "components": summary.get("components", {}), }, } async def get_indicators_cached( symbol: str, interval: str, limit: int, refresh: bool = False, min_context: int = INDICATOR_MIN_CONTEXT, ) -> Tuple[List[Dict[str, Any]], str, Dict[str, Any]]: data, source = await fetch_historical( symbol, interval, limit, refresh=refresh, min_context=max(limit, min_context), ) cache_key = f"ind_{_cache_prefix(symbol, interval)}len={len(data)}" if refresh: indicators_cache.delete(cache_key) else: cached = indicators_cache.get(cache_key) if cached is not None: try: return data, str(cached["source"]), dict(cached["indicators"]) except Exception: indicators_cache.delete(cache_key) inflight_key = f"{cache_key}:refresh={int(refresh)}" task = None if refresh else _INDICATORS_INFLIGHT.get(inflight_key) if task is None: task = asyncio.create_task( asyncio.to_thread(compute_indicators, data), name=f"indicators:{symbol}:{interval}:{len(data)}", ) _INDICATORS_INFLIGHT[inflight_key] = task try: indicators = await task finally: if _INDICATORS_INFLIGHT.get(inflight_key) is task: _INDICATORS_INFLIGHT.pop(inflight_key, None) indicators_cache.set( cache_key, {"source": source, "indicators": indicators}, ttl_seconds=indicators_ttl(interval), ) return data, source, indicators def _rebuild_blended_from_forecast_payload(cached_forecast: Optional[Dict[str, Any]], last_close: float) -> Optional[Dict[str, Any]]: """Reconstruct the minimum blended payload needed by the analysis engine.""" if not cached_forecast: return None rows = cached_forecast.get("forecast") or [] future_rows = [row for row in rows if not row.get("is_actual")] if not future_rows: return None ensemble = cached_forecast.get("ensemble", {}) return { "p10": np.array([float(row.get("p10") or last_close) for row in future_rows], dtype=float), "p50": np.array([float(row.get("p50") or last_close) for row in future_rows], dtype=float), "p90": np.array([float(row.get("p90") or last_close) for row in future_rows], dtype=float), "agreement": bool(ensemble.get("trend_agreement", True)), "scale": float(ensemble.get("alignment_scale", 1.0) or 1.0), "confidence": float(ensemble.get("confidence", 50.0) or 50.0), } 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, forecast_reference_price: Optional[float] = None, ) -> Dict[str, Any]: """ TradingView-style technical analysis dashboard (v6.1 Rework). Implements weighted scoring for Oscillators, MAs, and AI Forecast. """ if not data or len(data) < 30: return {"oscillators": {"data": []}, "moving_averages": {"data": []}, "summary": {"signal": "Trung lập", "buy": 0, "sell": 0, "neutral": 0}} closes = np.array([float(d["close"]) for d in data], dtype=float) highs = np.array([float(d["high"]) for d in data], dtype=float) 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 if isinstance(arr, (float, int)): return round(float(arr), 2) if isinstance(arr, pd.Series): arr = arr.values v = arr[-1] if (hasattr(arr, '__len__') and len(arr) > 0) else float('nan') if v is None or (isinstance(v, float) and math.isnan(v)): return None return round(float(v), 2) ema50_arr = _ema(closes, 50) ema50_current = _lv(ema50_arr) ema200_current = _lv(indicators["ema"].get("ema200")) ema50_prev = ( float(ema50_arr[-6]) if len(ema50_arr) >= 6 and not math.isnan(float(ema50_arr[-6])) else (ema50_current or last_close) ) regime, regime_label = _classify_regime(indicators, last_close, _safe(indicators.get("atr", {}).get("pct"), 1.0)) # ── 1. Oscillators ── osc_data = [] atr_scale = max(float(indicators.get("atr", {}).get("value") or last_close * 0.01), max(last_close * 0.0015, 1e-6)) def _add_osc(label, val, action_name, key=None, **kw): if isinstance(val, (np.ndarray, pd.Series, list)): v = _lv(val) else: v = round(float(val), 2) if val is not None else None osc_key = key or action_name osc_kwargs = { "last_close": last_close, "ema50": ema50_current or last_close, "ema200": ema200_current or last_close, "ema50_prev": ema50_prev, "atr": atr_scale, **kw, } act = _osc_action(action_name, v if v is not None else 0, **osc_kwargs) score = _osc_signal_score(action_name, v if v is not None else 0.0, **osc_kwargs) osc_data.append({"key": osc_key, "name": label, "value": v, "action": act, "score": round(score, 4)}) stoch_price_k, stoch_price_d = _stoch_kd(highs, lows, closes, 14, 3, 3) stoch_rsi_k, stoch_rsi_d = _stoch_rsi(closes, 14, 14, 3, 3) demarker14 = _demarker(highs, lows, 14) force_idx13 = _force_index(closes, vols, 13) force_scale = float(np.nanmean(np.abs(force_idx13[-20:]))) if len(force_idx13) else 1.0 if not math.isfinite(force_scale) or force_scale <= 1e-8: force_scale = max(float(np.nanmean(np.abs(force_idx13))) if len(force_idx13) else 1.0, 1.0) _add_osc("Chỉ số Sức mạnh tương đối (14)", _rsi(closes, 14), "rsi", key="rsi") _add_osc("Stochastic %K (14, 3, 3)", stoch_price_k, "stoch", key="stoch") _add_osc("Chỉ số Kênh hàng hóa (20)", _cci(highs, lows, closes, 20), "cci", key="cci") adx_vals = _adx(highs, lows, closes, 14) _add_osc("Chỉ số Định hướng Trung bình (14)", adx_vals[0], "adx", key="adx", plus_di=_lv(adx_vals[1]) or 0, minus_di=_lv(adx_vals[2]) or 0) ao_vals = _awesome_oscillator(highs, lows) ao_prev = _lv(ao_vals[:-1]) if len(ao_vals) > 1 else _lv(ao_vals) _add_osc("Chỉ số Dao động AO", ao_vals, "ao", key="ao", scale=atr_scale * 0.8, prev=ao_prev or 0.0) _add_osc("Xung lượng (10)", _momentum(closes, 10), "momentum", key="momentum", scale=atr_scale * 1.15) macd_vals = _macd(closes, 12, 26, 9) _add_osc("Cấp độ MACD (12, 26)", macd_vals[0], "macd", key="macd", signal=_lv(macd_vals[1]) or 0, scale=atr_scale * 0.18) _add_osc("Đường RSI Nhanh (3, 3, 14, 14)", stoch_rsi_k, "stoch_rsi", key="stoch_rsi") _add_osc("Vùng Phần trăm Williams (14)", _williams_r(highs, lows, closes, 14), "williams", key="williams") bbp_vals = _bull_bear_power(highs, lows, closes, 13) _add_osc("Sức Mạnh Giá Lên và Giá Xuống", bbp_vals, "bbp", key="bbp", scale=atr_scale * 0.9, bbp_norm=(_lv(bbp_vals) or 0.0) / atr_scale) _add_osc("Dao động Ultimate (7, 14, 28)", _ultimate_oscillator(highs, lows, closes, 7, 14, 28), "ultimate", key="ultimate") _add_osc("Tốc độ biến động ROC (12)", _roc(closes, 12), "roc", key="roc") _add_osc("TRIX (18)", _trix(closes, 18), "trix", key="trix") _add_osc("PPO (12, 26)", _ppo(closes, 12, 26), "ppo", key="ppo") _add_osc("CMO (14)", _cmo(closes, 14), "cmo", key="cmo") _add_osc("DPO (20)", _dpo(closes, 20), "dpo", key="dpo", scale=atr_scale * 0.75) _add_osc("Aroon Oscillator (25)", _aroon_oscillator(highs, lows, 25), "aroon", key="aroon") _add_osc("TSI (25, 13)", _tsi(closes, 25, 13), "tsi", key="tsi") _add_osc("DeMarker (14)", demarker14, "demarker", key="demarker") _add_osc("Force Index (13)", force_idx13, "force_index", key="force_index", scale=force_scale) osc_score = _calc_osc_score(osc_data, interval) # ── 2. Moving Averages ── ma_data = [] ema_map: Dict[int, np.ndarray] = { 1: closes.copy(), 5: _ema(closes, 5), 10: _ema(closes, 10), 20: _ema(closes, 20), 50: _ema(closes, 50), 100: _ema(closes, 100), 200: _ema(closes, 200), } def _add_ema_pair_row(fast_period: int, slow_period: int) -> None: fast_arr = ema_map[fast_period] slow_arr = ema_map[slow_period] fast_value = _lv(fast_arr) slow_value = _lv(slow_arr) pair_key = f"ema_{fast_period}_{slow_period}" act = _ema_pair_action(fast_value, slow_value) score = _ema_pair_score(fast_value, slow_value, pair_key) ma_data.append( { "key": pair_key, "name": f"EMA{fast_period} / EMA{slow_period}", "value": f"{fast_value if fast_value is not None else '—'} / {slow_value if slow_value is not None else '—'}", "action": act, "score": round(score, 4), "fast": fast_value, "slow": slow_value, } ) for fast_period, slow_period in [(1, 5), (5, 10), (10, 20), (20, 50), (50, 100), (100, 200)]: _add_ema_pair_row(fast_period, slow_period) ma_score = _calc_ma_score(ma_data, closes, interval) # ── 3. AI Forecast Gauge ── if not blended: # Fallback to simple blended if no forecast available blended = { "p50": [last_close * (1 + (confidence-50)/1000)], "p10": [last_close * 0.98], "p90": [last_close * 1.02], "agreement": True, "scale": 1.0 } ai_score = _calc_ai_forecast_score( blended, forecast_rows, last_close, indicators, len(forecast_rows) or 10, interval, model_reference_price=model_reference_price, ) # ── 4. Summary ── technical_score = _calc_technical_score_v2(osc_score, ma_score, regime, interval) summary = _calc_summary_score_v2(technical_score, ai_score, interval) # Pivot Points last_h = float(highs[-2]) if len(highs) > 1 else float(highs[-1]) last_l = float(lows[-2]) if len(lows) > 1 else float(lows[-1]) last_c = float(closes[-2]) if len(closes) > 1 else float(closes[-1]) pivots = _calc_pivot_points(last_h, last_l, last_c) dashboard = _build_dashboard_payload(last_close, forecast_rows, technical_score, ai_score, summary) return { "style": "tradingview", "regime": {"key": regime, "label": regime_label}, "dashboard": dashboard, "summary": summary, "technicals": technical_score, "oscillators": { "gauge": osc_score["gauge"], "signal": osc_score["signal"], "buy": osc_score["buy"], "sell": osc_score["sell"], "neutral": osc_score["neutral"], "buy_weight": osc_score.get("buy_weight", 0.0), "sell_weight": osc_score.get("sell_weight", 0.0), "neutral_weight": osc_score.get("neutral_weight", 0.0), "data": osc_data }, "moving_averages": { "gauge": ma_score["gauge"], "signal": ma_score["signal"], "buy": ma_score["buy"], "sell": ma_score["sell"], "neutral": ma_score["neutral"], "buy_weight": ma_score.get("buy_weight", 0.0), "sell_weight": ma_score.get("sell_weight", 0.0), "neutral_weight": ma_score.get("neutral_weight", 0.0), "golden_cross": ma_score["golden_cross"], "death_cross": ma_score["death_cross"], "data": ma_data }, "ai_gauge": ai_score, "pivot_points": pivots } START_TIME = time.time() async def _background_cleanup(): """Evict expired entries from TTL caches periodically (v6.0: Aggressive).""" while True: await asyncio.sleep(120) # Every 2 minutes try: h = historical_cache.evict_expired() f = forecast_cache.evict_expired() t = ticker_cache.evict_expired() persistent_cache.evict() logger.info("[cleanup] Evicted h=%d f=%d t=%d", h, f, t) except Exception as ex: logger.warning("[cleanup] Error: %s", ex) async def _periodic_health_check(): """Check circuit breakers and log source health every 2 minutes.""" while True: await asyncio.sleep(120) for name, cb in source_breakers.items(): if cb.state == "OPEN": logger.warning("[health] CB OPEN: %s (failures=%d)", name, cb.failures) def _clear_ip_limits(): """Remove stale IP rate-limit entries older than 60 seconds.""" stale_count = clear_stale_ip_limits(IP_LIMITS, now=time.time(), window_seconds=60) if stale_count: logger.debug("[ip-cleanup] Removed %d stale IPs", stale_count) @asynccontextmanager async def lifespan(app: FastAPI): # Startup logic logger.info("Starting AI Forecast Backend v%s...", APP_VERSION) background_tasks: List[asyncio.Task[Any]] = [] def _start_background_task(coro: Any, name: str) -> None: background_tasks.append(asyncio.create_task(coro, name=name)) # Async background tasks _start_background_task(persistent_cache.start_writer(), "persistent-cache-writer") _start_background_task(ws_manager.heartbeat(), "ws-heartbeat") _start_background_task(_background_cleanup(), "cache-cleanup") _start_background_task(_periodic_health_check(), "periodic-health-check") # F-2: Rate limit cleanup task async def _ip_cleanup_loop(): while True: await asyncio.sleep(300) # BUG-P1-07: reduced from 3600 to 300 _clear_ip_limits() _start_background_task(_ip_cleanup_loop(), "ip-limit-cleanup") # Quick source reachability check (non-blocking) _start_background_task(_source_selftest(), "source-selftest") if PRELOAD_TIMESFM and TIMESFM_AVAILABLE: _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 Forecast API...") for task in background_tasks: task.cancel() if background_tasks: await asyncio.gather(*background_tasks, return_exceptions=True) await GlobalHTTPClient.close() logger.info("Graceful shutdown completed.") # ────────────────────────────────────────────────────────────────────────────── # FastAPI App Instance (v6.0) # ────────────────────────────────────────────────────────────────────────────── app = FastAPI( title="AI Forecast API", version=APP_VERSION, description="OHLCV data, Kronos/TimesFM/Chronos forecasts, technical indicators, and real-time WebSocket prices", lifespan=lifespan, ) app.add_middleware( CORSMiddleware, allow_origins=CORS_ALLOW_ORIGINS, allow_credentials=CORS_ALLOW_CREDENTIALS, allow_methods=["*"], allow_headers=["*"], ) # ────────────────────────────────────────────────────────────────────────────── # Market Hours (UTC) # ────────────────────────────────────────────────────────────────────────────── MARKET_SESSIONS = MARKET_SESSIONS_DATA def market_status_now() -> List[Dict[str, Any]]: return compute_market_status_now() # ────────────────────────────────────────────────────────────────────────────── # ────────────────────────────────────────────────────────────────────────────── # TimesFM 2.5 Forecaster # ────────────────────────────────────────────────────────────────────────────── class TimesFMForecaster: """Async wrapper around Google TimesFM 2.5 (200M, PyTorch). model.forecast(horizon, inputs) → (point_forecast, quantile_forecast) point_forecast: np.ndarray (batch, horizon) quantile_forecast: np.ndarray (batch, horizon, 10) index mapping: [mean, q10, q20, q30, q40, q50, q60, q70, q80, q90] Input: OHLCV DataFrame; OHLC4 = (O+H+L+C)/4 computed internally. Output: p10/p50/p90 future OHLC4 values for `horizon` future bars. """ MODEL_HF_ID = "google/timesfm-2.5-200m-pytorch" # Keep context comfortably within 16384 limit; leaves room for any horizon MAX_CONTEXT = 8_192 MAX_HORIZON = 300 # hard cap — matches API limit of horizon<=300 # quantile axis indices in the (batch, horizon, 10) tensor _Q10 = 1 # 10th percentile _Q50 = 5 # 50th percentile (median) _Q90 = 9 # 90th percentile def __init__(self) -> None: self._model: Optional[Any] = None self._loaded = False self._lock: Optional[asyncio.Lock] = None 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() return self._lock async def _get_predict_lock(self) -> asyncio.Lock: if self._predict_lock is None: self._predict_lock = asyncio.Lock() return self._predict_lock @property def is_ready(self) -> bool: return self._loaded @property def device(self) -> str: if self._model is None: return "not_loaded" try: return str(self._model.model.device) except Exception: return "cpu" def _compile(self, horizon: int) -> None: """(Re-)compile with a max_horizon that covers `horizon`.""" import math output_patch = 128 # TimesFM 2.5 output_patch_len # max_horizon must be a multiple of output_patch_len max_h = math.ceil(horizon / output_patch) * output_patch max_h = max(max_h, output_patch) max_h = min(max_h, self.MAX_HORIZON) # context must be multiple of input_patch_len=32 ctx = self.MAX_CONTEXT # already a multiple of 32 self._model.compile( timesfm.ForecastConfig( max_context=ctx, max_horizon=max_h, normalize_inputs=True, use_continuous_quantile_head=True, force_flip_invariance=True, infer_is_positive=True, fix_quantile_crossing=True, ) ) 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: return lock = await self._get_lock() async with lock: if self._loaded: return if not TIMESFM_AVAILABLE: raise HTTPException( status_code=503, detail="TimesFM not installed. Clone https://github.com/google-research/timesfm and pip install -e .[torch]", ) try: logger.info("[TimesFM] Loading %s ...", self.MODEL_HF_ID) model = await asyncio.to_thread( timesfm.TimesFM_2p5_200M_torch.from_pretrained, self.MODEL_HF_ID, ) # Compile once with the default max_horizon self._model = model self._compile(self.MAX_HORIZON) self._loaded = True STARTUP_STATE["timesfm"]["loaded"] = True STARTUP_STATE["timesfm"]["device"] = self.device logger.info("[TimesFM] Ready on %s", self.device) except Exception as ex: STARTUP_STATE["timesfm"]["last_error"] = str(ex) logger.error("[TimesFM] Init failed: %s", ex, exc_info=True) raise HTTPException(status_code=500, detail=f"TimesFM init failed: {ex}") async def forecast( self, df: pd.DataFrame, horizon: int, # legacy kwargs — accepted and ignored for API compatibility x_timestamp=None, y_timestamp=None, sample_count: int = 0, ) -> Dict[str, Any]: await self._lazy_load() assert self._model is not None try: # 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: self._compile(horizon) t0 = time.time() predict_lock = await self._get_predict_lock() async with predict_lock: # TimesFM API: forecast(horizon: int, inputs: list[np.ndarray]) point_forecast, quantile_forecast = await asyncio.to_thread( self._model.forecast, horizon, [ohlc4_ctx], ) elapsed = time.time() - t0 logger.info( "[TimesFM] %.2fs | horizon=%d ctx=%d device=%s", elapsed, horizon, context_len, self.device, ) if torch is not None and torch.cuda.is_available(): torch.cuda.empty_cache() p10, p50, p90, output_validation = self._validate_output_tensors( point_forecast=point_forecast, quantile_forecast=quantile_forecast, horizon=horizon, ) return { "p10": p10, "p50": p50, "p90": p90, "model_name": self.MODEL_HF_ID, "context_length": context_len, "output_horizon": horizon, "input_semantics": { "feature_channels": ["ohlc4"], "active_forecast_channels": ["ohlc4"], "ignored_channels": ["volume"], "price_mode": "ohlc4_single_channel", "base_signal": "ohlc4", "volume_mode": "omitted", "amount_mode": "omitted", "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) raise HTTPException(status_code=500, detail=f"TimesFM prediction failed: {ex}") forecaster = TimesFMProviderForecaster(logger) kronos_forecaster = KronosForecaster(logger) chronos_forecaster = ChronosProviderForecaster(logger) # MODULE: Analysis Engine v2.0 (Relocated and Activated) # Legacy placeholders removed to avoid duplication with logic at line 1884. # ────────────────────────────────────────────────────────────────────────────── # WebSocket connection manager # ────────────────────────────────────────────────────────────────────────────── class ConnectionManager: def __init__(self): self.active: Dict[str, List[WebSocket]] = defaultdict(list) async def connect(self, ws: WebSocket, symbol: str): await ws.accept() self.active[symbol].append(ws) logger.info("[WS] connect: %s (total=%d)", symbol, len(self.active[symbol])) def disconnect(self, ws: WebSocket, symbol: str): self.active[symbol] = [c for c in self.active[symbol] if c is not ws] logger.info("[WS] disconnect: %s (total=%d)", symbol, len(self.active[symbol])) async def broadcast(self, symbol: str, data: Dict[str, Any]): """Reliable broadcast with automatic stale connection cleanup (P0).""" active_list = self.active.get(symbol, []) if not active_list: return dead = [] for ws in list(active_list): try: if ws.client_state == WebSocketState.CONNECTED: await ws.send_json(data) else: dead.append(ws) except Exception: dead.append(ws) for ws in dead: self.disconnect(ws, symbol) async def heartbeat(self): """Keep-alive loop: send ping to all active clients every 20s (BUG-13).""" while True: await asyncio.sleep(20) ping_msg = {"type": "ping", "ts": int(time.time())} for symbol in list(self.active.keys()): for ws in list(self.active[symbol]): try: if ws.client_state == WebSocketState.CONNECTED: await ws.send_json(ping_msg) except Exception: pass ws_manager = ConnectionManager() @app.websocket("/ws/price/{symbol}") async def websocket_price(websocket: WebSocket, symbol: str): """ B-6: Real-time price streaming via WebSocket. Fixes the 404/500 errors in chrome console. """ symbol = _get_canonical_symbol(symbol) if symbol not in SYMBOLS: await websocket.close(code=1008, reason=f"Unknown symbol: {symbol}") return interval = websocket.query_params.get("interval") or "1d" try: await ws_manager.connect(websocket, symbol) while True: try: # 1. State check if websocket.client_state != WebSocketState.CONNECTED: break # 2. Fetch fresh price ticker = await fetch_ticker(symbol, interval=interval) # 3. Final state check before send if websocket.client_state == WebSocketState.CONNECTED: await websocket.send_json({ "type": "price", "symbol": symbol, "price": ticker.get("price"), "change_pct": ticker.get("change_pct", 0), "ts": int(time.time()) }) except (WebSocketDisconnect, RuntimeError): break except Exception as ex: logger.debug("[WS] ticker fetch failed for %s: %s", symbol, ex) interval = 5 if SYMBOLS[symbol].category == "Crypto" else 30 await asyncio.sleep(interval) except Exception as ex: logger.error("[WS] handler error for %s: %s", symbol, ex) finally: ws_manager.disconnect(websocket, symbol) # ────────────────────────────────────────────────────────────────────────────── # Architecture v5: Security & Guardrails (Module F) # ────────────────────────────────────────────────────────────────────────────── IP_LIMITS: Dict[str, List[float]] = defaultdict(list) REQUEST_METRICS = RequestMetricsRegistry() def rate_limit_guard(request: Request): """ F-2: Simple IP-based rate limiting logic. BUG-P1-11: Whitelist health and metrics from rate limiting. """ return check_rate_limit(request, IP_LIMITS) def admin_only(request: Request): """F-1: Admin authentication guard for sensitive endpoints.""" return require_admin_token(request, ADMIN_TOKEN) def _validate_symbol_and_interval(symbol: str, interval: str) -> Tuple[str, str]: return validate_symbol_and_interval( symbol, interval, get_canonical_symbol=_get_canonical_symbol, symbols=SYMBOLS, supported_intervals=SUPPORTED_INTERVALS, ) def _normalize_watchlist_symbols(symbols: List[str]) -> Tuple[List[str], List[str], int]: return normalize_watchlist_symbols( symbols, get_canonical_symbol=_get_canonical_symbol, symbol_registry=SYMBOLS, ) def _validate_cache_target(target: str) -> str: return validate_cache_target(target) # ────────────────────────────────────────────────────────────────────────────── # Pydantic Models # ────────────────────────────────────────────────────────────────────────────── # BUG-P0-01: Removed duplicate v5 app instances and buggy lifespan assignment # Application v6.0 is defined earlier at line 2532 @app.middleware("http") async def rate_limit_middleware(request: Request, call_next): request_id = request.headers.get("X-Request-ID") or make_request_id() timer_start = start_timer() route_path = request.url.path if request.url.path not in RATE_LIMIT_WHITELIST and rate_limit_guard(request): duration_ms = elapsed_ms(timer_start) REQUEST_METRICS.record(route_path, 429, duration_ms) return JSONResponse( status_code=429, content={"detail": "Too many requests", "request_id": request_id}, headers={ "X-Request-ID": request_id, "X-Response-Time-Ms": f"{duration_ms:.2f}", }, ) try: response = await call_next(request) except Exception as ex: duration_ms = elapsed_ms(timer_start) REQUEST_METRICS.record(route_path, 500, duration_ms) logger.error( "[request] %s %s failed | request_id=%s | %.2fms | %s", request.method, route_path, request_id, duration_ms, ex, exc_info=True, ) return JSONResponse( status_code=500, content={"detail": "Internal server error", "request_id": request_id}, headers={ "X-Request-ID": request_id, "X-Response-Time-Ms": f"{duration_ms:.2f}", }, ) duration_ms = elapsed_ms(timer_start) REQUEST_METRICS.record(route_path, response.status_code, duration_ms) response.headers["X-Request-ID"] = request_id response.headers["X-Response-Time-Ms"] = f"{duration_ms:.2f}" return response async def _source_selftest(): """Ping data sources at startup to confirm reachability (1 attempt each).""" tests = build_source_selftest_urls(TWELVEDATA_API_KEY, FINNHUB_API_KEY) async with httpx.AsyncClient(timeout=8) as c: await run_source_selftest( client=c, tests=tests, startup_sources=STARTUP_STATE["sources"], logger=logger, ) async def _warmup_timesfm() -> None: """Load TimesFM in the background so the first forecast is fast.""" try: STARTUP_STATE["timesfm"]["warming"] = True await forecaster._lazy_load() STARTUP_STATE["timesfm"]["warming"] = False logger.info("[TimesFM] Warmup complete — ready on %s", forecaster.device) except Exception as ex: STARTUP_STATE["timesfm"]["warming"] = False 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") async def list_symbols( category: Optional[str] = Query(None, description="Filter by category"), ) -> Dict[str, Any]: global _SYMBOLS_CACHE if _SYMBOLS_CACHE is None: _SYMBOLS_CACHE = build_symbols_catalog(SYMBOLS, INTERVAL_ORDER) if category: return filter_symbols_catalog(_SYMBOLS_CACHE, category) return _SYMBOLS_CACHE async def _fetch_ticker_with_semaphore(symbol: str, semaphore: asyncio.Semaphore) -> Dict[str, Any]: async with semaphore: return await fetch_ticker(symbol) # ── Market Peers ────────────────────────────────────────────────────────────── @app.get("/api/market-peers") async def get_market_peers( symbol: str = Query("BTCUSD"), limit: int = Query(10, ge=1, le=20), ) -> Dict[str, Any]: symbol = _get_canonical_symbol(symbol) candidate_payload = build_market_peer_candidates(SYMBOLS, symbol, limit) peer_configs = candidate_payload.get("peer_configs", []) if not peer_configs: return candidate_payload tasks = [ _fetch_ticker_with_semaphore(peer.symbol, _market_peer_ticker_semaphore) for peer in peer_configs ] results = await asyncio.gather(*tasks, return_exceptions=True) return assemble_market_peer_payload( category=str(candidate_payload["category"]), symbol=str(candidate_payload["symbol"]), peer_configs=peer_configs, peer_results=results, ) # ── Symbol search ───────────────────────────────────────────────────────────── @app.get("/api/search") async def search_symbols( q: str = Query(..., min_length=1, description="Search query"), ) -> Dict[str, Any]: return search_symbols_catalog(SYMBOLS, q) # ── Historical OHLCV ────────────────────────────────────────────────────────── @app.get("/api/historical/{symbol}") async def get_historical( symbol: str, interval: str = Query("1h"), limit: int = Query(500, ge=50, le=2000), refresh: bool = Query(False), ) -> Dict[str, Any]: symbol = _get_canonical_symbol(symbol) if symbol not in SYMBOLS: raise HTTPException(404, f"Unknown symbol: {symbol}") if interval not in SUPPORTED_INTERVALS: raise HTTPException(400, f"Unsupported interval: {interval}") data, source = await fetch_historical( symbol, interval, limit, refresh=refresh, min_context=limit, ) return {"symbol": symbol, "interval": interval, "source": source, "count": len(data), "data": data, "generated_at": int(time.time()), "cache": {"refresh_requested": refresh}} # ── Technical Indicators ────────────────────────────────────────────────────── @app.get("/api/indicators/{symbol}") async def get_indicators( symbol: str, interval: str = Query("1h"), limit: int = Query(300, ge=50, le=1000), refresh: bool = Query(False), ) -> Dict[str, Any]: symbol = _get_canonical_symbol(symbol) if symbol not in SYMBOLS: raise HTTPException(404, f"Unknown symbol: {symbol}") if interval not in SUPPORTED_INTERVALS: raise HTTPException(400, f"Unsupported interval: {interval}") data, source, indicators = await get_indicators_cached( symbol, interval, limit, refresh=refresh, min_context=max(limit, INDICATOR_MIN_CONTEXT), ) return { "symbol": symbol, "interval": interval, "source": source, "candles": len(data), "generated_at": int(time.time()), "cache": {"refresh_requested": refresh}, "indicators": indicators, } # ── Technical Analysis Engine ───────────────────────────────────────────────── @app.get("/api/analysis/{symbol}") async def get_analysis( symbol: str, 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. E-2: Integrated Multi-timeframe (MTF) analysis. """ symbol = _get_canonical_symbol(symbol) if symbol not in SYMBOLS: raise HTTPException(404, f"Unknown symbol: {symbol}") # Fetch main context data, source, indicators = await get_indicators_cached( symbol, interval, 500, refresh=refresh, min_context=ANALYSIS_CONTEXT, ) if len(data) < 50: raise HTTPException(422, "Insufficient data for full analysis") # Fetch higher timeframe context (MTF) htf_interval = "1d" if interval in ["1h", "2h", "4h"] else "1h" if interval in ["5m", "15m"] else None htf_bias = "neutral" if htf_interval: try: _, _, htf_inds = await get_indicators_cached( symbol, htf_interval, 200, refresh=refresh, min_context=HIGHER_TIMEFRAME_CONTEXT, ) htf_bias = "bullish" if htf_inds["trend"].get("above_ema200") else "bearish" except Exception as ex: logger.warning( "[analysis] Higher timeframe context unavailable for %s %s via %s: %s", symbol, interval, htf_interval, ex, ) # Reuse cached forecast when available so /analysis and /forecast stay consistent. forecast_ret = 0.0 confidence = 50.0 forecast_rows: List[Dict[str, Any]] = [] blended: Optional[Dict[str, Any]] = None forecast_reference_price = float(data[-1]["close"]) try: 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: 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: logger.warning("[analysis] Failed to reuse cached forecast for %s %s: %s", symbol, interval, ex) # Build comprehensive analysis analysis = await asyncio.to_thread( _build_trade_analysis, symbol=symbol, interval=interval, data=data, indicators=indicators, forecast_rows=forecast_rows, confidence=confidence, source=source, blended=blended, forecast_reference_price=forecast_reference_price, ) # Inject MTF into analysis analysis["multi_timeframe"] = { "htf_interval": htf_interval, "htf_bias": htf_bias, "alignment": bool((analysis["summary"]["bias"] == "bullish" and htf_bias == "bullish") or (analysis["summary"]["bias"] == "bearish" and htf_bias == "bearish")) } return { "symbol": symbol, "interval": interval, "timestamp": int(time.time()), "analysis": analysis, "verdict": await get_local_ai_verdict(symbol, interval, analysis, forecast_ret, refresh=refresh), "verdict_source": "local_rules", "ai_rules": { "version": ai_rule_registry.version, "loaded_count": len(ai_rule_registry.documents), }, "cache": {"refresh_requested": refresh}, "indicators_snapshot": indicators if include_snapshot else None } # ── Real-time ticker ────────────────────────────────────────────────────────── def _derive_local_verdict(analysis: Dict[str, Any], forecast_pct: float) -> str: summary = analysis.get("summary", {}) technicals = analysis.get("technicals", {}) ai_gauge = analysis.get("ai_gauge", {}) multi_timeframe = analysis.get("multi_timeframe", {}) bias = str(summary.get("bias", "neutral")) tech_gauge = float(technicals.get("gauge") or 50.0) ai_score = float(ai_gauge.get("gauge") or 50.0) aligned = bool(multi_timeframe.get("alignment", False)) confidence = float(ai_gauge.get("confidence_pct") or 0.0) bullish_stack = bias == "bullish" and tech_gauge >= 58.0 and ai_score >= 56.0 bearish_stack = bias == "bearish" and tech_gauge <= 42.0 and ai_score <= 44.0 if bullish_stack and forecast_pct >= 0.6 and aligned and confidence >= 45.0: return "Mua ngay" if bearish_stack and forecast_pct <= -0.6 and aligned and confidence >= 45.0: return "Ban ngay" if bias == "bullish": return "Nen doi mua gia thap hon" if forecast_pct < 0.8 else "Mua ngay" if bias == "bearish": return "Nen doi ban gia cao hon" if forecast_pct > -0.8 else "Ban ngay" return "Nen doi mua gia thap hon" if forecast_pct >= 0 else "Nen doi ban gia cao hon" async def get_local_ai_verdict( symbol: str, interval: str, analysis: Dict[str, Any], forecast_pct: float = 0.0, refresh: bool = False, ) -> str: """Fast local verdict engine with cached deterministic output.""" cache_key = _analysis_verdict_cache_key(symbol, interval, analysis, forecast_pct) if not refresh: cached_verdict = ai_verdict_cache.get(cache_key) if cached_verdict: return str(cached_verdict) persisted_verdict = persistent_cache.get(cache_key) if persisted_verdict: verdict_text = str(persisted_verdict.get("verdict", "")).strip() if verdict_text: ai_verdict_cache.set(cache_key, verdict_text, ttl_seconds=verdict_ttl(interval)) return verdict_text verdict_text = _derive_local_verdict(analysis, forecast_pct) ai_verdict_cache.set(cache_key, verdict_text, ttl_seconds=verdict_ttl(interval)) persistent_cache.set(cache_key, {"verdict": verdict_text}, ttl=verdict_ttl(interval) * 4) return verdict_text @app.get("/api/ticker/{symbol}") async def get_ticker(symbol: str, interval: Optional[str] = None) -> Dict[str, Any]: symbol = _get_canonical_symbol(symbol) if symbol not in SYMBOLS: raise HTTPException(404, f"Unknown symbol: {symbol}") return await fetch_ticker(symbol, interval=interval) # ── Watchlist (batch ticker) ────────────────────────────────────────────────── @app.post("/api/watchlist/tickers") async def get_watchlist_tickers(body: WatchlistRequest) -> Dict[str, Any]: results: Dict[str, Any] = {} valid_symbols, invalid_symbols, duplicate_count = _normalize_watchlist_symbols(body.symbols) if not valid_symbols: raise HTTPException(status_code=422, detail="No valid symbols provided") tasks = [ _fetch_ticker_with_semaphore(s, _watchlist_ticker_semaphore) for s in valid_symbols ] tickers = await asyncio.gather(*tasks, return_exceptions=True) for sym, ticker in zip(valid_symbols, tickers): if isinstance(ticker, Exception): results[sym] = {"error": str(ticker)} else: results[sym] = ticker return { "tickers": results, "count": len(results), "invalid_symbols": invalid_symbols, "duplicate_count": duplicate_count, } 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 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("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 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]]: cached = forecast_cache.get(cache_key) if cached is not None and _forecast_payload_is_current(cached): cached["generated_at"] = int(time.time()) cached["cache"] = {"origin": "memory", "refresh_requested": False} return cached persisted = persistent_cache.get(cache_key) if persisted is not None and _forecast_payload_is_current(persisted): persisted["from_persistent_cache"] = True forecast_cache.set(cache_key, persisted, ttl_seconds=forecast_ttl(interval)) persisted["generated_at"] = int(time.time()) persisted["cache"] = {"origin": "persistent", "refresh_requested": False} return persisted return None async def _prepare_forecast_response_payload( symbol: str, interval: str, horizon: int, refresh: bool, cache_origin: str, ) -> Dict[str, Any]: data_list, source, indicators = await get_indicators_cached( symbol, interval, FORECAST_CONTEXT, refresh=refresh, min_context=FORECAST_CONTEXT, ) 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 not TIMESFM_AVAILABLE: return { "symbol": symbol, "interval": interval, "forecast_rows": [], "error": "TimesFM not installed. Run: pip install timesfm[torch]", "ai_runtime": {"mode": "local_only", "model": "offline"}, "_data_list": data_list, "indicators_snapshot": indicators, "_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_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, "input_semantics": { "feature_channels": ["ohlc4"], "active_forecast_channels": ["ohlc4"], "ignored_channels": ["volume"], "price_mode": "ohlc4_single_channel", "base_signal": "ohlc4", "volume_mode": "omitted", "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, } 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 = 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) ), } ) 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 logger.info( "[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"], analysis_bundle["agreement"], ) forecast_rows: List[Dict[str, Any]] = [ {"time": last_time, "p10": last_ohlc4, "p50": last_ohlc4, "p90": last_ohlc4, "is_actual": True} ] for i in range(horizon): forecast_rows.append({ "time": int(last_time + step * (i + 1)), "p10": round(float(analysis_bundle["p10"][i]), 6), "p50": round(float(analysis_bundle["p50"][i]), 6), "p90": round(float(analysis_bundle["p90"][i]), 6), }) return { "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_rows": forecast_rows, "from_persistent_cache": False, "model": { "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", TimesFMProviderForecaster.MODEL_HF_ID)), "device": forecaster.device, }, } def _build_raw_ohlc4_bundle( model_output: Dict[str, Any], last_ohlc4: float, ) -> Dict[str, Any]: """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_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 band_certainty = math.exp(-avg_band_pct / 4.0) path_consistency = path_metrics["path_consistency"] / 100.0 monotonicity = path_metrics["monotonicity"] / 100.0 move_pct = abs(path_metrics["final_return_pct"]) confidence = ( 28.0 + band_certainty * 34.0 + path_consistency * 18.0 + monotonicity * 12.0 + min(move_pct, 4.0) * 2.0 ) confidence = max(20.0, min(95.0, confidence)) final_sign = 0 if abs(path_metrics["final_return_pct"]) < 0.05 else (1 if path_metrics["final_return_pct"] > 0 else -1) weighted_sign = 0 if abs(path_metrics["weighted_return_pct"]) < 0.05 else (1 if path_metrics["weighted_return_pct"] > 0 else -1) agreement = final_sign == 0 or weighted_sign == 0 or final_sign == weighted_sign return { "p10": raw_p10, "p50": raw_p50, "p90": raw_p90, "model_weight": 1.0, "anchor_weight": 0.0, "agreement": agreement, "scale": 1.0, "confidence": round(confidence, 2), "model_bias_pct": 0.0, "path_metrics": path_metrics, "mode": "raw_timesfm_ohlc4", } async def _finalize_forecast_response_payload(payload: Dict[str, Any]) -> Dict[str, Any]: if payload.get("error"): response = { "symbol": payload["symbol"], "interval": payload["interval"], "forecast": payload.get("forecast_rows", []), "error": payload["error"], "path_checked": None, "display": { "mode": "raw_timesfm_ohlc4_line", "channels": ["ohlc4"], "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"}), } return make_json_compatible(response) analysis_bundle = payload["_analysis_bundle"] analysis = await asyncio.to_thread( _build_trade_analysis, symbol=payload["symbol"], interval=payload["interval"], data=payload["_data_list"], indicators=payload["indicators_snapshot"], forecast_rows=payload["forecast_rows"], 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 = { "symbol": payload["symbol"], "interval": payload["interval"], "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"], "display": { "mode": "raw_timesfm_ohlc4_line", "channels": ["ohlc4"], "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", "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, }, "model_diagnostics": { "raw_last_p10": round(float(payload["_model_output"]["p10"][-1]), 6), "raw_last_p50": round(float(payload["_model_output"]["p50"][-1]), 6), "raw_last_p90": round(float(payload["_model_output"]["p90"][-1]), 6), "display_last_p50": round(float(analysis_bundle["p50"][-1]), 6), }, "indicators_snapshot": payload["indicators_snapshot"], "analysis": analysis, "generated_at": int(time.time()), "cache": { "origin": payload["_cache_origin"], "refresh_requested": False, }, "ai_runtime": payload["ai_runtime"], } 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), 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: raise HTTPException(404, f"Unknown symbol: {symbol}") if interval not in SUPPORTED_INTERVALS: raise HTTPException(400, f"Unsupported interval: {interval}") 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: cached = _load_cached_forecast_response(cache_key, interval) if cached is not None: return cached else: logger.info("[forecast] Refresh requested for %s %s. Bypassing caches.", symbol, interval) cache_origin = "live_refresh" inflight_key = f"{cache_key}:refresh={int(refresh)}" inflight_task = None if refresh else _FORECAST_INFLIGHT.get(inflight_key) if inflight_task is not None: return await inflight_task async def _build_forecast_response() -> Dict[str, Any]: response = await _build_multi_model_forecast_response( symbol=symbol, interval=interval, horizon=horizon, refresh=refresh, cache_origin=cache_origin, model_selection=model_selection, ) forecast_cache.set(cache_key, response, ttl_seconds=forecast_ttl(interval)) persistent_cache.set(cache_key, response, ttl=forecast_ttl(interval) * 4) return response task = asyncio.create_task(_build_forecast_response(), name=f"forecast:{symbol}:{interval}:{horizon}") _FORECAST_INFLIGHT[inflight_key] = task try: return await task finally: if _FORECAST_INFLIGHT.get(inflight_key) is task: _FORECAST_INFLIGHT.pop(inflight_key, None) # ── Market Status ───────────────────────────────────────────────────────────── @app.get("/api/market-status") async def get_market_status() -> Dict[str, Any]: return { "utc_time": datetime.now(timezone.utc).isoformat(), "markets": market_status_now(), } # ── Switch endpoint ──────────────────────────────────────────────────────────── @app.post("/api/switch") async def switch_symbol_interval(body: SwitchRequest) -> Dict[str, Any]: symbol = body.symbol.upper() interval = body.interval if symbol not in SYMBOLS: raise HTTPException(404, f"Unknown symbol: {symbol}") if interval not in SUPPORTED_INTERVALS: raise HTTPException(400, f"Unsupported interval: {interval}") prefix = _cache_prefix(symbol, interval) h_cleared = historical_cache.delete_by_prefix(f"hist_{prefix}") f_cleared = forecast_cache.delete_by_prefix(f"forecast_{prefix}") i_cleared = indicators_cache.delete_by_prefix(f"ind_{prefix}") logger.info( "[switch] %s %s -> hist=%d forecast=%d indicators=%d", symbol, interval, h_cleared, f_cleared, i_cleared, ) return { "status": "cleared", "symbol": symbol, "interval": interval, "cleared": {"historical": h_cleared, "forecast": f_cleared, "indicators": i_cleared}, "message": "Cache cleared. Fetch fresh data now.", } # ── Cache management ────────────────────────────────────────────────────────── @app.delete("/api/cache/{symbol}/{interval}") async def clear_symbol_cache(symbol: str, interval: str, request: Request) -> Dict[str, Any]: admin_only(request) symbol, interval = _validate_symbol_and_interval(symbol, interval) prefix = _cache_prefix(symbol, interval) h_cleared = historical_cache.delete_by_prefix(f"hist_{prefix}") f_cleared = forecast_cache.delete_by_prefix(f"forecast_{prefix}") i_cleared = indicators_cache.delete_by_prefix(f"ind_{prefix}") v_cleared = ai_verdict_cache.delete_by_prefix(f"ai_verdict:{CACHE_VERSION}:{symbol}:{interval}:") return {"symbol": symbol, "interval": interval, "cleared": {"historical": h_cleared, "forecast": f_cleared, "indicators": i_cleared, "ai_verdict": v_cleared}} @app.delete("/api/cache") async def clear_all_cache( request: Request, target: str = Query("all", description="all | historical | forecast | indicators | ticker | ai_verdict"), ) -> Dict[str, Any]: admin_only(request) target = _validate_cache_target(target) h = historical_cache.clear() if target in ("all", "historical") else 0 f = forecast_cache.clear() if target in ("all", "forecast") else 0 i = indicators_cache.clear() if target in ("all", "indicators") else 0 t = ticker_cache.clear() if target in ("all", "ticker") else 0 v = ai_verdict_cache.clear() if target in ("all", "ai_verdict") else 0 return {"cleared": {"historical": h, "forecast": f, "indicators": i, "ticker": t, "ai_verdict": v}} @app.get("/api/cache/stats") async def cache_stats(request: Request) -> Dict[str, Any]: admin_only(request) return { "cache_version": CACHE_VERSION, "historical": historical_cache.stats(), "forecast": forecast_cache.stats(), "indicators": indicators_cache.stats(), "ticker": ticker_cache.stats(), "ai_verdict": ai_verdict_cache.stats(), } # ── Health ──────────────────────────────────────────────────────────────────── @app.get("/api/health") 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 any_model_available else "degraded", "version": APP_VERSION, "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, "python_version": sys.version.split()[0], "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"], } @app.get("/api/ping") async def ping() -> Dict[str, str]: return {"status": "ok", "version": APP_VERSION} # ─── EXTRA API ENDPOINTS (v5.0) ──────────────────────────────────────────────── @app.get("/api/ai/rules") 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).""" admin_only(request) now = time.time() uptime = now - START_TIME # Cache stats h_stats = historical_cache.stats() f_stats = forecast_cache.stats() t_stats = ticker_cache.stats() # CB stats cb_stats = {name: {"state": cb.state, "failures": cb.failures} for name, cb in source_breakers.items()} return { "uptime_seconds": round(uptime, 0), "request_metrics": REQUEST_METRICS.snapshot(), "cache": { "historical": h_stats, "forecast": f_stats, "ticker": t_stats, }, "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"], } @app.get("/api/backtest/{symbol}") async def get_backtest( symbol: str, interval: str = Query("1h"), limit: int = Query(500, ge=100, le=5000), ): """E-1: POC Backtesting: Calculates returns if following AI signals.""" symbol, interval = _validate_symbol_and_interval(symbol, interval) data, _ = await fetch_historical(symbol, interval, limit) if len(data) < 100: raise HTTPException(status_code=400, detail="Insufficient data for backtest") closes = np.array([float(d["close"]) for d in data]) rsi = _rsi(closes, 14) trades, position, entry_price, total_profit = [], 0, 0.0, 0.0 for i in range(50, len(closes)): if position == 0 and rsi[i] < 30: position, entry_price = 1, closes[i] elif position == 1 and rsi[i] > 70: p = (closes[i] - entry_price) / entry_price total_profit += p trades.append({"entry": entry_price, "exit": closes[i], "profit_pct": p * 100}) position = 0 return { "symbol": symbol, "total_trades": len(trades), "total_return_pct": round(total_profit * 100, 2), "trades": trades[-10:] } @app.get("/api/volume-profile/{symbol}") async def get_volume_profile( symbol: str, interval: str = Query("1h"), limit: int = Query(200, ge=20, le=5000), buckets: int = Query(20, ge=5, le=200), ): """E-4: POC Volume Profile.""" symbol, interval = _validate_symbol_and_interval(symbol, interval) data, _ = await fetch_historical(symbol, interval, limit) if not data: return {"buckets": [], "poc": None} highs, lows, vols = np.array([float(d["high"]) for d in data]), np.array([float(d["low"]) for d in data]), np.array([float(d["volume"]) for d in data]) p_min, p_max = np.min(lows), np.max(highs) if p_min == p_max: return {"buckets": [], "poc": None} bin_size, profile = (p_max - p_min) / buckets, defaultdict(float) for i in range(len(data)): b_start, b_end = int((lows[i] - p_min) / bin_size), int((highs[i] - p_min) / bin_size) num_bins = max(1, b_end - b_start + 1) v_per_bin = vols[i] / num_bins for b in range(b_start, min(b_end + 1, buckets)): profile[b] += v_per_bin sorted_profile = [{"price": round(p_min + (b * bin_size) + (bin_size / 2), 6), "volume": round(profile[b], 2)} for b in range(buckets)] return {"buckets": sorted_profile, "poc": max(sorted_profile, key=lambda x: x["volume"])["price"]} # ── Static Frontend ─────────────────────────────────────────────────────────── register_frontend_assets( app, project_root=PROJECT_ROOT, app_version=APP_VERSION, cache_version=CACHE_VERSION, logger=logger, )