荷蘭股票 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")