codex_jxs_code/抓取币安K线_ccxt.py

"""
基于 ccxt 的币安永续 ETHUSDT K线 + 秒级数据抓取
- K线: 1m、3m、5m、15m、30m、1h
- 秒级: fetch_trades 逐笔成交聚合
支持代理，与 抓取币安K线.py 共用同一数据库表
"""

import time
import datetime
import os
from pathlib import Path
from loguru import logger
from peewee import *

try:
    import ccxt
except ImportError:
    ccxt = None

DB_PATH = Path(__file__).parent / 'models' / 'database.db'
db = SqliteDatabase(str(DB_PATH))

KLINE_CONFIGS = {1: '1m', 3: '3m', 5: '5m', 15: '15m', 30: '30m', 60: '1h'}
CCXT_INTERVALS = {1: '1m', 3: '3m', 5: '5m', 15: '15m', 30: '30m', 60: '1h'}


class BinanceETHTrades(Model):
    id = BigIntegerField(primary_key=True)
    timestamp = BigIntegerField(index=True)
    price = FloatField()
    volume = FloatField()
    side = IntegerField()

    class Meta:
        database = db
        table_name = 'binance_eth_trades'


class BinanceETHSecond(Model):
    id = BigIntegerField(primary_key=True)
    open = FloatField(null=True)
    high = FloatField(null=True)
    low = FloatField(null=True)
    close = FloatField(null=True)
    volume = FloatField(null=True)
    trade_count = IntegerField(null=True)

    class Meta:
        database = db
        table_name = 'binance_eth_1s'


def create_kline_model(step: int):
    suffix = KLINE_CONFIGS.get(step, f'{step}m')
    tbl_name = f'binance_eth_{suffix}'
    attrs = {
        'id': BigIntegerField(primary_key=True),
        'open': FloatField(null=True),
        'high': FloatField(null=True),
        'low': FloatField(null=True),
        'close': FloatField(null=True),
    }
    meta = type('Meta', (), {'database': db, 'table_name': tbl_name})
    attrs['Meta'] = meta
    return type(f'BinanceETH{suffix}', (Model,), attrs)


