322 lines
11 KiB
Python
322 lines
11 KiB
Python
"""从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无法获取当前价格")
|