Advanced Quantitative Trading: Integrating Third-Party Stock APIs and Automating Telegram Trade Alerts

Real-time reception of trading signals, order statuses, and account updates is crucial in quantitative trading. Telegram Bot is a low-cost, highly reliable solution that allows you to stay informed about trading activities anytime, anywhere. This article will guide you on integrating Telegram Bot into your quantitative trading system to send notifications for key events (e.g., executions, stop-losses, strategy signals).

1. Choosing a Data Source

iTick, a professional financial data provider in China, offers real-time and historical data for stocks, futures, options, and more. This article will detail how to integrate iTick's data API into your quantitative trading system.

1. Overview of iTick Data Source

Main Data Types

Data Type	Update Frequency	Use Case
Real-time tick	Millisecond-level	High-frequency trading
Minute/Daily K-line	1 minute/day	Trend-following strategy

Integration Methods

Integration Method	Latency	Use Case
HTTP REST API	Medium (300ms+)	Low-frequency data
WebSocket	Low (<100ms)	Real-time data feed

2. Preparation

Register an Account: Complete in 30 seconds, no credit card required, supports third-party login.
Obtain API Key: Generate instantly in the console.
Read Documentation: Includes rich code examples and tutorials.
Install Python Dependencies

      pip install requests websocket-client pandas hmac hashlib

3. HTTP REST API Integration Example

Fetching Real-Time Quotes

      import requests

API_KEY = "YOUR_API_KEY"

headers = {
    "accept": "application/json",
    "token": API_KEY
}

def get_realtime_quote(symbolInfo):
    url = "https://api.itick.org/stock/quote"
    params = {
        "region": symbolInfo.region,
        "code": symbolInfo.code
    }

    response = requests.get(url, headers=headers, params=params)
    if response.status_code == 200:
        return response.json()["data"]
    else:
        raise Exception(f"Request failed: {response.text}")

# Fetch real-time quote for Kweichow Moutai
params = {
    "region": "SH",
    "code": "600519"
}
quote = get_realtime_quote(params)
print(f"Latest Price: {quote['ld']}, Volume: {quote['v']}")

    

Fetching Historical Data

      API_KEY = "YOUR_API_KEY"

headers = {
    "accept": "application/json",
    "token": API_KEY
}

def get_historical_kline(symbol, region, period="8", limit=100):
    url = "https://api.itick.org/stock/kline"
    params = {
        "region": region,
        "symbol": symbol,
        "kType": period,  # Supported intervals: 1 minute, 5 minutes, 10 minutes, 30 minutes, 1 hour, 2 hours, 4 hours, 1 day, 1 week, 1 month
        "et": int(time.time() * 1000),  # Query end time
        "limit": limit,
    }

    response = requests.get(url, headers=headers, params=params)
    return pd.DataFrame(response.json()["data"])

# Fetch daily K-line for Kweichow Moutai
df = get_historical_kline("600519", "SH", "8", 100)
df.to_csv("600519_daily.csv", index=False)

    

4. WebSocket Integration Example

      import websocket
import json
import threading

# API KEY
API_KEY = "YOUR_API_KEY"

# WebSocket server address
ws_url = "wss://api.itick.org/sws"

# Authentication message
auth_message = {
  "ac": "auth",
  "params": API_KEY
}

# Subscribe to Kweichow Moutai and CATL
subscribe_message = {
  "ac": "subscribe",
  "params": "600519$SH,300750$SZ",
  "types": "depth,quote"
}

def on_message(ws, message):
    data = json.loads(message)
    if "data" in data:
        print(f"Data: {data['data']}")

def on_error(ws, error):
    print("WebSocket Error:", error)

def on_close(ws):
    print("WebSocket Connection Closed")

def on_open(ws):
    # Send authentication message
    ws.send(json.dumps(auth_message))

    # Send subscription message
    ws.send(json.dumps(subscribe_message))

# Establish WebSocket connection
ws = websocket.WebSocketApp(
    ws_url,
    on_open=on_open,
    on_message=on_message,
    on_error=on_error,
    on_close=on_close
)

# Run WebSocket in the background
threading.Thread(target=ws.run_forever).start()

    

5. Advanced Application: Triggering Telegram Alerts for Trading Signals

Bollinger Bands Strategy Example

      import numpy as np

def bollinger_strategy(symbol, region):
    # Fetch historical data
    df = get_historical_kline(symbol, region, "3", 100)
    closes = df["c"].astype(float)

    # Calculate Bollinger Bands
    mean = closes.rolling(20).mean()
    std = closes.rolling(20).std()
    upper = mean + 2 * std
    lower = mean - 2 * std

    # Determine signals
    last_close = closes.iloc[-1]
    if last_close > upper.iloc[-1]:
        send_telegram_alert(f"🔔 {symbol} broke above the upper Bollinger Band! Consider selling.")
    elif last_close < lower.iloc[-1]:
        send_telegram_alert(f"🔔 {symbol} fell below the lower Bollinger Band! Consider buying.")

def send_telegram_alert(message):
    requests.post(
        f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage",
        json={"chat_id": CHAT_ID, "text": message}
    )

# Run every 10 minutes
schedule.every(10).minutes.do(bollinger_strategy, "600519", "SH")

    

6. Performance Optimization Tips

Data Caching: Cache fetched market data locally to avoid redundant requests.
Multithreading Optimization: Use multithreading or process pools to improve program performance.

7. Frequently Asked Questions

Q1: How to get free test data?

A1: iTick offers a free 30-day trial. You can access data in the test environment (use cautiously in the production environment).

Q2: How to handle WebSocket disconnections?

A2: Implement a reconnection mechanism to handle WebSocket disconnections.

      def run_ws():
    while True:
        try:
            ws.run_forever()
        except Exception as e:
            print(f"Connection lost, retrying in 5 seconds... Error: {e}")
            time.sleep(5)

threading.Thread(target=run_ws).start()

    