股指期货 API 入门指南：如何获取实时行情与构建交易系统

pip install requests pandas

import requests
import pandas as pd

def get_future_quote(token, region, code):
    url = f"https://api.itick.org/future/quote?region={region}&code={code}"
    headers = {"accept": "application/json", "token": token}
    try:
        response = requests.get(url, headers=headers)
        response.raise_for_status()  # 检查请求是否成功
        data = response.json()["data"]
        # 转换为DataFrame方便查看
        df = pd.DataFrame([data])
        return df
    except Exception as e:
        print(f"获取实时报价失败：{str(e)}")
        return None

def get_future_kline(token, region, code, ktype, limit):
    url = f"https://api.itick.org/future/kline?region={region}&code={code}&kType={ktype}&limit={limit}"
    headers = {"accept": "application/json", "token": token}
    try:
        response = requests.get(url, headers=headers)
        response.raise_for_status()  # 检查请求是否成功
        data = response.json()["data"]
        # 转换为DataFrame方便处理
        df = pd.DataFrame(data)
        df.columns = ["时间戳", "开盘价", "最高价", "最低价", "收盘价", "成交数量", "成交额"]
        return df
    except Exception as e:
        print(f"获取K线数据失败：{str(e)}")
        return None

# 调用示例：获取沪深300股指期货（IF）实时报价（假设region=CN）
token = "你的API密钥"
quote_df = get_future_quote(token, region="CN", code="IF")
print("实时报价：")
print(quote_df.head())

# 调用示例：获取沪深300股指期货（IF）1分钟K线，共100根
kline_df = get_future_kline(token, region="CN", code="IF", ktype=1, limit=100)
print("历史K线：")
print(kline_df.head())

import websocket
import json
import threading
import time

# WebSocket 连接地址和 token
WS_URL = "wss://api.itick.org/future"
API_TOKEN = "你的API密钥"

def on_message(ws, message):
    """处理接收到的消息"""
    data = json.loads(message)
    if data.get("code") == 1 and data.get("msg") == "Connected Successfully":
        print("连接成功")
    elif data.get("resAc") == "auth" and data.get("code") == 1:
        print("认证成功")
        subscribe(ws, code="IF")  # 订阅沪深300股指期货
    elif data.get("resAc") == "subscribe" and data.get("code") == 1:
        print("订阅成功")
    elif data.get("data"):
        market_data = data["data"]
        data_type = market_data.get("type")
        symbol = market_data.get("s")
        print(f"{data_type.upper()} 数据 for {symbol}: {market_data}")

def on_error(ws, error):
    print("错误:", error)

def on_close(ws, close_status_code, close_msg):
    print("连接关闭")

def on_open(ws):
    print("WebSocket 连接打开")

def subscribe(ws, code):
    subscribe_msg = {
        "ac": "subscribe",
        "params": code,  # 如"IF"或多个用逗号分隔
        "types": "quote,tick,depth"  # 订阅类型：quote报价、tick成交、depth盘口
    }
    ws.send(json.dumps(subscribe_msg))
    print("订阅消息已发送")

def send_ping(ws):
    """定期发送心跳包"""
    while True:
        time.sleep(30)  # 每30秒发送一次
        ping_msg = {
            "ac": "ping",
            "params": str(int(time.time() * 1000))
        }
        ws.send(json.dumps(ping_msg))
        print("心跳已发送")

if __name__ == "__main__":
    ws = websocket.WebSocketApp(
        WS_URL,
        header={"token": API_TOKEN},
        on_open=on_open,
        on_message=on_message,
        on_error=on_error,
        on_close=on_close
    )
    # 启动心跳线程
    ping_thread = threading.Thread(target=send_ping, args=(ws,))
    ping_thread.daemon = True
    ping_thread.start()
    # 启动 WebSocket
    ws.run_forever()