""" 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 hmac import hashlib from dotenv import load_dotenv # D-1: Load environment variables from .env file (v6.0) load_dotenv() import re import sys import time from collections import defaultdict from dataclasses import asdict, 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): # D-1: Security Hardening (v6.0) - Enforce environment variables twelvedata_api_key: Optional[str] = os.getenv("TWELVEDATA_API_KEY") finnhub_api_key: Optional[str] = os.getenv("FINNHUB_API_KEY") binance_api_key: Optional[str] = os.getenv("BINANCE_API_KEY") binance_api_secret: Optional[str] = os.getenv("BINANCE_API_SECRET") bybit_api_key: Optional[str] = os.getenv("BYBIT_API_KEY") bybit_api_secret: Optional[str] = os.getenv("BYBIT_API_SECRET") gemini_api_key: Optional[str] = os.getenv("GEMINI_API_KEY") alphavantage_api_key: Optional[str] = os.getenv("ALPHAVANTAGE_API_KEY") admin_token: str = os.getenv("ADMIN_TOKEN", "kronos_v6_default_secret") # Environment detection is_hf: bool = os.getenv("SPACE_ID") is not None # App Config host: str = os.getenv("HOST", "0.0.0.0") port: int = int(os.getenv("PORT", 8000)) preload_kronos: bool = os.getenv("PRELOAD_KRONOS", "True").lower() == "true" cache_version: str = "v6.0.0-final" # 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]: """Synchronous read is usually fast enough for SQLite.""" 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): """Asynchronous write: push to queue (P0).""" if hasattr(self, "_queue"): self._queue.put_nowait((key, payload, ttl)) else: # Fallback if queue not ready self._write_sync(key, payload, ttl) def _write_sync(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] Sync write error: %s", ex) async def start_writer(self): """Worker loop to process async writes.""" self._queue = asyncio.Queue() logger.info("[Persistence] Async writer started.") while True: key, payload, ttl = await self._queue.get() try: await asyncio.to_thread(self._write_sync, key, payload, ttl) except Exception as ex: logger.error("[Persistence] Writer loop error: %s", ex) finally: self._queue.task_done() def evict(self): try: with sqlite3.connect(self.db_path) as conn: conn.execute("DELETE FROM cache WHERE expiry < ?", (time.time(),)) except Exception as ex: logger.error("[Persistence] Eviction error: %s", ex) persistent_cache = PersistentCache(os.path.join(PROJECT_ROOT, "kronos_v5.db")) # B-12: Global Configuration Instances settings = Settings() CACHE_VERSION = settings.cache_version ADMIN_TOKEN = settings.admin_token TWELVEDATA_API_KEY = settings.twelvedata_api_key FINNHUB_API_KEY = settings.finnhub_api_key # Binance, Bybit, CoinGecko, yfinance, FRED — no key or optional key required # ────────────────────────────────────────────────────────────────────────────── # Constants # ────────────────────────────────────────────────────────────────────────────── SUPPORTED_INTERVALS: frozenset = frozenset({"1m", "5m", "15m", "1h", "4h", "1d", "1w"}) INTERVAL_ORDER: List[str] = ["1m", "5m", "15m", "1h", "4h", "1d", "1w"] TWELVE_INTERVAL_MAP: Dict[str, str] = { "1m": "1min", "5m": "5min", "15m": "15min", "1h": "1h", "4h": "4h", "1d": "1day", "1w": "1week", } FINNHUB_RESOLUTION_MAP: Dict[str, str] = { "1m": "1", "5m": "5", "15m": "15", "1h": "60","4h": "240","1d": "D", "1w": "W", } YF_INTERVAL_MAP: Dict[str, str] = { "1m": "1m", "5m": "5m", "15m": "15m", "1h": "60m","4h": "60m","1d": "1d", "1w": "1wk", } YF_PERIOD_MAP: Dict[str, str] = { "1m": "7d", "5m": "60d","15m": "60d", "1h": "730d","4h": "60d","1d": "max","1w": "max", } BINANCE_INTERVAL_MAP: Dict[str, str] = { "1m": "1m", "5m": "5m", "15m": "15m", "1h": "1h", "4h": "4h","1d": "1d", "1w": "1w", } BYBIT_INTERVAL_MAP: Dict[str, str] = { "1m": "1", "5m": "5", "15m": "15", "1h": "60", "4h": "240","1d": "D", "1w": "W", } STEP_SECONDS: Dict[str, int] = { "1m": 60, "5m": 300, "15m": 900, "1h": 3600, "4h": 14400,"1d": 86400,"1w": 604800, } # Source priority by asset category (first available mapping wins) CATEGORY_SOURCE_PRIORITY: Dict[str, List[str]] = { "Crypto": ["binance", "bybit", "coingecko", "yfinance", "finnhub"], "Cặp tiền": ["binance", "twelvedata", "finnhub", "yfinance"], "Kim loại": ["binance", "twelvedata", "yfinance", "finnhub"], "Năng lượng": ["twelvedata", "yfinance", "finnhub"], "Nông sản": ["twelvedata", "yfinance"], "Nguyên liệu CN": ["twelvedata", "yfinance"], "Chỉ số": ["binance", "twelvedata", "yfinance", "finnhub"], "Cổ phiếu Mỹ": ["binance", "twelvedata", "finnhub", "yfinance"], "Cổ phiếu VN": ["yfinance"], "Trái phiếu": ["twelvedata", "yfinance"], "ETF": ["binance", "twelvedata", "yfinance", "finnhub"], } # Fallback for unknown categories DEFAULT_SOURCE_PRIORITY: List[str] = ["binance", "yfinance", "twelvedata", "finnhub"] CACHE_VERSION = "v12" 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=1.5, 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" binance_type: str = "spot" # "spot" or "futures" 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", bin_type: str = "spot") -> SymbolConfig: return SymbolConfig(sym, label, label_en, cat, mappings, cg_id, bybit_cat, bin_type, desc) SYMBOLS: Dict[str, SymbolConfig] = { # ══════════════════════════════════════════════════════════════════════ # 1. KIM LOẠI (Metals) # ══════════════════════════════════════════════════════════════════════ "XAUUSD": _s("XAUUSD","Vàng (Gold)","Gold","Kim loại",{"binance":"PAXGUSDT","twelvedata":"XAU/USD","yfinance":"GC=F"}), "XAGUSD": _s("XAGUSD","Bạc (Silver)","Silver","Kim loại",{"twelvedata":"XAG/USD","yfinance":"SI=F"}), "XPTUSD": _s("XPTUSD","Bạch kim (Platinum)","Platinum","Kim loại",{"twelvedata":"XPT/USD","yfinance":"PL=F"}), "XPDUSD": _s("XPDUSD","Palladium","Palladium","Kim loại",{"twelvedata":"XPD/USD","yfinance":"PA=F"}), "HGUSD": _s("HGUSD","Đồng (Copper)","Copper","Kim loại",{"twelvedata":"COPPER","yfinance":"HG=F"}), "ALUSD": _s("ALUSD","Nhôm (Aluminum)","Aluminum","Kim loại",{"yfinance":"AL=F"}), "ZNUSD": _s("ZNUSD","Kẽm (Zinc)","Zinc","Kim loại",{"yfinance":"ZN=F"}), "NIUSD": _s("NIUSD","Niken (Nickel)","Nickel","Kim loại",{"yfinance":"NI=F"}), "PBUSD": _s("PBUSD","Chì (Lead)","Lead","Kim loại",{"yfinance":"LED=F"}), "SNUSD": _s("SNUSD","Thiếc (Tin)","Tin","Kim loại",{"yfinance":"SN=F"}), # ══════════════════════════════════════════════════════════════════════ # 2. NĂNG LƯỢNG (Energy) # ══════════════════════════════════════════════════════════════════════ "USOIL": _s("USOIL","Dầu thô WTI","WTI Oil","Năng lượng",{"twelvedata":"WTI","yfinance":"CL=F"}), "UKOIL": _s("UKOIL","Dầu Brent","Brent Oil","Năng lượng",{"twelvedata":"BRENT","yfinance":"BZ=F"}), "NGAS": _s("NGAS","Khí tự nhiên","Natural Gas","Năng lượng",{"twelvedata":"NATGAS","yfinance":"NG=F"}), "GASO": _s("GASO","Xăng RBOB","Gasoline","Năng lượng",{"yfinance":"RB=F"}), "HEAT": _s("HEAT","Dầu sưởi","Heating Oil","Năng lượng",{"yfinance":"HO=F"}), "COAL": _s("COAL","Than (Coal)","Coal","Năng lượng",{"yfinance":"MTF=F"}), # ══════════════════════════════════════════════════════════════════════ # 3. NGUYÊN LIỆU CÔNG NGHIỆP (Industrial Materials) # ══════════════════════════════════════════════════════════════════════ "COFFEE": _s("COFFEE","Cà phê Arabica","Arabica Coffee","Nguyên liệu CN",{"twelvedata":"COFFEE","yfinance":"KC=F"}), "ROBUSTA":_s("ROBUSTA","Cà phê Robusta","Robusta Coffee","Nguyên liệu CN",{"yfinance":"RC=F"}), "COCOA": _s("COCOA","Ca cao (Cocoa)","Cocoa","Nguyên liệu CN",{"yfinance":"CC=F"}), "SUGAR11":_s("SUGAR11","Đường 11","Sugar No.11","Nguyên liệu CN",{"twelvedata":"SUGAR","yfinance":"SB=F"}), "LSUGAR": _s("LSUGAR","Đường trắng","White Sugar","Nguyên liệu CN",{"yfinance":"LSU.L"}), "PALMOIL":_s("PALMOIL","Dầu cọ thô","Palm Oil","Nguyên liệu CN",{"yfinance":"FCPO=F"}), "COTTON": _s("COTTON","Bông (Cotton)","Cotton","Nguyên liệu CN",{"twelvedata":"COTTON","yfinance":"CT=F"}), "RUBBER": _s("RUBBER","Cao su RSS3","Rubber","Nguyên liệu CN",{"yfinance":"JR=F"}), "ORANGE": _s("ORANGE","Nước cam","Orange Juice","Nguyên liệu CN",{"yfinance":"OJ=F"}), # ══════════════════════════════════════════════════════════════════════ # 4. NÔNG SẢN (Agriculture) # ══════════════════════════════════════════════════════════════════════ "CORN": _s("CORN","Ngô (Corn)","Corn","Nông sản",{"twelvedata":"CORN","yfinance":"ZC=F"}), "WHEAT": _s("WHEAT","Lúa mì (Wheat)","Wheat","Nông sản",{"twelvedata":"WHEAT","yfinance":"ZW=F"}), "SOY": _s("SOY","Đậu tương","Soybeans","Nông sản",{"twelvedata":"SOYBEAN","yfinance":"ZS=F"}), "SOYOIL": _s("SOYOIL","Dầu đậu tương","Soybean Oil","Nông sản",{"yfinance":"ZL=F"}), "SOYMEAL":_s("SOYMEAL","Khô đậu tương","Soybean Meal","Nông sản",{"yfinance":"ZM=F"}), "RICE": _s("RICE","Gạo (Rice)","Rice","Nông sản",{"yfinance":"ZR=F"}), "OATS": _s("OATS","Yến mạch (Oats)","Oats","Nông sản",{"yfinance":"ZO=F"}), # ══════════════════════════════════════════════════════════════════════ # 5. CRYPTO (Top 50+) # ══════════════════════════════════════════════════════════════════════ "BTCUSD": _s("BTCUSD","Bitcoin (BTC)","Bitcoin","Crypto",{"binance":"BTCUSDT","bybit":"BTCUSDT","coingecko":"bitcoin"}), "ETHUSD": _s("ETHUSD","Ethereum (ETH)","Ethereum","Crypto",{"binance":"ETHUSDT","bybit":"ETHUSDT","coingecko":"ethereum"}), "BNBUSD": _s("BNBUSD","BNB (BNB)","BNB","Crypto",{"binance":"BNBUSDT","bybit":"BNBUSDT","coingecko":"binancecoin"}), "SOLUSD": _s("SOLUSD","Solana (SOL)","Solana","Crypto",{"binance":"SOLUSDT","bybit":"SOLUSDT","coingecko":"solana"}), "XRPUSD": _s("XRPUSD","Ripple (XRP)","XRP","Crypto",{"binance":"XRPUSDT","bybit":"XRPUSDT","coingecko":"ripple"}), "ADAUSD": _s("ADAUSD","Cardano (ADA)","Cardano","Crypto",{"binance":"ADAUSDT","bybit":"ADAUSDT","coingecko":"cardano"}), "AVAXUSD": _s("AVAXUSD","Avalanche (AVAX)","Avalanche","Crypto",{"binance":"AVAXUSDT","bybit":"AVAXUSDT","coingecko":"avalanche-2"}), "DOTUSD": _s("DOTUSD","Polkadot (DOT)","Polkadot","Crypto",{"binance":"DOTUSDT","bybit":"DOTUSDT","coingecko":"polkadot"}), "MATICUSD": _s("MATICUSD","Polygon (MATIC)","Polygon","Crypto",{"binance":"MATICUSDT","bybit":"MATICUSDT","coingecko":"matic-network"}), "LINKUSD": _s("LINKUSD","Chainlink (LINK)","Chainlink","Crypto",{"binance":"LINKUSDT","bybit":"LINKUSDT","coingecko":"chainlink"}), "LTCUSD": _s("LTCUSD","Litecoin (LTC)","Litecoin","Crypto",{"binance":"LTCUSDT","bybit":"LTCUSDT","coingecko":"litecoin"}), "UNIUSD": _s("UNIUSD","Uniswap (UNI)","Uniswap","Crypto",{"binance":"UNIUSDT","bybit":"UNIUSDT","coingecko":"uniswap"}), "ATOMUSD": _s("ATOMUSD","Cosmos (ATOM)","Cosmos","Crypto",{"binance":"ATOMUSDT"}), "NEARUSD": _s("NEARUSD","NEAR Protocol","NEAR","Crypto",{"binance":"NEARUSDT"}), "APTUSD": _s("APTUSD","Aptos (APT)","Aptos","Crypto",{"binance":"APTUSDT"}), "SUIUSD": _s("SUIUSD","Sui (SUI)","Sui","Crypto",{"binance":"SUIUSDT"}), "ARBUSD": _s("ARBUSD","Arbitrum (ARB)","Arbitrum","Crypto",{"binance":"ARBUSDT"}), "OPUSD": _s("OPUSD","Optimism (OP)","Optimism","Crypto",{"binance":"OPUSDT"}), "TONUSD": _s("TONUSD","Toncoin (TON)","Toncoin","Crypto",{"binance":"TONUSDT"}), "TRXUSD": _s("TRXUSD","Tron (TRX)","Tron","Crypto",{"binance":"TRXUSDT"}), "DOGEUSD":_s("DOGEUSD","Dogecoin (DOGE)","Dogecoin","Crypto",{"binance":"DOGEUSDT"}), "SHIBUSD":_s("SHIBUSD","Shiba Inu (SHIB)","Shiba Inu","Crypto",{"binance":"SHIBUSDT"}), "PEPEUSD":_s("PEPEUSD","PEPE Coin","PEPE","Crypto",{"binance":"PEPEUSDT"}), "FLOKIUSD":_s("FLOKIUSD","FLOKI Inu","FLOKI","Crypto",{"binance":"FLOKIUSDT"}), "BONKUSD":_s("BONKUSD","BONK Coin","BONK","Crypto",{"binance":"BONKUSDT"}), "WIFUSD": _s("WIFUSD","dogwifhat (WIF)","WIF","Crypto",{"binance":"WIFUSDT"}), "RNDRUSD":_s("RNDRUSD","Render (RNDR)","Render","Crypto",{"binance":"RNDRUSDT"}), "FETUSD": _s("FETUSD","Fetch.ai (FET)","Fetch.ai","Crypto",{"binance":"FETUSDT"}), "STXUSD": _s("STXUSD","Stacks (STX)","Stacks","Crypto",{"binance":"STXUSDT"}), "FILUSD": _s("FILUSD","Filecoin (FIL)","Filecoin","Crypto",{"binance":"FILUSDT"}), "VETUSD": _s("VETUSD","VeChain (VET)","VeChain","Crypto",{"binance":"VETUSDT"}), "ICPUSD": _s("ICPUSD","Internet Computer","ICP","Crypto",{"binance":"ICPUSDT"}), # ══════════════════════════════════════════════════════════════════════ # 6. CẶP TIỀN (Forex) # ══════════════════════════════════════════════════════════════════════ "DXY": _s("DXY","Chỉ số USD (DXY)","USD Index","Cặp tiền",{"twelvedata":"DXY","yfinance":"DX-Y.NYB"}), "EURUSD": _s("EURUSD","EUR/USD","EUR/USD","Cặp tiền",{"binance":"EURUSDT","twelvedata":"EUR/USD","yfinance":"EURUSD=X"}), "GBPUSD": _s("GBPUSD","GBP/USD","GBP/USD","Cặp tiền",{"binance":"GBPUSDT","twelvedata":"GBP/USD","yfinance":"GBPUSD=X"}), "USDJPY": _s("USDJPY","USD/JPY","USD/JPY","Cặp tiền",{"binance":"JPYUSDT","twelvedata":"USD/JPY","yfinance":"JPY=X"}), "USDCHF": _s("USDCHF","USD/CHF","USD/CHF","Cặp tiền",{"twelvedata":"USD/CHF","yfinance":"CHF=X"}), "AUDUSD": _s("AUDUSD","AUD/USD","AUD/USD","Cặp tiền",{"binance":"AUDUSDT","twelvedata":"AUD/USD","yfinance":"AUDUSD=X"}), "USDCAD": _s("USDCAD","USD/CAD","USD/CAD","Cặp tiền",{"twelvedata":"USD/CAD","yfinance":"CAD=X"}), "NZDUSD": _s("NZDUSD","NZD/USD","NZD/USD","Cặp tiền",{"twelvedata":"NZD/USD","yfinance":"NZDUSD=X"}), "GBPJPY": _s("GBPJPY","GBP/JPY","GBP/JPY","Cặp tiền",{"twelvedata":"GBP/JPY","yfinance":"GBPJPY=X"}), "EURJPY": _s("EURJPY","EUR/JPY","EUR/JPY","Cặp tiền",{"binance":"EURJPY","twelvedata":"EUR/JPY"}), "EURGBP": _s("EURGBP","EUR/GBP","EUR/GBP","Cặp tiền",{"binance":"EURGBP","twelvedata":"EUR/GBP"}), "USDVND": _s("USDVND","USD/VND","USD/VND","Cặp tiền",{"yfinance":"VND=X"}), "USDCNH": _s("USDCNH","USD/CNH","USD/CNH","Cặp tiền",{"twelvedata":"USD/CNH"}), "USDHKD": _s("USDHKD","USD/HKD","USD/HKD","Cặp tiền",{"twelvedata":"USD/HKD"}), "USDSGD": _s("USDSGD","USD/SGD","USD/SGD","Cặp tiền",{"twelvedata":"USD/SGD"}), # ══════════════════════════════════════════════════════════════════════ # 7. CHỈ SỐ THẾ GIỚI (Global Indices) # ══════════════════════════════════════════════════════════════════════ "SP500": _s("SP500","S&P 500","S&P 500","Chỉ số",{"yfinance":"^GSPC","twelvedata":"SPX"}, bin_type="futures"), "NASDAQ100":_s("NASDAQ100","Nasdaq 100","Nasdaq 100","Chỉ số",{"yfinance":"^NDX","twelvedata":"NDX"}, bin_type="futures"), "DOW30": _s("DOW30","Dow Jones 30","Dow Jones","Chỉ số",{"yfinance":"^DJI"}), "RUSSELL2000":_s("RUSSELL2000","Russell 2000","Russell 2000","Chỉ số",{"yfinance":"^RUT"}), "VIX": _s("VIX","Chỉ số sợ hãi (VIX)","VIX Index","Chỉ số",{"yfinance":"^VIX"}), "UK100": _s("UK100","FTSE 100 (UK)","FTSE 100","Chỉ số",{"yfinance":"^FTSE"}), "DAX40": _s("DAX40","DAX 40 (Đức)","DAX 40","Chỉ số",{"yfinance":"^GDAXI"}), "CAC40": _s("CAC40","CAC 40 (Pháp)","CAC 40","Chỉ số",{"yfinance":"^FCHI"}), "EU50": _s("EU50","Euro Stoxx 50","Euro Stoxx 50","Chỉ số",{"yfinance":"^STOXX50E"}), "NIKKEI225":_s("NIKKEI225","Nikkei 225 (Nhật)","Nikkei 225","Chỉ số",{"yfinance":"^N225"}), "HSI": _s("HSI","Hang Seng (HK)","Hang Seng","Chỉ số",{"yfinance":"^HSI"}), "VNINDEX": _s("VNINDEX","VN-Index","VN-Index","Chỉ số",{"yfinance":"^VNINDEX"}), "HNXINDEX": _s("HNXINDEX","HNX-Index","HNX-Index","Chỉ số",{"yfinance":"^HNX"}), # ══════════════════════════════════════════════════════════════════════ # 8. CỔ PHIẾU MỸ (US Stocks) # ══════════════════════════════════════════════════════════════════════ "AAPL": _s("AAPL","Apple Inc.","Apple","Cổ phiếu Mỹ",{"yfinance":"AAPL"}), "MSFT": _s("MSFT","Microsoft Corp.","Microsoft","Cổ phiếu Mỹ",{"yfinance":"MSFT"}), "GOOGL": _s("GOOGL","Alphabet (Google)","Google","Cổ phiếu Mỹ",{"yfinance":"GOOGL"}), "AMZN": _s("AMZN","Amazon.com","Amazon","Cổ phiếu Mỹ",{"yfinance":"AMZN"}), "NVDA": _s("NVDA","Nvidia Corp.","Nvidia","Cổ phiếu Mỹ",{"yfinance":"NVDA"}), "META": _s("META","Meta Platforms","Meta","Cổ phiếu Mỹ",{"yfinance":"META"}), "TSLA": _s("TSLA","Tesla Inc.","Tesla","Cổ phiếu Mỹ",{"yfinance":"TSLA"}), "AMD": _s("AMD","AMD","AMD","Cổ phiếu Mỹ",{"yfinance":"AMD"}), "INTC": _s("INTC","Intel Corp.","Intel","Cổ phiếu Mỹ",{"yfinance":"INTC"}), "TSM": _s("TSM","TSMC","TSMC","Cổ phiếu Mỹ",{"yfinance":"TSM"}), "AVGO": _s("AVGO","Broadcom","Broadcom","Cổ phiếu Mỹ",{"yfinance":"AVGO"}), "NFLX": _s("NFLX","Netflix","Netflix","Cổ phiếu Mỹ",{"yfinance":"NFLX"}), "COIN": _s("COIN","Coinbase","Coinbase","Cổ phiếu Mỹ",{"yfinance":"COIN"}), "MSTR": _s("MSTR","MicroStrategy","MicroStrategy","Cổ phiếu Mỹ",{"yfinance":"MSTR"}), "JPM": _s("JPM","JPMorgan Chase","JPMorgan","Cổ phiếu Mỹ",{"yfinance":"JPM"}), "GS": _s("GS","Goldman Sachs","Goldman Sachs","Cổ phiếu Mỹ",{"yfinance":"GS"}), "V": _s("V","Visa Inc.","Visa","Cổ phiếu Mỹ",{"yfinance":"V"}), "MA": _s("MA","Mastercard","Mastercard","Cổ phiếu Mỹ",{"yfinance":"MA"}), "WMT": _s("WMT","Walmart","Walmart","Cổ phiếu Mỹ",{"yfinance":"WMT"}), "XOM": _s("XOM","ExxonMobil","ExxonMobil","Cổ phiếu Mỹ",{"yfinance":"XOM"}), # ══════════════════════════════════════════════════════════════════════ # 9. CỔ PHIẾU VIỆT NAM (Vietnam Stocks) # ══════════════════════════════════════════════════════════════════════ "VCB": _s("VCB","Vietcombank (VCB)","Vietcombank","Cổ phiếu VN",{"yfinance":"VCB.VN"}), "FPT": _s("FPT","FPT Corp (FPT)","FPT","Cổ phiếu VN",{"yfinance":"FPT.VN"}), "VIC": _s("VIC","Vingroup (VIC)","Vingroup","Cổ phiếu VN",{"yfinance":"VIC.VN"}), "VHM": _s("VHM","Vinhomes (VHM)","Vinhomes","Cổ phiếu VN",{"yfinance":"VHM.VN"}), "HPG": _s("HPG","Hòa Phát (HPG)","Hoa Phat Steel","Cổ phiếu VN",{"yfinance":"HPG.VN"}), "SSI": _s("SSI","Chứng khoán SSI","SSI","Cổ phiếu VN",{"yfinance":"SSI.VN"}), "VND": _s("VND","VNDirect (VND)","VNDirect","Cổ phiếu VN",{"yfinance":"VND.VN"}), "MWG": _s("MWG","Thế Giới Di Động","MWG","Cổ phiếu VN",{"yfinance":"MWG.VN"}), "VNM": _s("VNM","Vinamilk (VNM)","Vinamilk","Cổ phiếu VN",{"yfinance":"VNM.VN"}), "MSN": _s("MSN","Masan Group","Masan","Cổ phiếu VN",{"yfinance":"MSN.VN"}), "GAS": _s("GAS","PV GAS","PV GAS","Cổ phiếu VN",{"yfinance":"GAS.VN"}), "BID": _s("BID","BIDV Bank","BIDV","Cổ phiếu VN",{"yfinance":"BID.VN"}), "CTG": _s("CTG","VietinBank","VietinBank","Cổ phiếu VN",{"yfinance":"CTG.VN"}), "TCB": _s("TCB","Techcombank","Techcombank","Cổ phiếu VN",{"yfinance":"TCB.VN"}), "MBB": _s("MBB","MB Bank","MB Bank","Cổ phiếu VN",{"yfinance":"MBB.VN"}), "STB": _s("STB","Sacombank","Sacombank","Cổ phiếu VN",{"yfinance":"STB.VN"}), "ACB": _s("ACB","ACB Bank","ACB","Cổ phiếu VN",{"yfinance":"ACB.VN"}), "HDB": _s("HDB","HDBank","HDBank","Cổ phiếu VN",{"yfinance":"HDB.VN"}), "VPB": _s("VPB","VPBank","VPBank","Cổ phiếu VN",{"yfinance":"VPB.VN"}), "DGC": _s("DGC","Hóa chất Đức Giang","DGC","Cổ phiếu VN",{"yfinance":"DGC.VN"}), # ══════════════════════════════════════════════════════════════════════ # 10. TRÁI PHIẾU & ETF # ══════════════════════════════════════════════════════════════════════ "US10Y": _s("US10Y","Trái phiếu Mỹ 10Y","US 10Y Treasury","Trái phiếu",{"yfinance":"^TNX"}), "US02Y": _s("US02Y","Trái phiếu Mỹ 2Y","US 2Y Treasury","Trái phiếu",{"yfinance":"^IRX"}), "SPY": _s("SPY","SPDR S&P 500 ETF","SPY","ETF",{"yfinance":"SPY"}), "QQQ": _s("QQQ","Invesco QQQ Trust","QQQ","ETF",{"yfinance":"QQQ"}), "DIA": _s("DIA","Dow Jones ETF","DIA","ETF",{"yfinance":"DIA"}), } # ────────────────────────────────────────────────────────────────────────────── # 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() # Explicitly clear on startup to ensure fresh v6.1+ format historical_cache.clear() forecast_cache.clear() ticker_cache.clear() def _cache_prefix(symbol: str, interval: str) -> str: return f"{CACHE_VERSION}:{symbol}:{interval}:" def interval_ttl(interval: str) -> int: if interval in {"1m", "5m"}: return 20 if interval == "15m": return 30 if interval in {"1h", "4h"}: return 60 return 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 (BUG-07 Fix) # Reduce threshold further for specific low-volatility asset classes symbol_cfg = None if records and "symbol" in records[0]: # Not always present in raw rows pass normalized: List[Dict[str, Any]] = [] for row in records: try: t = _parse_timestamp(row["time"]) o = float(row["open"]) h = float(row["high"]) l = float(row["low"]) c = float(row["close"]) v = float(row.get("volume") or 0.0) if any(math.isnan(x) or math.isinf(x) for x in [o, h, l, c, v]): continue # Adaptive threshold: permit smaller ranges for shorter timeframes # and specifically for assets where o > 0. # 1e-9 is effectively "almost zero" to catch true dead candles while preserving JPY/Bonds. if o > 0 and (h - l) / o < 1e-9: continue normalized.append({"time": t, "open": o, "high": h, "low": l, "close": c, "volume": v}) except Exception: continue normalized.sort(key=lambda x: x["time"]) seen: set = set() deduped: List[Dict[str, Any]] = [] for r in normalized: if r["time"] in seen: continue seen.add(r["time"]) deduped.append(r) return deduped # ────────────────────────────────────────────────────────────────────────────── # Data Sources # ────────────────────────────────────────────────────────────────────────────── async def fetch_binance(symbol: str, interval: str, limit: int) -> List[Dict[str, Any]]: await _rate_limit("binance") cfg = SYMBOLS[symbol] endpoint_symbol = cfg.mappings["binance"] # Define request details endpoint = "/fapi/v1/klines" if cfg.binance_type == "futures" else "/api/v3/klines" params = { "symbol": endpoint_symbol, "interval": BINANCE_INTERVAL_MAP.get(interval, "1h"), "limit": min(max(limit, 1), 1000), } # B-2: Endpoint Rotation for HF/Cloud environments endpoints_spot = ["https://api.binance.com", "https://api1.binance.com", "https://api2.binance.com", "https://api3.binance.com", "https://data-api.binance.com"] endpoints_fapi = ["https://fapi.binance.com"] # fapi usually more restricted, but try first selected_endpoints = endpoints_fapi if cfg.binance_type == "futures" else endpoints_spot # If on HF, we know api.binance.com is likely blocked, so we can try data-api or alternates faster if settings.is_hf and cfg.binance_type == "spot": # Move data-api to the front for HF spot selected_endpoints = ["https://data-api.binance.com", "https://api1.binance.com", "https://api2.binance.com", "https://api3.binance.com", "https://api.binance.com"] last_error = None for base_url in selected_endpoints: try: cb = source_breakers["binance"] if not cb.allow_request(): continue # Try next endpoint or fall through async def _do_fetch(): client = await GlobalHTTPClient.get_client() resp = await client.get(f"{base_url}{endpoint}", params=params, timeout=10.0) if resp.status_code == 451: logger.warning("[Binance] Endpoint %s blocked (451). Trying next...", base_url) raise RuntimeError("IP Blocked") if resp.status_code == 429: raise HTTPException(status_code=429, detail="Binance rate limit") resp.raise_for_status() cb.record_success() return resp.json() payload = await _retry(_do_fetch) # If successful, parse and return parsed = [ {"time": int(k[0])//1000, "open": k[1], "high": k[2], "low": k[3], "close": k[4], "volume": k[5]} for k in payload ] return _normalize_ohlcv(parsed, interval)[-limit:] except Exception as ex: last_error = ex logger.error("[Binance] Failed with %s: %s", base_url, ex) continue raise last_error or HTTPException(status_code=503, detail="Binance all endpoints failed") async def fetch_bybit(symbol: str, interval: str, limit: int) -> List[Dict[str, Any]]: """Bybit V5 kline endpoint — free, no API key.""" await _rate_limit("bybit") endpoint_symbol = 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), 1000), # Bybit V5 supports up to 1000 } logger.info("[Bybit] %s %s (cat=%s)", symbol, interval, bybit_cat) # B-3: Endpoint Rotation for Bybit bybit_endpoints = ["https://api.bybit.com", "https://api.bytick.com", "https://api.bybit.nl"] if settings.is_hf: # Prefer bytick on HF bybit_endpoints = ["https://api.bytick.com", "https://api.bybit.com", "https://api.bybit.nl"] last_error = None for base_url in bybit_endpoints: try: cb = source_breakers["bybit"] if not cb.allow_request(): continue async def _do_fetch(): client = await GlobalHTTPClient.get_client() url = f"{base_url}/v5/market/kline" headers = {} query_params = params.copy() if settings.bybit_api_key and settings.bybit_api_secret: timestamp = str(int(time.time() * 1000)) recv_window = "5000" sorted_params = "&".join([f"{k}={v}" for k, v in sorted(query_params.items())]) raw_str = timestamp + settings.bybit_api_key + recv_window + sorted_params signature = hmac.new(settings.bybit_api_secret.encode('utf-8'), raw_str.encode('utf-8'), hashlib.sha256).hexdigest() headers = { 'X-BAPI-API-KEY': settings.bybit_api_key, 'X-BAPI-TIMESTAMP': timestamp, 'X-BAPI-SIGN-TYPE': '2', 'X-BAPI-RECV-WINDOW': recv_window, 'X-BAPI-SIGN': signature } resp = await client.get(url, params=query_params, headers=headers, timeout=10.0) if resp.status_code == 403: logger.warning("[Bybit] Endpoint %s forbidden (403). Trying next...", base_url) raise RuntimeError("IP Blocked") resp.raise_for_status() cb.record_success() return resp.json() data = await _retry(_do_fetch) if data.get("retCode", -1) != 0: raise RuntimeError(f"Bybit Error: {data}") rows = data.get("result", {}).get("list", []) parsed = [ {"time": int(r[0])//1000, "open": r[1], "high": r[2], "low": r[3], "close": r[4], "volume": r[5]} for r in rows ] parsed.reverse() return _normalize_ohlcv(parsed, interval)[-limit:] except Exception as ex: last_error = ex logger.error("[Bybit] Failed with %s: %s", base_url, ex) continue raise last_error or HTTPException(status_code=503, detail="Bybit all endpoints failed") async def fetch_coingecko(symbol: str, interval: str, limit: int) -> List[Dict[str, Any]]: """ CoinGecko OHLCV — free, no API key (30 req/min). Days mapping: CoinGecko auto-selects granularity based on days requested. 1-2 days → hourly candles 3-90 days → daily candles (use for 1h/4h fallback) 90+ days → weekly candles """ await _rate_limit("coingecko") cg_id = 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]]: mappings = SYMBOLS[symbol].mappings if "finnhub" not in mappings: return [] cb = source_breakers.get("finnhub") if cb and not cb.allow_request(): return [] await _rate_limit("finnhub") client = await GlobalHTTPClient.get_client() async def _call(): r = await client.get( "https://finnhub.io/api/v1/stock/candle", params={ "symbol": mappings["finnhub"], "resolution": FINNHUB_RESOLUTION_MAP.get(interval, "D"), "count": limit, "token": FINNHUB_API_KEY, }, timeout=10, ) r.raise_for_status() return r.json() try: d = await _retry(_call) if d.get("s") != "ok": if cb: cb.record_success() return [] if cb: cb.record_success() return [ {"time": t, "open": o, "high": h, "low": l, "close": c, "volume": v} for t, o, h, l, c, v in zip(d["t"], d["o"], d["h"], d["l"], d["c"], d["v"]) ] except Exception as ex: if cb: cb.record_failure() logger.warning("[finnhub] %s failed: %s", symbol, ex) return [] def _resample_4h(df: pd.DataFrame) -> pd.DataFrame: if df.empty: return df out = df.resample("4h", label="left", closed="left").agg( {"Open": "first", "High": "max", "Low": "min", "Close": "last", "Volume": "sum"} ) return out.dropna(subset=["Open", "High", "Low", "Close"]) async def fetch_yfinance(symbol: str, interval: str, limit: int) -> List[Dict[str, Any]]: await _rate_limit("yfinance") ticker = 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: # B-5: Robust lookback (v6.0) - ensure EMA200 context p = "2y" if interval == "1d" else "60d" if "m" in interval else "1y" return yf.download(tickers=ticker, interval=yf_interval, period=p, progress=False, auto_adjust=False, threads=False) df = await asyncio.to_thread(_download) if df is None or df.empty: raise RuntimeError("yfinance: empty dataframe") if isinstance(df.columns, pd.MultiIndex): df.columns = df.columns.get_level_values(0) df.index = ( df.index.tz_localize("UTC") if df.index.tz is None else df.index.tz_convert("UTC") ) if interval == "4h": df = _resample_4h(df) parsed = [ {"time": int(idx.timestamp()), "open": row.get("Open"), "high": row.get("High"), "low": row.get("Low"), "close": row.get("Close"), "volume": row.get("Volume", 0)} for idx, row in df.iterrows() ] return _normalize_ohlcv(parsed, interval)[-limit:] async def fetch_alphavantage(symbol: str, interval: str, limit: int) -> List[Dict[str, Any]]: """Alpha Vantage - primarily for Macro/Historical. Needs API key.""" if not settings.alphavantage_api_key: raise ValueError("Alpha Vantage API key missing") await _rate_limit("alphavantage") endpoint_symbol = SYMBOLS[symbol].mappings.get("alphavantage", symbol) # AV has specific functions for daily/intraday func = "TIME_SERIES_INTRADAY" if interval in ["1m", "5m", "15m", "60min"] else "TIME_SERIES_DAILY" url = f"https://www.alphavantage.co/query?function={func}&symbol={endpoint_symbol}&apikey={settings.alphavantage_api_key}&outputsize=full" if "INTRADAY" in func: url += f"&interval={interval if interval != '1h' else '60min'}" async with httpx.AsyncClient(timeout=15) as client: resp = await client.get(url) data = resp.json() # Parse logic based on AV's weird nested JSON key = next((k for k in data.keys() if "Time Series" in k), None) if not key: raise RuntimeError(f"Alpha Vantage: {data.get('Note', data.get('Information', 'Unknown error'))}") series = data[key] parsed = [] for ts, vals in series.items(): parsed.append({ "time": int(datetime.strptime(ts, "%Y-%m-%d %H:%M:%S" if " " in ts else "%Y-%m-%d").replace(tzinfo=timezone.utc).timestamp()), "open": float(vals["1. open"]), "high": float(vals["2. high"]), "low": float(vals["3. low"]), "close": float(vals["4. close"]), "volume": float(vals.get("5. volume", 0)) }) parsed.sort(key=lambda x: x["time"]) return _normalize_ohlcv(parsed, interval)[-limit:] 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]: """ Fetch OHLCV data with fallback and caching. v6.0: Cache strategy optimized - always fetch max context, cache once. """ prefix = _cache_prefix(symbol, interval) key = f"hist_{prefix}" # BUG-P1-03: No limit in key to increase cache hits cached = historical_cache.get(key) if cached is not None: try: # v6.1: Cache stores (data, source) data_cached, source_cached = cached return data_cached[-limit:], source_cached except (ValueError, TypeError): # Handle old cache format gracefully historical_cache.delete(key) priority = _get_source_priority(symbol) errors: List[str] = [] # Fetch more than needed to ensure indicators have enough context (v6.0) fetch_limit = max(limit, 1000) for source in priority: try: if source == "binance": data = await fetch_binance(symbol, interval, fetch_limit) elif source == "bybit": data = await fetch_bybit(symbol, interval, fetch_limit) elif source == "coingecko": data = await fetch_coingecko(symbol, interval, fetch_limit) elif source == "twelvedata": data = await fetch_twelvedata(symbol, interval, fetch_limit) elif source == "finnhub": data = await fetch_finnhub(symbol, interval, fetch_limit) elif source == "yfinance": data = await fetch_yfinance(symbol, interval, fetch_limit) elif source == "alphavantage": data = await fetch_alphavantage(symbol, interval, fetch_limit) else: continue if len(data) >= 20: historical_cache.set(key, (data, source), ttl_seconds=interval_ttl(interval)) return data[-limit:], 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] priority = _get_source_priority(symbol) for source in priority: try: if source == "binance": await _rate_limit("binance") base_url = "https://fapi.binance.com" if cfg.binance_type == "futures" else "https://api.binance.com" endpoint = "/fapi/v1/ticker/24hr" if cfg.binance_type == "futures" else "/api/v3/ticker/24hr" async with httpx.AsyncClient(timeout=10) as c: r = await c.get(f"{base_url}{endpoint}", params={"symbol": cfg.mappings["binance"]}) d = r.json() # Futures and Spot use slightly different field names in some cases, but 24hr is mostly consistent res = { "price": float(d.get("lastPrice") or d.get("price") or 0), "change": float(d.get("priceChange", 0)), "change_pct": float(d.get("priceChangePercent", 0)), "high_24h": float(d.get("highPrice", 0)), "low_24h": float(d.get("lowPrice", 0)), "volume_24h": float(d.get("volume", 0)), "source": "binance" } elif source == "twelvedata": await _rate_limit("twelvedata") async with httpx.AsyncClient(timeout=10) as c: r = await c.get("https://api.twelvedata.com/quote", params={"symbol": cfg.mappings["twelvedata"], "apikey": TWELVEDATA_API_KEY}) d = r.json() if "close" not in d and "price" not in d: continue p = float(d.get("price") or d.get("close") or 0) pc = float(d.get("previous_close") or p) res = { "price": p, "change": p - pc, "change_pct": ((p - pc)/pc*100) if pc!=0 else 0, "high_24h": float(d.get("high") or p), "low_24h": float(d.get("low") or p), "volume_24h": float(d.get("volume") or 0), "source": "twelvedata" } elif source == "bybit": await _rate_limit("bybit") async with httpx.AsyncClient(timeout=10) as c: url = "https://api.bybit.com/v5/market/tickers" query_params = {"category": cfg.bybit_category, "symbol": cfg.mappings["bybit"]} headers = {} if settings.bybit_api_key and settings.bybit_api_secret: timestamp = str(int(time.time() * 1000)) recv_window = "5000" sorted_params = "&".join([f"{k}={v}" for k, v in sorted(query_params.items())]) raw_str = timestamp + settings.bybit_api_key + recv_window + sorted_params signature = hmac.new(settings.bybit_api_secret.encode('utf-8'), raw_str.encode('utf-8'), hashlib.sha256).hexdigest() headers = { 'X-BAPI-API-KEY': settings.bybit_api_key, 'X-BAPI-TIMESTAMP': timestamp, 'X-BAPI-SIGN-TYPE': '2', 'X-BAPI-RECV-WINDOW': recv_window, 'X-BAPI-SIGN': signature } r = await c.get(url, params=query_params, headers=headers) d = r.json() if d.get("retCode") == 0 and d.get("result", {}).get("list"): t = d["result"]["list"][0] res = { "price": float(t["lastPrice"]), "change": float(t["price24hPcnt"]) * float(t["lastPrice"]) / 100, "change_pct": float(t["price24hPcnt"]) * 100, "high_24h": float(t["highPrice24h"]), "low_24h": float(t["lowPrice24h"]), "volume_24h": float(t["volume24h"]), "source": "bybit" } elif source == "yfinance": def _yf_info(): t = yf.Ticker(cfg.mappings["yfinance"]) h = t.history(period="5d") if h.empty: return None lp, pc = float(h["Close"].iloc[-1]), float(h["Close"].iloc[-2]) if len(h)>1 else float(h["Close"].iloc[-1]) return { "price": lp, "change": lp - pc, "change_pct": ((lp - pc)/pc*100) if pc!=0 else 0, "high_24h": float(h["High"].iloc[-1]), "low_24h": float(h["Low"].iloc[-1]), "volume_24h": float(h["Volume"].iloc[-1]), "source": "yfinance" } res = await asyncio.to_thread(_yf_info) if not res: continue else: continue res.update({"symbol": symbol, "timestamp": int(time.time())}) ttl = 5 if cfg.category == "Crypto" else (10 if cfg.category in ("Cặp tiền", "Chỉ số") else 30) ticker_cache.set(f"ticker:{symbol}", res, ttl_seconds=ttl) return res except Exception as ex: logger.debug("[ticker] %s/%s failed: %s", symbol, source, ex) continue # Ultimate Fallback raise HTTPException(status_code=502, detail=f"Ticker failed for {symbol} after trying {priority}") # ────────────────────────────────────────────────────────────────────────────── # Technical Indicators # ────────────────────────────────────────────────────────────────────────────── # D-2: High-Performance Vectorized Indicators (NumPy) def _ema(arr: np.ndarray, period: int) -> np.ndarray: """Vectorized EMA using NumPy (replaces loops).""" if len(arr) == 0: return np.array([], dtype=float) alpha = 2.0 / (period + 1.0) # Use pandas ewm for robust vectorized calculation (v6.0) return pd.Series(arr).ewm(alpha=alpha, adjust=False).mean().values def _rsi(close: np.ndarray, period: int = 14) -> np.ndarray: """Vectorized RSI using NumPy/Pandas.""" delta = np.diff(close) gain = np.where(delta > 0, delta, 0.0) loss = np.where(delta < 0, -delta, 0.0) avg_gain = pd.Series(gain).ewm(alpha=1.0/period, adjust=False).mean() avg_loss = pd.Series(loss).ewm(alpha=1.0/period, adjust=False).mean() rs = avg_gain / avg_loss.replace(0, np.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 _macd(close: np.ndarray, fast=12, slow=26, signal=9 ) -> Tuple[np.ndarray, np.ndarray, np.ndarray]: ema_fast = _ema(close, fast) ema_slow = _ema(close, slow) macd_line = ema_fast - ema_slow sig_line = _ema(np.where(np.isnan(macd_line), 0, macd_line), signal) histogram = macd_line - sig_line return macd_line, sig_line, histogram def _atr(high: np.ndarray, low: np.ndarray, close: np.ndarray, period=14) -> np.ndarray: """Vectorized ATR.""" tr = np.maximum(high[1:] - low[1:], np.maximum(np.abs(high[1:] - close[:-1]), np.abs(low[1:] - close[:-1]))) tr = np.concatenate([[np.nan], tr]) return pd.Series(tr).ewm(alpha=1.0/period, adjust=False).mean().values def _stoch_rsi(close: np.ndarray, rsi_period=14, stoch_period=14, smooth_k=3, smooth_d=3) -> Tuple[np.ndarray, np.ndarray]: """Vectorized Stochastic RSI.""" rsi_vals = pd.Series(_rsi(close, rsi_period)) roll_min = rsi_vals.rolling(window=stoch_period).min() roll_max = rsi_vals.rolling(window=stoch_period).max() k = 100 * (rsi_vals - roll_min) / (roll_max - roll_min).replace(0, np.inf) k_smooth = k.rolling(window=smooth_k).mean() d_smooth = k_smooth.rolling(window=smooth_d).mean() return k_smooth.values, d_smooth.values def _sma(arr: np.ndarray, period: int) -> np.ndarray: """Vectorized Simple Moving Average.""" if len(arr) == 0: return np.array([], dtype=float) return pd.Series(arr).rolling(window=period).mean().values def _cci(high: np.ndarray, low: np.ndarray, close: np.ndarray, period: int = 20) -> np.ndarray: """Vectorized Commodity Channel Index.""" tp = (high + low + close) / 3.0 tp_series = pd.Series(tp) sma = tp_series.rolling(window=period).mean() # Optimized MAD calculation (Vectorized) mad = tp_series.rolling(window=period).apply(lambda x: np.abs(x - x.mean()).mean(), raw=True) return ((tp_series - sma) / (0.015 * mad.replace(0, np.inf))).values def _adx(high: np.ndarray, low: np.ndarray, close: np.ndarray, period: int = 14) -> tuple: """Vectorized Average Directional Index (v6.0).""" plus_dm = np.where((high[1:] - high[:-1] > low[:-1] - low[1:]) & (high[1:] - high[:-1] > 0), high[1:] - high[:-1], 0.0) minus_dm = np.where((low[:-1] - low[1:] > high[1:] - high[:-1]) & (low[:-1] - low[1:] > 0), low[:-1] - low[1:], 0.0) tr = np.maximum(high[1:] - low[1:], np.maximum(np.abs(high[1:] - close[:-1]), np.abs(low[1:] - close[:-1]))) # Pad first index plus_dm = np.concatenate([[0.0], plus_dm]) minus_dm = np.concatenate([[0.0], minus_dm]) tr = np.concatenate([[0.0], tr]) tr_sum = pd.Series(tr).ewm(alpha=1.0/period, adjust=False).mean() plus_di = 100 * pd.Series(plus_dm).ewm(alpha=1.0/period, adjust=False).mean() / tr_sum.replace(0, np.inf) minus_di = 100 * pd.Series(minus_dm).ewm(alpha=1.0/period, adjust=False).mean() / tr_sum.replace(0, np.inf) dx = 100 * np.abs(plus_di - minus_di) / (plus_di + minus_di).replace(0, np.inf) adx = dx.ewm(alpha=1.0/period, adjust=False).mean() return adx.values, plus_di.values, minus_di.values def _awesome_oscillator(high: np.ndarray, low: np.ndarray) -> np.ndarray: """Awesome Oscillator = SMA(5, median) - SMA(34, median).""" median = (high + low) / 2.0 sma5 = _sma(median, 5) sma34 = _sma(median, 34) return sma5 - sma34 def _momentum(close: np.ndarray, period: int = 10) -> np.ndarray: """Vectorized Momentum: close - close[n periods ago].""" return pd.Series(close).diff(period).values def _williams_r(high: np.ndarray, low: np.ndarray, close: np.ndarray, period: int = 14) -> np.ndarray: """Vectorized Williams %R.""" s = pd.Series(close) hh = pd.Series(high).rolling(window=period).max() ll = pd.Series(low).rolling(window=period).min() wr = -100 * (hh - s) / (hh - ll).replace(0, np.inf) return wr.values def _bull_bear_power(high: np.ndarray, low: np.ndarray, close: np.ndarray, period: int = 13) -> np.ndarray: """Vectorized Bull Bear Power (v6.0).""" ema_val = _ema(close, period) return (high - ema_val) + (low - ema_val) def _ultimate_oscillator(high: np.ndarray, low: np.ndarray, close: np.ndarray, p1: int = 7, p2: int = 14, p3: int = 28) -> np.ndarray: """Vectorized Ultimate Oscillator (v6.0).""" n = len(close) if n < p3 + 1: return np.full(n, np.nan) prev_close = pd.Series(close).shift(1) tl = np.minimum(low, prev_close.values) th = np.maximum(high, prev_close.values) bp = pd.Series(close - tl) tr = pd.Series(th - tl) def _avg(p): return bp.rolling(p).sum() / tr.rolling(p).sum().replace(0, np.inf) avg1 = _avg(p1) avg2 = _avg(p2) avg3 = _avg(p3) uo = 100 * (4 * avg1 + 2 * avg2 + avg3) / 7.0 return uo.values def _ichimoku_base(high: np.ndarray, low: np.ndarray, period: int = 26) -> np.ndarray: """Vectorized Ichimoku Base Line.""" hh = pd.Series(high).rolling(window=period).max() ll = pd.Series(low).rolling(window=period).min() return ((hh + ll) / 2.0).values def _vwma(close: np.ndarray, volume: np.ndarray, period: int = 20) -> np.ndarray: """Vectorized Volume Weighted Moving Average.""" cv = pd.Series(close * volume) v = pd.Series(volume) return (cv.rolling(period).sum() / v.rolling(period).sum().replace(0, np.inf)).values def _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 (Vectorized v6.0) vol_sma = _sma(vols, 20) last_close = closes[-1] last_atr = _last(atr14) or 0 return { "ema": { "ema9": _last(ema9), "ema21": _last(ema21), "ema50": _last(ema50), "ema200": _last(ema200), }, "rsi": { "value": _last(rsi14), "signal": ( "overbought" if (_last(rsi14) or 50) > 70 else "oversold" if (_last(rsi14) or 50) < 30 else "neutral" ), }, "macd": { "macd": _last(macd_l), "signal": _last(macd_s), "histogram": _last(macd_h), "cross": ( "bullish" if (_last(macd_h) or 0) > 0 else "bearish" if (_last(macd_h) or 0) < 0 else "neutral" ), }, "bollinger": { "upper": _last(bb_u), "middle": _last(bb_m), "lower": _last(bb_l), "bandwidth": round(((_last(bb_u) or 0) - (_last(bb_l) or 0)) / ((_last(bb_m) or 1)), 6), }, "atr": { "value": _last(atr14), "pct": round(last_atr / last_close * 100, 4) if last_close else None, }, "stoch_rsi": { "k": _last(stoch_k), "d": _last(stoch_d), "signal": ( "overbought" if (_last(stoch_k) or 50) > 80 else "oversold" if (_last(stoch_k) or 50) < 20 else "neutral" ), }, "volume": { "last": round(float(vols[-1]), 2), "sma20": round(float(vol_sma[-1]), 2), "above_avg": bool(vols[-1] > vol_sma[-1]), }, "trend": { "ema_bullish_stack": bool( all(x is not None for x in [_last(ema9),_last(ema21),_last(ema50)]) and _last(ema9) > _last(ema21) > _last(ema50) # type: ignore ), "above_ema200": bool(_last(ema200) is not None and last_close > (_last(ema200) or 0)), "close": round(float(last_close), 8), "short_momentum_ref": float(closes[max(0, len(closes)-6)]) if len(closes) else float(last_close), }, "short_momentum_ref": float(closes[max(0, len(closes)-6)]) if len(closes) else float(last_close), "series": { "ema9": _series(ema9), "ema21": _series(ema21), "ema50": _series(ema50), "bb_upper": _series(bb_u), "bb_mid": _series(bb_m), "bb_lower": _series(bb_l), }, } # ───────────────────────────────────────────────────────────────────────────── # UTILITIES # ───────────────────────────────────────────────────────────────────────────── def _clamp(v: float, lo: float, hi: float) -> float: return max(lo, min(hi, v)) def _pct(a: float, b: float) -> float: """(a - b) / b * 100, an toàn với b = 0.""" return (a - b) / b * 100.0 if b != 0 else 0.0 def _safe(v: Optional[float], default: float = 0.0) -> float: if v is None or math.isnan(v) or math.isinf(v): return default return float(v) def _round(v: float, d: int = 6) -> float: return round(float(v), d) def _build_anchor_forecast( data: List[Dict[str, Any]], indicators: Dict[str, Any], horizon: int, interval: str, ) -> Dict[str, np.ndarray]: """Fallback statistical forecast to cross-verify AI model stability.""" closes = np.array([float(d["close"]) for d in data], dtype=float) if len(closes) < 3: last_close = closes[-1] flat = np.full(horizon, last_close, dtype=float) return {"p10": flat.copy(), "p50": flat.copy(), "p90": flat.copy()} last_close = float(closes[-1]) ema21 = float(indicators["ema"].get("ema21") or last_close) ema50 = float(indicators["ema"].get("ema50") or last_close) rsi = float(indicators["rsi"].get("value") or 50.0) atr = float(indicators["atr"].get("value") or max(last_close * 0.006, 1.0)) band = float(indicators["bollinger"].get("bandwidth") or 0.02) returns = pd.Series(closes).pct_change().dropna().tail(96) fast_drift = float(returns.tail(min(12, len(returns))).mean()) if len(returns) else 0.0 slow_drift = float(returns.mean()) if len(returns) else 0.0 vol = float(returns.std()) if len(returns) > 1 else 0.0 ema_trend = ((last_close / ema21) - 1.0) * 0.30 + ((ema21 / ema50) - 1.0) * 0.25 mean_revert = -((last_close - ema21) / last_close) * 0.18 if last_close else 0.0 rsi_bias = 0.0018 if rsi < 35 else (-0.0018 if rsi > 65 else 0.0) volatility_drag = -vol * 0.35 step_return = _clamp( fast_drift * 0.55 + slow_drift * 0.20 + ema_trend + mean_revert + rsi_bias + volatility_drag, -0.03, 0.03, ) anchor_p50: List[float] = [] price = last_close for i in range(horizon): decay = max(0.35, 1.0 - (i / max(horizon * 1.6, 1))) price = price * (1.0 + step_return * decay) anchor_p50.append(price) step_scale = math.sqrt(max(STEP_SECONDS[interval], 60) / 86400.0) base_spread = max(atr * 0.70, last_close * max(vol * step_scale * 1.8, band * 0.22, 0.0035)) p50 = np.array(anchor_p50, dtype=float) p10 = np.array([max(0.0, x - base_spread * math.sqrt(i + 1)) for i, x in enumerate(anchor_p50)], dtype=float) p90 = np.array([x + base_spread * math.sqrt(i + 1) for i, x in enumerate(anchor_p50)], dtype=float) return {"p10": p10, "p50": p50, "p90": p90} def _blend_forecasts( last_close: float, model_output: Dict[str, Any], anchor_output: Dict[str, np.ndarray], indicators: Dict[str, Any], ) -> Dict[str, Any]: """Ensemble blending of AI model output and statistical anchor.""" raw_p10 = np.array(model_output["p10"], dtype=float) raw_p50 = np.array(model_output["p50"], dtype=float) raw_p90 = np.array(model_output["p90"], dtype=float) # A-6: Robust scale calculation using median of first 3 model predictions first_model_est = float(np.median(raw_p50[:3])) if len(raw_p50) >= 3 else (float(raw_p50[0]) if len(raw_p50) else last_close) scale = (last_close / first_model_est) if abs(first_model_est) > 1e-8 else 1.0 # Clip scale to [0.5, 2.0] to prevent extreme corrections on low-price coins (BUG-06) last_close_magnitude = math.floor(math.log10(max(abs(last_close), 1e-10))) clip_lo = 0.3 if last_close_magnitude < -4 else 0.5 clip_hi = 3.0 if last_close_magnitude < -4 else 2.0 scale = float(np.clip(scale, clip_lo, clip_hi)) if abs(scale - 1.0) > 0.3: logger.warning("[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) # B-2: Advanced weighting sensitive to RSI extremes (v6.0) rsi_val = indicators["rsi"].get("value") or 50.0 rsi_extreme = abs(rsi_val - 50.0) > 20.0 # <30 or >70 band_width_pct = abs(model_p90[-1] - model_p10[-1]) / last_close if last_close else 0.0 # If RSI is extreme, favor the AI model which better handles mean-reversion base_weight = 0.72 if agreement else 0.55 if rsi_extreme: base_weight += 0.08 model_weight = _clamp(base_weight - max(0.0, band_width_pct - 0.08) * 2.0, 0.30, 0.88) anchor_weight = 1.0 - model_weight blend_p10 = model_p10 * model_weight + anchor_p10 * anchor_weight blend_p50 = model_p50 * model_weight + anchor_p50 * anchor_weight blend_p90 = model_p90 * model_weight + anchor_p90 * anchor_weight bias_pct = abs((scale - 1.0) * 100.0) # Refined confidence score confidence = 60.0 confidence += 15.0 if agreement else -10.0 # Trend alignment (Bull/Bear stack) if indicators["trend"].get("ema_bullish_stack") and model_dir > 0: confidence += 10.0 elif not indicators["trend"].get("ema_bullish_stack") and model_dir < 0: # Bearish stack + down forecast confidence += 5.0 # Penalty for high uncertainty (wide bands) confidence -= min(25.0, band_width_pct * 150.0) # Penalty for extreme bias/scaling corrections confidence -= min(15.0, bias_pct * 0.5) # Penalty for low volume relative to average if not indicators["volume"].get("above_avg"): confidence -= 5.0 confidence = _clamp(confidence, 10.0, 95.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]) # Use previous COMPLETED candle for pivot calculation (BUG-P1-10) if len(highs) >= 2: last_h = float(highs[-2]) last_l = float(lows[-2]) last_c = float(closes[-2]) else: last_h, last_l, last_c = float(highs[-1]), float(lows[-1]), float(closes[-1]) pivots = _calc_pivot_points(last_h, last_l, last_c) # 1. Detect Pivot Highs and Lows (window=3) pivots_h = [] pivots_l = [] for i in range(3, len(highs)-3): if highs[i] == max(highs[i-3:i+4]): pivots_h.append(highs[i]) if lows[i] == min(lows[i-3:i+4]): pivots_l.append(lows[i]) # Fallback if no pivots if not pivots_h: pivots_h = [max(highs[-20:])] if not pivots_l: pivots_l = [min(lows[-20:])] # 2. Select nearest support and resistance ns = max([p for p in pivots_l if p < last_c], default=last_c - atr) nr = min([p for p in pivots_h if p > last_c], default=last_c + atr) return { "nearest_support": _round(ns, 6), "nearest_resistance":_round(nr, 6), "key_support": _round(ns - atr, 6), "key_resistance": _round(nr + atr, 6), "level_list": [ {"price": ns, "type": "support", "strength": 3, "label": f"Hỗ trợ {ns:.5g}"}, {"price": nr, "type": "resistance", "strength": 3, "label": f"Kháng cự {nr:.5g}"}, ] } def _classify_regime(indicators: Dict[str, Any], last_close: float, atr_pct: float) -> Tuple[str, str]: """Phân loại trạng thái thị trường dựa trên biến động và xu hướng.""" rsi = _safe(indicators["rsi"].get("value"), 50.0) ema9 = _safe(indicators["ema"].get("ema9"), last_close) ema50 = _safe(indicators["ema"].get("ema50"), last_close) ema200 = _safe(indicators["ema"].get("ema200"), last_close) # Logic 8 trạng thái if atr_pct > 3.5: if rsi > 60: return "volatile_bull", "Tăng trưởng biến động cao" if rsi < 40: return "volatile_bear", "Sụt giảm biến động cao" return "high_chaos", "Thị trường hỗn loạn / Vol cao" if last_close > ema200 and ema9 > ema50: if rsi > 70: return "overextended_bull", "Tăng trưởng quá mức (Overbought)" return "stable_bull", "Xu hướng tăng ổn định" if last_close < ema200 and ema9 < ema50: if rsi < 30: return "capitulation", "Hoảng loạn / Quá bán (Oversold)" return "stable_bear", "Xu hướng giảm ổn định" if abs(_pct(ema9, ema50)) < 0.5: return "tight_range", "Tích lũy biên độ hẹp (Squeeze)" return "sideways", "Thị trường đi ngang (Sideway)" def _calc_momentum(data: List[Dict[str, Any]]) -> Dict[str, Any]: """Tính toán momentum đa khung thời gian.""" if len(data) < 50: return {"short": 0, "mid": 0, "long": 0, "aligned": False} closes = [float(d["close"]) for d in data] last = closes[-1] short = _pct(last, closes[max(0, len(closes)-6)]) mid = _pct(last, closes[max(0, len(closes)-21)]) long = _pct(last, closes[max(0, len(closes)-51)]) aligned = (short > 0 and mid > 0 and long > 0) or (short < 0 and mid < 0 and long < 0) return { "short": round(short, 4), "mid": round(mid, 4), "long": round(long, 4), "short_lbl": f"{'Tăng' if short>0 else 'Giảm'} {abs(short):.2f}%", "aligned": aligned } @dataclass class ConfluenceResult: raw_score: float confluence: float bias: str conviction: str grade: str active_signals: int bull_count: int bear_count: int neutral_count: int def _calc_confluence(signals: List[Signal]) -> ConfluenceResult: tw = sum(s.weight for s in signals) ws = sum(s.value * s.weight for s in signals) raw = ws / tw if tw > 0 else 0.0 conf = _clamp(50.0 + raw * 50.0, 0.0, 100.0) bias = "bullish" if conf >= 60 else "bearish" if conf <= 40 else "neutral" return ConfluenceResult( raw_score=raw, confluence=conf, bias=bias, conviction="strong" if abs(conf-50)>20 else "moderate", grade="A" if abs(conf-50)>25 else "B", active_signals=len(signals), bull_count=sum(1 for s in signals if s.value > 0.2), bear_count=sum(1 for s in signals if s.value < -0.2), neutral_count=sum(1 for s in signals if abs(s.value) <= 0.2) ) @dataclass class TradeSetup: tier: str direction: str entry_low: float entry_high: float entry_mid: float stop_loss: float take_profit1: float take_profit2: float take_profit3: float risk_reward: float risk_pct: float valid: bool notes: str def _build_setups(bias: str, last_close: float, atr: float, levels: Dict[str, Any], projected: float) -> List[TradeSetup]: """Tạo các phương án giao dịch theo 3 cấp độ: Conservative, Standard, Aggressive.""" setups = [] ns = levels.get("nearest_support", last_close - atr) nr = levels.get("nearest_resistance", last_close + atr) ks = levels.get("key_support", ns - atr) kr = levels.get("key_resistance", nr + atr) if bias == "bullish": # 1. Conservative (Thận trọng - Chờ hồi) sl_c = ks - atr * 0.3 tp_c = nr setups.append(TradeSetup( tier="conservative", direction="long", entry_low=ks, entry_high=ns, entry_mid=(ks+ns)/2, stop_loss=sl_c, take_profit1=last_close + (last_close-sl_c), take_profit2=tp_c, take_profit3=tp_c + atr, risk_reward=_round((tp_c - ns) / (ns - sl_c), 2) if ns > sl_c else 0.0, risk_pct=_round((ns - sl_c) / ns * 100, 2), valid=True, notes="Đợi giá kiểm tra lại vùng hỗ trợ cứng trước khi tham gia." )) # 2. Standard (Tiêu chuẩn - Theo xu hướng) sl_s = ns - atr * 0.5 tp_s = projected setups.append(TradeSetup( tier="standard", direction="long", entry_low=ns, entry_high=last_close, entry_mid=(ns+last_close)/2, stop_loss=sl_s, take_profit1=last_close + atr, take_profit2=projected, take_profit3=max(projected, kr), risk_reward=_round((projected - last_close) / (last_close - sl_s), 2) if last_close > sl_s else 0.0, risk_pct=_round((last_close - sl_s) / last_close * 100, 2), valid=True, notes="Giao dịch theo đà tăng hiện tại, dừng lỗ dưới hỗ trợ gần nhất." )) # 3. Aggressive (Săn đuổi - Buy Stop / Breakout) sl_a = last_close - atr * 0.8 tp_a = max(projected, kr) + atr setups.append(TradeSetup( tier="aggressive", direction="long", entry_low=last_close, entry_high=last_close + atr*0.2, entry_mid=last_close+atr*0.1, stop_loss=sl_a, take_profit1=last_close + atr*1.5, take_profit2=tp_a - atr, take_profit3=tp_a, risk_reward=_round((tp_a - last_close) / (last_close - sl_a), 2) if last_close > sl_a else 0.0, risk_pct=_round((last_close - sl_a) / last_close * 100, 2), valid=True, notes="Vào lệnh trực tiếp để bắt kịp sóng tăng mạnh. Rủi ro cao hơn." )) elif bias == "bearish": # 1. Conservative (Thận trọng - Chờ hồi) sl_c = kr + atr * 0.3 tp_c = ns setups.append(TradeSetup( tier="conservative", direction="short", entry_low=nr, entry_high=kr, entry_mid=(nr+kr)/2, stop_loss=sl_c, take_profit1=last_close - (sl_c-last_close), take_profit2=tp_c, take_profit3=tp_c - atr, risk_reward=_round((nr - tp_c) / (sl_c - nr), 2) if sl_c > nr else 0.0, risk_pct=_round((sl_c - nr) / nr * 100, 2), valid=True, notes="Chờ giá hồi lên vùng kháng cự mạnh để tối ưu RR." )) # Standard Short sl_s = nr + atr * 0.5 tp_s = projected setups.append(TradeSetup( tier="standard", direction="short", entry_low=last_close, entry_high=nr, entry_mid=(last_close+nr)/2, stop_loss=sl_s, take_profit1=last_close - atr, take_profit2=projected, take_profit3=min(projected, ks), risk_reward=_round((last_close - projected) / (sl_s - last_close), 2) if sl_s > last_close else 0.0, risk_pct=_round((sl_s - last_close) / last_close * 100, 2), valid=True, notes="Bán theo xu hướng giảm, dừng lỗ trên kháng cự gần nhất." )) return [s for s in setups if s.valid] def _build_scenarios( bias: str, last_close: float, atr: float, levels: Dict[str, Any], forecast_rows: List[Dict[str, Any]], confluence: float, momentum: Dict[str, Any] ) -> List[Dict[str, Any]]: """Phân tích các kịch bản có thể xảy ra.""" fc_p50 = float(forecast_rows[-1]["p50"]) fc_p90 = float(forecast_rows[-1]["p90"]) fc_p10 = float(forecast_rows[-1]["p10"]) # Kịch bản cơ sở (P50) base = { "name": "base", "label": "Kịch bản Cơ sở", "probability": round(confluence, 1), "price_target": _round(fc_p50, 6), "return_pct": _round(_pct(fc_p50, last_close), 2), "catalyst": "Duy trì xu hướng hiện tại", "invalidation": "Phá vỡ vùng EMA21" } # BUG-10: Normalize probabilities to ensure sum = 100% # Using raw scores from confluence and bias bull_raw = (100 - confluence) * 0.4 + (20 if bias == 'bullish' else 0) base_raw = confluence bear_raw = max(5.0, 100.0 - base_raw - bull_raw) # Min 5% for bear in base calc # Normalize total = bull_raw + base_raw + bear_raw bull_prob = (bull_raw / total) * 100 base_prob = (base_raw / total) * 100 bear_prob = 100.0 - bull_prob - base_prob # Exact sum base["probability"] = round(base_prob, 1) bull = { "name": "bull", "label": "Kịch bản Tích cực", "probability": round(bull_prob, 1), "price_target": _round(fc_p90, 6), "return_pct": _round(_pct(fc_p90, last_close), 2), "catalyst": "Đột phá Momentum mạnh", "invalidation": "RSI quay đầu giảm" } bear = { "name": "bear", "label": "Kịch bản Tiêu cực", "probability": round(bear_prob, 1), "price_target": _round(fc_p10, 6), "return_pct": _round(_pct(fc_p10, last_close), 2), "catalyst": "Đảo chiều bất ngờ / Tin xấu", "invalidation": "Hỗ trợ cứng được giữ vững" } return [base, bull, bear] def _build_risk_framework( bias: str, last_close: float, atr: float, atr_pct: float, levels: Dict[str, Any], setups: List[TradeSetup], interval: str ) -> Dict[str, Any]: """Cung cấp hướng dẫn quản trị rủi ro.""" vol_adjusted_risk = _clamp(2.0 - (atr_pct / 5.0), 0.5, 1.5) advice = "Kích thước vị thế tiêu chuẩn (1% tài khoản)." if atr_pct > 4.0: advice = "Thị trường biến động mạnh: Giảm 50% khối lượng lệnh." elif atr_pct < 0.5: advice = "Biến động thấp: Cân nhắc chốt lời ngắn (Scalp)." return { "max_risk_pct": round(vol_adjusted_risk, 2), "position_size_advice": advice, "stop_loss_type": "Volatility-based (ATR)", "invalidation_point": levels.get("key_resistance" if bias=="bearish" else "key_support", last_close) } def _build_reasoning( bias: str, conviction: str, regime_key: str, regime_lbl: str, signals: List[Signal], indicators: Dict[str, Any], momentum: Dict[str, Any], levels: Dict[str, Any], last_close: float, atr_pct: float, interval: str, forecast_ret: float ) -> Tuple[List[str], List[str], List[str]]: """Tự động sinh các lý do, cảnh báo và cơ hội bằng tiếng Việt chuyên sâu.""" reasons, warnings, opportunities = [], [], [] # Logic Lý do if bias == "bullish": reasons.append(f"Xu hướng chủ đạo là TĂNG ({conviction}) trên khung {interval}.") reasons.append(f"Dự báo AI cho thấy tiềm năng tăng trưởng {forecast_ret:+.2f}% trong ngắn hạn.") if indicators["trend"].get("ema_bullish_stack"): reasons.append("Hệ thống EMA đang xếp chồng Bullish mạnh mẽ, xác nhận lực mua áp ảo.") elif bias == "bearish": reasons.append(f"Áp lực GIẢM giá chiếm ưu thế ({conviction}) trên khung {interval}.") reasons.append(f"AI nhận diện tín hiệu suy yếu với mục tiêu giảm về vùng {forecast_ret:+.2f}%.") else: reasons.append(f"Thị trường đang trong trạng thái TÍCH LŨY / ĐI NGANG trên khung {interval}.") reasons.append("Chưa có tín hiệu bứt phá rõ rệt từ các chỉ báo kỹ thuật quan trọng.") if regime_lbl: reasons.append(f"Trạng thái thị trường hiện tại: {regime_lbl}.") # Logic Cảnh báo rsi = _safe(indicators["rsi"].get("value"), 50.0) if rsi > 70: warnings.append("Chỉ số RSI đi vào vùng quá mua (>70), rủi ro đảo chiều kỹ thuật cao.") if atr_pct > 3.0: warnings.append("Biến động thị trường đang ở mức cao (ATR), ưu tiên quản lý vốn chặt chẽ.") # Logic Cơ hội if bias == "bullish": opportunities.append(f"Cơ hội Long khi giá điều chỉnh về vùng hỗ trợ {levels.get('nearest_support')}.") else: opportunities.append(f"Cơ hội Short khi giá hồi phục chạm kháng cự {levels.get('nearest_resistance')}.") return reasons, warnings, opportunities 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): if isinstance(arr, pd.Series): arr = arr.values v = arr[-1] if len(arr) else float('nan') return None if (v is None or (isinstance(v, float) and 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 # Handle scalar, Series or ndarray if isinstance(val, (np.ndarray, pd.Series, list)): v = _lv(val) else: v = 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 # Derive bias from signal if total_buy > total_sell + 3: summary_bias = "bullish" elif total_sell > total_buy + 3: summary_bias = "bearish" else: summary_bias = "neutral" # B-8: Integrated Summary Signals (v6.0) # Combine TV style with ensemble confidence if total_buy > total_sell + 6 and confidence > 65: total_signal = "Mua mạnh (Cực độ)" elif total_buy > total_sell + 3 and confidence > 55: total_signal = "Mua" elif total_sell > total_buy + 6 and confidence > 65: total_signal = "Bán mạnh (Cực độ)" elif total_sell > total_buy + 3 and confidence > 55: total_signal = "Bán" else: total_signal = "Trung lập (Thận trọng)" # B-4: Use last COMPLETED candle for pivot calculation to avoid flickering last_h = float(highs[-2]) if len(highs) > 1 else float(highs[-1]) last_l = float(lows[-2]) if len(lows) > 1 else float(lows[-1]) last_c = float(closes[-2]) if len(closes) > 1 else float(closes[-1]) pivots = _calc_pivot_points(last_h, last_l, last_c) return { "style": "tradingview", "summary": { "sell": total_sell, "neutral": total_neutral, "buy": total_buy, "signal": total_signal, "bias": summary_bias, }, "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, } START_TIME = time.time() async def _background_cleanup(): """Evict expired entries from TTL caches periodically (v6.0: Aggressive).""" while True: await asyncio.sleep(120) # Every 2 minutes try: h = historical_cache.evict_expired() f = forecast_cache.evict_expired() t = ticker_cache.evict_expired() persistent_cache.evict() logger.info("[cleanup] Evicted h=%d f=%d t=%d", h, f, t) except Exception as ex: logger.warning("[cleanup] Error: %s", ex) async def _periodic_health_check(): """Check circuit breakers and log source health every 2 minutes.""" while True: await asyncio.sleep(120) for name, cb in source_breakers.items(): if cb.state == "OPEN": logger.warning("[health] CB OPEN: %s (failures=%d)", name, cb.failures) def _clear_ip_limits(): """Remove stale IP rate-limit entries older than 60 seconds.""" now = time.time() stale = [ip for ip, ts_list in IP_LIMITS.items() if not any(now - t < 60 for t in ts_list)] for ip in stale: del IP_LIMITS[ip] if stale: logger.debug("[ip-cleanup] Removed %d stale IPs", len(stale)) @asynccontextmanager async def lifespan(app: FastAPI): # Startup logic logger.info("Starting Kronos AI Backend v6.0...") # Async background tasks asyncio.create_task(persistent_cache.start_writer()) # P0: Async Persistence asyncio.create_task(ws_manager.heartbeat()) asyncio.create_task(_background_cleanup()) asyncio.create_task(_periodic_health_check()) # F-2: Rate limit cleanup task async def _ip_cleanup_loop(): while True: await asyncio.sleep(300) # BUG-P1-07: reduced from 3600 to 300 _clear_ip_limits() asyncio.create_task(_ip_cleanup_loop()) # 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.") # ────────────────────────────────────────────────────────────────────────────── # FastAPI App Instance (v6.0) # ────────────────────────────────────────────────────────────────────────────── app = FastAPI( title="AI Trading Chart API", version="6.0.0", description="OHLCV data, hybrid AI forecasts, technical indicators, and real-time WebSocket prices", lifespan=lifespan, ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # ────────────────────────────────────────────────────────────────────────────── # 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: Optional[asyncio.Lock] = None async def _get_lock(self) -> asyncio.Lock: if self._lock is None: self._lock = asyncio.Lock() return self._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 lock = await self._get_lock() async with 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() # MODULE: Analysis Engine v2.0 (Relocated and Activated) # Legacy placeholders removed to avoid duplication with logic at line 1884. # ────────────────────────────────────────────────────────────────────────────── # WebSocket connection manager # ────────────────────────────────────────────────────────────────────────────── class ConnectionManager: def __init__(self): self.active: Dict[str, List[WebSocket]] = defaultdict(list) async def connect(self, ws: WebSocket, symbol: str): await ws.accept() self.active[symbol].append(ws) logger.info("[WS] connect: %s (total=%d)", symbol, len(self.active[symbol])) def disconnect(self, ws: WebSocket, symbol: str): self.active[symbol] = [c for c in self.active[symbol] if c is not ws] logger.info("[WS] disconnect: %s (total=%d)", symbol, len(self.active[symbol])) async def broadcast(self, symbol: str, data: Dict[str, Any]): """Reliable broadcast with automatic stale connection cleanup (P0).""" active_list = self.active.get(symbol, []) if not active_list: return dead = [] for ws in list(active_list): try: if ws.client_state == WebSocketState.CONNECTED: await ws.send_json(data) else: dead.append(ws) except Exception: dead.append(ws) for ws in dead: self.disconnect(ws, symbol) async def heartbeat(self): """Keep-alive loop: send ping to all active clients every 20s (BUG-13).""" while True: await asyncio.sleep(20) ping_msg = {"type": "ping", "ts": int(time.time())} for symbol in list(self.active.keys()): for ws in list(self.active[symbol]): try: if ws.client_state == WebSocketState.CONNECTED: await ws.send_json(ping_msg) except Exception: pass ws_manager = ConnectionManager() @app.websocket("/ws/price/{symbol}") async def websocket_price(websocket: WebSocket, symbol: str): """ B-6: Real-time price streaming via WebSocket. Fixes the 404/500 errors in chrome console. """ symbol = _get_canonical_symbol(symbol) if symbol not in SYMBOLS: await websocket.close(code=1008, reason=f"Unknown symbol: {symbol}") return try: await ws_manager.connect(websocket, symbol) while True: try: # 1. State check if websocket.client_state != WebSocketState.CONNECTED: break # 2. Fetch fresh price ticker = await get_ticker(symbol) # 3. Final state check before send if websocket.client_state == WebSocketState.CONNECTED: await websocket.send_json({ "type": "price", "symbol": symbol, "price": ticker.get("price"), "change_pct": ticker.get("change_pct", 0), "ts": int(time.time()) }) except (WebSocketDisconnect, RuntimeError): break except Exception as ex: logger.debug("[WS] ticker fetch failed for %s: %s", symbol, ex) interval = 5 if SYMBOLS[symbol].category == "Crypto" else 30 await asyncio.sleep(interval) except Exception as ex: logger.error("[WS] handler error for %s: %s", symbol, ex) finally: ws_manager.disconnect(websocket, symbol) # ────────────────────────────────────────────────────────────────────────────── # Architecture v5: Security & Guardrails (Module F) # ────────────────────────────────────────────────────────────────────────────── 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. BUG-P1-11: Whitelist health and metrics from rate limiting. """ if request.url.path in ["/api/health", "/api/metrics", "/api/ping"]: return False 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] # BUG-P0-01: Removed duplicate v5 app instances and buggy lifespan assignment # Application v6.0 is defined earlier at line 2532 @app.middleware("http") async def rate_limit_middleware(request: Request, call_next): # BUG-P1-11: Whitelist health endpoints if request.url.path in ["/api/health", "/api/metrics"]: return await call_next(request) if rate_limit_guard(request): return JSONResponse(status_code=429, content={"detail": "Too many requests"}) return await call_next(request) 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 ───────────────────────────────────────────────── _SYMBOLS_CACHE: Optional[Dict[str, Any]] = None @app.get("/api/symbols") async def list_symbols( category: Optional[str] = Query(None, description="Filter by category"), ) -> Dict[str, Any]: global _SYMBOLS_CACHE if _SYMBOLS_CACHE is None: all_results = list(SYMBOLS.values()) grouped: Dict[str, List[Dict]] = {} for s in all_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) _SYMBOLS_CACHE = { "total": len(all_results), "categories": sorted(grouped.keys()), "symbols": [ { "symbol": s.symbol, "label": s.label, "label_en": s.label_en, "category": s.category, "sources": list(s.mappings.keys()) } for s in all_results ], "symbols_by_category": grouped, "supported_intervals": INTERVAL_ORDER } if category: cat_symbols = _SYMBOLS_CACHE["symbols_by_category"].get(category, []) return { "total": len(cat_symbols), "categories": _SYMBOLS_CACHE["categories"], "symbols": cat_symbols, "symbols_by_category": { category: cat_symbols }, "supported_intervals": INTERVAL_ORDER } return _SYMBOLS_CACHE # ── Market Peers ────────────────────────────────────────────────────────────── # @app.get("/api/market-peers") # async def get_market_peers(symbol: str = Query("BTCUSD")) -> Dict[str, Any]: # # Fix: Ensure symbol is canonical to avoid category mismatches # symbol = _get_canonical_symbol(symbol.upper()) # # if symbol not in SYMBOLS: # # Fallback to Crypto if unknown # # return await crypto_market(top=10) # return {"category": "Unknown", "peers": []} # # cfg = SYMBOLS[symbol] # category = cfg.category # # # Find all peers in the same category # peers_list = [s for s in SYMBOLS.values() if s.category == category and s.symbol != symbol] # # # If too few peers in category, mix with others # if len(peers_list) < 3: # all_others = [s for s in SYMBOLS.values() if s.symbol != symbol] # peers_list.extend(all_others[:5]) # # # Format result with actual price data # result_peers = [] # # Batch fetch ticker data for peers # for p in peers_list[:12]: # try: # ticker = await fetch_ticker(p.symbol) # result_peers.append({ # "symbol": p.symbol, # "label": p.label, # "category": p.category, # "price": ticker.get("price", 0), # "change_24h": ticker.get("change_pct", 0) # }) # except: # result_peers.append({ # "symbol": p.symbol, "label": p.label, "category": p.category, # "price": 0, "change_24h": 0 # }) # # return { # "category": category, # "peers": result_peers # } # ── 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, } # ── Technical Analysis Engine ───────────────────────────────────────────────── @app.get("/api/analysis/{symbol}") async def get_analysis( symbol: str, interval: str = Query("1h"), ) -> Dict[str, Any]: """ A-5: Direct access to the comprehensive Analysis Engine. E-2: Integrated Multi-timeframe (MTF) analysis. """ symbol = _get_canonical_symbol(symbol) if symbol not in SYMBOLS: raise HTTPException(404, f"Unknown symbol: {symbol}") # Fetch main context data, source = await fetch_historical(symbol, interval, 500) if len(data) < 50: raise HTTPException(422, "Insufficient data for full analysis") # Fetch higher timeframe context (MTF) htf_interval = "1d" if interval in ["1h", "2h", "4h"] else "1h" if interval in ["5m", "15m"] else None htf_bias = "neutral" if htf_interval: try: htf_data, _ = await fetch_historical(symbol, htf_interval, 200) htf_inds = compute_indicators(htf_data) htf_bias = "bullish" if htf_inds["trend"].get("above_ema200") else "bearish" except Exception: pass # Compute Indicators indicators = compute_indicators(data) # We need a mock forecast return if no forecast is available forecast_ret = 0.0 confidence = 50.0 try: # Try to get from cache to avoid heavy re-computation f_prefix = _cache_prefix(symbol, interval) f_cache = forecast_cache.get(f"forecast_{f_prefix}10") if f_cache: forecast_ret = _pct(f_cache["forecast"][-1]["p50"], f_cache["last_close"]) confidence = f_cache["ensemble"]["confidence"] except Exception: pass # Build comprehensive analysis analysis = _build_trade_analysis( symbol=symbol, interval=interval, data=data, indicators=indicators, forecast_rows=[], # Optional here confidence=confidence, source=source, ) # Inject MTF into analysis analysis["multi_timeframe"] = { "htf_interval": htf_interval, "htf_bias": htf_bias, "alignment": bool((analysis["summary"]["bias"] == "bullish" and htf_bias == "bullish") or (analysis["summary"]["bias"] == "bearish" and htf_bias == "bearish")) } return { "symbol": symbol, "interval": interval, "timestamp": int(time.time()), "analysis": analysis, "verdict": await get_gemini_verdict(symbol, analysis, forecast_ret), "indicators_snapshot": indicators if Query(False) else None # Save bandwidth } async def get_gemini_verdict(symbol: str, analysis: Dict[str, Any], forecast_pct: float = 0.0) -> str: """ Get a strict 4-option verdict from Gemini based on technical analysis + AI Forecast. Options: 'Mua ngay', 'Bán ngay', 'Nên đợi mua giá thấp hơn', 'Nên đợi bán giá cao hơn' """ summary = analysis.get("summary", {}) bias = summary.get("bias", "neutral") signal = summary.get("signal", "Trung lập") # Translate forecast to readable text forecast_text = "Tăng giá" if forecast_pct > 0.5 else "Giảm giá" if forecast_pct < -0.5 else "Đi ngang" prompt = f""" Dựa trên phân tích kỹ thuật và dự báo AI cho mã {symbol}: - Xu hướng kỹ thuật: {bias} ({signal}) - Tín hiệu Oscillators: {analysis['oscillators']['signal']} - Tín hiệu Moving Averages: {analysis['moving_averages']['signal']} - Dự báo AI (24h tới): {forecast_text} (biến động {forecast_pct:.2f}%) Hãy đưa ra kết luận DUY NHẤT trong 4 lựa chọn sau: 1. Mua ngay 2. Bán ngay 3. Nên đợi mua giá thấp hơn 4. Nên đợi bán giá cao hơn CHỈ TRẢ VỀ CỤM TỪ KẾT LUẬN, KHÔNG GIẢI THÍCH. """ return await fetch_gemini_analysis(prompt) # ── 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}" # L1: RAM Cache cached = forecast_cache.get(cache_key) if cached is not None: return cached # L2: Persistent SQLite Cache (A-4) p_cached = persistent_cache.get(cache_key) if p_cached is not None: p_cached["from_persistent_cache"] = True # Backfill L1 forecast_cache.set(cache_key, p_cached, ttl_seconds=forecast_ttl(interval)) return p_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, "from_persistent_cache": False, "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, } # L1: RAM forecast_cache.set(cache_key, response, ttl_seconds=forecast_ttl(interval)) # L2: SQLite (long TTL = forecast_ttl * 4) persistent_cache.set(cache_key, response, ttl=forecast_ttl(interval) * 4) 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(), } # 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) ──────────────────────────────────────────────── async def fetch_gemini_analysis(prompt: str) -> str: """Fetch AI analysis from Google Gemini.""" if not settings.gemini_api_key: return "Gemini API key is not configured." url = f"https://generativelanguage.googleapis.com/v1beta/models/gemini-flash-latest:generateContent?key={settings.gemini_api_key}" headers = {"Content-Type": "application/json"} payload = { "contents": [{"parts": [{"text": prompt}]}] } try: async with httpx.AsyncClient(timeout=30.0) as client: resp = await client.post(url, headers=headers, json=payload) if resp.status_code != 200: logger.error("[Gemini] Error: %d - %s", resp.status_code, resp.text) return "Phân tích AI không khả dụng" data = resp.json() candidates = data.get("candidates", []) if candidates and candidates[0].get("content", {}).get("parts"): text = candidates[0]["content"]["parts"][0].get("text", "").strip() # Ensure we only return the verdict text if it's a verdict call # We'll rely on the prompt to enforce this, but can sanitize here return text return "Không có phản hồi từ AI" except Exception as ex: logger.error("[Gemini] Exception: %s", ex) return "Lỗi phân tích AI" @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)