diff --git a/datasvc/Dockerfile b/datasvc/Dockerfile index c21cce6..6c10c63 100644 --- a/datasvc/Dockerfile +++ b/datasvc/Dockerfile @@ -9,10 +9,10 @@ COPY requirements.txt /app/requirements.txt RUN pip install -r /app/requirements.txt COPY app /app/app +COPY pairs.json /app/pairs.json ENV DATA_DIR=/data \ EXCHANGE=binance \ - SYMBOLS=BTC/USDT:USDT,ETH/USDT:USDT \ TIMEFRAMES=1m,1h,1d,1w,1M \ START_FROM=2022-01-01 \ POLL_FACTOR=0.5 diff --git a/datasvc/README.md b/datasvc/README.md index 58ac665..7656ec9 100644 --- a/datasvc/README.md +++ b/datasvc/README.md @@ -17,10 +17,11 @@ | --- | --- | --- | | `DATA_DIR` | 本地 Parquet 存储目录 | `/data` | | `EXCHANGE` | 交易所标识（目前支持 binance） | `binance` | -| `SYMBOLS` | 逗号分隔的交易对列表 | `BTC/USDT:USDT,ETH/USDT:USDT` | | `TIMEFRAMES` | 基础抓取周期，逗号分隔 | `1m,1h,1d,1w,1M` | | `START_FROM` | 首次启动回补的起始 UTC 时间（ISO 字符串或毫秒时间戳） | `2022-01-01` | | `POLL_FACTOR` | 拉取间隔因子，实际间隔 = 周期毫秒 × factor | `0.5` | +| `REST_MAX_CONCURRENCY` | REST 历史拉取并发数 | `4` | +| `VERIFY_MAX_CONCURRENCY` | 校验请求并发数 | `2` | | `BACKOFF_BASE / BACKOFF_MAX` | 异常重试的指数退避参数 | `2.0 / 30.0` | > 衍生周期列表由程序自动推导，无需手动写入 `TIMEFRAMES`。 @@ -40,7 +41,6 @@ docker compose down # 关闭 ### 2.2 本地运行（无 Docker） ```bash export DATA_DIR=./data -export SYMBOLS="BTC/USDT:USDT" export TIMEFRAMES="1m,1h,1d" cd /Users/jack/Project/freqtrade @@ -134,7 +134,8 @@ ws://localhost:8000/ws?symbol=ETH/USDT:USDT&tf=15m&since=1700000000000 确认该周期在 `TIMEFRAMES` 或自动聚合列表中；若是衍生周期，需要确保对应基础周期已在运行。 3. **如何新增交易对/周期** - 修改环境变量或 docker-compose 配置后，重启服务即可；Parquet 文件会按需生成。 + - 交易对：编辑 `pairs.json`，每行一个字符串，保存后重启服务。 + - 周期：修改 `TIMEFRAMES` 环境变量（Docker 或本地启动命令）后重启。 --- diff --git a/datasvc/app/main.py b/datasvc/app/main.py index a3d114a..3626008 100644 --- a/datasvc/app/main.py +++ b/datasvc/app/main.py @@ -5,11 +5,13 @@ 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 @@ -24,6 +26,7 @@ from .storage import ( read_candles, upsert_candles, get_last_timestamp, + read_candle_exact, ) @@ -33,9 +36,16 @@ logging.basicConfig( format="%(asctime)s %(levelname)s [%(name)s] %(message)s", ) logger = logging.getLogger("datasvc") +BASE_DIR = Path(__file__).resolve().parent.parent +DEFAULT_PAIRS_FILE = BASE_DIR / "pairs.json" RESAMPLE_AVAILABLE = resample_to_interval is not None RESAMPLE_WARNING_EMITTED = False +REST_MAX_CONCURRENCY = int(os.environ.get("REST_MAX_CONCURRENCY", "4")) +VERIFY_MAX_CONCURRENCY = int(os.environ.get("VERIFY_MAX_CONCURRENCY", "2")) +REST_FETCH_SEMAPHORE = asyncio.Semaphore(max(1, REST_MAX_CONCURRENCY)) +VERIFY_FETCH_SEMAPHORE = asyncio.Semaphore(max(1, VERIFY_MAX_CONCURRENCY)) + AGGREGATION_PLAN: Dict[str, List[str]] = { "1m": ["2m", "3m", "4m", "5m", "10m", "15m", "20m", "25m", "30m"], "1h": ["2h", "3h", "4h", "6h", "8h", "12h", "16h"], @@ -61,6 +71,37 @@ def _unique_preserve(values: List[str]) -> List[str]: 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 @@ -80,11 +121,24 @@ def timeframe_to_minutes(tf: str) -> Optional[int]: 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 = _split_env_list(os.environ.get("SYMBOLS", "BTC/USDT:USDT,ETH/USDT:USDT")) -if not SYMBOLS: - SYMBOLS = ["BTC/USDT:USDT"] +SYMBOLS = _load_symbols() _default_timeframes = ["1m", "1h", "1d", "1w", "1M"] requested_timeframes = _split_env_list(os.environ.get("TIMEFRAMES", ",".join(_default_timeframes))) @@ -104,6 +158,7 @@ START_FROM = os.environ.get("START_FROM", "2022-01-01") # 首次启动拉取起 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) @@ -194,6 +249,7 @@ class Hub: hub = Hub() fetch_tasks: List[asyncio.Task] = [] +verification_queues: Dict[Tuple[str, str], asyncio.Queue[int]] = {} def resample_and_store(symbol: str, base_timeframe: str, derived_timeframes: List[str]) -> List[Tuple[str, List[CandleRow]]]: @@ -227,14 +283,14 @@ def resample_and_store(symbol: str, base_timeframe: str, derived_timeframes: Lis 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.view("int64") // 1_000_000) + 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.view("int64") // 1_000_000) + derived_df["timestamp"] = (idx.astype("int64") // 1_000_000) if "timestamp" not in derived_df.columns: logger.warning( "聚合结果缺少 timestamp 列，已跳过", @@ -280,6 +336,7 @@ class FetchState: 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]: @@ -295,20 +352,210 @@ class FetchState: "last_candle_ts": self.last_candle_ts, "consecutive_errors": self.consecutive_errors, "last_error": self.last_error, + "last_verified_ts": self.last_verified_ts, } fetch_states: Dict[Tuple[str, str], FetchState] = {} +async def process_candles( + symbol: str, + timeframe: str, + candles: List[CandleRow], + derived_timeframes: List[str], + tf_ms: int, + schedule_verification: bool, +) -> None: + if not candles: + return + state_key = (symbol, timeframe) + upsert_candles(DATA_DIR, symbol, timeframe, candles) + derived_updates: List[Tuple[str, List[CandleRow]]] = [] + if derived_timeframes and RESAMPLE_AVAILABLE: + derived_updates = await asyncio.to_thread( + resample_and_store, + symbol, + timeframe, + derived_timeframes, + ) + for row in candles[-3:]: + payload = { + "topic": f"candles.{symbol}.{timeframe}", + "type": "upsert", + "data": { + "t": row[0], + "o": row[1], + "h": row[2], + "l": row[3], + "c": row[4], + "v": row[5], + }, + } + await hub.publish(symbol, timeframe, payload) + for target_tf, rows in derived_updates: + if not rows: + continue + for row in rows: + ts = int(row[0]) + o, h, l, c, v = map(float, row[1:]) + payload = { + "topic": f"candles.{symbol}.{target_tf}", + "type": "upsert", + "data": { + "t": ts, + "o": o, + "h": h, + "l": l, + "c": c, + "v": v, + }, + } + await hub.publish(symbol, target_tf, payload) + state = fetch_states.get(state_key) + if state: + state.last_fetch_at = datetime.utcnow() + state.last_candle_ts = candles[-1][0] + state.consecutive_errors = 0 + state.last_error = None + if schedule_verification: + queue = verification_queues.get(state_key) + if queue: + now_ms = int(datetime.utcnow().timestamp() * 1000) + latest_ts = candles[-1][0] + if now_ms - latest_ts <= 2 * tf_ms: + verify_ts = latest_ts - tf_ms + if verify_ts > 0 and (state.last_verified_ts is None or verify_ts > state.last_verified_ts): + try: + queue.put_nowait(verify_ts) + except asyncio.QueueFull: + logger.warning( + "验证队列已满，丢弃此次校验请求", + extra={"symbol": symbol, "timeframe": timeframe, "timestamp": verify_ts}, + ) + + +async def rest_catchup( + symbol: str, + timeframe: str, + derived_timeframes: List[str], + tf_ms: int, + start_since: int, +) -> None: + state_key = (symbol, timeframe) + state = fetch_states[state_key] + exchange = build_exchange() + since = start_since + backoff = 1.0 + logger.info("开始 REST 补齐历史", extra={"symbol": symbol, "timeframe": timeframe, "since": since}) + try: + while True: + now_ms = int(datetime.utcnow().timestamp() * 1000) + if since >= now_ms - tf_ms: + break + try: + async with REST_FETCH_SEMAPHORE: + candles = await exchange.fetch_ohlcv(symbol, timeframe, since=since, limit=1000) + except asyncio.CancelledError: + raise + except (ccxt.NetworkError, ccxt.ExchangeNotAvailable, ccxt.RequestTimeout) as exc: + logger.warning( + "历史补齐网络异常，准备重试", + extra={"symbol": symbol, "timeframe": timeframe, "error": str(exc)}, + ) + state.last_error = str(exc) + state.consecutive_errors += 1 + backoff = min(backoff * BACKOFF_BASE, BACKOFF_MAX) + await asyncio.sleep(backoff) + continue + except Exception as exc: + logger.exception( + "历史补齐发生异常，准备重试", + extra={"symbol": symbol, "timeframe": timeframe}, + ) + state.last_error = str(exc) + state.consecutive_errors += 1 + backoff = min(backoff * BACKOFF_BASE, BACKOFF_MAX) + await asyncio.sleep(backoff) + continue + if not candles: + break + await process_candles(symbol, timeframe, candles, derived_timeframes, tf_ms, schedule_verification=False) + since = candles[-1][0] + tf_ms + state.consecutive_errors = 0 + state.last_error = None + backoff = 1.0 + + now_ms = int(datetime.utcnow().timestamp() * 1000) + lag = now_ms - since + if lag > tf_ms * 10: + await asyncio.sleep(0.2) + else: + await asyncio.sleep(max(1.0, tf_ms * POLL_FACTOR / 1000.0)) + finally: + with suppress(Exception): + await exchange.close() + logger.info("REST 补齐完成", extra={"symbol": symbol, "timeframe": timeframe, "latest": state.last_candle_ts}) + + +async def stream_loop(symbol: str, timeframe: str, derived_timeframes: List[str], tf_ms: int): + state_key = (symbol, timeframe) + url = build_stream_url(symbol, timeframe) + while True: + try: + async with websockets.connect(url, ping_interval=20, ping_timeout=20) as ws: + logger.info("WebSocket 已连接", extra={"symbol": symbol, "timeframe": timeframe, "url": url}) + async for message in ws: + data = json.loads(message) + kline = data.get("k") + if not kline or not kline.get("x"): + continue + row: CandleRow = [ + int(kline["t"]), + float(kline["o"]), + float(kline["h"]), + float(kline["l"]), + float(kline["c"]), + float(kline["v"]), + ] + await process_candles(symbol, timeframe, [row], derived_timeframes, tf_ms, schedule_verification=True) + except asyncio.CancelledError: + logger.info("取消 WebSocket 任务", extra={"symbol": symbol, "timeframe": timeframe}) + raise + except Exception as exc: + logger.warning( + "WebSocket 连接异常，准备重连", + extra={"symbol": symbol, "timeframe": timeframe, "error": str(exc)}, + ) + state = fetch_states.get(state_key) + start_since = None + if state and state.last_candle_ts: + start_since = state.last_candle_ts + tf_ms + if start_since: + await rest_catchup(symbol, timeframe, derived_timeframes, tf_ms, start_since) + await asyncio.sleep(5.0) + + def build_exchange(): if EXCHANGE.lower() == "binance": - return ccxt_async.binance({"enableRateLimit": True}) + 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: @@ -318,117 +565,124 @@ async def fetch_loop(symbol: str, timeframe: str): ) RESAMPLE_WARNING_EMITTED = True - exchange = build_exchange() tf_ms = tf_to_ms(timeframe) - start_since = parse_start_from_ms(START_FROM) - last_ts = get_last_timestamp(DATA_DIR, symbol, timeframe) - since = max(start_since, (last_ts + tf_ms) if last_ts else start_since) - backoff = 1.0 state_key = (symbol, timeframe) + last_ts = get_last_timestamp(DATA_DIR, symbol, timeframe) fetch_states[state_key] = FetchState(symbol=symbol, timeframe=timeframe, last_candle_ts=last_ts) + queue = verification_queues.get(state_key) + if queue is None: + queue = asyncio.Queue(maxsize=500) + verification_queues[state_key] = queue + if last_ts is not None: + try: + queue.put_nowait(last_ts) + except asyncio.QueueFull: + logger.warning( + "重启后无法排入校验任务，队列已满", + extra={"symbol": symbol, "timeframe": timeframe, "timestamp": last_ts}, + ) + + start_since = parse_start_from_ms(START_FROM) + if last_ts is not None: + rewind_since = max(0, last_ts - tf_ms) + initial_since = max(start_since, rewind_since) + else: + initial_since = start_since logger.info("启动拉取任务", extra={"symbol": symbol, "timeframe": timeframe}) try: - while True: - try: - candles = await exchange.fetch_ohlcv(symbol, timeframe, since=since, limit=1000) - if candles: - upsert_candles(DATA_DIR, symbol, timeframe, candles) - derived_updates: List[Tuple[str, List[CandleRow]]] = [] - if derived_timeframes and RESAMPLE_AVAILABLE: - derived_updates = await asyncio.to_thread( - resample_and_store, - symbol, - timeframe, - derived_timeframes, - ) - for row in candles[-3:]: - payload = { - "topic": f"candles.{symbol}.{timeframe}", - "type": "upsert", - "data": { - "t": row[0], - "o": row[1], - "h": row[2], - "l": row[3], - "c": row[4], - "v": row[5], - }, - } - await hub.publish(symbol, timeframe, payload) - for target_tf, rows in derived_updates: - if not rows: - continue - for row in rows: - ts = int(row[0]) - o, h, l, c, v = map(float, row[1:]) - payload = { - "topic": f"candles.{symbol}.{target_tf}", - "type": "upsert", - "data": { - "t": ts, - "o": o, - "h": h, - "l": l, - "c": c, - "v": v, - }, - } - await hub.publish(symbol, target_tf, payload) - since = candles[-1][0] + tf_ms - backoff = 1.0 - state = fetch_states[state_key] - state.last_fetch_at = datetime.utcnow() - state.last_candle_ts = candles[-1][0] - state.consecutive_errors = 0 - state.last_error = None - await asyncio.sleep(max(1.0, tf_ms * POLL_FACTOR / 1000.0)) - except asyncio.CancelledError: - raise - except (ccxt.NetworkError, ccxt.ExchangeNotAvailable, ccxt.RequestTimeout) as exc: - logger.warning( - "网络异常，准备重试", - extra={"symbol": symbol, "timeframe": timeframe, "error": str(exc)}, - ) - state = fetch_states[state_key] - state.last_error = str(exc) - state.consecutive_errors += 1 - backoff = min(backoff * BACKOFF_BASE, BACKOFF_MAX) - await asyncio.sleep(backoff) - except Exception as exc: - logger.exception( - "抓取循环发生异常，重建客户端后重试", - extra={"symbol": symbol, "timeframe": timeframe}, - ) - state = fetch_states[state_key] - state.last_error = str(exc) - state.consecutive_errors += 1 - await asyncio.sleep(backoff) - with suppress(Exception): - await exchange.close() - exchange = build_exchange() - backoff = min(backoff * BACKOFF_BASE, BACKOFF_MAX) + await rest_catchup(symbol, timeframe, derived_timeframes, tf_ms, initial_since) + await stream_loop(symbol, timeframe, derived_timeframes, tf_ms) except asyncio.CancelledError: logger.info("取消拉取任务", extra={"symbol": symbol, "timeframe": timeframe}) state = fetch_states.get(state_key) if state: state.last_error = "cancelled" raise + finally: + logger.info("拉取任务退出", extra={"symbol": symbol, "timeframe": timeframe}) + + +async def verification_worker(symbol: str, timeframe: str): + state_key = (symbol, timeframe) + queue = verification_queues.get(state_key) + if queue is None: + return + exchange = build_exchange() + try: + while True: + verify_ts = await queue.get() + try: + state = fetch_states.get(state_key) + if state and state.last_verified_ts is not None and verify_ts <= state.last_verified_ts: + queue.task_done() + continue + async with VERIFY_FETCH_SEMAPHORE: + verification = await exchange.fetch_ohlcv(symbol, timeframe, since=verify_ts, limit=2) + target_rows = [row for row in verification if row and row[0] == verify_ts] + if not target_rows: + logger.warning( + "验证未获取到目标数据", + extra={"symbol": symbol, "timeframe": timeframe, "timestamp": verify_ts}, + ) + queue.task_done() + continue + candidate = target_rows[-1] + stored = read_candle_exact(DATA_DIR, symbol, timeframe, verify_ts) + needs_upsert = stored.empty + reason = "missing" + if not needs_upsert: + stored_row = stored.iloc[0] + open_diff = abs(float(stored_row["open"]) - float(candidate[1])) + high_diff = abs(float(stored_row["high"]) - float(candidate[2])) + low_diff = abs(float(stored_row["low"]) - float(candidate[3])) + close_diff = abs(float(stored_row["close"]) - float(candidate[4])) + volume_diff = abs(float(stored_row["volume"]) - float(candidate[5])) + if any(diff > 1e-9 for diff in (open_diff, high_diff, low_diff, close_diff, volume_diff)): + needs_upsert = True + reason = "mismatch" + if needs_upsert: + upsert_candles(DATA_DIR, symbol, timeframe, [candidate]) + logger.info( + "验证回补完成", + extra={"symbol": symbol, "timeframe": timeframe, "timestamp": verify_ts, "reason": reason}, + ) + state = fetch_states.get(state_key) + if state: + state.last_verified_ts = verify_ts + queue.task_done() + except asyncio.CancelledError: + queue.task_done() + raise + except Exception as exc: + logger.warning( + "验证请求失败", + extra={"symbol": symbol, "timeframe": timeframe, "timestamp": verify_ts, "error": str(exc)}, + ) + queue.task_done() + await asyncio.sleep(1.0) + except asyncio.CancelledError: + raise finally: with suppress(Exception): await exchange.close() - logger.info("拉取任务退出", extra={"symbol": symbol, "timeframe": timeframe}) + logger.info("验证任务退出", extra={"symbol": symbol, "timeframe": timeframe}) @app.on_event("startup") async def on_start(): ensure_storage(DATA_DIR) fetch_tasks.clear() + verification_queues.clear() for s in SYMBOLS: for tf in FETCH_TIMEFRAMES: + state_key = (s, tf) + verification_queues[state_key] = asyncio.Queue(maxsize=500) task = asyncio.create_task(fetch_loop(s, tf), name=f"fetch::{s}::{tf}") fetch_tasks.append(task) + verify_task = asyncio.create_task(verification_worker(s, tf), name=f"verify::{s}::{tf}") + fetch_tasks.append(verify_task) @app.on_event("shutdown") @@ -444,6 +698,7 @@ async def on_shutdown(): if isinstance(result, Exception) and not isinstance(result, asyncio.CancelledError): logger.warning("任务停止时出现异常：%s", result) fetch_tasks.clear() + verification_queues.clear() @app.get("/health") diff --git a/datasvc/app/storage.py b/datasvc/app/storage.py index a73e45f..fd231b4 100644 --- a/datasvc/app/storage.py +++ b/datasvc/app/storage.py @@ -3,6 +3,7 @@ import threading from typing import List, Optional import pandas as pd +import pyarrow.dataset as ds _lock = threading.Lock() @@ -55,3 +56,14 @@ def get_last_timestamp(base_dir: str, symbol: str, timeframe: str) -> Optional[i return int(df["timestamp"].iloc[-1]) +def read_candle_exact(base_dir: str, symbol: str, timeframe: str, timestamp: int) -> pd.DataFrame: + p = _path(base_dir, symbol, timeframe) + if not os.path.exists(p): + return pd.DataFrame(columns=["timestamp", "open", "high", "low", "close", "volume"]) + dataset = ds.dataset(p, format="parquet") + table = dataset.to_table(filter=ds.field("timestamp") == int(timestamp)) + if table.num_rows == 0: + return pd.DataFrame(columns=["timestamp", "open", "high", "low", "close", "volume"]) + return table.to_pandas() + + diff --git a/datasvc/docker-compose.yml b/datasvc/docker-compose.yml index 5dc5cea..ed58b2b 100644 --- a/datasvc/docker-compose.yml +++ b/datasvc/docker-compose.yml @@ -1,4 +1,3 @@ -version: "3.9" services: datasvc: build: . @@ -6,7 +5,6 @@ services: restart: unless-stopped environment: - EXCHANGE=binance - - SYMBOLS=BTC/USDT:USDT,ETH/USDT:USDT - TIMEFRAMES=1m,1h,1d,1w,1M - START_FROM=2022-01-01 - POLL_FACTOR=0.5 diff --git a/datasvc/pairs.json b/datasvc/pairs.json new file mode 100644 index 0000000..ebb2365 --- /dev/null +++ b/datasvc/pairs.json @@ -0,0 +1,22 @@ +[ + "BTC/USDT:USDT", + "ETH/USDT:USDT", + "SOL/USDT:USDT", + "WIF/USDT:USDT", + "1000PEPE/USDT:USDT", + "DOGS/USDT:USDT", + "ORDI/USDT:USDT", + "AAVE/USDT:USDT", + "REEF/USDT:USDT", + "1000SATS/USDT:USDT", + "SUI/USDT:USDT", + "1INCH/USDT:USDT", + "DOGE/USDT:USDT", + "TON/USDT:USDT", + "UNI/USDT:USDT", + "XRP/USDT:USDT", + "SUN/USDT:USDT", + "NOT/USDT:USDT", + "RARE/USDT:USDT", + "RDNT/USDT:USDT" +] diff --git a/datasvc/requirements.txt b/datasvc/requirements.txt index ba66cd8..346bb9a 100644 --- a/datasvc/requirements.txt +++ b/datasvc/requirements.txt @@ -5,4 +5,5 @@ pandas==2.2.2 pyarrow==16.1.0 orjson==3.10.3 technical==1.5.0 +websockets==12.0 diff --git a/web/app.py b/web/app.py index 5025552..dfde4ef 100644 --- a/web/app.py +++ b/web/app.py @@ -221,7 +221,7 @@ def detect_symbol_type(symbol): else: return 'unknown' -def get_kl_data(symbol, timeframe, limit=1000, start_time=None, end_time=None): +def get_kl_data(symbol, timeframe, limit=100000, start_time=None, end_time=None): """获取K线数据，支持加密货币和A股""" symbol_type = detect_symbol_type(symbol) @@ -232,7 +232,7 @@ def get_kl_data(symbol, timeframe, limit=1000, start_time=None, end_time=None): else: return None -def _get_crypto_kl_data_via_ccxt(symbol, timeframe, limit=1000, start_time=None, end_time=None): +def _get_crypto_kl_data_via_ccxt(symbol, timeframe, limit=100000, start_time=None, end_time=None): """获取加密货币K线数据，支持分页加载确保获取指定时间范围内的所有数据""" try: # 初始化参数 @@ -346,7 +346,7 @@ def _get_crypto_kl_data_via_ccxt(symbol, timeframe, limit=1000, start_time=None, return None -def get_crypto_kl_data(symbol, timeframe, limit=1000, start_time=None, end_time=None): +def get_crypto_kl_data(symbol, timeframe, limit=100000, start_time=None, end_time=None): """优先通过本地数据服务获取加密货币K线，失败时回退至交易所API。""" start_ms = _parse_time_input(start_time) end_ms = _parse_time_input(end_time) @@ -369,7 +369,7 @@ def get_crypto_kl_data(symbol, timeframe, limit=1000, start_time=None, end_time= return _get_crypto_kl_data_via_ccxt(symbol, timeframe, limit, start_time, end_time) -def get_a_stock_kl_data(symbol, timeframe, limit=1000, start_time=None, end_time=None): +def get_a_stock_kl_data(symbol, timeframe, limit=100000, start_time=None, end_time=None): """获取A股K线数据""" try: # 处理时间戳参数转换为日期字符串