backend/main.py

"""
AI Trading Chart — Backend API v4.0
=====================================
FastAPI backend serving OHLCV data, AI forecasts, technical indicators,
real-time WebSocket prices, and a comprehensive symbol registry.

═══════════════════════════════════════════════════════════════════════════════
FREE DATA SOURCES (confirmed, no paid key required)
═══════════════════════════════════════════════════════════════════════════════
┌─────────────────┬─────────────────────────────────────┬──────────────────┐
│ Source          │ Best For                            │ Limit (Free)     │
├─────────────────┼─────────────────────────────────────┼──────────────────┤
│ Binance API     │ Crypto – fastest, deepest history   │ 1200 req/min     │
│ Bybit API       │ Crypto – backup to Binance          │ 120 req/min      │
│ CoinGecko API   │ Crypto OHLCV + market data          │ 30 req/min       │
│ yfinance        │ Stocks/ETF/Indices/Commodities/Forex│ Unlimited*       │
│ Finnhub         │ Forex/Stock/Crypto candles          │ 60 req/min       │
│ Twelvedata      │ Forex/Stocks/Crypto                 │ 800 credits/day  │
│ Alpha Vantage   │ Forex/Stocks (fallback)             │ 25 req/day       │
│ FRED (St.Louis) │ Macro indicators (DXY, rates...)    │ Unlimited        │
└─────────────────┴─────────────────────────────────────┴──────────────────┘
* yfinance scrapes Yahoo Finance — no official rate limit but use responsibly.

Source priority per asset class:
  Crypto    : binance → bybit → coingecko → yfinance → finnhub
  Forex     : twelvedata → finnhub → yfinance
  Stocks    : yfinance → finnhub → twelvedata
  Commodities: yfinance → twelvedata → finnhub
  Indices   : yfinance → twelvedata → finnhub
  VN Stocks : yfinance (.VN suffix)

═══════════════════════════════════════════════════════════════════════════════
Changelog v4.0 vs v3.0
═══════════════════════════════════════════════════════════════════════════════
[FEAT] 120+ symbols across 10 categories (Kim loại, Năng lượng, Nông sản,
       Nguyên liệu, Crypto, Cặp tiền, Chỉ số, Cổ phiếu Mỹ, Cổ phiếu VN,
       Trái phiếu/Lãi suất)
[FEAT] Bybit API — free crypto fallback after Binance
[FEAT] CoinGecko API — free crypto OHLCV + 24h stats
[FEAT] Per-category SOURCE_PRIORITY — each asset class uses its best source
[FEAT] GET /api/indicators/{symbol} — RSI, MACD, Bollinger Bands, EMA, ATR
[FEAT] GET /api/ticker/{symbol} — real-time last price + 24h stats
[FEAT] GET /api/market-status — which markets are open right now
[FEAT] GET /api/search?q= — fuzzy symbol search
[FEAT] WebSocket /ws/price/{symbol} — real-time price stream
[FEAT] GET /api/crypto/market — top-N crypto by market cap (CoinGecko)
[FEAT] Global rate-limiter with per-source token buckets
[FEAT] Exponential backoff with jitter on 429 / network errors
[FIX]  All v3.0 fixes retained (bias correction, dedup, 4h alignment…)
[IMPR] SymbolConfig extended: category_en, coingecko_id, bybit mapping
[IMPR] _normalize_ohlcv: gap detection — warns on suspiciously long candle gaps
[IMPR] Startup self-test pings all active sources and logs reachability
"""

from __future__ import annotations

import asyncio
import hmac
import logging
import math
import os
import hashlib
from dotenv import load_dotenv

# D-1: Load environment variables from .env file (v6.0)
load_dotenv()
os.environ["HF_HUB_DISABLE_SYMLINKS_WARNING"] = "1"
os.environ["HF_HUB_DISABLE_TELEMETRY"] = "1"
import re
import sys
import time
from collections import defaultdict
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional, Set, Tuple

import httpx
import numpy as np
import pandas as pd
import yfinance as yf
from backend.api_models import SwitchRequest, WatchlistRequest
from backend.cache_utils import PersistentCache, TTLCache
from backend.catalog_utils import build_symbols_catalog, filter_symbols_catalog
from backend.market_utils import MARKET_SESSIONS as MARKET_SESSIONS_DATA
from backend.market_utils import market_status_now as compute_market_status_now
from backend.observability_utils import (
    RequestMetricsRegistry,
    elapsed_ms,
    make_request_id,
    start_timer,
)
from backend.runtime_utils import (
    make_json_compatible,
    parse_cors_origins,
    resolve_runtime_paths,
)
from backend.security_utils import (
    RATE_LIMIT_WHITELIST,
    admin_only as require_admin_token,
    normalize_watchlist_symbols,
    rate_limit_guard as check_rate_limit,
    validate_cache_target,
    validate_symbol_and_interval,
)
from backend.startup_utils import (
    build_source_selftest_urls,
    clear_stale_ip_limits,
    run_source_selftest,
)
from backend.symbol_utils import (
    assemble_market_peer_payload,
    build_market_peer_candidates,
    search_symbols_catalog,
)
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException, Query, WebSocket, WebSocketDisconnect, Request
from fastapi.websockets import WebSocketState
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel

try:
    import torch
    TORCH_IMPORT_ERROR: Optional[str] = None
except Exception as exc:
    torch = None  # type: ignore[assignment]
    TORCH_IMPORT_ERROR = str(exc)

DEFAULT_ADMIN_TOKEN = "kronos_v6_default_secret"

# ──────────────────────────────────────────────────────────────────────────────
# Architecture v5: Configuration Management (C-1)
# ──────────────────────────────────────────────────────────────────────────────
class Settings(BaseModel):
    # D-1: Security Hardening (v6.0) - Enforce environment variables
    twelvedata_api_key: Optional[str] = os.getenv("TWELVEDATA_API_KEY")
    finnhub_api_key:    Optional[str] = os.getenv("FINNHUB_API_KEY")
    binance_api_key:    Optional[str] = os.getenv("BINANCE_API_KEY")
    binance_api_secret: Optional[str] = os.getenv("BINANCE_API_SECRET")
    bybit_api_key:      Optional[str] = os.getenv("BYBIT_API_KEY")
    bybit_api_secret:   Optional[str] = os.getenv("BYBIT_API_SECRET")
    alphavantage_api_key: Optional[str] = os.getenv("ALPHAVANTAGE_API_KEY")
    admin_token:        str           = os.getenv("ADMIN_TOKEN", DEFAULT_ADMIN_TOKEN)

    # Environment detection
    is_hf: bool = os.getenv("SPACE_ID") is not None

    # App Config
    host: str = os.getenv("HOST", "0.0.0.0")
    port: int = int(os.getenv("PORT", 8000))
    preload_kronos: bool = os.getenv("PRELOAD_KRONOS", "True").lower() == "true"
    cache_version:  str  = os.getenv("CACHE_VERSION", "v12")
    cors_allow_origins_raw: str = os.getenv("CORS_ALLOW_ORIGINS", "*")
    
    # WebSocket Config
    ws_heartbeat_interval: int = 20
    
    # Circuit Breaker Config
    cb_failure_threshold: int = 5
    cb_recovery_timeout:  int = 60

settings = Settings()


DEFAULT_AI_RULE_PATHS = [
    r"D:\Python\Rules\core-behavior.mdc",
    r"D:\Python\Rules\thinking-reasoning.mdc",
    r"D:\Python\Rules\tool-usage.mdc",
    r"D:\Python\Rules\agentic-coding.mdc",
    r"D:\Python\Rules\code-quality.mdc",
    r"D:\Python\Rules\output-format.mdc",
]


@dataclass
class AIRuleDocument:
    path: str
    name: str
    description: str
    always_apply: bool
    body: str
    sections: List[str]
    modified_at: float
    frontmatter: Dict[str, str] = field(default_factory=dict)


class AIRuleRegistry:
    def __init__(self, rule_paths: List[str]):
        self.rule_paths = [str(Path(p)) for p in rule_paths]
        self.documents: List[AIRuleDocument] = []
        self.missing_paths: List[str] = []
        self.load_errors: Dict[str, str] = {}
        self.combined_instruction = ""
        self.version = "uninitialized"
        self._mtimes: Dict[str, float] = {}
        self.refresh()

    def _parse_frontmatter(self, raw_text: str) -> Tuple[Dict[str, str], str]:
        lines = raw_text.splitlines()
        if len(lines) < 3 or lines[0].strip() != "---":
            return {}, raw_text.strip()

        meta: Dict[str, str] = {}
        closing_index = None
        for index in range(1, len(lines)):
            if lines[index].strip() == "---":
                closing_index = index
                break
            if ":" not in lines[index]:
                continue
            key, value = lines[index].split(":", 1)
            meta[key.strip()] = value.strip().strip('"').strip("'")

        if closing_index is None:
            return {}, raw_text.strip()
        body = "\n".join(lines[closing_index + 1:]).strip()
        return meta, body

    def _extract_sections(self, body: str) -> List[str]:
        sections: List[str] = []
        for line in body.splitlines():
            stripped = line.strip()
            if stripped.startswith("## "):
                sections.append(stripped[3:].strip())
        return sections

    def _load_document(self, path: str) -> AIRuleDocument:
        rule_path = Path(path)
        raw_text = rule_path.read_text(encoding="utf-8")
        meta, body = self._parse_frontmatter(raw_text)
        return AIRuleDocument(
            path=str(rule_path),
            name=rule_path.stem,
            description=meta.get("description", ""),
            always_apply=meta.get("alwaysApply", "").lower() == "true",
            body=body,
            sections=self._extract_sections(body),
            modified_at=rule_path.stat().st_mtime,
            frontmatter=meta,
        )

    def _compile_instruction(self, documents: List[AIRuleDocument]) -> str:
        blocks = [
            "You must follow the runtime policy below for every response.",
            "Treat these rules as active operating constraints, not optional guidance.",
        ]
        for doc in documents:
            header = doc.description or doc.name
            blocks.append(f"[{doc.name}] {header}")
            blocks.append(doc.body.strip())
        return "\n\n".join(blocks).strip()

    def refresh(self) -> None:
        documents: List[AIRuleDocument] = []
        missing_paths: List[str] = []
        load_errors: Dict[str, str] = {}
        mtimes: Dict[str, float] = {}

        for path in self.rule_paths:
            rule_path = Path(path)
            if not rule_path.exists():
                missing_paths.append(str(rule_path))
                continue
            try:
                doc = self._load_document(str(rule_path))
                documents.append(doc)
                mtimes[str(rule_path)] = doc.modified_at
            except Exception as exc:
                load_errors[str(rule_path)] = str(exc)

        documents.sort(key=lambda doc: self.rule_paths.index(doc.path))
        self.documents = documents
        self.missing_paths = missing_paths
        self.load_errors = load_errors
        self._mtimes = mtimes
        digest_source = "||".join(
            f"{doc.path}:{doc.modified_at}:{hashlib.sha256(doc.body.encode('utf-8')).hexdigest()}"
            for doc in documents
        )
        self.version = hashlib.sha256(digest_source.encode("utf-8")).hexdigest()[:12] if digest_source else "empty"
        self.combined_instruction = self._compile_instruction(documents)

    def refresh_if_needed(self) -> None:
        for path in self.rule_paths:
            rule_path = Path(path)
            current_mtime = rule_path.stat().st_mtime if rule_path.exists() else -1.0
            if self._mtimes.get(str(rule_path), -2.0) != current_mtime:
                self.refresh()
                return

    def build_system_instruction(self, task_contract: str) -> str:
        self.refresh_if_needed()
        parts = [self.combined_instruction] if self.combined_instruction else []
        if task_contract.strip():
            parts.append(task_contract.strip())
        return "\n\n".join(parts).strip()

    def snapshot(self) -> Dict[str, Any]:
        self.refresh_if_needed()
        return {
            "version": self.version,
            "loaded_count": len(self.documents),
            "missing_count": len(self.missing_paths),
            "error_count": len(self.load_errors),
            "missing_paths": self.missing_paths,
            "load_errors": self.load_errors,
            "documents": [
                {
                    "name": doc.name,
                    "path": doc.path,
                    "description": doc.description,
                    "always_apply": doc.always_apply,
                    "sections": doc.sections,
                    "modified_at": doc.modified_at,
                }
                for doc in self.documents
            ],
            "compiled_instruction_preview": self.combined_instruction[:2000],
        }


ai_rule_registry = AIRuleRegistry(DEFAULT_AI_RULE_PATHS)


# ──────────────────────────────────────────────────────────────────────────────
# Architecture v5: Global HTTP Client Pool (C-2)
# ──────────────────────────────────────────────────────────────────────────────
class GlobalHTTPClient:
    """Singleton HTTP client for optimized connection pooling (C-2)."""
    _client: Optional[httpx.AsyncClient] = None

    @classmethod
    async def get_client(cls) -> httpx.AsyncClient:
        if cls._client is None or cls._client.is_closed:
            cls._client = httpx.AsyncClient(
                timeout=httpx.Timeout(15.0, connect=5.0),
                limits=httpx.Limits(max_connections=100, max_keepalive_connections=20)
            )
        return cls._client

    @classmethod
    async def close(cls):
        if cls._client and not cls._client.is_closed:
            await cls._client.aclose()


# ──────────────────────────────────────────────────────────────────────────────
# Architecture v5: Circuit Breaker Pattern (C-3)
# ──────────────────────────────────────────────────────────────────────────────
class CircuitBreaker:
    """Protects against failing data sources (C-3)."""
    def __init__(self, name: str, threshold: int = 5, timeout: int = 60):
        self.name = name
        self.threshold = threshold
        self.timeout = timeout
        self.failures = 0
        self.last_failure_time = 0.0
        self.state = "CLOSED" # CLOSED, OPEN, HALF-OPEN

    def allow_request(self) -> bool:
        if self.state == "OPEN":
            if time.time() - self.last_failure_time > self.timeout:
                self.state = "HALF-OPEN"
                return True
            return False
        return True

    def record_success(self):
        self.failures = 0
        self.state = "CLOSED"

    def record_failure(self):
        self.failures += 1
        self.last_failure_time = time.time()
        if self.failures >= self.threshold:
            self.state = "OPEN"
            logger.error("[CB] %s is now OPEN - circuit broken", self.name)

    def open_for(self, timeout: Optional[int] = None) -> None:
        if timeout is not None:
            self.timeout = max(self.timeout, timeout)
        self.failures = max(self.failures, self.threshold)
        self.last_failure_time = time.time()
        self.state = "OPEN"
        logger.error("[CB] %s is OPEN for %ss", self.name, self.timeout)

# Instance per source
source_breakers: Dict[str, CircuitBreaker] = {
    s: CircuitBreaker(s, settings.cb_failure_threshold, settings.cb_recovery_timeout)
    for s in ["binance", "bybit", "coingecko", "twelvedata", "finnhub", "yfinance", "alphavantage"]
}
# ──────────────────────────────────────────────────────────────────────────────
# Logging
# ──────────────────────────────────────────────────────────────────────────────
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
)
logger = logging.getLogger("ai-trading-chart")
# Avoid leaking query-string API keys from httpx/httpcore INFO logs.
logging.getLogger("httpx").setLevel(logging.WARNING)
logging.getLogger("httpcore").setLevel(logging.WARNING)

APP_VERSION = "6.1.0"

# ─── PyInstaller / Frozen detection ─────────────────────────────────────────
IS_FROZEN = getattr(sys, 'frozen', False)
BUNDLE_DIR = getattr(sys, '_MEIPASS', os.path.dirname(os.path.abspath(__file__)))
RUNTIME_PATHS = resolve_runtime_paths(__file__, IS_FROZEN, BUNDLE_DIR)
CURRENT_DIR = RUNTIME_PATHS.current_dir
PROJECT_ROOT = RUNTIME_PATHS.project_root

if IS_FROZEN:
    logger.info("Running in FROZEN mode. BUNDLE_DIR: %s", BUNDLE_DIR)
else:
    logger.info("Running in DEV mode. PROJECT_ROOT: %s", PROJECT_ROOT)

# TimesFM 2.5 — replaces legacy local Kronos model
TIMESFM_AVAILABLE = False
TIMESFM_IMPORT_ERROR: Optional[str] = None

if TORCH_IMPORT_ERROR:
    TIMESFM_IMPORT_ERROR = TORCH_IMPORT_ERROR
    logger.warning("TimesFM disabled because torch is unavailable: %s", TORCH_IMPORT_ERROR)
else:
    try:
        import timesfm  # pip install timesfm[torch]
        TIMESFM_AVAILABLE = True
        logger.info("TimesFM library imported successfully")
    except Exception as _tfm_ex:
        TIMESFM_IMPORT_ERROR = str(_tfm_ex)
        logger.error("TimesFM import error: %s", _tfm_ex)
        logger.warning("TimesFM not installed — forecasting disabled. Run: pip install timesfm[torch]")

PRELOAD_TIMESFM = os.getenv("TIMESFM_PRELOAD", "1").strip().lower() not in {"0", "false", "no"}
STARTUP_STATE: Dict[str, Any] = {
    "timesfm": {
        "available": TIMESFM_AVAILABLE,
        "preload_enabled": PRELOAD_TIMESFM,
        "warming": False,
        "loaded": False,
        "device": "not_loaded",
        "last_error": TIMESFM_IMPORT_ERROR,
        "model": "google/timesfm-2.5-200m-pytorch",
    },
    "sources": {},
}

# ──────────────────────────────────────────────────────────────────────────────
# Architecture v5: Structured Logging & Persistence (C-4, C-6)
# ──────────────────────────────────────────────────────────────────────────────
import json

class StructuredLogger:
    @staticmethod
    def info(msg: str, **kwargs):
        payload = {"level": "INFO", "msg": msg, "ts": datetime.now().isoformat()}
        payload.update(kwargs)
        logger.info(msg) # Keeping standard log for dev, but payload is ready for JSON

persistent_cache = PersistentCache(
    db_path=os.path.join(PROJECT_ROOT, "data", "kronos_v5.db"),
    cache_version_getter=lambda: settings.cache_version,
    logger=logger,
)


# B-12: Global Configuration Instances
CACHE_VERSION = settings.cache_version
ADMIN_TOKEN = settings.admin_token
CORS_ALLOW_ORIGINS = parse_cors_origins(settings.cors_allow_origins_raw)
CORS_ALLOW_CREDENTIALS = CORS_ALLOW_ORIGINS != ["*"]


# ── API Key Pool (round-robin rotation for multi-pane) ────────────────────────
class APIKeyPool:
    """Round-robin key rotation to distribute API calls across multiple keys."""

    def __init__(self, primary_key, pool_csv, name=""):
        self._name = name
        self._keys = []
        self._index = 0
        self._exhausted = set()

        if pool_csv:
            self._keys = [k.strip() for k in pool_csv.split(",") if k.strip()]
        if not self._keys and primary_key:
            self._keys = [primary_key]

        logger.info("[APIKeyPool] %s: %d keys loaded", name, len(self._keys))

    @property
    def primary(self):
        return self._keys[0] if self._keys else None

    def next_key(self):
        if not self._keys:
            return None
        available = [k for k in self._keys if k not in self._exhausted]
        if not available:
            self._exhausted.clear()
            available = self._keys
        key = available[self._index % len(available)]
        self._index += 1
        return key

    def mark_exhausted(self, key):
        self._exhausted.add(key)
        remaining = len(self._keys) - len(self._exhausted)
        logger.warning("[APIKeyPool] %s: key ...%s exhausted (%d remaining)",
                       self._name, key[-8:], remaining)

    def reset(self):
        self._exhausted.clear()
        self._index = 0

    @property
    def pool_size(self):
        return len(self._keys)

    @property
    def available_count(self):
        return len(self._keys) - len(self._exhausted)


twelve_keys = [os.getenv(f"TWELVEDATA_API_KEY_{i}") for i in range(1, 9)]
twelvedata_pool = APIKeyPool(
    settings.twelvedata_api_key,
    ",".join([k for k in twelve_keys if k]),
    name="TwelveData",
)
finnhub_keys = [os.getenv(f"FINNHUB_API_KEY_{i}") for i in range(1, 9)]
finnhub_pool = APIKeyPool(
    settings.finnhub_api_key,
    ",".join([k for k in finnhub_keys if k]),
    name="Finnhub",
)

# Backward compat: keep single-key globals pointing to primary
TWELVEDATA_API_KEY = twelvedata_pool.primary
FINNHUB_API_KEY = finnhub_pool.primary
# Binance, Bybit, CoinGecko, yfinance, FRED — no key or optional key required

# ──────────────────────────────────────────────────────────────────────────────
# Constants
# ──────────────────────────────────────────────────────────────────────────────
SUPPORTED_INTERVALS: frozenset = frozenset({"1m", "5m", "15m", "1h", "4h", "1d", "1w"})
INTERVAL_ORDER:      List[str] = ["1m", "5m", "15m", "1h", "4h", "1d", "1w"]

TWELVE_INTERVAL_MAP: Dict[str, str] = {
    "1m": "1min", "5m": "5min", "15m": "15min",
    "1h": "1h",   "4h": "4h",  "1d": "1day", "1w": "1week",
}
FINNHUB_RESOLUTION_MAP: Dict[str, str] = {
    "1m": "1", "5m": "5", "15m": "15",
    "1h": "60","4h": "240","1d": "D", "1w": "W",
}
YF_INTERVAL_MAP: Dict[str, str] = {
    "1m": "1m", "5m": "5m", "15m": "15m",
    "1h": "60m","4h": "60m","1d": "1d", "1w": "1wk",
}
YF_PERIOD_MAP: Dict[str, str] = {
    "1m": "7d", "5m": "60d","15m": "60d",
    "1h": "730d","4h": "60d","1d": "max","1w": "max",
}
BINANCE_INTERVAL_MAP: Dict[str, str] = {
    "1m": "1m", "5m": "5m", "15m": "15m",
    "1h": "1h", "4h": "4h","1d": "1d",  "1w": "1w",
}
BYBIT_INTERVAL_MAP: Dict[str, str] = {
    "1m": "1",  "5m": "5",  "15m": "15",
    "1h": "60", "4h": "240","1d": "D",  "1w": "W",
}
STEP_SECONDS: Dict[str, int] = {
    "1m": 60,   "5m": 300,  "15m": 900,
    "1h": 3600, "4h": 14400,"1d": 86400,"1w": 604800,
}

# Source priority by asset category (first available mapping wins)
CATEGORY_SOURCE_PRIORITY: Dict[str, List[str]] = {
    "Crypto":               ["binance", "bybit", "coingecko", "yfinance", "finnhub"],
    "Cặp tiền":             ["twelvedata", "finnhub", "yfinance"],
    "Real Strength":        ["yfinance", "twelvedata", "finnhub"],
    "Kim loại":             ["binance", "twelvedata", "yfinance", "finnhub"],
    "Năng lượng":           ["twelvedata", "yfinance", "finnhub"],
    "Nông sản":             ["twelvedata", "yfinance"],
    "Nguyên liệu CN":       ["twelvedata", "yfinance"],
    "Chỉ số":               ["binance", "twelvedata", "yfinance", "finnhub"],
    "Cổ phiếu Mỹ":          ["binance", "twelvedata", "finnhub", "yfinance"],
    "Cổ phiếu VN":          ["yfinance"],
    "Trái phiếu":           ["twelvedata", "yfinance"],
    "ETF":                  ["binance", "twelvedata", "yfinance", "finnhub"],
}
# Fallback for unknown categories
DEFAULT_SOURCE_PRIORITY: List[str] = ["binance", "yfinance", "twelvedata", "finnhub"]

CLIP_DEFAULT  = 3.0
INDICATOR_MIN_CONTEXT = 240
ANALYSIS_CONTEXT = 750
FORECAST_CONTEXT = 1500
HIGHER_TIMEFRAME_CONTEXT = 240
WATCHLIST_TICKER_CONCURRENCY = 12
MARKET_PEER_TICKER_CONCURRENCY = 8


# ──────────────────────────────────────────────────────────────────────────────
# Rate Limiter (token-bucket, per source)
# ──────────────────────────────────────────────────────────────────────────────
class TokenBucket:
    """Simple async token-bucket rate limiter."""
    def __init__(self, rate: float, capacity: int) -> None:
        self._rate     = rate       # tokens per second
        self._capacity = capacity
        self._tokens   = float(capacity)
        self._last     = time.monotonic()

    async def acquire(self, tokens: int = 1) -> None:
        while True:
            now = time.monotonic()
            elapsed = now - self._last
            self._tokens = min(self._capacity, self._tokens + elapsed * self._rate)
            self._last = now
            if self._tokens >= tokens:
                self._tokens -= tokens
                return
            await asyncio.sleep(0.1)


# Per-source buckets (conservative — stays well within free limits)
_TWELVEDATA_POOL_SIZE = max(1, twelvedata_pool.pool_size)
_TWELVEDATA_RATE = min(1.0, 0.1 * _TWELVEDATA_POOL_SIZE)
_TWELVEDATA_CAPACITY = max(2, min(8, _TWELVEDATA_POOL_SIZE))
_rate_limiters: Dict[str, TokenBucket] = {
    "binance":     TokenBucket(rate=10.0, capacity=20),
    "bybit":       TokenBucket(rate=1.5,  capacity=5),
    "coingecko":   TokenBucket(rate=0.4,  capacity=3),
    "twelvedata":  TokenBucket(rate=_TWELVEDATA_RATE, capacity=_TWELVEDATA_CAPACITY),
    "finnhub":     TokenBucket(rate=1.0,  capacity=5),
    "yfinance":    TokenBucket(rate=5.0,  capacity=10),
    "alphavantage":TokenBucket(rate=0.02, capacity=1),   # 25/day
}

async def _rate_limit(source: str) -> None:
    bucket = _rate_limiters.get(source)
    if bucket:
        await bucket.acquire()


async def _retry(coro_fn, retries: int = 3, base_delay: float = 1.0):
    """Exponential backoff with jitter for transient failures."""
    for attempt in range(retries):
        try:
            return await coro_fn()
        except (httpx.ReadTimeout, httpx.ConnectTimeout, httpx.RemoteProtocolError, 
                httpx.ConnectError, httpx.PoolTimeout):
            if attempt == retries - 1:
                raise
            delay = base_delay * (2 ** attempt) + (0.1 * attempt)
            await asyncio.sleep(delay)


# ──────────────────────────────────────────────────────────────────────────────
# Symbol Registry
# ──────────────────────────────────────────────────────────────────────────────
@dataclass
class SymbolConfig:
    symbol:       str
    label:        str
    label_en:     str
    category:     str
    mappings:     Dict[str, str]
    coingecko_id: Optional[str] = None   # for CoinGecko OHLCV
    bybit_category: str = "linear"       # "linear" or "spot"
    binance_type:  str = "spot"          # "spot" or "futures"
    description:  str = ""


@dataclass(frozen=True)
class SyntheticComponentSpec:
    name: str
    mode: str
    left_symbol: str
    right_symbol: Optional[str] = None
    weight: float = 1.0
    enabled: bool = True


@dataclass(frozen=True)
class SyntheticSymbolConfig:
    symbol: str
    scale: float
    alpha: float
    wick_shrink: float
    components: Tuple[SyntheticComponentSpec, ...]


# ─── Helper to build entry quickly ────────────────────────────────────────────
def _s(sym: str, label: str, label_en: str, cat: str,
       mappings: Dict[str, str], cg_id: str = None, desc: str = "", 
       bybit_cat: str = "linear", bin_type: str = "spot") -> SymbolConfig:
    return SymbolConfig(sym, label, label_en, cat, mappings, cg_id, bybit_cat, bin_type, desc)


