220 lines
7.7 KiB
Python
220 lines
7.7 KiB
Python
"""从Binance获取加密货币OHLCV数据"""
|
||
|
||
import os
|
||
import time
|
||
from datetime import datetime, timedelta
|
||
from pathlib import Path
|
||
from typing import Optional, Literal
|
||
|
||
import pandas as pd
|
||
from binance.client import Client
|
||
import pytz
|
||
from dotenv import load_dotenv
|
||
|
||
from data.database import DatabaseManager
|
||
from utils.send_msg import lark_notifier
|
||
|
||
# 加载环境变量
|
||
load_dotenv()
|
||
|
||
|
||
class BinanceDataFetcher:
|
||
"""从Binance公开API获取和缓存OHLCV数据"""
|
||
|
||
INTERVALS = {
|
||
'1h': Client.KLINE_INTERVAL_1HOUR,
|
||
'15m': Client.KLINE_INTERVAL_15MINUTE,
|
||
}
|
||
|
||
def __init__(self, cache_dir: str = "./data/cache"):
|
||
"""初始化Binance客户端和数据库管理器
|
||
|
||
Args:
|
||
cache_dir: 缓存/数据库目录
|
||
"""
|
||
# 配置代理
|
||
proxies = None
|
||
use_proxy = os.getenv('USE_PROXY', 'false').lower() == 'true'
|
||
|
||
if use_proxy:
|
||
socks5_proxy = os.getenv('SOCKS5_PROXY', 'socks5://127.0.0.1:7890')
|
||
proxies = {
|
||
'http': socks5_proxy,
|
||
'https': socks5_proxy
|
||
}
|
||
print(f"✓ 使用SOCKS5代理: {socks5_proxy}")
|
||
|
||
try:
|
||
# 初始化Binance客户端（带代理）
|
||
if proxies:
|
||
self.client = Client(requests_params={'proxies': proxies})
|
||
else:
|
||
self.client = Client()
|
||
|
||
# 测试连接
|
||
self.client.ping()
|
||
print("✓ 已成功连接到Binance API")
|
||
except Exception as e:
|
||
print(f"⚠ Binance API连接失败: {e}")
|
||
print("提示：")
|
||
print(" 1. 检查代理设置 (.env文件)")
|
||
print(" 2. 确保代理服务正在运行")
|
||
print(" 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):
|
||
print(f"从数据库加载数据：{symbol} {timeframe} ({len(df)} rows)")
|
||
return df
|
||
|
||
# 2. 从API获取数据
|
||
if self.client is None:
|
||
if not df.empty:
|
||
print(f"⚠ API不可用，返回数据库现有数据 ({len(df)} rows)")
|
||
return df
|
||
|
||
raise ConnectionError(
|
||
"Binance客户端不可用。请使用VPN或确保数据库有数据。"
|
||
)
|
||
|
||
# Binance使用USDT交易对
|
||
pair = f"{symbol}USDT"
|
||
interval = self.INTERVALS[timeframe]
|
||
|
||
print(f"获取 {symbol} {timeframe} 数据，时间范围：{start_dt.date()} 至 {end_dt.date()}...")
|
||
|
||
klines = []
|
||
for attempt in range(max_retries):
|
||
try:
|
||
klines = self.client.get_historical_klines(
|
||
symbol=pair,
|
||
interval=interval,
|
||
start_str=str(int(start_dt.timestamp() * 1000)),
|
||
end_str=str(int(end_dt.timestamp() * 1000))
|
||
)
|
||
# 如果成功获取，跳出循环
|
||
break
|
||
except Exception as e:
|
||
# 如果是最后一次尝试，且依然失败
|
||
if attempt == max_retries - 1:
|
||
print(f"❌ 第 {attempt + 1} 次尝试失败，已达到最大重试次数。错误: {e}")
|
||
if not df.empty:
|
||
print(f"⚠ API请求最终失败，返回数据库现有数据")
|
||
return df
|
||
raise e # 如果没有缓存数据，抛出异常
|
||
|
||
# 计算等待时间 (指数退避: 2s, 4s, 8s...)
|
||
wait_time = 2 * (2 ** attempt)
|
||
print(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)
|
||
|
||
print(f"已获取 {len(new_df)} 行数据")
|
||
|
||
# 保存到数据库
|
||
self.db_manager.save_ohlcv(new_df, symbol, timeframe)
|
||
|
||
return new_df
|
||
|
||
def get_latest_price(self, symbol: Literal['BTC', 'ETH', 'SOL']) -> float:
|
||
"""获取符号的当前价格
|
||
|
||
Args:
|
||
symbol: 加密货币符号
|
||
|
||
Returns:
|
||
当前USDT价格
|
||
"""
|
||
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无法获取当前价格")
|