iTick Real-Time Stock Market API and Futu NiuNiu Trading Interface for Personal Quantitative Trading — High-Frequency Stock API - iTick

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}&region={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}&region={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