Singapore Stocks API Integration Guide: SGX Real-Time Quotes & Historical Data Access

      import requests
import datetime

# Replace with your actual token
API_TOKEN = "your_api_token_here"
BASE_URL = "https://api.itick.org"

def get_singapore_stock_quote(symbol):
    """
    Retrieve real-time Level-1 quote for an SGX-listed security
    :param symbol: SGX ticker symbol, e.g. "D05" (DBS)
    """
    url = f"{BASE_URL}/stock/quote"
    params = {
        "region": "SG",      # Singapore Exchange market code
        "code": symbol
    }
    headers = {
        "accept": "application/json",
        "token": API_TOKEN
    }

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

        if result.get("code") == 0:
            data = result.get("data", {})
            print(f"📊 Security: {data.get('n', 'N/A')}")
            print(f"Ticker: {data.get('s', 'N/A')}")
            print(f"Last Traded Price: {data.get('ld', 'N/A')} SGD")
            print(f"Open: {data.get('o', 'N/A')} SGD")
            print(f"Day High: {data.get('h', 'N/A')} SGD")
            print(f"Day Low: {data.get('l', 'N/A')} SGD")
            print(f"Volume: {data.get('v', 'N/A')} shares")
            print(f"Change: {data.get('chp', 'N/A')}%")

            # Convert millisecond timestamp
            ts = data.get('t', 0) / 1000
            if ts > 0:
                dt = datetime.datetime.fromtimestamp(ts)
                print(f"Data Timestamp: {dt.strftime('%Y-%m-%d %H:%M:%S')} (Singapore local time)")
            return data
        else:
            print(f"❌ API Error: {result.get('msg', 'Unknown error')}")
            return None

    except Exception as e:
        print(f"❌ Request exception: {str(e)}")
        return None

def get_multiple_quotes(symbols):
    """Batch fetch real-time quotes for multiple tickers"""
    for symbol in symbols:
        print(f"\n{'='*50}")
        print(f"Fetching real-time quote → {symbol}")
        get_singapore_stock_quote(symbol)

# Single ticker example
print("🔍 Fetching DBS Group (D05) real-time quote:")
get_singapore_stock_quote("D05")

# Batch example
sgx_bluechips = ["D05", "O39", "Z74"]
get_multiple_quotes(sgx_bluechips)

    

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

def get_singapore_stock_kline(symbol, ktype=8, limit=100):
    """
    Fetch historical candlestick (OHLCV) data for an SGX security
    :param symbol: ticker e.g. "Z74" (Singtel)
    :param ktype: bar resolution 
                 (1=1min, 2=5min, 3=15min, 4=30min, 5=60min, 
                  8=daily, 9=weekly, 10=monthly)
    :param limit: number of bars requested
    """
    url = f"{BASE_URL}/stock/kline"
    params = {
        "region": "SG",
        "code": symbol,
        "kType": ktype,
        "limit": limit
    }
    headers = {
        "accept": "application/json",
        "token": API_TOKEN
    }

    try:
        response = requests.get(url, params=params, headers=headers)
        result = response.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['datetime'] = df['datetime'].dt.tz_localize('UTC').dt.tz_convert('Asia/Singapore')
            df.set_index('datetime', 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], errors='coerce')

            print(f"✅ Retrieved {len(df)} bars")
            print(f"Range: {df.index[0].strftime('%Y-%m-%d')} → {df.index[-1].strftime('%Y-%m-%d')}")

            print("\nLatest 5 bars:")
            print(df[['open', 'high', 'low', 'close', 'volume']].tail())

            return df
        else:
            print(f"❌ API Error: {result.get('msg', 'Unknown error')}")
            return None

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

