"""从Binance获取加密货币OHLCV数据""" import os import json import time from datetime import datetime, timedelta from pathlib import Path from typing import Optional, Literal, Any import logging from urllib.parse import urlparse import pandas as pd from binance.client import Client import pytz from dotenv import load_dotenv from websocket import create_connection from data.database import DatabaseManager from utils.send_msg import lark_notifier # 加载环境变量 load_dotenv() logger = logging.getLogger("polymarket_trader") def _apply_global_proxy(proxy_url: str) -> None: """统一设置代理环境变量，确保第三方网络库也走代理。""" os.environ['HTTP_PROXY'] = proxy_url os.environ['HTTPS_PROXY'] = proxy_url os.environ['ALL_PROXY'] = proxy_url os.environ['http_proxy'] = proxy_url os.environ['https_proxy'] = proxy_url os.environ['all_proxy'] = proxy_url class BinanceDataFetcher: """从Binance公开API获取和缓存OHLCV数据""" INTERVALS = { '1h': Client.KLINE_INTERVAL_1HOUR, '15m': Client.KLINE_INTERVAL_15MINUTE, } INTERVAL_MS = { '1h': 60 * 60 * 1000, '15m': 15 * 60 * 1000, } WS_API_ENDPOINT = "wss://ws-api.binance.com/ws-api/v3" def __init__(self, cache_dir: str = "./data/cache"): """初始化Binance客户端和数据库管理器 Args: cache_dir: 缓存/数据库目录 """ # 配置代理 socks5_proxy = os.getenv('SOCKS5_PROXY', 'socks5://127.0.0.1:7890') _apply_global_proxy(socks5_proxy) proxies = { 'http': socks5_proxy, 'https': socks5_proxy } logger.info(f"强制使用SOCKS5代理: {socks5_proxy}") try: # 初始化Binance客户端（强制走代理） self.client = Client(requests_params={'proxies': proxies}) # 测试连接 self.client.ping() logger.info("已成功连接到Binance API") except Exception as e: logger.warning(f"Binance API连接失败: {e}") logger.warning("提示: 1) 检查代理设置 2) 确保代理服务运行 3) 或者使用缓存数据") self.client = None # 初始化数据库管理器 self.db_manager = DatabaseManager(os.path.join(cache_dir, "market_data.db")) def fetch_ohlcv( self, symbol: Literal['BTC', 'ETH', 'SOL'], timeframe: Literal['1h', '15m'], start_date: Optional[str] = None, end_date: Optional[str] = None, use_cache: bool = True, max_retries: int = 3 # [新增] 最大重试次数参数 ) -> pd.DataFrame: """获取加密货币对的OHLCV数据 Args: symbol: 加密货币符号 (BTC, ETH, SOL) timeframe: 时间框架间隔 (1h 或 15m) start_date: 开始日期，格式：'YYYY-MM-DD' (默认: 1年前) end_date: 结束日期，格式：'YYYY-MM-DD' (默认: 现在) use_cache: 是否使用缓存数据 Returns: 包含以下列的DataFrame: timestamp, open, high, low, close, volume """ # 设置日期范围 if end_date is None: end_dt = datetime.now(pytz.UTC) else: end_dt = datetime.strptime(end_date, '%Y-%m-%d').replace(tzinfo=pytz.UTC) if start_date is None: start_dt = end_dt - timedelta(days=365) # 1年前 else: start_dt = datetime.strptime(start_date, '%Y-%m-%d').replace(tzinfo=pytz.UTC) start_str = start_dt.strftime('%Y-%m-%d') end_str = end_dt.strftime('%Y-%m-%d') # 1. 尝试从数据库获取数据 df = pd.DataFrame() if use_cache: df = self.db_manager.get_ohlcv(symbol, timeframe, start_str, end_str) # 检查数据是否足够 if not df.empty: min_ts = df['timestamp'].min().replace(tzinfo=pytz.UTC) max_ts = df['timestamp'].max().replace(tzinfo=pytz.UTC) # 简单判断覆盖范围 (容忍少量缺失) if min_ts <= start_dt + timedelta(days=2) and max_ts >= end_dt - timedelta(days=2): logger.info(f"从数据库加载数据: {symbol} {timeframe} ({len(df)} rows)") return df # 2. 从API获取数据 if self.client is None: if not df.empty: logger.warning(f"API不可用，返回数据库现有数据 ({len(df)} rows)") return df raise ConnectionError( "Binance客户端不可用。请使用VPN或确保数据库有数据。" ) # Binance使用USDT交易对 pair = f"{symbol}USDT" interval = self.INTERVALS[timeframe] interval_ms = self.INTERVAL_MS[timeframe] logger.info(f"获取 {symbol} {timeframe} 数据，时间范围: {start_dt.date()} 至 {end_dt.date()}") klines = [] for attempt in range(max_retries): try: # 使用 Binance WebSocket API 的 klines 方法获取历史K线 klines = self._get_historical_klines_via_ws( symbol=pair, interval=interval, start_ts_ms=int(start_dt.timestamp() * 1000), end_ts_ms=int(end_dt.timestamp() * 1000), interval_ms=interval_ms ) # 如果成功获取，跳出循环 break except Exception as e: # 如果是最后一次尝试，且依然失败 if attempt == max_retries - 1: logger.error(f"第 {attempt + 1} 次尝试失败，已达到最大重试次数。错误: {e}") if not df.empty: logger.warning("WebSocket请求最终失败，返回数据库现有数据") return df raise e # 如果没有缓存数据，抛出异常 # 计算等待时间 (指数退避: 2s, 4s, 8s...) wait_time = 2 * (2 ** attempt) logger.warning(f"请求失败 (尝试 {attempt + 1}/{max_retries})，{wait_time}秒后重试。错误: {e}") time.sleep(wait_time) if not klines: lark_notifier.send_alert(f"数据获取失败: {symbol} {timeframe}", "连续3次请求或解析发生异常，已放弃重试") if not df.empty: return df raise ValueError(f"未获取到 {symbol} {timeframe} 的数据") # 转换为DataFrame new_df = pd.DataFrame(klines, columns=[ 'timestamp', 'open', 'high', 'low', 'close', 'volume', 'close_time', 'quote_volume', 'trades', 'taker_buy_base', 'taker_buy_quote', 'ignore' ]) # 选择并转换列 new_df = new_df[['timestamp', 'open', 'high', 'low', 'close', 'volume']].copy() new_df['timestamp'] = pd.to_datetime(new_df['timestamp'], unit='ms') for col in ['open', 'high', 'low', 'close', 'volume']: new_df[col] = new_df[col].astype(float) # 按时间排序 new_df = new_df.sort_values('timestamp').reset_index(drop=True) logger.info(f"已获取 {len(new_df)} 行数据") # 保存到数据库 self.db_manager.save_ohlcv(new_df, symbol, timeframe) return new_df def _get_historical_klines_via_ws( self, symbol: str, interval: str, start_ts_ms: int, end_ts_ms: int, interval_ms: int, limit: int = 1000 ): """通过 Binance WebSocket API 分页拉取历史K线。""" if end_ts_ms <= start_ts_ms: return [] all_klines = [] request_id = 1 current_start = start_ts_ms # WebSocket API 单次最多返回 1000 条，这里按窗口分页拉取。 while current_start < end_ts_ms: max_window_end = current_start + (limit * interval_ms) current_end = min(end_ts_ms, max_window_end) payload = { "id": request_id, "method": "klines", "params": { "symbol": symbol, "interval": interval, "startTime": current_start, "endTime": current_end, "limit": limit } } response_data = self._ws_request(payload) batch = response_data.get("result", []) if not batch: break all_klines.extend(batch) last_open_time = int(batch[-1][0]) next_start = last_open_time + interval_ms if next_start <= current_start: break current_start = next_start request_id += 1 return all_klines def _ws_request(self, payload: dict) -> dict: """发送单次 WebSocket API 请求并返回 JSON 响应。""" ws_kwargs: dict[str, Any] = { "timeout": 30 } socks5_proxy = os.getenv('SOCKS5_PROXY', 'socks5://127.0.0.1:7890') parsed = urlparse(socks5_proxy) if parsed.hostname and parsed.port: scheme = (parsed.scheme or 'socks5').lower() if scheme not in {'socks4', 'socks5', 'http'}: scheme = 'socks5' if scheme == 'socks5h': scheme = 'socks5' ws_kwargs.update({ "http_proxy_host": parsed.hostname, "http_proxy_port": parsed.port, "proxy_type": scheme }) ws = create_connection(self.WS_API_ENDPOINT, **ws_kwargs) try: ws.send(json.dumps(payload)) raw = ws.recv() data = json.loads(raw) status = data.get("status", 200) if status != 200: raise RuntimeError(f"Binance WS API返回异常状态: {data}") return data finally: ws.close() def get_latest_price(self, symbol: Literal['BTC', 'ETH', 'SOL']) -> float: """获取符号的当前价格 Args: symbol: 加密货币符号 Returns: 当前USDT价格 """ if self.client is None: raise ConnectionError("Binance客户端不可用，无法获取最新价格") pair = f"{symbol}USDT" ticker = self.client.get_symbol_ticker(symbol=pair) return float(ticker['price']) if __name__ == "__main__": # 测试数据获取器 fetcher = BinanceDataFetcher() # 获取BTC 1小时数据（1年） df = fetcher.fetch_ohlcv('BTC', '1h') print("\nBTC 1小时数据:") print(df.head()) print(f"\n形状：{df.shape}") print(f"日期范围：{df['timestamp'].min()} 至 {df['timestamp'].max()}") # 获取当前价格 try: current_price = fetcher.get_latest_price('BTC') print(f"\n当前BTC价格：${current_price:,.2f}") except: print("\n无法获取当前价格")