加拿大股票API對接：多倫多證券交易所（TSX）即時行情與量化實踐

      import requests
import json

# 配置區域
API_TOKEN = 'your_itick_token_here'  # 替換為你的真实Token
BASE_URL = 'https://api.itick.org'

def get_tsx_quote(symbol):
    """
    獲取多倫多交易所(TSX)股票即時報價
    :param symbol: TSX證券代碼，如 RY, ENB, BNS
    """
    headers = {
        'accept': 'application/json',
        'token': API_TOKEN
    }
    
    # 加拿大市場region固定為CA
    url = f'{BASE_URL}/stock/quote?region=CA&code={symbol}'
    
    try:
        response = requests.get(url, headers=headers, timeout=10)
        if response.status_code == 200:
            data = response.json()
            quote = data.get('data', {})
            return quote
        else:
            print(f"請求失敗 | 狀態碼: {response.status_code}")
            print(f"錯誤詳情: {response.text}")
            return None
    except Exception as e:
        print(f"網路異常: {str(e)}")
        return None

# 實戰調用：加拿大皇家銀行 RY
if __name__ == '__main__':
    quote = get_tsx_quote('RY')
    
    if quote:
        print("=" * 60)
        print(f"多倫多證券交易所 TSX - 即時行情")
        print("=" * 60)
        print(f"證券代碼: {quote.get('s', 'N/A')}")
        print(f"最新價: {quote.get('ld', 'N/A')} CAD")
        print(f"漲跌額: {quote.get('ch', 'N/A')}")
        print(f"漲跌幅: {quote.get('chp', 'N/A')}%")
        print(f"成交量: {quote.get('v', 'N/A')}")
        print(f"開盤: {quote.get('o', 'N/A')}")
        print(f"最高: {quote.get('h', 'N/A')}")
        print(f"最低: {quote.get('l', 'N/A')}")
        print(f"時間戳: {quote.get('t', 'N/A')}")
    else:
        print("未獲取到數據，請檢查Token或網路")

    

      import websocket
import json
import threading
import time

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

class TSXMonitor:
    def on_message(self, ws, message):
        try:
            data = json.loads(message)
            if 'data' in data:
                payload = data['data']
                symbol = payload.get('s')
                typ = payload.get('type')
                if typ == 'quote':
                    print(f"[報價] {symbol}: 最新 {payload.get('ld')} | 量 {payload.get('v')}")
                elif typ == 'tick':
                    print(f"[成交] {symbol}: 價 {payload.get('ld')} | 量 {payload.get('v')}")
        except Exception as e:
            print(f"解析錯誤: {e}")

    def on_open(self, ws):
        print("WebSocket 已連接，正在訂閱 TSX 股票...")
        sub_msg = {
            "ac": "subscribe",
            "params": "RY$CA,ENB$CA,ABX$CA",
            "types": "quote,tick"
        }
        ws.send(json.dumps(sub_msg))

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

    def on_close(self, ws, code, msg):
        print("連接關閉")

    def start(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
        )

        def heartbeat():
            while True:
                time.sleep(30)
                if self.ws.sock and self.ws.sock.connected:
                    self.ws.send(json.dumps({"ac": "ping", "params": str(int(time.time()*1000))}))

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

if __name__ == '__main__':
    monitor = TSXMonitor()
    monitor.start()

    

      import requests
import pandas as pd
from datetime import datetime

def get_tsx_kline(symbol, ktype=8, limit=20):
    """
    獲取TSX股票歷史K線
    :param symbol: 證券代碼，如 RY
    :param ktype: K線類型 1=1分鐘, 2=5分鐘, 3=15分鐘, 4=30分鐘, 5=1小時, 6=2小時, 7=4小時, 8=日線, 9=週線, 10=月線
    :param limit: 返回K線根數
    """
    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:
        data = resp.json().get('data', [])
        
        # 轉換為DataFrame便於分析
        df = pd.DataFrame(data)
        if not df.empty:
            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']]
    else:
        print(f"K線獲取失敗: {resp.status_code}")
        return None

# 示例：獲取Enbridge（ENB）最近10個交易日數據
df = get_tsx_kline('ENB', ktype=8, limit=10)
if df is not None:
    print(df.to_string(index=False))