diff --git a/data_provider/config.json b/data_provider/config.json index 5d442fd..90a7ff3 100644 --- a/data_provider/config.json +++ b/data_provider/config.json @@ -1,18 +1,31 @@ { "exchange": "binance", "symbols": [ - "BTC/USDT:USDT", - "ETH/USDT:USDT", + "BTC/USDT:USDT", + "ETH/USDT:USDT", "SOL/USDT:USDT", - "WIF/USDT:USDT", - "AAVE/USDT:USDT", - "SUI/USDT:USDT", - "1INCH/USDT:USDT", - "DOGE/USDT:USDT", - "UNI/USDT:USDT" + "XAU/USDT:USDT", + "XAG/USDT:USDT", + "SAGA/USDT:USDT", + "CL/USDT:USDT", + "ZEC/USDT:USDT", + "XRP/USDT:USDT", + "DOGE/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" } - diff --git a/data_provider/main.py b/data_provider/main.py index 7392878..321ca55 100644 --- a/data_provider/main.py +++ b/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, - }, - ) - history = self._fetch_history(symbol, timeframe, fetch_since) - merged = self._merge_candles(timeframe, existing, history) - with self._lock: - self.data.setdefault(symbol, {})[timeframe] = merged - self._write_to_disk(symbol, timeframe, merged) - self._ready.set() - logger.info("数据初始化完成") + fetch_since = self._get_start_time(timeframe) + try: + history = self._fetch_history(symbol, timeframe, fetch_since) + # 再次读取最新数据后合并，避免覆盖并发刷新线程写入的更新 + with self._lock: + 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._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"{base}{present}{', '.join(f'{d}' for d in derived_list)}" + + # 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"{tf}{span}" + + # 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""" + + + + + Data Provider — API Manual + + + + + + +
+
+

Data Provider v1.0.0

+

Cryptocurrency OHLCV data microservice — REST + WebSocket API for candlestick (K-line) data

+

Exchange: {provider.exchange_name} · Symbols: {len(symbols)} · Timeframes: {len(all_tfs)} · Status: {'Ready' if provider.is_ready() else 'Initializing…'}

+
+
+ +
+ + + + + +
+
+
What is Data Provider?
+
+

Data Provider 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.

+

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.

+
+
Key Features
+
+
+
    +
  • Fetches base OHLCV at 1m, 1h, 1d, 1w from Binance
    • +
  • Synthesizes {len(derived_tfs)} derived timeframes by resampling
    • +
  • Dual storage: in-memory cache + CSV persistence
    • +
  • Background refresh every {RECENT_FETCH_INTERVAL}s
    • +
+
+
+
    +
  • REST endpoint at /api/candles
    • +
  • WebSocket real-time push at /ws
    • +
  • Automatic disconnect recovery
    • +
  • Fully Dockerized, single JSON config file
    • +
+
+
+
+
+ +
+
Architecture
+
+ 
+┌───────────────────────┐
+│   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…)    │
+└────────────────────────┘
+
+
+ +
+
Endpoints at a Glance
+
+ + + + + + + + + + +
MethodPathDescription
GET/Service info & health summary
GET/healthStructured health check (JSON)
GET/timeframesList available base & derived timeframes
GET/api/candlesFetch OHLCV candles
WS/wsReal-time K-line streaming
GET/api-manualThis page
+
+
+ +
+
Data Model
+
+

Every candle object (REST response, WebSocket payload) has this structure:

+ + + + + + + + + + + +
FieldTypeExample
timestampint1704067200000
datetimestring"2024-01-01T00:00:00Z"
openfloat42314.0
highfloat44266.0
lowfloat42207.9
closefloat44230.2
volumefloat206424.144
+
+
+
+ + +
+ +
GET /health
+
+
+

Health check — use ready to determine if the service finished cold-start initialization.

+

Response 200 OK

+ 
{{
+  "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
+}}
+ +
+
+ +
GET /api/candles
+
+
+

The primary data endpoint. Returns OHLCV candlestick data for a given symbol and timeframe.

+

Query Parameters

+ + + + + + + + + +
ParamTypeRequiredDefaultDescription
symbolstringYes—Trading pair, e.g. BTC/USDT:USDT
tfstringNo1mTimeframe — any base or derived
startintNo—Start time, UTC milliseconds
endintNo—End time, UTC milliseconds
limitintNo—Max candles to return
+

Response 200 OK — array of candle objects

+ 
[
+  {{
+    "timestamp": 1704067200000,
+    "datetime": "2024-01-01T00:00:00Z",
+    "open": 42314.0,
+    "high": 44266.0,
+    "low": 42207.9,
+    "close": 44230.2,
+    "volume": 206424.144
+  }}
+]
+

At least one of start/end or limit should be provided. If none given, all in-memory candles are returned (subject to memory limits).

+
+
+ +
GET /timeframes
+
+
+

