import os
import asyncio
import json
import logging
from contextlib import suppress
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from pathlib import Path
from typing import Dict, List, Optional, Tuple, Union

import ccxt
import ccxt.async_support as ccxt_async
import pandas as pd
import websockets
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query, HTTPException, status
from fastapi.responses import JSONResponse
from fastapi.middleware.cors import CORSMiddleware

try:
    from technical.util import resample_to_interval  # type: ignore
except ImportError:  # pragma: no cover - 环境缺失依赖时自动降级
    resample_to_interval = None  # type: ignore

from .storage import (
    ensure_storage,
    read_candles,
    upsert_candles,
    get_last_timestamp,
    read_candle_exact,
)


LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO").upper()
logging.basicConfig(
    level=LOG_LEVEL,
    format="%(asctime)s %(levelname)s [%(name)s] %(message)s",
)
logger = logging.getLogger("datasvc")
BASE_DIR = Path(__file__).resolve().parent.parent
DEFAULT_PAIRS_FILE = BASE_DIR / "pairs.json"
RESAMPLE_AVAILABLE = resample_to_interval is not None
RESAMPLE_WARNING_EMITTED = False

REST_MAX_CONCURRENCY = int(os.environ.get("REST_MAX_CONCURRENCY", "4"))
VERIFY_MAX_CONCURRENCY = int(os.environ.get("VERIFY_MAX_CONCURRENCY", "2"))
REST_FETCH_SEMAPHORE = asyncio.Semaphore(max(1, REST_MAX_CONCURRENCY))
VERIFY_FETCH_SEMAPHORE = asyncio.Semaphore(max(1, VERIFY_MAX_CONCURRENCY))

AGGREGATION_PLAN: Dict[str, List[str]] = {
    "1m": ["2m", "3m", "4m", "5m", "10m", "15m", "20m", "25m", "30m"],
    "1h": ["2h", "3h", "4h", "6h", "8h", "12h", "16h"],
    "1d": ["2d", "3d", "4d", "5d", "6d"],
    "1w": ["2w"],
    "1M": ["2M", "3M", "6M"],
}

CandleRow = List[Union[int, float]]


def _split_env_list(value: str) -> List[str]:
    return [item.strip() for item in value.split(",") if item.strip()]


def _unique_preserve(values: List[str]) -> List[str]:
    seen = set()
    ordered: List[str] = []
    for item in values:
        if item not in seen:
            ordered.append(item)
            seen.add(item)
    return ordered


def _load_symbols() -> List[str]:
    path = DEFAULT_PAIRS_FILE
    if not path.is_file():
        default_symbols = ["BTC/USDT:USDT"]
        logger.warning("交易对配置文件不存在，使用默认值", extra={"file": str(path), "symbols": default_symbols})
        return default_symbols
    try:
        content = path.read_text(encoding="utf-8")
        data = json.loads(content)
    except Exception:
        default_symbols = ["BTC/USDT:USDT"]
        logger.exception("读取交易对配置文件失败，使用默认值", extra={"file": str(path), "symbols": default_symbols})
        return default_symbols

    raw_symbols: List[str] = []
    if isinstance(data, list):
        raw_symbols = [str(item).strip() for item in data if isinstance(item, str) and item.strip()]
    elif isinstance(data, dict):
        candidates = data.get("symbols") or data.get("pairs")
        if isinstance(candidates, list):
            raw_symbols = [str(item).strip() for item in candidates if isinstance(item, str) and item.strip()]
    if not raw_symbols:
        default_symbols = ["BTC/USDT:USDT"]
        logger.warning("交易对配置文件未提供有效列表，使用默认值", extra={"file": str(path), "symbols": default_symbols})
        return default_symbols

    symbols = _unique_preserve(raw_symbols)
    logger.info("已从配置文件载入交易对", extra={"file": str(path), "symbols": symbols})
    return symbols


def timeframe_to_minutes(tf: str) -> Optional[int]:
    if not tf:
        return None
    unit = tf[-1]
    try:
        value = int(tf[:-1])
    except ValueError:
        return None
    multiplier = {
        "m": 1,
        "h": 60,
        "d": 1440,
        "w": 10080,
        "M": 43200,  # 30 天近似
    }.get(unit)
    if multiplier is None:
        return None
    return value * multiplier