def plot_stock_data(df, symbol):
    """Plot closing price and 20-day simple moving average"""
    if df is None or df.empty:
        return

    plt.figure(figsize=(14, 7))
    plt.plot(df.index, df['close'], 'b-', linewidth=1.5, label='Close')
    plt.title(f'{symbol} – Closing Price (SGX)', fontsize=16)
    plt.xlabel('Date')
    plt.ylabel('Price (SGD)')
    plt.grid(True, alpha=0.3)
    plt.legend()

    df['MA20'] = df['close'].rolling(window=20).mean()
    plt.plot(df.index, df['MA20'], 'r--', linewidth=1, label='20-day SMA')
    plt.legend()

    plt.gcf().autofmt_xdate()
    plt.tight_layout()
    plt.show()

# Example: fetch & visualize Singtel daily data
print("\n🔍 Fetching Singtel (Z74) historical daily bars...")
singtel_df = get_singapore_stock_kline("Z74", ktype=8, limit=100)

if singtel_df is not None:
    plot_stock_data(singtel_df, "Z74 (Singtel)")

    print("\n📈 Summary Statistics:")
    print(f"Latest Close:          {singtel_df['close'].iloc[-1]:.3f} SGD")
    print(f"Period High:           {singtel_df['high'].max():.3f} SGD")
    print(f"Period Low:            {singtel_df['low'].min():.3f} SGD")
    print(f"Average Daily Volume:  {singtel_df['volume'].mean():,.0f} shares")
    print(f"Period Total Return:   {((singtel_df['close'].iloc[-1] / singtel_df['close'].iloc[0] - 1) * 100):.2f}%")

    

      import websocket
import json
import threading
import time

WS_URL = "wss://api.itick.org/stock"
API_TOKEN = "your_api_token_here"  # ← replace

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

    # Connection established
    if data.get("code") == 1 and data.get("msg") == "Connected Successfully":
        print("✅ WebSocket connected – awaiting authentication...")

    # Authentication result
    elif data.get("resAc") == "auth":
        if data.get("code") == 1:
            print("✅ Authentication successful – subscribing...")
            subscribe(ws)
        else:
            print("❌ Authentication failed")
            ws.close()

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

    # Market data payload
    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}] LTP: {market_data.get('ld')} | Chg%: {market_data.get('chp')}% | Vol: {market_data.get('v')}")
        elif data_type == "tick":
            print(f"[{symbol}] Trade: {market_data.get('ld')} | {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}] Bid: {bids} | Ask: {asks}")

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

def on_close(ws, close_status_code, close_msg):
    print(f"🔌 WebSocket closed: {close_msg}")

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

def subscribe(ws):
    subscribe_msg = {
        "ac": "subscribe",
        "params": "D05$SG,O39$SG,Z74$SG",
        "types": "tick,quote,depth"
    }
    ws.send(json.dumps(subscribe_msg))
    print(f"📤 Subscription sent: {subscribe_msg['params']}")

def send_ping(ws):
    while True:
        time.sleep(30)
        ping_msg = {
            "ac": "ping",
            "params": str(int(time.time() * 1000))
        }
        ws.send(json.dumps(ping_msg))
        print("💓 Ping 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
    )

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

    print("🚀 Starting WebSocket connection...")
    ws.run_forever()

    

      def get_singapore_tick_data(symbol):
    """
    Retrieve latest tick (last trade) information
    :param symbol: e.g. "D05"
    """
    url = f"{BASE_URL}/stock/tick"
    params = {
        "region": "SG",
        "code": symbol
    }
    headers = {
        "accept": "application/json",
        "token": API_TOKEN
    }

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

        if result.get("code") == 0:
            tick = result.get("data", {})
            print(f"📊 Ticker: {tick.get('s', 'N/A')}")
            print(f"Last Trade Price: {tick.get('ld', 'N/A')} SGD")
            print(f"Last Trade Size: {tick.get('v', 'N/A')} shares")

            ts = tick.get('t', 0) / 1000
            if ts > 0:
                dt = datetime.datetime.fromtimestamp(ts)
                print(f"Trade Timestamp: {dt.strftime('%Y-%m-%d %H:%M:%S')} (SG time)")
            return tick
        else:
            print(f"❌ API Error: {result.get('msg', 'Unknown error')}")
            return None

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

print("\n🔍 Fetching latest tick for DBS (D05):")
get_singapore_tick_data("D05")