期货 Tick 级数据与基金净值历史数据 API 接口详解

import requests

# API 配置
url = "https://api.itick.org/future/tick?region=US&code=NQ"  # NQ 为纳斯达克指数期货
headers = {
    "accept": "application/json",
    "token": "your_api_token"  # 替换为你的实际 Token
}

# 发送请求
response = requests.get(url, headers=headers)

if response.status_code == 200:
    data = response.json()
    if data.get("code") == 0:
        tick = data["data"]
        print(f"合约: {tick['s']}")
        print(f"最新价: {tick['ld']}")
        print(f"成交量: {tick['v']}")
        print(f"时间戳: {tick['t']}")
    else:
        print("API 错误:", data.get("msg"))
else:
    print("HTTP 错误:", response.status_code)

{
  "code": 0,
  "msg": null,
  "data": {
    "s": "NQ",
    "ld": 225.215,
    "t": 1754554087000,
    "v": 1134500
  }
}

import requests

# kType 参数说明：
# 1-1分钟、2-5分钟、3-15分钟、4-30分钟、5-60分钟、
# 6-日线、7-周线、8-月线、9-季线、10-年线

url = "https://api.itick.org/future/kline?region=US&code=NQ&kType=2&limit=10"
headers = {
    "accept": "application/json",
    "token": "your_api_token"
}

response = requests.get(url, headers=headers)

if response.status_code == 200:
    data = response.json()
    if data.get("code") == 0:
        klines = data["data"]
        for kline in klines:
            print(f"时间: {kline['t']}, 开盘: {kline['o']}, "
                  f"最高: {kline['h']}, 最低: {kline['l']}, "
                  f"收盘: {kline['c']}, 成交量: {kline['v']}")

import requests

url = "https://api.itick.org/fund/quote?region=US&code=QQQ"
headers = {
    "accept": "application/json",
    "token": "your_api_token"
}

response = requests.get(url, headers=headers)

if response.status_code == 200:
    data = response.json()
    if data.get("code") == 0:
        quote = data["data"]
        print(f"基金代码: {quote['s']}")
        print(f"最新净值: {quote['ld']}")
        print(f"开盘价: {quote.get('o')}")
        print(f"涨跌幅: {quote.get('chp')}%")

import requests

# 获取 QQQ 最近 30 条日线数据
url = "https://api.itick.org/fund/kline?region=US&code=QQQ&kType=6&limit=30"
headers = {
    "accept": "application/json",
    "token": "your_api_token"
}

response = requests.get(url, headers=headers)

if response.status_code == 200:
    data = response.json()
    if data.get("code") == 0:
        for nav in data["data"]:
            print(f"日期: {nav['t']}, 净值: {nav['c']}")

import websocket
import json
import threading
import time

WS_URL = "wss://api.itick.org/future"  # 期货数据端点，可替换为 stock/fund/forex
API_TOKEN = "your_api_token"

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":
        if data.get("code") == 1:
            print("✅ 认证通过，开始订阅数据...")
            subscribe(ws)
        else:
            print(f"❌ 认证失败：{data.get('msg')}")
    
    # 实时数据处理
    elif data.get("data"):
        market_data = data["data"]
        data_type = market_data.get("type")
        symbol = market_data.get("s")
        
        if data_type == "tick":
            print(f"📊 {symbol} Tick数据 - 最新价: {market_data['ld']}, 成交量: {market_data['v']}")
        elif data_type == "quote":
            print(f"📈 {symbol} 报价 - 最新: {market_data['ld']}, 涨跌幅: {market_data.get('chp')}%")
        elif data_type == "depth":
            print(f"📚 {symbol} 盘口深度 - 买一: {market_data['bids'][0] if market_data.get('bids') else 'N/A'}")

def on_error(ws, error):
    print(f"❌ 连接错误：{error}")

def on_close(ws, close_status_code, close_msg):
    """连接关闭时自动重连"""
    print(f"🔌 连接关闭，3秒后自动重连...")
    time.sleep(3)
    start_websocket()

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

def subscribe(ws):
    """发送订阅请求"""
    subscribe_msg = {
        "ac": "subscribe",
        "params": "NQ$US,CL$US",  # 多个标的用逗号分隔，格式：code$region
        "types": "tick,quote"      # 订阅数据类型
    }
    ws.send(json.dumps(subscribe_msg))
    print("📡 订阅消息已发送")

def send_ping(ws):
    """每30秒发送心跳包维持连接"""
    while True:
        time.sleep(30)
        try:
            ping_msg = {
                "ac": "ping",
                "params": str(int(time.time() * 1000))
            }
            ws.send(json.dumps(ping_msg))
        except Exception as e:
            print(f"❌ 发送心跳包失败：{e}")

def start_websocket():
    """启动 WebSocket 连接"""
    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()
    
    ws.run_forever()

if __name__ == "__main__":
    print("🚀 启动实时数据接收程序...")
    start_websocket()

import pandas as pd
from datetime import datetime

def aggregate_ticks_to_daily(ticks):
    """将 Tick 数据聚合成日线 OHLC"""
    df = pd.DataFrame(ticks)
    df['timestamp'] = pd.to_datetime(df['t'], unit='ms')
    df['date'] = df['timestamp'].dt.date
    
    daily = df.groupby('date').agg({
        'ld': ['first', 'max', 'min', 'last'],
        'v': 'sum'
    }).round(2)
    
    daily.columns = ['open', 'high', 'low', 'close', 'volume']
    return daily

def correlate_future_and_fund(future_ticks, fund_navs):
    """关联期货 Tick 聚合数据与基金净值"""
    daily_future = aggregate_ticks_to_daily(future_ticks)
    merged = pd.merge(daily_future, fund_navs, left_index=True, right_index=True)
    
    # 计算相关性
    correlation = merged['close'].corr(merged['nav'])
    print(f"期货收盘价与基金净值的相关性: {correlation:.4f}")
    
    return merged