Lists all available timeframes: base (fetched from exchange) and derived (synthesized by resampling).

+ 
{{
+  "base_timeframes": ["1m","1h","1d","1w"],
+  "derived_timeframes": ["2m","3m","4m","5m",...],
+  "timeframes": ["1m","2m","3m",...]
+}}
+
+
+ +
GET /
+
+
+

Root endpoint — returns service name, exchange, symbols, timeframes, and ready status. Same data as /health.

+
+
+
+ + +
+ +
+
WS /ws — Real-time K-line Streaming
+
+

Connect to receive live incremental candle updates. Subscribe to specific symbol/timeframe pairs; receive a full snapshot followed by real-time push.

+ +
Connection URL
+ 
ws://localhost:80/ws
+

Replace localhost:80 with the server address if connecting remotely.

+ +
Client → Server Messages
+ + + + + + + + + + + + + + + + + + + +
ActionPayloadDescription
subscribe
{{"action":"subscribe","symbol":"BTC/USDT:USDT","timeframe":"1m"}}
Subscribe to a symbol/timeframe. Server replies with subscribed, then a snapshot of recent candles.
unsubscribe
{{"action":"unsubscribe","symbol":"BTC/USDT:USDT","timeframe":"1m"}}
Stop receiving updates for this pair.
ping
{{"action":"ping"}}
Keepalive ping. Server replies with pong.
+ +
Server → Client Messages
+ + + + + + + + + + +
TypeDescription
subscribedConfirms subscription: {{"type":"subscribed","symbol":"...","timeframe":"..."}}
snapshotFull recent candles on subscribe: {{"type":"snapshot","symbol":"...","timeframe":"...","data":[...]}}
klineIncremental candle updates (last 2 candles): {{"type":"kline","symbol":"...","timeframe":"...","data":[...]}}
pongPing response.
unsubscribedUnsubscribe confirmation.
errorError message: {{"type":"error","message":"..."}}
+ +
Python Client Example
+ 
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())
+

Requires: pip install websockets. Subscribe to multiple symbol/timeframe pairs by sending additional subscribe messages.

+
+
+
+ + +
+ +
+
Base & Derived Timeframes
+
+

Base timeframes are fetched directly from the exchange. Derived timeframes are synthesized on-the-fly by aggregating base candles — no extra API calls needed.

+
+
+ + + {derived_rows} +
Base TimeframeConfiguredDerived Timeframes Synthesized
+
+
+ +
+
Memory Limits
+
+

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.

+
+
+ + + {memory_rows} +
TimeframeMax Candles in Memory
+
+
+ +
+
CSV Storage
+
+

Data is persisted under data/<timeframe>/ as one CSV file per symbol (e.g. data/1m/BTC_USDT_USDT.csv). Schema: timestamp,datetime,open,high,low,close,volume.

+
+
+
+ + +
+ +
+
config.json
+
+ 
{config_example}
+
+
+ + + + + + + + + + +
FieldTypeDescription
exchangestringCCXT exchange ID. Tested with binance (Binance Futures).
symbolsstring[]Trading pairs in Binance Futures format: BTC/USDT:USDT. Also accepts comma-separated string.
start_timestringGlobal earliest backfill date, ISO 8601 UTC. Used as fallback for timeframes not listed in start_time_per_tf.
start_time_per_tfobjectOptional. Per-timeframe override. Keys are timeframe strings (e.g. "1m"), values are ISO 8601 UTC dates. Useful for limiting high-frequency data volume (e.g. 1m from recent date, longer timeframes from earlier).
timeframesstring[]Base timeframes to fetch. Supported: 1m, 1h, 1d, 1w.
data_dirstringDirectory for CSV storage (created if missing).
+
+
+ +
+
Environment Variables
+
+ + + + + + + + +
VariableDefaultDescription
CONFIG_PATHconfig.jsonPath to the JSON config file.
UVICORN_HOST0.0.0.0Host to bind.
UVICORN_PORT80Port to bind.
UVICORN_LOG_LEVELinfoLog level: debug, info, warning, error.
+
+
+
+ + +
+ +
+
cURL — Fetch Last 500 5-Minute Candles
+
+ 
curl "http://localhost:80/api/candles?symbol=BTC/USDT:USDT&tf=5m&limit=500"
+
+
+ +
+
cURL — Fetch Date Range
+
+ 
# 4-hour candles for ETH, Jan–Mar 2024
+curl "http://localhost:80/api/candles?symbol=ETH/USDT:USDT&tf=4h&start=1704067200000&end=1711929600000"
+
+
+ +
+
Python — requests
+
+ 
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"])
+
+
+ +
+
JavaScript — WebSocket
+
+ 
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);
+
+
+ +
+
JavaScript — fetch (browser)
+
+ 
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`);
+
+
+
+ + +
+ + + +""" + @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"))