如何在pandas DataFrame中持续更新加密货币15分钟行情数据并实现历史对比

存储与对比逻辑选型结论 #

两种你提到的方案都可落地，更推荐内存存上一次拉取的DataFrame做对比+本地持久化单份基准数据的折中方案，兼顾运行效率和异常恢复能力：

• 新建pricechange2文件和旧文件对比：仅适合需要长期留存所有历史拉取原始数据的场景，缺点是频繁读写磁盘，运行效率低
• 直接在DataFrame增删列：纯内存操作速度快，但程序意外崩溃会丢失历史基准数据，无法继续对比

推荐的折中方案：只在内存里留存上一次拉取的结果做对比，每次拉取完把最新的基准数据覆盖写入本地一个固定文件，程序重启时直接读这个文件恢复上一次的基准数据，既避免频繁读写多份文件，又解决了程序崩溃数据丢失的问题。

完整实现代码 #

import ccxt
import re
import pandas as pd
import time
import os

# 封装单次USDT交易对数据拉取逻辑
def fetch_usdt_tickers(exchange):
    markets = exchange.fetchTickers()
    # 过滤符合要求的USDT交易对，排除杠杆代币
    usdt_pattern = re.compile(".*/USDT")
    exclude_keywords = ["DOWN", "UP", "BULL", "BEAR"]
    filtered_symbols = [
        symbol for symbol in markets.keys()
        if usdt_pattern.match(symbol) and all(key not in symbol for key in exclude_keywords)
    ]
    # 构造结构化DataFrame
    ticker_list = []
    for symbol in filtered_symbols:
        change_pct = float(markets[symbol]['info']['priceChangePercent'])
        ticker_list.append({"coinpair": symbol, "change_pct": change_pct})
    return pd.DataFrame(ticker_list).sort_values(by="change_pct", ascending=False)

if __name__ == "__main__":
    # 初始化交易所实例，按你自己的需求替换对应交易所和配置
    exchange = ccxt.binance({"enableRateLimit": True})
    # 本地持久化基准数据的路径
    BASE_DATA_PATH = "last_ticker_base.csv"
    # 拉取间隔15分钟，单位秒
    FETCH_INTERVAL = 15 * 60

    # 程序启动时加载已有的基准数据，避免重启后丢失上一次的对比基准
    last_ticker_df = None
    if os.path.exists(BASE_DATA_PATH):
        last_ticker_df = pd.read_csv(BASE_DATA_PATH)
        print("已加载本地存储的上一次基准数据")
    
    while True:
        current_time = time.strftime("%Y-%m-%d %H:%M:%S")
        print(f"\n===== {current_time} 拉取最新交易对数据 =====")
        current_ticker_df = fetch_usdt_tickers(exchange)

        # 非首次拉取时执行新旧数据对比
        if last_ticker_df is not None:
            # 按交易对合并两次数据
            compare_df = pd.merge(
                last_ticker_df,
                current_ticker_df,
                on="coinpair",
                suffixes=("_last", "_current")
            )
            # 计算15分钟内的波动值，你也可以按需替换为价格本身的涨跌幅计算逻辑
            compare_df["fluctuation"] = abs(compare_df["change_pct_current"] - compare_df["change_pct_last"])
            # 按波动从高到低排序
            compare_df = compare_df.sort_values(by="fluctuation", ascending=False)
            print("\n15分钟内波动最大的前10个币种：")
            print(compare_df[["coinpair", "change_pct_last", "change_pct_current", "fluctuation"]].head(10).to_string(index=False))
            print(f"\n本轮波动最高币种：{compare_df.iloc[0]['coinpair']}，波动幅度：{compare_df.iloc[0]['fluctuation']:.3f}%")
        
        # 更新基准数据，同时覆盖写入本地文件做持久化
        last_ticker_df = current_ticker_df.copy()
        last_ticker_df.to_csv(BASE_DATA_PATH, index=False)

        # 等待下一次拉取
        print(f"\n等待15分钟后执行下一次拉取...")
        time.sleep(FETCH_INTERVAL)

可选优化点 #

• 如果需要固定每小时的0、15、30、45分执行拉取，可替换time.sleep为schedule库的定时配置，时间精度更高
• 如果你要统计的是15分钟内价格本身的涨跌幅，而非交易所返回的24小时涨跌幅的变动，拉取数据时额外存储当前价格，对比时按(当前价-上次价)/上次价*100计算波动即可
• 如需留存所有历史拉取数据，每次拉取时可将当前数据追加写入带时间戳的历史文件中，方便后续回溯分析

内容的提问来源于stack exchange，提问作者rohan gala