Canada Stock API Integration:Real-Time TSX/TSXV Market Data & Quantitative Practices

      import requests

API_TOKEN = 'your_itick_token_here'     # ← replace with your actual token
BASE_URL  = 'https://api.itick.org'

def get_tsx_quote(symbol: str):
    """
    Retrieve real-time quote for a TSX-listed security.
    
    Args:
        symbol: TSX ticker (e.g. 'RY', 'ENB', 'BNS')
    """
    headers = {
        'accept': 'application/json',
        'token': API_TOKEN
    }
    
    # Region is fixed to 'CA' for all Canadian securities
    url = f'{BASE_URL}/stock/quote?region=CA&code={symbol}'
    
    try:
        resp = requests.get(url, headers=headers, timeout=10)
        if resp.status_code == 200:
            payload = resp.json()
            return payload.get('data', {})
        else:
            print(f"Request failed | HTTP {resp.status_code}")
            print(f"Error: {resp.text}")
            return None
    except Exception as e:
        print(f"Network error: {e}")
        return None


if __name__ == '__main__':
    quote = get_tsx_quote('RY')
    
    if quote:
        print("═" * 60)
        print("Toronto Stock Exchange (TSX) — Real-Time Quote")
        print("═" * 60)
        print(f"Ticker     : {quote.get('s', 'N/A')}")
        print(f"Last       : {quote.get('ld', 'N/A')} CAD")
        print(f"Change     : {quote.get('ch', 'N/A')}")
        print(f"Change %   : {quote.get('chp', 'N/A')}%")
        print(f"Volume     : {quote.get('v', 'N/A'):,}")
        print(f"Open       : {quote.get('o', 'N/A')}")
        print(f"High       : {quote.get('h', 'N/A')}")
        print(f"Low        : {quote.get('l', 'N/A')}")
        print(f"Timestamp  : {quote.get('t', 'N/A')}")
    else:
        print("No data returned. Check token, network, or symbol.")

    

      import websocket
import json
import threading
import time

WS_URL    = "wss://api.itick.org/stock"
API_TOKEN = 'your_itick_token_here'

class TSXRealTimeMonitor:
    def on_message(self, ws, message):
        try:
            msg = json.loads(message)
            if 'data' in msg:
                d = msg['data']
                symbol = d.get('s')
                typ    = d.get('type')
                
                if typ == 'quote':
                    print(f"[QUOTE] {symbol:6}  Last: {d.get('ld')}  Vol: {d.get('v'):,}")
                elif typ == 'tick':
                    print(f"[TRADE] {symbol:6}  Price: {d.get('ld')}  Size: {d.get('v'):,}")
        except Exception as e:
            print(f"Parse error: {e}")

    def on_open(self, ws):
        print("WebSocket connected. Subscribing to TSX symbols...")
        sub = {
            "ac": "subscribe",
            "params": "RY$CA,ENB$CA,ABX$CA",
            "types": "quote,tick"
        }
        ws.send(json.dumps(sub))

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

    def on_close(self, ws, code, msg):
        print("Connection closed")

    def run(self):
        headers = {"token": API_TOKEN}
        self.ws = websocket.WebSocketApp(
            WS_URL,
            header=headers,
            on_open=self.on_open,
            on_message=self.on_message,
            on_error=self.on_error,
            on_close=self.on_close
        )

        # Heartbeat thread (required to prevent idle disconnect)
        def heartbeat():
            while True:
                time.sleep(30)
                if self.ws.sock and self.ws.sock.connected:
                    ping = {"ac": "ping", "params": str(int(time.time()*1000))}
                    self.ws.send(json.dumps(ping))

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

if __name__ == '__main__':
    monitor = TSXRealTimeMonitor()
    monitor.run()

    

      import requests
import pandas as pd

def fetch_tsx_bars(symbol: str, ktype: int = 8, limit: int = 20):
    """
    Fetch historical OHLCV bars for a TSX-listed security.
    
    Args:
        symbol: Ticker (e.g. 'ENB')
        ktype : Bar resolution
                1=1min, 2=5min, 3=15min, 4=30min, 5=1H, 6=2H, 7=4H,
                8=Daily, 9=Weekly, 10=Monthly
        limit : Number of bars to return
    """
    headers = {'accept': 'application/json', 'token': 'your_itick_token_here'}
    
    url = f"https://api.itick.org/stock/kline?region=CA&code={symbol}&kType={ktype}&limit={limit}"
    
    resp = requests.get(url, headers=headers)
    if resp.status_code != 200:
        print(f"Failed: HTTP {resp.status_code}")
        return None
    
    bars = resp.json().get('data', [])
    if not bars:
        return None
    
    df = pd.DataFrame(bars)
    df.rename(columns={'t':'timestamp','o':'open','h':'high','l':'low','c':'close','v':'volume'}, inplace=True)
    df['datetime'] = pd.to_datetime(df['timestamp'], unit='ms')
    
    return df[['datetime','open','high','low','close','volume']]


if __name__ == '__main__':
    df = fetch_tsx_bars('ENB', ktype=8, limit=10)
    if df is not None:
        print(df.to_string(index=False))