意大利股票API接入指南:如何获取米兰证券交易所实时行情与历史数据

      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}%")