如何实现加密货币晴雨表：获取全币种多时段价格并统计总和

核心实现逻辑 #

• 交易对过滤：仅保留币安返回数据中以USDT结尾的现货交易对，排除其他计价币种、杠杆代币等无效标的，避免总价计算失真
• 历史价格获取：调用币安1小时级K线接口，传入对应历史时间的毫秒级时间戳，取单根K线收盘价作为对应时间点的历史价格
• 结果格式化：固定每列输出宽度实现多列对齐，按需求输出各币种价格及各时段总价

完整可运行代码 #

import requests
import json
import datetime
import time

def get_history_price(symbol, hours_ago):
    """获取指定币种n小时前的历史价格"""
    try:
        end_time = int((datetime.datetime.now() - datetime.timedelta(hours=hours_ago)).timestamp() * 1000)
        params = {
            "symbol": symbol,
            "interval": "1h",
            "endTime": end_time,
            "limit": 1
        }
        resp = requests.get("https://api.binance.com/api/v3/klines", params=params, timeout=5)
        kline_data = json.loads(resp.content)
        if kline_data:
            return float(kline_data[0][4])
        return 0.0
    except Exception as e:
        # 新币无历史数据、请求失败等场景默认返回0
        return 0.0

while True:
    # 获取实时价格数据
    api_request = requests.get("https://api.binance.com/api/v3/ticker/price", timeout=10)
    all_tickers = json.loads(api_request.content)
    
    # 过滤USDT交易对，可自行增加条件仅保留主流币种提升请求效率
    usdt_tickers = [t for t in all_tickers if t["symbol"].endswith("USDT") and "UP" not in t["symbol"] and "DOWN" not in t["symbol"]]
    
    # 存储各时段数据 结构：{symbol: {"current": xx, "1h": xx, "4h": xx, "8h": xx, "24h": xx}}
    price_data = {}
    sum_current = 0.0
    sum_1h = 0.0
    sum_4h = 0.0
    sum_8h = 0.0
    sum_24h = 0.0

    for ticker in usdt_tickers:
        symbol = ticker["symbol"]
        current_price = float(ticker["price"])
        h1_price = get_history_price(symbol, 1)
        h4_price = get_history_price(symbol, 4)
        h8_price = get_history_price(symbol, 8)
        h24_price = get_history_price(symbol, 24)
        
        price_data[symbol] = {
            "current": current_price,
            "1h": h1_price,
            "4h": h4_price,
            "8h": h8_price,
            "24h": h24_price
        }

        # 累加各时段总价
        sum_current += current_price
        sum_1h += h1_price
        sum_4h += h4_price
        sum_8h += h8_price
        sum_24h += h24_price
        # 全量请求可加0.05s间隔避免触发限流，仅查主流币可删除
        time.sleep(0.05)

    # 格式化输出
    print(f"{'Current':<22} {'1Hour':<22} {'4Hours':<22} {'8Hours':<22} {'24Hours':<22}")
    print("-"*110)
    for symbol, prices in price_data.items():
        cur_str = f"{symbol} ${prices['current']:.2f}"
        h1_str = f"{symbol} ${prices['1h']:.2f}"
        h4_str = f"{symbol} ${prices['4h']:.2f}"
        h8_str = f"{symbol} ${prices['8h']:.2f}"
        h24_str = f"{symbol} ${prices['24h']:.2f}"
        print(f"{cur_str:<22} {h1_str:<22} {h4_str:<22} {h8_str:<22} {h24_str:<22}")
    print("-"*110)
    sum_cur_str = f"sum : ${sum_current:.2f}"
    sum_1h_str = f"sum : ${sum_1h:.2f}"
    sum_4h_str = f"sum : ${sum_4h:.2f}"
    sum_8h_str = f"sum : ${sum_8h:.2f}"
    sum_24h_str = f"sum : ${sum_24h:.2f}"
    print(f"{sum_cur_str:<22} {sum_1h_str:<22} {sum_4h_str:<22} {sum_8h_str:<22} {sum_24h_str:<22}")

    # 每分钟刷新一次，可自行调整间隔
    time.sleep(60)

注意事项 #

• 全量拉取数千个交易对的历史数据耗时较长，建议仅保留BTC、ETH等主流币种进行查询，可在usdt_tickers过滤逻辑中增加筛选条件
• 币安公开API有请求速率限制，若触发限流可适当调高请求间隔时间，或申请API_KEY使用更高权限的接口
• 部分新上线币种不足24小时交易历史，默认返回0价，可根据需求修改异常处理逻辑，返回N/A等提示字符

内容的提问来源于stack exchange，提问作者Collin G.