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

# docker compose down && docker compose build --no-cache && docker compose up -d
# docker compose down && docker compose build && docker compose up -d

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,
)


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
WS_ENABLED = os.environ.get("WS_ENABLED", "false").lower() in {"1", "true", "yes"}
REST_POLL_INTERVAL = max(1.0, float(os.environ.get("REST_POLL_INTERVAL", "5")))
REST_POLL_WINDOW = max(1, int(os.environ.get("REST_POLL_WINDOW", "10")))

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", "2025-09-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)


def _load_verify_intervals() -> Dict[str, int]:
    mapping: Dict[str, int] = {}
    raw = os.environ.get("VERIFY_INTERVALS", "")
    if not raw:
        return mapping
    parts = [item.strip() for item in raw.split(",") if item.strip()]
    for part in parts:
        if "=" not in part:
            continue
        key, value = part.split("=", 1)
        key = key.strip()
        value = value.strip()
        if not key or not value:
            continue
        try:
            parsed = int(value)
        except ValueError:
            logger.warning("解析 VERIFY_INTERVALS 失败，已忽略条目", extra={"entry": part})
            continue
        if parsed <= 0:
            continue
        mapping[key] = parsed
    return mapping


DEFAULT_VERIFY_INTERVAL_MULTIPLIER = max(1, int(os.environ.get("VERIFY_DEFAULT_INTERVAL", "1")))
VERIFY_INTERVAL_MULTIPLIERS = _load_verify_intervals()

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(2025, 9, 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["VerificationJob"]] = {}


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.drop_duplicates(subset=["timestamp"], keep="last")
        .sort_values("timestamp")
        .reset_index(drop=True)
    )
    base_df["timestamp"] = base_df["timestamp"].astype("int64")

    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


def normalize_candles_for_timeframe(candles: List[CandleRow], tf_ms: int) -> Tuple[List[CandleRow], List[int]]:
    if not candles:
        return [], []
    normalized_map: Dict[int, CandleRow] = {}
    for row in candles:
        if not row:
            continue
        try:
            ts = int(row[0])
            o = float(row[1])
            h = float(row[2])
            l = float(row[3])
            c = float(row[4])
            v = float(row[5])
        except (TypeError, ValueError, IndexError):
            continue
        normalized_map[ts] = [ts, o, h, l, c, v]
    ordered_ts = sorted(normalized_map.keys())
    normalized: List[CandleRow] = []
    missing: List[int] = []
    last_ts: Optional[int] = None
    for ts in ordered_ts:
        normalized.append(normalized_map[ts])
        if last_ts is not None and tf_ms > 0:
            delta = ts - last_ts
            if delta > tf_ms:
                gap_ts = last_ts + tf_ms
                while gap_ts < ts:
                    missing.append(gap_ts)
                    gap_ts += tf_ms
        last_ts = ts
    return normalized, missing


