API Integration for Trading Systems

Modern trading systems often need external data beyond broker feeds. This guide covers integrating APIs for market data, news, and other services into your trading systems.

Common API Types

REST APIs

• Request/response model
• Simple to implement
• Good for periodic data
• HTTP-based

WebSocket APIs

• Real-time streaming
• Persistent connection
• Low latency
• Good for live prices

FIX Protocol

• Industry standard
• Direct broker connection
• Institutional grade
• Complex implementation

Working with REST APIs

Basic Request Structure

import requests
import json

class APIClient:
    def __init__(self, base_url, api_key=None):
        self.base_url = base_url
        self.headers = {'Content-Type': 'application/json'}
        if api_key:
            self.headers['Authorization'] = f'Bearer {api_key}'

    def get(self, endpoint, params=None):
        """GET request"""
        url = f"{self.base_url}/{endpoint}"
        response = requests.get(url, headers=self.headers, params=params)
        return self._handle_response(response)

    def post(self, endpoint, data):
        """POST request"""
        url = f"{self.base_url}/{endpoint}"
        response = requests.post(url, headers=self.headers, json=data)
        return self._handle_response(response)

    def _handle_response(self, response):
        """Handle API response"""
        if response.status_code == 200:
            return response.json()
        elif response.status_code == 429:
            raise Exception("Rate limit exceeded")
        else:
            raise Exception(f"API error: {response.status_code} - {response.text}")

Example: Economic Calendar API

class EconomicCalendarAPI:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://api.tradingeconomics.com"

    def get_events(self, country=None, start_date=None, end_date=None):
        """Get economic events"""
        params = {'c': self.api_key}

        if country:
            params['country'] = country
        if start_date:
            params['d1'] = start_date
        if end_date:
            params['d2'] = end_date

        response = requests.get(
            f"{self.base_url}/calendar",
            params=params
        )
        return response.json()

    def get_high_impact_events(self, date=None):
        """Get only high-impact events"""
        events = self.get_events(start_date=date, end_date=date)
        return [e for e in events if e.get('importance') == 3]

# Usage
calendar = EconomicCalendarAPI('your_api_key')
events = calendar.get_high_impact_events('2024-01-15')

for event in events:
    print(f"{event['date']} - {event['country']}: {event['event']}")

Rate Limiting

import time
from functools import wraps

class RateLimiter:
    def __init__(self, calls_per_second=1):
        self.calls_per_second = calls_per_second
        self.last_call = 0

    def wait(self):
        """Wait if necessary to respect rate limit"""
        now = time.time()
        time_since_last = now - self.last_call
        min_interval = 1.0 / self.calls_per_second

        if time_since_last < min_interval:
            time.sleep(min_interval - time_since_last)

        self.last_call = time.time()

def rate_limited(calls_per_second):
    """Decorator for rate limiting"""
    limiter = RateLimiter(calls_per_second)

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            limiter.wait()
            return func(*args, **kwargs)
        return wrapper
    return decorator

# Usage
@rate_limited(calls_per_second=2)
def fetch_price(symbol):
    # API call here
    pass

WebSocket Integration

Real-Time Price Feed

import websocket
import json
import threading

class PriceFeed:
    def __init__(self, url, on_price_callback):
        self.url = url
        self.on_price = on_price_callback
        self.ws = None
        self.running = False

    def on_message(self, ws, message):
        """Handle incoming message"""
        data = json.loads(message)
        self.on_price(data)

    def on_error(self, ws, error):
        """Handle error"""
        print(f"WebSocket error: {error}")

    def on_close(self, ws, close_status_code, close_msg):
        """Handle connection close"""
        print("WebSocket closed")
        if self.running:
            print("Reconnecting...")
            time.sleep(5)
            self.connect()

    def on_open(self, ws):
        """Handle connection open"""
        print("WebSocket connected")
        # Subscribe to symbols
        subscribe_msg = {
            "action": "subscribe",
            "symbols": ["EURUSD", "GBPUSD", "USDJPY"]
        }
        ws.send(json.dumps(subscribe_msg))

    def connect(self):
        """Start WebSocket connection"""
        self.running = True
        self.ws = websocket.WebSocketApp(
            self.url,
            on_message=self.on_message,
            on_error=self.on_error,
            on_close=self.on_close,
            on_open=self.on_open
        )

        # Run in background thread
        thread = threading.Thread(target=self.ws.run_forever)
        thread.daemon = True
        thread.start()

    def disconnect(self):
        """Close WebSocket connection"""
        self.running = False
        if self.ws:
            self.ws.close()

# Usage
def handle_price(data):
    print(f"Price update: {data['symbol']} - Bid: {data['bid']}, Ask: {data['ask']}")

feed = PriceFeed("wss://stream.example.com/prices", handle_price)
feed.connect()

Integrating News APIs

News Feed for Trading