def binance_stream_symbol(symbol: str) -> str:
    try:
        base, rest = symbol.split("/", 1)
    except ValueError:
        cleaned = symbol.replace("/", "").split(":")[0]
        return cleaned.lower()
    quote = rest.split(":")[0]
    return f"{base}{quote}".lower()


def build_stream_url(symbol: str, timeframe: str) -> str:
    stream_symbol = binance_stream_symbol(symbol)
    return f"{BINANCE_WS_BASE}/{stream_symbol}@kline_{timeframe}"

DATA_DIR = os.environ.get("DATA_DIR", "/data")
EXCHANGE = os.environ.get("EXCHANGE", "binance")
SYMBOLS = _load_symbols()

_default_timeframes = ["1m", "1h", "1d", "1w", "1M"]
requested_timeframes = _split_env_list(os.environ.get("TIMEFRAMES", ",".join(_default_timeframes)))
if not requested_timeframes:
    requested_timeframes = _default_timeframes

FETCH_TIMEFRAMES = _unique_preserve(requested_timeframes)
AVAILABLE_TIMEFRAMES = list(FETCH_TIMEFRAMES)
for base_tf in FETCH_TIMEFRAMES:
    for derived_tf in AGGREGATION_PLAN.get(base_tf, []):
        if derived_tf not in AVAILABLE_TIMEFRAMES:
            AVAILABLE_TIMEFRAMES.append(derived_tf)
DERIVED_TIMEFRAMES = [tf for tf in AVAILABLE_TIMEFRAMES if tf not in FETCH_TIMEFRAMES]
AGGREGATION_TARGETS = {tf: AGGREGATION_PLAN.get(tf, []) for tf in FETCH_TIMEFRAMES}

START_FROM = os.environ.get("START_FROM", "2022-01-01")  # 首次启动拉取起始日期（UTC）
POLL_FACTOR = float(os.environ.get("POLL_FACTOR", "0.5"))  # 轮询间隔 = tf_ms * factor
BACKOFF_BASE = float(os.environ.get("BACKOFF_BASE", "2.0"))
BACKOFF_MAX = float(os.environ.get("BACKOFF_MAX", "30.0"))
BINANCE_WS_BASE = os.environ.get("BINANCE_WS_BASE", "wss://fstream.binance.com/ws").rstrip("/")

VALID_SYMBOLS = set(SYMBOLS)
VALID_TIMEFRAMES = set(AVAILABLE_TIMEFRAMES)

ensure_storage(DATA_DIR)

app = FastAPI(title="Local Data Service", version="0.1.0")
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


def tf_to_ms(tf: str) -> int:
    minutes = timeframe_to_minutes(tf)
    if minutes is None:
        logger.warning("无法解析时间周期，默认使用 60 秒", extra={"timeframe": tf})
        return 60_000
    return minutes * 60_000


def ensure_symbol_timeframe(symbol: str, timeframe: str) -> None:
    if symbol not in VALID_SYMBOLS:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"symbol 必须为 {sorted(VALID_SYMBOLS)} 之一。",
        )
    if timeframe not in VALID_TIMEFRAMES:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"tf 必须为 {sorted(VALID_TIMEFRAMES)} 之一。",
        )


def parse_start_from_ms(val: str) -> int:
    """将 START_FROM 解析成毫秒级时间戳。
    支持两种格式：
    - YYYY-MM-DD（UTC 00:00:00）
    - 整型毫秒时间戳字符串
    """
    try:
        return int(val)
    except Exception:
        pass
    try:
        dt = datetime.fromisoformat(val)  # 允许 '2022-01-01' 或 '2022-01-01T00:00:00'
    except Exception:
        # 回退到固定日期
        dt = datetime(2022, 1, 1)
    return int(dt.timestamp() * 1000)


