Python for Quantitative Trading: How to Build a Real‑Time Trading Strategy with Market Data APIs

#api #python #tutorial #devops

Introduction

Real‑time market data forms the backbone of live algorithmic trading systems. For Python‑based quant projects, connecting reliable market data APIs allows developers to ingest live quotes, generate trading signals, and simulate strategy execution. Unlike centralized equity venues, forex and multi‑asset markets source quotes from distributed liquidity providers, which adds complexity to data ingestion. This article covers core concepts and walks through a practical working implementation using the AllTick API, focusing on live quote streaming, signal logic skeleton, and client‑side error handling.

Key Concepts

WebSocket Streaming: Unlike HTTP polling that repeatedly requests data, WebSocket maintains a persistent TCP connection. The server pushes new market updates as they occur, minimizing end‑to‑end latency for real‑time use‑cases. Polling introduces avoidable lag or excessive request volume, making it unsuitable for live strategy deployment.

Unified Market Schema: Normalize incoming payloads into consistent fields (symbol, timestamp, bid, ask, last). This decouples strategy business logic from vendor‑specific API response formats, simplifying future maintenance and asset expansion.

Signal Generation Skeleton: A real‑time strategy consumes streaming ticks, applies trading rules, and produces buy / sell / hold signals without blocking the data receiving thread.

Practical Implementation (Core Section)

Prerequisite: Install dependency pip install websocket-client. Obtain your API key from alltick.co. The code strictly follows AllTick public WebSocket API specification.

import json
import websocket
import time

# -------- Configuration --------
API_KEY = "YOUR_API_KEY_FROM_alltick.co"
WS_ENDPOINT = f"wss://quote.alltick.co/quote-b-ws-api?token={API_KEY}"
SUBSCRIBE_SYMBOLS = [{"code": "EURUSD"}, {"code": "USDJPY"}]

# Simple in‑memory state for strategy
strategy_state = {
    "latest_quotes": {},
    "last_signal": None
}

def generate_trading_signal(symbol: str, bid: float, ask: float):
    """Minimal placeholder signal logic: replace with your real strategy rules"""
    spread = ask - bid
    if spread > 0.0008:
        return "AVOID_TRADE"
    elif bid < 1.0800 and symbol == "EURUSD":
        return "SUGGEST_LONG"
    elif bid > 1.0950 and symbol == "EURUSD":
        return "SUGGEST_SHORT"
    return "HOLD"

def on_open(ws_app):
    """Triggered when WebSocket connection opens, send subscription request"""
    subscribe_payload = {
        "cmd_id": 22004,
        "seq_id": int(time.time()),
        "trace": "python-quant-strategy",
        "data": {
            "symbol_list": SUBSCRIBE_SYMBOLS
        }
    }
    ws_app.send(json.dumps(subscribe_payload))
    print("Subscription message sent, waiting for market data...")

def on_message(ws_app, raw_msg):
    """Process incoming market data and run strategy signal logic"""
    try:
        msg = json.loads(raw_msg)
        data_body = msg.get("data", {})
        symbol_code = data_body.get("symbol")
        bid_price = data_body.get("bid")
        ask_price = data_body.get("ask")

        if not all([symbol_code, bid_price, ask_price]):
            return

        # Cache latest normalized quote
        strategy_state["latest_quotes"][symbol_code] = {
            "bid": bid_price,
            "ask": ask_price,
            "timestamp": data_body.get("timestamp")
        }

        # Run strategy signal calculation
        current_signal = generate_trading_signal(symbol_code, bid_price, ask_price)
        strategy_state["last_signal"] = current_signal
        print(f"[{symbol_code}] bid:{bid_price} ask:{ask_price} | Signal: {current_signal}")

    except json.JSONDecodeError:
        return

def on_error(ws_app, error):
    print(f"WebSocket error occurred: {error}")

def on_close(ws_app, close_code, close_reason):
    print(f"Connection closed. Code:{close_code}, Reason:{close_reason}. Strategy will stop receiving ticks.")

if __name__ == "__main__":
    ws_client = websocket.WebSocketApp(
        WS_ENDPOINT,
        on_open=on_open,
        on_message=on_message,
        on_error=on_error,
        on_close=on_close
    )
    # run_forever blocks main thread; production code should add auto‑reconnection logic
    ws_client.run_forever()

Key Implementation Notes

Subscription Payload: cmd_id:22004 is the official AllTick command identifier for market data subscription.
Signal Logic: generate_trading_signal is only a demonstration template. Replace this function with your validated strategy rules. Do not use this placeholder for live capital deployment.
Production Improvement Points: The example omits auto‑reconnection, which you must implement for 24/7 operation. Add timestamp normalization to UTC, logging to file, and rate‑aware historical data fetching for backtesting validation.
Threading Warning: Avoid heavy synchronous computation in on_message, as it blocks incoming WebSocket message processing. Offload complex work to separate worker threads or queues.

Closing Remarks

This implementation demonstrates how to bootstrap a Python real‑time quant strategy with AllTick market data API. Real‑world production systems require further work: persistent storage, robust reconnection, risk control modules, and thorough backtesting against historical datasets. Always validate strategy performance before any live‑capital deployment.

Disclaimer: This article presents purely technical engineering examples. It is not investment advice. Algorithmic trading carries substantial financial risk.

"API Docs: https://alltick.co/apis/en
GitHub: https://github.com/alltick/alltick-realtime-forex-crypto-stock-tick-finance-websocket-api"

DEV Community