
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()
Q3: How to optimize high data latency?
A3: Upgrade to iTick's paid plans for reduced latency.