荷兰股票 API 接入教程：实时行情与历史 K 线数据完整实现

      import websocket
import json
import threading
import time

# WebSocket连接地址和token
WS_URL = "wss://api.itick.org/stock"
API_TOKEN = "your_token_here"

def on_message(ws, message):
    """处理接收到的消息"""
    data = json.loads(message)

    # 处理连接成功消息
    if data.get("code") == 1 and data.get("msg") == "Connected Successfully":
        print("✅ WebSocket连接成功，等待认证...")

    # 处理认证结果
    elif data.get("resAc") == "auth":
        if data.get("code") == 1:
            print("✅ 认证成功，开始订阅...")
            subscribe(ws)
        else:
            print("❌ 认证失败")
            ws.close()

    # 处理订阅结果
    elif data.get("resAc") == "subscribe":
        if data.get("code") == 1:
            print("✅ 订阅成功")
        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 == "quote":
            print(f"[{symbol}] 最新价: {market_data.get('ld')} EUR | "
                  f"涨跌幅: {market_data.get('chp')}% | "
                  f"成交量: {market_data.get('v')}")
        elif data_type == "tick":
            print(f"[{symbol}] 成交: {market_data.get('ld')} EUR")
        elif data_type == "depth":
            bids = market_data.get("b", [])[:3]
            asks = market_data.get("a", [])[:3]
            print(f"[{symbol}] 买盘: {bids} | 卖盘: {asks}")

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

def on_close(ws, close_status_code, close_msg):
    print(f"🔌 连接关闭: {close_msg}，5秒后重连...")
    time.sleep(5)
    connect()

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

def subscribe(ws):
    """订阅荷兰股票行情"""
    subscribe_msg = {
        "ac": "subscribe",
        "params": "UNA$NL,ASML$NL,INGA$NL",  # 联合利华、ASML、ING集团
        "types": "tick,quote,depth"
    }
    ws.send(json.dumps(subscribe_msg))
    print(f"📤 订阅请求已发送")

def send_ping(ws):
    """定期发送心跳包保持连接"""
    while True:
        time.sleep(30)
        if ws and ws.sock and ws.sock.connected:
            ping_msg = {
                "ac": "ping",
                "params": str(int(time.time() * 1000))
            }
            ws.send(json.dumps(ping_msg))
            print("💓 心跳已发送")

def connect():
    """建立WebSocket连接"""
    global ws
    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,), daemon=True)
    ping_thread.start()

    # 运行连接
    ws.run_forever()

if __name__ == "__main__":
    connect()

    

      import requests
import pandas as pd

API_TOKEN = "your_token_here"
BASE_URL = "https://api.itick.org"
HEADERS = {"token": API_TOKEN, "accept": "application/json"}

def get_kline(symbol, ktype=8, limit=100):
    """
    获取荷兰股票历史K线数据
    :param symbol: 股票代码，如"ASML"
    :param ktype: K线类型，8=日线
    :param limit: 获取数量
    :return: K线数据列表
    """
    url = f"{BASE_URL}/stock/kline"
    params = {
        "region": "NL",
        "code": symbol,
        "kType": ktype,
        "limit": str(limit)
    }

    resp = requests.get(url, params=params, headers=HEADERS)
    data = resp.json()

    if data.get("code") == 0:
        klines = data.get("data", [])

        # 周期类型映射
        ktype_map = {
            1: "1分钟", 2: "5分钟", 3: "15分钟", 4: "30分钟", 5: "60分钟",
            8: "日线", 9: "周线", 10: "月线"
        }

        print(f"✅ 获取到 {len(klines)} 条{ktype_map.get(ktype, '')}数据")

        # 打印最近3条数据
        for item in klines[-3:]:
            print(f"时间:{item['t']} 开:{item['o']} 高:{item['h']} 低:{item['l']} 收:{item['c']} 量:{item['v']}")

        return klines
    else:
        print(f"❌ 错误: {data.get('msg')}")
        return []

# 示例1：获取ASML日线数据
print("📊 ASML 日线数据:")
asml_daily = get_kline("ASML", ktype=8, limit=30)

# 示例2：获取联合利华周线数据
print("\n📊 联合利华 周线数据:")
una_weekly = get_kline("UNA", ktype=9, limit=20)

# 示例3：获取ING集团60分钟线
print("\n📊 ING集团 60分钟线:")
inga_hourly = get_kline("INGA", ktype=5, limit=50)

    

      def kline_to_dataframe(klines):
    """将K线数据转换为Pandas DataFrame"""
    if not klines:
        return None

    df = pd.DataFrame(klines)
    df['timestamp'] = pd.to_datetime(df['t'], unit='ms')
    df.set_index('timestamp', inplace=True)
    df.rename(columns={
        'o': 'open', 'h': 'high', 'l': 'low',
        'c': 'close', 'v': 'volume'
    }, inplace=True)

    # 转换为数值类型
    for col in ['open', 'high', 'low', 'close', 'volume']:
        df[col] = pd.to_numeric(df[col])

    return df[['open', 'high', 'low', 'close', 'volume']]

# 示例：获取ASML数据并转为DataFrame
klines = get_kline("ASML", ktype=8, limit=100)
if klines:
    df = kline_to_dataframe(klines)
    print("\n📈 DataFrame前5行:")
    print(df.head())

    # 简单统计
    print(f"\n📊 统计信息:")
    print(f"最新收盘价: {df['close'].iloc[-1]:.2f} EUR")
    print(f"期间最高: {df['high'].max():.2f} EUR")
    print(f"期间最低: {df['low'].min():.2f} EUR")