土耳其股票 API 技術接入指南：BIST 即時行情與歷史數據深度解析

      import requests

API_TOKEN = "your_token_here"  # 從iTick官網獲取
BASE_URL = "https://api.itick.org"

def get_headers():
    return {
        "accept": "application/json",
        "token": API_TOKEN
    }

    

      def get_turkey_quote(symbol):
    """獲取土耳其股票即時報價"""
    url = f"{BASE_URL}/stock/quote"
    params = {"region": "TR", "code": symbol}

    resp = requests.get(url, params=params, headers=get_headers(), timeout=5)
    data = resp.json()

    if data.get("code") == 0:
        quote = data["data"]
        print(f"📊 {quote.get('n')} ({quote.get('s')})")
        print(f"最新價: {quote.get('ld')} TRY")
        print(f"漲跌幅: {quote.get('chp')}%")
        print(f"成交量: {quote.get('v')}")
        return quote
    else:
        print(f"API錯誤: {data.get('msg')}")

# 測試
get_turkey_quote("THYAO")

    

      def get_turkey_kline(symbol, ktype=8, limit=100):
    """
    獲取土耳其股票歷史K線數據
    ktype參數說明：
        1: 1分鐘線
        2: 5分鐘線
        3: 15分鐘線
        4: 30分鐘線
        5: 60分鐘線
        8: 日線
        9: 週線
        10: 月線
    """
    url = f"{BASE_URL}/stock/kline"
    params = {
        "region": "TR",
        "code": symbol,
        "kType": ktype,
        "limit": str(limit)  # 注意：limit需要傳字串
    }

    resp = requests.get(url, params=params, headers=get_headers())
    data = resp.json()

    if data.get("code") == 0:
        klines = data.get("data", [])
        period_map = {1:"1分鐘",2:"5分鐘",3:"15分鐘",4:"30分鐘",5:"60分鐘",8:"日線",9:"週線",10:"月線"}
        print(f"✅ 獲取到 {len(klines)} 條{period_map.get(ktype, '')}數據")

        # 轉換為OHLCV格式
        for k in klines[-5:]:  # 顯示最近5條
            print(f"時間:{k['t']} 開:{k['o']} 高:{k['h']} 低:{k['l']} 收:{k['c']} 量:{k['v']}")
        return klines
    else:
        print(f"錯誤: {data.get('msg')}")

# 測試：獲取KCHOL日線數據
get_turkey_kline("KCHOL", ktype=8, limit=30)

# 測試：獲取EREGL週線數據
get_turkey_kline("EREGL", ktype=9, limit=20)

# 測試：獲取GARAN60分鐘線
get_turkey_kline("GARAN", ktype=5, limit=50)

    

      def get_turkey_depth(symbol):
    """獲取土耳其股票即時盤口數據"""
    url = f"{BASE_URL}/stock/depth"
    params = {"region": "TR", "code": symbol}

    resp = requests.get(url, params=params, headers=get_headers())
    data = resp.json()

    if data.get("code") == 0:
        depth = data.get("data", {})
        print(f"📊 {symbol} 盤口深度")
        print(f"--- 賣盤 (Ask) ---")
        for ask in depth.get('a', [])[:5]:  # 顯示前5檔
            print(f"檔位{ask.get('po')}: {ask.get('p')} TRY | 數量: {ask.get('v')}")
        print(f"--- 買盤 (Bid) ---")
        for bid in depth.get('b', [])[:5]:
            print(f"檔位{bid.get('po')}: {bid.get('p')} TRY | 數量: {bid.get('v')}")
        return depth
    else:
        print(f"錯誤: {data.get('msg')}")

# 測試
get_turkey_depth("KCHOL")

    

      def get_turkey_company_info(symbol):
    """獲取土耳其公司基本資訊"""
    url = f"{BASE_URL}/stock/info"
    params = {"region": "TR", "code": symbol}

    resp = requests.get(url, params=params, headers=get_headers())
    data = resp.json()

    if data.get("code") == 0:
        company = data.get("data", {})
        print(f"🏢 {company.get('n')} ({symbol})")
        print(f"行業: {company.get('i')}")
        print(f"板塊: {company.get('s')}")
        print(f"簡介: {company.get('bd')[:100]}...")
        print(f"總市值: {company.get('mcb')}")
        print(f"市盈率: {company.get('pet')}")
        return company
    else:
        print(f"錯誤: {data.get('msg')}")

get_turkey_company_info("KCHOL")

    

      import websocket
import json
import threading
import time
from typing import List, Callable