SYMBOLS: Dict[str, SymbolConfig] = {

    # ══════════════════════════════════════════════════════════════════════
    # 1. KIM LOẠI (Metals)
    # ══════════════════════════════════════════════════════════════════════
    "XAUUSD": _s("XAUUSD","Vàng (Gold)","Gold","Kim loại",{"binance":"PAXGUSDT","twelvedata":"XAU/USD","yfinance":"GC=F"}),
    "XAGUSD": _s("XAGUSD","Bạc (Silver)","Silver","Kim loại",{"twelvedata":"XAG/USD","yfinance":"SI=F"}),
    "XPTUSD": _s("XPTUSD","Bạch kim (Platinum)","Platinum","Kim loại",{"twelvedata":"XPT/USD","yfinance":"PL=F"}),
    "XPDUSD": _s("XPDUSD","Palladium","Palladium","Kim loại",{"twelvedata":"XPD/USD","yfinance":"PA=F"}),
    "HGUSD":  _s("HGUSD","Đồng (Copper)","Copper","Kim loại",{"twelvedata":"COPPER","yfinance":"HG=F"}),
    "ALUSD":  _s("ALUSD","Nhôm (Aluminum)","Aluminum","Kim loại",{"yfinance":"AL=F"}),
    "ZNUSD":  _s("ZNUSD","Kẽm (Zinc)","Zinc","Kim loại",{"yfinance":"ZN=F"}),
    "NIUSD":  _s("NIUSD","Niken (Nickel)","Nickel","Kim loại",{"yfinance":"NI=F"}),
    "PBUSD":  _s("PBUSD","Chì (Lead)","Lead","Kim loại",{"yfinance":"LED=F"}),
    "SNUSD":  _s("SNUSD","Thiếc (Tin)","Tin","Kim loại",{"yfinance":"SN=F"}),

    # ══════════════════════════════════════════════════════════════════════
    # 2. NĂNG LƯỢNG (Energy)
    # ══════════════════════════════════════════════════════════════════════
    "USOIL":  _s("USOIL","Dầu thô WTI","WTI Oil","Năng lượng",{"twelvedata":"WTI","yfinance":"CL=F"}),
    "UKOIL":  _s("UKOIL","Dầu Brent","Brent Oil","Năng lượng",{"twelvedata":"BRENT","yfinance":"BZ=F"}),
    "NGAS":   _s("NGAS","Khí tự nhiên","Natural Gas","Năng lượng",{"twelvedata":"NATGAS","yfinance":"NG=F"}),
    "GASO":   _s("GASO","Xăng RBOB","Gasoline","Năng lượng",{"yfinance":"RB=F"}),
    "HEAT":   _s("HEAT","Dầu sưởi","Heating Oil","Năng lượng",{"yfinance":"HO=F"}),
    "COAL":   _s("COAL","Than (Coal)","Coal","Năng lượng",{"yfinance":"MTF=F"}),

    # ══════════════════════════════════════════════════════════════════════
    # 3. NGUYÊN LIỆU CÔNG NGHIỆP (Industrial Materials)
    # ══════════════════════════════════════════════════════════════════════
    "COFFEE": _s("COFFEE","Cà phê Arabica","Arabica Coffee","Nguyên liệu CN",{"twelvedata":"COFFEE","yfinance":"KC=F"}),
    "ROBUSTA":_s("ROBUSTA","Cà phê Robusta","Robusta Coffee","Nguyên liệu CN",{"yfinance":"RC=F"}),
    "COCOA":  _s("COCOA","Ca cao (Cocoa)","Cocoa","Nguyên liệu CN",{"yfinance":"CC=F"}),
    "SUGAR11":_s("SUGAR11","Đường 11","Sugar No.11","Nguyên liệu CN",{"twelvedata":"SUGAR","yfinance":"SB=F"}),
    "LSUGAR": _s("LSUGAR","Đường trắng","White Sugar","Nguyên liệu CN",{"yfinance":"LSU.L"}),
    "PALMOIL":_s("PALMOIL","Dầu cọ thô","Palm Oil","Nguyên liệu CN",{"yfinance":"FCPO=F"}),
    "COTTON": _s("COTTON","Bông (Cotton)","Cotton","Nguyên liệu CN",{"twelvedata":"COTTON","yfinance":"CT=F"}),
    "RUBBER": _s("RUBBER","Cao su RSS3","Rubber","Nguyên liệu CN",{"yfinance":"JR=F"}),
    "ORANGE": _s("ORANGE","Nước cam","Orange Juice","Nguyên liệu CN",{"yfinance":"OJ=F"}),

    # ══════════════════════════════════════════════════════════════════════
    # 4. NÔNG SẢN (Agriculture)
    # ══════════════════════════════════════════════════════════════════════
    "CORN":   _s("CORN","Ngô (Corn)","Corn","Nông sản",{"twelvedata":"CORN","yfinance":"ZC=F"}),
    "WHEAT":  _s("WHEAT","Lúa mì (Wheat)","Wheat","Nông sản",{"twelvedata":"WHEAT","yfinance":"ZW=F"}),
    "SOY":    _s("SOY","Đậu tương","Soybeans","Nông sản",{"twelvedata":"SOYBEAN","yfinance":"ZS=F"}),
    "SOYOIL": _s("SOYOIL","Dầu đậu tương","Soybean Oil","Nông sản",{"yfinance":"ZL=F"}),
    "SOYMEAL":_s("SOYMEAL","Khô đậu tương","Soybean Meal","Nông sản",{"yfinance":"ZM=F"}),
    "RICE":   _s("RICE","Gạo (Rice)","Rice","Nông sản",{"yfinance":"ZR=F"}),
    "OATS":   _s("OATS","Yến mạch (Oats)","Oats","Nông sản",{"yfinance":"ZO=F"}),

    # ══════════════════════════════════════════════════════════════════════
    # 5. CRYPTO (Top 50+)
    # ══════════════════════════════════════════════════════════════════════
    "BTCUSD": _s("BTCUSD","Bitcoin (BTC)","Bitcoin","Crypto",{"binance":"BTCUSDT","bybit":"BTCUSDT","coingecko":"bitcoin"}),
    "ETHUSD": _s("ETHUSD","Ethereum (ETH)","Ethereum","Crypto",{"binance":"ETHUSDT","bybit":"ETHUSDT","coingecko":"ethereum"}),
    "BNBUSD": _s("BNBUSD","BNB (BNB)","BNB","Crypto",{"binance":"BNBUSDT","bybit":"BNBUSDT","coingecko":"binancecoin"}),
    "SOLUSD": _s("SOLUSD","Solana (SOL)","Solana","Crypto",{"binance":"SOLUSDT","bybit":"SOLUSDT","coingecko":"solana"}),
    "XRPUSD": _s("XRPUSD","Ripple (XRP)","XRP","Crypto",{"binance":"XRPUSDT","bybit":"XRPUSDT","coingecko":"ripple"}),
    "ADAUSD": _s("ADAUSD","Cardano (ADA)","Cardano","Crypto",{"binance":"ADAUSDT","bybit":"ADAUSDT","coingecko":"cardano"}),
    "AVAXUSD": _s("AVAXUSD","Avalanche (AVAX)","Avalanche","Crypto",{"binance":"AVAXUSDT","bybit":"AVAXUSDT","coingecko":"avalanche-2"}),
    "DOTUSD": _s("DOTUSD","Polkadot (DOT)","Polkadot","Crypto",{"binance":"DOTUSDT","bybit":"DOTUSDT","coingecko":"polkadot"}),
    "MATICUSD": _s("MATICUSD","Polygon (MATIC)","Polygon","Crypto",{"binance":"MATICUSDT","bybit":"MATICUSDT","coingecko":"matic-network"}),
    "LINKUSD": _s("LINKUSD","Chainlink (LINK)","Chainlink","Crypto",{"binance":"LINKUSDT","bybit":"LINKUSDT","coingecko":"chainlink"}),
    "LTCUSD": _s("LTCUSD","Litecoin (LTC)","Litecoin","Crypto",{"binance":"LTCUSDT","bybit":"LTCUSDT","coingecko":"litecoin"}),
    "UNIUSD": _s("UNIUSD","Uniswap (UNI)","Uniswap","Crypto",{"binance":"UNIUSDT","bybit":"UNIUSDT","coingecko":"uniswap"}),
    "ATOMUSD": _s("ATOMUSD","Cosmos (ATOM)","Cosmos","Crypto",{"binance":"ATOMUSDT"}),
    "NEARUSD": _s("NEARUSD","NEAR Protocol","NEAR","Crypto",{"binance":"NEARUSDT"}),
    "APTUSD": _s("APTUSD","Aptos (APT)","Aptos","Crypto",{"binance":"APTUSDT"}),
    "SUIUSD": _s("SUIUSD","Sui (SUI)","Sui","Crypto",{"binance":"SUIUSDT"}),
    "ARBUSD": _s("ARBUSD","Arbitrum (ARB)","Arbitrum","Crypto",{"binance":"ARBUSDT"}),
    "OPUSD":  _s("OPUSD","Optimism (OP)","Optimism","Crypto",{"binance":"OPUSDT"}),
    "TONUSD": _s("TONUSD","Toncoin (TON)","Toncoin","Crypto",{"binance":"TONUSDT"}),
    "TRXUSD": _s("TRXUSD","Tron (TRX)","Tron","Crypto",{"binance":"TRXUSDT"}),
    "DOGEUSD":_s("DOGEUSD","Dogecoin (DOGE)","Dogecoin","Crypto",{"binance":"DOGEUSDT"}),
    "SHIBUSD":_s("SHIBUSD","Shiba Inu (SHIB)","Shiba Inu","Crypto",{"binance":"SHIBUSDT"}),
    "PEPEUSD":_s("PEPEUSD","PEPE Coin","PEPE","Crypto",{"binance":"PEPEUSDT"}),
    "FLOKIUSD":_s("FLOKIUSD","FLOKI Inu","FLOKI","Crypto",{"binance":"FLOKIUSDT"}),
    "BONKUSD":_s("BONKUSD","BONK Coin","BONK","Crypto",{"binance":"BONKUSDT"}),
    "WIFUSD": _s("WIFUSD","dogwifhat (WIF)","WIF","Crypto",{"binance":"WIFUSDT"}),
    "RNDRUSD":_s("RNDRUSD","Render (RNDR)","Render","Crypto",{"binance":"RNDRUSDT"}),
    "FETUSD": _s("FETUSD","Fetch.ai (FET)","Fetch.ai","Crypto",{"binance":"FETUSDT"}),
    "STXUSD": _s("STXUSD","Stacks (STX)","Stacks","Crypto",{"binance":"STXUSDT"}),
    "FILUSD": _s("FILUSD","Filecoin (FIL)","Filecoin","Crypto",{"binance":"FILUSDT"}),
    "VETUSD": _s("VETUSD","VeChain (VET)","VeChain","Crypto",{"binance":"VETUSDT"}),
    "ICPUSD": _s("ICPUSD","Internet Computer","ICP","Crypto",{"binance":"ICPUSDT"}),

    # ══════════════════════════════════════════════════════════════════════
    # 6. CẶP TIỀN (Forex)
    # ══════════════════════════════════════════════════════════════════════
    "DXY":    _s("DXY","Chỉ số USD (DXY)","USD Index","Real Strength",{"twelvedata":"DXY","yfinance":"DX-Y.NYB"}),
    "USDX":   _s("USDX","Sức mạnh USD (USDx)","USD Strength (USDx)","Real Strength",{"synthetic":"USDX"}, desc="Synthetic USD strength index built from weighted USD crosses."),
    "EURX":   _s("EURX","Sức mạnh EUR (EURx)","EUR Strength (EURx)","Real Strength",{"synthetic":"EURX"}, desc="Synthetic EUR strength index built from weighted EUR crosses."),
    "GBPX":   _s("GBPX","Sức mạnh GBP (GBPx)","GBP Strength (GBPx)","Real Strength",{"synthetic":"GBPX"}, desc="Synthetic GBP strength index built from weighted GBP crosses."),
    "CHFX":   _s("CHFX","Sức mạnh CHF (CHFx)","CHF Strength (CHFx)","Real Strength",{"synthetic":"CHFX"}, desc="Synthetic CHF strength index built from weighted CHF crosses."),
    "JPYX":   _s("JPYX","Sức mạnh JPY (JPYx)","JPY Strength (JPYx)","Real Strength",{"synthetic":"JPYX"}, desc="Synthetic JPY strength index built from weighted JPY crosses."),
    "CADX":   _s("CADX","Sức mạnh CAD (CADx)","CAD Strength (CADx)","Real Strength",{"synthetic":"CADX"}, desc="Synthetic CAD strength index built from weighted CAD crosses."),
    "AUDX":   _s("AUDX","Sức mạnh AUD (AUDx)","AUD Strength (AUDx)","Real Strength",{"synthetic":"AUDX"}, desc="Synthetic AUD strength index built from weighted AUD crosses."),
    "NZDX":   _s("NZDX","Sức mạnh NZD (NZDx)","NZD Strength (NZDx)","Real Strength",{"synthetic":"NZDX"}, desc="Synthetic NZD strength index built from weighted NZD crosses."),
    "EURUSD": _s("EURUSD","EUR/USD","EUR/USD","Cặp tiền",{"twelvedata":"EUR/USD","yfinance":"EURUSD=X"}),
    "GBPUSD": _s("GBPUSD","GBP/USD","GBP/USD","Cặp tiền",{"twelvedata":"GBP/USD","yfinance":"GBPUSD=X"}),
    "USDJPY": _s("USDJPY","USD/JPY","USD/JPY","Cặp tiền",{"twelvedata":"USD/JPY","yfinance":"JPY=X"}),
    "USDCHF": _s("USDCHF","USD/CHF","USD/CHF","Cặp tiền",{"twelvedata":"USD/CHF","yfinance":"CHF=X"}),
    "AUDUSD": _s("AUDUSD","AUD/USD","AUD/USD","Cặp tiền",{"twelvedata":"AUD/USD","yfinance":"AUDUSD=X"}),
    "USDCAD": _s("USDCAD","USD/CAD","USD/CAD","Cặp tiền",{"twelvedata":"USD/CAD","yfinance":"CAD=X"}),
    "NZDUSD": _s("NZDUSD","NZD/USD","NZD/USD","Cặp tiền",{"twelvedata":"NZD/USD","yfinance":"NZDUSD=X"}),
    "EURGBP": _s("EURGBP","EUR/GBP","EUR/GBP","Cặp tiền",{"twelvedata":"EUR/GBP","yfinance":"EURGBP=X"}),
    "EURJPY": _s("EURJPY","EUR/JPY","EUR/JPY","Cặp tiền",{"twelvedata":"EUR/JPY","yfinance":"EURJPY=X"}),
    "EURCHF": _s("EURCHF","EUR/CHF","EUR/CHF","Cặp tiền",{"twelvedata":"EUR/CHF","yfinance":"EURCHF=X"}),
    "EURCAD": _s("EURCAD","EUR/CAD","EUR/CAD","Cặp tiền",{"twelvedata":"EUR/CAD","yfinance":"EURCAD=X"}),
    "EURAUD": _s("EURAUD","EUR/AUD","EUR/AUD","Cặp tiền",{"twelvedata":"EUR/AUD","yfinance":"EURAUD=X"}),
    "EURNZD": _s("EURNZD","EUR/NZD","EUR/NZD","Cặp tiền",{"twelvedata":"EUR/NZD","yfinance":"EURNZD=X"}),
    "GBPJPY": _s("GBPJPY","GBP/JPY","GBP/JPY","Cặp tiền",{"twelvedata":"GBP/JPY","yfinance":"GBPJPY=X"}),
    "GBPCHF": _s("GBPCHF","GBP/CHF","GBP/CHF","Cặp tiền",{"twelvedata":"GBP/CHF","yfinance":"GBPCHF=X"}),
    "GBPCAD": _s("GBPCAD","GBP/CAD","GBP/CAD","Cặp tiền",{"twelvedata":"GBP/CAD","yfinance":"GBPCAD=X"}),
    "GBPAUD": _s("GBPAUD","GBP/AUD","GBP/AUD","Cặp tiền",{"twelvedata":"GBP/AUD","yfinance":"GBPAUD=X"}),
    "GBPNZD": _s("GBPNZD","GBP/NZD","GBP/NZD","Cặp tiền",{"twelvedata":"GBP/NZD","yfinance":"GBPNZD=X"}),
    "CHFJPY": _s("CHFJPY","CHF/JPY","CHF/JPY","Cặp tiền",{"twelvedata":"CHF/JPY","yfinance":"CHFJPY=X"}),
    "CADCHF": _s("CADCHF","CAD/CHF","CAD/CHF","Cặp tiền",{"twelvedata":"CAD/CHF","yfinance":"CADCHF=X"}),
    "AUDCHF": _s("AUDCHF","AUD/CHF","AUD/CHF","Cặp tiền",{"twelvedata":"AUD/CHF","yfinance":"AUDCHF=X"}),
    "NZDCHF": _s("NZDCHF","NZD/CHF","NZD/CHF","Cặp tiền",{"twelvedata":"NZD/CHF","yfinance":"NZDCHF=X"}),
    "CADJPY": _s("CADJPY","CAD/JPY","CAD/JPY","Cặp tiền",{"twelvedata":"CAD/JPY","yfinance":"CADJPY=X"}),
    "AUDJPY": _s("AUDJPY","AUD/JPY","AUD/JPY","Cặp tiền",{"twelvedata":"AUD/JPY","yfinance":"AUDJPY=X"}),
    "NZDJPY": _s("NZDJPY","NZD/JPY","NZD/JPY","Cặp tiền",{"twelvedata":"NZD/JPY","yfinance":"NZDJPY=X"}),
    "AUDCAD": _s("AUDCAD","AUD/CAD","AUD/CAD","Cặp tiền",{"twelvedata":"AUD/CAD","yfinance":"AUDCAD=X"}),
    "NZDCAD": _s("NZDCAD","NZD/CAD","NZD/CAD","Cặp tiền",{"twelvedata":"NZD/CAD","yfinance":"NZDCAD=X"}),
    "AUDNZD": _s("AUDNZD","AUD/NZD","AUD/NZD","Cặp tiền",{"twelvedata":"AUD/NZD","yfinance":"AUDNZD=X"}),
    "USDVND": _s("USDVND","USD/VND","USD/VND","Cặp tiền",{"yfinance":"VND=X"}),
    "USDCNH": _s("USDCNH","USD/CNH","USD/CNH","Cặp tiền",{"twelvedata":"USD/CNH"}),
    "USDHKD": _s("USDHKD","USD/HKD","USD/HKD","Cặp tiền",{"twelvedata":"USD/HKD"}),
    "USDSGD": _s("USDSGD","USD/SGD","USD/SGD","Cặp tiền",{"twelvedata":"USD/SGD"}),

    # ══════════════════════════════════════════════════════════════════════
    # 7. CHỈ SỐ THẾ GIỚI (Global Indices)
    # ══════════════════════════════════════════════════════════════════════
    "SP500":    _s("SP500","S&P 500","S&P 500","Chỉ số",{"yfinance":"^GSPC","twelvedata":"SPX"}, bin_type="futures"),
    "NASDAQ100":_s("NASDAQ100","Nasdaq 100","Nasdaq 100","Chỉ số",{"yfinance":"^NDX","twelvedata":"NDX"}, bin_type="futures"),
    "DOW30":    _s("DOW30","Dow Jones 30","Dow Jones","Chỉ số",{"yfinance":"^DJI"}),
    "RUSSELL2000":_s("RUSSELL2000","Russell 2000","Russell 2000","Chỉ số",{"yfinance":"^RUT"}),
    "VIX":      _s("VIX","Chỉ số sợ hãi (VIX)","VIX Index","Chỉ số",{"yfinance":"^VIX"}),
    "UK100":    _s("UK100","FTSE 100 (UK)","FTSE 100","Chỉ số",{"yfinance":"^FTSE"}),
    "DAX40":    _s("DAX40","DAX 40 (Đức)","DAX 40","Chỉ số",{"yfinance":"^GDAXI"}),
    "CAC40":    _s("CAC40","CAC 40 (Pháp)","CAC 40","Chỉ số",{"yfinance":"^FCHI"}),
    "EU50":     _s("EU50","Euro Stoxx 50","Euro Stoxx 50","Chỉ số",{"yfinance":"^STOXX50E"}),
    "NIKKEI225":_s("NIKKEI225","Nikkei 225 (Nhật)","Nikkei 225","Chỉ số",{"yfinance":"^N225"}),
    "HSI":      _s("HSI","Hang Seng (HK)","Hang Seng","Chỉ số",{"yfinance":"^HSI"}),
    "VNINDEX":  _s("VNINDEX","VN-Index","VN-Index","Chỉ số",{"yfinance":"^VNINDEX"}),
    "HNXINDEX": _s("HNXINDEX","HNX-Index","HNX-Index","Chỉ số",{"yfinance":"^HNX"}),

    # ══════════════════════════════════════════════════════════════════════
    # 8. CỔ PHIẾU MỸ (US Stocks)
    # ══════════════════════════════════════════════════════════════════════
    "AAPL":  _s("AAPL","Apple Inc.","Apple","Cổ phiếu Mỹ",{"yfinance":"AAPL"}),
    "MSFT":  _s("MSFT","Microsoft Corp.","Microsoft","Cổ phiếu Mỹ",{"yfinance":"MSFT"}),
    "GOOGL": _s("GOOGL","Alphabet (Google)","Google","Cổ phiếu Mỹ",{"yfinance":"GOOGL"}),
    "AMZN":  _s("AMZN","Amazon.com","Amazon","Cổ phiếu Mỹ",{"yfinance":"AMZN"}),
    "NVDA":  _s("NVDA","Nvidia Corp.","Nvidia","Cổ phiếu Mỹ",{"yfinance":"NVDA"}),
    "META":  _s("META","Meta Platforms","Meta","Cổ phiếu Mỹ",{"yfinance":"META"}),
    "TSLA":  _s("TSLA","Tesla Inc.","Tesla","Cổ phiếu Mỹ",{"yfinance":"TSLA"}),
    "AMD":   _s("AMD","AMD","AMD","Cổ phiếu Mỹ",{"yfinance":"AMD"}),
    "INTC":  _s("INTC","Intel Corp.","Intel","Cổ phiếu Mỹ",{"yfinance":"INTC"}),
    "TSM":   _s("TSM","TSMC","TSMC","Cổ phiếu Mỹ",{"yfinance":"TSM"}),
    "AVGO":  _s("AVGO","Broadcom","Broadcom","Cổ phiếu Mỹ",{"yfinance":"AVGO"}),
    "NFLX":  _s("NFLX","Netflix","Netflix","Cổ phiếu Mỹ",{"yfinance":"NFLX"}),
    "COIN":  _s("COIN","Coinbase","Coinbase","Cổ phiếu Mỹ",{"yfinance":"COIN"}),
    "MSTR":  _s("MSTR","MicroStrategy","MicroStrategy","Cổ phiếu Mỹ",{"yfinance":"MSTR"}),
    "JPM":   _s("JPM","JPMorgan Chase","JPMorgan","Cổ phiếu Mỹ",{"yfinance":"JPM"}),
    "GS":    _s("GS","Goldman Sachs","Goldman Sachs","Cổ phiếu Mỹ",{"yfinance":"GS"}),
    "V":     _s("V","Visa Inc.","Visa","Cổ phiếu Mỹ",{"yfinance":"V"}),
    "MA":    _s("MA","Mastercard","Mastercard","Cổ phiếu Mỹ",{"yfinance":"MA"}),
    "WMT":   _s("WMT","Walmart","Walmart","Cổ phiếu Mỹ",{"yfinance":"WMT"}),
    "XOM":   _s("XOM","ExxonMobil","ExxonMobil","Cổ phiếu Mỹ",{"yfinance":"XOM"}),

    # ══════════════════════════════════════════════════════════════════════
    # 9. CỔ PHIẾU VIỆT NAM (Vietnam Stocks)
    # ══════════════════════════════════════════════════════════════════════
    "VCB":  _s("VCB","Vietcombank (VCB)","Vietcombank","Cổ phiếu VN",{"yfinance":"VCB.VN"}),
    "FPT":  _s("FPT","FPT Corp (FPT)","FPT","Cổ phiếu VN",{"yfinance":"FPT.VN"}),
    "VIC":  _s("VIC","Vingroup (VIC)","Vingroup","Cổ phiếu VN",{"yfinance":"VIC.VN"}),
    "VHM":  _s("VHM","Vinhomes (VHM)","Vinhomes","Cổ phiếu VN",{"yfinance":"VHM.VN"}),
    "HPG":  _s("HPG","Hòa Phát (HPG)","Hoa Phat Steel","Cổ phiếu VN",{"yfinance":"HPG.VN"}),
    "SSI":  _s("SSI","Chứng khoán SSI","SSI","Cổ phiếu VN",{"yfinance":"SSI.VN"}),
    "VND":  _s("VND","VNDirect (VND)","VNDirect","Cổ phiếu VN",{"yfinance":"VND.VN"}),
    "MWG":  _s("MWG","Thế Giới Di Động","MWG","Cổ phiếu VN",{"yfinance":"MWG.VN"}),
    "VNM":  _s("VNM","Vinamilk (VNM)","Vinamilk","Cổ phiếu VN",{"yfinance":"VNM.VN"}),
    "MSN":  _s("MSN","Masan Group","Masan","Cổ phiếu VN",{"yfinance":"MSN.VN"}),
    "GAS":  _s("GAS","PV GAS","PV GAS","Cổ phiếu VN",{"yfinance":"GAS.VN"}),
    "BID":  _s("BID","BIDV Bank","BIDV","Cổ phiếu VN",{"yfinance":"BID.VN"}),
    "CTG":  _s("CTG","VietinBank","VietinBank","Cổ phiếu VN",{"yfinance":"CTG.VN"}),
    "TCB":  _s("TCB","Techcombank","Techcombank","Cổ phiếu VN",{"yfinance":"TCB.VN"}),
    "MBB":  _s("MBB","MB Bank","MB Bank","Cổ phiếu VN",{"yfinance":"MBB.VN"}),
    "STB":  _s("STB","Sacombank","Sacombank","Cổ phiếu VN",{"yfinance":"STB.VN"}),
    "ACB":  _s("ACB","ACB Bank","ACB","Cổ phiếu VN",{"yfinance":"ACB.VN"}),
    "HDB":  _s("HDB","HDBank","HDBank","Cổ phiếu VN",{"yfinance":"HDB.VN"}),
    "VPB":  _s("VPB","VPBank","VPBank","Cổ phiếu VN",{"yfinance":"VPB.VN"}),
    "DGC":  _s("DGC","Hóa chất Đức Giang","DGC","Cổ phiếu VN",{"yfinance":"DGC.VN"}),

    # ══════════════════════════════════════════════════════════════════════
    # 10. TRÁI PHIẾU & ETF
    # ══════════════════════════════════════════════════════════════════════
    "US10Y": _s("US10Y","Trái phiếu Mỹ 10Y","US 10Y Treasury","Trái phiếu",{"yfinance":"^TNX"}),
    "US02Y": _s("US02Y","Trái phiếu Mỹ 2Y","US 2Y Treasury","Trái phiếu",{"yfinance":"^IRX"}),
    "SPY":   _s("SPY","SPDR S&P 500 ETF","SPY","ETF",{"yfinance":"SPY"}),
    "QQQ":   _s("QQQ","Invesco QQQ Trust","QQQ","ETF",{"yfinance":"QQQ"}),
    "DIA":   _s("DIA","Dow Jones ETF","DIA","ETF",{"yfinance":"DIA"}),
}

CRYPTO_QUOTE_SUFFIXES: Tuple[str, ...] = ("USDT", "USDC", "USD", "BTC", "ETH")
HF_CRYPTO_SOURCE_PRIORITY_DAILY: List[str] = [
    "yfinance",
    "twelvedata",
    "finnhub",
    "coingecko",
    "binance",
    "bybit",
]
HF_CRYPTO_SOURCE_PRIORITY_INTRADAY: List[str] = [
    "twelvedata",
    "yfinance",
    "finnhub",
    "coingecko",
    "binance",
    "bybit",
]


def _split_crypto_symbol(symbol: str) -> Tuple[Optional[str], Optional[str]]:
    upper_symbol = symbol.upper()
    for quote in CRYPTO_QUOTE_SUFFIXES:
        if upper_symbol.endswith(quote) and len(upper_symbol) > len(quote):
            return upper_symbol[: -len(quote)], quote
    return None, None


def _get_dynamic_crypto_mappings(symbol: str) -> Dict[str, str]:
    cfg = SYMBOLS[symbol]
    if cfg.category != "Crypto":
        return {}

    base, quote = _split_crypto_symbol(cfg.symbol)
    if not base or not quote:
        return {}

    derived: Dict[str, str] = {}
    normalized_quote = "USD" if quote == "USDT" else quote
    binance_symbol = cfg.mappings.get("binance")

    if "twelvedata" not in cfg.mappings:
        derived["twelvedata"] = f"{base}/{normalized_quote}"

    if normalized_quote == "USD" and "yfinance" not in cfg.mappings:
        derived["yfinance"] = f"{base}-USD"

    if "finnhub" not in cfg.mappings:
        if binance_symbol:
            derived["finnhub"] = f"BINANCE:{binance_symbol}"
        elif quote in {"USD", "USDT", "USDC"}:
            finnhub_quote = "USDT" if quote == "USD" else quote
            derived["finnhub"] = f"BINANCE:{base}{finnhub_quote}"

    return derived


def _get_symbol_mapping(symbol: str, source: str) -> Optional[str]:
    cfg = SYMBOLS[symbol]
    direct_mapping = cfg.mappings.get(source)
    if direct_mapping:
        return direct_mapping
    return _get_dynamic_crypto_mappings(symbol).get(source)


def _has_source_mapping(symbol: str, source: str) -> bool:
    return _get_symbol_mapping(symbol, source) is not None


# ──────────────────────────────────────────────────────────────────────────────
# TTL Cache (unchanged from v3, with improved stats)
# ──────────────────────────────────────────────────────────────────────────────

SYNTHETIC_SYMBOLS: Dict[str, SyntheticSymbolConfig] = {
    "USDX": SyntheticSymbolConfig(
        symbol="USDX",
        scale=43.0,
        alpha=1.0,
        wick_shrink=0.8,
        components=(
            SyntheticComponentSpec(name="EURUSD", mode="inverse", left_symbol="EURUSD"),
            SyntheticComponentSpec(name="GBPUSD", mode="inverse", left_symbol="GBPUSD"),
            SyntheticComponentSpec(name="USDCHF", mode="direct", left_symbol="USDCHF"),
            SyntheticComponentSpec(name="USDJPY", mode="direct", left_symbol="USDJPY"),
            SyntheticComponentSpec(name="USDCAD", mode="direct", left_symbol="USDCAD"),
            SyntheticComponentSpec(name="AUDUSD", mode="inverse", left_symbol="AUDUSD"),
            SyntheticComponentSpec(name="NZDUSD", mode="inverse", left_symbol="NZDUSD"),
        ),
    ),
    "EURX": SyntheticSymbolConfig(
        symbol="EURX",
        scale=43.0,
        alpha=1.0,
        wick_shrink=0.6,
        components=(
            SyntheticComponentSpec(name="EURUSD", mode="direct", left_symbol="EURUSD"),
            SyntheticComponentSpec(name="EURGBP", mode="direct", left_symbol="EURGBP"),
            SyntheticComponentSpec(name="EURCHF", mode="product", left_symbol="EURUSD", right_symbol="USDCHF"),
            SyntheticComponentSpec(name="EURJPY", mode="direct", left_symbol="EURJPY"),
            SyntheticComponentSpec(name="EURCAD", mode="product", left_symbol="EURUSD", right_symbol="USDCAD"),
            SyntheticComponentSpec(name="EURAUD", mode="ratio", left_symbol="EURUSD", right_symbol="AUDUSD"),
            SyntheticComponentSpec(name="EURNZD", mode="ratio", left_symbol="EURUSD", right_symbol="NZDUSD"),
        ),
    ),
    "GBPX": SyntheticSymbolConfig(
        symbol="GBPX",
        scale=43.0,
        alpha=1.0,
        wick_shrink=0.6,
        components=(
            SyntheticComponentSpec(name="GBPUSD", mode="direct", left_symbol="GBPUSD"),
            SyntheticComponentSpec(name="EURGBP", mode="inverse", left_symbol="EURGBP"),
            SyntheticComponentSpec(name="GBPCHF", mode="product", left_symbol="GBPUSD", right_symbol="USDCHF"),
            SyntheticComponentSpec(name="GBPJPY", mode="direct", left_symbol="GBPJPY"),
            SyntheticComponentSpec(name="GBPCAD", mode="product", left_symbol="GBPUSD", right_symbol="USDCAD"),
            SyntheticComponentSpec(name="GBPAUD", mode="ratio", left_symbol="GBPUSD", right_symbol="AUDUSD"),
            SyntheticComponentSpec(name="GBPNZD", mode="ratio", left_symbol="GBPUSD", right_symbol="NZDUSD"),
        ),
    ),
    "CHFX": SyntheticSymbolConfig(
        symbol="CHFX",
        scale=43.0,
        alpha=1.0,
        wick_shrink=0.8,
        components=(
            SyntheticComponentSpec(name="USDCHF", mode="inverse", left_symbol="USDCHF"),
            SyntheticComponentSpec(name="EURCHF", mode="inverse", left_symbol="EURCHF"),
            SyntheticComponentSpec(name="GBPCHF", mode="inverse", left_symbol="GBPCHF"),
            SyntheticComponentSpec(name="CHFJPY", mode="direct", left_symbol="CHFJPY"),
            SyntheticComponentSpec(name="CADCHF", mode="inverse", left_symbol="CADCHF"),
            SyntheticComponentSpec(name="AUDCHF", mode="inverse", left_symbol="AUDCHF"),
            SyntheticComponentSpec(name="NZDCHF", mode="inverse", left_symbol="NZDCHF"),
        ),
    ),
    "JPYX": SyntheticSymbolConfig(
        symbol="JPYX",
        scale=43.0,
        alpha=1.0,
        wick_shrink=1.0,
        components=(
            SyntheticComponentSpec(name="USDJPY", mode="inverse", left_symbol="USDJPY"),
            SyntheticComponentSpec(name="EURJPY", mode="inverse", left_symbol="EURJPY"),
            SyntheticComponentSpec(name="GBPJPY", mode="inverse", left_symbol="GBPJPY"),
            SyntheticComponentSpec(name="CHFJPY", mode="inverse", left_symbol="CHFJPY"),
            SyntheticComponentSpec(name="CADJPY", mode="inverse", left_symbol="CADJPY"),
            SyntheticComponentSpec(name="AUDJPY", mode="inverse", left_symbol="AUDJPY"),
            SyntheticComponentSpec(name="NZDJPY", mode="inverse", left_symbol="NZDJPY"),
        ),
    ),
    "CADX": SyntheticSymbolConfig(
        symbol="CADX",
        scale=44.0,
        alpha=1.0,
        wick_shrink=0.8,
        components=(
            SyntheticComponentSpec(name="USDCAD", mode="inverse", left_symbol="USDCAD"),
            SyntheticComponentSpec(name="EURCAD", mode="inverse", left_symbol="EURCAD"),
            SyntheticComponentSpec(name="GBPCAD", mode="inverse", left_symbol="GBPCAD"),
            SyntheticComponentSpec(name="CADCHF", mode="direct", left_symbol="CADCHF"),
            SyntheticComponentSpec(name="CADJPY", mode="direct", left_symbol="CADJPY"),
            SyntheticComponentSpec(name="AUDCAD", mode="inverse", left_symbol="AUDCAD"),
            SyntheticComponentSpec(name="NZDCAD", mode="inverse", left_symbol="NZDCAD"),
        ),
    ),
    "AUDX": SyntheticSymbolConfig(
        symbol="AUDX",
        scale=44.0,
        alpha=1.0,
        wick_shrink=0.6,
        components=(
            SyntheticComponentSpec(name="AUDUSD", mode="direct", left_symbol="AUDUSD"),
            SyntheticComponentSpec(name="EURAUD", mode="inverse", left_symbol="EURAUD"),
            SyntheticComponentSpec(name="GBPAUD", mode="inverse", left_symbol="GBPAUD"),
            SyntheticComponentSpec(name="AUDCHF", mode="direct", left_symbol="AUDCHF"),
            SyntheticComponentSpec(name="AUDJPY", mode="direct", left_symbol="AUDJPY"),
            SyntheticComponentSpec(name="AUDCAD", mode="direct", left_symbol="AUDCAD"),
            SyntheticComponentSpec(name="AUDNZD", mode="direct", left_symbol="AUDNZD"),
        ),
    ),
    "NZDX": SyntheticSymbolConfig(
        symbol="NZDX",
        scale=44.0,
        alpha=1.0,
        wick_shrink=0.8,
        components=(
            SyntheticComponentSpec(name="NZDUSD", mode="direct", left_symbol="NZDUSD"),
            SyntheticComponentSpec(name="EURNZD", mode="inverse", left_symbol="EURNZD"),
            SyntheticComponentSpec(name="GBPNZD", mode="inverse", left_symbol="GBPNZD"),
            SyntheticComponentSpec(name="NZDCHF", mode="direct", left_symbol="NZDCHF"),
            SyntheticComponentSpec(name="NZDJPY", mode="direct", left_symbol="NZDJPY"),
            SyntheticComponentSpec(name="NZDCAD", mode="direct", left_symbol="NZDCAD"),
            SyntheticComponentSpec(name="AUDNZD", mode="inverse", left_symbol="AUDNZD"),
        ),
    ),
}

