feat: multi-symbol support and per-tf start_time in data_provider

data_provider/config.json

@@ -4,15 +4,28 @@
     "BTC/USDT:USDT",
     "ETH/USDT:USDT",
     "SOL/USDT:USDT",
-    "WIF/USDT:USDT",  
-    "AAVE/USDT:USDT", 
-    "SUI/USDT:USDT", 
-    "1INCH/USDT:USDT", 
+    "XAU/USDT:USDT",
+    "XAG/USDT:USDT",
+    "SAGA/USDT:USDT",
+    "CL/USDT:USDT",
+    "ZEC/USDT:USDT",
+    "XRP/USDT:USDT",
     "DOGE/USDT:USDT",
-    "UNI/USDT:USDT"
+    "BNB/USDT:USDT",
+    "SUI/USDT:USDT",
+    "BILL/USDT:USDT",
+    "BZ/USDT:USDT",
+    "LAB/USDT:USDT",
+    "TON/USDT:USDT",
+    "CRCL/USDT:USDT",
+    "SNDK/USDT:USDT",
+    "1000PEPE/USDT:USDT",
+    "CHIP/USDT:USDT"
   ],
   "start_time": "2024-01-01T00:00:00Z",
+  "start_time_per_tf": {
+    "1m": "2026-01-01T00:00:00Z"
+  },
   "timeframes": ["1m", "1h", "1d", "1w"],
   "data_dir": "./data"
 }
-

data_provider/main.py

@@ -19,6 +19,7 @@ import ccxt  # type: ignore
 import pandas as pd  # type: ignore
 from fastapi import FastAPI, HTTPException, Query, WebSocket, WebSocketDisconnect
 from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import HTMLResponse
 import uvicorn
 from technical.util import resample_to_interval
 
@@ -35,6 +36,10 @@ TIMEFRAME_TO_MS: Dict[str, int] = {
     "1d": 86_400_000,
     "1w": 604_800_000,
 }
