如何用Python实现币安XRP/USDT期货实时价格监控及跌幅告警

核心实现步骤 #

• 替换为实时价格接口：当前K线接口适合获取历史数据，实时价格用/fapi/v1/ticker/price接口更高效，能拿到最新成交价。
• 维护过去一小时最高价：
  1. 初始化时拉取最近1小时K线的最高价作为初始值；
  2. 后续实时获取价格时，若当前价超过记录的最高价，更新最高价；
  3. 每到整点（或每小时）重新拉取K线数据，重置最高价，确保时间窗口始终是过去一小时。
• 循环持续监控：用while True构建循环，加入合理延迟（比如1-2秒），避免触发币安API限流。
• 下跌触发判断：每次获取实时价格后，检查是否满足当前价格 <= 过去一小时最高价 * 0.99，满足则输出提示。

优化后的完整代码 #

import requests
import time
from datetime import datetime

# 币安测试网API地址
BASE_URL = 'https://testnet.binancefuture.com'
SYMBOL = 'XRPUSDT'

def get_hourly_high():
    """获取过去一小时的最高价"""
    params = {'symbol': SYMBOL, 'interval': '1h', 'limit': 1}
    response = requests.get(f"{BASE_URL}/fapi/v1/klines", params=params)
    if response.status_code == 200:
        # K线数据中第2位是最高价（索引从0开始）
        return float(response.json()[0][2])
    else:
        print("获取小时K线失败")
        return None

def get_current_price():
    """获取当前实时价格"""
    params = {'symbol': SYMBOL}
    response = requests.get(f"{BASE_URL}/fapi/v1/ticker/price", params=params)
    if response.status_code == 200:
        return float(response.json()['price'])
    else:
        print("获取实时价格失败")
        return None

def main():
    # 初始化过去一小时最高价
    hourly_high = get_hourly_high()
    if not hourly_high:
        return
    
    # 记录上一次更新最高价的整点时间
    last_hour = datetime.now().hour
    
    print(f"开始监控XRP/USDT期货价格，初始一小时最高价: {hourly_high:.4f}")
    
    while True:
        current_price = get_current_price()
        if not current_price:
            time.sleep(2)
            continue
        
        # 检查是否进入新的小时，重置最高价
        current_hour = datetime.now().hour
        if current_hour != last_hour:
            new_hourly_high = get_hourly_high()
            if new_hourly_high:
                hourly_high = new_hourly_high
                last_hour = current_hour
                print(f"\n进入新小时，更新一小时最高价为: {hourly_high:.4f}")
        
        # 更新最高价：如果当前价格超过记录的最高价，更新
        if current_price > hourly_high:
            hourly_high = current_price
            print(f"更新一小时最高价为: {hourly_high:.4f}")
        
        # 判断是否下跌1%
        if current_price <= hourly_high * 0.99:
            print(f"\n⚠️ 价格预警！当前价格 {current_price:.4f}，较一小时最高价 {hourly_high:.4f} 下跌超1%")
        
        # 控制请求频率，避免限流
        time.sleep(1)

if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("\n监控已停止")

代码说明 #

• get_hourly_high：拉取最近1小时K线的最高价，用于初始化和每小时重置。
• get_current_price：调用实时价格接口，获取最新成交价。
• 循环逻辑：持续获取价格，每小时重置一次最高价，同时实时更新最高价（如果当前价更高），满足下跌条件时输出提示。
• 限流处理：加入time.sleep(1)控制请求频率，避免触发币安API的限流规则。

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