def compute_live_derived_updates(
    symbol: str,
    base_timeframe: str,
    derived_timeframes: List[str],
    base_tf_ms: int,
    candles: List[CandleRow],
    last_closed_ts: Optional[int],
) -> Dict[str, List[Tuple[CandleRow, bool]]]:
    updates: Dict[str, List[Tuple[CandleRow, bool]]] = {}
    if not candles or not derived_timeframes or base_tf_ms <= 0:
        return updates

    derived_ms_map: Dict[str, int] = {}
    max_multiplier = 1
    for target_tf in derived_timeframes:
        derived_ms = tf_to_ms(target_tf)
        if derived_ms is None or derived_ms <= 0 or derived_ms % base_tf_ms != 0:
            continue
        multiplier = derived_ms // base_tf_ms
        derived_ms_map[target_tf] = derived_ms
        if multiplier > max_multiplier:
            max_multiplier = multiplier
    if not derived_ms_map:
        return updates

    window_ms = max_multiplier * base_tf_ms
    newest_ts = max(int(row[0]) for row in candles if row)
    base_start = newest_ts - window_ms + base_tf_ms
    if base_start < 0:
        base_start = 0

    base_df = read_candles(DATA_DIR, symbol, base_timeframe, base_start, newest_ts)
    if base_df.empty:
        return updates
    base_df = base_df.sort_values("timestamp")

    base_rows: List[Tuple[int, float, float, float, float, float]] = []
    for record in candles:
        try:
            ts = int(record[0])
            if ts < base_start:
                continue
            base_rows.append(
                (
                    ts,
                    float(record[1]),
                    float(record[2]),
                    float(record[3]),
                    float(record[4]),
                    float(record[5]),
                )
            )
        except (TypeError, ValueError, IndexError):
            continue

    if base_rows:
        temp_df = pd.DataFrame(
            base_rows,
            columns=["timestamp", "open", "high", "low", "close", "volume"],
        )
        base_df = pd.concat([base_df, temp_df], ignore_index=True)

    if base_df.empty:
        return updates

    base_df = (
        base_df.drop_duplicates(subset=["timestamp"], keep="last")
        .sort_values("timestamp")
        .reset_index(drop=True)
    )

    base_df_indexed = base_df.set_index("timestamp", drop=False)
    if base_df_indexed.empty:
        return updates

    for target_tf, derived_ms in derived_ms_map.items():
        multiplier = derived_ms // base_tf_ms
        rows_with_status: List[Tuple[CandleRow, bool]] = []
        latest_available_ts = int(base_df_indexed.index.max())
        candidate_start = max(base_start, int(base_df_indexed.index.min()))
        first_bucket = (candidate_start // derived_ms) * derived_ms
        if first_bucket < candidate_start:
            first_bucket += derived_ms
        last_possible_start = latest_available_ts - (multiplier - 1) * base_tf_ms
        current_start = first_bucket
        while current_start <= last_possible_start:
            expected_ts = [current_start + i * base_tf_ms for i in range(multiplier)]
            subset = base_df_indexed.reindex(expected_ts)
            if subset.isna().any().any():
                current_start += derived_ms
                continue
            start_ts = current_start
            end_ts = start_ts + derived_ms - base_tf_ms
            row: CandleRow = [
                start_ts,
                float(subset.iloc[0]["open"]),
                float(subset["high"].max()),
                float(subset["low"].min()),
                float(subset.iloc[-1]["close"]),
                float(subset["volume"].sum()),
            ]
            closed = last_closed_ts is not None and last_closed_ts >= end_ts
            upsert_candles(DATA_DIR, symbol, target_tf, [row])
            rows_with_status.append((row, closed))
            current_start += derived_ms
        if rows_with_status:
            updates[target_tf] = rows_with_status
    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,
        }


@dataclass
class VerificationJob:
    timestamp: int
    count: int = 1


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


def enqueue_verification_job(symbol: str, timeframe: str, timestamp: int, count: int) -> None:
    state_key = (symbol, timeframe)
    queue = verification_queues.get(state_key)
    if queue is None:
        return
    state = fetch_states.get(state_key)
    if state is None:
        state = FetchState(symbol=symbol, timeframe=timeframe)
        fetch_states[state_key] = state
    tf_ms = tf_to_ms(timeframe)
    interval_multiplier = VERIFY_INTERVAL_MULTIPLIERS.get(timeframe, DEFAULT_VERIFY_INTERVAL_MULTIPLIER)
    min_gap_ms: Optional[int] = None
    if tf_ms and tf_ms > 0:
        min_gap_ms = tf_ms * max(1, interval_multiplier)
    if state.last_verified_ts is not None and timestamp <= state.last_verified_ts:
        logger.debug(
            "跳过校验任务，已验证更晚时间",
            extra={"symbol": symbol, "timeframe": timeframe, "timestamp": timestamp, "last_verified": state.last_verified_ts},
        )
        return
    if min_gap_ms is not None and state.last_verified_ts is not None:
        gap = timestamp - state.last_verified_ts
        if gap < min_gap_ms:
            logger.debug(
                "跳过校验任务，间隔不足",
                extra={
                    "symbol": symbol,
                    "timeframe": timeframe,
                    "timestamp": timestamp,
                    "last_verified": state.last_verified_ts,
                    "required_gap_ms": min_gap_ms,
                    "actual_gap_ms": gap,
                },
            )
            return
    job = VerificationJob(timestamp=int(timestamp), count=max(1, int(count)))
    try:
        queue.put_nowait(job)
        logger.info(
            "排入校验任务",
            extra={"symbol": symbol, "timeframe": timeframe, "timestamp": timestamp, "count": count},
        )
    except asyncio.QueueFull:
        logger.warning(
            "验证队列已满，丢弃校验任务",
            extra={"symbol": symbol, "timeframe": timeframe, "timestamp": timestamp, "count": count},
        )