+# 内存最大缓存的 K 线数量（超出部分从 CSV 按需读取，不限制则不限量）
+MAX_CANDLES_IN_MEMORY: Dict[str, int] = {
+    "1m": 129_600,   # 3 个月（3 * 30 * 24 * 60）
+}
 # 每个基础周期可派生出的合成周期列表（由该基础周期 K 线 resample 得到）
 DERIVED_TIMEFRAME_PLAN: Dict[str, List[str]] = {
     "1m": ["2m", "3m", "4m", "5m", "10m", "15m", "20m", "25m", "30m", "45m"],
@@ -213,6 +218,14 @@ class DataProvider:
         if start is None:
             raise ValueError("配置文件必须包含 start_time 字段")
         self.start_time_ms: int = start
+        # 按周期独立 start_time（从 start_time_per_tf 字段读取），未配置则回退到全局 start_time
+        self.start_time_map: Dict[str, int] = {}
+        start_per_tf = self.config.get("start_time_per_tf", {})
+        if isinstance(start_per_tf, dict):
+            for tf, ts_str in start_per_tf.items():
+                tf_start = parse_timestamp(ts_str)
+                if tf_start is not None:
+                    self.start_time_map[tf] = tf_start
         self.exchange = self._init_exchange()
         self.data: Dict[str, Dict[str, List[Dict[str, float]]]] = {
             symbol: {tf: [] for tf in self.timeframes} for symbol in self.symbols
@@ -236,6 +249,8 @@ class DataProvider:
         self._stop_event = threading.Event()
         self._fetch_thread: Optional[threading.Thread] = None
         self._persist_thread: Optional[threading.Thread] = None
+        self._backfill_thread: Optional[threading.Thread] = None
+        self._cold_start_done = False
         # 记录断线后需要从哪个 since 重新拉取（symbol -> timeframe -> since_ms）
         self._resume_since: Dict[str, Dict[str, int]] = {}
         # 恢复点持久化文件
@@ -270,6 +285,10 @@ class DataProvider:
                 unique.append(item)
         return unique
 
+    def _get_start_time(self, timeframe: str) -> int:
+        """获取指定周期的 start_time（毫秒）。优先使用 start_time_per_tf，否则回退全局。"""
+        return self.start_time_map.get(timeframe, self.start_time_ms)
+
     def _validate_timeframes(self, configured: Optional[Iterable[str]]) -> List[str]:
         """校验周期在允许集合内；未配置则默认 TIMEFRAME_ORDER 全部；顺序优先按 TIMEFRAME_ORDER。"""
         if not configured:
@@ -348,6 +367,19 @@ class DataProvider:
         logger.debug("时间周期 %s 合并后K线数量: %s", timeframe, len(ordered))
         return ordered
 
+    def _trim(self, symbol: str, timeframe: str) -> None:
+        """将内存缓存裁剪到 MAX_CANDLES_IN_MEMORY 上限（全量数据已在 CSV 中）。"""
+        max_count = MAX_CANDLES_IN_MEMORY.get(timeframe)
+        if max_count is None:
+            return
+        sym_data = self.data.get(symbol)
+        if sym_data is None:
+            return
+        candles = sym_data.get(timeframe)
+        if candles is None or len(candles) <= max_count:
+            return
+        sym_data[timeframe] = candles[-max_count:]
+
     def _write_to_disk(self, symbol: str, timeframe: str, data: List[Dict[str, float]]) -> None:
         """先写临时文件再 replace，避免写入中断导致 CSV 损坏。"""
         path = self._data_file_path(symbol, timeframe)
@@ -368,75 +400,159 @@ class DataProvider:
         logger.info("交易对 %s 时间周期 %s 已写入磁盘 (%s 根K线)", symbol, timeframe, len(data))
 
     def _fetch_history(self, symbol: str, timeframe: str, since_ms: int) -> List[List[float]]:
-        """从 since_ms 分页拉取直到接近当前时间；遇限频则 sleep 重试。"""
+        """从 since_ms 分页拉取直到接近当前时间。
+
+        当 since_ms 早于交易所数据保留期时，先从最新数据往回探测有效起点，
+        再向后分页补齐。"""
         results: List[List[float]] = []
         limit = 1500
         now_ms = self.exchange.milliseconds()
         tf_ms = TIMEFRAME_TO_MS[timeframe]
-        fetch_since = since_ms
+
+        # 先尝试直接拉取
+        first_batch = self._fetch_ohlcv_retry(symbol, timeframe, since=since_ms, limit=limit)
+
+        # 若返回空，或首根时间远晚于 since（交易所忽略了过老的 since），启用反向探测
+        gap_ms = tf_ms * limit * 2
+        first_skipped = (
+            not first_batch
+            or (first_batch[0][0] - since_ms > gap_ms and first_batch[-1][0] >= now_ms - tf_ms * 10)
+        )
+        if first_skipped:
+            logger.info("since 过老被交易所跳过，启用反向探测 %s %s (since=%s)", symbol, timeframe, to_utc_iso(since_ms))
+            # 反向探测：从当前时间往前翻页，找到实际最早可用的 K 线
+            all_collected: List[List[float]] = []
+            search_since = max(0, now_ms - tf_ms * limit)
+            search_rounds = 0
+            max_search = 200
+            earliest_ts = None
+            while search_rounds < max_search:
+                search_rounds += 1
+                batch = self._fetch_ohlcv_retry(symbol, timeframe, since=search_since, limit=limit)
+                if not batch:
+                    break
+                all_collected = batch + all_collected
+                first_ts = batch[0][0]
+                if first_ts == search_since or first_ts <= since_ms:
+                    earliest_ts = first_ts
+                    break
+                search_since = max(0, first_ts - tf_ms * limit)
+                time.sleep(self.exchange.rateLimit / 1000 if self.exchange.rateLimit else 0.2)
+            if earliest_ts is None and all_collected:
+                earliest_ts = all_collected[0][0]
+            if earliest_ts is not None:
+                logger.info("反向探测找到最早 %s: %s (%s)", symbol, to_utc_iso(earliest_ts), timeframe)
+            # 用收集到的数据作为起点，继续向前分页到 now
+            results = all_collected
+            if results:
+                last_ts = results[-1][0]
+                if last_ts < now_ms - tf_ms:
+                    extra = self._fetch_ohlcv_paginate(symbol, timeframe, last_ts + tf_ms, now_ms, limit)
+                    results.extend(extra)
+            logger.info("交易对 %s 时间周期 %s 拉取历史K线数量: %s", symbol, timeframe, len(results))
+            return results
+
+        # 直接拉取成功，正常向后分页
+        results = list(first_batch)
+        last_ts = results[-1][0]
+        if last_ts < now_ms - tf_ms:
+            extra = self._fetch_ohlcv_paginate(symbol, timeframe, last_ts + tf_ms, now_ms, limit)
+            results.extend(extra)
+        logger.info("交易对 %s 时间周期 %s 拉取历史K线数量: %s", symbol, timeframe, len(results))
+        return results
+
+    def _fetch_ohlcv_retry(self, symbol: str, timeframe: str, since: int, limit: int) -> List[List[float]]:
+        """带重试的单次 fetch_ohlcv，限频/网络错误时等待重试。"""
+        max_attempts = 5
+        for attempt in range(max_attempts):
+            try:
+                return self.exchange.fetch_ohlcv(symbol, timeframe=timeframe, since=since, limit=limit)
+            except ccxt.RateLimitExceeded as exc:
+                wait_s = (self.exchange.rateLimit / 1000 if self.exchange.rateLimit else 1) * (attempt + 1)
+                logger.warning("触发频率限制，等待 %.1fs: %s", wait_s, exc)
+                time.sleep(wait_s)
+            except ccxt.BaseError as exc:
+                logger.error("拉取K线失败 (%s, %s, since=%s): %s (attempt %s/%s)",
+                             timeframe, symbol, since, exc, attempt + 1, max_attempts)
+                time.sleep(3 * (attempt + 1))
+        logger.error("拉取K线失败，已达最大重试次数: %s %s", symbol, timeframe)
+        return []
+
+    def _fetch_ohlcv_paginate(self, symbol: str, timeframe: str, start_ms: int, end_ms: int, limit: int) -> List[List[float]]:
+        """从 start_ms 向后分页拉取到 end_ms。"""
+        results: List[List[float]] = []
+        tf_ms = TIMEFRAME_TO_MS[timeframe]
+        fetch_since = start_ms
         max_rounds = 5000
         rounds = 0
-        while fetch_since < now_ms and rounds < max_rounds:
+        while fetch_since < end_ms and rounds < max_rounds:
             rounds += 1
-            try:
-                candles = self.exchange.fetch_ohlcv(
-                    symbol,
-                    timeframe=timeframe,
-                    since=fetch_since,
-                    limit=limit,
-                )
-            except ccxt.RateLimitExceeded as exc:
-                logger.warning("触发频率限制，等待: %s", exc)
-                time.sleep(self.exchange.rateLimit / 1000 if self.exchange.rateLimit else 1)
-                continue
-            except ccxt.BaseError as exc:
-                logger.error("拉取历史K线失败 (%s, %s): %s", timeframe, fetch_since, exc)
-                time.sleep(5)
-                continue
+            candles = self._fetch_ohlcv_retry(symbol, timeframe, since=fetch_since, limit=limit)
             if not candles:
                 break
             results.extend(candles)
             last_ts = candles[-1][0]
             fetch_since = last_ts + tf_ms
-            if last_ts >= now_ms - tf_ms:
+            if last_ts >= end_ms - tf_ms:
                 break
             time.sleep(self.exchange.rateLimit / 1000 if self.exchange.rateLimit else 0.2)
-        logger.info("交易对 %s 时间周期 %s 拉取历史K线数量: %s", symbol, timeframe, len(results))
         return results
 
     def initialize(self) -> None:
-        """阻塞式启动：加载本地、从倒数第二根或配置起点补历史、写盘并 set _ready。"""
-        logger.info("开始初始化数据提供商")
+        """快速启动：仅加载本地 CSV 到内存，立即设 _ready 让服务可用。后台再补拉交易所数据。"""
+        logger.info("开始初始化数据提供商（仅加载本地数据）")
         for symbol in self.symbols:
             for timeframe in self.timeframes:
                 existing = self._load_local(symbol, timeframe)
+                with self._lock:
+                    self.data.setdefault(symbol, {})[timeframe] = existing
+                self._trim(symbol, timeframe)
+                logger.info("交易对 %s 时间周期 %s 已加载本地K线: %s 根", symbol, timeframe, len(existing))
+        self._ready.set()
+        logger.info("数据初始化完成，服务已就绪（后台将补拉交易所数据）")
+
+    def _cold_start_backfill(self) -> None:
+        """后台一次性回填：从交易所拉取每个 symbol/timeframe 的缺失历史，合并后写盘。"""
+        logger.info("开始后台回填交易所数据")
+        for symbol in self.symbols:
+            for timeframe in self.timeframes:
+                if self._stop_event.is_set():
+                    return
+                with self._lock:
+                    existing = list(self.data.get(symbol, {}).get(timeframe, []))
                 tf_ms = TIMEFRAME_TO_MS[timeframe]
                 last_ts = existing[-1]["timestamp"] if existing else None
+
+                # 计算 fetch_since：优先从数据末尾补，但若数据量极少
+                #（刷新线程抢先写入的少量近期数据），则回退到 start_time 做全量回填
                 if last_ts is not None:
-                    if len(existing) >= 2:
-                        # 从倒数第二根起拉，避免最后一根未收盘重复/缺口
-                        fetch_since = existing[-2]["timestamp"]
+                    if len(existing) > RECENT_CANDLE_LIMIT * 3:
+                        # 已有足够历史数据，仅增量补拉
+                        fetch_since = (existing[-2]["timestamp"] if len(existing) >= 2
+                                       else max(0, existing[-1]["timestamp"] - tf_ms))
                     else:
-                        fetch_since = max(0, last_ts - tf_ms)
+                        # 只有刷新线程抢先写入的少量数据，从 start_time 全量回填
+                        fetch_since = self._get_start_time(timeframe)
+                        logger.info("数据量不足 (%s 根)，从 start_time 回填 %s %s",
+                                    len(existing), symbol, timeframe)
                 else:
-                    fetch_since = self.start_time_ms
-                logger.debug(
-                    "初始化拉取参数",
-                    extra={
-                        "symbol": symbol,
-                        "timeframe": timeframe,
-                        "existing_last": last_ts,
-                        "fetch_since": fetch_since,
-                        "tf_ms": tf_ms,
-                    },
-                )
+                    fetch_since = self._get_start_time(timeframe)
+                try:
                     history = self._fetch_history(symbol, timeframe, fetch_since)
-                merged = self._merge_candles(timeframe, existing, history)
+                    # 再次读取最新数据后合并，避免覆盖并发刷新线程写入的更新
                     with self._lock:
-                    self.data.setdefault(symbol, {})[timeframe] = merged
+                        current = list(self.data.get(symbol, {}).get(timeframe, []))
+                        merged = self._merge_candles(timeframe, current, history)
+                        self.data[symbol][timeframe] = merged
                     self._write_to_disk(symbol, timeframe, merged)
-        self._ready.set()
-        logger.info("数据初始化完成")
+                    self._trim(symbol, timeframe)
+                    self._notify_update(symbol, timeframe)
+                    logger.info("回填完成 %s %s: +%s 根新K线", symbol, timeframe, len(history))
+                except Exception as exc:
+                    logger.error("回填失败 %s %s: %s", symbol, timeframe, exc)
+                    self._set_resume_since(symbol, timeframe, fetch_since)
+        logger.info("后台回填全部完成")
+        self._cold_start_done = True
         
     def resample_df(self, df: pd.DataFrame, interval: int) -> pd.DataFrame:
         """将基础周期 DataFrame 聚合为 interval 分钟周期（freqtrade technical.util）。"""
@@ -549,23 +665,28 @@ class DataProvider:
                 logger.error("数据更新回调异常: %s", exc)
 
     def start_background_workers(self) -> None:
-        """启动增量刷新线程与周期性落盘线程。"""
-        if self._fetch_thread and self._fetch_thread.is_alive():
-            return
+        """启动后台线程：冷启动回填、增量刷新、周期性落盘。"""
         self._stop_event.clear()
+        self._backfill_thread = threading.Thread(target=self._cold_start_backfill, name="backfill-loop", daemon=True)
         self._fetch_thread = threading.Thread(target=self._refresh_loop, name="refresh-loop", daemon=True)
         self._persist_thread = threading.Thread(target=self._persist_loop, name="persist-loop", daemon=True)
+        self._backfill_thread.start()
         self._fetch_thread.start()
         self._persist_thread.start()
-        logger.info("后台线程已启动")
+        logger.info("后台线程已启动（回填 + 增量刷新 + 落盘）")
 
     def stop(self) -> None:
         """停止后台线程（应用关闭时 lifespan finally 调用）。"""
         self._stop_event.set()
-        if self._fetch_thread:
-            self._fetch_thread.join(timeout=5)
-        if self._persist_thread:
-            self._persist_thread.join(timeout=5)
+        for thread, name in [
+            (self._backfill_thread, "backfill"),
+            (self._fetch_thread, "fetch"),
+            (self._persist_thread, "persist"),
+        ]:
+            if thread:
+                thread.join(timeout=5)
+                if thread.is_alive():
+                    logger.warning("后台线程 %s 未能在 5s 内结束", name)
         logger.info("数据提供商已停止")
 
     def _refresh_loop(self) -> None:
@@ -591,6 +712,7 @@ class DataProvider:
                                 self.data[symbol][timeframe] = merged
                             self._notify_update(symbol, timeframe)
                             self._clear_resume_since(symbol, timeframe)
+                            self._trim(symbol, timeframe)
                         else:
                             # 正常增量获取最近若干根K线
                             candles = self.exchange.fetch_ohlcv(
@@ -605,6 +727,7 @@ class DataProvider:
                                 merged = self._merge_candles(timeframe, current, candles)
                                 self.data[symbol][timeframe] = merged
                             self._notify_update(symbol, timeframe)
+                            self._trim(symbol, timeframe)
                     except ccxt.BaseError as exc:
                         logger.error("更新最新K线失败 (%s %s): %s", symbol, timeframe, exc)
                         # 记录应当从何时恢复拉取，避免重连后从当前时间开始导致丢K
@@ -613,10 +736,10 @@ class DataProvider:
                         if current:
                             last_ts = int(current[-1]["timestamp"])
                         else:
-                            last_ts = self.start_time_ms
+                            last_ts = self._get_start_time(timeframe)
                         tf_ms = TIMEFRAME_TO_MS[timeframe]
                         # 回退一个周期，确保包含可能未完全收盘的K线，去重由 _merge_candles 处理
-                        since_ms = max(self.start_time_ms, last_ts - tf_ms)
+                        since_ms = max(self._get_start_time(timeframe), last_ts - tf_ms)
                         self._set_resume_since(symbol, timeframe, since_ms)
                         time.sleep(2)
                         continue
@@ -629,7 +752,7 @@ class DataProvider:
             self._persist_all()
 
     def _persist_all(self) -> None:
-        """在锁内复制 data 后落盘，避免长时间持锁。"""
+        """在锁内复制 data 后落盘，避免长时间持锁。有限制的时间周期跳过不写（CSV 已有全量历史）。"""
         if not self._ready.is_set():
             return
         with self._lock:
@@ -639,6 +762,8 @@ class DataProvider:
             }
         for symbol, tf_map in snapshot.items():
             for timeframe, data in tf_map.items():
+                if timeframe in MAX_CANDLES_IN_MEMORY:
+                    continue  # 内存只有尾部 N 根，不覆盖 CSV 全量
                 self._write_to_disk(symbol, timeframe, data)
         # 周期性也保存一次恢复点，保证一致性
         self._save_resume_since()
@@ -663,9 +788,18 @@ class DataProvider:
         end_ms: Optional[int],
         limit: Optional[int],
     ) -> List[Dict[str, float]]:
-        """从内存读取已缓存的基础周期 K 线并按时间/limit 裁剪。"""
+        """从内存读取已缓存的基础周期 K 线并按时间/limit 裁剪；内存不足时从 CSV 补充。"""
         with self._lock:
             candles = list(self.data.get(symbol, {}).get(timeframe, []))
+        # 有限制的时间周期，若内存不够老则从 CSV 补充历史
+        if timeframe in MAX_CANDLES_IN_MEMORY and candles and start_ms is not None:
+            if start_ms < candles[0]["timestamp"]:
+                csv_candles = self._load_local(symbol, timeframe)
+                if csv_candles:
+                    merged = {c["timestamp"]: c for c in csv_candles}
+                    for c in candles:
+                        merged[c["timestamp"]] = c
+                    candles = sorted(merged.values(), key=lambda x: x["timestamp"])
         if start_ms is not None:
             candles = [row for row in candles if row["timestamp"] >= start_ms]
         if end_ms is not None:
@@ -853,6 +987,532 @@ def create_app(provider: DataProvider) -> FastAPI:
             "ready": provider.is_ready(),
         }
 
