義大利股票 API 接入指南：使用 iTick 獲取即時行情與歷史資料

      import websocket
import json
import threading
import time

# WebSocket 連線地址與 token
WS_URL = "wss://api.itick.org/stock"
API_TOKEN = "your_api_token_here"  # 請替換為您的實際 Token

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 | "
                  f"時間: {time.strftime('%H:%M:%S', time.localtime(market_data.get('t')/1000))}")
        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"❌ WebSocket 錯誤: {error}")

def on_close(ws, close_status_code, close_msg):
    """連線關閉回調"""
    print(f"🔌 WebSocket 連線關閉: {close_msg}")

def on_open(ws):
    """連線建立後的回調"""
    print("🌐 WebSocket 連線已開啟")

def subscribe(ws):
    """訂閱義大利股票行情資料"""
    subscribe_msg = {
        "ac": "subscribe",
        "params": "ENI$IT,FERR$IT,ISP$IT",  # 義大利股票代碼：埃尼、法拉利、聯合聖保羅銀行
        "types": "tick,quote,depth"          # 訂閱成交、報價與盤口資料
    }
    ws.send(json.dumps(subscribe_msg))
    print(f"📤 訂閱請求已發送: {subscribe_msg['params']}")

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__":
    # 建立 WebSocket 連線，透過 header 傳遞 token
    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()

    print("🚀 啟動 WebSocket 連線...")
    # 啟動 WebSocket 連線
    ws.run_forever()

    

      import requests
import pandas as pd
import matplotlib.pyplot as plt
from datetime import datetime

def get_italian_stock_kline(symbol, ktype=8, limit=100):
    """
    獲取義大利股票歷史 K 線資料
    :param symbol: 股票代碼，如 "ENI"
    :param ktype: K 線類型，8 為日線
    :param limit: 獲取 K 線數量
    :return: DataFrame 格式的 K 線資料
    """
    url = "https://api.itick.org/stock/kline"
    params = {
        "region": "IT",      # 義大利市場代碼
        "code": symbol,
        "kType": ktype,
        "limit": limit
    }
    headers = {
        "accept": "application/json",
        "token": "your_api_token_here"  # 請替換為您的實際 Token
    }

    try:
        response = requests.get(url, params=params, headers=headers)
        result = response.json()

        if result.get("code") == 0:  # 0 表示成功
            kline_data = result.get("data", [])

            if not kline_data:
                print(f"❌ 未獲取到 {symbol} 的資料")
                return None

            # 轉換為 DataFrame
            df = pd.DataFrame(kline_data)

            # 轉換時間戳並設定索引
            df['datetime'] = pd.to_datetime(df['t'], unit='ms')
            df.set_index('datetime', inplace=True)

            # 重新命名欄位
            df.rename(columns={
                'o': 'open', 'h': 'high', 'l': 'low',
                'c': 'close', 'v': 'volume', 'tu': 'turnover'
            }, inplace=True)

            # 轉換為數值型態
            for col in ['open', 'high', 'low', 'close', 'volume', 'turnover']:
                if col in df.columns:
                    df[col] = pd.to_numeric(df[col])

            print(f"✅ 獲取到 {len(df)} 條 K 線資料")
            print(f"資料範圍: {df.index[0]} 至 {df.index[-1]}")

            # 顯示最新 5 條資料
            print("\n📊 最新 5 條 K 線資料：")
            print(df[['open', 'high', 'low', 'close', 'volume']].tail())

            return df
        else:
            print(f"❌ API 錯誤: {result.get('msg', '未知錯誤')}")
            return None

    except Exception as e:
        print(f"❌ 請求異常: {str(e)}")
        return None

def plot_stock_data(df, symbol):
    """繪製股票 K 線收盤價走勢"""
    if df is None or df.empty:
        return

    plt.figure(figsize=(14, 7))
    plt.plot(df.index, df['close'], 'b-', linewidth=1.5, label='收盤價')
    plt.title(f'{symbol} 收盤價走勢圖', fontsize=16)
    plt.xlabel('日期')
    plt.ylabel('價格 (EUR)')
    plt.grid(True, alpha=0.3)
    plt.legend()

    # 格式化 x 軸日期
    plt.gcf().autofmt_xdate()

    # 加入簡單移動平均線
    df['MA20'] = df['close'].rolling(window=20).mean()
    plt.plot(df.index, df['MA20'], 'r--', linewidth=1, label='20日均線')
    plt.legend()

    plt.tight_layout()
    plt.show()

# 獲取埃尼公司最近 100 個交易日資料
eni_df = get_italian_stock_kline("ENI", ktype=8, limit=100)

if eni_df is not None:
    # 繪製走勢圖
    plot_stock_data(eni_df, "ENI (埃尼)")

    # 計算簡單統計指標
    print(f"\n📈 統計指標：")
    print(f"最新收盤價: {eni_df['close'].iloc[-1]:.3f} EUR")
    print(f"期間最高價: {eni_df['high'].max():.3f} EUR")
    print(f"期間最低價: {eni_df['low'].min():.3f} EUR")
    print(f"平均成交量: {eni_df['volume'].mean():.0f} 股")
    print(f"期間漲跌幅: {((eni_df['close'].iloc[-1] / eni_df['close'].iloc[0] - 1) * 100):.2f}%")