async def process_candles(
    symbol: str,
    timeframe: str,
    candles: List[CandleRow],
    derived_timeframes: List[str],
    tf_ms: int,
    finalized: bool,
    allow_verification: bool,
    closed_flags: Optional[List[bool]] = None,
) -> None:
    if not candles:
        return
    if closed_flags is None or len(closed_flags) != len(candles):
        closed_flags = [finalized] * len(candles)
    state_key = (symbol, timeframe)
    upsert_candles(DATA_DIR, symbol, timeframe, candles)
    base_records = list(zip(candles, closed_flags))
    last_closed_ts: Optional[int] = None
    for row, is_closed in base_records:
        if is_closed:
            if last_closed_ts is None or row[0] > last_closed_ts:
                last_closed_ts = row[0]
    if last_closed_ts is None:
        last_closed_ts = candles[-1][0] - tf_ms
    logger.info(
        "基础周期 K 线更新完成",
        extra={
            "symbol": symbol,
            "timeframe": timeframe,
            "count": len(candles),
            "finalized": finalized,
            "last_closed_ts": last_closed_ts,
        },
    )
    derived_updates: List[Tuple[str, List[CandleRow]]] = []
    live_derived_updates: Dict[str, List[Tuple[CandleRow, bool]]] = {}
    derived_closed_ts: Dict[str, int] = {}
    if derived_timeframes:
        needs_resample = finalized or len(candles) > 1
        if needs_resample and RESAMPLE_AVAILABLE:
            derived_updates = await asyncio.to_thread(
                resample_and_store,
                symbol,
                timeframe,
                derived_timeframes,
            )
            if derived_updates:
                logger.info(
                    "衍生周期批量聚合完成",
                    extra={
                        "symbol": symbol,
                        "base_timeframe": timeframe,
                        "targets": [item[0] for item in derived_updates],
                        "origin": "resample",
                    },
                )
        live_derived_updates = await asyncio.to_thread(
            compute_live_derived_updates,
            symbol,
            timeframe,
            derived_timeframes,
            tf_ms,
            candles,
            last_closed_ts,
        )
        if live_derived_updates:
            logger.info(
                "衍生周期实时聚合完成",
                extra={
                    "symbol": symbol,
                    "base_timeframe": timeframe,
                    "targets": list(live_derived_updates.keys()),
                    "origin": "live",
                },
            )
    for row, is_closed in base_records[-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],
                "closed": bool(is_closed),
            },
        }
        await hub.publish(symbol, timeframe, payload)
    last_closed_ts_for_derived = last_closed_ts
    for target_tf, rows in derived_updates:
        if not rows:
            continue
        target_tf_ms = tf_to_ms(target_tf)
        for row in rows:
            ts = int(row[0])
            o, h, l, c, v = map(float, row[1:])
            if target_tf_ms and target_tf_ms > 0:
                derived_closed = last_closed_ts_for_derived is not None and last_closed_ts_for_derived >= ts + target_tf_ms - tf_ms
            else:
                derived_closed = last_closed_ts_for_derived is not None and last_closed_ts_for_derived >= ts
            if derived_closed:
                previous = derived_closed_ts.get(target_tf)
                if previous is None or ts > previous:
                    derived_closed_ts[target_tf] = ts
            derived_state_key = (symbol, target_tf)
            derived_state = fetch_states.get(derived_state_key)
            if derived_state is None:
                derived_state = FetchState(symbol=symbol, timeframe=target_tf)
                fetch_states[derived_state_key] = derived_state
            derived_state.last_fetch_at = datetime.utcnow()
            derived_state.last_candle_ts = ts
            derived_state.consecutive_errors = 0
            derived_state.last_error = None
            payload = {
                "topic": f"candles.{symbol}.{target_tf}",
                "type": "upsert",
                "data": {
                    "t": ts,
                    "o": o,
                    "h": h,
                    "l": l,
                    "c": c,
                    "v": v,
                    "closed": bool(derived_closed),
                },
            }
            await hub.publish(symbol, target_tf, payload)
    if live_derived_updates:
        for target_tf, items in live_derived_updates.items():
            if not items:
                continue
            for row, derived_closed in items:
                ts = int(row[0])
                if derived_closed:
                    previous = derived_closed_ts.get(target_tf)
                    if previous is None or ts > previous:
                        derived_closed_ts[target_tf] = ts
                derived_state_key = (symbol, target_tf)
                derived_state = fetch_states.get(derived_state_key)
                if derived_state is None:
                    derived_state = FetchState(symbol=symbol, timeframe=target_tf)
                    fetch_states[derived_state_key] = derived_state
                derived_state.last_fetch_at = datetime.utcnow()
                derived_state.last_candle_ts = ts
                derived_state.consecutive_errors = 0
                derived_state.last_error = None
                payload = {
                    "topic": f"candles.{symbol}.{target_tf}",
                    "type": "upsert",
                    "data": {
                        "t": ts,
                        "o": float(row[1]),
                        "h": float(row[2]),
                        "l": float(row[3]),
                        "c": float(row[4]),
                        "v": float(row[5]),
                        "closed": bool(derived_closed),
                    },
                }
                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
        last_closed_flag = closed_flags[-1] if closed_flags else finalized
        if allow_verification and last_closed_flag:
            latest_ts = candles[-1][0]
            now_ms = int(datetime.utcnow().timestamp() * 1000)
            if latest_ts > 0 and now_ms - latest_ts <= 2 * tf_ms:
                range_count = 10 if timeframe == "1m" else 1
                enqueue_verification_job(symbol, timeframe, latest_ts, range_count)
        if allow_verification and derived_closed_ts:
            for target_tf, closed_ts in derived_closed_ts.items():
                enqueue_verification_job(symbol, target_tf, closed_ts, 10)


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
    gap_retry: Dict[int, int] = {}
    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
            candles, missing_ts = normalize_candles_for_timeframe(candles, tf_ms)
            if not candles:
                since += tf_ms
                backoff = 1.0
                await asyncio.sleep(0.2)
                continue
            await process_candles(
                symbol,
                timeframe,
                candles,
                derived_timeframes,
                tf_ms,
                finalized=True,
                allow_verification=False,
                closed_flags=[True] * len(candles),
            )
            state.consecutive_errors = 0
            state.last_error = None
            backoff = 1.0
            if missing_ts:
                gap_start = missing_ts[0]
                attempts = gap_retry.get(gap_start, 0) + 1
                gap_retry[gap_start] = attempts
                if attempts <= 3:
                    logger.warning(
                        "检测到缺失 K 线，准备回补",
                        extra={
                            "symbol": symbol,
                            "timeframe": timeframe,
                            "missing_from": gap_start,
                            "missing_to": missing_ts[-1],
                            "attempt": attempts,
                        },
                    )
                    since = gap_start
                    await asyncio.sleep(0.2)
                    continue
                logger.error(
                    "缺失 K 线多次回补失败，已跳过",
                    extra={
                        "symbol": symbol,
                        "timeframe": timeframe,
                        "missing_from": gap_start,
                        "missing_to": missing_ts[-1],
                    },
                )
                gap_retry.pop(gap_start, None)
            else:
                gap_retry.clear()

            since = candles[-1][0] + tf_ms

            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 rest_poll_loop(
    symbol: str,
    timeframe: str,
    derived_timeframes: List[str],
    tf_ms: int,
) -> None:
    state_key = (symbol, timeframe)
    window = max(REST_POLL_WINDOW, 1)
    interval = max(REST_POLL_INTERVAL, 1.0)
    exchange = build_exchange()
    try:
        while True:
            state = fetch_states.get(state_key)
            latest_ts = state.last_candle_ts if state else None
            if latest_ts is None or latest_ts <= 0:
                since = parse_start_from_ms(START_FROM)
            else:
                since = max(0, latest_ts - (window - 1) * tf_ms)
            try:
                async with REST_FETCH_SEMAPHORE:
                    candles = await exchange.fetch_ohlcv(
                        symbol,
                        timeframe,
                        since=since,
                        limit=max(window + 2, window),
                    )
            except asyncio.CancelledError:
                raise
            except (ccxt.NetworkError, ccxt.ExchangeNotAvailable, ccxt.RequestTimeout) as exc:
                logger.warning(
                    "实时轮询网络异常，准备重试",
                    extra={"symbol": symbol, "timeframe": timeframe, "error": str(exc)},
                )
                await asyncio.sleep(interval)
                continue
            except Exception as exc:
                logger.exception(
                    "实时轮询发生异常",
                    extra={"symbol": symbol, "timeframe": timeframe},
                )
                await asyncio.sleep(interval)
                continue
            candles, _ = normalize_candles_for_timeframe(candles, tf_ms)
            if candles:
                closed_flags = [True] * len(candles)
                logger.info(
                    "轮询拉取基础周期完成",
                    extra={
                        "symbol": symbol,
                        "timeframe": timeframe,
                        "count": len(candles),
                        "since": since,
                        "mode": "rest_poll",
                    },
                )
                await process_candles(
                    symbol,
                    timeframe,
                    candles,
                    derived_timeframes,
                    tf_ms,
                    finalized=True,
                    allow_verification=True,
                    closed_flags=closed_flags,
                )
            await asyncio.sleep(interval)
    except asyncio.CancelledError:
        raise
    finally:
        with suppress(Exception):
            await exchange.close()
        logger.info("轮询任务退出", extra={"symbol": symbol, "timeframe": timeframe})


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:
                        continue
                    is_closed = bool(kline.get("x"))
                    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,
                        finalized=is_closed,
                        allow_verification=is_closed,
                        closed_flags=[is_closed],
                    )
        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 and last_ts > 0:
        initial_count = 10 if timeframe == "1m" else 1
        try:
            queue.put_nowait(VerificationJob(timestamp=last_ts, count=initial_count))
        except asyncio.QueueFull:
            logger.warning(
                "重启后无法排入校验任务，队列已满",
                extra={"symbol": symbol, "timeframe": timeframe, "timestamp": last_ts, "count": initial_count},
            )

    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)
        if WS_ENABLED:
            await stream_loop(symbol, timeframe, derived_timeframes, tf_ms)
        else:
            await rest_poll_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
    interval_ms = tf_to_ms(timeframe)
    exchange = build_exchange()
    try:
        while True:
            job = await queue.get()
            try:
                state = fetch_states.get(state_key)
                if state and state.last_verified_ts is not None and job.timestamp <= state.last_verified_ts:
                    logger.debug(
                        "跳过校验任务，时间戳已验证",
                        extra={"symbol": symbol, "timeframe": timeframe, "timestamp": job.timestamp},
                    )
                    continue
                verify_count = max(1, job.count)
                if interval_ms <= 0:
                    interval_ms = tf_to_ms(timeframe)
                start_ts = job.timestamp - (verify_count - 1) * interval_ms
                if start_ts < 0:
                    start_ts = 0
                limit = max(verify_count + 2, 2)
                async with VERIFY_FETCH_SEMAPHORE:
                    fetched = await exchange.fetch_ohlcv(symbol, timeframe, since=start_ts, limit=limit)
                normalized, _ = normalize_candles_for_timeframe(fetched, interval_ms)
                if not normalized:
                    logger.warning(
                        "验证未获取到任何数据",
                        extra={
                            "symbol": symbol,
                            "timeframe": timeframe,
                            "timestamp": job.timestamp,
                            "count": verify_count,
                        },
                    )
                    continue
                logger.info(
                    "开始校验 K 线",
                    extra={
                        "symbol": symbol,
                        "timeframe": timeframe,
                        "timestamp": job.timestamp,
                        "count": verify_count,
                    },
                )
                remote_map = {int(row[0]): row for row in normalized}
                target_ts_list = [start_ts + i * interval_ms for i in range(verify_count)]
                local_df = read_candles(DATA_DIR, symbol, timeframe, start_ts, job.timestamp)
                updated = False
                for ts in target_ts_list:
                    candidate = remote_map.get(ts)
                    if candidate is None:
                        logger.warning(
                            "验证缺失远端数据",
                            extra={"symbol": symbol, "timeframe": timeframe, "timestamp": ts},
                        )
                        continue
                    stored_rows = local_df[local_df["timestamp"] == ts]
                    needs_upsert = stored_rows.empty
                    reason = "missing"
                    if not needs_upsert:
                        stored_row = stored_rows.iloc[0]
                        diffs = (
                            abs(float(stored_row["open"]) - float(candidate[1])),
                            abs(float(stored_row["high"]) - float(candidate[2])),
                            abs(float(stored_row["low"]) - float(candidate[3])),
                            abs(float(stored_row["close"]) - float(candidate[4])),
                            abs(float(stored_row["volume"]) - float(candidate[5])),
                        )
                        if any(diff > 1e-9 for diff in diffs):
                            needs_upsert = True
                            reason = "mismatch"
                    if needs_upsert:
                        upsert_candles(DATA_DIR, symbol, timeframe, [candidate])
                        updated = True
                        logger.info(
                            "验证回补完成",
                            extra={"symbol": symbol, "timeframe": timeframe, "timestamp": ts, "reason": reason},
                        )
                state = fetch_states.get(state_key)
                if state:
                    if state.last_verified_ts is None or job.timestamp > state.last_verified_ts:
                        state.last_verified_ts = job.timestamp
                    if updated:
                        state.last_error = None
            except asyncio.CancelledError:
                raise
            except Exception as exc:
                logger.warning(
                    "验证请求失败",
                    extra={
                        "symbol": symbol,
                        "timeframe": timeframe,
                        "timestamp": job.timestamp,
                        "count": job.count,
                        "error": str(exc),
                    },
                )
                await asyncio.sleep(1.0)
            finally:
                queue.task_done()
    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)
        for tf in DERIVED_TIMEFRAMES:
            state_key = (s, tf)
            if state_key not in verification_queues:
                verification_queues[state_key] = asyncio.Queue(maxsize=500)
            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,
    }