+    @app.get("/api-manual", response_class=HTMLResponse)
+    async def api_manual():
+        """数据服务 API 使用手册页面。"""
+        derived_plan = DERIVED_TIMEFRAME_PLAN
+        timeframe_ms = TIMEFRAME_TO_MS
+        memory_limits = MAX_CANDLES_IN_MEMORY
+        base_tfs = provider.timeframes
+        derived_tfs = provider.get_derived_timeframes()
+        all_tfs = provider.get_available_timeframes()
+        symbols = provider.symbols
+
+        # Build derived timeframe table rows
+        derived_rows = ""
+        for base, derived_list in derived_plan.items():
+            present = "✓" if base in base_tfs else "✗"
+            derived_rows += f"<tr><td><code>{base}</code></td><td>{present}</td><td>{', '.join(f'<code>{d}</code>' for d in derived_list)}</td></tr>"
+
+        # Build memory limits table rows
+        memory_rows = ""
+        for tf in base_tfs:
+            limit = memory_limits.get(tf)
+            if limit is None:
+                span = "unlimited"
+            elif tf == "1m":
+                span = f"{limit:,} candles (~3 months)"
+            elif tf == "1h":
+                span = f"{limit:,} candles (~1 year)"
+            else:
+                span = f"{limit:,} candles"
+            memory_rows += f"<tr><td><code>{tf}</code></td><td>{span}</td></tr>"
+
+        # Pre-build config JSON example to avoid f-string escaping issues
+        import json as _json
+        _config_obj = {
+            "exchange": "binance",
+            "symbols": ["BTC/USDT:USDT", "ETH/USDT:USDT"],
+            "start_time": "2024-01-01T00:00:00Z",
+            "start_time_per_tf": {"1m": "2026-01-01T00:00:00Z"},
+            "timeframes": ["1m", "1h", "1d", "1w"],
+            "data_dir": "./data",
+        }
+        config_example = _json.dumps(_config_obj, indent=2, ensure_ascii=False)
+
+        return f"""<!DOCTYPE html>
+<html lang="zh-CN">
+<head>
+    <meta charset="utf-8">
+    <meta name="viewport" content="width=device-width, initial-scale=1.0">
+    <title>Data Provider — API Manual</title>
+    <link href="https://cdn.jsdelivr.net/npm/bootstrap@5.1.3/dist/css/bootstrap.min.css" rel="stylesheet">
+    <link href="https://cdn.jsdelivr.net/npm/bootstrap-icons@1.8.1/font/bootstrap-icons.css" rel="stylesheet">
+    <style>
+        body {{
+            font-family: "Helvetica Neue", Arial, "PingFang SC", "Microsoft YaHei", sans-serif;
+            background: #f8f9fa; color: #333; padding: 0;
+        }}
+        .hero {{
+            background: linear-gradient(135deg, #1a1a2e 0%, #16213e 50%, #0f3460 100%);
+            color: #fff; padding: 48px 0 40px; margin-bottom: 32px;
+        }}
+        .hero h1 {{ font-size: 2rem; font-weight: 700; margin: 0; }}
+        .hero .badge {{ font-size: 0.8rem; vertical-align: middle; margin-left: 10px; }}
+        .hero p {{ opacity: 0.85; margin: 8px 0 0; font-size: 0.95rem; }}
+        .nav-pills .nav-link {{ color: #495057; border-radius: 6px; padding: 6px 16px; font-size: 0.9rem; }}
+        .nav-pills .nav-link.active {{ background: #0d6efd; }}
+        .section {{ display: none; }}
+        .section.active {{ display: block; }}
+        .card {{ border: 1px solid #e9ecef; box-shadow: 0 1px 3px rgba(0,0,0,0.04); margin-bottom: 20px; }}
+        .card-header {{ background: #fff; border-bottom: 1px solid #e9ecef; font-weight: 600; font-size: 0.95rem; padding: 12px 18px; }}
+        pre {{ background: #1e1e2e; color: #cdd6f4; border-radius: 8px; padding: 16px; font-size: 0.85rem; line-height: 1.6; overflow-x: auto; }}
+        pre code {{ color: inherit; }}
+        .method-badge {{ display: inline-block; padding: 2px 8px; border-radius: 4px; font-size: 0.75rem; font-weight: 700; margin-right: 8px; width: 52px; text-align: center; }}
+        .method-GET {{ background: #d1fae5; color: #065f46; }}
+        .method-WS  {{ background: #ede9fe; color: #5b21b6; }}
+        .endpoint-path {{ font-family: monospace; font-size: 0.9rem; }}
+        .table {{ font-size: 0.875rem; }}
+        .table code {{ background: #f1f3f5; padding: 1px 5px; border-radius: 3px; font-size: 0.82rem; }}
+        .status-ready {{ color: #198754; }}
+        .status-notready {{ color: #dc3545; }}
+        .param-required {{ color: #dc3545; font-weight: 600; }}
+        .toc-item {{ cursor: pointer; }}
+        footer {{ border-top: 1px solid #dee2e6; margin-top: 48px; padding: 20px 0; color: #868e96; font-size: 0.85rem; text-align: center; }}
+        .endpoint-card {{ border-left: 3px solid #0d6efd; }}
+        .endpoint-card.ws {{ border-left-color: #7c3aed; }}
+    </style>
+</head>
+<body>
+
+<div class="hero">
+    <div class="container">
+        <h1>Data Provider <span class="badge bg-success">v1.0.0</span></h1>
+        <p>Cryptocurrency OHLCV data microservice — REST + WebSocket API for candlestick (K-line) data</p>
+        <p style="font-size:0.82rem; opacity:0.7;">Exchange: {provider.exchange_name} &middot; Symbols: {len(symbols)} &middot; Timeframes: {len(all_tfs)} &middot; Status: <span class="{'status-ready' if provider.is_ready() else 'status-notready'}">{'Ready' if provider.is_ready() else 'Initializing…'}</span></p>
+    </div>
+</div>
+
+<div class="container" style="max-width: 960px;">
+
+    <!-- Tabs -->
+    <ul class="nav nav-pills mb-4" id="manualTabs" role="tablist">
+        <li class="nav-item"><button class="nav-link active" data-section="overview">Overview</button></li>
+        <li class="nav-item"><button class="nav-link" data-section="rest">REST API</button></li>
+        <li class="nav-item"><button class="nav-link" data-section="ws">WebSocket</button></li>
+        <li class="nav-item"><button class="nav-link" data-section="timeframes">Timeframes</button></li>
+        <li class="nav-item"><button class="nav-link" data-section="config">Configuration</button></li>
+        <li class="nav-item"><button class="nav-link" data-section="examples">Code Examples</button></li>
+    </ul>
+
+    <!-- ==================== OVERVIEW ==================== -->
+    <div class="section active" id="section-overview">
+        <div class="card">
+            <div class="card-header">What is Data Provider?</div>
+            <div class="card-body">
+                <p><strong>Data Provider</strong> is a standalone microservice that fetches, caches, and serves cryptocurrency OHLCV (Open/High/Low/Close/Volume) candlestick data from the Binance Futures exchange via CCXT.</p>
+                <p>It decouples data acquisition from analysis — consumers request historical or real-time market data through a simple REST or WebSocket API instead of talking to exchange APIs directly.</p>
+                <hr>
+                <h6>Key Features</h6>
+                <div class="row">
+                    <div class="col-md-6">
+                        <ul class="small">
+                            <li>Fetches base OHLCV at <b>1m, 1h, 1d, 1w</b> from Binance</li>
+                            <li>Synthesizes <b>{len(derived_tfs)} derived timeframes</b> by resampling</li>
+                            <li>Dual storage: in-memory cache + CSV persistence</li>
+                            <li>Background refresh every <b>{RECENT_FETCH_INTERVAL}s</b></li>
+                        </ul>
+                    </div>
+                    <div class="col-md-6">
+                        <ul class="small">
+                            <li>REST endpoint at <code>/api/candles</code></li>
+                            <li>WebSocket real-time push at <code>/ws</code></li>
+                            <li>Automatic disconnect recovery</li>
+                            <li>Fully Dockerized, single JSON config file</li>
+                        </ul>
+                    </div>
+                </div>
+            </div>
+        </div>
+
+        <div class="card">
+            <div class="card-header">Architecture</div>
+            <div class="card-body">
+                <pre style="background:#f8f9fa; color:#333; text-align:center; font-size:0.8rem; line-height:1.4;">
+┌───────────────────────┐
+│   Binance Futures     │
+│   (ccxt driver)       │
+└──────────┬────────────┘
+           │ fetch_ohlcv()
+┌──────────┴────────────┐
+│    DataProvider        │
+│                        │
+│  · In-memory cache     │
+│  · CSV persistence     │
+│  · Background refresh  │
+│  · Derived timeframe   │
+│    resample engine     │
+└──────┬────────┬────────┘
+       │        │
+  REST │        │ WebSocket
+       │        │
+┌──────┴────────┴────────┐
+│  Consumers (Flask app, │
+│  browser, scripts…)    │
+└────────────────────────┘</pre>
+            </div>
+        </div>
+
+        <div class="card">
+            <div class="card-header">Endpoints at a Glance</div>
+            <div class="card-body p-0">
+                <table class="table table-hover mb-0">
+                    <thead class="table-light"><tr><th>Method</th><th>Path</th><th>Description</th></tr></thead>
+                    <tbody>
+                        <tr><td><span class="method-badge method-GET">GET</span></td><td class="endpoint-path">/</td><td>Service info &amp; health summary</td></tr>
+                        <tr><td><span class="method-badge method-GET">GET</span></td><td class="endpoint-path">/health</td><td>Structured health check (JSON)</td></tr>
+                        <tr><td><span class="method-badge method-GET">GET</span></td><td class="endpoint-path">/timeframes</td><td>List available base &amp; derived timeframes</td></tr>
+                        <tr><td><span class="method-badge method-GET">GET</span></td><td class="endpoint-path">/api/candles</td><td>Fetch OHLCV candles</td></tr>
+                        <tr><td><span class="method-badge method-WS">WS</span></td><td class="endpoint-path">/ws</td><td>Real-time K-line streaming</td></tr>
+                        <tr><td><span class="method-badge method-GET">GET</span></td><td class="endpoint-path">/api-manual</td><td>This page</td></tr>
+                    </tbody>
+                </table>
+            </div>
+        </div>
+
+        <div class="card">
+            <div class="card-header">Data Model</div>
+            <div class="card-body">
+                <p class="small">Every candle object (REST response, WebSocket payload) has this structure:</p>
+                <table class="table table-bordered table-sm">
+                    <thead class="table-light"><tr><th>Field</th><th>Type</th><th>Example</th></tr></thead>
+                    <tbody>
+                        <tr><td><code>timestamp</code></td><td>int</td><td><code>1704067200000</code></td></tr>
+                        <tr><td><code>datetime</code></td><td>string</td><td><code>"2024-01-01T00:00:00Z"</code></td></tr>
+                        <tr><td><code>open</code></td><td>float</td><td><code>42314.0</code></td></tr>
+                        <tr><td><code>high</code></td><td>float</td><td><code>44266.0</code></td></tr>
+                        <tr><td><code>low</code></td><td>float</td><td><code>42207.9</code></td></tr>
+                        <tr><td><code>close</code></td><td>float</td><td><code>44230.2</code></td></tr>
+                        <tr><td><code>volume</code></td><td>float</td><td><code>206424.144</code></td></tr>
+                    </tbody>
+                </table>
+            </div>
+        </div>
+    </div>
+
+    <!-- ==================== REST API ==================== -->
+    <div class="section" id="section-rest">
+
+        <h5 class="mb-3"><span class="method-badge method-GET">GET</span> /health</h5>
+        <div class="card endpoint-card">
+            <div class="card-body">
+                <p class="small">Health check — use <code>ready</code> to determine if the service finished cold-start initialization.</p>
+                <p><strong>Response</strong> <span class="text-muted small">200 OK</span></p>
+                <pre><code>{{
+  "status": "ok",
+  "exchange": "binance",
+  "symbols": ["BTC/USDT:USDT", "ETH/USDT:USDT"],
+  "base_timeframes": ["1m", "1h", "1d", "1w"],
+  "derived_timeframes": ["2m","3m","4m","5m",...],
+  "timeframes": ["1m","2m","3m",...],
+  "ready": true
+}}</code></pre>
+                <button class="btn btn-sm btn-outline-secondary" onclick="fetch('/health').then(r=>r.json()).then(d=>alert(JSON.stringify(d,null,2)))">Try it</button>
+            </div>
+        </div>
+
+        <h5 class="mb-3 mt-4"><span class="method-badge method-GET">GET</span> /api/candles</h5>
+        <div class="card endpoint-card">
+            <div class="card-body">
+                <p class="small">The primary data endpoint. Returns OHLCV candlestick data for a given symbol and timeframe.</p>
+                <p><strong>Query Parameters</strong></p>
+                <table class="table table-bordered table-sm">
+                    <thead class="table-light"><tr><th>Param</th><th>Type</th><th>Required</th><th>Default</th><th>Description</th></tr></thead>
+                    <tbody>
+                        <tr><td><code>symbol</code></td><td>string</td><td><span class="param-required">Yes</span></td><td>—</td><td>Trading pair, e.g. <code>BTC/USDT:USDT</code></td></tr>
+                        <tr><td><code>tf</code></td><td>string</td><td>No</td><td><code>1m</code></td><td>Timeframe — any base or derived</td></tr>
+                        <tr><td><code>start</code></td><td>int</td><td>No</td><td>—</td><td>Start time, UTC milliseconds</td></tr>
+                        <tr><td><code>end</code></td><td>int</td><td>No</td><td>—</td><td>End time, UTC milliseconds</td></tr>
+                        <tr><td><code>limit</code></td><td>int</td><td>No</td><td>—</td><td>Max candles to return</td></tr>
+                    </tbody>
+                </table>
+                <p><strong>Response</strong> <span class="text-muted small">200 OK — array of candle objects</span></p>
+                <pre><code>[
+  {{
+    "timestamp": 1704067200000,
+    "datetime": "2024-01-01T00:00:00Z",
+    "open": 42314.0,
+    "high": 44266.0,
+    "low": 42207.9,
+    "close": 44230.2,
+    "volume": 206424.144
+  }}
+]</code></pre>
+                <p class="small text-muted">At least one of <code>start</code>/<code>end</code> or <code>limit</code> should be provided. If none given, all in-memory candles are returned (subject to memory limits).</p>
+            </div>
+        </div>
+
+        <h5 class="mb-3 mt-4"><span class="method-badge method-GET">GET</span> /timeframes</h5>
+        <div class="card endpoint-card">
+            <div class="card-body">
+                <p class="small">Lists all available timeframes: base (fetched from exchange) and derived (synthesized by resampling).</p>
+                <pre><code>{{
+  "base_timeframes": ["1m","1h","1d","1w"],
+  "derived_timeframes": ["2m","3m","4m","5m",...],
+  "timeframes": ["1m","2m","3m",...]
+}}</code></pre>
+            </div>
+        </div>
+
+        <h5 class="mb-3 mt-4"><span class="method-badge method-GET">GET</span> /</h5>
+        <div class="card endpoint-card">
+            <div class="card-body">
+                <p class="small">Root endpoint — returns service name, exchange, symbols, timeframes, and ready status. Same data as <code>/health</code>.</p>
+            </div>
+        </div>
+    </div>
+
+    <!-- ==================== WEBSOCKET ==================== -->
+    <div class="section" id="section-ws">
+
+        <div class="card endpoint-card ws">
+            <div class="card-header"><span class="method-badge method-WS">WS</span> /ws — Real-time K-line Streaming</div>
+            <div class="card-body">
+                <p class="small">Connect to receive live incremental candle updates. Subscribe to specific symbol/timeframe pairs; receive a full snapshot followed by real-time push.</p>
+
+                <h6 class="mt-3">Connection URL</h6>
+                <pre><code>ws://localhost:80/ws</code></pre>
+                <p class="small text-muted">Replace <code>localhost:80</code> with the server address if connecting remotely.</p>
+
+                <h6 class="mt-3">Client → Server Messages</h6>
+                <table class="table table-bordered table-sm">
+                    <thead class="table-light"><tr><th>Action</th><th>Payload</th><th>Description</th></tr></thead>
+                    <tbody>
+                        <tr>
+                            <td><code>subscribe</code></td>
+                            <td><pre style="margin:0;padding:4px 8px;font-size:0.78rem;"><code>{{"action":"subscribe","symbol":"BTC/USDT:USDT","timeframe":"1m"}}</code></pre></td>
+                            <td>Subscribe to a symbol/timeframe. Server replies with <code>subscribed</code>, then a <code>snapshot</code> of recent candles.</td>
+                        </tr>
+                        <tr>
+                            <td><code>unsubscribe</code></td>
+                            <td><pre style="margin:0;padding:4px 8px;font-size:0.78rem;"><code>{{"action":"unsubscribe","symbol":"BTC/USDT:USDT","timeframe":"1m"}}</code></pre></td>
+                            <td>Stop receiving updates for this pair.</td>
+                        </tr>
+                        <tr>
+                            <td><code>ping</code></td>
+                            <td><pre style="margin:0;padding:4px 8px;font-size:0.78rem;"><code>{{"action":"ping"}}</code></pre></td>
+                            <td>Keepalive ping. Server replies with <code>pong</code>.</td>
+                        </tr>
+                    </tbody>
+                </table>
+
+                <h6 class="mt-3">Server → Client Messages</h6>
+                <table class="table table-bordered table-sm">
+                    <thead class="table-light"><tr><th>Type</th><th>Description</th></tr></thead>
+                    <tbody>
+                        <tr><td><code>subscribed</code></td><td>Confirms subscription: <code>{{"type":"subscribed","symbol":"...","timeframe":"..."}}</code></td></tr>
+                        <tr><td><code>snapshot</code></td><td>Full recent candles on subscribe: <code>{{"type":"snapshot","symbol":"...","timeframe":"...","data":[...]}}</code></td></tr>
+                        <tr><td><code>kline</code></td><td>Incremental candle updates (last 2 candles): <code>{{"type":"kline","symbol":"...","timeframe":"...","data":[...]}}</code></td></tr>
+                        <tr><td><code>pong</code></td><td>Ping response.</td></tr>
+                        <tr><td><code>unsubscribed</code></td><td>Unsubscribe confirmation.</td></tr>
+                        <tr><td><code>error</code></td><td>Error message: <code>{{"type":"error","message":"..."}}</code></td></tr>
+                    </tbody>
+                </table>
+
+                <h6 class="mt-3">Python Client Example</h6>
+                <pre><code>import asyncio, json
+import websockets
+
+async def listen():
+    async with websockets.connect("ws://localhost:80/ws") as ws:
+        # Subscribe to BTC 1m candles
+        await ws.send(json.dumps({{
+            "action": "subscribe",
+            "symbol": "BTC/USDT:USDT",
+            "timeframe": "1m"
+        }}))
+        while True:
+            msg = json.loads(await ws.recv())
+            print(f"[{{msg['type']}}] {{msg.get('symbol','')}} {{msg.get('timeframe','')}}")
+            if msg['type'] == 'kline':
+                for candle in msg.get('data', []):
+                    print(f"  o={{candle['open']}} h={{candle['high']}} "
+                          f"l={{candle['low']}} c={{candle['close']}}")
+
+asyncio.run(listen())</code></pre>
+                <p class="small text-muted">Requires: <code>pip install websockets</code>. Subscribe to multiple symbol/timeframe pairs by sending additional subscribe messages.</p>
+            </div>
+        </div>
+    </div>
+
+    <!-- ==================== TIMEFRAMES ==================== -->
+    <div class="section" id="section-timeframes">
+
+        <div class="card">
+            <div class="card-header">Base & Derived Timeframes</div>
+            <div class="card-body">
+                <p class="small"><strong>Base</strong> timeframes are fetched directly from the exchange. <strong>Derived</strong> timeframes are synthesized on-the-fly by aggregating base candles — no extra API calls needed.</p>
+            </div>
+            <div class="card-body p-0">
+                <table class="table table-hover mb-0">
+                    <thead class="table-light"><tr><th>Base Timeframe</th><th>Configured</th><th>Derived Timeframes Synthesized</th></tr></thead>
+                    <tbody>{derived_rows}</tbody>
+                </table>
+            </div>
+        </div>
+
+        <div class="card">
+            <div class="card-header">Memory Limits</div>
+            <div class="card-body">
+                <p class="small">To bound memory usage, only a sliding window is held in RAM for high-frequency timeframes. Data beyond the window is served from CSV on disk lazily.</p>
+            </div>
+            <div class="card-body p-0">
+                <table class="table table-hover mb-0">
+                    <thead class="table-light"><tr><th>Timeframe</th><th>Max Candles in Memory</th></tr></thead>
+                    <tbody>{memory_rows}</tbody>
+                </table>
+            </div>
+        </div>
+
+        <div class="card">
+            <div class="card-header">CSV Storage</div>
+            <div class="card-body">
+                <p class="small">Data is persisted under <code>data/&lt;timeframe&gt;/</code> as one CSV file per symbol (e.g. <code>data/1m/BTC_USDT_USDT.csv</code>). Schema: <code>timestamp,datetime,open,high,low,close,volume</code>.</p>
+            </div>
+        </div>
+    </div>
+
+    <!-- ==================== CONFIGURATION ==================== -->
+    <div class="section" id="section-config">
+
+        <div class="card">
+            <div class="card-header">config.json</div>
+            <div class="card-body">
+                <pre><code>{config_example}</code></pre>
+            </div>
+            <div class="card-body p-0">
+                <table class="table table-hover mb-0">
+                    <thead class="table-light"><tr><th>Field</th><th>Type</th><th>Description</th></tr></thead>
+                    <tbody>
+                        <tr><td><code>exchange</code></td><td>string</td><td>CCXT exchange ID. Tested with <code>binance</code> (Binance Futures).</td></tr>
+                        <tr><td><code>symbols</code></td><td>string[]</td><td>Trading pairs in Binance Futures format: <code>BTC/USDT:USDT</code>. Also accepts comma-separated string.</td></tr>
+                        <tr><td><code>start_time</code></td><td>string</td><td>Global earliest backfill date, ISO 8601 UTC. Used as fallback for timeframes not listed in <code>start_time_per_tf</code>.</td></tr>
+                        <tr><td><code>start_time_per_tf</code></td><td>object</td><td><em>Optional.</em> Per-timeframe override. Keys are timeframe strings (e.g. <code>"1m"</code>), values are ISO 8601 UTC dates. Useful for limiting high-frequency data volume (e.g. 1m from recent date, longer timeframes from earlier).</td></tr>
+                        <tr><td><code>timeframes</code></td><td>string[]</td><td>Base timeframes to fetch. Supported: <code>1m</code>, <code>1h</code>, <code>1d</code>, <code>1w</code>.</td></tr>
+                        <tr><td><code>data_dir</code></td><td>string</td><td>Directory for CSV storage (created if missing).</td></tr>
+                    </tbody>
+                </table>
+            </div>
+        </div>
+
+        <div class="card">
+            <div class="card-header">Environment Variables</div>
+            <div class="card-body p-0">
+                <table class="table table-hover mb-0">
+                    <thead class="table-light"><tr><th>Variable</th><th>Default</th><th>Description</th></tr></thead>
+                    <tbody>
+                        <tr><td><code>CONFIG_PATH</code></td><td><code>config.json</code></td><td>Path to the JSON config file.</td></tr>
+                        <tr><td><code>UVICORN_HOST</code></td><td><code>0.0.0.0</code></td><td>Host to bind.</td></tr>
+                        <tr><td><code>UVICORN_PORT</code></td><td><code>80</code></td><td>Port to bind.</td></tr>
+                        <tr><td><code>UVICORN_LOG_LEVEL</code></td><td><code>info</code></td><td>Log level: <code>debug</code>, <code>info</code>, <code>warning</code>, <code>error</code>.</td></tr>
+                    </tbody>
+                </table>
+            </div>
+        </div>
+    </div>
+
+    <!-- ==================== EXAMPLES ==================== -->
+    <div class="section" id="section-examples">
+
+        <div class="card">
+            <div class="card-header">cURL — Fetch Last 500 5-Minute Candles</div>
+            <div class="card-body">
+                <pre><code>curl "http://localhost:80/api/candles?symbol=BTC/USDT:USDT&tf=5m&limit=500"</code></pre>
+            </div>
+        </div>
+
+        <div class="card">
+            <div class="card-header">cURL — Fetch Date Range</div>
+            <div class="card-body">
+                <pre><code># 4-hour candles for ETH, Jan–Mar 2024
+curl "http://localhost:80/api/candles?symbol=ETH/USDT:USDT&tf=4h&start=1704067200000&end=1711929600000"</code></pre>
+            </div>
+        </div>
+
+        <div class="card">
+            <div class="card-header">Python — requests</div>
+            <div class="card-body">
+                <pre><code>import requests
+
+# Check service is ready
+resp = requests.get("http://localhost:80/health")
+meta = resp.json()
+print(f"Ready: {{meta['ready']}}")
+
+# Fetch candles
+params = {{
+    "symbol": "BTC/USDT:USDT",
+    "tf": "1h",
+    "limit": 200,
+}}
+resp = requests.get("http://localhost:80/api/candles", params=params)
+candles = resp.json()
+for c in candles[:3]:
+    print(c["datetime"], c["open"], c["close"])</code></pre>
+            </div>
+        </div>
+
+        <div class="card">
+            <div class="card-header">JavaScript — WebSocket</div>
+            <div class="card-body">
+                <pre><code>const ws = new WebSocket("ws://localhost:80/ws");
+
+ws.onopen = () => {{
+  ws.send(JSON.stringify({{
+    action: "subscribe",
+    symbol: "BTC/USDT:USDT",
+    timeframe: "1m"
+  }}));
+}};
+
+ws.onmessage = (event) => {{
+  const msg = JSON.parse(event.data);
+  if (msg.type === "snapshot") {{
+    console.log("Initial data:", msg.data.length, "candles");
+  }} else if (msg.type === "kline") {{
+    console.log("Update:", msg.data);
+  }}
+}};
+
+// Keepalive every 30s
+setInterval(() => ws.send(JSON.stringify({{action: "ping"}})), 30000);</code></pre>
+            </div>
+        </div>
+
+        <div class="card">
+            <div class="card-header">JavaScript — fetch (browser)</div>
+            <div class="card-body">
+                <pre><code>const params = new URLSearchParams({{
+  symbol: "BTC/USDT:USDT",
+  tf: "5m",
+  limit: "500"
+}});
+const resp = await fetch(`/api/candles?${{params}}`);
+const candles = await resp.json();
+console.log(`Got ${{candles.length}} candles`);</code></pre>
+            </div>
+        </div>
+    </div>
+
+    <footer>
+        Data Provider v1.0.0 &middot; Chan Project &middot; Served from {provider.exchange_name}
+    </footer>
+</div>
+
+<script>
+// Simple tab switcher
+document.querySelectorAll('#manualTabs .nav-link').forEach(btn => {{
+    btn.addEventListener('click', () => {{
+        document.querySelectorAll('#manualTabs .nav-link').forEach(b => b.classList.remove('active'));
+        document.querySelectorAll('.section').forEach(s => s.classList.remove('active'));
+        btn.classList.add('active');
+        document.getElementById('section-' + btn.dataset.section).classList.add('active');
+    }});
+}});
+</script>
+</body>
+</html>"""
+
     @app.websocket("/ws")
     async def websocket_endpoint(ws: WebSocket):
         """WebSocket 实时 K 线推送。
@@ -937,7 +1597,7 @@ app = build_app()
 def main() -> None:
     """直接运行本模块时启动 uvicorn（监听 UVICORN_HOST / UVICORN_PORT）。"""
     host = os.getenv("UVICORN_HOST", "0.0.0.0")
-    port = int(os.getenv("UVICORN_PORT", "9009"))
+    port = int(os.getenv("UVICORN_PORT", "80"))
 
     uvicorn.run(app, host=host, port=port, log_level=os.getenv("UVICORN_LOG_LEVEL", "info"))