def _get_canonical_symbol(sym: str) -> str:
    """Try to find the registry ID for a given symbol or alias."""
    s = sym.upper()
    if s in SYMBOLS:
        return s
    # Try common suffixes
    for suffix in ["USD", "USDT"]:
        if f"{s}{suffix}" in SYMBOLS:
            return f"{s}{suffix}"
    # Try prefix removal
    if s.startswith("BINANCE:") and s[8:] in SYMBOLS:
        return s[8:]
    # Search in mappings
    for reg_id in SYMBOLS:
        for source_name in ("binance", "bybit", "coingecko", "twelvedata", "yfinance", "finnhub"):
            mapping = _get_symbol_mapping(reg_id, source_name)
            if mapping and s == mapping.upper():
                return reg_id
    return s


historical_cache = TTLCache()
forecast_cache   = TTLCache()
ticker_cache     = TTLCache()
ai_verdict_cache = TTLCache()
indicators_cache = TTLCache()
source_history_cache = TTLCache()

_watchlist_ticker_semaphore = asyncio.Semaphore(WATCHLIST_TICKER_CONCURRENCY)
_market_peer_ticker_semaphore = asyncio.Semaphore(MARKET_PEER_TICKER_CONCURRENCY)

# Explicitly clear on startup to ensure fresh v6.1+ format
historical_cache.clear()
forecast_cache.clear()
ticker_cache.clear()
ai_verdict_cache.clear()
indicators_cache.clear()
source_history_cache.clear()


def _cache_prefix(symbol: str, interval: str) -> str:
    return f"{CACHE_VERSION}:{symbol}:{interval}:"


def interval_ttl(interval: str) -> int:
    if interval in {"1m", "5m"}:    return 20
    if interval == "15m":           return 30
    if interval in {"1h", "4h"}:    return 60
    return 900


def forecast_ttl(interval: str) -> int:
    if interval in {"1m", "5m"}:    return 60
    if interval in {"15m", "1h"}:   return 300
    if interval == "4h":            return 600
    return 1800


def indicators_ttl(interval: str) -> int:
    if interval in {"1m", "5m"}:    return 45
    if interval in {"15m", "30m"}:  return 120
    if interval in {"1h", "4h"}:    return 300
    return 900


def verdict_ttl(interval: str) -> int:
    if interval in {"1m", "5m"}:
        return 45
    if interval == "15m":
        return 90
    if interval in {"1h", "4h"}:
        return 300
    return 900


def _analysis_verdict_cache_key(
    symbol: str,
    interval: str,
    analysis: Dict[str, Any],
    forecast_pct: float,
) -> str:
    summary = analysis.get("summary", {})
    payload = {
        "symbol": symbol,
        "interval": interval,
        "summary": {
            "bias": summary.get("bias"),
            "signal": summary.get("signal"),
            "score": round(float(summary.get("score", 0.0) or 0.0), 2),
        },
        "osc_signal": analysis.get("oscillators", {}).get("signal"),
        "ma_signal": analysis.get("moving_averages", {}).get("signal"),
        "mtf": analysis.get("multi_timeframe", {}),
        "forecast_pct": round(float(forecast_pct), 2),
        "rules_version": ai_rule_registry.version,
    }
    digest = hashlib.sha256(
        json.dumps(payload, ensure_ascii=False, sort_keys=True).encode("utf-8")
    ).hexdigest()
    return f"ai_verdict:{CACHE_VERSION}:{symbol}:{interval}:{digest[:24]}"


_HISTORICAL_INFLIGHT: Dict[str, "asyncio.Task[Tuple[List[Dict[str, Any]], str]]"] = {}
_INDICATORS_INFLIGHT: Dict[str, "asyncio.Task[Dict[str, Any]]"] = {}
_FORECAST_INFLIGHT: Dict[str, "asyncio.Task[Dict[str, Any]]"] = {}
_TICKER_INFLIGHT: Dict[str, "asyncio.Task[Dict[str, Any]]"] = {}
_SOURCE_HISTORY_INFLIGHT: Dict[str, "asyncio.Task[List[Dict[str, Any]]]"] = {}


