How to Use iTick API for Italian Stock Data: Real-Time Quotes & Backtesting Examples

      import websocket
import json
import threading
import time

WS_URL = "wss://api.itick.org/stock"
API_TOKEN = "your_api_token_here"  # ← Replace with your token

def on_message(ws, message):
    data = json.loads(message)

    # Connection & auth handling
    if data.get("code") == 1 and data.get("msg") == "Connected Successfully":
        print("WebSocket connected successfully")
    elif data.get("resAc") == "auth":
        if data.get("code") == 1:
            print("Authentication successful → subscribing...")
            subscribe(ws)
        else:
            print(f"Authentication failed: {data.get('msg')}")
            ws.close()

    # Subscription confirmation
    elif data.get("resAc") == "subscribe":
        print("Subscription accepted" if data.get("code") == 1 else f"Subscription failed: {data.get('msg')}")

    # Market data
    elif "data" in data:
        payload = data["data"]
        symbol = payload.get("s")
        msg_type = payload.get("type")

        if msg_type == "quote":
            print(f"[{symbol}] Last: {payload.get('ld')} | Chg%: {payload.get('chp')}% | Vol: {payload.get('v')}")
        elif msg_type == "tick":
            ts = payload.get('t', 0) / 1000
            print(f"[{symbol}] Trade @ {payload.get('ld')} | {time.strftime('%H:%M:%S', time.localtime(ts))}")
        elif msg_type == "depth":
            bids = payload.get("b", [])[:3]
            asks = payload.get("a", [])[:3]
            print(f"[{symbol}] Bids: {bids} | Asks: {asks}")

def on_error(ws, error):
    print(f"WebSocket error: {error}")

def on_close(ws, code, reason):
    print(f"WebSocket closed: {reason}")

def on_open(ws):
    print("WebSocket connection opened")

def subscribe(ws):
    sub = {
        "ac": "subscribe",
        "params": "ENI$IT,FERR$IT,ISP$IT",
        "types": "tick,quote,depth"
    }
    ws.send(json.dumps(sub))
    print(f"Sent subscription: {sub['params']}")

def heartbeat(ws):
    while True:
        time.sleep(30)
        ws.send(json.dumps({
            "ac": "ping",
            "params": str(int(time.time() * 1000))
        }))
        print("Heartbeat sent")

if __name__ == "__main__":
    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
    )

    threading.Thread(target=heartbeat, args=(ws,), daemon=True).start()

    print("Starting WebSocket client...")
    ws.run_forever()

    

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

def get_italian_kline(symbol: str, ktype: int = 8, limit: int = 100):
    url = "https://api.itick.org/stock/kline"
    params = {
        "region": "IT",
        "code": symbol,
        "kType": str(ktype),
        "limit": str(limit)
    }
    headers = {
        "accept": "application/json",
        "token": "your_api_token_here"
    }

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

        if result.get("code") == 0:
            bars = result.get("data", [])
            if not bars:
                print(f"No data returned for {symbol}")
                return None

            df = pd.DataFrame(bars)
            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], errors='coerce')

            print(f"Retrieved {len(df)} bars | Range: {df.index[0]:%Y-%m-%d} → {df.index[-1]:%Y-%m-%d}")
            print(df[['open','high','low','close','volume']].tail())

            return df

        else:
            print(f"API error: {result.get('msg')}")
            return None

    except Exception as e:
        print(f"Request failed: {e}")
        return None


def plot_price_with_ma(df, symbol):
    if df is None or df.empty:
        return

    plt.figure(figsize=(14,7))
    plt.plot(df.index, df['close'], label='Close', color='royalblue')
    df['MA20'] = df['close'].rolling(20).mean()
    plt.plot(df.index, df['MA20'], '--', label='20-day MA', color='darkred')

    plt.title(f"{symbol} – Daily Closing Price & 20-day MA")
    plt.xlabel("Date")
    plt.ylabel("Price (EUR)")
    plt.grid(True, alpha=0.3)
    plt.legend()
    plt.tight_layout()
    plt.gcf().autofmt_xdate()
    plt.show()


# Download & visualize
eni_df = get_italian_kline("ENI", ktype=8, limit=100)
if eni_df is not None:
    plot_price_with_ma(eni_df, "ENI")
    print("\nKey Statistics:")
    print(f"Latest Close:      {eni_df['close'].iloc[-1]:.3f} EUR")
    print(f"Period High:       {eni_df['high'].max():.3f} EUR")
    print(f"Period Low:        {eni_df['low'].min():.3f} EUR")
    print(f"Avg Daily Volume:  {eni_df['volume'].mean():,.0f} shares")
    print(f"Period Return:     {((eni_df['close'].iloc[-1]/eni_df['close'].iloc[0]-1)*100):+.2f}%")