In today's digital financial era, personal investors can achieve quantitative trading through API interfaces. This article provides a detailed guide on how to use the iTick real-time stock market API for Hong Kong and U.S. stocks, combined with the Futu NiuNiu (Futu OpenAPI) trading interface, to build a complete personal quantitative trading system. Starting from system architecture design, we will delve into specific implementations and provide complete Python code and backtesting strategies.
I. System Architecture Design
The system adopts a modular design with loosely coupled components, making it easy to extend and maintain. Below is the complete system architecture:
II. Technology Selection and Preparation
1. Market Data Source - iTick API
iTick provides stable and reliable real-time stock market data for Hong Kong and U.S. stocks, with the following features:
- Low Latency: Millisecond-level market data push
- Comprehensive Coverage: Global markets including Hong Kong, U.S., and A-shares (via Hong Kong Stock Connect)
- Rich Data Types: Tick data, minute-level data, daily data, etc.
- High Reliability: 99.9% availability guarantee
2. Trading Interface - Futu NiuNiu Futu OpenAPI
Futu NiuNiu's open platform provides a comprehensive trading API:
- Supports trading for Hong Kong, U.S., and A-shares (via Hong Kong Stock Connect)
- Offers a simulated trading environment
- Comprehensive documentation and community support
- Relatively high trade execution speed
3. Development Environment Preparation
# Required Python Libraries
import requests # For API calls
import pandas as pd # Data processing
import numpy as np # Numerical computations
from futu import * # Futu OpenAPI
import matplotlib.pyplot as plt # Visualization
import sqlite3 # Local data storage
import time # Time control
import threading # Multithreading
III. Core System Modules Implementation
1. Market Data Fetching Module
class ITickDataFetcher:
def __init__(self):
self.headers = {
"accept": "application/json",
"token": ITICK_API_KEY
}
def get_realtime_data(self, symbol, region):
"""Fetch real-time market data"""
url = f"https://api.itick.org/stock/quote?symbol={symbol}®ion={region}"
try:
response = requests.get(url, headers=self.headers)
data = response.json()
return self._parse_data(data)
except Exception as e:
print(f"Failed to fetch real-time data: {str(e)}")
return None
def get_historical_data(self, symbol, region, freq='1'):
"""Fetch historical data"""
url = f"https://api.itick.org/stock/kline?symbol={symbol}®ion={region}&kType={freq}"
try:
response = requests.get(url, headers=self.headers)
data = response.json()
df = self._parse_historical_data(data)
except Exception as e:
print(f"Failed to fetch historical data for {symbol}: {str(e)}")
return None
def _parse_data(self, raw_data):
"""Parse real-time data"""
# Implement data parsing logic
pass
def _parse_historical_data(self, raw_data):
"""Parse historical data"""
# Implement historical data parsing logic
pass
2. Data Preprocessing Module
class DataProcessor:
def __init__(self):
self.conn = sqlite3.connect('quant_trading.db')
def clean_data(self, df):
"""Clean data"""
# Handle missing values
df = df.dropna()
# Remove outliers
df = df[(df['v'] > 0) & (df['c'] > 0)]
return df
def calculate_technical_indicators(self, df):
"""Calculate technical indicators"""
# Moving Averages
df['ma5'] = df['c'].rolling(5).mean()
df['ma20'] = df['c'].rolling(20).mean()
# RSI
delta = df['c'].diff()
gain = (delta.where(delta > 0, 0)).rolling(14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(14).mean()
rs = gain / loss
df['rsi'] = 100 - (100 / (1 + rs))
# MACD
exp12 = df['c'].ewm(span=12, adjust=False).mean()
exp26 = df['c'].ewm(span=26, adjust=False).mean()
df['macd'] = exp12 - exp26
df['signal'] = df['macd'].ewm(span=9, adjust=False).mean()
return df
def save_to_db(self, df, table_name):
"""Save data to database"""
df.to_sql(table_name, self.conn, if_exists='append', index=False)
def load_from_db(self, table_name):
"""Load data from database"""
return pd.read_sql(f"SELECT * FROM {table_name}", self.conn)
3. Strategy Engine
class DualMovingAverageStrategy:
def __init__(self, fast_window=5, slow_window=20):
self.fast_window = fast_window
self.slow_window = slow_window
self.position = 0 # 0: No position, 1: Long, -1: Short
self.signals = []
def generate_signals(self, df):
"""Generate trading signals"""
signals = pd.DataFrame(index=df.index)
signals['price'] = df['c']
signals['fast_ma'] = df['c'].rolling(self.fast_window).mean()
signals['slow_ma'] = df['c'].rolling(self.slow_window).mean()
# Generate trading signals
signals['signal'] = 0
signals['signal'][self.fast_window:] = np.where(
signals['fast_ma'][self.fast_window:] > signals['slow_ma'][self.fast_window:], 1, 0)
# Calculate actual buy/sell signals
signals['positions'] = signals['signal'].diff()
return signals
def on_tick(self, tick_data):
"""Handle real-time market data"""
# Implement real-time market data handling logic
pass
4. Trade Execution Module
class FutuTrader:
def __init__(self, host='127.0.0.1', port=11111):
self.quote_ctx = OpenQuoteContext(host=host, port=port)
self.trade_ctx = OpenTradeContext(host=host, port=port)
def get_account_info(self):
"""Fetch account information"""
ret, data = self.trade_ctx.accinfo_query()
if ret == RET_OK:
return data
else:
print(f"Failed to fetch account info: {data}")
return None
def place_order(self, symbol, price, quantity, trd_side, order_type=OrderType.NORMAL):
"""Place an order"""
ret, data = self.trade_ctx.place_order(
price=price,
qty=quantity,
code=symbol,
trd_side=trd_side,
order_type=order_type,
trd_env=TrdEnv.SIMULATE # Simulated trading, change to TrdEnv.REAL for live trading
)
if ret == RET_OK:
print(f"Order placed successfully: {data}")
return data
else:
print(f"Failed to place order: {data}")
return None
def close_all_positions(self):
"""Close all positions"""
ret, data = self.trade_ctx.position_list_query()
if ret == RET_OK:
for _, row in data.iterrows():
if row['qty'] > 0:
self.place_order(
row['code'],
row['cost_price'],
row['qty'],
TrdSide.SELL
)
else:
print(f"Failed to fetch positions: {data}")
def subscribe(self, symbols, subtype_list=[SubType.QUOTE]):
"""Subscribe to market data"""
ret, data = self.quote_ctx.subscribe(symbols, subtype_list)
if ret != RET_OK:
print(f"Failed to subscribe to market data: {data}")
def unsubscribe(self, symbols, subtype_list=[SubType.QUOTE]):
"""Unsubscribe from market data"""
ret, data = self.quote_ctx.unsubscribe(symbols, subtype_list)
if ret != RET_OK:
print(f"Failed to unsubscribe from market data: {data}")
def __del__(self):
self.quote_ctx.close()
self.trade_ctx.close()
IV. Backtesting System Implementation
class BacktestEngine:
def __init__(self, initial_capital=100000):
self.initial_capital = initial_capital
self.results = None
def run_backtest(self, strategy, data):
"""Run backtest"""
signals = strategy.generate_signals(data)
# Initialize portfolio
portfolio = pd.DataFrame(index=signals.index)
portfolio['signal'] = signals['positions']
portfolio['price'] = signals['price']
portfolio['cash'] = self.initial_capital
portfolio['shares'] = 0
portfolio['total'] = self.initial_capital
# Execute trades
for i in range(1, len(portfolio)):
# Buy signal
if portfolio['signal'][i] == 1:
shares_to_buy = portfolio['cash'][i-1] // portfolio['price'][i]
portfolio['shares'][i] = portfolio['shares'][i-1] + shares_to_buy
portfolio['cash'][i] = portfolio['cash'][i-1] - shares_to_buy * portfolio['price'][i]
# Sell signal
elif portfolio['signal'][i] == -1:
portfolio['cash'][i] = portfolio['cash'][i-1] + portfolio['shares'][i-1] * portfolio['price'][i]
portfolio['shares'][i] = 0
# No signal
else:
portfolio['shares'][i] = portfolio['shares'][i-1]
portfolio['cash'][i] = portfolio['cash'][i-1]
# Update total assets
portfolio['total'][i] = portfolio['cash'][i] + portfolio['shares'][i] * portfolio['price'][i]
self.results = portfolio
return portfolio
def analyze_results(self):
"""Analyze backtest results"""
if self.results is None:
print("Please run the backtest first")
return None
returns = self.results['total'].pct_change()
cumulative_returns = (returns + 1).cumprod()
# Calculate annualized return
total_days = len(self.results)
annualized_return = (cumulative_returns.iloc[-1] ** (252/total_days)) - 1
# Calculate maximum drawdown
max_drawdown = (self.results['total'].cummax() - self.results['total']).max()
# Calculate Sharpe ratio
sharpe_ratio = returns.mean() / returns.std() * np.sqrt(252)
# Visualization
plt.figure(figsize=(12, 6))
plt.plot(cumulative_returns)
plt.title("Cumulative Returns")
plt.xlabel("Date")
plt.ylabel("Returns")
plt.grid(True)
plt.show()
return {
'annualized_return': annualized_return,
'max_drawdown': max_drawdown,
'sharpe_ratio': sharpe_ratio,
'final_value': self.results['total'].iloc[-1]
}
V. Conclusion
This article provides a detailed guide on how to build a personal quantitative trading system using the iTick real-time stock market API and the Futu NiuNiu trading interface. The system includes the following core components:
- Market Data Fetching Module: Fetch real-time and historical data from the iTick API
- Data Preprocessing Module: Clean data and calculate technical indicators
- Strategy Engine: Implement a dual moving average trading strategy
- Trade Execution Module: Execute trades via the Futu OpenAPI
- Backtesting System: Evaluate the historical performance of strategies