class TurkeyWebSocketClient:
    """土耳其股票WebSocket即時行情客戶端（支援自動重連）"""

    def __init__(self, token: str, symbols: List[str], callback: Callable, types: str = "quote"):
        self.token = token
        self.symbols = [f"{s}$TR" for s in symbols]  # 格式：代碼$TR
        self.callback = callback
        self.types = types
        self.ws = None
        self.running = False
        self.reconnect_interval = 5
        self.ws_url = "wss://api.itick.org/stock"

    def start(self):
        self.running = True
        self._connect()

    def _connect(self):
        headers = {"token": self.token}
        self.ws = websocket.WebSocketApp(
            self.ws_url,
            header=headers,
            on_open=self._on_open,
            on_message=self._on_message,
            on_error=self._on_error,
            on_close=self._on_close
        )

        # 在獨立執行緒運行
        wst = threading.Thread(target=self.ws.run_forever, daemon=True)
        wst.start()

    def _on_open(self, ws):
        print("✅ WebSocket連接成功，開始訂閱...")
        subscribe_msg = {
            "ac": "subscribe",
            "params": ",".join(self.symbols),
            "types": self.types  # quote/tick/depth
        }
        ws.send(json.dumps(subscribe_msg))
        print(f"📤 訂閱請求: {subscribe_msg['params']}")

        # 啟動心跳
        self._start_heartbeat()

    def _start_heartbeat(self):
        def heartbeat():
            while self.running:
                time.sleep(30)
                if self.ws and self.ws.sock and self.ws.sock.connected:
                    ping = {"ac": "ping", "params": str(int(time.time()*1000))}
                    self.ws.send(json.dumps(ping))
                    print("💓 心跳已發送")
        threading.Thread(target=heartbeat, daemon=True).start()

    def _on_message(self, ws, message):
        try:
            data = json.loads(message)
            if "data" in data:
                self.callback(data["data"])
        except Exception as e:
            print(f"訊息解析錯誤: {e}")

    def _on_error(self, ws, error):
        print(f"❌ WebSocket錯誤: {error}")

    def _on_close(self, ws, close_status, close_msg):
        print(f"🔌 連接關閉: {close_msg}")
        if self.running:
            print(f"⏱️ {self.reconnect_interval}秒後重連...")
            time.sleep(self.reconnect_interval)
            self._connect()

    def stop(self):
        self.running = False
        if self.ws:
            self.ws.close()

# 使用範例
def handle_market_data(data):
    """自訂數據處理函數"""
    data_type = data.get("type")
    symbol = data.get("s")

    if data_type == "quote":
        print(f"[{symbol}] 最新:{data.get('ld')} TRY | 漲跌:{data.get('chp')}% | 量:{data.get('v')}")
    elif data_type == "tick":
        print(f"[{symbol}] 成交:{data.get('ld')} TRY | 時間:{data.get('t')}")
    elif data_type == "depth":
        bids = data.get("b", [])[:3]
        asks = data.get("a", [])[:3]
        print(f"[{symbol}] 買盤:{bids} | 賣盤:{asks}")

# 訂閱土耳其主流股票
client = TurkeyWebSocketClient(
    token="your_token_here",
    symbols=["KCHOL", "EREGL", "THYAO", "GARAN"],
    types="quote,tick,depth",
    callback=handle_market_data
)
client.start()

# 保持主執行緒運行
try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    client.stop()

    

      import pandas as pd

def kline_to_dataframe(klines):
    """將K線數據轉換為Pandas DataFrame"""
    df = pd.DataFrame(klines)
    df['timestamp'] = pd.to_datetime(df['t'], unit='ms')
    df.set_index('timestamp', 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])

    return df

# 獲取數據並轉換為DataFrame
klines = get_turkey_kline("KCHOL", ktype=8, limit=100)
if klines:
    df = kline_to_dataframe(klines)
    print(df.head())

    

      def calculate_technical_indicators(df):
    """計算常用技術指標"""
    # 移動平均線
    df['MA20'] = df['close'].rolling(window=20).mean()
    df['MA50'] = df['close'].rolling(window=50).mean()

    # RSI
    delta = df['close'].diff()
    gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
    loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
    rs = gain / loss
    df['RSI'] = 100 - (100 / (1 + rs))

    # MACD
    exp12 = df['close'].ewm(span=12, adjust=False).mean()
    exp26 = df['close'].ewm(span=26, adjust=False).mean()
    df['MACD'] = exp12 - exp26
    df['Signal'] = df['MACD'].ewm(span=9, adjust=False).mean()

    return df

# 計算指標
df_with_indicators = calculate_technical_indicators(df)
print(df_with_indicators[['close', 'MA20', 'RSI', 'MACD']].tail())

    

      import time
from functools import wraps

def rate_limit(max_calls_per_second=2):
    """限流裝飾器"""
    def decorator(func):
        last_called = [0.0]

        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_called[0]
            left_to_wait = 1.0/max_calls_per_second - elapsed
            if left_to_wait > 0:
                time.sleep(left_to_wait)
            ret = func(*args, **kwargs)
            last_called[0] = time.time()
            return ret
        return wrapper
    return decorator

@rate_limit(max_calls_per_second=2)
def rate_limited_api_call(symbol):
    return get_turkey_quote(symbol)

    

      class TurkeyDataCache:
    """簡單的本地快取，減少API呼叫"""

    def __init__(self, ttl_seconds=60):
        self.cache = {}
        self.ttl = ttl_seconds

    def get(self, key):
        if key in self.cache:
            data, timestamp = self.cache[key]
            if time.time() - timestamp < self.ttl:
                return data
        return None

    def set(self, key, data):
        self.cache[key] = (data, time.time())

# 使用範例
cache = TurkeyDataCache(ttl_seconds=30)

def get_cached_quote(symbol):
    cached = cache.get(f"quote_{symbol}")
    if cached:
        print("使用快取數據")
        return cached

    print("呼叫API獲取")
    data = get_turkey_quote(symbol)
    if data:
        cache.set(f"quote_{symbol}", data)
    return data