# ──────────────────────────────────────────────────────────────────────────────
# Data Helpers
# ──────────────────────────────────────────────────────────────────────────────
def _parse_timestamp(ts: Any) -> int:
    if isinstance(ts, (int, float, np.integer, np.floating)):
        return int(ts // 1000) if ts > 10_000_000_000 else int(ts)
    if isinstance(ts, str):
        return int(pd.to_datetime(ts, utc=True).timestamp())
    if isinstance(ts, datetime):
        if ts.tzinfo is None:
            ts = ts.replace(tzinfo=timezone.utc)
        return int(ts.timestamp())
    raise ValueError(f"Unsupported timestamp type: {type(ts)}")


def _normalize_ohlcv(records: List[Dict[str, Any]], interval: str = "1h") -> List[Dict[str, Any]]:
    """Parse, validate (zombie-candle filter), deduplicate (timestamp-only)."""
    if interval in {"1w", "1d"}:
        threshold = 0.005
    elif interval in {"4h", "1h"}:
        threshold = 0.001
    else:
        threshold = 0.0001

    # B-1: Dynamic adaptive zombie-candle threshold (BUG-07 Fix)
    # Reduce threshold further for specific low-volatility asset classes
    symbol_cfg = None
    if records and "symbol" in records[0]: # Not always present in raw rows
         pass 

    normalized: List[Dict[str, Any]] = []
    for row in records:
        try:
            t = _parse_timestamp(row["time"])
            o = float(row["open"])
            h = float(row["high"])
            l = float(row["low"])
            c = float(row["close"])
            v = float(row.get("volume") or 0.0)
            if any(math.isnan(x) or math.isinf(x) for x in [o, h, l, c, v]):
                continue
            
            # Adaptive threshold: permit smaller ranges for shorter timeframes
            # and specifically for assets where o > 0.
            # 1e-9 is effectively "almost zero" to catch true dead candles while preserving JPY/Bonds.
            if o > 0 and (h - l) / o < 1e-9:
                continue
                
            normalized.append({"time": t, "open": o, "high": h, "low": l, "close": c, "volume": v})
        except Exception:
            continue

    normalized.sort(key=lambda x: x["time"])
    seen: set = set()
    deduped: List[Dict[str, Any]] = []
    for r in normalized:
        if r["time"] in seen:
            continue
        seen.add(r["time"])
        deduped.append(r)
    return deduped


# ──────────────────────────────────────────────────────────────────────────────
# Data Sources
# ──────────────────────────────────────────────────────────────────────────────

async def fetch_binance(symbol: str, interval: str, limit: int) -> List[Dict[str, Any]]:
    await _rate_limit("binance")
    cfg = SYMBOLS[symbol]
    endpoint_symbol = _get_symbol_mapping(symbol, "binance")
    if not endpoint_symbol:
        raise RuntimeError("No Binance mapping for this symbol")
    
    # Define request details
    endpoint = "/fapi/v1/klines" if cfg.binance_type == "futures" else "/api/v3/klines"
    params = {
        "symbol":   endpoint_symbol,
        "interval": BINANCE_INTERVAL_MAP.get(interval, "1h"),
        "limit":    min(max(limit, 1), 1000),
    }

    # B-2: Endpoint Rotation for HF/Cloud environments
    endpoints_spot = ["https://api.binance.com", "https://api1.binance.com", "https://api2.binance.com", "https://api3.binance.com", "https://data-api.binance.com"]
    endpoints_fapi = ["https://fapi.binance.com"] # fapi usually more restricted, but try first
    
    selected_endpoints = endpoints_fapi if cfg.binance_type == "futures" else endpoints_spot
    
    # If on HF, we know api.binance.com is likely blocked, so we can try data-api or alternates faster
    if settings.is_hf and cfg.binance_type == "spot":
        # Move data-api to the front for HF spot
        selected_endpoints = ["https://data-api.binance.com", "https://api1.binance.com", "https://api2.binance.com", "https://api3.binance.com", "https://api.binance.com"]

    last_error = None
    for base_url in selected_endpoints:
        try:
            cb = source_breakers["binance"]
            if not cb.allow_request():
                continue # Try next endpoint or fall through
                
            async def _do_fetch():
                client = await GlobalHTTPClient.get_client()
                resp = await client.get(f"{base_url}{endpoint}", params=params, timeout=10.0)
                if resp.status_code == 451:
                    logger.warning("[Binance] Endpoint %s blocked (451). Trying next...", base_url)
                    raise RuntimeError("IP Blocked")
                if resp.status_code == 429:
                    raise HTTPException(status_code=429, detail="Binance rate limit")
                resp.raise_for_status()
                cb.record_success()
                return resp.json()

            payload = await _retry(_do_fetch)
            # If successful, parse and return
            parsed = [
                {"time": int(k[0])//1000, "open": k[1], "high": k[2],
                 "low": k[3], "close": k[4], "volume": k[5]}
                for k in payload
            ]
            return _normalize_ohlcv(parsed, interval)[-limit:]
            
        except Exception as ex:
            last_error = ex
            logger.error("[Binance] Failed with %s: %s", base_url, ex)
            continue
            
    raise last_error or HTTPException(status_code=503, detail="Binance all endpoints failed")


async def fetch_bybit(symbol: str, interval: str, limit: int) -> List[Dict[str, Any]]:
    """Bybit V5 kline endpoint — free, no API key."""
    await _rate_limit("bybit")
    endpoint_symbol = _get_symbol_mapping(symbol, "bybit")
    if not endpoint_symbol:
        raise RuntimeError("No Bybit mapping for this symbol")
    bybit_cat = SYMBOLS[symbol].bybit_category
    params = {
        "category": bybit_cat,
        "symbol":   endpoint_symbol,
        "interval": BYBIT_INTERVAL_MAP.get(interval, "60"),
        "limit":    min(max(limit, 1), 1000), # Bybit V5 supports up to 1000
    }
    logger.info("[Bybit] %s %s (cat=%s)", symbol, interval, bybit_cat)

    # B-3: Endpoint Rotation for Bybit
    bybit_endpoints = ["https://api.bybit.com", "https://api.bytick.com", "https://api.bybit.nl"]
    if settings.is_hf:
        # Prefer bytick on HF
        bybit_endpoints = ["https://api.bytick.com", "https://api.bybit.com", "https://api.bybit.nl"]

    last_error = None
    for base_url in bybit_endpoints:
        try:
            cb = source_breakers["bybit"]
            if not cb.allow_request():
                continue
                
            async def _do_fetch():
                client = await GlobalHTTPClient.get_client()
                url = f"{base_url}/v5/market/kline"
                
                headers = {}
                query_params = params.copy()
                
                if settings.bybit_api_key and settings.bybit_api_secret:
                    timestamp = str(int(time.time() * 1000))
                    recv_window = "5000"
                    sorted_params = "&".join([f"{k}={v}" for k, v in sorted(query_params.items())])
                    raw_str = timestamp + settings.bybit_api_key + recv_window + sorted_params
                    signature = hmac.new(settings.bybit_api_secret.encode('utf-8'), 
                                        raw_str.encode('utf-8'), hashlib.sha256).hexdigest()
                    
                    headers = {
                        'X-BAPI-API-KEY': settings.bybit_api_key,
                        'X-BAPI-TIMESTAMP': timestamp,
                        'X-BAPI-SIGN-TYPE': '2',
                        'X-BAPI-RECV-WINDOW': recv_window,
                        'X-BAPI-SIGN': signature
                    }

                resp = await client.get(url, params=query_params, headers=headers, timeout=10.0)
                if resp.status_code == 403:
                    logger.warning("[Bybit] Endpoint %s forbidden (403). Trying next...", base_url)
                    raise RuntimeError("IP Blocked")
                resp.raise_for_status()
                cb.record_success()
                return resp.json()

            data = await _retry(_do_fetch)
            if data.get("retCode", -1) != 0:
                raise RuntimeError(f"Bybit Error: {data}")

            rows = data.get("result", {}).get("list", [])
            parsed = [
                {"time": int(r[0])//1000, "open": r[1], "high": r[2],
                 "low": r[3], "close": r[4], "volume": r[5]}
                for r in rows
            ]
            parsed.reverse()
            return _normalize_ohlcv(parsed, interval)[-limit:]
            
        except Exception as ex:
            last_error = ex
            logger.error("[Bybit] Failed with %s: %s", base_url, ex)
            continue

    raise last_error or HTTPException(status_code=503, detail="Bybit all endpoints failed")


async def fetch_coingecko(symbol: str, interval: str, limit: int) -> List[Dict[str, Any]]:
    """
    CoinGecko OHLCV — free, no API key (30 req/min).
    Days mapping: CoinGecko auto-selects granularity based on days requested.
      1-2 days   → hourly candles
      3-90 days  → daily candles (use for 1h/4h fallback)
      90+ days   → weekly candles
    """
    await _rate_limit("coingecko")
    cg_id = _get_symbol_mapping(symbol, "coingecko") or SYMBOLS[symbol].coingecko_id
    if not cg_id:
        raise RuntimeError("No CoinGecko ID for this symbol")

    # BUG-03: skip sub-hourly crypto if using coingecko (not supported by free OHLC)
    if interval in {"1m", "5m", "15m"}:
        raise RuntimeError(f"CoinGecko does not support {interval} OHLC granularity")

    # Map interval to days
    days_map = {"1h": 14, "4h": 90, "1d": 365, "1w": 730}
    days = days_map.get(interval, 14)
    logger.info("[CoinGecko] %s %s (days=%d)", symbol, interval, days)

    async def _fetch():
        cb = source_breakers["coingecko"]
        if not cb.allow_request():
            raise HTTPException(status_code=503, detail="CoinGecko circuit is OPEN")
            
        try:
            client = await GlobalHTTPClient.get_client()
            resp = await client.get(
                f"https://api.coingecko.com/api/v3/coins/{cg_id}/ohlc",
                params={"vs_currency": "usd", "days": days},
            )
            if resp.status_code == 429:
                cb.record_failure()
                raise HTTPException(status_code=429, detail="CoinGecko rate limit")
            if resp.status_code >= 500:
                cb.record_failure()
                raise HTTPException(status_code=resp.status_code, detail="CoinGecko server error")
            
            cb.record_success()
            return resp.json()
        except Exception as ex:
            cb.record_failure()
            raise ex

    rows = await _retry(_fetch)
    # CoinGecko OHLC: [timestamp_ms, open, high, low, close]
    parsed = [
        {"time": int(r[0])//1000, "open": r[1], "high": r[2],
         "low": r[3], "close": r[4], "volume": 0.0}
        for r in rows
    ]
    return _normalize_ohlcv(parsed, interval)[-limit:]


async def fetch_twelvedata(symbol: str, interval: str, limit: int) -> List[Dict[str, Any]]:
    await _rate_limit("twelvedata")
    endpoint_symbol = _get_symbol_mapping(symbol, "twelvedata")
    if not endpoint_symbol:
        raise RuntimeError("No TwelveData mapping for this symbol")
    params = {
        "symbol":     endpoint_symbol,
        "interval":   TWELVE_INTERVAL_MAP[interval],
        "outputsize": min(max(limit, 30), 5000),
        "apikey":     twelvedata_pool.next_key(),
        "format":     "JSON",
    }
    logger.info("[TwelveData] %s %s", symbol, interval)

    async def _fetch():
        cb = source_breakers["twelvedata"]
        if not cb.allow_request():
            raise HTTPException(status_code=503, detail="TwelveData circuit is OPEN")
            
        try:
            client = await GlobalHTTPClient.get_client()
            resp = await client.get("https://api.twelvedata.com/time_series", params=params)
            if resp.status_code == 429:
                cb.record_failure()
                twelvedata_pool.mark_exhausted(params["apikey"])
                cb.open_for(timeout=1800)
                raise RuntimeError("TwelveData rate limit")
            if resp.status_code >= 500:
                cb.record_failure()
                raise HTTPException(status_code=resp.status_code, detail="TwelveData server error")
            
            cb.record_success()
            return resp.json()
        except Exception as ex:
            cb.record_failure()
            raise ex

    payload = await _retry(_fetch)
    if payload.get("code") == 429:
        cb.open_for(timeout=1800)
        raise RuntimeError(f"TwelveData error: {payload}")
    if "code" in payload and payload.get("code") == 400:
        raise RuntimeError(f"TwelveData error: {payload}")
    values = payload.get("values", [])
    parsed = [
        {"time": v.get("datetime"), "open": v.get("open"), "high": v.get("high"),
         "low": v.get("low"), "close": v.get("close"), "volume": v.get("volume", 0)}
        for v in values
    ]
    parsed.reverse()
    return _normalize_ohlcv(parsed, interval)[-limit:]


async def fetch_finnhub(symbol: str, interval: str, limit: int) -> List[Dict[str, Any]]:
    finnhub_symbol = _get_symbol_mapping(symbol, "finnhub")
    if not finnhub_symbol:
        return []

    cb = source_breakers.get("finnhub")
    if cb and not cb.allow_request():
        return []

    await _rate_limit("finnhub")
    client = await GlobalHTTPClient.get_client()

    async def _call():
        r = await client.get(
            "https://finnhub.io/api/v1/stock/candle",
            params={
                "symbol":   finnhub_symbol,
                "resolution": FINNHUB_RESOLUTION_MAP.get(interval, "D"),
                "count":    limit,
                "token":    finnhub_pool.next_key(),
            },
            timeout=10,
        )
        r.raise_for_status()
        return r.json()

    try:
        d = await _retry(_call)
        if d.get("s") != "ok":
            if cb: cb.record_success()
            return []
        
        if cb: cb.record_success()
        return [
            {"time": t, "open": o, "high": h, "low": l, "close": c, "volume": v}
            for t, o, h, l, c, v in zip(d["t"], d["o"], d["h"], d["l"], d["c"], d["v"])
        ]
    except Exception as ex:
        if cb: cb.record_failure()
        logger.warning("[finnhub] %s failed: %s", symbol, ex)
        return []


def _resample_4h(df: pd.DataFrame) -> pd.DataFrame:
    if df.empty:
        return df
    out = df.resample("4h", label="left", closed="left").agg(
        {"Open": "first", "High": "max", "Low": "min", "Close": "last", "Volume": "sum"}
    )
    return out.dropna(subset=["Open", "High", "Low", "Close"])


async def fetch_yfinance(symbol: str, interval: str, limit: int) -> List[Dict[str, Any]]:
    await _rate_limit("yfinance")
    ticker = _get_symbol_mapping(symbol, "yfinance")
    if not ticker:
        raise RuntimeError("No yfinance mapping for this symbol")
    yf_interval = YF_INTERVAL_MAP[interval]
    period      = YF_PERIOD_MAP[interval]
    logger.info("[yfinance] %s %s", symbol, interval)

    def _download() -> pd.DataFrame:
        # B-5: Robust lookback (v6.0) - ensure EMA200 context
        p = "2y" if interval == "1d" else "60d" if "m" in interval else "1y"
        return yf.download(tickers=ticker, interval=yf_interval, period=p,
                           progress=False, auto_adjust=False, threads=False)

    df = await asyncio.to_thread(_download)
    if df is None or df.empty:
        raise RuntimeError("yfinance: empty dataframe")

    if isinstance(df.columns, pd.MultiIndex):
        df.columns = df.columns.get_level_values(0)

    df.index = (
        df.index.tz_localize("UTC") if df.index.tz is None else df.index.tz_convert("UTC")
    )
    if interval == "4h":
        df = _resample_4h(df)

    parsed = [
        {"time": int(idx.timestamp()), "open": row.get("Open"), "high": row.get("High"),
         "low": row.get("Low"), "close": row.get("Close"), "volume": row.get("Volume", 0)}
        for idx, row in df.iterrows()
    ]
    return _normalize_ohlcv(parsed, interval)[-limit:]


async def fetch_alphavantage(symbol: str, interval: str, limit: int) -> List[Dict[str, Any]]:
    """Alpha Vantage - primarily for Macro/Historical. Needs API key."""
    if not settings.alphavantage_api_key:
        raise ValueError("Alpha Vantage API key missing")
    
    await _rate_limit("alphavantage")
    endpoint_symbol = SYMBOLS[symbol].mappings.get("alphavantage", symbol)
    
    # AV has specific functions for daily/intraday
    func = "TIME_SERIES_INTRADAY" if interval in ["1m", "5m", "15m", "60min"] else "TIME_SERIES_DAILY"
    url = f"https://www.alphavantage.co/query?function={func}&symbol={endpoint_symbol}&apikey={settings.alphavantage_api_key}&outputsize=full"
    if "INTRADAY" in func:
        url += f"&interval={interval if interval != '1h' else '60min'}"

    async with httpx.AsyncClient(timeout=15) as client:
        resp = await client.get(url)
        data = resp.json()
        
        # Parse logic based on AV's weird nested JSON
        key = next((k for k in data.keys() if "Time Series" in k), None)
        if not key:
            raise RuntimeError(f"Alpha Vantage: {data.get('Note', data.get('Information', 'Unknown error'))}")
        
        series = data[key]
        parsed = []
        for ts, vals in series.items():
            parsed.append({
                "time": int(datetime.strptime(ts, "%Y-%m-%d %H:%M:%S" if " " in ts else "%Y-%m-%d").replace(tzinfo=timezone.utc).timestamp()),
                "open": float(vals["1. open"]),
                "high": float(vals["2. high"]),
                "low": float(vals["3. low"]),
                "close": float(vals["4. close"]),
                "volume": float(vals.get("5. volume", 0))
            })
        parsed.sort(key=lambda x: x["time"])
        return _normalize_ohlcv(parsed, interval)[-limit:]


async def _fetch_historical_from_source(
    source: str,
    symbol: str,
    interval: str,
    limit: int,
) -> List[Dict[str, Any]]:
    if source == "binance":
        return await fetch_binance(symbol, interval, limit)
    if source == "bybit":
        return await fetch_bybit(symbol, interval, limit)
    if source == "coingecko":
        return await fetch_coingecko(symbol, interval, limit)
    if source == "twelvedata":
        return await fetch_twelvedata(symbol, interval, limit)
    if source == "finnhub":
        return await fetch_finnhub(symbol, interval, limit)
    if source == "yfinance":
        return await fetch_yfinance(symbol, interval, limit)
    if source == "alphavantage":
        return await fetch_alphavantage(symbol, interval, limit)
    raise ValueError(f"Unsupported source fetcher: {source}")


async def _fetch_historical_from_source_cached(
    source: str,
    symbol: str,
    interval: str,
    limit: int,
) -> List[Dict[str, Any]]:
    """
    Deduplicate repeated component fetches used by synthetic symbols.

    This keeps Real Strength baskets from re-downloading the same FX cross
    multiple times across EURX/GBPX/CHFX/... within one refresh window.
    """
    cache_key = f"source_hist:{CACHE_VERSION}:{source}:{symbol}:{interval}:{limit}"
    cached = source_history_cache.get(cache_key)
    if cached is not None:
        return cached

    inflight_task = _SOURCE_HISTORY_INFLIGHT.get(cache_key)
    if inflight_task is not None:
        return await inflight_task

    task = asyncio.create_task(
        _fetch_historical_from_source(source, symbol, interval, limit),
        name=f"source_hist:{source}:{symbol}:{interval}:{limit}",
    )
    _SOURCE_HISTORY_INFLIGHT[cache_key] = task
    try:
        data = await task
        source_history_cache.set(cache_key, data, ttl_seconds=interval_ttl(interval))
        return data
    finally:
        if _SOURCE_HISTORY_INFLIGHT.get(cache_key) is task:
            _SOURCE_HISTORY_INFLIGHT.pop(cache_key, None)


def _is_synthetic_symbol(symbol: str) -> bool:
    return symbol in SYNTHETIC_SYMBOLS


def _extract_candle_value(candle: Dict[str, Any], field_name: str) -> float:
    return float(candle[field_name])


def _combine_component_candles(
    spec: SyntheticComponentSpec,
    left_candle: Dict[str, Any],
    right_candle: Optional[Dict[str, Any]] = None,
    use_body_only_extrema: bool = False,
) -> Dict[str, float]:
    if spec.mode == "direct":
        open_value = _extract_candle_value(left_candle, "open")
        close_value = _extract_candle_value(left_candle, "close")
        if use_body_only_extrema:
            return {
                "open": open_value,
                "high": max(open_value, close_value),
                "low": min(open_value, close_value),
                "close": close_value,
            }
        return {
            "open": open_value,
            "high": _extract_candle_value(left_candle, "high"),
            "low": _extract_candle_value(left_candle, "low"),
            "close": close_value,
        }

    if spec.mode == "inverse":
        lo = _extract_candle_value(left_candle, "open")
        lh = _extract_candle_value(left_candle, "high")
        ll = _extract_candle_value(left_candle, "low")
        lc = _extract_candle_value(left_candle, "close")
        if min(lo, lh, ll, lc) <= 0:
            raise ValueError(f"Synthetic component {spec.name} has non-positive inverse values")
        open_value = 1.0 / lo
        close_value = 1.0 / lc
        if use_body_only_extrema:
            return {
                "open": open_value,
                "high": max(open_value, close_value),
                "low": min(open_value, close_value),
                "close": close_value,
            }
        return {
            "open": open_value,
            "high": 1.0 / ll,
            "low": 1.0 / lh,
            "close": close_value,
        }

    if right_candle is None:
        raise ValueError(f"Synthetic component {spec.name} requires a right candle")

    lo = _extract_candle_value(left_candle, "open")
    lh = _extract_candle_value(left_candle, "high")
    ll = _extract_candle_value(left_candle, "low")
    lc = _extract_candle_value(left_candle, "close")

    ro = _extract_candle_value(right_candle, "open")
    rh = _extract_candle_value(right_candle, "high")
    rl = _extract_candle_value(right_candle, "low")
    rc = _extract_candle_value(right_candle, "close")

    if spec.mode == "product":
        open_value = lo * ro
        close_value = lc * rc
        if use_body_only_extrema:
            return {
                "open": open_value,
                "high": max(open_value, close_value),
                "low": min(open_value, close_value),
                "close": close_value,
            }
        return {
            "open": open_value,
            "high": lh * rh,
            "low": ll * rl,
            "close": close_value,
        }

    if spec.mode == "ratio":
        if min(ro, rh, rl, rc) <= 0:
            raise ValueError(f"Synthetic component {spec.name} has non-positive divisor values")
        open_value = lo / ro
        close_value = lc / rc
        if use_body_only_extrema:
            return {
                "open": open_value,
                "high": max(open_value, close_value),
                "low": min(open_value, close_value),
                "close": close_value,
            }
        return {
            "open": open_value,
            "high": lh / rl,
            "low": ll / rh,
            "close": close_value,
        }

    raise ValueError(f"Unsupported synthetic component mode: {spec.mode}")


def _weighted_geometric_mean(
    values: List[Tuple[float, float]],
    scale: float,
    alpha: float,
) -> float:
    sum_ln = 0.0
    sum_weight = 0.0
    for value, weight in values:
        if value <= 0 or weight <= 0:
            continue
        sum_ln += weight * math.log(value)
        sum_weight += weight
    if sum_weight <= 0:
        raise ValueError("Synthetic symbol received no valid positive component values")
    return scale * math.exp(alpha * (sum_ln / sum_weight))


def _build_synthetic_ohlc(
    component_values: Dict[str, Dict[str, float]],
    config: SyntheticSymbolConfig,
) -> Dict[str, float]:
    open_value = _weighted_geometric_mean(
        [(component_values[spec.name]["open"], spec.weight) for spec in config.components if spec.enabled],
        config.scale,
        config.alpha,
    )
    close_value = _weighted_geometric_mean(
        [(component_values[spec.name]["close"], spec.weight) for spec in config.components if spec.enabled],
        config.scale,
        config.alpha,
    )
    high_raw = _weighted_geometric_mean(
        [(component_values[spec.name]["high"], spec.weight) for spec in config.components if spec.enabled],
        config.scale,
        config.alpha,
    )
    low_raw = _weighted_geometric_mean(
        [(component_values[spec.name]["low"], spec.weight) for spec in config.components if spec.enabled],
        config.scale,
        config.alpha,
    )

    midpoint = (open_value + close_value) / 2.0
    high_value = midpoint + (high_raw - midpoint) * config.wick_shrink
    low_value = midpoint + (low_raw - midpoint) * config.wick_shrink

    return {
        "open": open_value,
        "high": max(high_value, open_value, close_value),
        "low": min(low_value, open_value, close_value),
        "close": close_value,
    }


async def _build_synthetic_symbol_history(
    symbol: str,
    interval: str,
    fetch_limit: int,
    cache_key: str,
) -> Tuple[List[Dict[str, Any]], str]:
    config = SYNTHETIC_SYMBOLS[symbol]
    headroom = max(50, fetch_limit // 2)
    component_limit = min(2000, fetch_limit + headroom)
    required_symbols = {
        spec.left_symbol
        for spec in config.components
        if spec.enabled
    } | {
        spec.right_symbol
        for spec in config.components
        if spec.enabled and spec.right_symbol
    }

    minimum_required = min(20, fetch_limit)
    preferred_sources = (
        ["yfinance", "twelvedata", "finnhub"]
        if interval in {"1d", "1w"}
        else ["twelvedata", "finnhub", "yfinance"]
    )

    component_rows: Dict[str, List[Dict[str, Any]]] = {}
    component_sources: Dict[str, str] = {}
    common_times: List[int] = []
    last_error_messages: List[str] = []
    use_body_only_extrema = interval not in {"1d", "1w"}

    async def _load_component_rows(
        source_name: str,
        component_symbol: str,
    ) -> Tuple[str, Optional[List[Dict[str, Any]]], Optional[str]]:
        if source_name not in SYMBOLS[component_symbol].mappings:
            return component_symbol, None, f"{component_symbol}: missing {source_name} mapping"

        try:
            rows = await _fetch_historical_from_source_cached(
                source_name,
                component_symbol,
                interval,
                component_limit,
            )
            return component_symbol, rows, None
        except Exception as exc:
            return component_symbol, None, f"{component_symbol}: {exc}"

    for source in preferred_sources:
        current_rows: Dict[str, List[Dict[str, Any]]] = {}
        current_sources: Dict[str, str] = {}
        source_errors: List[str] = []

        component_results = await asyncio.gather(
            *[
                _load_component_rows(source, component_symbol)
                for component_symbol in sorted(required_symbols)
            ]
        )
        for component_symbol, rows, error_message in component_results:
            if error_message:
                source_errors.append(error_message)
                continue
            if rows is None:
                source_errors.append(f"{component_symbol}: empty rows")
                continue
            current_rows[component_symbol] = rows
            current_sources[component_symbol] = source

        if source_errors:
            last_error_messages = source_errors
            continue

        time_sets = [
            {int(row["time"]) for row in rows}
            for rows in current_rows.values()
            if rows
        ]
        current_common_times = sorted(set.intersection(*time_sets)) if time_sets else []
        if len(current_common_times) < minimum_required:
            last_error_messages = [
                f"{source}: aligned={len(current_common_times)} required>={minimum_required}"
            ]
            continue

        component_rows = current_rows
        component_sources = current_sources
        common_times = current_common_times
        break

    if len(common_times) < minimum_required:
        raise HTTPException(
            status_code=502,
            detail={
                "message": f"Insufficient aligned candles to build synthetic symbol {symbol}",
                "errors": last_error_messages,
            },
        )

    candles_by_symbol = {
        component_symbol: {int(row["time"]): row for row in rows}
        for component_symbol, rows in component_rows.items()
    }

    synthetic_rows: List[Dict[str, Any]] = []
    for timestamp in common_times[-fetch_limit:]:
        component_values: Dict[str, Dict[str, float]] = {}
        for spec in config.components:
            if not spec.enabled:
                continue
            left_candle = candles_by_symbol[spec.left_symbol][timestamp]
            right_candle = (
                candles_by_symbol[spec.right_symbol][timestamp]
                if spec.right_symbol
                else None
            )
            component_values[spec.name] = _combine_component_candles(
                spec,
                left_candle,
                right_candle,
                use_body_only_extrema=use_body_only_extrema,
            )

        synthetic_ohlc = _build_synthetic_ohlc(component_values, config)
        synthetic_rows.append(
            {
                "time": timestamp,
                "open": round(float(synthetic_ohlc["open"]), 8),
                "high": round(float(synthetic_ohlc["high"]), 8),
                "low": round(float(synthetic_ohlc["low"]), 8),
                "close": round(float(synthetic_ohlc["close"]), 8),
                "volume": 0.0,
            }
        )

    source = "synthetic:" + ",".join(sorted(set(component_sources.values())))
    historical_cache.set(cache_key, (synthetic_rows, source), ttl_seconds=interval_ttl(interval))
    return synthetic_rows, source


def _get_source_priority(symbol: str, interval: Optional[str] = None) -> List[str]:
    cfg = SYMBOLS[symbol]
    priority = CATEGORY_SOURCE_PRIORITY.get(cfg.category, DEFAULT_SOURCE_PRIORITY)
    if settings.is_hf and cfg.category == "Crypto":
        priority = (
            HF_CRYPTO_SOURCE_PRIORITY_DAILY
            if interval in {"1d", "1w"}
            else HF_CRYPTO_SOURCE_PRIORITY_INTRADAY
        )
    return [source_name for source_name in priority if _has_source_mapping(symbol, source_name)]


async def _run_historical_fetch(
    symbol: str,
    interval: str,
    fetch_limit: int,
    cache_key: str,
) -> Tuple[List[Dict[str, Any]], str]:
    if _is_synthetic_symbol(symbol):
        return await _build_synthetic_symbol_history(symbol, interval, fetch_limit, cache_key)

    priority = _get_source_priority(symbol, interval)
    errors: List[str] = []

    for source in priority:
        try:
            if source == "binance":
                data = await fetch_binance(symbol, interval, fetch_limit)
            elif source == "bybit":
                data = await fetch_bybit(symbol, interval, fetch_limit)
            elif source == "coingecko":
                data = await fetch_coingecko(symbol, interval, fetch_limit)
            elif source == "twelvedata":
                data = await fetch_twelvedata(symbol, interval, fetch_limit)
            elif source == "finnhub":
                data = await fetch_finnhub(symbol, interval, fetch_limit)
            elif source == "yfinance":
                data = await fetch_yfinance(symbol, interval, fetch_limit)
            elif source == "alphavantage":
                data = await fetch_alphavantage(symbol, interval, fetch_limit)
            else:
                continue

            if len(data) >= 20:
                historical_cache.set(cache_key, (data, source), ttl_seconds=interval_ttl(interval))
                return data, source
            errors.append(f"{source}: insufficient ({len(data)} candles)")
        except HTTPException as ex:
            errors.append(f"{source}: HTTP {ex.status_code} {ex.detail}")
            logger.warning(
                "[fetch_historical] %s/%s %s HTTP %s: %s",
                symbol,
                interval,
                source,
                ex.status_code,
                ex.detail,
            )
            continue
        except Exception as ex:
            errors.append(f"{source}: {ex}")
            logger.warning("[fetch_historical] %s/%s %s: %s", symbol, interval, source, ex)

    raise HTTPException(
        status_code=502,
        detail={"message": f"All sources failed: {symbol}/{interval}", "errors": errors},
    )


async def fetch_historical(
    symbol: str,
    interval: str,
    limit: int,
    refresh: bool = False,
    min_context: Optional[int] = None,
) -> Tuple[List[Dict[str, Any]], str]:
    """
    Fetch OHLCV data with fallback and caching.
    v6.2: Fetch only the context each caller actually needs while keeping
    AI routes supplied with deeper history for indicator and Kronos quality.
    """
    prefix = _cache_prefix(symbol, interval)
    key    = f"hist_{prefix}"  # BUG-P1-03: No limit in key to increase cache hits
    required_context = max(limit, min_context or limit)
    if refresh:
        historical_cache.delete(key)
    else:
        cached = historical_cache.get(key)
        if cached is not None:
            try:
                # v6.1: Cache stores (data, source)
                data_cached, source_cached = cached
                if len(data_cached) >= required_context:
                    return data_cached[-limit:], source_cached
            except (ValueError, TypeError):
                # Handle old cache format gracefully
                historical_cache.delete(key)

    fetch_limit = required_context
    inflight_key = f"{key}:refresh={int(refresh)}:fetch={fetch_limit}"
    task = None if refresh else _HISTORICAL_INFLIGHT.get(inflight_key)
    if task is None:
        task = asyncio.create_task(
            _run_historical_fetch(symbol, interval, fetch_limit, key),
            name=f"hist:{symbol}:{interval}:{fetch_limit}",
        )
        _HISTORICAL_INFLIGHT[inflight_key] = task
    try:
        data, source = await task
        return data[-limit:], source
    finally:
        if _HISTORICAL_INFLIGHT.get(inflight_key) is task:
            _HISTORICAL_INFLIGHT.pop(inflight_key, None)


# ──────────────────────────────────────────────────────────────────────────────
# Real-time Ticker (last price + 24h stats)
# ──────────────────────────────────────────────────────────────────────────────
async def fetch_ticker(symbol: str, interval: Optional[str] = None) -> Dict[str, Any]:
    interval_key = interval or "default"
    cache_key = f"ticker:{symbol}:{interval_key}"
    cached = ticker_cache.get(cache_key)
    if cached:
        return cached

    inflight_key = cache_key
    inflight_task = _TICKER_INFLIGHT.get(inflight_key)
    if inflight_task is not None:
        return await inflight_task

    async def _run() -> Dict[str, Any]:
        if _is_synthetic_symbol(symbol):
            synthetic_interval = interval or "1d"
            rows, source = await fetch_historical(symbol, synthetic_interval, 2, min_context=2)
            current_row = rows[-1]
            previous_row = rows[-2] if len(rows) > 1 else current_row
            current_price = float(current_row["close"])
            previous_close = float(previous_row["close"])
            high_24h = float(current_row["high"])
            low_24h = float(current_row["low"])
            change_value = current_price - previous_close
            result = {
                "symbol": symbol,
                "price": current_price,
                "change": change_value,
                "change_pct": ((change_value / previous_close) * 100.0) if previous_close else 0.0,
                "high_24h": high_24h,
                "low_24h": low_24h,
                "volume_24h": 0.0,
                "source": source,
                "timestamp": int(time.time()),
            }
            ticker_cache.set(cache_key, result, ttl_seconds=10)
            return result

        cfg = SYMBOLS[symbol]
        priority = _get_source_priority(symbol, interval=interval)
        client = await GlobalHTTPClient.get_client()

        for source in priority:
            try:
                if source == "binance":
                    binance_symbol = _get_symbol_mapping(symbol, "binance")
                    if not binance_symbol:
                        continue
                    await _rate_limit("binance")
                    base_url = "https://fapi.binance.com" if cfg.binance_type == "futures" else "https://api.binance.com"
                    endpoint = "/fapi/v1/ticker/24hr" if cfg.binance_type == "futures" else "/api/v3/ticker/24hr"
                    r = await client.get(
                        f"{base_url}{endpoint}",
                        params={"symbol": binance_symbol},
                        timeout=10.0,
                    )
                    d = r.json()
                    res = {
                        "price": float(d.get("lastPrice") or d.get("price") or 0),
                        "change": float(d.get("priceChange", 0)),
                        "change_pct": float(d.get("priceChangePercent", 0)),
                        "high_24h": float(d.get("highPrice", 0)),
                        "low_24h": float(d.get("lowPrice", 0)),
                        "volume_24h": float(d.get("volume", 0)),
                        "source": "binance",
                    }
                elif source == "twelvedata":
                    twelvedata_symbol = _get_symbol_mapping(symbol, "twelvedata")
                    if not twelvedata_symbol:
                        continue
                    await _rate_limit("twelvedata")
                    r = await client.get(
                        "https://api.twelvedata.com/quote",
                        params={"symbol": twelvedata_symbol, "apikey": twelvedata_pool.next_key()},
                        timeout=10.0,
                    )
                    d = r.json()
                    if "close" not in d and "price" not in d:
                        continue
                    p = float(d.get("price") or d.get("close") or 0)
                    pc = float(d.get("previous_close") or p)
                    res = {
                        "price": p,
                        "change": p - pc,
                        "change_pct": ((p - pc) / pc * 100) if pc != 0 else 0,
                        "high_24h": float(d.get("high") or p),
                        "low_24h": float(d.get("low") or p),
                        "volume_24h": float(d.get("volume") or 0),
                        "source": "twelvedata",
                    }
                elif source == "bybit":
                    bybit_symbol = _get_symbol_mapping(symbol, "bybit")
                    if not bybit_symbol:
                        continue
                    await _rate_limit("bybit")
                    url = "https://api.bybit.com/v5/market/tickers"
                    query_params = {"category": cfg.bybit_category, "symbol": bybit_symbol}
                    headers = {}

                    if settings.bybit_api_key and settings.bybit_api_secret:
                        timestamp = str(int(time.time() * 1000))
                        recv_window = "5000"
                        sorted_params = "&".join([f"{k}={v}" for k, v in sorted(query_params.items())])
                        raw_str = timestamp + settings.bybit_api_key + recv_window + sorted_params
                        signature = hmac.new(
                            settings.bybit_api_secret.encode("utf-8"),
                            raw_str.encode("utf-8"),
                            hashlib.sha256,
                        ).hexdigest()
                        headers = {
                            "X-BAPI-API-KEY": settings.bybit_api_key,
                            "X-BAPI-TIMESTAMP": timestamp,
                            "X-BAPI-SIGN-TYPE": "2",
                            "X-BAPI-RECV-WINDOW": recv_window,
                            "X-BAPI-SIGN": signature,
                        }

                    r = await client.get(url, params=query_params, headers=headers, timeout=10.0)
                    d = r.json()
                    if d.get("retCode") != 0 or not d.get("result", {}).get("list"):
                        continue
                    t = d["result"]["list"][0]
                    res = {
                        "price": float(t["lastPrice"]),
                        "change": float(t["price24hPcnt"]) * float(t["lastPrice"]) / 100,
                        "change_pct": float(t["price24hPcnt"]) * 100,
                        "high_24h": float(t["highPrice24h"]),
                        "low_24h": float(t["lowPrice24h"]),
                        "volume_24h": float(t["volume24h"]),
                        "source": "bybit",
                    }
                elif source == "yfinance":
                    yfinance_symbol = _get_symbol_mapping(symbol, "yfinance")
                    if not yfinance_symbol:
                        continue
                    def _yf_info():
                        ticker = yf.Ticker(yfinance_symbol)
                        history = ticker.history(period="5d")
                        if history.empty:
                            return None
                        last_price = float(history["Close"].iloc[-1])
                        prev_close = float(history["Close"].iloc[-2]) if len(history) > 1 else last_price
                        return {
                            "price": last_price,
                            "change": last_price - prev_close,
                            "change_pct": ((last_price - prev_close) / prev_close * 100) if prev_close != 0 else 0,
                            "high_24h": float(history["High"].iloc[-1]),
                            "low_24h": float(history["Low"].iloc[-1]),
                            "volume_24h": float(history["Volume"].iloc[-1]),
                            "source": "yfinance",
                        }

                    res = await asyncio.to_thread(_yf_info)
                    if not res:
                        continue
                else:
                    continue

                res.update({"symbol": symbol, "timestamp": int(time.time())})
                ttl = 5 if cfg.category == "Crypto" else (10 if cfg.category in ("Cặp tiền", "Chỉ số", "Real Strength") else 30)
                ticker_cache.set(cache_key, res, ttl_seconds=ttl)
                return res
            except Exception as ex:
                logger.debug("[ticker] %s/%s failed: %s", symbol, source, ex)
                continue

        raise HTTPException(status_code=502, detail=f"Ticker failed for {symbol} after trying {priority}")

    task = asyncio.create_task(_run(), name=f"ticker:{symbol}:{interval_key}")
    _TICKER_INFLIGHT[inflight_key] = task
    try:
        return await task
    finally:
        if _TICKER_INFLIGHT.get(inflight_key) is task:
            _TICKER_INFLIGHT.pop(inflight_key, None)



# ──────────────────────────────────────────────────────────────────────────────
# Technical Indicators
# ──────────────────────────────────────────────────────────────────────────────
# D-2: High-Performance Vectorized Indicators (NumPy)
def _ema(arr: np.ndarray, period: int) -> np.ndarray:
    """Vectorized EMA using NumPy (replaces loops)."""
    if len(arr) == 0: return np.array([], dtype=float)
    alpha = 2.0 / (period + 1.0)
    # Use pandas ewm for robust vectorized calculation (v6.0)
    return pd.Series(arr).ewm(alpha=alpha, adjust=False).mean().values

def _rsi(close: np.ndarray, period: int = 14) -> np.ndarray:
    """Vectorized RSI using NumPy/Pandas."""
    delta = np.diff(close)
    gain = np.where(delta > 0, delta, 0.0)
    loss = np.where(delta < 0, -delta, 0.0)
    
    avg_gain = pd.Series(gain).ewm(alpha=1.0/period, adjust=False).mean()
    avg_loss = pd.Series(loss).ewm(alpha=1.0/period, adjust=False).mean()

    rs = avg_gain / avg_loss.replace(0, np.nan)
    rsi = 100 - (100 / (1 + rs))
    rsi = rsi.where(avg_loss > 0, 100.0)
    rsi = rsi.where(avg_gain > 0, 0.0)
    rsi = rsi.where(~((avg_gain == 0) & (avg_loss == 0)), 50.0)
    # Prepend NaN to match original array length
    return np.concatenate([[np.nan], rsi.values])

def _bollinger(close: np.ndarray, period=20, k=2.0) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
    """Vectorized Bollinger Bands."""
    s = pd.Series(close)
    mid = s.rolling(window=period).mean()
    std = s.rolling(window=period).std(ddof=0)
    return (mid + k*std).values, mid.values, (mid - k*std).values


def _macd(close: np.ndarray, fast=12, slow=26, signal=9
          ) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
    ema_fast = _ema(close, fast)
    ema_slow = _ema(close, slow)
    macd_line = ema_fast - ema_slow
    sig_line  = _ema(np.where(np.isnan(macd_line), 0, macd_line), signal)
    histogram  = macd_line - sig_line
    return macd_line, sig_line, histogram


def _atr(high: np.ndarray, low: np.ndarray, close: np.ndarray, period=14) -> np.ndarray:
    """Vectorized ATR."""
    tr = np.maximum(high[1:] - low[1:], 
                    np.maximum(np.abs(high[1:] - close[:-1]), 
                               np.abs(low[1:] - close[:-1])))
    tr = np.concatenate([[np.nan], tr])
    return pd.Series(tr).ewm(alpha=1.0/period, adjust=False).mean().values


def _stoch_rsi(close: np.ndarray, rsi_period=14, stoch_period=14,
               smooth_k=3, smooth_d=3) -> Tuple[np.ndarray, np.ndarray]:
    """Vectorized Stochastic RSI."""
    rsi_vals = pd.Series(_rsi(close, rsi_period))
    roll_min = rsi_vals.rolling(window=stoch_period).min()
    roll_max = rsi_vals.rolling(window=stoch_period).max()
    
    k = 100 * (rsi_vals - roll_min) / (roll_max - roll_min).replace(0, np.inf)
    k_smooth = k.rolling(window=smooth_k).mean()
    d_smooth = k_smooth.rolling(window=smooth_d).mean()
    return k_smooth.values, d_smooth.values


def _stoch_kd(
    high: np.ndarray,
    low: np.ndarray,
    close: np.ndarray,
    k_period: int = 14,
    smooth_k: int = 3,
    d_period: int = 3,
) -> Tuple[np.ndarray, np.ndarray]:
    """Standard stochastic oscillator computed from price high/low/close."""
    high_s = pd.Series(high)
    low_s = pd.Series(low)
    close_s = pd.Series(close)
    hh = high_s.rolling(window=k_period).max()
    ll = low_s.rolling(window=k_period).min()
    raw_k = 100.0 * (close_s - ll) / (hh - ll).replace(0, np.nan)
    k = raw_k.rolling(window=smooth_k).mean()
    d = k.rolling(window=d_period).mean()
    return k.values, d.values


def _sma(arr: np.ndarray, period: int) -> np.ndarray:
    """Vectorized Simple Moving Average."""
    if len(arr) == 0: return np.array([], dtype=float)
    return pd.Series(arr).rolling(window=period).mean().values


def _cci(high: np.ndarray, low: np.ndarray, close: np.ndarray, period: int = 20) -> np.ndarray:
    """Vectorized Commodity Channel Index."""
    tp = (high + low + close) / 3.0
    tp_series = pd.Series(tp)
    sma = tp_series.rolling(window=period).mean()
    # Optimized MAD calculation (Vectorized)
    mad = tp_series.rolling(window=period).apply(lambda x: np.abs(x - x.mean()).mean(), raw=True)
    return ((tp_series - sma) / (0.015 * mad.replace(0, np.inf))).values


def _adx(high: np.ndarray, low: np.ndarray, close: np.ndarray, period: int = 14) -> tuple:
    """Vectorized Average Directional Index (v6.0)."""
    plus_dm = np.where((high[1:] - high[:-1] > low[:-1] - low[1:]) & (high[1:] - high[:-1] > 0), high[1:] - high[:-1], 0.0)
    minus_dm = np.where((low[:-1] - low[1:] > high[1:] - high[:-1]) & (low[:-1] - low[1:] > 0), low[:-1] - low[1:], 0.0)
    tr = np.maximum(high[1:] - low[1:], np.maximum(np.abs(high[1:] - close[:-1]), np.abs(low[1:] - close[:-1])))
    
    # Pad first index
    plus_dm = np.concatenate([[0.0], plus_dm])
    minus_dm = np.concatenate([[0.0], minus_dm])
    tr = np.concatenate([[0.0], tr])
    
    tr_sum = pd.Series(tr).ewm(alpha=1.0/period, adjust=False).mean()
    plus_di = 100 * pd.Series(plus_dm).ewm(alpha=1.0/period, adjust=False).mean() / tr_sum.replace(0, np.inf)
    minus_di = 100 * pd.Series(minus_dm).ewm(alpha=1.0/period, adjust=False).mean() / tr_sum.replace(0, np.inf)
    
    dx = 100 * np.abs(plus_di - minus_di) / (plus_di + minus_di).replace(0, np.inf)
    adx = dx.ewm(alpha=1.0/period, adjust=False).mean()
    return adx.values, plus_di.values, minus_di.values


def _awesome_oscillator(high: np.ndarray, low: np.ndarray) -> np.ndarray:
    """Awesome Oscillator = SMA(5, median) - SMA(34, median)."""
    median = (high + low) / 2.0
    sma5 = _sma(median, 5)
    sma34 = _sma(median, 34)
    return sma5 - sma34


def _momentum(close: np.ndarray, period: int = 10) -> np.ndarray:
    """Vectorized Momentum Index centered at 100."""
    base = pd.Series(close).shift(period)
    return (pd.Series(close) / base.replace(0, np.nan) * 100.0).values


def _roc(close: np.ndarray, period: int = 12) -> np.ndarray:
    """Rate of Change in percentage."""
    base = pd.Series(close).shift(period)
    return (pd.Series(close) / base.replace(0, np.nan) - 1.0).mul(100.0).values


def _trix(close: np.ndarray, period: int = 18) -> np.ndarray:
    """Triple EMA oscillator in percentage."""
    ema1 = pd.Series(_ema(close, period))
    ema2 = ema1.ewm(span=period, adjust=False).mean()
    ema3 = ema2.ewm(span=period, adjust=False).mean()
    return ema3.pct_change().mul(100.0).values


def _ppo(close: np.ndarray, fast: int = 12, slow: int = 26) -> np.ndarray:
    """Percentage Price Oscillator."""
    ema_fast = _ema(close, fast)
    ema_slow = _ema(close, slow)
    return ((ema_fast - ema_slow) / np.where(np.abs(ema_slow) < 1e-8, np.nan, ema_slow) * 100.0)


def _cmo(close: np.ndarray, period: int = 14) -> np.ndarray:
    """Chande Momentum Oscillator."""
    delta = pd.Series(close).diff()
    up = delta.clip(lower=0.0).rolling(period).sum()
    down = (-delta.clip(upper=0.0)).rolling(period).sum()
    return ((up - down) / (up + down).replace(0, np.nan) * 100.0).values


def _dpo(close: np.ndarray, period: int = 20) -> np.ndarray:
    """Detrended Price Oscillator."""
    offset = int(period / 2) + 1
    sma = pd.Series(close).rolling(period).mean()
    return (pd.Series(close) - sma.shift(offset)).values


def _aroon_oscillator(high: np.ndarray, low: np.ndarray, period: int = 25) -> np.ndarray:
    """Aroon Oscillator = Aroon Up - Aroon Down."""
    hs = pd.Series(high)
    ls = pd.Series(low)
    aroon_up = hs.rolling(period).apply(lambda x: ((period - 1 - (len(x) - 1 - int(np.argmax(x)))) / (period - 1)) * 100.0, raw=True)
    aroon_down = ls.rolling(period).apply(lambda x: ((period - 1 - (len(x) - 1 - int(np.argmin(x)))) / (period - 1)) * 100.0, raw=True)
    return (aroon_up - aroon_down).values


def _tsi(close: np.ndarray, long_period: int = 25, short_period: int = 13) -> np.ndarray:
    """True Strength Index."""
    delta = pd.Series(close).diff()
    abs_delta = delta.abs()
    ema1 = delta.ewm(span=long_period, adjust=False).mean()
    ema2 = ema1.ewm(span=short_period, adjust=False).mean()
    abs_ema1 = abs_delta.ewm(span=long_period, adjust=False).mean()
    abs_ema2 = abs_ema1.ewm(span=short_period, adjust=False).mean()
    return (ema2 / abs_ema2.replace(0, np.nan) * 100.0).values


def _demarker(high: np.ndarray, low: np.ndarray, period: int = 14) -> np.ndarray:
    """DeMarker oscillator in range 0..100."""
    high_delta = pd.Series(high).diff()
    low_delta = -pd.Series(low).diff()
    demax = high_delta.clip(lower=0.0)
    demin = low_delta.clip(lower=0.0)
    demax_avg = demax.rolling(period).sum()
    demin_avg = demin.rolling(period).sum()
    denom = (demax_avg + demin_avg).replace(0, np.nan)
    return (demax_avg / denom * 100.0).values


def _force_index(close: np.ndarray, volume: np.ndarray, period: int = 13) -> np.ndarray:
    """Force Index smoothed with an EMA."""
    raw = pd.Series(close).diff() * pd.Series(volume)
    return raw.ewm(span=period, adjust=False).mean().values


def _williams_r(high: np.ndarray, low: np.ndarray, close: np.ndarray,
                period: int = 14) -> np.ndarray:
    """Vectorized Williams %R."""
    s = pd.Series(close)
    hh = pd.Series(high).rolling(window=period).max()
    ll = pd.Series(low).rolling(window=period).min()
    wr = -100 * (hh - s) / (hh - ll).replace(0, np.inf)
    return wr.values


def _bull_bear_power(high: np.ndarray, low: np.ndarray, close: np.ndarray,
                     period: int = 13) -> np.ndarray:
    """Vectorized Bull Bear Power (v6.0)."""
    ema_val = _ema(close, period)
    return (high - ema_val) + (low - ema_val)


def _ultimate_oscillator(high: np.ndarray, low: np.ndarray, close: np.ndarray,
                         p1: int = 7, p2: int = 14, p3: int = 28) -> np.ndarray:
    """Vectorized Ultimate Oscillator (v6.0)."""
    n = len(close)
    if n < p3 + 1: return np.full(n, np.nan)

    prev_close = pd.Series(close).shift(1)
    tl = np.minimum(low, prev_close.values)
    th = np.maximum(high, prev_close.values)
    
    bp = pd.Series(close - tl)
    tr = pd.Series(th - tl)
    
    def _avg(p):
        return bp.rolling(p).sum() / tr.rolling(p).sum().replace(0, np.inf)

    avg1 = _avg(p1)
    avg2 = _avg(p2)
    avg3 = _avg(p3)
    
    uo = 100 * (4 * avg1 + 2 * avg2 + avg3) / 7.0
    return uo.values


def _ichimoku_base(high: np.ndarray, low: np.ndarray, period: int = 26) -> np.ndarray:
    """Vectorized Ichimoku Base Line."""
    hh = pd.Series(high).rolling(window=period).max()
    ll = pd.Series(low).rolling(window=period).min()
    return ((hh + ll) / 2.0).values


def _ichimoku_cloud(high: np.ndarray, low: np.ndarray) -> Dict[str, np.ndarray]:
    """Basic Ichimoku lines for cloud-position scoring."""
    hs = pd.Series(high)
    ls = pd.Series(low)
    tenkan = ((hs.rolling(9).max() + ls.rolling(9).min()) / 2.0).values
    kijun = ((hs.rolling(26).max() + ls.rolling(26).min()) / 2.0).values
    span_a = ((pd.Series(tenkan) + pd.Series(kijun)) / 2.0).values
    span_b = ((hs.rolling(52).max() + ls.rolling(52).min()) / 2.0).values
    return {
        "tenkan": tenkan,
        "kijun": kijun,
        "span_a": span_a,
        "span_b": span_b,
    }


def _vwma(close: np.ndarray, volume: np.ndarray, period: int = 20) -> np.ndarray:
    """Vectorized Volume Weighted Moving Average."""
    cv = pd.Series(close * volume)
    v  = pd.Series(volume)
    return (cv.rolling(period).sum() / v.rolling(period).sum().replace(0, np.inf)).values


def _wma(arr: np.ndarray, period: int) -> np.ndarray:
    """Weighted moving average used by Hull MA."""
    if len(arr) == 0:
        return np.array([], dtype=float)
    weights = np.arange(1, period + 1, dtype=float)
    return pd.Series(arr).rolling(window=period).apply(
        lambda x: float(np.dot(x, weights) / weights.sum()),
        raw=True,
    ).values


def _hull_ma(close: np.ndarray, period: int = 9) -> np.ndarray:
    """Hull Moving Average."""
    half = max(period // 2, 1)
    sqrt_p = max(int(math.sqrt(period)), 1)
    wma_half = _wma(close, half)
    wma_full = _wma(close, period)
    diff = 2 * wma_half - wma_full
    hull = _wma(np.where(np.isnan(diff), close, diff), sqrt_p)
    return hull


def _calc_pivot_points(high_val: float, low_val: float, close_val: float
                       ) -> Dict[str, Any]:
    """Calculate pivot points using 5 methods: Classic, Fibonacci, Camarilla, Woodie, DM."""
    p_classic = (high_val + low_val + close_val) / 3.0
    r1_c = 2 * p_classic - low_val
    s1_c = 2 * p_classic - high_val
    r2_c = p_classic + (high_val - low_val)
    s2_c = p_classic - (high_val - low_val)
    r3_c = high_val + 2 * (p_classic - low_val)
    s3_c = low_val - 2 * (high_val - p_classic)

    # Fibonacci
    diff = high_val - low_val
    r1_f = p_classic + 0.382 * diff
    r2_f = p_classic + 0.618 * diff
    r3_f = p_classic + 1.0 * diff
    s1_f = p_classic - 0.382 * diff
    s2_f = p_classic - 0.618 * diff
    s3_f = p_classic - 1.0 * diff

    # Camarilla
    r1_cam = close_val + diff * 1.1 / 12
    r2_cam = close_val + diff * 1.1 / 6
    r3_cam = close_val + diff * 1.1 / 4
    s1_cam = close_val - diff * 1.1 / 12
    s2_cam = close_val - diff * 1.1 / 6
    s3_cam = close_val - diff * 1.1 / 4

    # Woodie
    p_w = (high_val + low_val + 2 * close_val) / 4.0
    r1_w = 2 * p_w - low_val
    s1_w = 2 * p_w - high_val
    r2_w = p_w + diff
    s2_w = p_w - diff
    r3_w = high_val + 2 * (p_w - low_val)
    s3_w = low_val - 2 * (high_val - p_w)

    # DM (Demark)
    if close_val < p_classic:
        x = high_val + 2 * low_val + close_val
    elif close_val > p_classic:
        x = 2 * high_val + low_val + close_val
    else:
        x = high_val + low_val + 2 * close_val
    p_dm = x / 4.0
    r1_dm = x / 2.0 - low_val
    s1_dm = x / 2.0 - high_val

    def _r(v):
        return round(float(v), 2)

    levels = []
    for lbl, cl, fb, cm, wd, dm_r, dm_s in [
        ("R3", r3_c, r3_f, r3_cam, r3_w, None, None),
        ("R2", r2_c, r2_f, r2_cam, r2_w, None, None),
        ("R1", r1_c, r1_f, r1_cam, r1_w, r1_dm, None),
        ("P",  p_classic, p_classic, p_classic, p_w, p_dm, None),
        ("S1", s1_c, s1_f, s1_cam, s1_w, None, s1_dm),
        ("S2", s2_c, s2_f, s2_cam, s2_w, None, None),
        ("S3", s3_c, s3_f, s3_cam, s3_w, None, None),
    ]:
        row = {
            "level": lbl,
            "classic": _r(cl),
            "fibonacci": _r(fb),
            "camarilla": _r(cm),
            "woodie": _r(wd),
        }
        if lbl == "R1":
            row["dm"] = _r(r1_dm)
        elif lbl == "P":
            row["dm"] = _r(p_dm)
        elif lbl == "S1":
            row["dm"] = _r(s1_dm)
        else:
            row["dm"] = None
        levels.append(row)

    return {"data": levels}


# ─────────────────────────────────────────────────────────────────────────────
# TradingView-style action classification helpers
# ─────────────────────────────────────────────────────────────────────────────

def _osc_action(name: str, value: float, **kw) -> str:
    """Classify oscillator value as 'Mua' / 'Bán' / 'Trung lập'."""
    if value is None or math.isnan(value):
        return "Trung lập"
    atr = max(float(kw.get("atr", 0.0) or 0.0), 1e-8)

    def _bounded_action(
        current: float,
        min_value: float,
        max_value: float,
        *,
        bullish_high: bool = True,
    ) -> str:
        span = max(max_value - min_value, 1e-8)
        ratio = _clamp((current - min_value) / span, 0.0, 1.0)
        if not bullish_high:
            ratio = 1.0 - ratio
        if ratio > (2.0 / 3.0):
            return "Mua"
        if ratio < (1.0 / 3.0):
            return "Bán"
        return "Trung lập"

    if name == "rsi":
        return _bounded_action(value, 0.0, 100.0)
    if name in {"stoch", "stoch_rsi", "ultimate", "demarker"}:
        return _bounded_action(value, 0.0, 100.0)
    if name == "williams":
        return _bounded_action(value, -100.0, 0.0, bullish_high=True)
    if name == "cci":
        return "Mua" if value > 100 else "Bán" if value < -100 else "Trung lập"
    if name == "momentum":
        return "Mua" if value > 100 else "Bán" if value < 100 else "Trung lập"
    if name == "roc":
        return "Mua" if value > 1.0 else "Bán" if value < -1.0 else "Trung lập"
    if name == "trix":
        return "Mua" if value > 0.0 else "Bán" if value < 0.0 else "Trung lập"
    if name == "ppo":
        return "Mua" if value > 0.35 else "Bán" if value < -0.35 else "Trung lập"
    if name == "cmo":
        return "Mua" if value > 33.0 else "Bán" if value < -33.0 else "Trung lập"
    if name == "dpo":
        return "Mua" if value > 0.0 else "Bán" if value < 0.0 else "Trung lập"
    if name == "aroon":
        return "Mua" if value > 33.0 else "Bán" if value < -33.0 else "Trung lập"
    if name == "tsi":
        return "Mua" if value > 25.0 else "Bán" if value < -25.0 else "Trung lập"
    if name == "adx":
        plus_di = kw.get("plus_di", 0)
        minus_di = kw.get("minus_di", 0)
        if value < 20:
            return "Trung lập"
        return "Mua" if plus_di > minus_di else "Bán"
    if name == "ao":
        prev = float(kw.get("prev", value) or value)
        if value > 0 and value > prev:
            return "Mua"
        if value < 0 and value < prev:
            return "Bán"
        return "Trung lập"
    if name == "macd":
        signal = kw.get("signal", 0)
        return "Mua" if value > signal else "Bán" if value < signal else "Trung lập"
    if name == "bbp":
        bbp_norm = float(kw.get("bbp_norm", value / atr))
        return "Mua" if bbp_norm > 0.5 else "Bán" if bbp_norm < -0.5 else "Trung lập"
    if name == "force_index":
        scale = max(float(kw.get("scale", 1.0) or 1.0), 1e-8)
        norm = value / scale
        return "Mua" if norm > 0.25 else "Bán" if norm < -0.25 else "Trung lập"
    return "Trung lập"


def _ema_pair_action(fast_val: Optional[float], slow_val: Optional[float]) -> str:
    """EMA chain classification based on relative ordering only."""
    if fast_val is None or slow_val is None:
        return "Trung lập"
    if math.isnan(fast_val) or math.isnan(slow_val):
        return "Trung lập"
    if fast_val > slow_val:
        return "Mua"
    if fast_val < slow_val:
        return "Bán"
    return "Trung lập"


def _ema_pair_threshold_pct(pair_key: str) -> float:
    thresholds = {
        "ema_1_5": 0.08,
        "ema_5_10": 0.10,
        "ema_10_20": 0.14,
        "ema_20_50": 0.22,
        "ema_50_100": 0.35,
        "ema_100_200": 0.55,
    }
    return thresholds.get(pair_key, 0.18)


def _ema_pair_score(fast_val: Optional[float], slow_val: Optional[float], pair_key: str) -> float:
    if fast_val is None or slow_val is None:
        return 0.0
    if math.isnan(fast_val) or math.isnan(slow_val):
        return 0.0
    threshold = max(_ema_pair_threshold_pct(pair_key), 1e-4)
    spread_pct = _pct(fast_val, slow_val)
    return _clamp(math.tanh(spread_pct / threshold), -1.0, 1.0)


def _osc_signal_score(name: str, value: float, **kw) -> float:
    """Continuous oscillator score in [-1, 1] for weighting strength internally."""
    if value is None or math.isnan(value):
        return 0.0

    scale = max(float(kw.get("scale", 1.0) or 1.0), 1e-8)

    def _bounded_score(
        current: float,
        min_value: float,
        max_value: float,
        *,
        bullish_high: bool = True,
    ) -> float:
        span = max(max_value - min_value, 1e-8)
        ratio = _clamp((current - min_value) / span, 0.0, 1.0)
        score = (ratio * 2.0) - 1.0
        return _clamp(score if bullish_high else -score, -1.0, 1.0)

    if name == "rsi":
        return _bounded_score(value, 0.0, 100.0)
    if name in {"stoch", "stoch_rsi", "ultimate", "demarker"}:
        return _bounded_score(value, 0.0, 100.0)
    if name == "cci":
        return _clamp(value / 200.0, -1.0, 1.0)
    if name == "adx":
        plus_di = float(kw.get("plus_di", 0.0) or 0.0)
        minus_di = float(kw.get("minus_di", 0.0) or 0.0)
        strength = _clamp((value - 18.0) / 22.0, 0.0, 1.0)
        direction = math.tanh((plus_di - minus_di) / max(plus_di + minus_di, 10.0) * 3.0)
        return _clamp(direction * strength, -1.0, 1.0)
    if name == "ao":
        return _clamp(math.tanh(value / scale), -1.0, 1.0)
    if name == "momentum":
        return _clamp((value - 100.0) / 6.0, -1.0, 1.0)
    if name == "bbp":
        bbp_norm = float(kw.get("bbp_norm", value / max(float(kw.get("atr", scale) or scale), 1e-8)))
        return _clamp(math.tanh(bbp_norm / 1.0), -1.0, 1.0)
    if name == "dpo":
        return _clamp(math.tanh(value / scale), -1.0, 1.0)
    if name == "macd":
        signal = float(kw.get("signal", 0.0) or 0.0)
        return _clamp(math.tanh((value - signal) / scale), -1.0, 1.0)
    if name == "williams":
        return _bounded_score(value, -100.0, 0.0)
    if name == "roc":
        return _clamp(math.tanh(value / 3.0), -1.0, 1.0)
    if name == "trix":
        return _clamp(math.tanh(value / 0.35), -1.0, 1.0)
    if name == "ppo":
        return _clamp(math.tanh(value / 0.8), -1.0, 1.0)
    if name == "cmo":
        return _clamp(value / 50.0, -1.0, 1.0)
    if name == "aroon":
        return _clamp(value / 60.0, -1.0, 1.0)
    if name == "tsi":
        return _clamp(value / 25.0, -1.0, 1.0)
    if name == "force_index":
        return _clamp(math.tanh(value / scale), -1.0, 1.0)
    return 0.0


def compute_indicators(data: List[Dict[str, Any]]) -> Dict[str, Any]:
    """Compute a full suite of technical indicators on OHLCV data."""
    if len(data) < 30:
        return {"error": "Insufficient data (need ≥30 candles)"}

    closes = np.array([d["close"] for d in data], dtype=float)
    highs  = np.array([d["high"]  for d in data], dtype=float)
    lows   = np.array([d["low"]   for d in data], dtype=float)
    vols   = np.array([d["volume"]for d in data], dtype=float)
    times  = [d["time"] for d in data]

    def _last(arr):
        v = arr[-1]
        return None if math.isnan(v) else round(float(v), 8)

    def _series(arr, n: int = 50):
        out = []
        window = arr[-min(n, len(arr)):]
        offset = len(arr) - len(window)
        for i, v in enumerate(window):
            idx = offset + i
            out.append({"time": times[idx], "value": None if math.isnan(v) else round(float(v), 8)})
        return out

    ema9   = _ema(closes, 9)
    ema21  = _ema(closes, 21)
    ema50  = _ema(closes, 50)
    ema200 = _ema(closes, 200)
    rsi14  = _rsi(closes, 14)
    macd_l, macd_s, macd_h = _macd(closes)
    bb_u, bb_m, bb_l       = _bollinger(closes)
    atr14  = _atr(highs, lows, closes, 14)
    stoch_k, stoch_d       = _stoch_rsi(closes)
    roc12  = _roc(closes, 12)
    trix18 = _trix(closes, 18)
    ppo12  = _ppo(closes, 12, 26)
    cmo14  = _cmo(closes, 14)
    dpo20  = _dpo(closes, 20)
    aroon25 = _aroon_oscillator(highs, lows, 25)
    tsi25  = _tsi(closes, 25, 13)

    # Volume SMA 20 (Vectorized v6.0)
    vol_sma = _sma(vols, 20)

    last_close = closes[-1]
    last_atr   = _last(atr14) or 0

    return {
        "ema": {
            "ema9":   _last(ema9),   "ema21":  _last(ema21),
            "ema50":  _last(ema50),  "ema200": _last(ema200),
        },
        "rsi": {
            "value":  _last(rsi14),
            "signal": (
                "overbought" if (_last(rsi14) or 50) > 70
                else "oversold" if (_last(rsi14) or 50) < 30
                else "neutral"
            ),
        },
        "macd": {
            "macd":      _last(macd_l),
            "signal":    _last(macd_s),
            "histogram": _last(macd_h),
            "cross":     (
                "bullish" if (_last(macd_h) or 0) > 0
                else "bearish" if (_last(macd_h) or 0) < 0
                else "neutral"
            ),
        },
        "bollinger": {
            "upper":  _last(bb_u),
            "middle": _last(bb_m),
            "lower":  _last(bb_l),
            "bandwidth": round(((_last(bb_u) or 0) - (_last(bb_l) or 0)) / ((_last(bb_m) or 1)), 6),
        },
        "atr": {
            "value":   _last(atr14),
            "pct":     round(last_atr / last_close * 100, 4) if last_close else None,
        },
        "stoch_rsi": {
            "k": _last(stoch_k),
            "d": _last(stoch_d),
            "signal": (
                "overbought" if (_last(stoch_k) or 50) > 80
                else "oversold" if (_last(stoch_k) or 50) < 20
                else "neutral"
            ),
        },
        "roc": {
            "value": _last(roc12),
            "signal": "bullish" if (_last(roc12) or 0) > 1.0 else "bearish" if (_last(roc12) or 0) < -1.0 else "neutral",
        },
        "trix": {
            "value": _last(trix18),
            "signal": "bullish" if (_last(trix18) or 0) > 0 else "bearish" if (_last(trix18) or 0) < 0 else "neutral",
        },
        "ppo": {
            "value": _last(ppo12),
            "signal": "bullish" if (_last(ppo12) or 0) > 0.35 else "bearish" if (_last(ppo12) or 0) < -0.35 else "neutral",
        },
        "cmo": {
            "value": _last(cmo14),
            "signal": "bullish" if (_last(cmo14) or 0) > 20 else "bearish" if (_last(cmo14) or 0) < -20 else "neutral",
        },
        "dpo": {
            "value": _last(dpo20),
            "signal": "bullish" if (_last(dpo20) or 0) > 0 else "bearish" if (_last(dpo20) or 0) < 0 else "neutral",
        },
        "aroon": {
            "value": _last(aroon25),
            "signal": "bullish" if (_last(aroon25) or 0) > 25 else "bearish" if (_last(aroon25) or 0) < -25 else "neutral",
        },
        "tsi": {
            "value": _last(tsi25),
            "signal": "bullish" if (_last(tsi25) or 0) > 5 else "bearish" if (_last(tsi25) or 0) < -5 else "neutral",
        },
        "volume": {
            "last":       round(float(vols[-1]), 2),
            "sma20":      round(float(vol_sma[-1]), 2),
            "above_avg":  bool(vols[-1] > vol_sma[-1]),
        },
        "trend": {
            "ema_bullish_stack": bool(
                all(x is not None for x in [_last(ema9),_last(ema21),_last(ema50)]) and
                _last(ema9) > _last(ema21) > _last(ema50)   # type: ignore
            ),
            "above_ema200": bool(_last(ema200) is not None and last_close > (_last(ema200) or 0)),
            "close":        round(float(last_close), 8),
            "short_momentum_ref": float(closes[max(0, len(closes)-6)]) if len(closes) else float(last_close),
        },
        "short_momentum_ref": float(closes[max(0, len(closes)-6)]) if len(closes) else float(last_close),
        "series": {
            "ema9":   _series(ema9),
            "ema21":  _series(ema21),
            "ema50":  _series(ema50),
            "bb_upper": _series(bb_u),
            "bb_mid":   _series(bb_m),
            "bb_lower": _series(bb_l),
        },
    }


# ─────────────────────────────────────────────────────────────────────────────
# UTILITIES
# ─────────────────────────────────────────────────────────────────────────────

def _clamp(v: float, lo: float, hi: float) -> float:
    return max(lo, min(hi, v))


def _pct(a: float, b: float) -> float:
    """(a - b) / b * 100, an toàn với b = 0."""
    return (a - b) / b * 100.0 if b != 0 else 0.0


def _safe(v: Optional[float], default: float = 0.0) -> float:
    if v is None or math.isnan(v) or math.isinf(v):
        return default
    return float(v)


def _round(v: float, d: int = 6) -> float:
    return round(float(v), d)


def _build_anchor_forecast(
    data: List[Dict[str, Any]],
    indicators: Dict[str, Any],
    horizon: int,
    interval: str,
) -> Dict[str, np.ndarray]:
    """Fallback statistical forecast to cross-verify AI model stability."""
    closes = np.array([float(d["close"]) for d in data], dtype=float)
    if len(closes) < 3:
        last_close = closes[-1]
        flat = np.full(horizon, last_close, dtype=float)
        return {"p10": flat.copy(), "p50": flat.copy(), "p90": flat.copy()}

    last_close = float(closes[-1])
    ema21 = float(indicators["ema"].get("ema21") or last_close)
    ema50 = float(indicators["ema"].get("ema50") or last_close)
    rsi = float(indicators["rsi"].get("value") or 50.0)
    atr = float(indicators["atr"].get("value") or max(last_close * 0.006, 1.0))
    band = float(indicators["bollinger"].get("bandwidth") or 0.02)

    returns = pd.Series(closes).pct_change().dropna().tail(96)
    fast_drift = float(returns.tail(min(12, len(returns))).mean()) if len(returns) else 0.0
    slow_drift = float(returns.mean()) if len(returns) else 0.0
    vol = float(returns.std()) if len(returns) > 1 else 0.0

    ema_trend = ((last_close / ema21) - 1.0) * 0.30 + ((ema21 / ema50) - 1.0) * 0.25
    mean_revert = -((last_close - ema21) / last_close) * 0.18 if last_close else 0.0
    rsi_bias = 0.0018 if rsi < 35 else (-0.0018 if rsi > 65 else 0.0)
    volatility_drag = -vol * 0.35

    step_return = _clamp(
        fast_drift * 0.55 + slow_drift * 0.20 + ema_trend + mean_revert + rsi_bias + volatility_drag,
        -0.03,
        0.03,
    )

    anchor_p50: List[float] = []
    price = last_close
    for i in range(horizon):
        decay = max(0.35, 1.0 - (i / max(horizon * 1.6, 1)))
        price = price * (1.0 + step_return * decay)
        anchor_p50.append(price)

    step_scale = math.sqrt(max(STEP_SECONDS[interval], 60) / 86400.0)
    base_spread = max(atr * 0.70, last_close * max(vol * step_scale * 1.8, band * 0.22, 0.0035))
    p50 = np.array(anchor_p50, dtype=float)
    p10 = np.array([max(0.0, x - base_spread * math.sqrt(i + 1)) for i, x in enumerate(anchor_p50)], dtype=float)
    p90 = np.array([x + base_spread * math.sqrt(i + 1) for i, x in enumerate(anchor_p50)], dtype=float)
    return {"p10": p10, "p50": p50, "p90": p90}


def _blend_forecasts(
    last_close: float,
    model_output: Dict[str, Any],
    anchor_output: Dict[str, np.ndarray],
    indicators: Dict[str, Any],
) -> Dict[str, Any]:
    """Ensemble blending of AI model output and statistical anchor."""
    raw_p10 = np.array(model_output["p10"], dtype=float)
    raw_p50 = np.array(model_output["p50"], dtype=float)
    raw_p90 = np.array(model_output["p90"], dtype=float)

    # A-6: Robust scale calculation using median of first 3 model predictions
    first_model_est = float(np.median(raw_p50[:3])) if len(raw_p50) >= 3 else (float(raw_p50[0]) if len(raw_p50) else last_close)
    scale = (last_close / first_model_est) if abs(first_model_est) > 1e-8 else 1.0
    
    # Clip scale to [0.5, 2.0] to prevent extreme corrections on low-price coins (BUG-06)
    last_close_magnitude = math.floor(math.log10(max(abs(last_close), 1e-10)))
    clip_lo = 0.3 if last_close_magnitude < -4 else 0.5
    clip_hi = 3.0 if last_close_magnitude < -4 else 2.0
    scale = float(np.clip(scale, clip_lo, clip_hi))
    if abs(scale - 1.0) > 0.3:
        logger.warning("[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)
    
    # B-2: Advanced confidence (v6.1 Rework)
    # We use a simplified version of the new AI scoring logic here to keep blended dict consistent
    forecast_ret_pct = abs((blend_p50[-1] - last_close) / last_close * 100) if last_close else 0
    atr_pct = float(indicators["atr"].get("pct") or 1.0)
    
    # Certainty (band width)
    band_pct = abs(blend_p90[-1] - blend_p10[-1]) / abs(blend_p50[-1]) if abs(blend_p50[-1]) > 1e-8 else 0.1
    certainty = math.exp(-band_pct * 3.0)
    
    # Confidence: derived from agreement, magnitude (vs ATR), and certainty
    base_conf = 65.0 if agreement else 45.0
    magnitude_bonus = min(20.0, (forecast_ret_pct / max(atr_pct, 0.1)) * 5.0)
    
    confidence = (base_conf + magnitude_bonus) * certainty
    
    # Penalize scale bias
    bias_penalty = min(15.0, abs(scale - 1.0) * 30.0)
    confidence = max(10.0, min(95.0, confidence - bias_penalty))

    return {
        "p10": blend_p10,
        "p50": blend_p50,
        "p90": blend_p90,
        "model_weight": round(model_weight, 4),
        "anchor_weight": round(anchor_weight, 4),
        "agreement": agreement,
        "scale": round(scale, 6),
        "confidence": round(confidence, 2),
        "model_bias_pct": round(bias_pct, 4),
    }


# ─────────────────────────────────────────────────────────────────────────────
# MODULE: Analysis Engine v2.0
# ─────────────────────────────────────────────────────────────────────────────


@dataclass
class Signal:
    name:        str
    value:       float
    weight:      float
    label:       str
    detail:      str = ""


def _build_signals(
    indicators: Dict[str, Any],
    last_close:  float,
    forecast_return_pct: float,
    confidence:  float,
    interval:    str,
) -> List[Signal]:
    """
    Tạo danh sách 18 tín hiệu từ bộ indicators đã tính sẵn.
    Mỗi tín hiệu trả về giá trị trong [-1, +1]:
      +1  = bullish mạnh nhất
      -1  = bearish mạnh nhất
       0  = trung tính
    """
    signals: List[Signal] = []

    # ── Lấy giá trị ──────────────────────────────────────────────────────────
    rsi      = _safe(indicators["rsi"].get("value"), 50.0)
    ema9     = _safe(indicators["ema"].get("ema9"),  last_close)
    ema21    = _safe(indicators["ema"].get("ema21"), last_close)
    ema50    = _safe(indicators["ema"].get("ema50"), last_close)
    ema200   = _safe(indicators["ema"].get("ema200"), last_close)
    macd_v   = _safe(indicators["macd"].get("macd"), 0.0)
    macd_sig = _safe(indicators["macd"].get("signal"), 0.0)
    macd_h   = _safe(indicators["macd"].get("histogram"), 0.0)
    bb_upper = _safe(indicators["bollinger"].get("upper"), last_close * 1.02)
    bb_mid   = _safe(indicators["bollinger"].get("middle"), last_close)
    bb_lower = _safe(indicators["bollinger"].get("lower"), last_close * 0.98)
    bb_bw    = _safe(indicators["bollinger"].get("bandwidth"), 0.02)
    atr      = _safe(indicators["atr"].get("value"), last_close * 0.01)
    atr_pct  = _safe(indicators["atr"].get("pct"), 1.0)
    stoch_k  = _safe(indicators["stoch_rsi"].get("k"), 50.0)
    stoch_d  = _safe(indicators["stoch_rsi"].get("d"), 50.0)
    vol_cur  = _safe(indicators["volume"].get("last"), 0.0)
    vol_avg  = _safe(indicators["volume"].get("sma20"), 1.0)

    # ── 1: RSI momentum ──
    if rsi > 70:
        rsi_score = _clamp(-((rsi - 70) / 30), -1.0, -0.1)
        rsi_lbl   = f"RSI quá mua ({rsi:.1f})"
    elif rsi < 30:
        rsi_score = _clamp((30 - rsi) / 30, 0.1, 1.0)
        rsi_lbl   = f"RSI quá bán ({rsi:.1f})"
    else:
        rsi_score = _clamp((rsi - 50) / 20.0, -1.0, 1.0)
        rsi_lbl   = f"RSI trung tính ({rsi:.1f})"
    signals.append(Signal("rsi_momentum", rsi_score, 1.6, rsi_lbl))

    # ── 2: Stochastic RSI ──
    st_score = _clamp((stoch_k - 50) / 30.0, -1.0, 1.0)
    kd_cross = 0.4 if stoch_k > stoch_d else -0.4
    signals.append(Signal("stoch_rsi", _clamp(st_score + kd_cross * 0.3, -1, 1), 1.2, f"StochRSI K{'>' if kd_cross>0 else '<'}D ({stoch_k:.1f})"))

    # ── 3: MACD Crossover ──
    m_score = 1.0 if macd_v > macd_sig else -1.0
    signals.append(Signal("macd_cross", m_score, 1.5, f"MACD {'Bullish' if m_score>0 else 'Bearish'}"))

    # ── 4: EMA Stack (9/21/50) ──
    ema_bull = (ema9 > ema21 > ema50)
    ema_bear = (ema9 < ema21 < ema50)
    e_score = 1.0 if ema_bull else -1.0 if ema_bear else 0.0
    signals.append(Signal("ema_stack", e_score, 1.8, "Cấu trúc EMA stack"))

    # ── 5: Price vs EMA200 ──
    dist_200 = _pct(last_close, ema200)
    signals.append(Signal("price_vs_ema200", _clamp(dist_200 / 10.0, -1.0, 1.0), 1.6, f"Giá vs EMA200 ({dist_200:+.2f}%)"))

    # ── 6: EMA9 Slope ──
    dist_9 = _pct(last_close, ema9)
    signals.append(Signal("ema9_slope", _clamp(dist_9 / 1.5, -1.0, 1.0), 1.0, f"Giá vs EMA9 ({dist_9:+.2f}%)"))

    # ── 7: Bollinger Position ──
    bb_range = bb_upper - bb_lower
    bb_pos = (last_close - bb_lower) / bb_range if bb_range > 0 else 0.5
    bb_score = 0.0
    if bb_pos < 0.1: bb_score = 0.8
    elif bb_pos > 0.9: bb_score = -0.8
    signals.append(Signal("bollinger_position", bb_score, 1.1, f"Vị trí BB ({bb_pos:.0%})"))

    # ── 8: BB Squeeze ──
    signals.append(Signal("bb_squeeze", 0.0, 0.5, "BB Squeeze" if bb_bw < 0.015 else "BB Bình thường"))

    # ── 9: Volume Confirmation ──
    vol_ratio = vol_cur / vol_avg if vol_avg > 0 else 1.0
    signals.append(Signal("volume_confirm", _clamp((vol_ratio - 1.0) * 0.5, -1, 1), 1.3, f"KLGD ({vol_ratio:.1f}x TB)"))

    # ── 10: AI Forecast ──
    signals.append(Signal("ai_forecast", _clamp(forecast_return_pct / 3.0, -1, 1) * (confidence/100.0), 2.0, f"AI {forecast_return_pct:+.2f}%"))

    # ── 11: EMA Cross (9 vs 21) ──
    signals.append(Signal("ema_cross_9_21", 1.0 if ema9 > ema21 else -1.0, 0.6, f"EMA9 {'>' if ema9>ema21 else '<'} EMA21"))

    # ── 12: RSI Extremes ──
    rsi_ext = 1.0 if rsi < 20 else -1.0 if rsi > 80 else 0.0
    signals.append(Signal("rsi_extreme", rsi_ext, 0.7, "RSI cực đại" if rsi_ext !=0 else "RSI ổn định"))

    # ── 13: MACD Hist Momentum ──
    signals.append(Signal("macd_hist_momentum", _clamp(macd_h / (atr or 1), -1, 1), 0.9, f"MACD Hist ({macd_h:+.4f})"))

    # ── 14: ATR Volatility ──
    signals.append(Signal("atr_volatility", 0.0, 0.3, f"ATR % ({atr_pct:.1f}%)"))

    # ── 15: Price Momentum (5-candle) ──
    # BUG-01 Fix: Read from trend or top-level correctly
    mom_ref = _safe(indicators.get("short_momentum_ref") or indicators.get("trend", {}).get("short_momentum_ref"), last_close)
    signals.append(Signal("short_momentum", _pct(last_close, mom_ref), 0.8, "Momentum ngắn hạn"))

    # ── 16: Volatility Expansion ──
    signals.append(Signal("vol_expansion", 0.5 if vol_ratio > 2.0 else 0.0, 0.6, "Bùng nổ KLGD" if vol_ratio > 2.0 else "KLGD ổn định"))

    # ── 17: EMA50 Support/Resist ──
    signals.append(Signal("price_vs_ema50", _clamp(_pct(last_close, ema50) / 2.0, -1, 1), 1.0, f"Giá vs EMA50 ({_pct(last_close, ema50):+.2f}%)"))

    # ── 18: Trend Alignment ──
    align = 1.0 if (ema9 > ema200 and forecast_return_pct > 0) else -1.0 if (ema9 < ema200 and forecast_return_pct < 0) else 0.0
    signals.append(Signal("trend_alignment", align, 1.2, "Đồng thuận xu hướng"))

    return signals


@dataclass
class PriceLevel:
    price:    float
    type:     str
    strength: int
    label:    str


def _calc_price_levels(data: List[Dict[str, Any]], atr: float) -> Dict[str, Any]:
    """B-15: Proper swing detection for support and resistance (BUG-15)."""
    if len(data) < 20: return {}
    closes = np.array([float(d["close"]) for d in data])
    highs  = np.array([float(d["high"])  for d in data])
    lows   = np.array([float(d["low"])   for d in data])
    
    # Use previous COMPLETED candle for pivot calculation (BUG-P1-10)
    if len(highs) >= 2:
        last_h = float(highs[-2])
        last_l = float(lows[-2])
        last_c = float(closes[-2])
    else:
        last_h, last_l, last_c = float(highs[-1]), float(lows[-1]), float(closes[-1])
    
    pivots = _calc_pivot_points(last_h, last_l, last_c)
    
    # 1. Detect Pivot Highs and Lows (window=3)
    pivots_h = []
    pivots_l = []
    for i in range(3, len(highs)-3):
        if highs[i] == max(highs[i-3:i+4]): pivots_h.append(highs[i])
        if lows[i] == min(lows[i-3:i+4]): pivots_l.append(lows[i])
    
    # Fallback if no pivots
    if not pivots_h: pivots_h = [max(highs[-20:])]
    if not pivots_l: pivots_l = [min(lows[-20:])]
    
    # 2. Select nearest support and resistance
    ns = max([p for p in pivots_l if p < last_c], default=last_c - atr)
    nr = min([p for p in pivots_h if p > last_c], default=last_c + atr)
    
    return {
        "nearest_support":   _round(ns, 6),
        "nearest_resistance":_round(nr, 6),
        "key_support":       _round(ns - atr, 6),
        "key_resistance":    _round(nr + atr, 6),
        "level_list": [
            {"price": ns, "type": "support", "strength": 3, "label": f"Hỗ trợ {ns:.5g}"},
            {"price": nr, "type": "resistance", "strength": 3, "label": f"Kháng cự {nr:.5g}"},
        ]
    }


def _classify_regime(indicators: Dict[str, Any], last_close: float, atr_pct: float) -> Tuple[str, str]:
    """Phân loại trạng thái thị trường dựa trên biến động và xu hướng."""
    rsi = _safe(indicators["rsi"].get("value"), 50.0)
    ema9 = _safe(indicators["ema"].get("ema9"), last_close)
    ema50 = _safe(indicators["ema"].get("ema50"), last_close)
    ema200 = _safe(indicators["ema"].get("ema200"), last_close)
    
    # Logic 8 trạng thái
    if atr_pct > 3.5:
        if rsi > 60: return "volatile_bull", "Tăng trưởng biến động cao"
        if rsi < 40: return "volatile_bear", "Sụt giảm biến động cao"
        return "high_chaos", "Thị trường hỗn loạn / Vol cao"
    
    if last_close > ema200 and ema9 > ema50:
        if rsi > 70: return "overextended_bull", "Tăng trưởng quá mức (Overbought)"
        return "stable_bull", "Xu hướng tăng ổn định"
    
    if last_close < ema200 and ema9 < ema50:
        if rsi < 30: return "capitulation", "Hoảng loạn / Quá bán (Oversold)"
        return "stable_bear", "Xu hướng giảm ổn định"
    
    if abs(_pct(ema9, ema50)) < 0.5:
        return "tight_range", "Tích lũy biên độ hẹp (Squeeze)"
        
    return "sideways", "Thị trường đi ngang (Sideway)"


def _calc_momentum(data: List[Dict[str, Any]]) -> Dict[str, Any]:
    """Tính toán momentum đa khung thời gian."""
    if len(data) < 50: return {"short": 0, "mid": 0, "long": 0, "aligned": False}
    closes = [float(d["close"]) for d in data]
    last = closes[-1]
    
    short = _pct(last, closes[max(0, len(closes)-6)])
    mid   = _pct(last, closes[max(0, len(closes)-21)])
    long  = _pct(last, closes[max(0, len(closes)-51)])
    
    aligned = (short > 0 and mid > 0 and long > 0) or (short < 0 and mid < 0 and long < 0)
    
    return {
        "short": round(short, 4),
        "mid":   round(mid, 4),
        "long":  round(long, 4),
        "short_lbl": f"{'Tăng' if short>0 else 'Giảm'} {abs(short):.2f}%",
        "aligned": aligned
    }


@dataclass
class ConfluenceResult:
    raw_score:      float
    confluence:     float
    bias:           str
    conviction:     str
    grade:          str
    active_signals: int
    bull_count:     int
    bear_count:     int
    neutral_count:  int


def _calc_confluence(signals: List[Signal]) -> ConfluenceResult:
    tw = sum(s.weight for s in signals)
    ws = sum(s.value * s.weight for s in signals)
    raw = ws / tw if tw > 0 else 0.0
    conf = _clamp(50.0 + raw * 50.0, 0.0, 100.0)
    bias = "bullish" if conf >= 60 else "bearish" if conf <= 40 else "neutral"
    
    return ConfluenceResult(
        raw_score=raw, confluence=conf, bias=bias,
        conviction="strong" if abs(conf-50)>20 else "moderate",
        grade="A" if abs(conf-50)>25 else "B",
        active_signals=len(signals),
        bull_count=sum(1 for s in signals if s.value > 0.2),
        bear_count=sum(1 for s in signals if s.value < -0.2),
        neutral_count=sum(1 for s in signals if abs(s.value) <= 0.2)
    )


@dataclass
class TradeSetup:
    tier:         str
    direction:    str
    entry_low:    float
    entry_high:   float
    entry_mid:    float
    stop_loss:    float
    take_profit1: float
    take_profit2: float
    take_profit3: float
    risk_reward:  float
    risk_pct:     float
    valid:        bool
    notes:        str


def _build_setups(bias: str, last_close: float, atr: float, levels: Dict[str, Any], projected: float) -> List[TradeSetup]:
    """Tạo các phương án giao dịch theo 3 cấp độ: Conservative, Standard, Aggressive."""
    setups = []
    ns = levels.get("nearest_support", last_close - atr)
    nr = levels.get("nearest_resistance", last_close + atr)
    ks = levels.get("key_support", ns - atr)
    kr = levels.get("key_resistance", nr + atr)
    
    if bias == "bullish":
        # 1. Conservative (Thận trọng - Chờ hồi)
        sl_c = ks - atr * 0.3
        tp_c = nr
        setups.append(TradeSetup(
            tier="conservative", direction="long",
            entry_low=ks, entry_high=ns, entry_mid=(ks+ns)/2,
            stop_loss=sl_c, take_profit1=last_close + (last_close-sl_c),
            take_profit2=tp_c, take_profit3=tp_c + atr,
            risk_reward=_round((tp_c - ns) / (ns - sl_c), 2) if ns > sl_c else 0.0,
            risk_pct=_round((ns - sl_c) / ns * 100, 2),
            valid=True, notes="Đợi giá kiểm tra lại vùng hỗ trợ cứng trước khi tham gia."
        ))
        
        # 2. Standard (Tiêu chuẩn - Theo xu hướng)
        sl_s = ns - atr * 0.5
        tp_s = projected
        setups.append(TradeSetup(
            tier="standard", direction="long",
            entry_low=ns, entry_high=last_close, entry_mid=(ns+last_close)/2,
            stop_loss=sl_s, take_profit1=last_close + atr,
            take_profit2=projected, take_profit3=max(projected, kr),
            risk_reward=_round((projected - last_close) / (last_close - sl_s), 2) if last_close > sl_s else 0.0,
            risk_pct=_round((last_close - sl_s) / last_close * 100, 2),
            valid=True, notes="Giao dịch theo đà tăng hiện tại, dừng lỗ dưới hỗ trợ gần nhất."
        ))
        
        # 3. Aggressive (Săn đuổi - Buy Stop / Breakout)
        sl_a = last_close - atr * 0.8
        tp_a = max(projected, kr) + atr
        setups.append(TradeSetup(
            tier="aggressive", direction="long",
            entry_low=last_close, entry_high=last_close + atr*0.2, entry_mid=last_close+atr*0.1,
            stop_loss=sl_a, take_profit1=last_close + atr*1.5,
            take_profit2=tp_a - atr, take_profit3=tp_a,
            risk_reward=_round((tp_a - last_close) / (last_close - sl_a), 2) if last_close > sl_a else 0.0,
            risk_pct=_round((last_close - sl_a) / last_close * 100, 2),
            valid=True, notes="Vào lệnh trực tiếp để bắt kịp sóng tăng mạnh. Rủi ro cao hơn."
        ))
        
    elif bias == "bearish":
        # 1. Conservative (Thận trọng - Chờ hồi)
        sl_c = kr + atr * 0.3
        tp_c = ns
        setups.append(TradeSetup(
            tier="conservative", direction="short",
            entry_low=nr, entry_high=kr, entry_mid=(nr+kr)/2,
            stop_loss=sl_c, take_profit1=last_close - (sl_c-last_close),
            take_profit2=tp_c, take_profit3=tp_c - atr,
            risk_reward=_round((nr - tp_c) / (sl_c - nr), 2) if sl_c > nr else 0.0,
            risk_pct=_round((sl_c - nr) / nr * 100, 2),
            valid=True, notes="Chờ giá hồi lên vùng kháng cự mạnh để tối ưu RR."
        ))
        
        # Standard Short
        sl_s = nr + atr * 0.5
        tp_s = projected
        setups.append(TradeSetup(
            tier="standard", direction="short",
            entry_low=last_close, entry_high=nr, entry_mid=(last_close+nr)/2,
            stop_loss=sl_s, take_profit1=last_close - atr,
            take_profit2=projected, take_profit3=min(projected, ks),
            risk_reward=_round((last_close - projected) / (sl_s - last_close), 2) if sl_s > last_close else 0.0,
            risk_pct=_round((sl_s - last_close) / last_close * 100, 2),
            valid=True, notes="Bán theo xu hướng giảm, dừng lỗ trên kháng cự gần nhất."
        ))
        
    return [s for s in setups if s.valid]


def _build_scenarios(
    bias: str, last_close: float, atr: float, levels: Dict[str, Any], 
    forecast_rows: List[Dict[str, Any]], confluence: float, momentum: Dict[str, Any]
) -> List[Dict[str, Any]]:
    """Phân tích các kịch bản có thể xảy ra."""
    fc_p50 = float(forecast_rows[-1]["p50"])
    fc_p90 = float(forecast_rows[-1]["p90"])
    fc_p10 = float(forecast_rows[-1]["p10"])
    
    # Kịch bản cơ sở (P50)
    base = {
        "name": "base", "label": "Kịch bản Cơ sở", "probability": round(confluence, 1),
        "price_target": _round(fc_p50, 6), "return_pct": _round(_pct(fc_p50, last_close), 2),
        "catalyst": "Duy trì xu hướng hiện tại", "invalidation": "Phá vỡ vùng EMA21"
    }
    
    # BUG-10: Normalize probabilities to ensure sum = 100%
    # Using raw scores from confluence and bias
    bull_raw = (100 - confluence) * 0.4 + (20 if bias == 'bullish' else 0)
    base_raw = confluence
    bear_raw = max(5.0, 100.0 - base_raw - bull_raw) # Min 5% for bear in base calc
    
    # Normalize
    total = bull_raw + base_raw + bear_raw
    bull_prob = (bull_raw / total) * 100
    base_prob = (base_raw / total) * 100
    bear_prob = 100.0 - bull_prob - base_prob  # Exact sum
    
    base["probability"] = round(base_prob, 1)
    bull = {
        "name": "bull", "label": "Kịch bản Tích cực", "probability": round(bull_prob, 1),
        "price_target": _round(fc_p90, 6), "return_pct": _round(_pct(fc_p90, last_close), 2),
        "catalyst": "Đột phá Momentum mạnh", "invalidation": "RSI quay đầu giảm"
    }
    bear = {
        "name": "bear", "label": "Kịch bản Tiêu cực", "probability": round(bear_prob, 1),
        "price_target": _round(fc_p10, 6), "return_pct": _round(_pct(fc_p10, last_close), 2),
        "catalyst": "Đảo chiều bất ngờ / Tin xấu", "invalidation": "Hỗ trợ cứng được giữ vững"
    }
    
    return [base, bull, bear]


def _build_risk_framework(
    bias: str, last_close: float, atr: float, atr_pct: float, 
    levels: Dict[str, Any], setups: List[TradeSetup], interval: str
) -> Dict[str, Any]:
    """Cung cấp hướng dẫn quản trị rủi ro."""
    vol_adjusted_risk = _clamp(2.0 - (atr_pct / 5.0), 0.5, 1.5)
    
    advice = "Kích thước vị thế tiêu chuẩn (1% tài khoản)."
    if atr_pct > 4.0:
        advice = "Thị trường biến động mạnh: Giảm 50% khối lượng lệnh."
    elif atr_pct < 0.5:
        advice = "Biến động thấp: Cân nhắc chốt lời ngắn (Scalp)."
        
    return {
        "max_risk_pct": round(vol_adjusted_risk, 2),
        "position_size_advice": advice,
        "stop_loss_type": "Volatility-based (ATR)",
        "invalidation_point": levels.get("key_resistance" if bias=="bearish" else "key_support", last_close)
    }


def _build_reasoning(
    bias: str, conviction: str, regime_key: str, regime_lbl: str, 
    signals: List[Signal], indicators: Dict[str, Any], momentum: Dict[str, Any], 
    levels: Dict[str, Any], last_close: float, atr_pct: float, interval: str, forecast_ret: float
) -> Tuple[List[str], List[str], List[str]]:
    """Tự động sinh các lý do, cảnh báo và cơ hội bằng tiếng Việt chuyên sâu."""
    reasons, warnings, opportunities = [], [], []
    
    # Logic Lý do
    if bias == "bullish":
        reasons.append(f"Xu hướng chủ đạo là TĂNG ({conviction}) trên khung {interval}.")
        reasons.append(f"Dự báo AI cho thấy tiềm năng tăng trưởng {forecast_ret:+.2f}% trong ngắn hạn.")
        if indicators["trend"].get("ema_bullish_stack"):
            reasons.append("Hệ thống EMA đang xếp chồng Bullish mạnh mẽ, xác nhận lực mua áp ảo.")
    elif bias == "bearish":
        reasons.append(f"Áp lực GIẢM giá chiếm ưu thế ({conviction}) trên khung {interval}.")
        reasons.append(f"AI nhận diện tín hiệu suy yếu với mục tiêu giảm về vùng {forecast_ret:+.2f}%.")
    else:
        reasons.append(f"Thị trường đang trong trạng thái TÍCH LŨY / ĐI NGANG trên khung {interval}.")
        reasons.append("Chưa có tín hiệu bứt phá rõ rệt từ các chỉ báo kỹ thuật quan trọng.")
        
    if regime_lbl:
        reasons.append(f"Trạng thái thị trường hiện tại: {regime_lbl}.")
        
    # Logic Cảnh báo
    rsi = _safe(indicators["rsi"].get("value"), 50.0)
    if rsi > 70: warnings.append("Chỉ số RSI đi vào vùng quá mua (>70), rủi ro đảo chiều kỹ thuật cao.")
    if atr_pct > 3.0: warnings.append("Biến động thị trường đang ở mức cao (ATR), ưu tiên quản lý vốn chặt chẽ.")
    
    # Logic Cơ hội
    if bias == "bullish":
        opportunities.append(f"Cơ hội Long khi giá điều chỉnh về vùng hỗ trợ {levels.get('nearest_support')}.")
    else:
        opportunities.append(f"Cơ hội Short khi giá hồi phục chạm kháng cự {levels.get('nearest_resistance')}.")
        
    return reasons, warnings, opportunities


# ── Technical Analysis Weights & Constants (Dashboard Rework v6.1) ────────────
OSC_WEIGHTS = {
    "rsi": 2.1,
    "macd": 2.4,
    "stoch_rsi": 1.1,
    "stoch": 1.2,
    "cci": 1.7,
    "adx": 1.8,
    "williams": 1.5,
    "ultimate": 1.1,
    "bbp": 1.0,
    "ao": 0.9,
    "momentum": 1.6,
    "roc": 1.3,
    "trix": 1.4,
    "ppo": 1.4,
    "cmo": 1.2,
    "dpo": 0.8,
    "aroon": 1.3,
    "tsi": 1.4,
    "demarker": 1.2,
    "force_index": 1.4,
}

MA_WEIGHT_MAP = {
    "ema_1_5": 0.9,
    "ema_5_10": 1.1,
    "ema_10_20": 1.35,
    "ema_20_50": 1.8,
    "ema_50_100": 2.2,
    "ema_100_200": 2.7,
}

def _extract_osc_key(label: str) -> str:
    l = label.lower()
    if "sức mạnh tương đối" in l or ("rsi" in l and "nhanh" not in l): return "rsi"
    if "macd" in l: return "macd"
    if "stochastic %k" in l: return "stoch"
    if "nhanh" in l or "stoch_rsi" in l: return "stoch_rsi"
    if "kênh hàng hóa" in l or "cci" in l: return "cci"
    if "định hướng" in l or "adx" in l: return "adx"
    if "williams" in l: return "williams"
    if "ultimate" in l: return "ultimate"
    if "bbp" in l or "sức mạnh giá" in l: return "bbp"
    if "ao" in l: return "ao"
    if "xung lượng" in l or "momentum" in l: return "momentum"
    if "roc" in l: return "roc"
    if "trix" in l: return "trix"
    if "ppo" in l: return "ppo"
    if "cmo" in l: return "cmo"
    if "dpo" in l: return "dpo"
    if "aroon" in l: return "aroon"
    if "tsi" in l: return "tsi"
    if "demarker" in l: return "demarker"
    if "force index" in l: return "force_index"
    return "unknown"

def _get_ma_weight(label: str) -> float:
    l = label.lower()
    if l in MA_WEIGHT_MAP:
        return MA_WEIGHT_MAP[l]
    pair = re.findall(r"\d+", l)
    if len(pair) >= 2:
        return MA_WEIGHT_MAP.get(f"ema_{pair[0]}_{pair[1]}", 1.0)
    return 1.0

def _gauge_to_signal(gauge: float, interval: str = "1h") -> str:
    """Interval-aware 5-level signal converter."""
    thresholds = {
        "1m": (80, 62, 38, 20),
        "5m": (80, 62, 38, 20),
        "15m": (77, 60, 40, 23),
        "1h": (77, 60, 40, 23),
        "4h": (75, 58, 42, 25),
        "1d": (75, 58, 42, 25),
        "1w": (72, 56, 44, 28),
    }
    strong_buy, buy, sell, strong_sell = thresholds.get(interval, thresholds["1h"])
    if gauge >= strong_buy: return "Mua mạnh"
    if gauge >= buy: return "Mua"
    if gauge > sell: return "Trung lập"
    if gauge > strong_sell: return "Bán"
    return "Bán mạnh"


def _gauge_to_normalized_score(gauge: float) -> float:
    """Convert a 0..100 gauge into a -1..1 frontend-friendly scale."""
    return _clamp((float(gauge) - 50.0) / 50.0, -1.0, 1.0)


def _forecast_path_metrics(p50_path: np.ndarray, last_close: float) -> Dict[str, float]:
    """Measure forecast quality from the full path, not only the final endpoint."""
    if len(p50_path) == 0 or abs(last_close) <= 1e-8:
        return {
            "weighted_return_pct": 0.0,
            "final_return_pct": 0.0,
            "path_consistency": 50.0,
            "monotonicity": 50.0,
            "max_adverse_excursion_pct": 0.0,
            "mean_step_return_pct": 0.0,
        }

    ret_path = ((p50_path / last_close) - 1.0) * 100.0
    step_weights = np.linspace(1.0, 0.65, len(ret_path))
    weighted_ret_pct = float(np.average(ret_path, weights=step_weights))
    final_ret_pct = float(ret_path[-1])
    final_sign = 0 if abs(final_ret_pct) < 0.05 else (1 if final_ret_pct > 0 else -1)

    if len(ret_path) > 1 and final_sign != 0:
        signed_steps = [
            1.0 if np.sign(curr - prev) == final_sign else 0.0
            for prev, curr in zip(ret_path[:-1], ret_path[1:])
            if abs(curr - prev) >= 0.02
        ]
        path_consistency = float(sum(signed_steps) / len(signed_steps)) if signed_steps else 0.5
    else:
        path_consistency = 0.5

    if len(ret_path) > 1:
        monotonicity = float(np.mean(np.diff(ret_path) >= 0)) if final_sign >= 0 else float(np.mean(np.diff(ret_path) <= 0))
    else:
        monotonicity = 0.5

    if final_sign > 0:
        adverse = abs(float(np.min(ret_path)))
    elif final_sign < 0:
        adverse = abs(float(np.max(ret_path)))
    else:
        adverse = max(abs(float(np.min(ret_path))), abs(float(np.max(ret_path))))

    mean_step_return_pct = float(np.mean(np.diff(ret_path))) if len(ret_path) > 1 else final_ret_pct
    return {
        "weighted_return_pct": round(weighted_ret_pct, 2),
        "final_return_pct": round(final_ret_pct, 2),
        "path_consistency": round(path_consistency * 100.0, 1),
        "monotonicity": round(monotonicity * 100.0, 1),
        "max_adverse_excursion_pct": round(adverse, 2),
        "mean_step_return_pct": round(mean_step_return_pct, 3),
    }


def _calc_vote_gauge(buy: int, sell: int, neutral: int) -> float:
    """Equal-weight vote gauge for oscillator and MA signals."""
    buy_f = float(max(0, buy))
    sell_f = float(max(0, sell))
    neutral_f = float(max(0, neutral))
    denominator = buy_f + sell_f + neutral_f
    if denominator <= 0:
        return 50.0
    raw_gauge = 50.0 + 50.0 * ((buy_f - sell_f) / denominator)
    return max(0.0, min(100.0, raw_gauge))

def _calc_osc_score(osc_data: list, interval: str = "1h") -> dict:
    """MODULE 1: Equal-weight vote scoring for oscillators."""
    buy = sell = neutral = 0
    buy_weight = sell_weight = neutral_weight = 0.0
    
    for item in osc_data:
        key = item.get("key") or _extract_osc_key(item["name"])
        w = OSC_WEIGHTS.get(key, 1.0)
        # Robust case-insensitive check
        act = str(item.get("action", "")).strip().lower()
        if act == "mua":
            buy += 1
            buy_weight += w
        elif act == "bán":
            sell += 1
            sell_weight += w
        else:
            neutral += 1
            neutral_weight += w

    gauge = _calc_vote_gauge(buy, sell, neutral)
    normalized = _gauge_to_normalized_score(gauge)
    return {
        "gauge": round(gauge, 1),
        "normalized_score": round(normalized, 4),
        "signal": _gauge_to_signal(gauge, interval),
        "buy": buy, "sell": sell, "neutral": neutral,
        "buy_weight": round(buy_weight, 2),
        "sell_weight": round(sell_weight, 2),
        "neutral_weight": round(neutral_weight, 2),
    }

def _calc_ma_score(ma_data: list, closes: np.ndarray, interval: str = "1h") -> dict:
    """MODULE 2: Equal-weight vote scoring for moving averages."""
    buy = sell = neutral = 0
    buy_weight = sell_weight = neutral_weight = 0.0
    
    for item in ma_data:
        w = _get_ma_weight(item.get("key") or item["name"])
        act = str(item.get("action", "")).strip().lower()
        if act == "mua":
            buy += 1
            buy_weight += w
        elif act == "bán":
            sell += 1
            sell_weight += w
        else:
            neutral += 1
            neutral_weight += w

    gauge = _calc_vote_gauge(buy, sell, neutral)
    normalized = _gauge_to_normalized_score(gauge)
    ema50 = _ema(closes, 50)[-1] if len(closes) >= 50 else float("nan")
    ema200 = _ema(closes, 200)[-1] if len(closes) >= 200 else float("nan")
    
    return {
        "gauge": round(gauge, 1),
        "normalized_score": round(normalized, 4),
        "signal": _gauge_to_signal(gauge, interval),
        "buy": buy, "sell": sell, "neutral": neutral,
        "buy_weight": round(buy_weight, 2),
        "sell_weight": round(sell_weight, 2),
        "neutral_weight": round(neutral_weight, 2),
        "structure_bonus": 0.0,
        "golden_cross": bool(not any(math.isnan(x) for x in (ema50, ema200)) and ema50 > ema200),
        "death_cross": bool(not any(math.isnan(x) for x in (ema50, ema200)) and ema50 < ema200),
    }

def _calc_ai_forecast_score(
    blended: dict,
    forecast_rows: List[Dict[str, Any]],
    last_close: float,
    indicators: dict,
    horizon: int,
    interval: str,
) -> dict:
    """MODULE 3: AI score from path quality, trend alignment, magnitude, and certainty."""
    trend = indicators.get("trend", {})
    atr_pct = max(float(indicators.get("atr", {}).get("pct") or 0.0), 0.1)
    rsi = float(indicators.get("rsi", {}).get("value") or 50.0)

    p50_path = np.array(blended.get("p50", []), dtype=float)
    p10_path = np.array(blended.get("p10", []), dtype=float)
    p90_path = np.array(blended.get("p90", []), dtype=float)

    if not len(p50_path):
        p50_path = np.array([last_close], dtype=float)
    if not len(p10_path):
        p10_path = np.array([last_close], dtype=float)
    if not len(p90_path):
        p90_path = np.array([last_close], dtype=float)

    path_len = len(p50_path)
    path_metrics = _forecast_path_metrics(p50_path, last_close)
    forecast_ret_path = ((p50_path / max(last_close, 1e-8)) - 1.0) * 100.0
    final_ret_pct = path_metrics["final_return_pct"]
    weighted_ret_pct = path_metrics["weighted_return_pct"]
    directional_edge_pct = (weighted_ret_pct * 0.60) + (final_ret_pct * 0.40)
    direction_norm = math.tanh(directional_edge_pct / max(atr_pct * 1.35, 0.35))
    final_sign = 0 if abs(final_ret_pct) < 0.05 else (1 if final_ret_pct > 0 else -1)

    path_consistency = path_metrics["path_consistency"] / 100.0
    monotonicity = path_metrics["monotonicity"] / 100.0
    adverse_excursion_pct = path_metrics["max_adverse_excursion_pct"]

    avg_band_pct = float(np.mean((p90_path - p10_path) / np.maximum(np.abs(p50_path), 1e-8)) * 100.0)
    band_certainty = math.exp(-avg_band_pct / 4.0)
    ensemble_conf = max(0.0, min(1.0, float(blended.get("confidence", 50.0)) / 100.0))
    scale_penalty = min(0.18, abs(float(blended.get("scale", 1.0)) - 1.0) * 0.35)
    stability_penalty = min(0.20, adverse_excursion_pct / max(atr_pct * 5.0, 1.0) * 0.12)
    certainty_score = (
        ensemble_conf * 0.45 +
        band_certainty * 0.35 +
        path_consistency * 0.12 +
        monotonicity * 0.08
    ) - scale_penalty - stability_penalty
    certainty_score = max(0.08, min(0.98, certainty_score))

    move_in_atr = abs(final_ret_pct) / atr_pct
    magnitude_score = math.tanh(move_in_atr / 1.6)
    horizon_decay = max(0.70, 1.0 - max(horizon - 12, 0) * 0.012)
    magnitude_score *= horizon_decay

    ema_stack = bool(trend.get("ema_bullish_stack", False))
    above_200 = bool(trend.get("above_ema200", False))
    trend_alignment = 0.0
    if final_sign > 0 and ema_stack and above_200:
        trend_alignment = 0.10
    elif final_sign < 0 and (not ema_stack) and (not above_200):
        trend_alignment = -0.10
    elif final_sign > 0 and not above_200:
        trend_alignment = -0.07
    elif final_sign < 0 and above_200:
        trend_alignment = 0.07

    exhaustion_penalty = 0.0
    if final_sign > 0 and rsi >= 74:
        exhaustion_penalty = min(0.12, (rsi - 74.0) / 100.0)
    elif final_sign < 0 and rsi <= 26:
        exhaustion_penalty = -min(0.12, (26.0 - rsi) / 100.0)

    path_strength = 0.55 + 0.25 * path_consistency + 0.20 * monotonicity
    effective_strength = direction_norm * path_strength * (0.45 + 0.55 * certainty_score)
    directional_push = effective_strength * (22.0 + 18.0 * magnitude_score)
    alignment_push = trend_alignment * 35.0
    exhaustion_push = -exhaustion_penalty * 35.0
    gauge = max(8.0, min(92.0, 50.0 + directional_push + alignment_push + exhaustion_push))

    confidence_pct = 35.0 + certainty_score * 55.0 + min(move_in_atr, 1.5) * 6.0
    confidence_pct = max(20.0, min(95.0, confidence_pct))
    direction_label = "bullish" if gauge >= 58 else "bearish" if gauge <= 42 else "neutral"

    return {
        "gauge": round(gauge, 1),
        "normalized_score": round(_gauge_to_normalized_score(gauge), 4),
        "confidence_pct": round(confidence_pct, 1),
        "forecast_return_pct": round(final_ret_pct, 2),
        "weighted_return_pct": round(weighted_ret_pct, 2),
        "direction": direction_label,
        "magnitude_vs_atr": round(move_in_atr, 2),
        "band_uncertainty_pct": round(avg_band_pct, 2),
        "certainty": round(certainty_score * 100.0, 1),
        "path_consistency": round(path_consistency * 100.0, 1),
        "monotonicity": round(monotonicity * 100.0, 1),
        "max_adverse_excursion_pct": round(adverse_excursion_pct, 2),
        "path_metrics": path_metrics,
        "signal": _gauge_to_signal(gauge, interval)
    }

def _calc_summary_score(osc_score: dict, ma_score: dict, ai_score: dict) -> dict:
    """MODULE 4: Normalize + Weighted combine with compatibility fields."""
    W_OSC = 0.35
    W_MA  = 0.35
    W_AI  = 0.30
    
    composite = (osc_score["gauge"] * W_OSC + ma_score["gauge"] * W_MA + ai_score["gauge"] * W_AI)
    
    # Confidence multiplier: pull towards neutral if AI certainty is low
    certainty_factor = ai_score["certainty"] / 100.0
    pulled_to_neutral = composite + (50.0 - composite) * (1.0 - certainty_factor) * 0.20
    final_gauge = max(5.0, min(95.0, pulled_to_neutral))
    
    dist = abs(final_gauge - 50.0)
    conviction = "Rất mạnh" if dist >= 25 else "Mạnh" if dist >= 15 else "Trung bình" if dist >= 8 else "Yếu"
    
    # Compatibility fields for legacy frontend (Total votes)
    buy = osc_score["buy"] + ma_score["buy"]
    sell = osc_score["sell"] + ma_score["sell"]
    neutral = osc_score["neutral"] + ma_score["neutral"]
    
    bias = "neutral"
    if final_gauge >= 58: bias = "bullish"
    elif final_gauge <= 42: bias = "bearish"
    
    return {
        "gauge": round(final_gauge, 1),
        "signal": _gauge_to_signal(final_gauge),
        "conviction": conviction,
        "bias": bias,
        "buy": buy, "sell": sell, "neutral": neutral,
        "components": {
            "oscillators": round(osc_score["gauge"], 1),
            "moving_averages": round(ma_score["gauge"], 1),
            "ai_forecast": round(ai_score["gauge"], 1)
        }
    }

def _calc_technical_score_v2(osc_score: dict, ma_score: dict, regime: str = "sideways", interval: str = "1h") -> dict:
    """Blend oscillators and moving averages with equal credibility per vote."""
    buy = osc_score["buy"] + ma_score["buy"]
    sell = osc_score["sell"] + ma_score["sell"]
    neutral = osc_score["neutral"] + ma_score["neutral"]
    final_gauge = _calc_vote_gauge(buy, sell, neutral)
    osc_delta = osc_score["gauge"] - 50.0
    ma_delta = ma_score["gauge"] - 50.0
    osc_dir = math.copysign(1.0, osc_delta) if abs(osc_delta) >= 1.0 else 0.0
    ma_dir = math.copysign(1.0, ma_delta) if abs(ma_delta) >= 1.0 else 0.0

    return {
        "gauge": round(final_gauge, 1),
        "normalized_score": round(_gauge_to_normalized_score(final_gauge), 4),
        "signal": _gauge_to_signal(final_gauge, interval),
        "buy": buy,
        "sell": sell,
        "neutral": neutral,
        "buy_weight": round(osc_score.get("buy_weight", 0.0) + ma_score.get("buy_weight", 0.0), 2),
        "sell_weight": round(osc_score.get("sell_weight", 0.0) + ma_score.get("sell_weight", 0.0), 2),
        "neutral_weight": round(osc_score.get("neutral_weight", 0.0) + ma_score.get("neutral_weight", 0.0), 2),
        "alignment": osc_dir == ma_dir and osc_dir != 0.0,
        "components": {
            "oscillators": round(osc_score["gauge"], 1),
            "moving_averages": round(ma_score["gauge"], 1),
        }
    }

def _calc_summary_score_v2(tech_score: dict, ai_score: dict, interval: str = "1h") -> dict:
    """Final decision score derived from AI certainty weighting."""
    tech_gauge = float(tech_score["gauge"])
    ai_gauge = float(ai_score["gauge"])
    ai_certainty = _clamp(float(ai_score.get("certainty", 50.0)) / 100.0, 0.0, 1.0)
    tech_weight = 1.0 - ai_certainty
    ai_weight = ai_certainty
    final_gauge = max(5.0, min(95.0, (ai_gauge * ai_weight) + (tech_gauge * tech_weight)))
    tech_delta = tech_gauge - 50.0
    ai_delta = ai_gauge - 50.0
    tech_dir = 0 if abs(tech_delta) < 2.0 else (1 if tech_delta > 0 else -1)
    ai_dir = 0 if abs(ai_delta) < 2.0 else (1 if ai_delta > 0 else -1)
    dist = abs(final_gauge - 50.0)
    conviction = "R?t m?nh" if dist >= 25 else "M?nh" if dist >= 15 else "Trung b?nh" if dist >= 8 else "Y?u"

    bias = "neutral"
    if final_gauge >= 58:
        bias = "bullish"
    elif final_gauge <= 42:
        bias = "bearish"

    return {
        "gauge": round(final_gauge, 1),
        "normalized_score": round(_gauge_to_normalized_score(final_gauge), 4),
        "signal": _gauge_to_signal(final_gauge, interval),
        "conviction": conviction,
        "bias": bias,
        "agreement": tech_dir != 0 and tech_dir == ai_dir,
        "buy": tech_score["buy"],
        "sell": tech_score["sell"],
        "neutral": tech_score["neutral"],
        "buy_weight": tech_score.get("buy_weight", 0.0),
        "sell_weight": tech_score.get("sell_weight", 0.0),
        "neutral_weight": tech_score.get("neutral_weight", 0.0),
        "components": {
            "technical": round(tech_gauge, 1),
            "oscillators": round(tech_score["components"]["oscillators"], 1),
            "moving_averages": round(tech_score["components"]["moving_averages"], 1),
            "ai_forecast": round(ai_gauge, 1),
            "ai_weight": round(ai_weight, 2),
            "technical_weight": round(tech_weight, 2),
            "certainty_pct": round(ai_certainty * 100.0, 1),
            "formula": "ai_gauge * certainty + technical_gauge * (1 - certainty)",
        }
    }


def _build_dashboard_payload(
    last_close: float,
    forecast_rows: List[Dict[str, Any]],
    technical_score: Dict[str, Any],
    ai_score: Dict[str, Any],
    summary: Dict[str, Any],
) -> Dict[str, Any]:
    """Single source of truth for hero gauges consumed by the frontend."""
    forecast_end = last_close
    if forecast_rows:
        forecast_end = float(forecast_rows[-1].get("p50") or last_close)

    return {
        "technical": {
            "gauge": technical_score["gauge"],
            "normalized_score": technical_score["normalized_score"],
            "signal": technical_score["signal"],
            "buy": technical_score["buy"],
            "sell": technical_score["sell"],
            "neutral": technical_score["neutral"],
            "buy_weight": technical_score.get("buy_weight", 0.0),
            "sell_weight": technical_score.get("sell_weight", 0.0),
            "neutral_weight": technical_score.get("neutral_weight", 0.0),
        },
        "ai": {
            "gauge": ai_score["gauge"],
            "normalized_score": ai_score["normalized_score"],
            "signal": ai_score["signal"],
            "forecast_return_pct": ai_score["forecast_return_pct"],
            "weighted_return_pct": ai_score["weighted_return_pct"],
            "confidence_pct": ai_score["confidence_pct"],
            "certainty": ai_score["certainty"],
            "path_consistency": ai_score["path_consistency"],
            "monotonicity": ai_score.get("monotonicity", 50.0),
            "max_adverse_excursion_pct": ai_score.get("max_adverse_excursion_pct", 0.0),
            "current_price": round(float(last_close), 6),
            "forecast_price": round(float(forecast_end), 6),
        },
        "summary": {
            "gauge": summary["gauge"],
            "normalized_score": summary["normalized_score"],
            "signal": summary["signal"],
            "conviction": summary["conviction"],
            "bias": summary["bias"],
            "agreement": summary.get("agreement", False),
            "buy_weight": summary.get("buy_weight", 0.0),
            "sell_weight": summary.get("sell_weight", 0.0),
            "neutral_weight": summary.get("neutral_weight", 0.0),
            "components": summary.get("components", {}),
        },
    }


async def get_indicators_cached(
    symbol: str,
    interval: str,
    limit: int,
    refresh: bool = False,
    min_context: int = INDICATOR_MIN_CONTEXT,
) -> Tuple[List[Dict[str, Any]], str, Dict[str, Any]]:
    data, source = await fetch_historical(
        symbol,
        interval,
        limit,
        refresh=refresh,
        min_context=max(limit, min_context),
    )
    cache_key = f"ind_{_cache_prefix(symbol, interval)}len={len(data)}"

    if refresh:
        indicators_cache.delete(cache_key)
    else:
        cached = indicators_cache.get(cache_key)
        if cached is not None:
            try:
                return data, str(cached["source"]), dict(cached["indicators"])
            except Exception:
                indicators_cache.delete(cache_key)

    inflight_key = f"{cache_key}:refresh={int(refresh)}"
    task = None if refresh else _INDICATORS_INFLIGHT.get(inflight_key)
    if task is None:
        task = asyncio.create_task(
            asyncio.to_thread(compute_indicators, data),
            name=f"indicators:{symbol}:{interval}:{len(data)}",
        )
        _INDICATORS_INFLIGHT[inflight_key] = task

    try:
        indicators = await task
    finally:
        if _INDICATORS_INFLIGHT.get(inflight_key) is task:
            _INDICATORS_INFLIGHT.pop(inflight_key, None)

    indicators_cache.set(
        cache_key,
        {"source": source, "indicators": indicators},
        ttl_seconds=indicators_ttl(interval),
    )
    return data, source, indicators


def _rebuild_blended_from_forecast_payload(cached_forecast: Optional[Dict[str, Any]], last_close: float) -> Optional[Dict[str, Any]]:
    """Reconstruct the minimum blended payload needed by the analysis engine."""
    if not cached_forecast:
        return None

    rows = cached_forecast.get("forecast") or []
    future_rows = [row for row in rows if not row.get("is_actual")]
    if not future_rows:
        return None

    ensemble = cached_forecast.get("ensemble", {})
    return {
        "p10": np.array([float(row.get("p10") or last_close) for row in future_rows], dtype=float),
        "p50": np.array([float(row.get("p50") or last_close) for row in future_rows], dtype=float),
        "p90": np.array([float(row.get("p90") or last_close) for row in future_rows], dtype=float),
        "agreement": bool(ensemble.get("trend_agreement", True)),
        "scale": float(ensemble.get("alignment_scale", 1.0) or 1.0),
        "confidence": float(ensemble.get("confidence", 50.0) or 50.0),
    }

def _build_trade_analysis(
    symbol: str, interval: str, data: List[Dict[str, Any]], indicators: Dict[str, Any],
    forecast_rows: List[Dict[str, Any]], confidence: float, source: str,
    blended: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
    """
    TradingView-style technical analysis dashboard (v6.1 Rework).
    Implements weighted scoring for Oscillators, MAs, and AI Forecast.
    """
    if not data or len(data) < 30:
        return {"oscillators": {"data": []}, "moving_averages": {"data": []}, "summary": {"signal": "Trung lập", "buy": 0, "sell": 0, "neutral": 0}}

    closes = np.array([float(d["close"]) for d in data], dtype=float)
    highs  = np.array([float(d["high"])  for d in data], dtype=float)
    lows   = np.array([float(d["low"])   for d in data], dtype=float)
    vols   = np.array([float(d.get("volume", 0)) for d in data], dtype=float)
    last_close = closes[-1]

    def _lv(arr):
        if arr is None: return None
        if isinstance(arr, (float, int)): return round(float(arr), 2)
        if isinstance(arr, pd.Series): arr = arr.values
        v = arr[-1] if (hasattr(arr, '__len__') and len(arr) > 0) else float('nan')
        if v is None or (isinstance(v, float) and math.isnan(v)): return None
        return round(float(v), 2)

    ema50_arr = _ema(closes, 50)
    ema50_current = _lv(ema50_arr)
    ema200_current = _lv(indicators["ema"].get("ema200"))
    ema50_prev = (
        float(ema50_arr[-6])
        if len(ema50_arr) >= 6 and not math.isnan(float(ema50_arr[-6]))
        else (ema50_current or last_close)
    )

    regime, regime_label = _classify_regime(indicators, last_close, _safe(indicators.get("atr", {}).get("pct"), 1.0))

    # ── 1. Oscillators ──
    osc_data = []
    atr_scale = max(float(indicators.get("atr", {}).get("value") or last_close * 0.01), max(last_close * 0.0015, 1e-6))
    def _add_osc(label, val, action_name, key=None, **kw):
        if isinstance(val, (np.ndarray, pd.Series, list)): v = _lv(val)
        else: v = round(float(val), 2) if val is not None else None
        osc_key = key or action_name
        osc_kwargs = {
            "last_close": last_close,
            "ema50": ema50_current or last_close,
            "ema200": ema200_current or last_close,
            "ema50_prev": ema50_prev,
            "atr": atr_scale,
            **kw,
        }
        act = _osc_action(action_name, v if v is not None else 0, **osc_kwargs)
        score = _osc_signal_score(action_name, v if v is not None else 0.0, **osc_kwargs)
        osc_data.append({"key": osc_key, "name": label, "value": v, "action": act, "score": round(score, 4)})

    stoch_price_k, stoch_price_d = _stoch_kd(highs, lows, closes, 14, 3, 3)
    stoch_rsi_k, stoch_rsi_d = _stoch_rsi(closes, 14, 14, 3, 3)
    demarker14 = _demarker(highs, lows, 14)
    force_idx13 = _force_index(closes, vols, 13)
    force_scale = float(np.nanmean(np.abs(force_idx13[-20:]))) if len(force_idx13) else 1.0
    if not math.isfinite(force_scale) or force_scale <= 1e-8:
        force_scale = max(float(np.nanmean(np.abs(force_idx13))) if len(force_idx13) else 1.0, 1.0)
    _add_osc("Chỉ số Sức mạnh tương đối (14)", _rsi(closes, 14), "rsi", key="rsi")
    _add_osc("Stochastic %K (14, 3, 3)", stoch_price_k, "stoch", key="stoch")
    _add_osc("Chỉ số Kênh hàng hóa (20)", _cci(highs, lows, closes, 20), "cci", key="cci")
    adx_vals = _adx(highs, lows, closes, 14)
    _add_osc("Chỉ số Định hướng Trung bình (14)", adx_vals[0], "adx", key="adx", plus_di=_lv(adx_vals[1]) or 0, minus_di=_lv(adx_vals[2]) or 0)
    ao_vals = _awesome_oscillator(highs, lows)
    ao_prev = _lv(ao_vals[:-1]) if len(ao_vals) > 1 else _lv(ao_vals)
    _add_osc("Chỉ số Dao động AO", ao_vals, "ao", key="ao", scale=atr_scale * 0.8, prev=ao_prev or 0.0)
    _add_osc("Xung lượng (10)", _momentum(closes, 10), "momentum", key="momentum", scale=atr_scale * 1.15)
    macd_vals = _macd(closes, 12, 26, 9)
    _add_osc("Cấp độ MACD (12, 26)", macd_vals[0], "macd", key="macd", signal=_lv(macd_vals[1]) or 0, scale=atr_scale * 0.18)
    _add_osc("Đường RSI Nhanh (3, 3, 14, 14)", stoch_rsi_k, "stoch_rsi", key="stoch_rsi")
    _add_osc("Vùng Phần trăm Williams (14)", _williams_r(highs, lows, closes, 14), "williams", key="williams")
    bbp_vals = _bull_bear_power(highs, lows, closes, 13)
    _add_osc("Sức Mạnh Giá Lên và Giá Xuống", bbp_vals, "bbp", key="bbp", scale=atr_scale * 0.9, bbp_norm=(_lv(bbp_vals) or 0.0) / atr_scale)
    _add_osc("Dao động Ultimate (7, 14, 28)", _ultimate_oscillator(highs, lows, closes, 7, 14, 28), "ultimate", key="ultimate")
    _add_osc("Tốc độ biến động ROC (12)", _roc(closes, 12), "roc", key="roc")
    _add_osc("TRIX (18)", _trix(closes, 18), "trix", key="trix")
    _add_osc("PPO (12, 26)", _ppo(closes, 12, 26), "ppo", key="ppo")
    _add_osc("CMO (14)", _cmo(closes, 14), "cmo", key="cmo")
    _add_osc("DPO (20)", _dpo(closes, 20), "dpo", key="dpo", scale=atr_scale * 0.75)
    _add_osc("Aroon Oscillator (25)", _aroon_oscillator(highs, lows, 25), "aroon", key="aroon")
    _add_osc("TSI (25, 13)", _tsi(closes, 25, 13), "tsi", key="tsi")
    _add_osc("DeMarker (14)", demarker14, "demarker", key="demarker")
    _add_osc("Force Index (13)", force_idx13, "force_index", key="force_index", scale=force_scale)

    osc_score = _calc_osc_score(osc_data, interval)

    # ── 2. Moving Averages ──
    ma_data = []
    ema_map: Dict[int, np.ndarray] = {
        1: closes.copy(),
        5: _ema(closes, 5),
        10: _ema(closes, 10),
        20: _ema(closes, 20),
        50: _ema(closes, 50),
        100: _ema(closes, 100),
        200: _ema(closes, 200),
    }

    def _add_ema_pair_row(fast_period: int, slow_period: int) -> None:
        fast_arr = ema_map[fast_period]
        slow_arr = ema_map[slow_period]
        fast_value = _lv(fast_arr)
        slow_value = _lv(slow_arr)
        pair_key = f"ema_{fast_period}_{slow_period}"
        act = _ema_pair_action(fast_value, slow_value)
        score = _ema_pair_score(fast_value, slow_value, pair_key)
        ma_data.append(
            {
                "key": pair_key,
                "name": f"EMA{fast_period} / EMA{slow_period}",
                "value": f"{fast_value if fast_value is not None else '—'} / {slow_value if slow_value is not None else '—'}",
                "action": act,
                "score": round(score, 4),
                "fast": fast_value,
                "slow": slow_value,
            }
        )

    for fast_period, slow_period in [(1, 5), (5, 10), (10, 20), (20, 50), (50, 100), (100, 200)]:
        _add_ema_pair_row(fast_period, slow_period)

    ma_score = _calc_ma_score(ma_data, closes, interval)

    # ── 3. AI Forecast Gauge ──
    if not blended:
        # Fallback to simple blended if no forecast available
        blended = {
            "p50": [last_close * (1 + (confidence-50)/1000)],
            "p10": [last_close * 0.98], "p90": [last_close * 1.02],
            "agreement": True, "scale": 1.0
        }
    
    ai_score = _calc_ai_forecast_score(
        blended,
        forecast_rows,
        last_close,
        indicators,
        len(forecast_rows) or 10,
        interval,
    )

    # ── 4. Summary ──
    technical_score = _calc_technical_score_v2(osc_score, ma_score, regime, interval)
    summary = _calc_summary_score_v2(technical_score, ai_score, interval)

    # Pivot Points
    last_h = float(highs[-2]) if len(highs) > 1 else float(highs[-1])
    last_l = float(lows[-2]) if len(lows) > 1 else float(lows[-1])
    last_c = float(closes[-2]) if len(closes) > 1 else float(closes[-1])
    pivots = _calc_pivot_points(last_h, last_l, last_c)
    dashboard = _build_dashboard_payload(last_close, forecast_rows, technical_score, ai_score, summary)

    return {
        "style": "tradingview",
        "regime": {"key": regime, "label": regime_label},
        "dashboard": dashboard,
        "summary": summary,
        "technicals": technical_score,
        "oscillators": {
            "gauge": osc_score["gauge"],
            "signal": osc_score["signal"],
            "buy": osc_score["buy"],
            "sell": osc_score["sell"],
            "neutral": osc_score["neutral"],
            "buy_weight": osc_score.get("buy_weight", 0.0),
            "sell_weight": osc_score.get("sell_weight", 0.0),
            "neutral_weight": osc_score.get("neutral_weight", 0.0),
            "data": osc_data
        },
        "moving_averages": {
            "gauge": ma_score["gauge"],
            "signal": ma_score["signal"],
            "buy": ma_score["buy"],
            "sell": ma_score["sell"],
            "neutral": ma_score["neutral"],
            "buy_weight": ma_score.get("buy_weight", 0.0),
            "sell_weight": ma_score.get("sell_weight", 0.0),
            "neutral_weight": ma_score.get("neutral_weight", 0.0),
            "golden_cross": ma_score["golden_cross"],
            "death_cross": ma_score["death_cross"],
            "data": ma_data
        },
        "ai_gauge": ai_score,
        "pivot_points": pivots
    }



START_TIME = time.time()

async def _background_cleanup():
    """Evict expired entries from TTL caches periodically (v6.0: Aggressive)."""
    while True:
        await asyncio.sleep(120)  # Every 2 minutes
        try:
            h = historical_cache.evict_expired()
            f = forecast_cache.evict_expired()
            t = ticker_cache.evict_expired()
            persistent_cache.evict()
            logger.info("[cleanup] Evicted h=%d f=%d t=%d", h, f, t)
        except Exception as ex:
            logger.warning("[cleanup] Error: %s", ex)

async def _periodic_health_check():
    """Check circuit breakers and log source health every 2 minutes."""
    while True:
        await asyncio.sleep(120)
        for name, cb in source_breakers.items():
            if cb.state == "OPEN":
                logger.warning("[health] CB OPEN: %s (failures=%d)", name, cb.failures)

def _clear_ip_limits():
    """Remove stale IP rate-limit entries older than 60 seconds."""
    stale_count = clear_stale_ip_limits(IP_LIMITS, now=time.time(), window_seconds=60)
    if stale_count:
        logger.debug("[ip-cleanup] Removed %d stale IPs", stale_count)


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup logic
    logger.info("Starting Kronos AI Backend v%s...", APP_VERSION)
    background_tasks: List[asyncio.Task[Any]] = []

    def _start_background_task(coro: Any, name: str) -> None:
        background_tasks.append(asyncio.create_task(coro, name=name))

    # Async background tasks
    _start_background_task(persistent_cache.start_writer(), "persistent-cache-writer")
    _start_background_task(ws_manager.heartbeat(), "ws-heartbeat")
    _start_background_task(_background_cleanup(), "cache-cleanup")
    _start_background_task(_periodic_health_check(), "periodic-health-check")
    
    # F-2: Rate limit cleanup task
    async def _ip_cleanup_loop():
        while True:
            await asyncio.sleep(300) # BUG-P1-07: reduced from 3600 to 300
            _clear_ip_limits()
    _start_background_task(_ip_cleanup_loop(), "ip-limit-cleanup")

    # Quick source reachability check (non-blocking)
    _start_background_task(_source_selftest(), "source-selftest")
    
    if PRELOAD_TIMESFM and TIMESFM_AVAILABLE:
        _start_background_task(_warmup_timesfm(), "timesfm-warmup")
    elif not TIMESFM_AVAILABLE:
        STARTUP_STATE["timesfm"]["last_error"] = TIMESFM_IMPORT_ERROR or "TimesFM import failed"
    
    try:
        yield
    finally:
        logger.info("Shutting down AI Trading Chart API...")
        for task in background_tasks:
            task.cancel()
        if background_tasks:
            await asyncio.gather(*background_tasks, return_exceptions=True)
        await GlobalHTTPClient.close()
        logger.info("Graceful shutdown completed.")


# ──────────────────────────────────────────────────────────────────────────────
# FastAPI App Instance (v6.0)
# ──────────────────────────────────────────────────────────────────────────────
app = FastAPI(
    title="AI Trading Chart API",
    version=APP_VERSION,
    description="OHLCV data, hybrid AI forecasts, technical indicators, and real-time WebSocket prices",
    lifespan=lifespan,
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=CORS_ALLOW_ORIGINS,
    allow_credentials=CORS_ALLOW_CREDENTIALS,
    allow_methods=["*"], allow_headers=["*"],
)


# ──────────────────────────────────────────────────────────────────────────────
# Market Hours (UTC)
# ──────────────────────────────────────────────────────────────────────────────
MARKET_SESSIONS = MARKET_SESSIONS_DATA


def market_status_now() -> List[Dict[str, Any]]:
    return compute_market_status_now()


# ──────────────────────────────────────────────────────────────────────────────
# ──────────────────────────────────────────────────────────────────────────────
# TimesFM 2.5 Forecaster
# ──────────────────────────────────────────────────────────────────────────────
class TimesFMForecaster:
    """Async wrapper around Google TimesFM 2.5 (200M, PyTorch).

    model.forecast(horizon, inputs) → (point_forecast, quantile_forecast)
      point_forecast:    np.ndarray (batch, horizon)
      quantile_forecast: np.ndarray (batch, horizon, 10)
        index mapping: [mean, q10, q20, q30, q40, q50, q60, q70, q80, q90]

    Input:  OHLCV DataFrame; OHLC4 = (O+H+L+C)/4 computed internally.
    Output: p10/p50/p90 future OHLC4 values for `horizon` future bars.
    """
    MODEL_HF_ID = "google/timesfm-2.5-200m-pytorch"
    # Keep context comfortably within 16384 limit; leaves room for any horizon
    MAX_CONTEXT   = 8_192
    MAX_HORIZON   = 300    # hard cap — matches API limit of horizon<=300
    # quantile axis indices in the (batch, horizon, 10) tensor
    _Q10 = 1   # 10th percentile
    _Q50 = 5   # 50th percentile (median)
    _Q90 = 9   # 90th percentile

    def __init__(self) -> None:
        self._model: Optional[Any] = None
        self._loaded = False
        self._lock: Optional[asyncio.Lock] = None
        self._predict_lock: Optional[asyncio.Lock] = None
        self._compiled_horizon: Optional[int] = None  # last compiled max_horizon

    async def _get_lock(self) -> asyncio.Lock:
        if self._lock is None:
            self._lock = asyncio.Lock()
        return self._lock

    async def _get_predict_lock(self) -> asyncio.Lock:
        if self._predict_lock is None:
            self._predict_lock = asyncio.Lock()
        return self._predict_lock

    @property
    def is_ready(self) -> bool:
        return self._loaded

    @property
    def device(self) -> str:
        if self._model is None:
            return "not_loaded"
        try:
            return str(self._model.model.device)
        except Exception:
            return "cpu"

    def _compile(self, horizon: int) -> None:
        """(Re-)compile with a max_horizon that covers `horizon`."""
        import math
        output_patch = 128  # TimesFM 2.5 output_patch_len
        # max_horizon must be a multiple of output_patch_len
        max_h = math.ceil(horizon / output_patch) * output_patch
        max_h = max(max_h, output_patch)
        max_h = min(max_h, self.MAX_HORIZON)
        # context must be multiple of input_patch_len=32
        ctx = self.MAX_CONTEXT  # already a multiple of 32
        self._model.compile(
            timesfm.ForecastConfig(
                max_context=ctx,
                max_horizon=max_h,
                normalize_inputs=True,
                use_continuous_quantile_head=True,
                force_flip_invariance=True,
                infer_is_positive=True,
                fix_quantile_crossing=True,
            )
        )
        self._compiled_horizon = max_h
        logger.info("[TimesFM] Compiled: ctx=%d max_horizon=%d", ctx, max_h)

    async def _lazy_load(self) -> None:
        if self._loaded:
            return
        lock = await self._get_lock()
        async with lock:
            if self._loaded:
                return
            if not TIMESFM_AVAILABLE:
                raise HTTPException(
                    status_code=503,
                    detail="TimesFM not installed. Clone https://github.com/google-research/timesfm and pip install -e .[torch]",
                )
            try:
                logger.info("[TimesFM] Loading %s ...", self.MODEL_HF_ID)
                model = await asyncio.to_thread(
                    timesfm.TimesFM_2p5_200M_torch.from_pretrained,
                    self.MODEL_HF_ID,
                )
                # Compile once with the default max_horizon
                self._model = model
                self._compile(self.MAX_HORIZON)
                self._loaded = True
                STARTUP_STATE["timesfm"]["loaded"] = True
                STARTUP_STATE["timesfm"]["device"] = self.device
                logger.info("[TimesFM] Ready on %s", self.device)
            except Exception as ex:
                STARTUP_STATE["timesfm"]["last_error"] = str(ex)
                logger.error("[TimesFM] Init failed: %s", ex, exc_info=True)
                raise HTTPException(status_code=500, detail=f"TimesFM init failed: {ex}")

    async def forecast(
        self,
        df: pd.DataFrame,
        horizon: int,
        # legacy kwargs — accepted and ignored for API compatibility
        x_timestamp=None,
        y_timestamp=None,
        sample_count: int = 0,
    ) -> Dict[str, Any]:
        await self._lazy_load()
        assert self._model is not None

        try:
            # Build single OHLC4 series: (O+H+L+C)/4
            ohlc4 = (
                df[["open", "high", "low", "close"]]
                .mean(axis=1)
                .astype(np.float32)
                .values
            )
            context_len = min(len(ohlc4), self.MAX_CONTEXT)
            ohlc4_ctx = ohlc4[-context_len:]

            # Re-compile if horizon exceeds current compiled max_horizon
            if self._compiled_horizon is None or horizon > self._compiled_horizon:
                self._compile(horizon)

            t0 = time.time()
            predict_lock = await self._get_predict_lock()
            async with predict_lock:
                # TimesFM API: forecast(horizon: int, inputs: list[np.ndarray])
                point_forecast, quantile_forecast = await asyncio.to_thread(
                    self._model.forecast,
                    horizon,
                    [ohlc4_ctx],
                )
            elapsed = time.time() - t0
            logger.info(
                "[TimesFM] %.2fs | horizon=%d ctx=%d device=%s",
                elapsed, horizon, context_len, self.device,
            )

            if torch is not None and torch.cuda.is_available():
                torch.cuda.empty_cache()

            # point_forecast:    (1, horizon)
            # quantile_forecast: (1, horizon, 10)
            p10 = quantile_forecast[0, :horizon, self._Q10].astype(float)
            p50 = quantile_forecast[0, :horizon, self._Q50].astype(float)
            p90 = quantile_forecast[0, :horizon, self._Q90].astype(float)

            return {
                "p10": p10,
                "p50": p50,
                "p90": p90,
                "model_name": self.MODEL_HF_ID,
                "context_length": context_len,
                "output_horizon": horizon,
                "input_semantics": {
                    "feature_channels": ["ohlc4"],
                    "active_forecast_channels": ["ohlc4"],
                    "ignored_channels": ["volume"],
                    "price_mode": "ohlc4_single_channel",
                    "base_signal": "ohlc4",
                    "volume_mode": "omitted",
                    "amount_mode": "omitted",
                    "adapter_mode": "timesfm_native",
                    "normalization": "timesfm_internal_revin",
                },
                "output_semantics": {
                    "forecast_channel": "ohlc4",
                    "forecast_mode": "single_future_ohlc4_line",
                    "quantile_fields": ["p10", "p50", "p90"],
                    "candle_projection": "omitted",
                },
            }
        except Exception as ex:
            logger.error("[TimesFM] Forecast failed: %s", ex, exc_info=True)
            raise HTTPException(status_code=500, detail=f"TimesFM prediction failed: {ex}")




forecaster = TimesFMForecaster()



# MODULE: Analysis Engine v2.0 (Relocated and Activated)
# Legacy placeholders removed to avoid duplication with logic at line 1884.


# ──────────────────────────────────────────────────────────────────────────────
# WebSocket connection manager
# ──────────────────────────────────────────────────────────────────────────────
class ConnectionManager:
    def __init__(self):
        self.active: Dict[str, List[WebSocket]] = defaultdict(list)

    async def connect(self, ws: WebSocket, symbol: str):
        await ws.accept()
        self.active[symbol].append(ws)
        logger.info("[WS] connect: %s (total=%d)", symbol, len(self.active[symbol]))

    def disconnect(self, ws: WebSocket, symbol: str):
        self.active[symbol] = [c for c in self.active[symbol] if c is not ws]
        logger.info("[WS] disconnect: %s (total=%d)", symbol, len(self.active[symbol]))

    async def broadcast(self, symbol: str, data: Dict[str, Any]):
        """Reliable broadcast with automatic stale connection cleanup (P0)."""
        active_list = self.active.get(symbol, [])
        if not active_list:
            return

        dead = []
        for ws in list(active_list):
            try:
                if ws.client_state == WebSocketState.CONNECTED:
                    await ws.send_json(data)
                else:
                    dead.append(ws)
            except Exception:
                dead.append(ws)
        
        for ws in dead:
            self.disconnect(ws, symbol)

    async def heartbeat(self):
        """Keep-alive loop: send ping to all active clients every 20s (BUG-13)."""
        while True:
            await asyncio.sleep(20)
            ping_msg = {"type": "ping", "ts": int(time.time())}
            for symbol in list(self.active.keys()):
                for ws in list(self.active[symbol]):
                    try:
                        if ws.client_state == WebSocketState.CONNECTED:
                            await ws.send_json(ping_msg)
                    except Exception:
                        pass


ws_manager = ConnectionManager()
@app.websocket("/ws/price/{symbol}")
async def websocket_price(websocket: WebSocket, symbol: str):
    """
    B-6: Real-time price streaming via WebSocket.
    Fixes the 404/500 errors in chrome console.
    """
    symbol = _get_canonical_symbol(symbol)
    if symbol not in SYMBOLS:
        await websocket.close(code=1008, reason=f"Unknown symbol: {symbol}")
        return
    interval = websocket.query_params.get("interval") or "1d"

    try:
        await ws_manager.connect(websocket, symbol)
        while True:
            try:
                # 1. State check
                if websocket.client_state != WebSocketState.CONNECTED:
                    break

                # 2. Fetch fresh price
                ticker = await fetch_ticker(symbol, interval=interval)
                
                # 3. Final state check before send
                if websocket.client_state == WebSocketState.CONNECTED:
                    await websocket.send_json({
                        "type": "price",
                        "symbol": symbol,
                        "price": ticker.get("price"),
                        "change_pct": ticker.get("change_pct", 0),
                        "ts": int(time.time())
                    })
            except (WebSocketDisconnect, RuntimeError):
                break
            except Exception as ex:
                logger.debug("[WS] ticker fetch failed for %s: %s", symbol, ex)
            
            interval = 5 if SYMBOLS[symbol].category == "Crypto" else 30
            await asyncio.sleep(interval)
    except Exception as ex:
        logger.error("[WS] handler error for %s: %s", symbol, ex)
    finally:
        ws_manager.disconnect(websocket, symbol)

# ──────────────────────────────────────────────────────────────────────────────
# Architecture v5: Security & Guardrails (Module F)
# ──────────────────────────────────────────────────────────────────────────────
from fastapi.responses import FileResponse, HTMLResponse, JSONResponse

IP_LIMITS: Dict[str, List[float]] = defaultdict(list)
REQUEST_METRICS = RequestMetricsRegistry()

def rate_limit_guard(request: Request):
    """
    F-2: Simple IP-based rate limiting logic.
    BUG-P1-11: Whitelist health and metrics from rate limiting.
    """
    return check_rate_limit(request, IP_LIMITS)

def admin_only(request: Request):
    """F-1: Admin authentication guard for sensitive endpoints."""
    return require_admin_token(request, ADMIN_TOKEN)


def _validate_symbol_and_interval(symbol: str, interval: str) -> Tuple[str, str]:
    return validate_symbol_and_interval(
        symbol,
        interval,
        get_canonical_symbol=_get_canonical_symbol,
        symbols=SYMBOLS,
        supported_intervals=SUPPORTED_INTERVALS,
    )


def _normalize_watchlist_symbols(symbols: List[str]) -> Tuple[List[str], List[str], int]:
    return normalize_watchlist_symbols(
        symbols,
        get_canonical_symbol=_get_canonical_symbol,
        symbol_registry=SYMBOLS,
    )


def _validate_cache_target(target: str) -> str:
    return validate_cache_target(target)


# ──────────────────────────────────────────────────────────────────────────────
# Pydantic Models
# ──────────────────────────────────────────────────────────────────────────────
# BUG-P0-01: Removed duplicate v5 app instances and buggy lifespan assignment
# Application v6.0 is defined earlier at line 2532

@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    request_id = request.headers.get("X-Request-ID") or make_request_id()
    timer_start = start_timer()
    route_path = request.url.path

    if request.url.path not in RATE_LIMIT_WHITELIST and rate_limit_guard(request):
        duration_ms = elapsed_ms(timer_start)
        REQUEST_METRICS.record(route_path, 429, duration_ms)
        return JSONResponse(
            status_code=429,
            content={"detail": "Too many requests", "request_id": request_id},
            headers={
                "X-Request-ID": request_id,
                "X-Response-Time-Ms": f"{duration_ms:.2f}",
            },
        )

    try:
        response = await call_next(request)
    except Exception as ex:
        duration_ms = elapsed_ms(timer_start)
        REQUEST_METRICS.record(route_path, 500, duration_ms)
        logger.error(
            "[request] %s %s failed | request_id=%s | %.2fms | %s",
            request.method,
            route_path,
            request_id,
            duration_ms,
            ex,
            exc_info=True,
        )
        return JSONResponse(
            status_code=500,
            content={"detail": "Internal server error", "request_id": request_id},
            headers={
                "X-Request-ID": request_id,
                "X-Response-Time-Ms": f"{duration_ms:.2f}",
            },
        )

    duration_ms = elapsed_ms(timer_start)
    REQUEST_METRICS.record(route_path, response.status_code, duration_ms)
    response.headers["X-Request-ID"] = request_id
    response.headers["X-Response-Time-Ms"] = f"{duration_ms:.2f}"
    return response



async def _source_selftest():
    """Ping data sources at startup to confirm reachability (1 attempt each)."""
    tests = build_source_selftest_urls(TWELVEDATA_API_KEY, FINNHUB_API_KEY)
    async with httpx.AsyncClient(timeout=8) as c:
        await run_source_selftest(
            client=c,
            tests=tests,
            startup_sources=STARTUP_STATE["sources"],
            logger=logger,
        )


async def _warmup_timesfm() -> None:
    """Load TimesFM in the background so the first forecast is fast."""
    try:
        STARTUP_STATE["timesfm"]["warming"] = True
        await forecaster._lazy_load()
        STARTUP_STATE["timesfm"]["warming"] = False
        logger.info("[TimesFM] Warmup complete — ready on %s", forecaster.device)
    except Exception as ex:
        STARTUP_STATE["timesfm"]["warming"] = False
        logger.error("[TimesFM] Warmup failed: %s", ex)


# ── 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:
        _SYMBOLS_CACHE = build_symbols_catalog(SYMBOLS, INTERVAL_ORDER)

    if category:
        return filter_symbols_catalog(_SYMBOLS_CACHE, category)

    return _SYMBOLS_CACHE


async def _fetch_ticker_with_semaphore(symbol: str, semaphore: asyncio.Semaphore) -> Dict[str, Any]:
    async with semaphore:
        return await fetch_ticker(symbol)


# ── Market Peers ──────────────────────────────────────────────────────────────
@app.get("/api/market-peers")
async def get_market_peers(
    symbol: str = Query("BTCUSD"),
    limit: int = Query(10, ge=1, le=20),
) -> Dict[str, Any]:
    symbol = _get_canonical_symbol(symbol)
    candidate_payload = build_market_peer_candidates(SYMBOLS, symbol, limit)
    peer_configs = candidate_payload.get("peer_configs", [])
    if not peer_configs:
        return candidate_payload

    tasks = [
        _fetch_ticker_with_semaphore(peer.symbol, _market_peer_ticker_semaphore)
        for peer in peer_configs
    ]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return assemble_market_peer_payload(
        category=str(candidate_payload["category"]),
        symbol=str(candidate_payload["symbol"]),
        peer_configs=peer_configs,
        peer_results=results,
    )



# ── Symbol search ─────────────────────────────────────────────────────────────
@app.get("/api/search")
async def search_symbols(
    q: str = Query(..., min_length=1, description="Search query"),
) -> Dict[str, Any]:
    return search_symbols_catalog(SYMBOLS, q)


# ── Historical OHLCV ──────────────────────────────────────────────────────────
@app.get("/api/historical/{symbol}")
async def get_historical(
    symbol:   str,
    interval: str = Query("1h"),
    limit:    int = Query(500, ge=50, le=2000),
    refresh:  bool = Query(False),
) -> Dict[str, Any]:
    symbol = _get_canonical_symbol(symbol)
    if symbol not in SYMBOLS:
        raise HTTPException(404, f"Unknown symbol: {symbol}")
    if interval not in SUPPORTED_INTERVALS:
        raise HTTPException(400, f"Unsupported interval: {interval}")
    data, source = await fetch_historical(
        symbol,
        interval,
        limit,
        refresh=refresh,
        min_context=limit,
    )
    return {"symbol": symbol, "interval": interval, "source": source,
            "count": len(data), "data": data,
            "generated_at": int(time.time()),
            "cache": {"refresh_requested": refresh}}


# ── Technical Indicators ──────────────────────────────────────────────────────
@app.get("/api/indicators/{symbol}")
async def get_indicators(
    symbol:   str,
    interval: str = Query("1h"),
    limit:    int = Query(300, ge=50, le=1000),
    refresh:  bool = Query(False),
) -> Dict[str, Any]:
    symbol = _get_canonical_symbol(symbol)
    if symbol not in SYMBOLS:
        raise HTTPException(404, f"Unknown symbol: {symbol}")
    if interval not in SUPPORTED_INTERVALS:
        raise HTTPException(400, f"Unsupported interval: {interval}")

    data, source, indicators = await get_indicators_cached(
        symbol,
        interval,
        limit,
        refresh=refresh,
        min_context=max(limit, INDICATOR_MIN_CONTEXT),
    )
    return {
        "symbol":   symbol,
        "interval": interval,
        "source":   source,
        "candles":  len(data),
        "generated_at": int(time.time()),
        "cache": {"refresh_requested": refresh},
        "indicators": indicators,
    }


# ── Technical Analysis Engine ─────────────────────────────────────────────────
@app.get("/api/analysis/{symbol}")
async def get_analysis(
    symbol:   str,
    interval: str = Query("1h"),
    refresh:  bool = Query(False),
    include_snapshot: bool = Query(False),
) -> Dict[str, Any]:
    """
    A-5: Direct access to the comprehensive Analysis Engine.
    E-2: Integrated Multi-timeframe (MTF) analysis.
    """
    symbol = _get_canonical_symbol(symbol)
    if symbol not in SYMBOLS:
        raise HTTPException(404, f"Unknown symbol: {symbol}")

    # Fetch main context
    data, source, indicators = await get_indicators_cached(
        symbol,
        interval,
        500,
        refresh=refresh,
        min_context=ANALYSIS_CONTEXT,
    )
    if len(data) < 50:
        raise HTTPException(422, "Insufficient data for full analysis")

    # Fetch higher timeframe context (MTF)
    htf_interval = "1d" if interval in ["1h", "2h", "4h"] else "1h" if interval in ["5m", "15m"] else None
    htf_bias = "neutral"
    if htf_interval:
        try:
            _, _, htf_inds = await get_indicators_cached(
                symbol,
                htf_interval,
                200,
                refresh=refresh,
                min_context=HIGHER_TIMEFRAME_CONTEXT,
            )
            htf_bias = "bullish" if htf_inds["trend"].get("above_ema200") else "bearish"
        except Exception as ex:
            logger.warning(
                "[analysis] Higher timeframe context unavailable for %s %s via %s: %s",
                symbol,
                interval,
                htf_interval,
                ex,
            )

    # Reuse cached forecast when available so /analysis and /forecast stay consistent.
    forecast_ret = 0.0
    confidence = 50.0
    forecast_rows: List[Dict[str, Any]] = []
    blended: Optional[Dict[str, Any]] = None
    try:
        f_prefix = _cache_prefix(symbol, interval)
        f_cache = forecast_cache.get(f"forecast_{f_prefix}10")
        if f_cache:
            forecast_rows = f_cache.get("forecast") or []
            if forecast_rows:
                forecast_ret = _pct(float(forecast_rows[-1]["p50"]), float(f_cache.get("last_close") or data[-1]["close"]))
            confidence = float(f_cache.get("ensemble", {}).get("confidence") or 50.0)
            blended = _rebuild_blended_from_forecast_payload(f_cache, float(data[-1]["close"]))
    except Exception as ex:
        logger.warning("[analysis] Failed to reuse cached forecast for %s %s: %s", symbol, interval, ex)

    # Build comprehensive analysis
    analysis = await asyncio.to_thread(
        _build_trade_analysis,
        symbol=symbol,
        interval=interval,
        data=data,
        indicators=indicators,
        forecast_rows=forecast_rows,
        confidence=confidence,
        source=source,
        blended=blended,
    )
    
    # Inject MTF into analysis
    analysis["multi_timeframe"] = {
        "htf_interval": htf_interval,
        "htf_bias": htf_bias,
        "alignment": bool((analysis["summary"]["bias"] == "bullish" and htf_bias == "bullish") or
                          (analysis["summary"]["bias"] == "bearish" and htf_bias == "bearish"))
    }

    return {
        "symbol": symbol,
        "interval": interval,
        "timestamp": int(time.time()),
        "analysis": analysis,
        "verdict": await get_local_ai_verdict(symbol, interval, analysis, forecast_ret, refresh=refresh),
        "verdict_source": "local_rules",
        "ai_rules": {
            "version": ai_rule_registry.version,
            "loaded_count": len(ai_rule_registry.documents),
        },
        "cache": {"refresh_requested": refresh},
        "indicators_snapshot": indicators if include_snapshot else None
    }


# ── Real-time ticker ──────────────────────────────────────────────────────────
def _derive_local_verdict(analysis: Dict[str, Any], forecast_pct: float) -> str:
    summary = analysis.get("summary", {})
    technicals = analysis.get("technicals", {})
    ai_gauge = analysis.get("ai_gauge", {})
    multi_timeframe = analysis.get("multi_timeframe", {})

    bias = str(summary.get("bias", "neutral"))
    tech_gauge = float(technicals.get("gauge") or 50.0)
    ai_score = float(ai_gauge.get("gauge") or 50.0)
    aligned = bool(multi_timeframe.get("alignment", False))
    confidence = float(ai_gauge.get("confidence_pct") or 0.0)

    bullish_stack = bias == "bullish" and tech_gauge >= 58.0 and ai_score >= 56.0
    bearish_stack = bias == "bearish" and tech_gauge <= 42.0 and ai_score <= 44.0

    if bullish_stack and forecast_pct >= 0.6 and aligned and confidence >= 45.0:
        return "Mua ngay"
    if bearish_stack and forecast_pct <= -0.6 and aligned and confidence >= 45.0:
        return "Ban ngay"
    if bias == "bullish":
        return "Nen doi mua gia thap hon" if forecast_pct < 0.8 else "Mua ngay"
    if bias == "bearish":
        return "Nen doi ban gia cao hon" if forecast_pct > -0.8 else "Ban ngay"
    return "Nen doi mua gia thap hon" if forecast_pct >= 0 else "Nen doi ban gia cao hon"


async def get_local_ai_verdict(
    symbol: str,
    interval: str,
    analysis: Dict[str, Any],
    forecast_pct: float = 0.0,
    refresh: bool = False,
) -> str:
    """Fast local verdict engine with cached deterministic output."""
    cache_key = _analysis_verdict_cache_key(symbol, interval, analysis, forecast_pct)
    if not refresh:
        cached_verdict = ai_verdict_cache.get(cache_key)
        if cached_verdict:
            return str(cached_verdict)

        persisted_verdict = persistent_cache.get(cache_key)
        if persisted_verdict:
            verdict_text = str(persisted_verdict.get("verdict", "")).strip()
            if verdict_text:
                ai_verdict_cache.set(cache_key, verdict_text, ttl_seconds=verdict_ttl(interval))
                return verdict_text

    verdict_text = _derive_local_verdict(analysis, forecast_pct)
    ai_verdict_cache.set(cache_key, verdict_text, ttl_seconds=verdict_ttl(interval))
    persistent_cache.set(cache_key, {"verdict": verdict_text}, ttl=verdict_ttl(interval) * 4)
    return verdict_text


@app.get("/api/ticker/{symbol}")
async def get_ticker(symbol: str, interval: Optional[str] = None) -> Dict[str, Any]:
    symbol = _get_canonical_symbol(symbol)
    if symbol not in SYMBOLS:
        raise HTTPException(404, f"Unknown symbol: {symbol}")
    return await fetch_ticker(symbol, interval=interval)


# ── Watchlist (batch ticker) ──────────────────────────────────────────────────
@app.post("/api/watchlist/tickers")
async def get_watchlist_tickers(body: WatchlistRequest) -> Dict[str, Any]:
    results: Dict[str, Any] = {}
    valid_symbols, invalid_symbols, duplicate_count = _normalize_watchlist_symbols(body.symbols)
    if not valid_symbols:
        raise HTTPException(status_code=422, detail="No valid symbols provided")

    tasks = [
        _fetch_ticker_with_semaphore(s, _watchlist_ticker_semaphore)
        for s in valid_symbols
    ]
    tickers = await asyncio.gather(*tasks, return_exceptions=True)
    
    for sym, ticker in zip(valid_symbols, tickers):
        if isinstance(ticker, Exception):
            results[sym] = {"error": str(ticker)}
        else:
            results[sym] = ticker
    return {
        "tickers": results,
        "count": len(results),
        "invalid_symbols": invalid_symbols,
        "duplicate_count": duplicate_count,
    }


def _forecast_cache_key(symbol: str, interval: str, horizon: int) -> str:
    return f"forecast_{_cache_prefix(symbol, interval)}{horizon}"


def _forecast_payload_is_current(cached: Optional[Dict[str, Any]]) -> bool:
    if not isinstance(cached, dict):
        return False
    if "forecast_candles" in cached:
        return False

    display = cached.get("display") or {}
    if (
        display.get("mode") != "raw_timesfm_ohlc4_line"
        or display.get("output_mode") != "single_future_ohlc4_line"
        or display.get("channels") != ["ohlc4"]
    ):
        return False

    model_meta = cached.get("model") or {}
    semantics = model_meta.get("input_semantics") or {}
    output_semantics = model_meta.get("output_semantics") or {}
    return (
        semantics.get("volume_mode") == "omitted"
        and semantics.get("amount_mode") == "omitted"
        and semantics.get("active_forecast_channels") == ["ohlc4"]
        and semantics.get("feature_channels") == ["ohlc4"]
        and semantics.get("price_mode") == "ohlc4_single_channel"
        and semantics.get("base_signal") == "ohlc4"
        and semantics.get("adapter_mode") == "timesfm_native"
        and output_semantics.get("forecast_channel") == "ohlc4"
        and output_semantics.get("forecast_mode") == "single_future_ohlc4_line"
        and output_semantics.get("candle_projection") == "omitted"
    )


def _load_cached_forecast_response(cache_key: str, interval: str) -> Optional[Dict[str, Any]]:
    cached = forecast_cache.get(cache_key)
    if cached is not None and _forecast_payload_is_current(cached):
        cached["generated_at"] = int(time.time())
        cached["cache"] = {"origin": "memory", "refresh_requested": False}
        return cached

    persisted = persistent_cache.get(cache_key)
    if persisted is not None and _forecast_payload_is_current(persisted):
        persisted["from_persistent_cache"] = True
        forecast_cache.set(cache_key, persisted, ttl_seconds=forecast_ttl(interval))
        persisted["generated_at"] = int(time.time())
        persisted["cache"] = {"origin": "persistent", "refresh_requested": False}
        return persisted

    return None


async def _prepare_forecast_response_payload(
    symbol: str,
    interval: str,
    horizon: int,
    refresh: bool,
    cache_origin: str,
) -> Dict[str, Any]:
    data_list, source, indicators = await get_indicators_cached(
        symbol,
        interval,
        FORECAST_CONTEXT,
        refresh=refresh,
        min_context=FORECAST_CONTEXT,
    )
    if not TIMESFM_AVAILABLE:
        last_ohlc4 = (
            float(
                np.mean(
                    [
                        float(data_list[-1]["open"]),
                        float(data_list[-1]["high"]),
                        float(data_list[-1]["low"]),
                        float(data_list[-1]["close"]),
                    ]
                )
            )
            if data_list
            else 0.0
        )
        return {
            "symbol": symbol,
            "interval": interval,
            "forecast_rows": [],
            "error": "TimesFM not installed. Run: pip install timesfm[torch]",
            "ai_runtime": {"mode": "local_only", "model": "offline"},
            "_data_list": data_list,
            "indicators_snapshot": indicators,
            "_blended": {"confidence": 0.0, "agreement": False, "scale": 1.0, "model_weight": 0.0, "anchor_weight": 1.0, "model_bias_pct": 0.0},
            "source": source,
            "horizon": horizon,
            "last_close": last_ohlc4,
            "model": {
                "name": "offline",
                "context_length": 0,
                "quantiles": [0.1, 0.5, 0.9],
                "cache_version": CACHE_VERSION,
                "sample_count": 0,
                "input_semantics": {
                    "feature_channels": ["ohlc4"],
                    "active_forecast_channels": ["ohlc4"],
                    "ignored_channels": ["volume"],
                    "price_mode": "ohlc4_single_channel",
                    "base_signal": "ohlc4",
                    "volume_mode": "omitted",
                    "amount_mode": "omitted",
                    "adapter_mode": "timesfm_native",
                },
                "output_semantics": {
                    "forecast_channel": "ohlc4",
                    "forecast_mode": "single_future_ohlc4_line",
                    "quantile_fields": ["p10", "p50", "p90"],
                    "candle_projection": "omitted",
                },
            },
            "_cache_origin": cache_origin,
        }

    if len(data_list) < 40:
        raise HTTPException(422, "Insufficient historical data for forecasting")

    df_hist = pd.DataFrame(data_list)
    df_hist["timestamps"] = pd.to_datetime(df_hist["time"], unit="s", utc=True)

    if "amount" not in df_hist.columns or df_hist["amount"].isna().all() or df_hist["amount"].sum() == 0:
        typical = (df_hist["high"] + df_hist["low"] + df_hist["close"]) / 3
        df_hist["amount"] = (df_hist["volume"] * typical).fillna(0)
    else:
        df_hist["amount"] = df_hist["amount"].fillna(0)

    # TimesFM handles context internally; pass the full available history
    context_len = min(len(df_hist), TimesFMForecaster.MAX_CONTEXT)
    df_context = df_hist.tail(context_len).reset_index(drop=True)

    logger.info("[forecast] %s %s | ctx=%d/%d | horizon=%d", symbol, interval, context_len, len(df_hist), horizon)

    last_time = int(df_hist["time"].iloc[-1])
    step = STEP_SECONDS[interval]

    model_output = await forecaster.forecast(
        df=df_context[["open", "high", "low", "close", "volume"]],
        horizon=horizon,
    )

    last_ohlc4 = float(df_hist[["open", "high", "low", "close"]].mean(axis=1).iloc[-1])
    last_close = last_ohlc4
    analysis_bundle = _build_raw_close_bundle(model_output, last_ohlc4)
    logger.info(
        "[forecast] raw-ohlc4-line | %s %s | confidence=%.1f agreement=%s",
        symbol,
        interval,
        analysis_bundle["confidence"],
        analysis_bundle["agreement"],
    )

    forecast_rows: List[Dict[str, Any]] = [
        {"time": last_time, "p10": last_ohlc4, "p50": last_ohlc4, "p90": last_ohlc4, "is_actual": True}
    ]
    for i in range(horizon):
        forecast_rows.append({
            "time": int(last_time + step * (i + 1)),
            "p10": round(float(analysis_bundle["p10"][i]), 6),
            "p50": round(float(analysis_bundle["p50"][i]), 6),
            "p90": round(float(analysis_bundle["p90"][i]), 6),
        })

    return {
        "symbol": symbol,
        "interval": interval,
        "source": source,
        "horizon": horizon,
        "last_close": last_close,
        "forecast_rows": forecast_rows,
        "from_persistent_cache": False,
        "model": {
            "name": model_output.get("model_name", TimesFMForecaster.MODEL_HF_ID),
            "context_length": int(model_output.get("context_length", context_len)),
            "quantiles": [0.1, 0.5, 0.9],
            "cache_version": CACHE_VERSION,
            "sample_count": 0,  # TimesFM is deterministic; no sampling needed
            "input_semantics": model_output.get("input_semantics", {}),
            "output_semantics": model_output.get("output_semantics", {}),
        },
        "indicators_snapshot": indicators,
        "_data_list": data_list,
        "_analysis_bundle": analysis_bundle,
        "_model_output": model_output,
        "_cache_origin": cache_origin,
        "ai_runtime": {
            "mode": "local_only",
            "model": str(model_output.get("model_name", TimesFMForecaster.MODEL_HF_ID)),
            "device": forecaster.device,
        },
    }


def _build_raw_close_bundle(
    model_output: Dict[str, Any],
    last_close: float,
) -> Dict[str, Any]:
    """Build the close-path bundle directly from raw Kronos output."""
    raw_p10 = np.array(model_output["p10"], dtype=float)
    raw_p50 = np.array(model_output["p50"], dtype=float)
    raw_p90 = np.array(model_output["p90"], dtype=float)

    path_metrics = _forecast_path_metrics(raw_p50, last_close)
    avg_band_pct = float(
        np.mean((raw_p90 - raw_p10) / np.maximum(np.abs(raw_p50), 1e-8)) * 100.0
    ) if len(raw_p50) else 0.0
    band_certainty = math.exp(-avg_band_pct / 4.0)
    path_consistency = path_metrics["path_consistency"] / 100.0
    monotonicity = path_metrics["monotonicity"] / 100.0
    move_pct = abs(path_metrics["final_return_pct"])

    confidence = (
        28.0
        + band_certainty * 34.0
        + path_consistency * 18.0
        + monotonicity * 12.0
        + min(move_pct, 4.0) * 2.0
    )
    confidence = max(20.0, min(95.0, confidence))

    final_sign = 0 if abs(path_metrics["final_return_pct"]) < 0.05 else (1 if path_metrics["final_return_pct"] > 0 else -1)
    weighted_sign = 0 if abs(path_metrics["weighted_return_pct"]) < 0.05 else (1 if path_metrics["weighted_return_pct"] > 0 else -1)
    agreement = final_sign == 0 or weighted_sign == 0 or final_sign == weighted_sign

    return {
        "p10": raw_p10,
        "p50": raw_p50,
        "p90": raw_p90,
        "model_weight": 1.0,
        "anchor_weight": 0.0,
        "agreement": agreement,
        "scale": 1.0,
        "confidence": round(confidence, 2),
        "model_bias_pct": 0.0,
        "path_metrics": path_metrics,
        "mode": "raw_timesfm_ohlc4",
    }


async def _finalize_forecast_response_payload(payload: Dict[str, Any]) -> Dict[str, Any]:
    if payload.get("error"):
        response = {
            "symbol": payload["symbol"],
            "interval": payload["interval"],
            "forecast": payload.get("forecast_rows", []),
            "error": payload["error"],
            "path_checked": None,
            "display": {
                "mode": "raw_timesfm_ohlc4_line",
                "channels": ["ohlc4"],
                "output_mode": "single_future_ohlc4_line",
                "uncertainty_source": "timesfm_quantile_head",
                "uses_anchor_blending": False,
            },
            "ai_runtime": payload.get("ai_runtime", {"mode": "local_only", "model": "offline"}),
        }
        return make_json_compatible(response)

    analysis_bundle = payload["_analysis_bundle"]
    analysis = await asyncio.to_thread(
        _build_trade_analysis,
        symbol=payload["symbol"],
        interval=payload["interval"],
        data=payload["_data_list"],
        indicators=payload["indicators_snapshot"],
        forecast_rows=payload["forecast_rows"],
        confidence=float(analysis_bundle.get("confidence", 50.0)),
        source=payload["source"],
        blended=analysis_bundle,
    )

    response = {
        "symbol": payload["symbol"],
        "interval": payload["interval"],
        "source": payload["source"],
        "horizon": payload["horizon"],
        "last_close": payload["last_close"],
        "forecast": payload["forecast_rows"],
        "from_persistent_cache": payload.get("from_persistent_cache", False),
        "model": payload["model"],
        "display": {
            "mode": "raw_timesfm_ohlc4_line",
            "channels": ["ohlc4"],
            "output_mode": "single_future_ohlc4_line",
            "uncertainty_source": "timesfm_quantile_head",
            "uses_anchor_blending": False,
        },
        "ensemble": {
            "mode": "raw_timesfm_ohlc4",
            "model_weight": analysis_bundle["model_weight"],
            "anchor_weight": analysis_bundle["anchor_weight"],
            "trend_agreement": analysis_bundle["agreement"],
            "confidence": analysis_bundle["confidence"],
            "model_bias_pct": analysis_bundle["model_bias_pct"],
            "alignment_scale": analysis_bundle["scale"],
            "used_for_display": True,
        },
        "model_diagnostics": {
            "raw_last_p10": round(float(payload["_model_output"]["p10"][-1]), 6),
            "raw_last_p50": round(float(payload["_model_output"]["p50"][-1]), 6),
            "raw_last_p90": round(float(payload["_model_output"]["p90"][-1]), 6),
            "display_last_p50": round(float(analysis_bundle["p50"][-1]), 6),
        },
        "indicators_snapshot": payload["indicators_snapshot"],
        "analysis": analysis,
        "generated_at": int(time.time()),
        "cache": {
            "origin": payload["_cache_origin"],
            "refresh_requested": False,
        },
        "ai_runtime": payload["ai_runtime"],
    }

    return make_json_compatible(response)


# ── Forecast ──────────────────────────────────────────────────────────────────
@app.get("/api/forecast/{symbol}")
async def get_forecast(
    symbol:   str,
    interval: str = Query("1h"),
    horizon:  int = Query(10, ge=5, le=300),
    refresh:  bool = Query(False)
) -> 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}")

    cache_key = _forecast_cache_key(symbol, interval, horizon)
    cache_origin = "live"

    if not refresh:
        cached = _load_cached_forecast_response(cache_key, interval)
        if cached is not None:
            return cached
    else:
        logger.info("[forecast] Refresh requested for %s %s. Bypassing caches.", symbol, interval)
        cache_origin = "live_refresh"

    inflight_key = f"{cache_key}:refresh={int(refresh)}"
    inflight_task = None if refresh else _FORECAST_INFLIGHT.get(inflight_key)
    if inflight_task is not None:
        return await inflight_task

    async def _build_forecast_response() -> Dict[str, Any]:
        payload = await _prepare_forecast_response_payload(
            symbol=symbol,
            interval=interval,
            horizon=horizon,
            refresh=refresh,
            cache_origin=cache_origin,
        )
        response = await _finalize_forecast_response_payload(payload)
        forecast_cache.set(cache_key, response, ttl_seconds=forecast_ttl(interval))
        persistent_cache.set(cache_key, response, ttl=forecast_ttl(interval) * 4)
        return response

    task = asyncio.create_task(_build_forecast_response(), name=f"forecast:{symbol}:{interval}:{horizon}")
    _FORECAST_INFLIGHT[inflight_key] = task
    try:
        return await task
    finally:
        if _FORECAST_INFLIGHT.get(inflight_key) is task:
            _FORECAST_INFLIGHT.pop(inflight_key, None)



# ── Market Status ─────────────────────────────────────────────────────────────
@app.get("/api/market-status")
async def get_market_status() -> Dict[str, Any]:
    return {
        "utc_time": datetime.now(timezone.utc).isoformat(),
        "markets":  market_status_now(),
    }


# ── Switch endpoint ────────────────────────────────────────────────────────────
@app.post("/api/switch")
async def switch_symbol_interval(body: SwitchRequest) -> Dict[str, Any]:
    symbol = body.symbol.upper()
    interval = body.interval
    if symbol not in SYMBOLS:
        raise HTTPException(404, f"Unknown symbol: {symbol}")
    if interval not in SUPPORTED_INTERVALS:
        raise HTTPException(400, f"Unsupported interval: {interval}")

    prefix = _cache_prefix(symbol, interval)
    h_cleared = historical_cache.delete_by_prefix(f"hist_{prefix}")
    f_cleared = forecast_cache.delete_by_prefix(f"forecast_{prefix}")
    i_cleared = indicators_cache.delete_by_prefix(f"ind_{prefix}")

    logger.info(
        "[switch] %s %s -> hist=%d forecast=%d indicators=%d",
        symbol,
        interval,
        h_cleared,
        f_cleared,
        i_cleared,
    )

    return {
        "status": "cleared",
        "symbol": symbol,
        "interval": interval,
        "cleared": {"historical": h_cleared, "forecast": f_cleared, "indicators": i_cleared},
        "message": "Cache cleared. Fetch fresh data now.",
    }


# ── Cache management ──────────────────────────────────────────────────────────
@app.delete("/api/cache/{symbol}/{interval}")
async def clear_symbol_cache(symbol: str, interval: str, request: Request) -> Dict[str, Any]:
    admin_only(request)
    symbol, interval = _validate_symbol_and_interval(symbol, interval)
    prefix    = _cache_prefix(symbol, interval)
    h_cleared = historical_cache.delete_by_prefix(f"hist_{prefix}")
    f_cleared = forecast_cache.delete_by_prefix(f"forecast_{prefix}")
    i_cleared = indicators_cache.delete_by_prefix(f"ind_{prefix}")
    v_cleared = ai_verdict_cache.delete_by_prefix(f"ai_verdict:{CACHE_VERSION}:{symbol}:{interval}:")
    return {"symbol": symbol, "interval": interval,
            "cleared": {"historical": h_cleared, "forecast": f_cleared, "indicators": i_cleared, "ai_verdict": v_cleared}}


@app.delete("/api/cache")
async def clear_all_cache(
    request: Request,
    target: str = Query("all", description="all | historical | forecast | indicators | ticker | ai_verdict"),
) -> Dict[str, Any]:
    admin_only(request)
    target = _validate_cache_target(target)
    h = historical_cache.clear() if target in ("all", "historical") else 0
    f = forecast_cache.clear()   if target in ("all", "forecast")   else 0
    i = indicators_cache.clear() if target in ("all", "indicators") else 0
    t = ticker_cache.clear()     if target in ("all", "ticker")     else 0
    v = ai_verdict_cache.clear() if target in ("all", "ai_verdict") else 0
    return {"cleared": {"historical": h, "forecast": f, "indicators": i, "ticker": t, "ai_verdict": v}}


@app.get("/api/cache/stats")
async def cache_stats(request: Request) -> Dict[str, Any]:
    admin_only(request)
    return {
        "cache_version": CACHE_VERSION,
        "historical":    historical_cache.stats(),
        "forecast":      forecast_cache.stats(),
        "indicators":    indicators_cache.stats(),
        "ticker":        ticker_cache.stats(),
        "ai_verdict":    ai_verdict_cache.stats(),
    }


# ── Health ────────────────────────────────────────────────────────────────────
@app.get("/api/health")
async def health_check() -> Dict[str, Any]:
    STARTUP_STATE["timesfm"]["loaded"] = forecaster.is_ready
    STARTUP_STATE["timesfm"]["device"] = forecaster.device
    return {
        "status":        "online" if TIMESFM_AVAILABLE else "degraded",
        "version":       APP_VERSION,
        "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],
        "cors_allow_origins": CORS_ALLOW_ORIGINS,
        "admin_auth_configured": ADMIN_TOKEN != DEFAULT_ADMIN_TOKEN,
        "request_metrics": REQUEST_METRICS.snapshot(),
        "timesfm":      STARTUP_STATE["timesfm"],
        "startup_checks": STARTUP_STATE["sources"],
    }


@app.get("/api/ping")
async def ping() -> Dict[str, str]:
    return {"status": "ok", "version": APP_VERSION}


# ─── EXTRA API ENDPOINTS (v5.0) ────────────────────────────────────────────────

@app.get("/api/ai/rules")
async def get_ai_rules() -> Dict[str, Any]:
    return ai_rule_registry.snapshot()


@app.get("/api/metrics")
async def get_metrics(request: Request):
    """Export Prometheus-ready metrics (latencies, cache hits, CB states)."""
    admin_only(request)
    now = time.time()
    uptime = now - START_TIME
    
    # Cache stats
    h_stats = historical_cache.stats()
    f_stats = forecast_cache.stats()
    t_stats = ticker_cache.stats()
    
    # CB stats
    cb_stats = {name: {"state": cb.state, "failures": cb.failures} 
                for name, cb in source_breakers.items()}
    
    return {
        "uptime_seconds": round(uptime, 0),
        "request_metrics": REQUEST_METRICS.snapshot(),
        "cache": {
            "historical": h_stats,
            "forecast":   f_stats,
            "ticker":     t_stats,
        },
        "circuit_breakers": cb_stats,
        "sources": STARTUP_STATE["sources"],
        "timesfm_status": STARTUP_STATE["timesfm"]["loaded"],
    }


@app.get("/api/backtest/{symbol}")
async def get_backtest(
    symbol: str,
    interval: str = Query("1h"),
    limit: int = Query(500, ge=100, le=5000),
):
    """E-1: POC Backtesting: Calculates returns if following AI signals."""
    symbol, interval = _validate_symbol_and_interval(symbol, interval)
    data, _ = await fetch_historical(symbol, interval, limit)
    if len(data) < 100:
        raise HTTPException(status_code=400, detail="Insufficient data for backtest")
    
    closes = np.array([float(d["close"]) for d in data])
    rsi = _rsi(closes, 14)
    
    trades, position, entry_price, total_profit = [], 0, 0.0, 0.0
    for i in range(50, len(closes)):
        if position == 0 and rsi[i] < 30:
            position, entry_price = 1, closes[i]
        elif position == 1 and rsi[i] > 70:
            p = (closes[i] - entry_price) / entry_price
            total_profit += p
            trades.append({"entry": entry_price, "exit": closes[i], "profit_pct": p * 100})
            position = 0
            
    return {
        "symbol": symbol,
        "total_trades": len(trades),
        "total_return_pct": round(total_profit * 100, 2),
        "trades": trades[-10:]
    }


@app.get("/api/volume-profile/{symbol}")
async def get_volume_profile(
    symbol: str,
    interval: str = Query("1h"),
    limit: int = Query(200, ge=20, le=5000),
    buckets: int = Query(20, ge=5, le=200),
):
    """E-4: POC Volume Profile."""
    symbol, interval = _validate_symbol_and_interval(symbol, interval)
    data, _ = await fetch_historical(symbol, interval, limit)
    if not data:
        return {"buckets": [], "poc": None}
    
    highs, lows, vols = np.array([float(d["high"]) for d in data]), np.array([float(d["low"]) for d in data]), np.array([float(d["volume"]) for d in data])
    
    p_min, p_max = np.min(lows), np.max(highs)
    if p_min == p_max:
        return {"buckets": [], "poc": None}
    
    bin_size, profile = (p_max - p_min) / buckets, defaultdict(float)
    for i in range(len(data)):
        b_start, b_end = int((lows[i] - p_min) / bin_size), int((highs[i] - p_min) / bin_size)
        num_bins = max(1, b_end - b_start + 1)
        v_per_bin = vols[i] / num_bins
        for b in range(b_start, min(b_end + 1, buckets)):
            profile[b] += v_per_bin
            
    sorted_profile = [{"price": round(p_min + (b * bin_size) + (bin_size / 2), 6), "volume": round(profile[b], 2)} for b in range(buckets)]
        
    return {"buckets": sorted_profile, "poc": max(sorted_profile, key=lambda x: x["volume"])["price"]}


# ── Static Frontend ───────────────────────────────────────────────────────────
FRONTEND_PATH = os.path.join(PROJECT_ROOT, "frontend")
if os.path.exists(FRONTEND_PATH):
    INDEX_PATH = os.path.join(FRONTEND_PATH, "index.html")
    AIBG_PATH = os.path.join(FRONTEND_PATH, "AIBG.png")
    FAVICON_PATH = os.path.join(FRONTEND_PATH, "favicon.svg")
    WORKSPACE_JS_PATH = os.path.join(FRONTEND_PATH, "workspace.js")
    WORKSPACE_CSS_PATH = os.path.join(FRONTEND_PATH, "workspace.css")

    def _frontend_asset_headers() -> Dict[str, str]:
        return {
            "Cache-Control": "no-store, no-cache, must-revalidate, max-age=0",
            "Pragma": "no-cache",
            "Expires": "0",
        }

    def _frontend_asset_version() -> str:
        asset_paths = [INDEX_PATH, WORKSPACE_JS_PATH, WORKSPACE_CSS_PATH]
        version_parts: List[str] = [APP_VERSION, CACHE_VERSION]
        for asset_path in asset_paths:
            if os.path.exists(asset_path):
                version_parts.append(str(int(os.path.getmtime(asset_path))))
        return "-".join(version_parts)

    @app.get("/", include_in_schema=False)
    @app.get("/index.html", include_in_schema=False)
    async def serve_frontend_index() -> HTMLResponse:
        if not os.path.exists(INDEX_PATH):
            raise HTTPException(status_code=404, detail="Frontend index not found")

        html = Path(INDEX_PATH).read_text(encoding="utf-8")
        html = html.replace("__FRONTEND_ASSET_VERSION__", _frontend_asset_version())
        headers = _frontend_asset_headers()
        return HTMLResponse(content=html, headers=headers)

    @app.get("/workspace.js", include_in_schema=False)
    async def serve_workspace_js() -> FileResponse:
        if not os.path.exists(WORKSPACE_JS_PATH):
            raise HTTPException(status_code=404, detail="Workspace JS asset not found")

        return FileResponse(
            WORKSPACE_JS_PATH,
            media_type="application/javascript",
            headers=_frontend_asset_headers(),
        )

    @app.get("/workspace.css", include_in_schema=False)
    async def serve_workspace_css() -> FileResponse:
        if not os.path.exists(WORKSPACE_CSS_PATH):
            raise HTTPException(status_code=404, detail="Workspace CSS asset not found")

        return FileResponse(
            WORKSPACE_CSS_PATH,
            media_type="text/css",
            headers=_frontend_asset_headers(),
        )

    @app.get("/AIBG.png", include_in_schema=False)
    async def serve_aibg() -> FileResponse:
        if not os.path.exists(AIBG_PATH):
            raise HTTPException(status_code=404, detail="AIBG asset not found")

        return FileResponse(
            AIBG_PATH,
            media_type="image/png",
            headers=_frontend_asset_headers(),
        )

    @app.get("/favicon.svg", include_in_schema=False)
    @app.get("/favicon.ico", include_in_schema=False)
    async def serve_favicon() -> FileResponse:
        if not os.path.exists(FAVICON_PATH):
            raise HTTPException(status_code=404, detail="Favicon not found")

        return FileResponse(
            FAVICON_PATH,
            media_type="image/svg+xml",
            headers=_frontend_asset_headers(),
        )

    app.mount("/", StaticFiles(directory=FRONTEND_PATH, html=True), name="frontend")
    logger.info("Mounted frontend: %s", FRONTEND_PATH)
else:
    logger.warning("Frontend path not found: %s", FRONTEND_PATH)