class BinanceCCXTCollector:
    """基于 ccxt 的币安永续 K 线抓取"""

    def __init__(self, symbol: str = "ETHUSDT", proxy: str = None):
        self.symbol = symbol
        self.ccxt_symbol = "ETH/USDT:USDT"
        self.proxy = proxy or os.environ.get('BINANCE_PROXY', '').strip() or None
        self.models = {}
        self.exchange = None
        self._init_exchange()
        self._init_database()

    def _init_exchange(self):
        if not ccxt:
            raise ImportError("请安装 ccxt: pip install ccxt")
        options = {'defaultType': 'future', 'adjustForTimeDifference': True}
        config = {'timeout': 30000, 'enableRateLimit': True}
        if self.proxy:
            config['proxy'] = self.proxy
        self.exchange = ccxt.binanceusdm(config=config)
        self.exchange.options.update(options)
        logger.info(f"ccxt 交易所: binanceusdm" + (f" 代理: {self.proxy}" if self.proxy else ""))

    def _init_database(self):
        DB_PATH.parent.mkdir(parents=True, exist_ok=True)
        db.connect(reuse_if_open=True)
        for step in KLINE_CONFIGS.keys():
            model = create_kline_model(step)
            self.models[step] = model
            db.create_tables([model], safe=True)
        db.create_tables([BinanceETHTrades, BinanceETHSecond], safe=True)
        logger.info("数据库表已就绪")

    def get_db_time_range(self, step: int):
        model = self.models.get(step)
        if not model:
            return None, None
        try:
            lo = model.select(fn.MIN(model.id)).scalar()
            hi = model.select(fn.MAX(model.id)).scalar()
            return lo, hi
        except Exception:
            return None, None

    def fetch_klines(self, step: int, start_sec: int, end_sec: int, max_retries: int = 5):
        """ccxt fetch_ohlcv，按批次拉取"""
        interval = CCXT_INTERVALS.get(step, f'{step}m')
        since_ms = int(start_sec) * 1000
        end_ms = int(end_sec) * 1000
        out = []
        while since_ms < end_ms:
            for attempt in range(max_retries):
                try:
                    ohlcv = self.exchange.fetch_ohlcv(
                        self.ccxt_symbol,
                        interval,
                        since=since_ms,
                        limit=1500,
                    )
                    if not ohlcv:
                        break
                    for row in ohlcv:
                        ts, o, h, l, c = int(row[0]), row[1], row[2], row[3], row[4]
                        if ts >= end_ms:
                            break
                        out.append({'id': ts, 'open': float(o), 'high': float(h),
                                    'low': float(l), 'close': float(c)})
                    since_ms = int(ohlcv[-1][0]) + 1
                    time.sleep(0.15)
                    break
                except Exception as e:
                    delay = 2 + attempt * 2
                    logger.warning(f"fetch_ohlcv 异常 ({attempt+1}/{max_retries}): {e}，{delay}s 后重试")
                    if attempt < max_retries - 1:
                        time.sleep(delay)
            else:
                return []
        return out

    def save_klines(self, step: int, klines: list):
        model = self.models.get(step)
        if not model:
            return 0
        n = 0
        for k in klines:
            try:
                _, created = model.get_or_create(
                    id=k['id'],
                    defaults={'open': k['open'], 'high': k['high'], 'low': k['low'], 'close': k['close']}
                )
                if created:
                    n += 1
            except Exception as e:
                logger.error(f"保存失败 {k['id']}: {e}")
        return n

    def get_batch_seconds(self, step: int):
        return {1: 3600*4, 3: 3600*8, 5: 3600*12, 15: 3600*24, 30: 3600*48}.get(step, 3600*72)

    def collect_period_range(self, step: int, target_start: int, target_end: int):
        suffix = KLINE_CONFIGS.get(step, f'{step}m')
        batch = self.get_batch_seconds(step)
        db_lo, db_hi = self.get_db_time_range(step)
        db_lo_sec = db_lo // 1000 if db_lo else None
        db_hi_sec = db_hi // 1000 if db_hi else None

        total = 0

        backward_end = db_lo_sec if db_lo_sec else target_end
        if backward_end > target_start:
            logger.info(f"[{suffix}] 向前抓取历史")
            current = backward_end
            fail = 0
            while current > target_start and fail < 5:
                start = max(current - batch, target_start)
                klines = self.fetch_klines(step, start, current)
                if klines:
                    saved = self.save_klines(step, klines)
                    total += saved
                    logger.info(f"[{suffix}] ← {time.strftime('%Y-%m-%d %H:%M', time.localtime(start))} ~ "
                               f"{time.strftime('%Y-%m-%d %H:%M', time.localtime(current))} | "
                               f"获取 {len(klines)} 新增 {saved}")
                    fail = 0
                else:
                    fail += 1
                current = start
                time.sleep(0.3)

        forward_start = db_hi_sec if db_hi_sec else target_start
        if forward_start < target_end:
            logger.info(f"[{suffix}] 向后抓取最新")
            current = forward_start
            fail = 0
            while current < target_end and fail < 3:
                end = min(current + batch, target_end)
                klines = self.fetch_klines(step, current, end)
                if klines:
                    saved = self.save_klines(step, klines)
                    total += saved
                    logger.info(f"[{suffix}] → {time.strftime('%Y-%m-%d %H:%M', time.localtime(current))} ~ "
                               f"{time.strftime('%Y-%m-%d %H:%M', time.localtime(end))} | "
                               f"获取 {len(klines)} 新增 {saved}")
                    fail = 0
                else:
                    fail += 1
                current = end
                time.sleep(0.3)

        lo, hi = self.get_db_time_range(step)
        if lo and hi:
            logger.success(f"[{suffix}] 完成 新增 {total} | "
                          f"{time.strftime('%Y-%m-%d', time.localtime(lo//1000))} ~ "
                          f"{time.strftime('%Y-%m-%d', time.localtime(hi//1000))}")
        return total

    def collect_from_date(self, start_date: str, periods: list = None):
        if periods is None:
            periods = list(KLINE_CONFIGS.keys())
        start_ts = int(datetime.datetime.strptime(start_date, '%Y-%m-%d').timestamp())
        end_ts = int(time.time())
        logger.info(f"币安 ETHUSDT 永续(ccxt) | {start_date} ~ 当前 | 周期: {[KLINE_CONFIGS[p] for p in periods]}")
        results = {}
        for step in periods:
            results[KLINE_CONFIGS[step]] = self.collect_period_range(step, start_ts, end_ts)
            time.sleep(0.5)
        return results

    # ==================== 秒级 ====================

    def fetch_trades(self, since_ms: int = None, limit: int = 1000, max_retries: int = 5):
        for attempt in range(max_retries):
            try:
                trades = self.exchange.fetch_trades(
                    self.ccxt_symbol,
                    since=since_ms,
                    limit=limit,
                )
                out = []
                for t in trades:
                    out.append({
                        'id': t['id'],
                        'timestamp': t['timestamp'],
                        'price': float(t['price']),
                        'volume': float(t['amount']),
                        'side': 1 if t.get('side') == 'buy' else 0,
                    })
                return out
            except Exception as e:
                delay = 2 + attempt * 2
                logger.warning(f"fetch_trades 异常 ({attempt+1}/{max_retries}): {e}，{delay}s 后重试")
                if attempt < max_retries - 1:
                    time.sleep(delay)
        return []

    def save_trades(self, trades: list):
        n = 0
        for t in trades:
            try:
                _, created = BinanceETHTrades.get_or_create(
                    id=t['id'],
                    defaults={'timestamp': t['timestamp'], 'price': t['price'],
                              'volume': t['volume'], 'side': t['side']}
                )
                if created:
                    n += 1
            except Exception:
                pass
        return n

    def aggregate_to_seconds(self, start_ms: int = None, end_ms: int = None):
        q = BinanceETHTrades.select().order_by(BinanceETHTrades.timestamp)
        if start_ms:
            q = q.where(BinanceETHTrades.timestamp >= start_ms)
        if end_ms:
            q = q.where(BinanceETHTrades.timestamp <= end_ms)
        sec_data = {}
        for t in q:
            ts = (t.timestamp // 1000) * 1000
            if ts not in sec_data:
                sec_data[ts] = {
                    'open': t.price, 'high': t.price, 'low': t.price, 'close': t.price,
                    'volume': t.volume, 'trade_count': 1,
                }
            else:
                sec_data[ts]['high'] = max(sec_data[ts]['high'], t.price)
                sec_data[ts]['low'] = min(sec_data[ts]['low'], t.price)
                sec_data[ts]['close'] = t.price
                sec_data[ts]['volume'] += t.volume
                sec_data[ts]['trade_count'] += 1

        for ts, d in sec_data.items():
            try:
                BinanceETHSecond.insert(
                    id=ts, open=d['open'], high=d['high'], low=d['low'], close=d['close'],
                    volume=d['volume'], trade_count=d['trade_count']
                ).on_conflict(
                    conflict_target=[BinanceETHSecond.id],
                    update={
                        BinanceETHSecond.open: d['open'], BinanceETHSecond.high: d['high'],
                        BinanceETHSecond.low: d['low'], BinanceETHSecond.close: d['close'],
                        BinanceETHSecond.volume: d['volume'],
                        BinanceETHSecond.trade_count: d['trade_count'],
                    }
                ).execute()
            except Exception as e:
                logger.error(f"保存秒级失败 {ts}: {e}")
        logger.info(f"聚合 {len(sec_data)} 条秒级数据")

    def collect_second_data(self, start_date: str, end_date: str = None):
        start_ts = int(datetime.datetime.strptime(start_date, '%Y-%m-%d').timestamp())
        end_ts = int(time.time()) if end_date is None else int(
            datetime.datetime.strptime(end_date, '%Y-%m-%d').timestamp())
        try:
            hi = BinanceETHSecond.select(fn.MAX(BinanceETHSecond.id)).scalar()
            fetch_start = (hi + 1000) if hi else start_ts * 1000
        except Exception:
            fetch_start = start_ts * 1000

        total_saved = 0
        since = fetch_start
        while since < end_ts * 1000:
            trades = self.fetch_trades(since_ms=since, limit=1000)
            if trades:
                saved = self.save_trades(trades)
                total_saved += saved
                since = trades[-1]['timestamp'] + 1
                logger.info(f"[1s] {datetime.datetime.fromtimestamp(since/1000).strftime('%Y-%m-%d %H:%M:%S')} | "
                           f"成交 {len(trades)} 新增 {saved}")
            else:
                since += 3600 * 1000
            time.sleep(0.2)

        if total_saved > 0:
            self.aggregate_to_seconds(start_ms=fetch_start, end_ms=end_ts*1000)
        logger.success(f"秒级抓取完成 新增 {total_saved}")
        return total_saved

    def collect_trades_realtime(self, duration_seconds: int = 3600, interval: float = 0.5):
        end = time.time() + duration_seconds
        total = 0
        since = None
        while time.time() < end:
            try:
                trades = self.fetch_trades(since_ms=since, limit=1000)
                if trades:
                    total += self.save_trades(trades)
                    since = trades[-1]['timestamp'] + 1
            except Exception as e:
                logger.warning(f"实时采集异常: {e}")
            time.sleep(interval)
        self.aggregate_to_seconds()
        logger.success(f"实时采集完成 新增 {total}")
        return total

    def get_stats(self):
        logger.info("币安(ccxt) 数据库统计:")
        for step, model in self.models.items():
            try:
                c = model.select().count()
                lo, hi = self.get_db_time_range(step)
                s = f"  {KLINE_CONFIGS[step]:>4}: {c:>8} 条"
                if lo and hi:
                    s += f" | {time.strftime('%Y-%m-%d %H:%M', time.localtime(lo//1000))} ~ " \
                         f"{time.strftime('%Y-%m-%d %H:%M', time.localtime(hi//1000))}"
                logger.info(s)
            except Exception as e:
                logger.error(f"  {KLINE_CONFIGS[step]}: {e}")
        try:
            tc = BinanceETHTrades.select().count()
            sc = BinanceETHSecond.select().count()
            if sc > 0:
                lo = BinanceETHSecond.select(fn.MIN(BinanceETHSecond.id)).scalar()
                hi = BinanceETHSecond.select(fn.MAX(BinanceETHSecond.id)).scalar()
                logger.info(f"  trades: {tc:>8} | 1s: {sc:>8} | "
                           f"{datetime.datetime.fromtimestamp(lo/1000).strftime('%Y-%m-%d %H:%M:%S')} ~ "
                           f"{datetime.datetime.fromtimestamp(hi/1000).strftime('%Y-%m-%d %H:%M:%S')}")
            else:
                logger.info(f"  trades: {tc:>8} | 1s: {sc:>8}")
        except Exception as e:
            logger.error(f"  trades/1s: {e}")

    def close(self):
        if not db.is_closed():
            db.close()


if __name__ == '__main__':
    proxy = os.environ.get('BINANCE_PROXY') or None
    collector = BinanceCCXTCollector(symbol="ETHUSDT", proxy=proxy)
    try:
        collector.get_stats()

        # K线（不指定 periods 则抓取全部：1m、3m、5m、15m、30m、1h）
        collector.collect_from_date(start_date='2025-01-01')

        # 秒级
        collector.collect_second_data(start_date='2025-01-01')

        collector.get_stats()
    finally:
        collector.close()