第四章：资金费率数据抓取

做资金费率套利，第一步就是搞定数据。没有数据，策略就是空中楼阁。我个人习惯把数据抓取分成三个层次：实时数据、历史数据、数据存储。今天咱们一个一个来拆解。

4.1 使用CCXT获取实时资金费率

CCXT这个库，说白了就是交易所的统一接口。不管你用币安、OKX还是Bybit，代码写法几乎一样。我刚开始做套利时，每个交易所单独写一套API调用，维护起来简直想哭。后来换了CCXT，真香。

import ccxt
import time

# 初始化交易所
exchange = ccxt.binance({
    'apiKey': '你的API_KEY',
    'secret': '你的SECRET',
    'enableRateLimit': True,  # 这个很重要，防止被交易所封
})

# 获取BTC/USDT永续合约的资金费率
def get_funding_rate(symbol='BTC/USDT'):
    try:
        funding = exchange.fetch_funding_rate(symbol)
        return {
            'symbol': symbol,
            'funding_rate': funding['fundingRate'],
            'next_funding_time': funding['nextFundingTime'],
            'timestamp': funding['timestamp']
        }
    except Exception as e:
        print(f"抓取失败: {e}")
        return None

# 循环抓取，每5秒一次
while True:
    data = get_funding_rate()
    if data:
        print(f"当前费率: {data['funding_rate']*100:.4f}%")
    time.sleep(5)

嗯，这里要注意：不同交易所的symbol格式可能不一样。币安用BTC/USDT，OKX用BTC-USDT-SWAP。我建议统一用CCXT的market_id来转换，省得踩坑。

4.2 历史费率数据下载

实时数据只是开胃菜，真正做回测需要历史数据。我曾经在回测时发现策略表现很好，结果上线就亏钱——后来查出来是历史数据采样频率不对。

为什么会这样？因为资金费率每8小时结算一次，但交易所每分钟都会更新预期费率。如果你只拿结算时的数据，会漏掉很多波动。

import pandas as pd
from datetime import datetime, timedelta

def download_historical_funding(symbol='BTC/USDT', days=30):
    """
    下载历史资金费率数据
    币安支持最多500条历史记录，需要分批抓取
    """
    exchange = ccxt.binance()
    all_data = []
    
    # 计算起始时间
    since = exchange.parse8601(
        (datetime.now() - timedelta(days=days)).isoformat()
    )
    
    while True:
        try:
            # 每次抓取500条
            funding_history = exchange.fetch_funding_rate_history(
                symbol, 
                since=since,
                limit=500
            )
            
            if not funding_history:
                break
                
            all_data.extend(funding_history)
            # 更新起始时间，避免重复
            since = funding_history[-1]['timestamp'] + 1
            
            print(f"已下载 {len(all_data)} 条记录")
            
        except Exception as e:
            print(f"下载中断: {e}")
            break
    
    # 转为DataFrame
    df = pd.DataFrame(all_data)
    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
    df['funding_rate'] = df['fundingRate'].astype(float)
    
    return df[['timestamp', 'funding_rate']]

# 下载30天数据
df = download_historical_funding('BTC/USDT', days=30)
print(df.head())

4.3 数据清洗与存储

数据拿到手，别急着用。你想想看，交易所偶尔会返回空值、重复数据或者异常值。我遇到过最离谱的一次，币安某个交易对突然返回了1000%的资金费率——明显是bug。

4.3.1 清洗步骤

def clean_funding_data(df):
    """数据清洗流水线"""
    # 1. 去重
    df = df.drop_duplicates(subset=['timestamp'])
    
    # 2. 排序
    df = df.sort_values('timestamp')
    
    # 3. 处理缺失值
    df = df.dropna(subset=['funding_rate'])
    
    # 4. 异常值过滤（资金费率一般在-0.1%到0.1%之间）
    df = df[df['funding_rate'].abs() < 0.01]  # 超过1%的视为异常
    
    # 5. 重采样到分钟级别（填充缺失的分钟数据）
    df = df.set_index('timestamp')
    df = df.resample('1T').ffill()  # 向前填充
    
    return df.reset_index()

df_clean = clean_funding_data(df)
print(f"清洗前: {len(df)} 条, 清洗后: {len(df_clean)} 条")

4.3.2 存储到CSV

# CSV存储，简单直接
df_clean.to_csv('funding_rate_btc_usdt.csv', index=False)
print("数据已保存到 funding_rate_btc_usdt.csv")

4.3.3 存储到数据库

数据量大了以后，CSV就不够用了。我建议用SQLite起步，轻量又方便。

import sqlite3

def save_to_database(df, db_path='funding.db'):
    """存储到SQLite数据库"""
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    
    # 创建表
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS funding_rates (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            timestamp TEXT NOT NULL,
            funding_rate REAL NOT NULL,
            symbol TEXT NOT NULL,
            UNIQUE(timestamp, symbol)
        )
    ''')
    
    # 批量插入
    df['symbol'] = 'BTC/USDT'
    df.to_sql('funding_rates', conn, if_exists='append', index=False)
    
    conn.commit()
    conn.close()
    print(f"数据已存入 {db_path}")

save_to_database(df_clean)

4.4 知识体系总览

下面这张图，是我做数据抓取时的完整流程。你看一遍就能明白整个链路。

4.5 完整代码整合

最后，我把上面所有代码整合成一个完整的抓取脚本。你直接复制就能用。

import ccxt
import pandas as pd
import sqlite3
from datetime import datetime, timedelta

class FundingRateCollector:
    def __init__(self, exchange_id='binance'):
        self.exchange = getattr(ccxt, exchange_id)({
            'enableRateLimit': True
        })
    
    def fetch_realtime(self, symbol='BTC/USDT'):
        """获取实时资金费率"""
        data = self.exchange.fetch_funding_rate(symbol)
        return {
            'timestamp': pd.to_datetime(data['timestamp'], unit='ms'),
            'funding_rate': data['fundingRate']
        }
    
    def fetch_history(self, symbol='BTC/USDT', days=30):
        """获取历史数据"""
        all_data = []
        since = self.exchange.parse8601(
            (datetime.now() - timedelta(days=days)).isoformat()
        )
        
        while True:
            records = self.exchange.fetch_funding_rate_history(
                symbol, since=since, limit=500
            )
            if not records:
                break
            all_data.extend(records)
            since = records[-1]['timestamp'] + 1
        
        df = pd.DataFrame(all_data)
        df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
        df['funding_rate'] = df['fundingRate'].astype(float)
        return df[['timestamp', 'funding_rate']]
    
    def clean_data(self, df):
        """数据清洗"""
        df = df.drop_duplicates(subset=['timestamp'])
        df = df.sort_values('timestamp')
        df = df.dropna()
        df = df[df['funding_rate'].abs() < 0.01]
        df = df.set_index('timestamp').resample('1T').ffill().reset_index()
        return df
    
    def save_csv(self, df, filename):
        df.to_csv(filename, index=False)
        print(f"保存到 {filename}")
    
    def save_db(self, df, db_path='funding.db'):
        conn = sqlite3.connect(db_path)
        df.to_sql('funding_rates', conn, if_exists='append', index=False)
        conn.close()
        print(f"保存到数据库 {db_path}")

# 使用示例
collector = FundingRateCollector()
df = collector.fetch_history('BTC/USDT', days=7)
df_clean = collector.clean_data(df)
collector.save_csv(df_clean, 'btc_funding_7d.csv')
collector.save_db(df_clean)