Turkey Stock API Integration Guide: BIST Real-Time Quotes & Historical Data

      import requests

API_TOKEN = "your_token_here"  # Obtain from iTick website
BASE_URL = "https://api.itick.org"

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

    

      def get_turkey_quote(symbol):
    """Retrieve real-time quote for Turkish stocks"""
    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"Last: {quote.get('ld')} TRY")
        print(f"Change: {quote.get('chp')}%")
        print(f"Volume: {quote.get('v')}")
        return quote
    else:
        print(f"API Error: {data.get('msg')}")

# Test
get_turkey_quote("THYAO")

    

      def get_turkey_kline(symbol, ktype=8, limit=100):
    """
    Retrieve historical K-line data for Turkish stocks
    ktype reference:
        1: 1-minute
        2: 5-minute
        3: 15-minute
        4: 30-minute
        5: 60-minute
        8: Daily
        9: Weekly
        10: Monthly
    """
    url = f"{BASE_URL}/stock/kline"
    params = {
        "region": "TR",
        "code": symbol,
        "kType": ktype,
        "limit": str(limit)  # Note: limit must be passed as string
    }

    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 min",2:"5 min",3:"15 min",4:"30 min",5:"60 min",8:"Daily",9:"Weekly",10:"Monthly"}
        print(f"✅ Retrieved {len(klines)} {period_map.get(ktype, '')} bars")

        # Display last 5 bars in OHLCV format
        for k in klines[-5:]:
            print(f"Time:{k['t']} O:{k['o']} H:{k['h']} L:{k['l']} C:{k['c']} Vol:{k['v']}")
        return klines
    else:
        print(f"Error: {data.get('msg')}")

# Examples
get_turkey_kline("KCHOL", ktype=8, limit=30)     # Koç Holding daily
get_turkey_kline("EREGL", ktype=9, limit=20)     # Ereğli weekly
get_turkey_kline("GARAN", ktype=5, limit=50)     # Garanti 60-min

    

      def get_turkey_depth(symbol):
    """Retrieve real-time order book depth for Turkish stocks"""
    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} Order Book Depth")
        print(f"--- Ask Side ---")
        for ask in depth.get('a', [])[:5]:
            print(f"Level {ask.get('po')}: {ask.get('p')} TRY | Size: {ask.get('v')}")
        print(f"--- Bid Side ---")
        for bid in depth.get('b', [])[:5]:
            print(f"Level {bid.get('po')}: {bid.get('p')} TRY | Size: {bid.get('v')}")
        return depth
    else:
        print(f"Error: {data.get('msg')}")

# Test
get_turkey_depth("KCHOL")

    

      def get_turkey_company_info(symbol):
    """Retrieve company fundamental information"""
    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"Industry: {company.get('i')}")
        print(f"Sector: {company.get('s')}")
        print(f"Description: {company.get('bd')[:100]}...")
        print(f"Market Cap: {company.get('mcb')}")
        print(f"P/E Ratio: {company.get('pet')}")
        return company
    else:
        print(f"Error: {data.get('msg')}")

get_turkey_company_info("KCHOL")

    

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

class TurkeyWebSocketClient:
    """Production-grade WebSocket client for Turkish equity real-time data (with auto-reconnect)"""

    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]  # Format: ticker$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 connected. Subscribing...")
        subscribe_msg = {
            "ac": "subscribe",
            "params": ",".join(self.symbols),
            "types": self.types  # quote,tick,depth
        }
        ws.send(json.dumps(subscribe_msg))
        print(f"📤 Subscription sent: {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("💓 Heartbeat sent")
        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"Message parse error: {e}")

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

    def _on_close(self, ws, close_status, close_msg):
        print(f"🔌 Connection closed: {close_msg}")
        if self.running:
            print(f"⏱️ Reconnecting in {self.reconnect_interval} seconds...")
            time.sleep(self.reconnect_interval)
            self._connect()

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

# Usage example
def handle_market_data(data):
    """Custom callback for market data"""
    data_type = data.get("type")
    symbol = data.get("s")

    if data_type == "quote":
        print(f"[{symbol}] Last: {data.get('ld')} TRY | Chg: {data.get('chp')}% | Vol: {data.get('v')}")
    elif data_type == "tick":
        print(f"[{symbol}] Trade: {data.get('ld')} TRY | Time: {data.get('t')}")
    elif data_type == "depth":
        bids = data.get("b", [])[:3]
        asks = data.get("a", [])[:3]
        print(f"[{symbol}] Bids: {bids} | Asks: {asks}")

# Subscribe to major Turkish stocks
client = TurkeyWebSocketClient(
    token="your_token_here",
    symbols=["KCHOL", "EREGL", "THYAO", "GARAN"],
    types="quote,tick,depth",
    callback=handle_market_data
)
client.start()

# Keep main thread alive
try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    client.stop()

    

      import pandas as pd

def kline_to_dataframe(klines):
    """Convert K-line list to 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

# Fetch and convert
klines = get_turkey_kline("KCHOL", ktype=8, limit=100)
if klines:
    df = kline_to_dataframe(klines)
    print(df.head())

    

      def calculate_technical_indicators(df):
    """Calculate common technical indicators"""
    # Moving Averages
    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

# Apply indicators
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):
    """Rate-limiting decorator"""
    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:
    """Simple in-memory cache to reduce API calls"""

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

# Usage
cache = TurkeyDataCache(ttl_seconds=30)

def get_cached_quote(symbol):
    cached = cache.get(f"quote_{symbol}")
    if cached:
        print("Using cached data")
        return cached

    print("Fetching from API")
    data = get_turkey_quote(symbol)
    if data:
        cache.set(f"quote_{symbol}", data)
    return data