class NewsAPI:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://newsapi.org/v2"

    def get_forex_news(self, query="forex currency"):
        """Get forex-related news"""
        params = {
            'apiKey': self.api_key,
            'q': query,
            'language': 'en',
            'sortBy': 'publishedAt'
        }

        response = requests.get(f"{self.base_url}/everything", params=params)
        data = response.json()

        return data.get('articles', [])

    def analyze_sentiment(self, articles):
        """Simple sentiment analysis on headlines"""
        positive_words = ['surge', 'gain', 'rise', 'bull', 'up', 'high', 'strong']
        negative_words = ['fall', 'drop', 'crash', 'bear', 'down', 'low', 'weak']

        sentiment_score = 0

        for article in articles:
            headline = article['title'].lower()

            for word in positive_words:
                if word in headline:
                    sentiment_score += 1

            for word in negative_words:
                if word in headline:
                    sentiment_score -= 1

        return sentiment_score

# Usage
news = NewsAPI('your_api_key')
articles = news.get_forex_news('EUR USD currency')
sentiment = news.analyze_sentiment(articles)

print(f"Market sentiment score: {sentiment}")
if sentiment > 2:
    print("Bullish sentiment detected")
elif sentiment < -2:
    print("Bearish sentiment detected")

MQL5 WebRequest

Making HTTP Requests from EA

string SendAPIRequest(string url, string method, string body = "")
{
    char post[];
    char result[];
    string headers;
    string resultHeaders;

    if(body != "")
    {
        StringToCharArray(body, post);
        ArrayResize(post, ArraySize(post) - 1);
    }

    int timeout = 5000;

    int res = WebRequest(method, url, headers, timeout, post, result, resultHeaders);

    if(res == -1)
    {
        Print("WebRequest error: ", GetLastError());
        return "";
    }

    return CharArrayToString(result);
}

// Example: Get economic events
string GetEconomicEvents()
{
    string url = "https://api.example.com/calendar?date=today";
    string response = SendAPIRequest(url, "GET");

    return response;
}

Remember to add URL to allowed list in MT5: Tools > Options > Expert Advisors > Allow WebRequest for listed URLs

class TradingDataAggregator:
    def __init__(self):
        self.price_feed = None
        self.news_api = None
        self.calendar_api = None
        self.data = {
            'prices': {},
            'news_sentiment': 0,
            'economic_events': []
        }

    def setup(self, config):
        """Initialize all data sources"""
        self.price_feed = PriceFeed(
            config['price_feed_url'],
            self._on_price_update
        )
        self.news_api = NewsAPI(config['news_api_key'])
        self.calendar_api = EconomicCalendarAPI(config['calendar_api_key'])

    def _on_price_update(self, data):
        """Handle price updates"""
        self.data['prices'][data['symbol']] = {
            'bid': data['bid'],
            'ask': data['ask'],
            'time': data['time']
        }

    def update_news(self):
        """Update news sentiment"""
        articles = self.news_api.get_forex_news()
        self.data['news_sentiment'] = self.news_api.analyze_sentiment(articles)

    def update_calendar(self, date):
        """Update economic calendar"""
        self.data['economic_events'] = self.calendar_api.get_high_impact_events(date)

    def should_trade(self, symbol):
        """Check if trading conditions are met"""
        # Avoid trading during high-impact events
        if self._has_upcoming_event(symbol):
            return False

        # Check if we have recent price data
        if symbol not in self.data['prices']:
            return False

        price_age = time.time() - self.data['prices'][symbol]['time']
        if price_age > 60:  # Price older than 1 minute
            return False

        return True

    def _has_upcoming_event(self, symbol):
        """Check for upcoming high-impact events"""
        currencies = self._get_currencies(symbol)

        for event in self.data['economic_events']:
            if event['country'] in currencies:
                event_time = datetime.fromisoformat(event['date'])
                time_until = (event_time - datetime.now()).total_seconds()

                # Don't trade 30 min before/after event
                if -1800 < time_until < 1800:
                    return True

        return False

    def _get_currencies(self, symbol):
        """Extract currencies from symbol"""
        return [symbol[:3], symbol[3:]]

import time
from functools import wraps

def retry_on_failure(max_retries=3, delay=1, backoff=2):
    """Decorator for retrying failed API calls"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            retries = 0
            current_delay = delay

            while retries < max_retries:
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    retries += 1
                    if retries >= max_retries:
                        raise

                    print(f"Retry {retries}/{max_retries} after {current_delay}s: {e}")
                    time.sleep(current_delay)
                    current_delay *= backoff

        return wrapper
    return decorator

class ResilientAPIClient:
    def __init__(self, base_url):
        self.base_url = base_url
        self.session = requests.Session()

    @retry_on_failure(max_retries=3, delay=1)
    def get(self, endpoint, params=None):
        """GET with retry logic"""
        response = self.session.get(
            f"{self.base_url}/{endpoint}",
            params=params,
            timeout=10
        )
        response.raise_for_status()
        return response.json()