class Hub:
    def __init__(self) -> None:
        self.subscribers: Dict[str, List[WebSocket]] = {}

    def topic(self, symbol: str, timeframe: str) -> str:
        return f"candles::{symbol}::{timeframe}"

    async def subscribe(self, ws: WebSocket, symbol: str, timeframe: str):
        topic = self.topic(symbol, timeframe)
        await ws.accept()
        self.subscribers.setdefault(topic, []).append(ws)

    def _clean(self, topic: str):
        conns = self.subscribers.get(topic, [])
        self.subscribers[topic] = [w for w in conns if not w.client_state.name == "DISCONNECTED"]

    async def publish(self, symbol: str, timeframe: str, payload: dict):
        topic = self.topic(symbol, timeframe)
        conns = self.subscribers.get(topic, [])
        if not conns:
            return
        message = json.dumps(payload, ensure_ascii=False)
        dead: List[WebSocket] = []
        for ws in conns:
            try:
                await ws.send_text(message)
            except Exception:
                dead.append(ws)
        if dead:
            self.subscribers[topic] = [w for w in conns if w not in dead]


hub = Hub()

fetch_tasks: List[asyncio.Task] = []
verification_queues: Dict[Tuple[str, str], asyncio.Queue[int]] = {}


def resample_and_store(symbol: str, base_timeframe: str, derived_timeframes: List[str]) -> List[Tuple[str, List[CandleRow]]]:
    if not RESAMPLE_AVAILABLE or not derived_timeframes:
        return []
    base_df = read_candles(DATA_DIR, symbol, base_timeframe, None, None)
    if base_df.empty:
        return []
    base_df = base_df.copy()
    if "date" not in base_df.columns:
        base_df["date"] = pd.to_datetime(base_df["timestamp"], unit="ms", utc=True)
    base_df = base_df.sort_values("timestamp")

    updates: List[Tuple[str, List[List[float]]]] = []
    for target_tf in derived_timeframes:
        minutes = timeframe_to_minutes(target_tf)
        if minutes is None:
            logger.warning("无法解析聚合周期", extra={"target_timeframe": target_tf})
            continue
        try:
            derived_df = resample_to_interval(base_df, minutes)  # type: ignore[misc]
        except Exception:
            logger.exception(
                "聚合周期计算失败",
                extra={"symbol": symbol, "base_timeframe": base_timeframe, "target_timeframe": target_tf},
            )
            continue
        if derived_df is None or derived_df.empty:
            continue
        derived_df = derived_df.copy()
        if "timestamp" not in derived_df.columns:
            if "date" in derived_df.columns:
                dates = pd.to_datetime(derived_df["date"], utc=True, errors="coerce")
                derived_df["timestamp"] = (dates.astype("int64") // 1_000_000)
            elif isinstance(derived_df.index, pd.DatetimeIndex):
                idx = derived_df.index
                if idx.tz is None:
                    idx = idx.tz_localize("UTC")
                else:
                    idx = idx.tz_convert("UTC")
                derived_df["timestamp"] = (idx.astype("int64") // 1_000_000)
        if "timestamp" not in derived_df.columns:
            logger.warning(
                "聚合结果缺少 timestamp 列，已跳过",
                extra={"target_timeframe": target_tf},
            )
            continue
        derived_df = derived_df.dropna(subset=["timestamp", "open", "high", "low", "close", "volume"])
        if derived_df.empty:
            continue
        derived_df["timestamp"] = derived_df["timestamp"].astype("int64")
        derived_df = derived_df.sort_values("timestamp")
        last_ts = get_last_timestamp(DATA_DIR, symbol, target_tf)
        if last_ts is not None:
            derived_df = derived_df[derived_df["timestamp"] > last_ts]
        if derived_df.empty:
            continue
        numpy_rows = derived_df[["timestamp", "open", "high", "low", "close", "volume"]].to_numpy()
        records: List[CandleRow] = []
        for ts, o, h, l, c, v in numpy_rows:
            records.append(
                [
                    int(ts),
                    float(o),
                    float(h),
                    float(l),
                    float(c),
                    float(v),
                ]
            )
        if not records:
            continue
        upsert_candles(DATA_DIR, symbol, target_tf, records)
        updates.append((target_tf, records[-3:] if len(records) > 3 else records))
    return updates


@dataclass
class FetchState:
    symbol: str
    timeframe: str
    started_at: datetime = field(default_factory=datetime.utcnow)
    last_fetch_at: Optional[datetime] = None
    last_candle_ts: Optional[int] = None
    consecutive_errors: int = 0
    last_error: Optional[str] = None
    last_verified_ts: Optional[int] = None

    def to_payload(self) -> dict:
        def serialize_dt(dt: Optional[datetime]) -> Optional[str]:
            if not dt:
                return None
            return dt.replace(microsecond=0).isoformat() + "Z"

        return {
            "symbol": self.symbol,
            "timeframe": self.timeframe,
            "started_at": serialize_dt(self.started_at),
            "last_fetch_at": serialize_dt(self.last_fetch_at),
            "last_candle_ts": self.last_candle_ts,
            "consecutive_errors": self.consecutive_errors,
            "last_error": self.last_error,
            "last_verified_ts": self.last_verified_ts,
        }


fetch_states: Dict[Tuple[str, str], FetchState] = {}


async def process_candles(
    symbol: str,
    timeframe: str,
    candles: List[CandleRow],
    derived_timeframes: List[str],
    tf_ms: int,
    schedule_verification: bool,
) -> None:
    if not candles:
        return
    state_key = (symbol, timeframe)
    upsert_candles(DATA_DIR, symbol, timeframe, candles)
    derived_updates: List[Tuple[str, List[CandleRow]]] = []
    if derived_timeframes and RESAMPLE_AVAILABLE:
        derived_updates = await asyncio.to_thread(
            resample_and_store,
            symbol,
            timeframe,
            derived_timeframes,
        )
    for row in candles[-3:]:
        payload = {
            "topic": f"candles.{symbol}.{timeframe}",
            "type": "upsert",
            "data": {
                "t": row[0],
                "o": row[1],
                "h": row[2],
                "l": row[3],
                "c": row[4],
                "v": row[5],
            },
        }
        await hub.publish(symbol, timeframe, payload)
    for target_tf, rows in derived_updates:
        if not rows:
            continue
        for row in rows:
            ts = int(row[0])
            o, h, l, c, v = map(float, row[1:])
            payload = {
                "topic": f"candles.{symbol}.{target_tf}",
                "type": "upsert",
                "data": {
                    "t": ts,
                    "o": o,
                    "h": h,
                    "l": l,
                    "c": c,
                    "v": v,
                },
            }
            await hub.publish(symbol, target_tf, payload)
    state = fetch_states.get(state_key)
    if state:
        state.last_fetch_at = datetime.utcnow()
        state.last_candle_ts = candles[-1][0]
        state.consecutive_errors = 0
        state.last_error = None
        if schedule_verification:
            queue = verification_queues.get(state_key)
            if queue:
                now_ms = int(datetime.utcnow().timestamp() * 1000)
                latest_ts = candles[-1][0]
                if now_ms - latest_ts <= 2 * tf_ms:
                    verify_ts = latest_ts - tf_ms
                    if verify_ts > 0 and (state.last_verified_ts is None or verify_ts > state.last_verified_ts):
                        try:
                            queue.put_nowait(verify_ts)
                        except asyncio.QueueFull:
                            logger.warning(
                                "验证队列已满，丢弃此次校验请求",
                                extra={"symbol": symbol, "timeframe": timeframe, "timestamp": verify_ts},
                            )


async def rest_catchup(
    symbol: str,
    timeframe: str,
    derived_timeframes: List[str],
    tf_ms: int,
    start_since: int,
) -> None:
    state_key = (symbol, timeframe)
    state = fetch_states[state_key]
    exchange = build_exchange()
    since = start_since
    backoff = 1.0
    logger.info("开始 REST 补齐历史", extra={"symbol": symbol, "timeframe": timeframe, "since": since})
    try:
        while True:
            now_ms = int(datetime.utcnow().timestamp() * 1000)
            if since >= now_ms - tf_ms:
                break
            try:
                async with REST_FETCH_SEMAPHORE:
                    candles = await exchange.fetch_ohlcv(symbol, timeframe, since=since, limit=1000)
            except asyncio.CancelledError:
                raise
            except (ccxt.NetworkError, ccxt.ExchangeNotAvailable, ccxt.RequestTimeout) as exc:
                logger.warning(
                    "历史补齐网络异常，准备重试",
                    extra={"symbol": symbol, "timeframe": timeframe, "error": str(exc)},
                )
                state.last_error = str(exc)
                state.consecutive_errors += 1
                backoff = min(backoff * BACKOFF_BASE, BACKOFF_MAX)
                await asyncio.sleep(backoff)
                continue
            except Exception as exc:
                logger.exception(
                    "历史补齐发生异常，准备重试",
                    extra={"symbol": symbol, "timeframe": timeframe},
                )
                state.last_error = str(exc)
                state.consecutive_errors += 1
                backoff = min(backoff * BACKOFF_BASE, BACKOFF_MAX)
                await asyncio.sleep(backoff)
                continue
            if not candles:
                break
            await process_candles(symbol, timeframe, candles, derived_timeframes, tf_ms, schedule_verification=False)
            since = candles[-1][0] + tf_ms
            state.consecutive_errors = 0
            state.last_error = None
            backoff = 1.0

            now_ms = int(datetime.utcnow().timestamp() * 1000)
            lag = now_ms - since
            if lag > tf_ms * 10:
                await asyncio.sleep(0.2)
            else:
                await asyncio.sleep(max(1.0, tf_ms * POLL_FACTOR / 1000.0))
    finally:
        with suppress(Exception):
            await exchange.close()
    logger.info("REST 补齐完成", extra={"symbol": symbol, "timeframe": timeframe, "latest": state.last_candle_ts})


async def stream_loop(symbol: str, timeframe: str, derived_timeframes: List[str], tf_ms: int):
    state_key = (symbol, timeframe)
    url = build_stream_url(symbol, timeframe)
    while True:
        try:
            async with websockets.connect(url, ping_interval=20, ping_timeout=20) as ws:
                logger.info("WebSocket 已连接", extra={"symbol": symbol, "timeframe": timeframe, "url": url})
                async for message in ws:
                    data = json.loads(message)
                    kline = data.get("k")
                    if not kline or not kline.get("x"):
                        continue
                    row: CandleRow = [
                        int(kline["t"]),
                        float(kline["o"]),
                        float(kline["h"]),
                        float(kline["l"]),
                        float(kline["c"]),
                        float(kline["v"]),
                    ]
                    await process_candles(symbol, timeframe, [row], derived_timeframes, tf_ms, schedule_verification=True)
        except asyncio.CancelledError:
            logger.info("取消 WebSocket 任务", extra={"symbol": symbol, "timeframe": timeframe})
            raise
        except Exception as exc:
            logger.warning(
                "WebSocket 连接异常，准备重连",
                extra={"symbol": symbol, "timeframe": timeframe, "error": str(exc)},
            )
            state = fetch_states.get(state_key)
            start_since = None
            if state and state.last_candle_ts:
                start_since = state.last_candle_ts + tf_ms
            if start_since:
                await rest_catchup(symbol, timeframe, derived_timeframes, tf_ms, start_since)
            await asyncio.sleep(5.0)


def build_exchange():
    if EXCHANGE.lower() == "binance":
        return ccxt_async.binance(
            {
                "enableRateLimit": True,
                "timeout": 20_000,
                "options": {
                    "adjustForTimeDifference": True,
                    "defaultType": "future",
                    "defaultSubType": "linear",
                    "defaultMarket": "future",
                    "defaultSettle": "USDT",
                },
            }
        )
    raise RuntimeError(f"Unsupported EXCHANGE: {EXCHANGE}")


async def fetch_loop(symbol: str, timeframe: str):
    """初次通过 REST 补齐历史，随后转入 Binance WebSocket 拉取增量。"""
    derived_timeframes = AGGREGATION_TARGETS.get(timeframe, [])
    global RESAMPLE_WARNING_EMITTED
    if derived_timeframes and not RESAMPLE_AVAILABLE and not RESAMPLE_WARNING_EMITTED:
        logger.warning(
            "缺少 technical.util.resample_to_interval 模块，聚合时间周期生成已跳过",
            extra={"timeframe": timeframe},
        )
        RESAMPLE_WARNING_EMITTED = True

    tf_ms = tf_to_ms(timeframe)
    state_key = (symbol, timeframe)
    last_ts = get_last_timestamp(DATA_DIR, symbol, timeframe)
    fetch_states[state_key] = FetchState(symbol=symbol, timeframe=timeframe, last_candle_ts=last_ts)
    queue = verification_queues.get(state_key)
    if queue is None:
        queue = asyncio.Queue(maxsize=500)
        verification_queues[state_key] = queue
    if last_ts is not None:
        try:
            queue.put_nowait(last_ts)
        except asyncio.QueueFull:
            logger.warning(
                "重启后无法排入校验任务，队列已满",
                extra={"symbol": symbol, "timeframe": timeframe, "timestamp": last_ts},
            )

    start_since = parse_start_from_ms(START_FROM)
    if last_ts is not None:
        rewind_since = max(0, last_ts - tf_ms)
        initial_since = max(start_since, rewind_since)
    else:
        initial_since = start_since

    logger.info("启动拉取任务", extra={"symbol": symbol, "timeframe": timeframe})

    try:
        await rest_catchup(symbol, timeframe, derived_timeframes, tf_ms, initial_since)
        await stream_loop(symbol, timeframe, derived_timeframes, tf_ms)
    except asyncio.CancelledError:
        logger.info("取消拉取任务", extra={"symbol": symbol, "timeframe": timeframe})
        state = fetch_states.get(state_key)
        if state:
            state.last_error = "cancelled"
        raise
    finally:
        logger.info("拉取任务退出", extra={"symbol": symbol, "timeframe": timeframe})


async def verification_worker(symbol: str, timeframe: str):
    state_key = (symbol, timeframe)
    queue = verification_queues.get(state_key)
    if queue is None:
        return
    exchange = build_exchange()
    try:
        while True:
            verify_ts = await queue.get()
            try:
                state = fetch_states.get(state_key)
                if state and state.last_verified_ts is not None and verify_ts <= state.last_verified_ts:
                    queue.task_done()
                    continue
                async with VERIFY_FETCH_SEMAPHORE:
                    verification = await exchange.fetch_ohlcv(symbol, timeframe, since=verify_ts, limit=2)
                target_rows = [row for row in verification if row and row[0] == verify_ts]
                if not target_rows:
                    logger.warning(
                        "验证未获取到目标数据",
                        extra={"symbol": symbol, "timeframe": timeframe, "timestamp": verify_ts},
                    )
                    queue.task_done()
                    continue
                candidate = target_rows[-1]
                stored = read_candle_exact(DATA_DIR, symbol, timeframe, verify_ts)
                needs_upsert = stored.empty
                reason = "missing"
                if not needs_upsert:
                    stored_row = stored.iloc[0]
                    open_diff = abs(float(stored_row["open"]) - float(candidate[1]))
                    high_diff = abs(float(stored_row["high"]) - float(candidate[2]))
                    low_diff = abs(float(stored_row["low"]) - float(candidate[3]))
                    close_diff = abs(float(stored_row["close"]) - float(candidate[4]))
                    volume_diff = abs(float(stored_row["volume"]) - float(candidate[5]))
                    if any(diff > 1e-9 for diff in (open_diff, high_diff, low_diff, close_diff, volume_diff)):
                        needs_upsert = True
                        reason = "mismatch"
                if needs_upsert:
                    upsert_candles(DATA_DIR, symbol, timeframe, [candidate])
                    logger.info(
                        "验证回补完成",
                        extra={"symbol": symbol, "timeframe": timeframe, "timestamp": verify_ts, "reason": reason},
                    )
                state = fetch_states.get(state_key)
                if state:
                    state.last_verified_ts = verify_ts
                queue.task_done()
            except asyncio.CancelledError:
                queue.task_done()
                raise
            except Exception as exc:
                logger.warning(
                    "验证请求失败",
                    extra={"symbol": symbol, "timeframe": timeframe, "timestamp": verify_ts, "error": str(exc)},
                )
                queue.task_done()
                await asyncio.sleep(1.0)
    except asyncio.CancelledError:
        raise
    finally:
        with suppress(Exception):
            await exchange.close()
        logger.info("验证任务退出", extra={"symbol": symbol, "timeframe": timeframe})


@app.on_event("startup")
async def on_start():
    ensure_storage(DATA_DIR)
    fetch_tasks.clear()
    verification_queues.clear()
    for s in SYMBOLS:
        for tf in FETCH_TIMEFRAMES:
            state_key = (s, tf)
            verification_queues[state_key] = asyncio.Queue(maxsize=500)
            task = asyncio.create_task(fetch_loop(s, tf), name=f"fetch::{s}::{tf}")
            fetch_tasks.append(task)
            verify_task = asyncio.create_task(verification_worker(s, tf), name=f"verify::{s}::{tf}")
            fetch_tasks.append(verify_task)


@app.on_event("shutdown")
async def on_shutdown():
    if not fetch_tasks:
        return
    logger.info("正在停止拉取任务")
    tasks = list(fetch_tasks)
    for task in tasks:
        task.cancel()
    results = await asyncio.gather(*tasks, return_exceptions=True)
    for result in results:
        if isinstance(result, Exception) and not isinstance(result, asyncio.CancelledError):
            logger.warning("任务停止时出现异常：%s", result)
    fetch_tasks.clear()
    verification_queues.clear()


@app.get("/health")
async def health():
    now = datetime.utcnow().replace(microsecond=0).isoformat() + "Z"
    return {
        "status": "ok",
        "time": now,
        "exchange": EXCHANGE,
        "symbols": SYMBOLS,
        "base_timeframes": FETCH_TIMEFRAMES,
        "derived_timeframes": DERIVED_TIMEFRAMES,
        "timeframes": AVAILABLE_TIMEFRAMES,
        "tasks": [state.to_payload() for state in fetch_states.values()],
    }


@app.get("/api/candles")
def api_candles(
    symbol: str = Query(..., description="如 BTC/USDT:USDT"),
    tf: str = Query("1m", description="时间周期"),
    start: Optional[int] = Query(None, description="开始时间戳(ms)"),
    end: Optional[int] = Query(None, description="结束时间戳(ms)"),
):
    try:
        ensure_symbol_timeframe(symbol, tf)
        df = read_candles(DATA_DIR, symbol, tf, start, end)
        records = df.to_dict("records") if not df.empty else []
        return JSONResponse(records)
    except Exception as e:
        return JSONResponse({"error": str(e)}, status_code=500)


@app.websocket("/ws")
async def ws_endpoint(websocket: WebSocket, symbol: str, tf: str, since: Optional[int] = None):
    if symbol not in VALID_SYMBOLS or tf not in VALID_TIMEFRAMES:
        await websocket.close(code=status.WS_1008_POLICY_VIOLATION, reason="invalid symbol/timeframe")
        return
    await hub.subscribe(websocket, symbol, tf)
    try:
        snap = read_candles(DATA_DIR, symbol, tf, since, None)
        await websocket.send_text(
            json.dumps(
                {
                    "topic": f"candles.{symbol}.{tf}",
                    "type": "snapshot",
                    "data": [
                        {"t": int(r["timestamp"]), "o": r["open"], "h": r["high"], "l": r["low"], "c": r["close"], "v": r["volume"]}
                        for _, r in snap.iterrows()
                    ],
                },
                ensure_ascii=False,
            )
        )
    except Exception:
        pass
    try:
        while True:
            await asyncio.sleep(30)
            await websocket.send_text(json.dumps({"type": "ping", "ts": int(datetime.utcnow().timestamp() * 1000)}))
    except WebSocketDisconnect:
        return


@app.get("/")
def root():
    return {
        "service": "Local Data Service",
        "exchange": EXCHANGE,
        "symbols": SYMBOLS,
        "base_timeframes": FETCH_TIMEFRAMES,
        "derived_timeframes": DERIVED_TIMEFRAMES,
        "timeframes": AVAILABLE_TIMEFRAMES,
    }