diff --git a/.DS_Store b/.DS_Store index fd04eab..ca5e059 100644 Binary files a/.DS_Store and b/.DS_Store differ diff --git a/.gitignore b/.gitignore index 8e52078..448c5fd 100644 --- a/.gitignore +++ b/.gitignore @@ -30,3 +30,5 @@ feature_meta 交易记录/~$交易规则.docx /datasvc/data .DS_Store +.DS_Store +/data_provider/data diff --git a/data_provider/Dockerfile b/data_provider/Dockerfile new file mode 100644 index 0000000..d07a269 --- /dev/null +++ b/data_provider/Dockerfile @@ -0,0 +1,20 @@ +FROM python:3.11-slim + +ENV PYTHONUNBUFFERED=1 \ + PIP_NO_CACHE_DIR=1 + +WORKDIR /app + +COPY requirements.txt /app/requirements.txt +RUN pip install --no-cache-dir -r requirements.txt + +COPY . /app + +ENV CONFIG_PATH=/app/config.json \ + UVICORN_HOST=0.0.0.0 \ + UVICORN_PORT=9009 + +EXPOSE 9009 + +CMD ["python", "-m", "main"] + diff --git a/data_provider/config.json b/data_provider/config.json new file mode 100644 index 0000000..b1a4ce4 --- /dev/null +++ b/data_provider/config.json @@ -0,0 +1,24 @@ +{ + "exchange": "binance", + "symbols": [ + "BTC/USDT:USDT", + "ETH/USDT:USDT", + "SOL/USDT:USDT", + "WIF/USDT:USDT", + "LINK/USDT:USDT", + "ORDI/USDT:USDT", + "AAVE/USDT:USDT", + "REEF/USDT:USDT", + "1INCH/USDT:USDT", + "DOGE/USDT:USDT", + "TON/USDT:USDT", + "UNI/USDT:USDT", + "XRP/USDT:USDT", + "SUN/USDT:USDT", + "SUI/USDT:USDT" + ], + "start_time": "2025-01-01T00:00:00Z", + "timeframes": ["1m", "1h", "1d", "1w"], + "data_dir": "./data" +} + diff --git a/data_provider/docker-compose.yml b/data_provider/docker-compose.yml new file mode 100644 index 0000000..4906e8f --- /dev/null +++ b/data_provider/docker-compose.yml @@ -0,0 +1,15 @@ +services: + data_provider: + build: . + container_name: data-provider + restart: unless-stopped + environment: + CONFIG_PATH: /app/config.json + UVICORN_HOST: 0.0.0.0 + UVICORN_PORT: "9009" + volumes: + - ./config.json:/app/config.json:ro + - ./data:/app/data + ports: + - "9009:9009" + diff --git a/data_provider/main.py b/data_provider/main.py new file mode 100644 index 0000000..be7e64e --- /dev/null +++ b/data_provider/main.py @@ -0,0 +1,583 @@ +import asyncio +import csv +import json +import logging +import os +import threading +import time +from contextlib import asynccontextmanager +from datetime import datetime, timezone +from pathlib import Path +from typing import Dict, Iterable, List, Optional + +import ccxt # type: ignore +import pandas as pd # type: ignore +from fastapi import FastAPI, HTTPException, Query +from fastapi.middleware.cors import CORSMiddleware +import uvicorn +from technical.util import resample_to_interval + +# docker compose logs --tail=200 +# docker compose down && docker compose build --no-cache && docker compose up -d + +TIMEFRAME_ORDER = ["1m", "1h", "1d", "1w"] +ALLOWED_TIMEFRAMES = set(TIMEFRAME_ORDER) +TIMEFRAME_TO_MS: Dict[str, int] = { + "1m": 60_000, + "1h": 3_600_000, + "1d": 86_400_000, + "1w": 604_800_000, +} +DERIVED_TIMEFRAME_PLAN: Dict[str, List[str]] = { + "1m": ["2m", "3m", "4m", "5m", "10m", "15m", "20m", "25m", "30m"], + "1h": ["2h", "3h", "4h", "6h", "8h", "12h", "16h"], + "1d": ["2d", "3d", "4d", "5d", "6d"], + "1w": ["2w"], +} +CSV_FIELDNAMES = ["timestamp", "datetime", "open", "high", "low", "close", "volume"] +DEFAULT_LIMIT = 500 +RECENT_CANDLE_LIMIT = 10 +RECENT_FETCH_INTERVAL = 5 +PERSIST_INTERVAL = 600 + + +logger = logging.getLogger("data_provider") +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", +) + + +def to_utc_iso(timestamp_ms: int) -> str: + dt = datetime.fromtimestamp(timestamp_ms / 1000, tz=timezone.utc) + return dt.isoformat().replace("+00:00", "Z") + + +def parse_timestamp(value: Optional[object]) -> Optional[int]: + if value is None: + return None + if isinstance(value, (int, float)): + return int(value) + if isinstance(value, str): + text = value.strip() + if not text: + return None + if text.isdigit(): + return int(text) + if text.endswith("Z"): + text = text[:-1] + "+00:00" + try: + dt = datetime.fromisoformat(text) + except ValueError as exc: # pragma: no cover - informative logging + raise ValueError(f"无法解析时间字符串: {value}") from exc + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + else: + dt = dt.astimezone(timezone.utc) + return int(dt.timestamp() * 1000) + raise ValueError(f"不支持的时间格式: {value}") + + +def candle_to_dict(candle: Iterable[float]) -> Dict[str, float]: + ts = int(candle[0]) + return { + "timestamp": ts, + "datetime": to_utc_iso(ts), + "open": float(candle[1]), + "high": float(candle[2]), + "low": float(candle[3]), + "close": float(candle[4]), + "volume": float(candle[5]), + } + + +def timeframe_to_minutes(tf: str) -> Optional[int]: + if not tf: + return None + unit = tf[-1] + try: + value = int(tf[:-1]) + except ValueError: + return None + multiplier = { + "m": 1, + "h": 60, + "d": 1_440, + "w": 10_080, + }.get(unit) + if multiplier is None: + return None + return value * multiplier + + +class DataProvider: + def __init__(self, config_path: Path) -> None: + self.config_path = config_path + self.config = self._load_config() + self.exchange_name: str = self.config["exchange"] + self.symbols: List[str] = self._load_symbols(self.config) + self.timeframes: List[str] = self._validate_timeframes(self.config.get("timeframes")) + self.data_dir = Path(self.config.get("data_dir", "./data")).expanduser() + start = parse_timestamp(self.config.get("start_time")) + if start is None: + raise ValueError("配置文件必须包含 start_time 字段") + self.start_time_ms: int = 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 + } + self.derived_map: Dict[str, str] = {} + for base_tf in self.timeframes: + for derived_tf in DERIVED_TIMEFRAME_PLAN.get(base_tf, []): + self.derived_map.setdefault(derived_tf, base_tf) + derived_order: List[str] = [] + for base_tf in TIMEFRAME_ORDER: + if base_tf not in self.timeframes: + continue + for derived_tf in DERIVED_TIMEFRAME_PLAN.get(base_tf, []): + if derived_tf in self.derived_map and derived_tf not in derived_order: + derived_order.append(derived_tf) + self.available_timeframes: List[str] = list(self.timeframes) + derived_order + self._lock = threading.RLock() + self._ready = threading.Event() + self._stop_event = threading.Event() + self._fetch_thread: Optional[threading.Thread] = None + self._persist_thread: Optional[threading.Thread] = None + + def _load_config(self) -> Dict[str, object]: + if not self.config_path.exists(): + raise FileNotFoundError(f"未找到配置文件: {self.config_path}") + with self.config_path.open("r", encoding="utf-8") as fp: + return json.load(fp) + + def _load_symbols(self, config: Dict[str, object]) -> List[str]: + raw_symbols: List[str] = [] + symbols_value = config.get("symbols") + if isinstance(symbols_value, list): + raw_symbols = [str(item).strip() for item in symbols_value if isinstance(item, str) and item.strip()] + elif isinstance(symbols_value, str) and symbols_value.strip(): + raw_symbols = [item.strip() for item in symbols_value.split(",") if item.strip()] + symbol_single = config.get("symbol") + if not raw_symbols and isinstance(symbol_single, str) and symbol_single.strip(): + raw_symbols = [symbol_single.strip()] + if not raw_symbols: + raise ValueError("配置文件必须提供 symbols（列表或逗号分隔字符串）或 symbol 字段") + unique: List[str] = [] + for item in raw_symbols: + if item not in unique: + unique.append(item) + return unique + + def _validate_timeframes(self, configured: Optional[Iterable[str]]) -> List[str]: + if not configured: + return list(TIMEFRAME_ORDER) + invalid = [tf for tf in configured if tf not in ALLOWED_TIMEFRAMES] + if invalid: + raise ValueError(f"不支持的时间周期: {invalid}. 允许值: {sorted(ALLOWED_TIMEFRAMES)}") + unique = [] + seen = set() + for tf in TIMEFRAME_ORDER: + if tf in configured and tf not in seen: + unique.append(tf) + seen.add(tf) + for tf in configured: + if tf not in seen: + unique.append(tf) + seen.add(tf) + return unique + + def _init_exchange(self): + if not hasattr(ccxt, self.exchange_name): + raise ValueError(f"不支持的交易所: {self.exchange_name}") + exchange_class = getattr(ccxt, self.exchange_name) + exchange = exchange_class({"enableRateLimit": True}) + if exchange.id == "binance": + exchange.options.setdefault("defaultType", "future") + exchange.load_markets() + logger.info("已初始化交易所 %s", exchange.id) + return exchange + + def _data_file_path(self, symbol: str, timeframe: str) -> Path: + symbol_safe = symbol.replace("/", "_").replace(":", "_") + return self.data_dir / timeframe / f"{self.exchange.id}_{symbol_safe}_{timeframe}.csv" + + def _load_local(self, symbol: str, timeframe: str) -> List[Dict[str, float]]: + path = self._data_file_path(symbol, timeframe) + if not path.exists(): + return [] + loaded: List[Dict[str, float]] = [] + with path.open("r", encoding="utf-8", newline="") as fp: + reader = csv.DictReader(fp) + for row in reader: + try: + loaded.append( + { + "timestamp": int(row["timestamp"]), + "datetime": row.get("datetime") or to_utc_iso(int(row["timestamp"])), + "open": float(row["open"]), + "high": float(row["high"]), + "low": float(row["low"]), + "close": float(row["close"]), + "volume": float(row["volume"]), + } + ) + except (KeyError, ValueError): + logger.warning("忽略损坏的行: %s", row) + loaded.sort(key=lambda item: item["timestamp"]) + logger.info("交易对 %s 时间周期 %s 加载本地K线数量: %s", symbol, timeframe, len(loaded)) + return loaded + + def _merge_candles( + self, + timeframe: str, + base: List[Dict[str, float]], + new_candles: Iterable[Iterable[float]], + ) -> List[Dict[str, float]]: + merged = {entry["timestamp"]: entry for entry in base} + for candle in new_candles: + entry = candle_to_dict(candle) + merged[entry["timestamp"]] = entry + ordered = list(sorted(merged.values(), key=lambda item: item["timestamp"])) + logger.debug("时间周期 %s 合并后K线数量: %s", timeframe, len(ordered)) + return ordered + + def _write_to_disk(self, symbol: str, timeframe: str, data: List[Dict[str, float]]) -> None: + path = self._data_file_path(symbol, timeframe) + path.parent.mkdir(parents=True, exist_ok=True) + tmp_path = path.with_suffix(path.suffix + ".tmp") + try: + with tmp_path.open("w", encoding="utf-8", newline="") as fp: + writer = csv.DictWriter(fp, fieldnames=CSV_FIELDNAMES) + writer.writeheader() + writer.writerows(data) + os.replace(tmp_path, path) + finally: + if tmp_path.exists(): + try: + tmp_path.unlink() + except OSError: + pass + logger.info("交易对 %s 时间周期 %s 已写入磁盘 (%s 根K线)", symbol, timeframe, len(data)) + + def _fetch_history(self, symbol: str, timeframe: str, since_ms: int) -> List[List[float]]: + results: List[List[float]] = [] + limit = 1500 + now_ms = self.exchange.milliseconds() + tf_ms = TIMEFRAME_TO_MS[timeframe] + fetch_since = since_ms + max_rounds = 5000 + rounds = 0 + while fetch_since < now_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 + 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: + 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: + logger.info("开始初始化数据提供商") + for symbol in self.symbols: + for timeframe in self.timeframes: + existing = self._load_local(symbol, timeframe) + tf_ms = TIMEFRAME_TO_MS[timeframe] + last_ts = existing[-1]["timestamp"] if existing else None + if last_ts is not None: + if len(existing) >= 2: + fetch_since = existing[-2]["timestamp"] + else: + fetch_since = max(0, last_ts - tf_ms) + 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("数据初始化完成") + + def resample_df(self, df: pd.DataFrame, interval: int) -> pd.DataFrame: + return resample_to_interval(df, interval) + + def start_background_workers(self) -> None: + if self._fetch_thread and self._fetch_thread.is_alive(): + return + self._stop_event.clear() + 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._fetch_thread.start() + self._persist_thread.start() + logger.info("后台线程已启动") + + def stop(self) -> None: + self._stop_event.set() + if self._fetch_thread: + self._fetch_thread.join(timeout=5) + if self._persist_thread: + self._persist_thread.join(timeout=5) + logger.info("数据提供商已停止") + + def _refresh_loop(self) -> None: + while not self._stop_event.is_set(): + for symbol in self.symbols: + for timeframe in self.timeframes: + try: + candles = self.exchange.fetch_ohlcv( + symbol, + timeframe=timeframe, + limit=RECENT_CANDLE_LIMIT, + ) + except ccxt.BaseError as exc: + logger.error("更新最新K线失败 (%s %s): %s", symbol, timeframe, exc) + time.sleep(2) + continue + if not candles: + continue + with self._lock: + current = self.data.setdefault(symbol, {}).get(timeframe, []) + merged = self._merge_candles(timeframe, current, candles) + self.data[symbol][timeframe] = merged + if self._stop_event.wait(RECENT_FETCH_INTERVAL): + break + + def _persist_loop(self) -> None: + while not self._stop_event.wait(PERSIST_INTERVAL): + self._persist_all() + + def _persist_all(self) -> None: + if not self._ready.is_set(): + return + with self._lock: + snapshot = { + symbol: {tf: list(data) for tf, data in tf_map.items()} + for symbol, tf_map in self.data.items() + } + for symbol, tf_map in snapshot.items(): + for timeframe, data in tf_map.items(): + self._write_to_disk(symbol, timeframe, data) + + def is_ready(self) -> bool: + return self._ready.is_set() + + def wait_ready(self, timeout: Optional[float] = None) -> bool: + return self._ready.wait(timeout) + + def get_available_timeframes(self) -> List[str]: + return list(self.available_timeframes) + + def get_derived_timeframes(self) -> List[str]: + return list(self.derived_map.keys()) + + def _get_base_klines( + self, + symbol: str, + timeframe: str, + start_ms: Optional[int], + end_ms: Optional[int], + limit: Optional[int], + ) -> List[Dict[str, float]]: + with self._lock: + candles = list(self.data.get(symbol, {}).get(timeframe, [])) + if start_ms is not None: + candles = [row for row in candles if row["timestamp"] >= start_ms] + if end_ms is not None: + candles = [row for row in candles if row["timestamp"] <= end_ms] + if limit: + candles = candles[-limit:] + return candles + + def get_klines( + self, + symbol: str, + timeframe: str, + start_time: Optional[object] = None, + end_time: Optional[object] = None, + limit: Optional[int] = None, + ) -> List[Dict[str, float]]: + if symbol not in self.symbols: + raise HTTPException(status_code=404, detail=f"symbol {symbol} 不可用") + self.wait_ready() + start_ms = parse_timestamp(start_time) + end_ms = parse_timestamp(end_time) + if timeframe in self.timeframes: + return self._get_base_klines(symbol, timeframe, start_ms, end_ms, limit) + base_tf = self.derived_map.get(timeframe) + if not base_tf: + raise HTTPException(status_code=404, detail=f"{symbol} 时间周期 {timeframe} 不可用") + target_minutes = timeframe_to_minutes(timeframe) + if target_minutes is None: + raise HTTPException(status_code=400, detail=f"不支持的时间周期: {timeframe}") + target_ms = target_minutes * 60_000 + adjusted_start = None if start_ms is None else max(0, start_ms - target_ms) + base_candles = self._get_base_klines(symbol, base_tf, adjusted_start, end_ms, None) + if not base_candles: + return [] + df = pd.DataFrame(base_candles) + if df.empty: + return [] + df = df.drop_duplicates(subset=["timestamp"], keep="last").sort_values("timestamp") + df["date"] = pd.to_datetime(df["timestamp"], unit="ms", utc=True) + resampled = self.resample_df(df, target_minutes) + if resampled is None or resampled.empty: + return [] + if "timestamp" in resampled.columns: + resampled_df = resampled.copy() + else: + resampled_df = resampled.copy() + if "date" in resampled_df.columns: + dates = pd.to_datetime(resampled_df["date"], utc=True, errors="coerce") + resampled_df["timestamp"] = (dates.view("int64") // 1_000_000).astype("int64") + elif isinstance(resampled_df.index, pd.DatetimeIndex): + idx = resampled_df.index + if idx.tz is None: + idx = idx.tz_localize("UTC") + else: + idx = idx.tz_convert("UTC") + resampled_df["timestamp"] = (idx.view("int64") // 1_000_000).astype("int64") + else: + raise HTTPException(status_code=500, detail=f"聚合结果缺少 timestamp 列 ({timeframe})") + resampled_df = resampled_df.dropna(subset=["timestamp"]).sort_values("timestamp") + if start_ms is not None: + resampled_df = resampled_df[resampled_df["timestamp"] >= start_ms] + if end_ms is not None: + resampled_df = resampled_df[resampled_df["timestamp"] <= end_ms] + if resampled_df.empty: + return [] + resampled_df["datetime"] = resampled_df["timestamp"].apply(to_utc_iso) + for column in ["open", "high", "low", "close", "volume"]: + if column not in resampled_df.columns: + resampled_df[column] = 0.0 + resampled_df = resampled_df[["timestamp", "datetime", "open", "high", "low", "close", "volume"]] + result = resampled_df.to_dict("records") + if limit: + result = result[-limit:] + logger.debug( + "衍生周期返回", + extra={ + "symbol": symbol, + "timeframe": timeframe, + "base_timeframe": base_tf, + "count": len(result), + }, + ) + return result + + +def create_app(provider: DataProvider) -> FastAPI: + @asynccontextmanager + async def lifespan(app: FastAPI): + loop = asyncio.get_running_loop() + await loop.run_in_executor(None, provider.initialize) + provider.start_background_workers() + try: + yield + finally: + provider.stop() + + app = FastAPI(title="Chan 数据提供商", version="1.0.0", lifespan=lifespan) + + app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], + ) + + @app.get("/health") + async def health() -> Dict[str, object]: + return { + "status": "ok", + "exchange": provider.exchange_name, + "symbols": provider.symbols, + "base_timeframes": provider.timeframes, + "derived_timeframes": provider.get_derived_timeframes(), + "timeframes": provider.get_available_timeframes(), + "ready": provider.is_ready(), + } + + @app.get("/timeframes") + async def list_timeframes() -> Dict[str, List[str]]: + provider.wait_ready() + return { + "base_timeframes": provider.timeframes, + "derived_timeframes": provider.get_derived_timeframes(), + "timeframes": provider.get_available_timeframes(), + } + + @app.get("/api/candles") + async def api_candles( + symbol: str = Query(..., description="如 BTC/USDT"), + tf: str = Query("1m", description="时间周期"), + start: Optional[int] = Query(None, description="开始时间戳(ms)"), + end: Optional[int] = Query(None, description="结束时间戳(ms)"), + limit: Optional[int] = Query(None, description="可选，限制返回数量"), + ): + data = provider.get_klines(symbol=symbol, timeframe=tf, start_time=start, end_time=end, limit=limit) + return data + + @app.get("/") + async def root() -> Dict[str, object]: + return { + "service": "Data Provider", + "exchange": provider.exchange_name, + "symbols": provider.symbols, + "base_timeframes": provider.timeframes, + "derived_timeframes": provider.get_derived_timeframes(), + "timeframes": provider.get_available_timeframes(), + "ready": provider.is_ready(), + } + + return app + + +def build_app() -> FastAPI: + config_path = Path(os.getenv("CONFIG_PATH", "config.json")) + provider = DataProvider(config_path) + return create_app(provider) + + +app = build_app() + + +def main() -> None: + host = os.getenv("UVICORN_HOST", "0.0.0.0") + port = int(os.getenv("UVICORN_PORT", "9009")) + + uvicorn.run(app, host=host, port=port, log_level=os.getenv("UVICORN_LOG_LEVEL", "info")) + + +if __name__ == "__main__": + main() + diff --git a/data_provider/requirements.txt b/data_provider/requirements.txt new file mode 100644 index 0000000..5e8afd5 --- /dev/null +++ b/data_provider/requirements.txt @@ -0,0 +1,5 @@ +ccxt>=4.0.0,<5.0.0 +fastapi>=0.110.0,<1.0.0 +uvicorn[standard]>=0.23.0,<1.0.0 +pandas>=2.0.0,<3.0.0 +technical==1.5.0 diff --git a/datasvc/Dockerfile b/datasvc/Dockerfile index 7307616..e0cdf94 100644 --- a/datasvc/Dockerfile +++ b/datasvc/Dockerfile @@ -14,7 +14,7 @@ COPY pairs.json /app/pairs.json ENV DATA_DIR=/data \ EXCHANGE=binance \ TIMEFRAMES=1m,1h,1d,1w,1M \ - START_FROM=2025-09-01 \ + START_FROM=2025-01-01 \ POLL_FACTOR=0.5 VOLUME ["/data"] diff --git a/datasvc/app/main.py b/datasvc/app/main.py index d165777..f8b17ba 100644 --- a/datasvc/app/main.py +++ b/datasvc/app/main.py @@ -22,10 +22,8 @@ from fastapi.middleware.cors import CORSMiddleware # docker compose down && docker compose build --no-cache && docker compose up -d # docker compose down && docker compose build && docker compose up -d -try: - from technical.util import resample_to_interval # type: ignore -except ImportError: # pragma: no cover - 环境缺失依赖时自动降级 - resample_to_interval = None # type: ignore +from technical.util import resample_to_interval # type: ignore + from .storage import candle_path, ensure_storage, read_candles, write_candles_snapshot @@ -450,7 +448,7 @@ DERIVED_TIMEFRAMES = [tf for tf in AVAILABLE_TIMEFRAMES if tf not in FETCH_TIMEF AGGREGATION_TARGETS = {tf: AGGREGATION_PLAN.get(tf, []) for tf in FETCH_TIMEFRAMES} BASE_TIMEFRAMES = set(FETCH_TIMEFRAMES) -START_FROM = os.environ.get("START_FROM", "2025-09-01") # 首次启动拉取起始日期（UTC） +START_FROM = os.environ.get("START_FROM", "2025-01-01") # 首次启动拉取起始日期（UTC） POLL_FACTOR = float(os.environ.get("POLL_FACTOR", "0.5")) # 轮询间隔 = tf_ms * factor BACKOFF_BASE = float(os.environ.get("BACKOFF_BASE", "2.0")) BACKOFF_MAX = float(os.environ.get("BACKOFF_MAX", "30.0")) @@ -513,7 +511,7 @@ def parse_start_from_ms(val: str) -> int: dt = datetime.fromisoformat(val) # 允许 '2022-01-01' 或 '2022-01-01T00:00:00' except Exception: # 回退到固定日期 - dt = datetime(2025, 9, 1) + dt = datetime(2025, 1, 1) return int(dt.timestamp() * 1000) diff --git a/datasvc/docker-compose.yml b/datasvc/docker-compose.yml index 3319b58..2c3cc02 100644 --- a/datasvc/docker-compose.yml +++ b/datasvc/docker-compose.yml @@ -6,7 +6,7 @@ services: environment: - EXCHANGE=binance - TIMEFRAMES=1m,1h,1d,1w,1M - - START_FROM=2025-09-01 + - START_FROM=2025-01-01 - POLL_FACTOR=0.5 - DATA_DIR=/data - TZ=Asia/Shanghai diff --git a/web/app.py b/web/app.py index dfde4ef..207fe8e 100644 --- a/web/app.py +++ b/web/app.py @@ -47,7 +47,7 @@ china_stock = ChinaStockData() logger = logging.getLogger(__name__) -DATA_SERVICE_URL = os.environ.get("DATA_SERVICE_URL", os.environ.get("DATASVC_URL", "http://127.0.0.1:9000")) +DATA_SERVICE_URL = os.environ.get("DATA_SERVICE_URL", os.environ.get("DATASVC_URL", "http://127.0.0.1:9009")) DEFAULT_TIMEFRAME_LABELS = OrderedDict([ ("1m", "1分钟"),