土耳其股票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