期货数据API对接实战指南

news2025/4/18 23:01:40

一、期货数据接口概述

StockTV提供全球主要期货市场的实时行情与历史数据接口，覆盖以下品种：

商品期货：原油、黄金、白银、铜、天然气、农产品等
金融期货：股指期货、国债期货
特色品种：马棕油、铁矿石等区域特色期货

二、环境准备与配置

1. API密钥获取

API_KEY = "your_futures_api_key"  # 通过官网申请
BASE_URL = "https://api.stocktv.top"

2. 安装必要库

pip install requests pandas matplotlib websocket-client

三、期货行情数据对接

1. 获取期货合约列表

def get_futures_list():
    """获取可交易期货合约列表"""
    url = f"{BASE_URL}/futures/list"
    params = {"key": API_KEY}
    response = requests.get(url, params=params)
    return response.json()

# 示例调用
futures_list = get_futures_list()
print("可用期货合约:", [f"{x['symbol']} ({x['name']})" for x in futures_list['data'][:5]])

2. 查询特定合约行情

def get_futures_quote(symbol):
    """获取期货合约实时行情"""
    url = f"{BASE_URL}/futures/quote"
    params = {
        "symbol": symbol,
        "key": API_KEY
    }
    response = requests.get(url, params=params)
    return response.json()

# 获取原油期货行情
crude_oil = get_futures_quote("CL1!")
print(f"WTI原油最新价: {crude_oil['data']['last']} 涨跌: {crude_oil['data']['change']}")

四、期货K线数据获取

1. 历史K线数据接口

def get_futures_kline(symbol, interval="1d", limit=100):
    """
    获取期货K线数据
    :param symbol: 合约代码
    :param interval: 时间间隔(1m/5m/15m/1h/1d)
    :param limit: 数据条数
    """
    url = f"{BASE_URL}/futures/kline"
    params = {
        "symbol": symbol,
        "interval": interval,
        "limit": limit,
        "key": API_KEY
    }
    response = requests.get(url, params=params)
    data = response.json()
    
    # 转换为DataFrame
    df = pd.DataFrame(data['data'])
    df['time'] = pd.to_datetime(df['time'], unit='ms')
    return df

# 获取黄金期货15分钟K线
gold_kline = get_futures_kline("GC1!", "15m")

2. K线数据可视化

import matplotlib.pyplot as plt

def plot_futures_kline(df, title):
    plt.figure(figsize=(12,6))
    plt.title(title)
    
    # 绘制蜡烛图
    for i, row in df.iterrows():
        color = 'red' if row['close'] > row['open'] else 'green'
        plt.plot([i, i], [row['low'], row['high']], color=color)
        plt.plot([i-0.2, i+0.2], [row['open'], row['open']], color=color)
        plt.plot([i-0.2, i+0.2], [row['close'], row['close']], color=color)
    
    plt.xlabel('时间')
    plt.ylabel('价格')
    plt.grid()
    plt.show()

plot_futures_kline(gold_kline, "COMEX黄金期货15分钟K线")

五、期货交易数据存储方案

1. 数据库设计（SQL示例）

import sqlite3

def init_db():
    conn = sqlite3.connect('futures_data.db')
    c = conn.cursor()
    
    c.execute('''CREATE TABLE IF NOT EXISTS futures_quotes
                 (symbol text, last real, volume integer, 
                  time timestamp, PRIMARY KEY (symbol, time))''')
    
    c.execute('''CREATE TABLE IF NOT EXISTS futures_kline
                 (symbol text, open real, high real, low real, 
                  close real, volume integer, time timestamp,
                  PRIMARY KEY (symbol, time))''')
    
    conn.commit()
    conn.close()

init_db()

2. 数据存储实现

def save_futures_quote(data):
    conn = sqlite3.connect('futures_data.db')
    c = conn.cursor()
    
    c.execute('''INSERT INTO futures_quotes 
                 VALUES (?, ?, ?, ?)''',
              (data['symbol'], data['last'], data['volume'], data['time']))
    
    conn.commit()
    conn.close()

def save_futures_kline(symbol, kline_data):
    conn = sqlite3.connect('futures_data.db')
    c = conn.cursor()
    
    for row in kline_data:
        c.execute('''INSERT INTO futures_kline 
                     VALUES (?, ?, ?, ?, ?, ?, ?)''',
                  (symbol, row['open'], row['high'], row['low'],
                   row['close'], row['volume'], row['time']))
    
    conn.commit()
    conn.close()

六、生产环境注意事项

错误处理与重试机制

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), 
       wait=wait_exponential(multiplier=1, min=2, max=10))
def safe_futures_api_call(url, params):
    try:
        response = requests.get(url, params=params, timeout=5)
        response.raise_for_status()
        return response.json()
    except Exception as e:
        print(f"API调用失败: {e}")
        raise

性能优化建议

使用Redis缓存高频访问的合约信息
批量获取多个合约数据减少API调用次数
对历史K线数据实现本地存储

七、完整示例：期货监控系统

import schedule
import time

class FuturesMonitor:
    def __init__(self):
        self.tracked_symbols = ["CL1!", "GC1!"]
    
    def update_data(self):
        for symbol in self.tracked_symbols:
            # 获取实时行情
            quote = get_futures_quote(symbol)
            print(f"{symbol} 最新价: {quote['data']['last']}")
            
            # 获取K线数据
            kline = get_futures_kline(symbol, "15m")
            print(f"最近3根K线: {kline.tail(3)}")
            
            # 存储数据
            save_futures_quote(quote['data'])
    
    def run(self):
        # 每15秒更新一次数据
        schedule.every(15).seconds.do(self.update_data)
        
        while True:
            schedule.run_pending()
            time.sleep(1)

# 启动监控
monitor = FuturesMonitor()
monitor.run()