加拿大股票API对接：多伦多证券交易所（TSX/TSXV）实时行情与量化实践

      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))