192 lines
8.0 KiB
Python
192 lines
8.0 KiB
Python
import asyncio
|
||
import time
|
||
from datetime import datetime, timezone
|
||
from typing import Iterable
|
||
|
||
import aiohttp
|
||
import aiohttp.resolver
|
||
import ccxt.async_support as ccxt
|
||
import pandas as pd
|
||
|
||
import logging
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class BinanceFetcher:
|
||
def __init__(self, config: dict):
|
||
self.config = config
|
||
self.quote_currencies = self._resolve_quote_currencies(config)
|
||
self.exchange = self._build_exchange(config)
|
||
self._markets_meta: dict[str, dict] = {} # symbol -> market dict（含 onchainDate 等）
|
||
|
||
@staticmethod
|
||
def _resolve_quote_currencies(config: dict) -> list[str]:
|
||
"""优先取 quote_currencies 列表，回退到 quote_currency 单值。"""
|
||
base = config.get('base_filter', {})
|
||
quotes = base.get('quote_currencies')
|
||
if quotes:
|
||
return [q.upper() for q in quotes]
|
||
single = base.get('quote_currency')
|
||
return [single.upper()] if single else ['USDT']
|
||
|
||
@staticmethod
|
||
def _build_exchange(config: dict):
|
||
exchange_class = getattr(ccxt, config['exchange']['name'])
|
||
kwargs = {
|
||
'enableRateLimit': True,
|
||
'options': {'defaultType': config['exchange']['market']},
|
||
}
|
||
proxy = config['exchange'].get('proxy')
|
||
if proxy:
|
||
kwargs['httpsProxy'] = proxy
|
||
exchange = exchange_class(kwargs)
|
||
|
||
# 若需代理，注入自定义 aiohttp session（ThreadedResolver 绕过 aiodns DNS 失败）
|
||
if proxy:
|
||
resolver = aiohttp.resolver.ThreadedResolver()
|
||
connector = aiohttp.TCPConnector(resolver=resolver, enable_cleanup_closed=True)
|
||
exchange._custom_session = aiohttp.ClientSession(connector=connector)
|
||
exchange._own_session = False # 阻止 ccxt 在 open() 中重建 session
|
||
exchange.session = exchange._custom_session
|
||
return exchange
|
||
|
||
async def close(self):
|
||
await self.exchange.close()
|
||
custom = getattr(self.exchange, '_custom_session', None)
|
||
if custom and not custom.closed:
|
||
await custom.close()
|
||
|
||
# ------------------------------------------------------------------
|
||
# Phase 1: 流动性 + 上市天数过滤
|
||
# ------------------------------------------------------------------
|
||
|
||
async def get_liquid_symbols(self) -> list[str]:
|
||
"""按所有配置的报价币拉取并过滤。"""
|
||
logger.info("正在加载交易所市场信息...")
|
||
await self.exchange.load_markets()
|
||
self._markets_meta = dict(self.exchange.markets)
|
||
|
||
tickers = await self.exchange.fetch_tickers()
|
||
|
||
base_conf = self.config['base_filter']
|
||
min_listing_days = int(base_conf.get('min_listing_days', 0))
|
||
exclude_stablecoins = bool(base_conf.get('exclude_stablecoins', True))
|
||
stablecoin_map = base_conf.get('stablecoin_bases', {})
|
||
|
||
# 通用稳定币列表（用于报价币不在 stablecoin_map 时的回退）
|
||
default_stable = {'USDT', 'USDC', 'FDUSD', 'DAI', 'TUSD', 'BUSD', 'EUR', 'TRY', 'WBTC'}
|
||
|
||
# 按报价币解析成交额门槛 —— 未配置时沿用默认值
|
||
default_volume = float(base_conf['min_quote_volume_24h'])
|
||
volume_thresholds: dict[str, float] = {
|
||
q.upper(): float(v) for q, v in (base_conf.get('volume_thresholds') or {}).items()
|
||
}
|
||
for q in self.quote_currencies:
|
||
t = volume_thresholds.get(q, default_volume)
|
||
logger.info(f" 报价币 {q}: 成交额门槛 = {t:,.0f}")
|
||
|
||
valid: list[str] = []
|
||
|
||
for symbol, ticker in tickers.items():
|
||
market = self._markets_meta.get(symbol, {})
|
||
quote = market.get('quote')
|
||
base = market.get('base')
|
||
if quote not in self.quote_currencies:
|
||
continue
|
||
|
||
# 1. 成交额（按 quote 独立阈值）
|
||
qv = ticker.get('quoteVolume') or 0
|
||
threshold = volume_thresholds.get(quote.upper(), default_volume)
|
||
if qv < threshold:
|
||
continue
|
||
|
||
# 2. 价差
|
||
bid, ask = ticker.get('bid'), ticker.get('ask')
|
||
if bid and ask and bid > 0:
|
||
spread = (ask - bid) / bid * 100
|
||
if spread > base_conf['max_spread_pct']:
|
||
continue
|
||
|
||
# 3. 上市天数（币安 market.active 通常为 True；onchainDate/listingDate 不一定存在）
|
||
if min_listing_days > 0 and not self._passes_listing_filter(market, min_listing_days):
|
||
continue
|
||
|
||
# 4. 稳定币过滤
|
||
if exclude_stablecoins:
|
||
banned = set(stablecoin_map.get(quote, default_stable))
|
||
if base in banned:
|
||
continue
|
||
|
||
valid.append(symbol)
|
||
|
||
logger.info(
|
||
f"基础流动性过滤完成：报价币={self.quote_currencies}，剩余 {len(valid)} 个标的。"
|
||
)
|
||
return valid
|
||
|
||
@staticmethod
|
||
def _passes_listing_filter(market: dict, min_days: int) -> bool:
|
||
"""检查市场是否满足上市天数要求。"""
|
||
# 币安 markets 通常不直接返回 listing date；这里使用 active 字段做软校验
|
||
if not market.get('active', True):
|
||
return False
|
||
# 部分交易所/合约类型会返回 info.createdAt / info.listDate
|
||
info = market.get('info', {}) if isinstance(market, dict) else {}
|
||
ts_ms = info.get('onboardDate') or info.get('listDate') or info.get('createdAt')
|
||
if ts_ms is None:
|
||
# 拿不到日期，按通过处理（避免误杀）
|
||
return True
|
||
try:
|
||
ts_ms = int(ts_ms)
|
||
except (TypeError, ValueError):
|
||
return True
|
||
onboard = datetime.fromtimestamp(ts_ms / 1000, tz=timezone.utc)
|
||
elapsed = (datetime.now(tz=timezone.utc) - onboard).days
|
||
return elapsed >= min_days
|
||
|
||
# ------------------------------------------------------------------
|
||
# Phase 2: 批量 K 线拉取（带重试）
|
||
# ------------------------------------------------------------------
|
||
|
||
async def fetch_klines_batch(self, symbols: list[str]) -> dict[str, pd.DataFrame]:
|
||
timeframe = self.config['data']['timeframe']
|
||
limit = self.config['data']['lookback_candles']
|
||
concurrency = int(self.config['data'].get('max_concurrency', 15))
|
||
|
||
fetcher_conf = self.config.get('fetcher', {})
|
||
max_retries = int(fetcher_conf.get('max_retries', 3))
|
||
retry_delay = float(fetcher_conf.get('retry_delay', 1.0))
|
||
|
||
sem = asyncio.Semaphore(concurrency)
|
||
|
||
async def fetch_single(sym):
|
||
async with sem:
|
||
return sym, await self._fetch_with_retry(sym, timeframe, limit, max_retries, retry_delay)
|
||
|
||
logger.info(f"开始异步拉取 {len(symbols)} 个标的的 {timeframe} K线...")
|
||
results = await asyncio.gather(*(fetch_single(s) for s in symbols))
|
||
return {sym: df for sym, df in results if df is not None}
|
||
|
||
async def _fetch_with_retry(
|
||
self, symbol: str, timeframe: str, limit: int, max_retries: int, retry_delay: float
|
||
) -> pd.DataFrame | None:
|
||
last_err = None
|
||
for attempt in range(1, max_retries + 1):
|
||
try:
|
||
ohlcv = await self.exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=limit)
|
||
if not ohlcv:
|
||
raise ValueError("empty ohlcv response")
|
||
df = pd.DataFrame(
|
||
ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume']
|
||
)
|
||
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
|
||
df.set_index('timestamp', inplace=True)
|
||
return df
|
||
except Exception as e:
|
||
last_err = e
|
||
logger.warning(f"[{symbol}] 第 {attempt}/{max_retries} 次拉取失败: {e}")
|
||
if attempt < max_retries:
|
||
await asyncio.sleep(retry_delay * attempt)
|
||
logger.error(f"[{symbol}] 拉取失败，已重试 {max_retries} 次，最终放弃。last_err={last_err}")
|
||
return None |