""" 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 logging import math import os import re import sys import time from collections import defaultdict from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Any, Dict, List, Optional, Tuple import httpx import numpy as np import pandas as pd import torch import yfinance as yf from contextlib import asynccontextmanager from fastapi import FastAPI, HTTPException, Query, WebSocket, WebSocketDisconnect, Request from fastapi.websockets import WebSocketState from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles from pydantic import BaseModel, ConfigDict # ────────────────────────────────────────────────────────────────────────────── # Architecture v5: Configuration Management (C-1) # ────────────────────────────────────────────────────────────────────────────── class Settings(BaseModel): # API Keys (Hardcoded as requested) twelvedata_api_key: str = "ede46395568d4bdf8195970d19b72880" finnhub_api_key: str = "d7ebfk1r01qu8k17ui80d7ebfk1r01qu8k17ui8g" alpha_vantage_key: str = "" # App Config host: str = "0.0.0.0" port: int = 8000 preload_kronos: bool = True cache_version: str = "v11" # WebSocket Config ws_heartbeat_interval: int = 20 # Circuit Breaker Config cb_failure_threshold: int = 5 cb_recovery_timeout: int = 60 settings = Settings() # ────────────────────────────────────────────────────────────────────────────── # 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) # 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-trading-chart") # ─── PyInstaller / Frozen detection ───────────────────────────────────────── IS_FROZEN = getattr(sys, 'frozen', False) BUNDLE_DIR = getattr(sys, '_MEIPASS', os.path.dirname(os.path.abspath(__file__))) if IS_FROZEN: # In .exe, assets like frontend/ and Kronos-master/ are at the bundle root PROJECT_ROOT = BUNDLE_DIR logger.info("Running in FROZEN mode. BUNDLE_DIR: %s", BUNDLE_DIR) else: # In dev, PROJECT_ROOT is up one level from backend/main.py CURRENT_DIR = os.path.dirname(os.path.abspath(__file__)) PROJECT_ROOT = os.path.dirname(CURRENT_DIR) logger.info("Running in DEV mode. PROJECT_ROOT: %s", PROJECT_ROOT) KRONOS_PATH = os.path.join(PROJECT_ROOT, "Kronos-master") if KRONOS_PATH not in sys.path: sys.path.append(KRONOS_PATH) try: from model.kronos import Kronos, KronosTokenizer, KronosPredictor, calc_time_stamps KRONOS_AVAILABLE = True logger.info("Kronos loaded from: %s", KRONOS_PATH) except Exception as ex: KRONOS_AVAILABLE = False logger.error("Kronos load error: %s", ex) logger.warning("Kronos not found or failed at: %s — forecasting disabled", KRONOS_PATH) PRELOAD_KRONOS = os.getenv("KRONOS_PRELOAD", "1").strip().lower() not in {"0", "false", "no"} STARTUP_STATE: Dict[str, Any] = { "kronos": { "available": KRONOS_AVAILABLE, "preload_enabled": PRELOAD_KRONOS, "warming": False, "loaded": False, "device": "not_loaded", "last_error": None, "path": KRONOS_PATH, }, "sources": {}, } # ────────────────────────────────────────────────────────────────────────────── # Architecture v5: Structured Logging & Persistence (C-4, C-6) # ────────────────────────────────────────────────────────────────────────────── import json import sqlite3 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 class PersistentCache: """SQLite-based persistent cache layer (C-6).""" def __init__(self, db_path: str = "kronos_cache.db"): self.db_path = db_path self._init_db() def _init_db(self): with sqlite3.connect(self.db_path) as conn: conn.execute(""" CREATE TABLE IF NOT EXISTS cache ( key TEXT PRIMARY KEY, payload BLOB, expiry REAL, version TEXT ) """) conn.execute("CREATE INDEX IF NOT EXISTS idx_expiry ON cache(expiry)") def get(self, key: str) -> Optional[Any]: try: with sqlite3.connect(self.db_path) as conn: cur = conn.execute("SELECT payload, expiry FROM cache WHERE key = ?", (key,)) row = cur.fetchone() if row: payload, expiry = row if time.time() < expiry: return json.loads(payload) else: conn.execute("DELETE FROM cache WHERE key = ?", (key,)) except Exception as ex: logger.error("[Persistence] Read error: %s", ex) return None def set(self, key: str, payload: Any, ttl: int): try: with sqlite3.connect(self.db_path) as conn: conn.execute( "INSERT OR REPLACE INTO cache (key, payload, expiry, version) VALUES (?, ?, ?, ?)", (key, json.dumps(payload), time.time() + ttl, settings.cache_version) ) except Exception as ex: logger.error("[Persistence] Write error: %s", ex) def evict(self): with sqlite3.connect(self.db_path) as conn: conn.execute("DELETE FROM cache WHERE expiry < ?", (time.time(),)) persistent_cache = PersistentCache(os.path.join(PROJECT_ROOT, "kronos_v5.db")) # ────────────────────────────────────────────────────────────────────────────── # API Keys (Migrated to Settings) # ────────────────────────────────────────────────────────────────────────────── TWELVEDATA_API_KEY = settings.twelvedata_api_key FINNHUB_API_KEY = settings.finnhub_api_key ALPHA_VANTAGE_KEY = settings.alpha_vantage_key if ALPHA_VANTAGE_KEY == "demo" or not ALPHA_VANTAGE_KEY: logger.warning("[AlphaVantage] No valid key - restricting to sample data only") # Binance, Bybit, CoinGecko, yfinance, FRED — no 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"], "Kim loại": ["yfinance", "twelvedata", "finnhub"], "Năng lượng": ["yfinance", "twelvedata", "finnhub"], "Nông sản": ["yfinance", "twelvedata"], "Nguyên liệu CN": ["yfinance", "twelvedata"], "Chỉ số": ["yfinance", "twelvedata", "finnhub"], "Cổ phiếu Mỹ": ["yfinance", "finnhub", "twelvedata"], "Cổ phiếu VN": ["yfinance"], "Trái phiếu": ["yfinance"], "ETF": ["yfinance", "finnhub"], } # Fallback for unknown categories DEFAULT_SOURCE_PRIORITY: List[str] = ["yfinance", "twelvedata", "finnhub", "binance"] CACHE_VERSION = "v11" CLIP_DEFAULT = 3.0 # ────────────────────────────────────────────────────────────────────────────── # 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) _rate_limiters: Dict[str, TokenBucket] = { "binance": TokenBucket(rate=10.0, capacity=20), "bybit": TokenBucket(rate=2.0, capacity=5), "coingecko": TokenBucket(rate=0.4, capacity=3), "twelvedata": TokenBucket(rate=0.1, capacity=2), # 8/min free = ~0.13/s "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" description: str = "" # ─── 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") -> SymbolConfig: return SymbolConfig(sym, label, label_en, cat, mappings, cg_id, bybit_cat, desc) SYMBOLS: Dict[str, SymbolConfig] = { # ══════════════════════════════════════════════════════════════════════ # 1. KIM LOẠI QUÝ & CÔNG NGHIỆP (Metals) # ══════════════════════════════════════════════════════════════════════ "XAUUSD": _s("XAUUSD","Vàng (XAU/USD)","Gold","Kim loại", {"twelvedata":"XAU/USD","finnhub":"OANDA:XAU_USD","yfinance":"GC=F"}, desc="Vàng giao ngay – chuẩn mực trú ẩn an toàn toàn cầu"), "XAGUSD": _s("XAGUSD","Bạc (XAG/USD)","Silver","Kim loại", {"twelvedata":"XAG/USD","finnhub":"OANDA:XAG_USD","yfinance":"SI=F"}), "XPTUSD": _s("XPTUSD","Bạch kim (XPT/USD)","Platinum","Kim loại", {"twelvedata":"XPT/USD","yfinance":"PL=F"}), "XPDUSD": _s("XPDUSD","Palladium (XPD/USD)","Palladium","Kim loại", {"twelvedata":"XPD/USD","yfinance":"PA=F"}), "COPPER": _s("COPPER","Đồng COMEX (HG)","Copper COMEX","Kim loại", {"yfinance":"HG=F","twelvedata":"HG1"}), "COPPER_LME": _s("COPPER_LME","Đồng LME","Copper LME","Kim loại", {"twelvedata":"XCU/USD","yfinance":"HG=F"}), "ALUMINUM": _s("ALUMINUM","Nhôm (ALI)","Aluminum","Kim loại", {"yfinance":"ALI=F"}), "ALUMINUM_LME": _s("ALUMINUM_LME","Nhôm LME","Aluminum LME","Kim loại", {"twelvedata":"LMAHDS03"}), "NICKEL_LME": _s("NICKEL_LME","Niken LME","Nickel LME","Kim loại", {"twelvedata":"NI1"}), "ZINC_LME": _s("ZINC_LME","Kẽm LME","Zinc LME","Kim loại", {"twelvedata":"ZN1"}), "IRON_ORE": _s("IRON_ORE","Quặng sắt (SGX)","Iron Ore","Kim loại", {"yfinance":"TIO=F"}), "STEEL_HRC": _s("STEEL_HRC","Thép cuộn cán nóng","Steel HRC","Kim loại", {"yfinance":"HRC=F"}), "LITHIUM": _s("LITHIUM","Lithium Futures","Lithium","Kim loại", {"yfinance":"ALB"}), # ALB proxy for lithium # ══════════════════════════════════════════════════════════════════════ # 2. NĂNG LƯỢNG (Energy) # ══════════════════════════════════════════════════════════════════════ "WTI": _s("WTI","Dầu thô WTI (CL)","WTI Crude Oil","Năng lượng", {"yfinance":"CL=F","twelvedata":"CL=F","finnhub":"NYMEX:CL"}, desc="Dầu thô West Texas Intermediate – chuẩn Mỹ"), "BRENT": _s("BRENT","Dầu Brent (BZ)","Brent Crude","Năng lượng", {"yfinance":"BZ=F","twelvedata":"CB1"}, desc="Dầu thô Brent – chuẩn quốc tế"), "NATURAL_GAS": _s("NATURAL_GAS","Khí tự nhiên (NG)","Natural Gas","Năng lượng", {"yfinance":"NG=F","twelvedata":"NG=F"}), "GASOLINE_RBOB": _s("GASOLINE_RBOB","Xăng RBOB","Gasoline RBOB","Năng lượng", {"yfinance":"RB=F"}), "HEATING_OIL": _s("HEATING_OIL","Dầu sưởi (HO)","Heating Oil","Năng lượng", {"yfinance":"HO=F"}), "LNG": _s("LNG","Khí tự nhiên hóa lỏng","LNG","Năng lượng", {"yfinance":"LNG"}), # Cheniere proxy "COAL": _s("COAL","Than Newcastle (XAB)","Coal Newcastle","Năng lượng", {"yfinance":"MTF=F"}), "URANIUM": _s("URANIUM","Uranium (UX)","Uranium","Năng lượng", {"yfinance":"UX=F"}), # ══════════════════════════════════════════════════════════════════════ # 3. NÔNG SẢN (Agricultural) # ══════════════════════════════════════════════════════════════════════ "CORN": _s("CORN","Ngô (ZC)","Corn","Nông sản", {"yfinance":"ZC=F","twelvedata":"ZC=F"}), "WHEAT": _s("WHEAT","Lúa mì (ZW)","Wheat","Nông sản", {"yfinance":"ZW=F","twelvedata":"ZW=F"}), "SOYBEAN": _s("SOYBEAN","Đậu tương (ZS)","Soybean","Nông sản", {"yfinance":"ZS=F","twelvedata":"ZS=F"}), "SOYBEAN_OIL": _s("SOYBEAN_OIL","Dầu đậu tương (ZL)","Soybean Oil","Nông sản", {"yfinance":"ZL=F","twelvedata":"ZL=F"}), "SOYBEAN_MEAL": _s("SOYBEAN_MEAL","Khô đậu tương (ZM)","Soybean Meal","Nông sản", {"yfinance":"ZM=F","twelvedata":"ZM=F"}), "RICE": _s("RICE","Gạo thô (ZR)","Rough Rice","Nông sản", {"yfinance":"ZR=F"}), "OATS": _s("OATS","Yến mạch (ZO)","Oats","Nông sản", {"yfinance":"ZO=F"}), # ══════════════════════════════════════════════════════════════════════ # 4. NGUYÊN LIỆU CÔNG NGHIỆP (Soft Commodities) # ══════════════════════════════════════════════════════════════════════ "COFFEE_ARABICA": _s("COFFEE_ARABICA","Cà phê Arabica (KC)","Coffee Arabica","Nguyên liệu CN", {"yfinance":"KC=F"}), "COFFEE_ROBUSTA": _s("COFFEE_ROBUSTA","Cà phê Robusta (RC)","Coffee Robusta","Nguyên liệu CN", {"yfinance":"RC=F"}), "COCOA": _s("COCOA","Ca cao (CC)","Cocoa","Nguyên liệu CN", {"yfinance":"CC=F"}), "SUGAR_11": _s("SUGAR_11","Đường 11 (SB)","Sugar No.11","Nguyên liệu CN", {"yfinance":"SB=F"}), "WHITE_SUGAR": _s("WHITE_SUGAR","Đường trắng (LSW)","White Sugar","Nguyên liệu CN", {"yfinance":"LSW=F"}), "COTTON": _s("COTTON","Bông (CT)","Cotton","Nguyên liệu CN", {"yfinance":"CT=F"}), "ORANGE_JUICE": _s("ORANGE_JUICE","Nước cam (OJ)","Orange Juice","Nguyên liệu CN", {"yfinance":"OJ=F"}), "LUMBER": _s("LUMBER","Gỗ xẻ (LBS)","Lumber","Nguyên liệu CN", {"yfinance":"LBS=F"}), "RUBBER_RSS3": _s("RUBBER_RSS3","Cao su RSS3","Rubber RSS3","Nguyên liệu CN", {"yfinance":"JRU=F"}), "PALM_OIL": _s("PALM_OIL","Dầu cọ thô (FCPO)","Palm Oil","Nguyên liệu CN", {"yfinance":"FCPO=F"}), # ══════════════════════════════════════════════════════════════════════ # 5. CRYPTO — Top 30 by market cap (nguồn: Binance + Bybit + CoinGecko) # ══════════════════════════════════════════════════════════════════════ "BTCUSD": _s("BTCUSD","Bitcoin (BTC)","Bitcoin","Crypto", {"binance":"BTCUSDT","bybit":"BTCUSDT","coingecko":"bitcoin", "finnhub":"BINANCE:BTCUSDT","twelvedata":"BTC/USD","yfinance":"BTC-USD"}, cg_id="bitcoin", desc="Đồng tiền mã hóa đầu tiên và lớn nhất thế giới"), "ETHUSD": _s("ETHUSD","Ethereum (ETH)","Ethereum","Crypto", {"binance":"ETHUSDT","bybit":"ETHUSDT","coingecko":"ethereum", "finnhub":"BINANCE:ETHUSDT","twelvedata":"ETH/USD","yfinance":"ETH-USD"}, cg_id="ethereum"), "BNBUSD": _s("BNBUSD","BNB (BNB)","BNB","Crypto", {"binance":"BNBUSDT","bybit":"BNBUSDT","coingecko":"binancecoin", "finnhub":"BINANCE:BNBUSDT","yfinance":"BNB-USD"}, cg_id="binancecoin"), "SOLUSD": _s("SOLUSD","Solana (SOL)","Solana","Crypto", {"binance":"SOLUSDT","bybit":"SOLUSDT","coingecko":"solana", "finnhub":"BINANCE:SOLUSDT","yfinance":"SOL-USD"}, cg_id="solana"), "XRPUSD": _s("XRPUSD","Ripple (XRP)","XRP","Crypto", {"binance":"XRPUSDT","bybit":"XRPUSDT","coingecko":"ripple", "finnhub":"BINANCE:XRPUSDT","yfinance":"XRP-USD"}, cg_id="ripple"), "ADAUSD": _s("ADAUSD","Cardano (ADA)","Cardano","Crypto", {"binance":"ADAUSDT","bybit":"ADAUSDT","coingecko":"cardano","yfinance":"ADA-USD"}, cg_id="cardano"), "AVAXUSD": _s("AVAXUSD","Avalanche (AVAX)","Avalanche","Crypto", {"binance":"AVAXUSDT","bybit":"AVAXUSDT","coingecko":"avalanche-2","yfinance":"AVAX-USD"}, cg_id="avalanche-2"), "DOTUSD": _s("DOTUSD","Polkadot (DOT)","Polkadot","Crypto", {"binance":"DOTUSDT","bybit":"DOTUSDT","coingecko":"polkadot","yfinance":"DOT-USD"}, cg_id="polkadot"), "MATICUSD": _s("MATICUSD","Polygon (MATIC/POL)","Polygon","Crypto", {"binance":"MATICUSDT","bybit":"MATICUSDT","coingecko":"matic-network","yfinance":"MATIC-USD"}, cg_id="matic-network"), "LINKUSD": _s("LINKUSD","Chainlink (LINK)","Chainlink","Crypto", {"binance":"LINKUSDT","bybit":"LINKUSDT","coingecko":"chainlink","yfinance":"LINK-USD"}, cg_id="chainlink"), "LTCUSD": _s("LTCUSD","Litecoin (LTC)","Litecoin","Crypto", {"binance":"LTCUSDT","bybit":"LTCUSDT","coingecko":"litecoin","yfinance":"LTC-USD"}, cg_id="litecoin"), "UNIUSD": _s("UNIUSD","Uniswap (UNI)","Uniswap","Crypto", {"binance":"UNIUSDT","bybit":"UNIUSDT","coingecko":"uniswap","yfinance":"UNI-USD"}, cg_id="uniswap"), "ATOMUSD": _s("ATOMUSD","Cosmos (ATOM)","Cosmos","Crypto", {"binance":"ATOMUSDT","bybit":"ATOMUSDT","coingecko":"cosmos","yfinance":"ATOM-USD"}, cg_id="cosmos"), "NEARUSD": _s("NEARUSD","NEAR Protocol","NEAR","Crypto", {"binance":"NEARUSDT","bybit":"NEARUSDT","coingecko":"near","yfinance":"NEAR-USD"}, cg_id="near"), "APTUSD": _s("APTUSD","Aptos (APT)","Aptos","Crypto", {"binance":"APTUSDT","bybit":"APTUSDT","coingecko":"aptos","yfinance":"APT-USD"}, cg_id="aptos"), "SUIUSD": _s("SUIUSD","Sui (SUI)","Sui","Crypto", {"binance":"SUIUSDT","bybit":"SUIUSDT","coingecko":"sui","yfinance":"SUI20947-USD"}, cg_id="sui"), "ARBUSD": _s("ARBUSD","Arbitrum (ARB)","Arbitrum","Crypto", {"binance":"ARBUSDT","bybit":"ARBUSDT","coingecko":"arbitrum","yfinance":"ARB-USD"}, cg_id="arbitrum"), "OPUSD": _s("OPUSD","Optimism (OP)","Optimism","Crypto", {"binance":"OPUSDT","bybit":"OPUSDT","coingecko":"optimism","yfinance":"OP-USD"}, cg_id="optimism"), "INJUSD": _s("INJUSD","Injective (INJ)","Injective","Crypto", {"binance":"INJUSDT","bybit":"INJUSDT","coingecko":"injective-protocol"}, cg_id="injective-protocol"), "TONUSD": _s("TONUSD","Toncoin (TON)","Toncoin","Crypto", {"binance":"TONUSDT","bybit":"TONUSDT","coingecko":"the-open-network"}, cg_id="the-open-network"), "TRXUSD": _s("TRXUSD","Tron (TRX)","Tron","Crypto", {"binance":"TRXUSDT","bybit":"TRXUSDT","coingecko":"tron","yfinance":"TRX-USD"}, cg_id="tron"), "XMRUSD": _s("XMRUSD","Monero (XMR)","Monero","Crypto", {"bybit":"XMRUSDT","coingecko":"monero","yfinance":"XMR-USD"}, cg_id="monero"), "DOGEUSD": _s("DOGEUSD","Dogecoin (DOGE)","Dogecoin","Crypto", {"binance":"DOGEUSDT","bybit":"DOGEUSDT","coingecko":"dogecoin","yfinance":"DOGE-USD"}, cg_id="dogecoin"), "SHIBAUSD": _s("SHIBAUSD","Shiba Inu (SHIB)","Shiba Inu","Crypto", {"binance":"SHIBUSDT","coingecko":"shiba-inu","yfinance":"SHIB-USD"}, cg_id="shiba-inu"), "PEPE": _s("PEPE","PEPE Coin","PEPE","Crypto", {"binance":"PEPEUSDT","coingecko":"pepe","yfinance":"PEPE24478-USD"}, cg_id="pepe"), # ══════════════════════════════════════════════════════════════════════ # 6. CẶP TIỀN (Forex) # ══════════════════════════════════════════════════════════════════════ "DXY": _s("DXY","Chỉ số USD (DXY)","USD Index","Cặp tiền", {"yfinance":"DX-Y.NYB"}, desc="Rổ 6 đồng tiền chính so với USD"), "EURUSD": _s("EURUSD","EUR/USD","EUR/USD","Cặp tiền", {"twelvedata":"EUR/USD","finnhub":"OANDA:EUR_USD","yfinance":"EURUSD=X"}), "GBPUSD": _s("GBPUSD","GBP/USD","GBP/USD","Cặp tiền", {"twelvedata":"GBP/USD","finnhub":"OANDA:GBP_USD","yfinance":"GBPUSD=X"}), "USDJPY": _s("USDJPY","USD/JPY","USD/JPY","Cặp tiền", {"twelvedata":"USD/JPY","finnhub":"OANDA:USD_JPY","yfinance":"JPY=X"}), "USDCHF": _s("USDCHF","USD/CHF","USD/CHF","Cặp tiền", {"twelvedata":"USD/CHF","finnhub":"OANDA:USD_CHF","yfinance":"CHF=X"}), "AUDUSD": _s("AUDUSD","AUD/USD","AUD/USD","Cặp tiền", {"twelvedata":"AUD/USD","finnhub":"OANDA:AUD_USD","yfinance":"AUDUSD=X"}), "USDCAD": _s("USDCAD","USD/CAD","USD/CAD","Cặp tiền", {"twelvedata":"USD/CAD","finnhub":"OANDA:USD_CAD","yfinance":"CAD=X"}), "NZDUSD": _s("NZDUSD","NZD/USD","NZD/USD","Cặp tiền", {"twelvedata":"NZD/USD","finnhub":"OANDA:NZD_USD","yfinance":"NZDUSD=X"}), "GBPJPY": _s("GBPJPY","GBP/JPY","GBP/JPY","Cặp tiền", {"twelvedata":"GBP/JPY","finnhub":"OANDA:GBP_JPY","yfinance":"GBPJPY=X"}), "EURJPY": _s("EURJPY","EUR/JPY","EUR/JPY","Cặp tiền", {"twelvedata":"EUR/JPY","finnhub":"OANDA:EUR_JPY","yfinance":"EURJPY=X"}), "EURGBP": _s("EURGBP","EUR/GBP","EUR/GBP","Cặp tiền", {"twelvedata":"EUR/GBP","finnhub":"OANDA:EUR_GBP","yfinance":"EURGBP=X"}), "CADCHF": _s("CADCHF","CAD/CHF","CAD/CHF","Cặp tiền", {"twelvedata":"CAD/CHF","finnhub":"OANDA:CAD_CHF","yfinance":"CADCHF=X"}), "AUDNZD": _s("AUDNZD","AUD/NZD","AUD/NZD","Cặp tiền", {"twelvedata":"AUD/NZD","finnhub":"OANDA:AUD_NZD","yfinance":"AUDNZD=X"}), "USDVND": _s("USDVND","USD/VND","USD/VND","Cặp tiền", {"yfinance":"VND=X"}, desc="Tỷ giá đô la Mỹ – đồng Việt Nam"), "USDHKD": _s("USDHKD","USD/HKD","USD/HKD","Cặp tiền", {"twelvedata":"USD/HKD","yfinance":"HKD=X"}), "USDSGD": _s("USDSGD","USD/SGD","USD/SGD","Cặp tiền", {"twelvedata":"USD/SGD","yfinance":"SGD=X"}), "USDCNY": _s("USDCNY","USD/CNY (Offshore)","USD/CNH","Cặp tiền", {"twelvedata":"USD/CNH","yfinance":"CNY=X"}), "USDINR": _s("USDINR","USD/INR","USD/INR","Cặp tiền", {"yfinance":"INR=X"}), "USDBRL": _s("USDBRL","USD/BRL","USD/BRL","Cặp tiền", {"yfinance":"BRL=X"}), # ══════════════════════════════════════════════════════════════════════ # 7. CHỈ SỐ CHỨNG KHOÁN (Stock Indices) # ══════════════════════════════════════════════════════════════════════ "SP500": _s("SP500","S&P 500","S&P 500","Chỉ số", {"yfinance":"^GSPC","twelvedata":"SPY","finnhub":"INDEX:SPX"}, desc="500 cổ phiếu vốn hóa lớn nhất Mỹ"), "NASDAQ": _s("NASDAQ","Nasdaq Composite","Nasdaq Composite","Chỉ số", {"yfinance":"^IXIC","twelvedata":"QQQ"}), "NASDAQ100":_s("NASDAQ100","Nasdaq 100 (NDX)","Nasdaq 100","Chỉ số", {"yfinance":"^NDX","twelvedata":"QQQ"}), "DOW30": _s("DOW30","Dow Jones 30 (DJIA)","Dow Jones","Chỉ số", {"yfinance":"^DJI","twelvedata":"DIA"}), "RUSSELL2000":_s("RUSSELL2000","Russell 2000","Russell 2000","Chỉ số", {"yfinance":"^RUT","twelvedata":"IWM"}), "VIX": _s("VIX","VIX (Fear Index)","VIX","Chỉ số", {"yfinance":"^VIX"}, desc="Chỉ số biến động CBOE – thước đo nỗi sợ thị trường"), "VNINDEX": _s("VNINDEX","VN-Index","VN-Index","Chỉ số", {"yfinance":"^VNINDEX"}, desc="Chỉ số thị trường chứng khoán Việt Nam – HOSE"), "HNX30": _s("HNX30","HNX 30","HNX 30","Chỉ số", {"yfinance":"^HNX30"}, desc="30 cổ phiếu lớn nhất sàn Hà Nội"), "UK100": _s("UK100","FTSE 100 (UK)","FTSE 100","Chỉ số", {"yfinance":"^FTSE","finnhub":"INDEX:UKX"}), "DAX40": _s("DAX40","DAX 40 (Đức)","DAX 40","Chỉ số", {"yfinance":"^GDAXI","finnhub":"INDEX:DAX"}), "EU50": _s("EU50","Euro Stoxx 50","Euro Stoxx 50","Chỉ số", {"yfinance":"^STOXX50E"}), "CAC40": _s("CAC40","CAC 40 (Pháp)","CAC 40","Chỉ số", {"yfinance":"^FCHI"}), "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"}), "SSE50": _s("SSE50","SSE 50 (Thượng Hải)","SSE 50","Chỉ số", {"yfinance":"000016.SS"}), "CSI300": _s("CSI300","CSI 300 (Trung Quốc)","CSI 300","Chỉ số", {"yfinance":"000300.SS"}), "ASX200": _s("ASX200","ASX 200 (Úc)","ASX 200","Chỉ số", {"yfinance":"^AXJO"}), "SENSEX": _s("SENSEX","BSE Sensex (Ấn Độ)","Sensex","Chỉ số", {"yfinance":"^BSESN"}), "KOSPI": _s("KOSPI","KOSPI (Hàn Quốc)","KOSPI","Chỉ số", {"yfinance":"^KS11"}), "SGX": _s("SGX","Straits Times (Singapore)","STI","Chỉ số", {"yfinance":"^STI"}), # ══════════════════════════════════════════════════════════════════════ # 8. CỔ PHIẾU MỸ (US Stocks) # ══════════════════════════════════════════════════════════════════════ "AAPL": _s("AAPL","Apple Inc.","Apple","Cổ phiếu Mỹ",{"yfinance":"AAPL","finnhub":"AAPL","twelvedata":"AAPL"}), "MSFT": _s("MSFT","Microsoft Corp.","Microsoft","Cổ phiếu Mỹ",{"yfinance":"MSFT","finnhub":"MSFT","twelvedata":"MSFT"}), "NVDA": _s("NVDA","Nvidia Corp.","Nvidia","Cổ phiếu Mỹ",{"yfinance":"NVDA","finnhub":"NVDA","twelvedata":"NVDA"}), "GOOGL": _s("GOOGL","Alphabet (Google)","Google","Cổ phiếu Mỹ",{"yfinance":"GOOGL","finnhub":"GOOGL","twelvedata":"GOOGL"}), "AMZN": _s("AMZN","Amazon.com","Amazon","Cổ phiếu Mỹ",{"yfinance":"AMZN","finnhub":"AMZN","twelvedata":"AMZN"}), "META": _s("META","Meta Platforms","Meta","Cổ phiếu Mỹ",{"yfinance":"META","finnhub":"META","twelvedata":"META"}), "TSLA": _s("TSLA","Tesla Inc.","Tesla","Cổ phiếu Mỹ",{"yfinance":"TSLA","finnhub":"TSLA","twelvedata":"TSLA"}), "AVGO": _s("AVGO","Broadcom Inc.","Broadcom","Cổ phiếu Mỹ",{"yfinance":"AVGO","finnhub":"AVGO"}), "JPM": _s("JPM","JPMorgan Chase","JPMorgan","Cổ phiếu Mỹ",{"yfinance":"JPM","finnhub":"JPM"}), "V": _s("V","Visa Inc.","Visa","Cổ phiếu Mỹ",{"yfinance":"V","finnhub":"V"}), "MA": _s("MA","Mastercard","Mastercard","Cổ phiếu Mỹ",{"yfinance":"MA","finnhub":"MA"}), "XOM": _s("XOM","ExxonMobil","ExxonMobil","Cổ phiếu Mỹ",{"yfinance":"XOM","finnhub":"XOM"}), "WMT": _s("WMT","Walmart","Walmart","Cổ phiếu Mỹ",{"yfinance":"WMT","finnhub":"WMT"}), "BAC": _s("BAC","Bank of America","BofA","Cổ phiếu Mỹ",{"yfinance":"BAC","finnhub":"BAC"}), "GS": _s("GS","Goldman Sachs","Goldman Sachs","Cổ phiếu Mỹ",{"yfinance":"GS","finnhub":"GS"}), "AMD": _s("AMD","AMD (Advanced Micro Devices)","AMD","Cổ phiếu Mỹ",{"yfinance":"AMD","finnhub":"AMD"}), "INTC": _s("INTC","Intel Corp.","Intel","Cổ phiếu Mỹ",{"yfinance":"INTC","finnhub":"INTC"}), "NFLX": _s("NFLX","Netflix","Netflix","Cổ phiếu Mỹ",{"yfinance":"NFLX","finnhub":"NFLX"}), "DIS": _s("DIS","Walt Disney","Disney","Cổ phiếu Mỹ",{"yfinance":"DIS","finnhub":"DIS"}), "COIN": _s("COIN","Coinbase Global","Coinbase","Cổ phiếu Mỹ",{"yfinance":"COIN","finnhub":"COIN"}), "MSTR": _s("MSTR","MicroStrategy (MSTR)","MicroStrategy","Cổ phiếu Mỹ",{"yfinance":"MSTR","finnhub":"MSTR"}), # ══════════════════════════════════════════════════════════════════════ # 9. CỔ PHIẾU VIỆT NAM (Vietnam Stocks — HOSE/HNX via yfinance .VN) # ══════════════════════════════════════════════════════════════════════ "VCB": _s("VCB","Vietcombank (VCB)","Vietcombank","Cổ phiếu VN",{"yfinance":"VCB.VN"}), "BID": _s("BID","BIDV (BID)","BIDV","Cổ phiếu VN",{"yfinance":"BID.VN"}), "CTG": _s("CTG","VietinBank (CTG)","VietinBank","Cổ phiếu VN",{"yfinance":"CTG.VN"}), "TCB": _s("TCB","Techcombank (TCB)","Techcombank","Cổ phiếu VN",{"yfinance":"TCB.VN"}), "MBB": _s("MBB","MB Bank (MBB)","MB Bank","Cổ phiếu VN",{"yfinance":"MBB.VN"}), "VPB": _s("VPB","VPBank (VPB)","VPBank","Cổ phiếu VN",{"yfinance":"VPB.VN"}), "ACB": _s("ACB","ACB Bank (ACB)","ACB","Cổ phiếu VN",{"yfinance":"ACB.VN"}), "HDB": _s("HDB","HDBank (HDB)","HDBank","Cổ phiếu VN",{"yfinance":"HDB.VN"}), "SHB": _s("SHB","SHB Bank (SHB)","SHB","Cổ phiếu VN",{"yfinance":"SHB.VN"}), "STB": _s("STB","Sacombank (STB)","Sacombank","Cổ phiếu VN",{"yfinance":"STB.VN"}), "FPT": _s("FPT","FPT Corp (FPT)","FPT","Cổ phiếu VN",{"yfinance":"FPT.VN"}, desc="Tập đoàn công nghệ FPT"), "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"}), "VNM": _s("VNM","Vinamilk (VNM)","Vinamilk","Cổ phiếu VN",{"yfinance":"VNM.VN"}), "MSN": _s("MSN","Masan Group (MSN)","Masan","Cổ phiếu VN",{"yfinance":"MSN.VN"}), "GAS": _s("GAS","PV Gas (GAS)","PV Gas","Cổ phiếu VN",{"yfinance":"GAS.VN"}), "SAB": _s("SAB","Sabeco (SAB)","Sabeco","Cổ phiếu VN",{"yfinance":"SAB.VN"}), "HPG": _s("HPG","Hòa Phát (HPG)","Hoa Phat Steel","Cổ phiếu VN",{"yfinance":"HPG.VN"}), "MWG": _s("MWG","Thế Giới Di Động (MWG)","MWG","Cổ phiếu VN",{"yfinance":"MWG.VN"}), "DGC": _s("DGC","Hóa chất Đức Giang (DGC)","DGC","Cổ phiếu VN",{"yfinance":"DGC.VN"}), "PLX": _s("PLX","Petrolimex (PLX)","Petrolimex","Cổ phiếu VN",{"yfinance":"PLX.VN"}), "POW": _s("POW","PV Power (POW)","PV Power","Cổ phiếu VN",{"yfinance":"POW.VN"}), "SSI": _s("SSI","SSI Securities (SSI)","SSI","Cổ phiếu VN",{"yfinance":"SSI.VN"}), "VND": _s("VND","VNDirect (VND)","VNDirect","Cổ phiếu VN",{"yfinance":"VND.VN"}), "HCM": _s("HCM","HCMC Securities (HCM)","HCM","Cổ phiếu VN",{"yfinance":"HCM.VN"}), # ══════════════════════════════════════════════════════════════════════ # 10. TRÁI PHIẾU & LÃI SUẤT (Bonds & Rates) # ══════════════════════════════════════════════════════════════════════ "US10Y": _s("US10Y","Trái phiếu Mỹ 10 năm","US 10Y Treasury","Trái phiếu", {"yfinance":"^TNX"}, desc="Lợi suất trái phiếu kho bạc Mỹ kỳ hạn 10 năm – chuẩn lãi suất toàn cầu"), "US02Y": _s("US02Y","Trái phiếu Mỹ 2 năm","US 2Y Treasury","Trái phiếu", {"yfinance":"^IRX"}), "US30Y": _s("US30Y","Trái phiếu Mỹ 30 năm","US 30Y Treasury","Trái phiếu", {"yfinance":"^TYX"}), "DE10Y": _s("DE10Y","Bund Đức 10 năm","DE 10Y Bund","Trái phiếu", {"yfinance":"^DE10YT=RR"}), "JP10Y": _s("JP10Y","JGB Nhật 10 năm","JP 10Y JGB","Trái phiếu", {"yfinance":"^JP10YT=RR"}), } # ────────────────────────────────────────────────────────────────────────────── # TTL Cache (unchanged from v3, with improved stats) # ────────────────────────────────────────────────────────────────────────────── 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, cfg in SYMBOLS.items(): if s == cfg.mappings.get("binance") or s == cfg.mappings.get("coingecko") or s == cfg.mappings.get("twelvedata"): return reg_id return s class TTLCache: def __init__(self) -> None: self._store: Dict[str, Tuple[float, Any]] = {} def get(self, key: str) -> Optional[Any]: entry = self._store.get(key) if entry is None: return None exp, payload = entry if time.time() > exp: self._store.pop(key, None) return None return payload def set(self, key: str, payload: Any, ttl_seconds: int) -> None: self._store[key] = (time.time() + ttl_seconds, payload) def delete(self, key: str) -> bool: return self._store.pop(key, None) is not None def delete_by_prefix(self, prefix: str) -> int: victims = [k for k in list(self._store) if k.startswith(prefix)] for k in victims: self._store.pop(k, None) return len(victims) def clear(self) -> int: n = len(self._store) self._store.clear() return n def evict_expired(self) -> int: """Explicitly remove all expired entries from memory (BUG-05).""" now = time.time() expired = [k for k, (exp, _) in self._store.items() if now > exp] for k in expired: self._store.pop(k, None) return len(expired) def stats(self) -> Dict[str, Any]: now = time.time() alive = sum(1 for exp, _ in self._store.values() if now <= exp) return {"total_keys": len(self._store), "alive_keys": alive, "expired_keys": len(self._store) - alive} historical_cache = TTLCache() forecast_cache = TTLCache() ticker_cache = TTLCache() 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 300 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 # ────────────────────────────────────────────────────────────────────────────── # 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 # Instead of hardcoded values, use a permissive baseline and dynamic check in v5 vol_threshold = threshold 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 # Permissive filter: only discard if range is truly zero or extremely suspicious (< 0.00001) # This prevents discarding JPY pairs or US bonds (BUG-07) if o > 0 and (h - l) / o < 1e-6: 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") endpoint_symbol = SYMBOLS[symbol].mappings["binance"] params = { "symbol": endpoint_symbol, "interval": BINANCE_INTERVAL_MAP.get(interval, interval), "limit": min(max(limit, 30), 1000), } logger.info("[Binance] %s %s", symbol, interval) async def _fetch(): cb = source_breakers["binance"] if not cb.allow_request(): raise HTTPException(status_code=503, detail="Binance circuit is OPEN") try: client = await GlobalHTTPClient.get_client() resp = await client.get("https://api.binance.com/api/v3/klines", params=params) if resp.status_code == 429: cb.record_failure() raise HTTPException(status_code=429, detail="Binance rate limit") if resp.status_code >= 500: cb.record_failure() raise HTTPException(status_code=resp.status_code, detail="Binance server error") cb.record_success() return resp.json() except Exception as ex: cb.record_failure() raise ex payload = await _retry(_fetch) 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:] 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 = SYMBOLS[symbol].mappings["bybit"] 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), 200), } logger.info("[Bybit] %s %s (cat=%s)", symbol, interval, bybit_cat) async def _fetch(): cb = source_breakers["bybit"] if not cb.allow_request(): raise HTTPException(status_code=503, detail="Bybit circuit is OPEN") try: client = await GlobalHTTPClient.get_client() resp = await client.get("https://api.bybit.com/v5/market/kline", params=params) if resp.status_code == 429: cb.record_failure() raise HTTPException(status_code=429, detail="Bybit rate limit") if resp.status_code >= 500: cb.record_failure() raise HTTPException(status_code=resp.status_code, detail="Bybit server error") cb.record_success() return resp.json() except Exception as ex: cb.record_failure() raise ex data = await _retry(_fetch) if data.get("retCode", -1) != 0: raise RuntimeError(f"Bybit: {data}") data = await _retry(_fetch) rows = data.get("result", {}).get("list", []) # Bybit returns: [startTime, open, high, low, close, volume, turnover] 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() # Bybit: newest first return _normalize_ohlcv(parsed, interval)[-limit:] 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 = 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 = SYMBOLS[symbol].mappings["twelvedata"] params = { "symbol": endpoint_symbol, "interval": TWELVE_INTERVAL_MAP[interval], "outputsize": min(max(limit, 30), 5000), "apikey": TWELVEDATA_API_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() raise HTTPException(status_code=429, detail="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 "code" in payload and payload.get("code") in [429, 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]]: await _rate_limit("finnhub") endpoint_symbol = SYMBOLS[symbol].mappings["finnhub"] resolution = FINNHUB_RESOLUTION_MAP[interval] now = int(time.time()) # BUG-14: Limit lookback to prevent overflow or errors on long intervals (1w) # Cap lookback at 5 years (approx 157,788,000 seconds) max_lookback = 5 * 365 * 86400 lookback = min(int(STEP_SECONDS[interval] * max(limit * 3, 150)), max_lookback) if endpoint_symbol.startswith("OANDA:"): url = "https://finnhub.io/api/v1/forex/candle" elif endpoint_symbol.startswith("BINANCE:"): url = "https://finnhub.io/api/v1/crypto/candle" elif endpoint_symbol.startswith("INDEX:"): url = "https://finnhub.io/api/v1/index/historical" else: url = "https://finnhub.io/api/v1/stock/candle" params = { "symbol": endpoint_symbol, "resolution": resolution, "from": now - lookback, "to": now, "token": FINNHUB_API_KEY, } logger.info("[Finnhub] %s %s", symbol, interval) async def _fetch(): async with httpx.AsyncClient(timeout=20) as client: resp = await client.get(url, params=params) if resp.status_code == 429: raise HTTPException(status_code=429, detail="Finnhub rate limit") return resp.json() payload = await _retry(_fetch) if payload.get("s") != "ok": raise RuntimeError(f"Finnhub: {payload}") timestamps = payload.get("t", []) n = len(timestamps) volumes = payload.get("v") or [0.0] * n parsed = [ {"time": timestamps[i], "open": payload["o"][i], "high": payload["h"][i], "low": payload["l"][i], "close": payload["c"][i], "volume": volumes[i] if i < len(volumes) else 0.0} for i in range(n) ] return _normalize_ohlcv(parsed, interval)[-limit:] 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 = SYMBOLS[symbol].mappings["yfinance"] yf_interval = YF_INTERVAL_MAP[interval] period = YF_PERIOD_MAP[interval] logger.info("[yfinance] %s %s", symbol, interval) def _download() -> pd.DataFrame: return yf.download(tickers=ticker, interval=yf_interval, period=period, 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:] def _get_source_priority(symbol: str) -> List[str]: cfg = SYMBOLS[symbol] priority = CATEGORY_SOURCE_PRIORITY.get(cfg.category, DEFAULT_SOURCE_PRIORITY) return [s for s in priority if s in cfg.mappings] async def fetch_historical( symbol: str, interval: str, limit: int ) -> Tuple[List[Dict[str, Any]], str]: prefix = _cache_prefix(symbol, interval) key = f"hist_{prefix}{limit}" cached = historical_cache.get(key) if cached is not None: return cached, "cache" priority = _get_source_priority(symbol) errors: List[str] = [] for source in priority: try: if source == "binance": data = await fetch_binance(symbol, interval, limit) elif source == "bybit": data = await fetch_bybit(symbol, interval, limit) elif source == "coingecko": data = await fetch_coingecko(symbol, interval, limit) elif source == "twelvedata": data = await fetch_twelvedata(symbol, interval, limit) elif source == "finnhub": data = await fetch_finnhub(symbol, interval, limit) elif source == "yfinance": data = await fetch_yfinance(symbol, interval, limit) else: continue if len(data) >= 20: historical_cache.set(key, data, ttl_seconds=interval_ttl(interval)) return data, source errors.append(f"{source}: insufficient ({len(data)} candles)") except HTTPException: raise 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}, ) # ────────────────────────────────────────────────────────────────────────────── # Real-time Ticker (last price + 24h stats) # ────────────────────────────────────────────────────────────────────────────── async def fetch_ticker(symbol: str) -> Dict[str, Any]: cached = ticker_cache.get(f"ticker:{symbol}") if cached: return cached cfg = SYMBOLS[symbol] result = {} category = cfg.category try: if "binance" in cfg.mappings and category == "Crypto": await _rate_limit("binance") async with httpx.AsyncClient(timeout=10) as c: r = await c.get( "https://api.binance.com/api/v3/ticker/24hr", params={"symbol": cfg.mappings["binance"]}, ) d = r.json() result = { "price": float(d["lastPrice"]), "change": float(d["priceChange"]), "change_pct": float(d["priceChangePercent"]), "high_24h": float(d["highPrice"]), "low_24h": float(d["lowPrice"]), "volume_24h": float(d["volume"]), "source": "binance", } else: # yfinance history for robust prev_close def _yf_info(): symbol_mapped = cfg.mappings.get("yfinance", "") t = yf.Ticker(symbol_mapped) # Fetch 5d to ensure we have at least one previous close even across weekends/holidays h = t.history(period="5d") if h.empty: # Fallback to fast_info if history fails fi = t.fast_info return { "price": getattr(fi, "last_price", 0) or 0, "prev_close": getattr(fi, "previous_close", None), "source": "yfinance_fast" } last_price = float(h["Close"].iloc[-1]) prev_close = float(h["Close"].iloc[-2]) if len(h) > 1 else last_price return { "price": last_price, "prev_close": prev_close, "high_24h": float(h["High"].iloc[-1]), "low_24h": float(h["Low"].iloc[-1]), "volume_24h": float(h["Volume"].iloc[-1]), "source": "yfinance_hist", } info = await asyncio.to_thread(_yf_info) price = info.get("price") or 0 prev = info.get("prev_close") # If prev is still None, use price (result 0%) as ultimate fallback if prev is None: prev = price info["change"] = price - prev info["change_pct"] = ((price - prev) / prev * 100) if prev != 0 else 0 result = info result["symbol"] = symbol result["timestamp"] = int(time.time()) ticker_cache.set(f"ticker:{symbol}", result, ttl_seconds=15) return result except Exception as ex: logger.warning("[ticker] %s failed: %s", symbol, ex) raise HTTPException(status_code=502, detail=f"Ticker fetch failed: {ex}") # ────────────────────────────────────────────────────────────────────────────── # 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 (standard industry practice) 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.inf) rsi = 100 - (100 / (1 + rs)) # 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() return (mid + k*std).values, mid.values, (mid - k*std).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() 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: tr = np.maximum(high[1:] - low[1:], np.maximum(np.abs(high[1:] - close[:-1]), np.abs(low[1:] - close[:-1]))) atr = np.full(len(close), np.nan) if len(tr) < period: return atr atr[period] = np.mean(tr[:period]) for i in range(period + 1, len(close)): atr[i] = (atr[i - 1] * (period - 1) + tr[i - 1]) / period return atr def _stoch_rsi(close: np.ndarray, rsi_period=14, stoch_period=14, smooth_k=3, smooth_d=3) -> Tuple[np.ndarray, np.ndarray]: rsi_vals = _rsi(close, rsi_period) k = np.full_like(close, np.nan) for i in range(stoch_period - 1, len(rsi_vals)): window = rsi_vals[i - stoch_period + 1: i + 1] if np.any(np.isnan(window)): continue lo, hi = np.min(window), np.max(window) k[i] = (rsi_vals[i] - lo) / (hi - lo) * 100 if hi != lo else 50 k_smooth = _ema(np.where(np.isnan(k), 50, k), smooth_k) d_smooth = _ema(np.where(np.isnan(k_smooth), 50, k_smooth), smooth_d) return k_smooth, d_smooth def _sma(arr: np.ndarray, period: int) -> np.ndarray: """Simple Moving Average.""" out = np.full_like(arr, np.nan) for i in range(period - 1, len(arr)): out[i] = np.mean(arr[i - period + 1: i + 1]) return out def _cci(high: np.ndarray, low: np.ndarray, close: np.ndarray, period: int = 20) -> np.ndarray: """Commodity Channel Index.""" tp = (high + low + close) / 3.0 out = np.full_like(close, np.nan) for i in range(period - 1, len(tp)): window = tp[i - period + 1: i + 1] sma_val = np.mean(window) mean_dev = np.mean(np.abs(window - sma_val)) out[i] = (tp[i] - sma_val) / (0.015 * mean_dev) if mean_dev > 0 else 0.0 return out def _adx(high: np.ndarray, low: np.ndarray, close: np.ndarray, period: int = 14 ) -> tuple: """Average Directional Index. Returns (adx, plus_di, minus_di).""" n = len(close) plus_dm = np.zeros(n) minus_dm = np.zeros(n) tr = np.zeros(n) for i in range(1, n): up = high[i] - high[i - 1] dn = low[i - 1] - low[i] plus_dm[i] = up if (up > dn and up > 0) else 0.0 minus_dm[i] = dn if (dn > up and dn > 0) else 0.0 tr[i] = max(high[i] - low[i], abs(high[i] - close[i - 1]), abs(low[i] - close[i - 1])) atr_arr = np.full(n, np.nan) plus_di_arr = np.full(n, np.nan) minus_di_arr = np.full(n, np.nan) dx_arr = np.full(n, np.nan) adx_arr = np.full(n, np.nan) if n <= period: return adx_arr, plus_di_arr, minus_di_arr atr_arr[period] = np.mean(tr[1:period + 1]) sm_plus = np.mean(plus_dm[1:period + 1]) sm_minus = np.mean(minus_dm[1:period + 1]) for i in range(period + 1, n): atr_arr[i] = (atr_arr[i - 1] * (period - 1) + tr[i]) / period sm_plus = (sm_plus * (period - 1) + plus_dm[i]) / period sm_minus = (sm_minus * (period - 1) + minus_dm[i]) / period plus_di_arr[i] = (sm_plus / atr_arr[i]) * 100 if atr_arr[i] > 0 else 0 minus_di_arr[i] = (sm_minus / atr_arr[i]) * 100 if atr_arr[i] > 0 else 0 di_sum = plus_di_arr[i] + minus_di_arr[i] dx_arr[i] = abs(plus_di_arr[i] - minus_di_arr[i]) / di_sum * 100 if di_sum > 0 else 0 # B-5: Critical fix for ADX IndexError and logic (BUG-11) # Ensure start_idx doesn't exceed n and at least 1 DX value is used valid_dx = [dx_arr[i] for i in range(period + 1, min(2 * period + 1, n)) if not np.isnan(dx_arr[i])] if valid_dx and len(valid_dx) > 0: start_idx = min(period + len(valid_dx), n - 1) adx_arr[start_idx] = np.mean(valid_dx) for i in range(start_idx + 1, n): if not np.isnan(dx_arr[i]): adx_arr[i] = (adx_arr[i - 1] * (period - 1) + dx_arr[i]) / period return adx_arr, plus_di_arr, minus_di_arr 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: """Momentum = close - close[n periods ago].""" out = np.full_like(close, np.nan) for i in range(period, len(close)): out[i] = close[i] - close[i - period] return out def _williams_r(high: np.ndarray, low: np.ndarray, close: np.ndarray, period: int = 14) -> np.ndarray: """Williams %R.""" out = np.full_like(close, np.nan) for i in range(period - 1, len(close)): hh = np.max(high[i - period + 1: i + 1]) ll = np.min(low[i - period + 1: i + 1]) out[i] = -100 * (hh - close[i]) / (hh - ll) if (hh - ll) > 0 else 0.0 return out def _bull_bear_power(high: np.ndarray, low: np.ndarray, close: np.ndarray, period: int = 13) -> np.ndarray: """Bull Bear Power = Bull Power + Bear Power.""" ema_val = _ema(close, period) bull = high - ema_val bear = low - ema_val return bull + bear def _ultimate_oscillator(high: np.ndarray, low: np.ndarray, close: np.ndarray, p1: int = 7, p2: int = 14, p3: int = 28) -> np.ndarray: """Ultimate Oscillator (7, 14, 28).""" n = len(close) out = np.full(n, np.nan) if n < p3 + 1: return out bp = np.zeros(n) tr = np.zeros(n) for i in range(1, n): tl = min(low[i], close[i - 1]) bp[i] = close[i] - tl tr[i] = max(high[i], close[i - 1]) - tl for i in range(p3, n): s1 = np.sum(bp[i - p1 + 1:i + 1]) / max(np.sum(tr[i - p1 + 1:i + 1]), 1e-10) s2 = np.sum(bp[i - p2 + 1:i + 1]) / max(np.sum(tr[i - p2 + 1:i + 1]), 1e-10) s3 = np.sum(bp[i - p3 + 1:i + 1]) / max(np.sum(tr[i - p3 + 1:i + 1]), 1e-10) out[i] = 100 * (4 * s1 + 2 * s2 + s3) / 7.0 return out def _ichimoku_base(high: np.ndarray, low: np.ndarray, period: int = 26) -> np.ndarray: """Ichimoku Base Line (Kijun-sen).""" out = np.full_like(high, np.nan) for i in range(period - 1, len(high)): hh = np.max(high[i - period + 1: i + 1]) ll = np.min(low[i - period + 1: i + 1]) out[i] = (hh + ll) / 2.0 return out def _vwma(close: np.ndarray, volume: np.ndarray, period: int = 20) -> np.ndarray: """Volume Weighted Moving Average.""" out = np.full_like(close, np.nan) for i in range(period - 1, len(close)): cv = close[i - period + 1: i + 1] * volume[i - period + 1: i + 1] v_sum = np.sum(volume[i - period + 1: i + 1]) out[i] = np.sum(cv) / v_sum if v_sum > 0 else close[i] return out 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 = _sma(close, half) wma_full = _sma(close, period) diff = 2 * wma_half - wma_full hull = _sma(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" if name == "rsi": return "Bán" if value > 70 else "Mua" if value < 30 else "Trung lập" if name == "stoch": return "Bán" if value > 80 else "Mua" if value < 20 else "Trung lập" if name == "cci": return "Bán" if value > 100 else "Mua" if value < -100 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": return "Mua" if value > 0 else "Bán" if value < 0 else "Trung lập" if name == "momentum": return "Mua" if value > 0 else "Bán" if value < 0 else "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 == "stoch_rsi": return "Bán" if value > 80 else "Mua" if value < 20 else "Trung lập" if name == "williams": return "Mua" if value < -80 else "Bán" if value > -20 else "Trung lập" if name == "bbp": return "Mua" if value > 0 else "Bán" if value < 0 else "Trung lập" if name == "ultimate": return "Bán" if value > 70 else "Mua" if value < 30 else "Trung lập" return "Trung lập" def _ma_action(price: float, ma_val: float) -> str: """Classify MA as 'Mua' (price > MA) or 'Bán' (price < MA).""" if ma_val is None or math.isnan(ma_val): return "Trung lập" return "Mua" if price > ma_val else "Bán" 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) # Volume SMA 20 vol_sma = np.array([ np.mean(vols[max(0, i-19):i+1]) for i in range(len(vols)) ]) 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" ), }, "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) scale = float(np.clip(scale, 0.5, 2.0)) if abs(scale - 1.0) > 0.3: logger.warning("[Kronos] High scale correction: %.4f (first_model=%.8f, last=%.8f)", scale, first_model_est, last_close) 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) band_width_pct = abs(model_p90[-1] - model_p10[-1]) / last_close if last_close else 0.0 base_weight = 0.68 if agreement else 0.52 model_weight = _clamp(base_weight - max(0.0, band_width_pct - 0.08) * 1.8, 0.35, 0.82) 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) confidence = 58.0 confidence += 12.0 if agreement else -8.0 confidence += 8.0 if indicators["trend"].get("ema_bullish_stack") else 0.0 confidence -= min(20.0, band_width_pct * 140.0) confidence -= min(15.0, bias_pct * 0.45) confidence = _clamp(confidence, 12.0, 92.0) 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, 1.2, 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]) last_c = closes[-1] # 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": # (Similar logic for short positions) 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 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, ) -> Dict[str, Any]: """ TradingView-style technical analysis dashboard. Returns pure numerical data: oscillators, moving averages, pivot points. No trade setups, no entry/SL/TP, no reasoning text. """ if not data or len(data) < 30: return {"oscillators": {"data": []}, "moving_averages": {"data": []}, "pivot_points": {"data": []}, "summary": {"signal": "Trung lập", "sell": 0, "neutral": 0, "buy": 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] def _lv(arr): v = arr[-1] if len(arr) else float('nan') return None if (v is None or math.isnan(v)) else round(float(v), 2) # ── Compute all oscillators ── rsi14 = _rsi(closes, 14) stoch_k_arr, stoch_d_arr = _stoch_rsi(closes, 14, 14, 3, 3) cci20 = _cci(highs, lows, closes, 20) adx14, plus_di14, minus_di14 = _adx(highs, lows, closes, 14) ao = _awesome_oscillator(highs, lows) mom10 = _momentum(closes, 10) macd_l, macd_s, macd_h = _macd(closes, 12, 26, 9) stoch_rsi_k, stoch_rsi_d = _stoch_rsi(closes, 14, 14, 3, 3) wr14 = _williams_r(highs, lows, closes, 14) bbp = _bull_bear_power(highs, lows, closes, 13) uo = _ultimate_oscillator(highs, lows, closes, 7, 14, 28) osc_data = [] osc_buy = osc_sell = osc_neutral = 0 def _add_osc(label, val, action_name, **kw): nonlocal osc_buy, osc_sell, osc_neutral v = _lv(val) if isinstance(val, np.ndarray) else (round(float(val), 2) if val is not None else None) act = _osc_action(action_name, v if v is not None else 0, **kw) osc_data.append({"name": label, "value": v, "action": act}) if act == "Mua": osc_buy += 1 elif act == "Bán": osc_sell += 1 else: osc_neutral += 1 _add_osc("Chỉ số Sức mạnh tương đối (14)", rsi14, "rsi") _add_osc("Stochastic %K (14, 3, 3)", stoch_k_arr, "stoch") _add_osc("Chỉ số Kênh hàng hóa (20)", cci20, "cci") _add_osc("Chỉ số Định hướng Trung bình (14)", adx14, "adx", plus_di=_lv(plus_di14) or 0, minus_di=_lv(minus_di14) or 0) _add_osc("Chỉ số Dao động AO", ao, "ao") _add_osc("Xung lượng (10)", mom10, "momentum") _add_osc("Cấp độ MACD (12, 26)", macd_l, "macd", signal=_lv(macd_s) or 0) _add_osc("Đường RSI Nhanh (3, 3, 14, 14)", stoch_rsi_k, "stoch_rsi") _add_osc("Vùng Phần trăm Williams (14)", wr14, "williams") _add_osc("Sức Mạnh Giá Lên và Giá Xuống", bbp, "bbp") _add_osc("Dao động Ultimate (7, 14, 28)", uo, "ultimate") osc_total = osc_buy + osc_sell + osc_neutral if osc_buy > osc_sell + 2: osc_signal = "Mua" elif osc_sell > osc_buy + 2: osc_signal = "Bán" else: osc_signal = "Trung lập" # ── Compute all Moving Averages ── ma_data = [] ma_buy = ma_sell = ma_neutral = 0 def _add_ma(label, val_arr): nonlocal ma_buy, ma_sell, ma_neutral v = _lv(val_arr) act = _ma_action(last_close, v if v is not None else last_close) ma_data.append({"name": label, "value": v, "action": act}) if act == "Mua": ma_buy += 1 elif act == "Bán": ma_sell += 1 else: ma_neutral += 1 # EMA periods for p in [10, 20, 30, 50, 100, 200]: _add_ma(f"Trung bình Trượt Hàm mũ ({p})", _ema(closes, p)) _add_ma(f"Đường Trung bình trượt Đơn giản ({p})", _sma(closes, p)) # Ichimoku ichi = _ichimoku_base(highs, lows, 26) _add_ma("Đường cơ sở Ichimoku (9, 26, 52, 26)", ichi) # VWMA vwma_arr = _vwma(closes, vols, 20) _add_ma("Đường Trung bình di động Tỷ trọng tuyến tính (20)", vwma_arr) # Hull MA hull = _hull_ma(closes, 9) _add_ma("Đường trung bình trượt Hull (9)", hull) if ma_buy > ma_sell + 2: ma_signal = "Mua" elif ma_sell > ma_buy + 2: ma_signal = "Bán" else: ma_signal = "Trung lập" # ── Summary (Total) ── total_buy = osc_buy + ma_buy total_sell = osc_sell + ma_sell total_neutral = osc_neutral + ma_neutral if total_buy > total_sell + 6: total_signal = "Mua mạnh" elif total_buy > total_sell + 3: total_signal = "Mua" elif total_sell > total_buy + 6: total_signal = "Bán mạnh" elif total_sell > total_buy + 3: total_signal = "Bán" else: total_signal = "Trung lập" # ── Pivot Points ── # Use last completed candle for pivot calculation last_h = float(highs[-1]) last_l = float(lows[-1]) last_c = float(closes[-1]) pivots = _calc_pivot_points(last_h, last_l, last_c) return { "style": "tradingview", "summary": { "sell": total_sell, "neutral": total_neutral, "buy": total_buy, "signal": total_signal, }, "oscillators": { "sell": osc_sell, "neutral": osc_neutral, "buy": osc_buy, "signal": osc_signal, "data": osc_data, }, "moving_averages": { "sell": ma_sell, "neutral": ma_neutral, "buy": ma_buy, "signal": ma_signal, "data": ma_data, }, "pivot_points": pivots, } # ────────────────────────────────────────────────────────────────────────────── # Market Hours (UTC) # ────────────────────────────────────────────────────────────────────────────── MARKET_SESSIONS = { "NYSE/NASDAQ": {"tz": "US/Eastern", "open": (9, 30), "close": (16, 0), "days": range(5)}, "London (LSE)": {"tz": "Europe/London","open": (8, 0), "close": (16, 30), "days": range(5)}, "Frankfurt (DAX)":{"tz": "Europe/Berlin","open": (9, 0), "close": (17, 30), "days": range(5)}, "Tokyo (TSE)": {"tz": "Asia/Tokyo", "open": (9, 0), "close": (15, 30), "days": range(5)}, "Hong Kong (HKEX)":{"tz":"Asia/Hong_Kong","open":(9,30), "close": (16, 0), "days": range(5)}, "Shanghai (SSE)": {"tz": "Asia/Shanghai","open": (9, 30), "close": (15, 0), "days": range(5)}, "Ho Chi Minh (HOSE)":{"tz":"Asia/Ho_Chi_Minh","open":(9,0),"close":(15,0), "days": range(5)}, "Forex (24/5)": {"always_open": True, "days": range(5), "note": "Closes Sat-Sun UTC"}, "Crypto (24/7)": {"always_open": True, "note": "Never closes"}, } def market_status_now() -> List[Dict[str, Any]]: try: import pytz except ImportError: return [{"market": m, "status": "unknown", "note": "pytz not installed"} for m in MARKET_SESSIONS] results = [] utcnow = datetime.now(timezone.utc) for name, info in MARKET_SESSIONS.items(): if info.get("always_open"): day_ok = utcnow.weekday() in info.get("days", range(7)) results.append({ "market": name, "status": "open" if day_ok else "closed", "note": info.get("note", ""), }) continue try: tz = pytz.timezone(info["tz"]) local = utcnow.astimezone(tz) oh, om = info["open"] ch, cm = info["close"] is_day = local.weekday() in info["days"] is_hours = (local.hour, local.minute) >= (oh, om) and (local.hour, local.minute) < (ch, cm) results.append({ "market": name, "status": "open" if (is_day and is_hours) else "closed", "local_time": local.strftime("%H:%M"), "opens_at": f"{oh:02d}:{om:02d}", "closes_at": f"{ch:02d}:{cm:02d}", }) except Exception as ex: results.append({"market": name, "status": "unknown", "error": str(ex)}) return results # ────────────────────────────────────────────────────────────────────────────── # Kronos Forecaster (identical to v3, with CLIP_DEFAULT fix retained) # ────────────────────────────────────────────────────────────────────────────── class KronosForecaster: MAX_CONTEXT = 512 MODEL_NAME = "NeoQuasar/Kronos-base" def __init__(self) -> None: self._predictor: Optional[Any] = None self._loaded = False self._lock = asyncio.Lock() @property def is_ready(self) -> bool: return self._loaded @property def device(self) -> str: return str(self._predictor.device) if self._predictor else "not_loaded" @property def _clip(self) -> float: if self._predictor is None: return CLIP_DEFAULT return getattr(self._predictor, "clip", CLIP_DEFAULT) async def _lazy_load(self) -> None: if self._loaded: return async with self._lock: if self._loaded: return if not KRONOS_AVAILABLE: raise HTTPException(status_code=500, detail="Kronos not available") try: device = ("cuda:0" if torch.cuda.is_available() else "mps" if (hasattr(torch.backends, "mps") and torch.backends.mps.is_available()) else "cpu") logger.info("[Kronos] Loading on %s …", device) tokenizer = await asyncio.to_thread(KronosTokenizer.from_pretrained, "NeoQuasar/Kronos-Tokenizer-base") model = await asyncio.to_thread(Kronos.from_pretrained, self.MODEL_NAME) self._predictor = KronosPredictor(model, tokenizer, device=device, max_context=self.MAX_CONTEXT) self._loaded = True logger.info("[Kronos] Ready on %s", device) except Exception as ex: logger.error("[Kronos] Init failed: %s", ex) raise HTTPException(status_code=500, detail=f"Kronos init failed: {ex}") @staticmethod def _safe_volume(df: pd.DataFrame) -> np.ndarray: vol = df["volume"].values.astype(np.float32) if vol.sum() == 0: typical = ((df["high"] + df["low"] + df["close"]) / 3).values.astype(np.float32) vol = np.full_like(typical, typical.mean()) return vol async def forecast(self, df: pd.DataFrame, x_timestamp: pd.Series, y_timestamp: pd.Series, horizon: int, sample_count: int = 10) -> Dict[str, Any]: await self._lazy_load() assert self._predictor is not None try: if not isinstance(x_timestamp, pd.Series): x_timestamp = pd.Series(x_timestamp.values if hasattr(x_timestamp, "values") else x_timestamp) if not isinstance(y_timestamp, pd.Series): y_timestamp = pd.Series(y_timestamp.values if hasattr(y_timestamp, "values") else y_timestamp) vol = self._safe_volume(df) if "amount" in df.columns and df["amount"].sum() != 0: amount = df["amount"].values.astype(np.float32) else: typical = ((df["high"] + df["low"] + df["close"]) / 3).values.astype(np.float32) amount = vol * typical x = np.stack([df["open"].values, df["high"].values, df["low"].values, df["close"].values, vol, amount], axis=1).astype(np.float32) x_stamp = calc_time_stamps(x_timestamp).values.astype(np.float32) y_stamp = calc_time_stamps(y_timestamp).values.astype(np.float32) x_mean = np.mean(x, axis=0) x_std = np.std(x, axis=0) x_std_safe = np.where(x_std < 1e-8, 1.0, x_std) x_norm = np.clip((x - x_mean) / x_std_safe, -self._clip, self._clip) x_norm = x_norm[np.newaxis, :] x_stamp = x_stamp[np.newaxis, :] y_stamp = y_stamp[np.newaxis, :] t0 = time.time() samples = await asyncio.to_thread( self._predictor.generate, x=x_norm, x_stamp=x_stamp, y_stamp=y_stamp, pred_len=horizon, T=1.0, top_k=0, top_p=0.9, sample_count=sample_count, verbose=False, return_samples=True, ) logger.info("[Kronos] %.2fs | horizon=%d samples=%d ctx=%d", time.time() - t0, horizon, sample_count, len(df)) if "cuda" in self.device: torch.cuda.empty_cache() close_samples = np.asarray(samples[0, :, :, 3], dtype=float) # Some Kronos checkpoints return the full decoded sequence rather than # only the requested pred_len. Keep the most recent horizon window so # downstream logic always receives a forecast-length vector. if close_samples.shape[-1] > horizon: close_samples = close_samples[:, -horizon:] elif close_samples.shape[-1] < horizon: pad_width = horizon - close_samples.shape[-1] close_samples = np.pad(close_samples, ((0, 0), (0, pad_width)), mode="edge") close_samples = close_samples * x_std_safe[3] + x_mean[3] return { "p10": np.percentile(close_samples, 10, axis=0), "p50": np.percentile(close_samples, 50, axis=0), "p90": np.percentile(close_samples, 90, axis=0), "model_name": self.MODEL_NAME, "context_length": len(df), "output_horizon": int(close_samples.shape[-1]), } except Exception as ex: logger.error("[Kronos] Forecast failed: %s", ex, exc_info=True) raise HTTPException(status_code=500, detail=f"Kronos prediction failed: {ex}") forecaster = KronosForecaster() # ────────────────────────────────────────────────────────────────────────────── # 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]): dead = [] for ws in list(self.active.get(symbol, [])): try: await ws.send_json(data) 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() # ────────────────────────────────────────────────────────────────────────────── # Architecture v5: Security & Guardrails (Module F) # ────────────────────────────────────────────────────────────────────────────── from fastapi import Request from fastapi.responses import JSONResponse ADMIN_TOKEN = "kronos_v5_secret_admin" # Hardcoded as requested IP_LIMITS: Dict[str, List[float]] = defaultdict(list) def rate_limit_guard(request: Request): """F-2: Simple IP-based rate limiting logic.""" ip = request.client.host if request.client else "unknown" now = time.time() IP_LIMITS[ip] = [ts for ts in IP_LIMITS[ip] if now - ts < 60] if len(IP_LIMITS[ip]) >= 60: return True IP_LIMITS[ip].append(now) return False def admin_only(request: Request): """F-1: Admin authentication guard for sensitive endpoints.""" token = request.headers.get("X-Admin-Token") if token != ADMIN_TOKEN: raise HTTPException(status_code=401, detail="Admin access required") return True # ────────────────────────────────────────────────────────────────────────────── # Pydantic Models # ────────────────────────────────────────────────────────────────────────────── class SwitchRequest(BaseModel): symbol: str interval: str class WatchlistRequest(BaseModel): symbols: List[str] app = FastAPI( title="AI Trading Chart API", version="5.0.0", description="OHLCV data, hybrid AI forecasts, technical indicators, and real-time WebSocket prices", ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) @asynccontextmanager async def lifespan(app: FastAPI): # Startup logic logger.info("Starting Kronos AI Backend v5.0...") # Async background tasks asyncio.create_task(ws_manager.heartbeat()) asyncio.create_task(_background_cleanup()) asyncio.create_task(_periodic_health_check()) asyncio.create_task(_prefetch_popular_symbols()) # D-3: Prefetcher # Quick source reachability check (non-blocking) asyncio.create_task(_source_selftest()) if PRELOAD_KRONOS and KRONOS_AVAILABLE: asyncio.create_task(_warmup_kronos()) elif not KRONOS_AVAILABLE: STARTUP_STATE["kronos"]["last_error"] = "Kronos source import failed" yield # Shutdown logic logger.info("Shutting down AI Trading Chart API...") # F-5: Graceful termination of background tasks await GlobalHTTPClient.close() logger.info("Graceful shutdown completed.") async def _background_cleanup(): """Periodic task to evict expired cache entries (BUG-05).""" while True: await asyncio.sleep(300) # Every 5 mins evicted = historical_cache.evict_expired() evicted += forecast_cache.evict_expired() evicted += ticker_cache.evict_expired() if evicted > 0: logger.info("[Cache] Evicted %d expired entries", evicted) async def _periodic_health_check(): """Re-check source health every 30 minutes (BUG-16).""" while True: await asyncio.sleep(1800) await _source_selftest() # Set app lifespan after definition app.router.lifespan_context = lifespan @app.middleware("http") async def rate_limit_middleware(request: Request, call_next): if rate_limit_guard(request): return JSONResponse(status_code=429, content={"detail": "Too many requests"}) return await call_next(request) async def _prefetch_popular_symbols(): """D-3: Prefetch data for major symbols to reduce first-load latency.""" popular = ["XAUUSD", "BTCUSD", "ETHUSD", "DXY", "SP500"] while True: logger.info("[Prefetch] Refreshing popular symbols...") for sym in popular: try: await fetch_historical(sym, "1h", 200) await asyncio.sleep(1) # Gentle throttling except Exception: pass await asyncio.sleep(300) # Every 5 mins async def _source_selftest(): """Ping data sources at startup to confirm reachability (1 attempt each).""" tests = [ ("binance", "https://api.binance.com/api/v3/ping"), ("bybit", "https://api.bybit.com/v5/market/time"), ("coingecko", "https://api.coingecko.com/api/v3/ping"), ("twelvedata", f"https://api.twelvedata.com/api_usage?apikey={TWELVEDATA_API_KEY}"), ("finnhub", f"https://finnhub.io/api/v1/status?token={FINNHUB_API_KEY}"), ] async with httpx.AsyncClient(timeout=8) as c: for name, url in tests: try: r = await c.get(url) STARTUP_STATE["sources"][name] = { "reachable": True, "status_code": r.status_code, "checked_at": datetime.now(timezone.utc).isoformat(), } logger.info("[selftest] %-15s HTTP %d ✓", name, r.status_code) except Exception as ex: STARTUP_STATE["sources"][name] = { "reachable": False, "error": str(ex), "checked_at": datetime.now(timezone.utc).isoformat(), } logger.warning("[selftest] %-15s FAILED: %s", name, ex) async def _warmup_kronos() -> None: """Load Kronos in the background so the first forecast is fast and health is explicit.""" STARTUP_STATE["kronos"]["warming"] = True STARTUP_STATE["kronos"]["last_error"] = None try: await forecaster._lazy_load() STARTUP_STATE["kronos"]["loaded"] = forecaster.is_ready STARTUP_STATE["kronos"]["device"] = forecaster.device logger.info("[startup] Kronos warmup finished on %s", forecaster.device) except Exception as ex: STARTUP_STATE["kronos"]["loaded"] = False STARTUP_STATE["kronos"]["device"] = forecaster.device STARTUP_STATE["kronos"]["last_error"] = str(ex) logger.warning("[startup] Kronos warmup failed: %s", ex) finally: STARTUP_STATE["kronos"]["warming"] = False # ── Symbol / Interval listing ───────────────────────────────────────────────── @app.get("/api/symbols") async def list_symbols( category: Optional[str] = Query(None, description="Filter by category"), ) -> Dict[str, Any]: results = list(SYMBOLS.values()) if category: results = [s for s in results if s.category.lower() == category.lower()] # Group by category for structured response grouped: Dict[str, List[Dict]] = {} for s in results: entry = { "symbol": s.symbol, "label": s.label, "label_en": s.label_en, "category": s.category, "description": s.description, "sources": list(s.mappings.keys()), } grouped.setdefault(s.category, []).append(entry) return { "total": len(results), "supported_intervals": INTERVAL_ORDER, "categories": sorted(grouped.keys()), "symbols_by_category": grouped, "symbols": [ {"symbol": s.symbol, "label": s.label, "label_en": s.label_en, "category": s.category, "sources": list(s.mappings.keys())} for s in results ], } # ── Symbol search ───────────────────────────────────────────────────────────── @app.get("/api/search") async def search_symbols( q: str = Query(..., min_length=1, description="Search query"), ) -> Dict[str, Any]: q_lower = q.lower() results = [] for sym, cfg in SYMBOLS.items(): score = 0 if q_lower == sym.lower(): score = 100 elif q_lower in sym.lower(): score = 80 elif q_lower in cfg.label.lower(): score = 60 elif q_lower in cfg.label_en.lower(): score = 50 elif q_lower in cfg.category.lower(): score = 30 if score > 0: results.append({ "symbol": sym, "label": cfg.label, "label_en": cfg.label_en, "category": cfg.category, "score": score, }) results.sort(key=lambda x: -x["score"]) return {"query": q, "count": len(results), "results": results[:20]} # ── 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), ) -> 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) return {"symbol": symbol, "interval": interval, "source": source, "count": len(data), "data": data} # ── 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), ) -> 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) indicators = compute_indicators(data) return { "symbol": symbol, "interval": interval, "source": source, "candles": len(data), "indicators": indicators, } # ── Real-time ticker ────────────────────────────────────────────────────────── @app.get("/api/ticker/{symbol}") async def get_ticker(symbol: str) -> 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) # ── Watchlist (batch ticker) ────────────────────────────────────────────────── @app.post("/api/watchlist/tickers") async def get_watchlist_tickers(body: WatchlistRequest) -> Dict[str, Any]: results = {} valid_symbols = [s.upper() for s in body.symbols if s.upper() in SYMBOLS] tasks = [fetch_ticker(s) 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)} # ── Forecast ────────────────────────────────────────────────────────────────── @app.get("/api/forecast/{symbol}") async def get_forecast( symbol: str, interval: str = Query("1h"), horizon: int = Query(10, ge=5, le=300), ) -> 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}") prefix = _cache_prefix(symbol, interval) cache_key = f"forecast_{prefix}{horizon}" cached = forecast_cache.get(cache_key) if cached is not None: return cached data_list, source = await fetch_historical(symbol, interval, 1500) if not KRONOS_AVAILABLE: # Return graceful empty forecast so UI doesn't break return { "symbol": symbol, "interval": interval, "forecast": [], "error": "AI Forecaster is currently offline or not found in bundle.", "path_checked": KRONOS_PATH } 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) context_len = min(len(df_hist), KronosForecaster.MAX_CONTEXT) df_context = df_hist.tail(context_len).reset_index(drop=True) logger.info("[forecast] %s %s | ctx=%d/%d | horizon=%d", symbol, interval, context_len, len(df_hist), horizon) last_time = int(df_hist["time"].iloc[-1]) step = STEP_SECONDS[interval] y_timestamps = pd.Series(pd.to_datetime( [last_time + step * (i + 1) for i in range(horizon)], unit="s", utc=True )) indicators = compute_indicators(data_list) sample_count = 10 if forecaster.device in {"not_loaded", "cpu"} else 15 model_output = await forecaster.forecast( df=df_context[["open", "high", "low", "close", "volume", "amount"]], x_timestamp=df_context["timestamps"], y_timestamp=y_timestamps, horizon=horizon, sample_count=sample_count, ) last_close = float(df_hist["close"].iloc[-1]) anchor_output = _build_anchor_forecast(data_list, indicators, horizon, interval) blended = _blend_forecasts(last_close, model_output, anchor_output, indicators) logger.info( "[forecast] ensemble | %s %s | model_weight=%.2f anchor_weight=%.2f confidence=%.1f agreement=%s", symbol, interval, blended["model_weight"], blended["anchor_weight"], blended["confidence"], blended["agreement"], ) forecast_rows: List[Dict[str, Any]] = [ {"time": last_time, "p10": last_close, "p50": last_close, "p90": last_close, "is_actual": True} ] for i in range(horizon): forecast_rows.append({ "time": int(last_time + step * (i + 1)), "p10": round(float(blended["p10"][i]), 6), "p50": round(float(blended["p50"][i]), 6), "p90": round(float(blended["p90"][i]), 6), }) analysis = _build_trade_analysis( symbol=symbol, interval=interval, data=data_list, indicators=indicators, forecast_rows=forecast_rows, confidence=float(blended["confidence"]), source=source, ) response = { "symbol": symbol, "interval": interval, "source": source, "horizon": horizon, "last_close": last_close, "forecast": forecast_rows, "model": { "name": model_output.get("model_name", "Kronos-base"), "context_length": int(model_output.get("context_length", context_len)), "quantiles": [0.1, 0.5, 0.9], "cache_version": CACHE_VERSION, "sample_count": sample_count, }, "ensemble": { "mode": "kronos_plus_anchor", "model_weight": blended["model_weight"], "anchor_weight": blended["anchor_weight"], "trend_agreement": blended["agreement"], "confidence": blended["confidence"], "model_bias_pct": blended["model_bias_pct"], "alignment_scale": blended["scale"], }, "indicators_snapshot": indicators, "analysis": analysis, } forecast_cache.set(cache_key, response, ttl_seconds=forecast_ttl(interval)) return response # ── 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(), } # ── Crypto market overview (CoinGecko) ─────────────────────────────────────── @app.get("/api/crypto/market") async def crypto_market(top: int = Query(20, ge=5, le=100)) -> Dict[str, Any]: cached = ticker_cache.get(f"cg_market:{top}") if cached: return cached await _rate_limit("coingecko") async with httpx.AsyncClient(timeout=15) as c: r = await c.get( "https://api.coingecko.com/api/v3/coins/markets", params={"vs_currency": "usd", "order": "market_cap_desc", "per_page": top, "page": 1, "sparkline": False}, ) if r.status_code != 200: raise HTTPException(502, "CoinGecko markets unavailable") coins = r.json() result = { "count": len(coins), "coins": [{ "rank": c["market_cap_rank"], "id": c["id"], "symbol": _get_canonical_symbol(c["symbol"]), "name": c["name"], "price": c["current_price"], "change_24h": c["price_change_percentage_24h"], "market_cap": c["market_cap"], "volume_24h": c["total_volume"], "high_24h": c["high_24h"], "low_24h": c["low_24h"], } for c in coins], "fetched_at": int(time.time()), } ticker_cache.set(f"cg_market:{top}", result, ttl_seconds=60) return result @app.get("/api/market-peers") async def get_market_peers(symbol: str = Query("BTCUSD")) -> Dict[str, Any]: symbol = symbol.upper() if symbol not in SYMBOLS: # Fallback to general market if unknown return await crypto_market(top=10) cfg = SYMBOLS[symbol] category = cfg.category # 1. Find all peers in the same category peers = [s for s in SYMBOLS.values() if s.category == category and s.symbol != symbol] # 2. Limit to top 10 (or first 10 in registry) peers = peers[:10] # 3. Fetch tickers for peers tasks = [fetch_ticker(p.symbol) for p in peers] results = await asyncio.gather(*tasks, return_exceptions=True) peer_data = [] for p, res in zip(peers, results): if isinstance(res, Exception): continue peer_data.append({ "symbol": p.symbol, "name": p.label, "price": res.get("price", 0), "change_24h": res.get("change_pct", res.get("change_24h", 0)), "description": p.description }) return { "category": category, "symbol": symbol, "peers": peer_data, "count": len(peer_data) } # ── 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}") logger.info("[switch] %s %s → hist=%d forecast=%d", symbol, interval, h_cleared, f_cleared) return { "status": "cleared", "symbol": symbol, "interval": interval, "cleared": {"historical": h_cleared, "forecast": f_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) -> Dict[str, Any]: symbol = symbol.upper() interval = interval.lower() 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}") return {"symbol": symbol, "interval": interval, "cleared": {"historical": h_cleared, "forecast": f_cleared}} @app.delete("/api/cache") async def clear_all_cache( target: str = Query("all", description="all | historical | forecast | ticker"), ) -> Dict[str, Any]: h = historical_cache.clear() if target in ("all", "historical") else 0 f = forecast_cache.clear() if target in ("all", "forecast") else 0 t = ticker_cache.clear() if target in ("all", "ticker") else 0 return {"cleared": {"historical": h, "forecast": f, "ticker": t}} @app.get("/api/cache/stats") async def cache_stats() -> Dict[str, Any]: return { "cache_version": CACHE_VERSION, "historical": historical_cache.stats(), "forecast": forecast_cache.stats(), "ticker": ticker_cache.stats(), } # ── Health ──────────────────────────────────────────────────────────────────── @app.get("/api/health") async def health_check() -> Dict[str, Any]: STARTUP_STATE["kronos"]["loaded"] = forecaster.is_ready STARTUP_STATE["kronos"]["device"] = forecaster.device return { "status": "online" if KRONOS_AVAILABLE else "degraded", "version": "5.0.0", "model_ready": forecaster.is_ready, "device": forecaster.device, "symbols_count": len(SYMBOLS), "timestamp": datetime.now(timezone.utc).isoformat(), "cache_version": CACHE_VERSION, "python_version": sys.version.split()[0], "kronos": STARTUP_STATE["kronos"], "startup_checks": STARTUP_STATE["sources"], } # ─── EXTRA API ENDPOINTS (v5.0) ──────────────────────────────────────────────── @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), "cache": { "historical": h_stats, "forecast": f_stats, "ticker": t_stats, }, "circuit_breakers": cb_stats, "sources": STARTUP_STATE["sources"], "kronos_status": STARTUP_STATE["kronos"]["loaded"], } @app.get("/api/backtest/{symbol}") async def get_backtest(symbol: str, interval: str = "1h", limit: int = 500): """E-1: POC Backtesting: Calculates returns if following AI signals.""" symbol = _get_canonical_symbol(symbol) 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 = "1h", limit: int = 200, buckets: int = 20): """E-4: POC Volume Profile.""" symbol = _get_canonical_symbol(symbol) data, _ = await fetch_historical(symbol, interval, limit) 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": []} 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 ─────────────────────────────────────────────────────────── FRONTEND_PATH = os.path.join(PROJECT_ROOT, "frontend") if os.path.exists(FRONTEND_PATH): app.mount("/", StaticFiles(directory=FRONTEND_PATH, html=True), name="frontend") logger.info("Mounted frontend: %s", FRONTEND_PATH) else: logger.warning("Frontend path not found: %s", FRONTEND_PATH)