Trading API Base blueprint

[lukeTradingLearner]

Hi im new to ig and automation . im interested in the trading API and making automated scripts. I am interested in making sculping scripts. I have a few questions that i cant find.

• What's the api Request limit. there is usually a cap per month or something?
• How fast does the api place orders
• When I made my fist complete script where is the best place to host a server that uses the script to reduce latency(what area are the ig servers located I want to host serve close to them)
• How do i find out what times the market is open. im using EUR/USD in this script.

I am working on the foundation to all my trading strategies. i want to download historical data and store it to a csv file. every 5 min it will add new data to the end of the file. also it checks if market is open so does not waste api requests. Im very new to coding and automation and there hardly anything online. if anyone has suggestions/tips to deploy trading strategies would be very helpful. 

Here is my foundation of my script that i made so far so you can see my approach of using the api:

import requests
import pandas as pd
import json
from datetime import datetime
import time
from datetime import time as dt_time
api_key = 'api_key'
ig_username = 'usename'
ig_password = 'password'
base_url = 'https://demo-api.ig.com/gateway/deal' #change for live

headers = {
    'Content-Type': 'application/json; charset=UTF-8',
    'Accept': 'application/json; charset=UTF-8',
    'X-IG-API-KEY': api_key,
    'Version': '2'
}
def authenticate():
    data = {
        "identifier": ig_username,
        "password": ig_password
    }
    response = requests.post(f"{base_url}/session", headers=headers, json=data)
    if response.status_code == 200:
        if 'CST' in response.headers and 'X-SECURITY-TOKEN' in response.headers:
            return response.headers['CST'], response.headers['X-SECURITY-TOKEN']
        else:
            raise ValueError(f"Error: 'CST' or 'X-SECURITY-TOKEN' not found in response headers. Response: {response.text}")
    else:
        raise ValueError(f"Error {response.status_code}: {response.text}")

def get_eur_usd_data(cst, x_security_token, start_date, end_date):
    headers_with_token = headers.copy()
    headers_with_token['CST'] = cst
    headers_with_token['X-SECURITY-TOKEN'] = x_security_token
    resolution = 'MINUTE_5'
    params = {
        'resolution': resolution,
        'from': start_date,
        'to': end_date
    }
    request_url = f"{base_url}/prices/CS.D.EURUSD.MINI.IP"
    response = requests.get(f"{base_url}/prices/CS.D.EURUSD.MINI.IP/"+resolution+"/"+start_date+"/"+end_date,headers=headers_with_token)
    if response.status_code == 200:
        data = json.loads(response.text)
        return data['prices']
    else:
        raise ValueError(f"Error {response.status_code}: {response.text}")

def is_market_open_manual():
    # Define market hours (e.g., 5 PM Sunday to 5 PM Friday EST)
    open_time = dt_time(17, 0, 0)
    close_time = dt_time(17, 0, 0)
    start_of_week = 6  # Sunday
    end_of_week = 4  # Friday

    now = datetime.utcnow()
    now_time = now.time()
    now_weekday = now.weekday()

    if start_of_week <= now_weekday <= end_of_week:
        if start_of_week == now_weekday:
            return now_time >= open_time
        elif end_of_week == now_weekday:
            return now_time < close_time
        else:
            return True
    else:
        return False
def is_market_open(cst, x_security_token): #Note uses api request have not uses yet
    headers_with_token = headers.copy()
    headers_with_token['CST'] = cst
    headers_with_token['X-SECURITY-TOKEN'] = x_security_token
    epic = 'CS.D.EURUSD.MINI.IP'

    request_url = f"{base_url}/markets/{epic}"
    response = requests.get(request_url, headers=headers_with_token)
    if response.status_code == 200:
        data = json.loads(response.text)
        return data['snapshot']['marketStatus'] == 'TRADEABLE'
    else:
        raise ValueError(f"Error {response.status_code}: {response.text}")

def update_csv(filename, data):
    df = pd.DataFrame(data)
    df['datetime'] = pd.to_datetime(df['snapshotTime'], unit='ms')
    df.set_index('datetime', inplace=True)

    if not df.empty:
        try:
            existing_df = pd.read_csv(filename, index_col='datetime', parse_dates=True)
            df = existing_df.append(df)
            df.to_csv(filename)
        except FileNotFoundError:
            df.to_csv(filename)
def main():
    filename = 'eur_usd_5min_data.csv'
    retry_limit = 3
    retry_delay = 60  # Time in seconds between retries

    while True:
        try:
            cst, x_security_token = authenticate()

            if is_market_open_manual():
                try:
                    existing_df = pd.read_csv(filename, index_col='datetime', parse_dates=True)
                    start_date = existing_df.index[-1].strftime('%Y-%m-%d %H:%M:%S')
                except FileNotFoundError:
                    start_date = '2023-03-01 00:00:00'

                end_date = datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')

                data = get_eur_usd_data(cst, x_security_token, start_date, end_date)
                update_csv(filename, data)
                print("Data updated.")
            else:
                print("Market is closed, no data is retrieved.")

            # Sleep for 5 minutes (300 seconds)
            time.sleep(300)

        except (ValueError, requests.exceptions.RequestException) as e:
            print(f"An error occurred: {e}")
            retry_limit -= 1
            if retry_limit > 0:
                print(f"Retrying... {retry_limit} attempts remaining.")
                time.sleep(retry_delay)
            else:
                print("Retry limit reached. Exiting.")
                break

if __name__ == '__main__':
    main()
 

[OfentseIG]

1 hour ago, lukeTradingLearner said:

Hi im new to ig and automation . im interested in the trading API and making automated scripts. I am interested in making sculping scripts. I have a few questions that i cant find.

• What's the api Request limit. there is usually a cap per month or something?
• How fast does the api place orders
• When I made my fist complete script where is the best place to host a server that uses the script to reduce latency(what area are the ig servers located I want to host serve close to them)
• How do i find out what times the market is open. im using EUR/USD in this script.

I am working on the foundation to all my trading strategies. i want to download historical data and store it to a csv file. every 5 min it will add new data to the end of the file. also it checks if market is open so does not waste api requests. Im very new to coding and automation and there hardly anything online. if anyone has suggestions/tips to deploy trading strategies would be very helpful. 

Here is my foundation of my script that i made so far so you can see my approach of using the api:

import requests
import pandas as pd
import json
from datetime import datetime
import time
from datetime import time as dt_time
api_key = 'api_key'
ig_username = 'usename'
ig_password = 'password'
base_url = 'https://demo-api.ig.com/gateway/deal' #change for live

headers = {
    'Content-Type': 'application/json; charset=UTF-8',
    'Accept': 'application/json; charset=UTF-8',
    'X-IG-API-KEY': api_key,
    'Version': '2'
}
def authenticate():
    data = {
        "identifier": ig_username,
        "password": ig_password
    }
    response = requests.post(f"{base_url}/session", headers=headers, json=data)
    if response.status_code == 200:
        if 'CST' in response.headers and 'X-SECURITY-TOKEN' in response.headers:
            return response.headers['CST'], response.headers['X-SECURITY-TOKEN']
        else:
            raise ValueError(f"Error: 'CST' or 'X-SECURITY-TOKEN' not found in response headers. Response: {response.text}")
    else:
        raise ValueError(f"Error {response.status_code}: {response.text}")

def get_eur_usd_data(cst, x_security_token, start_date, end_date):
    headers_with_token = headers.copy()
    headers_with_token['CST'] = cst
    headers_with_token['X-SECURITY-TOKEN'] = x_security_token
    resolution = 'MINUTE_5'
    params = {
        'resolution': resolution,
        'from': start_date,
        'to': end_date
    }
    request_url = f"{base_url}/prices/CS.D.EURUSD.MINI.IP"
    response = requests.get(f"{base_url}/prices/CS.D.EURUSD.MINI.IP/"+resolution+"/"+start_date+"/"+end_date,headers=headers_with_token)
    if response.status_code == 200:
        data = json.loads(response.text)
        return data['prices']
    else:
        raise ValueError(f"Error {response.status_code}: {response.text}")

def is_market_open_manual():
    # Define market hours (e.g., 5 PM Sunday to 5 PM Friday EST)
    open_time = dt_time(17, 0, 0)
    close_time = dt_time(17, 0, 0)
    start_of_week = 6  # Sunday
    end_of_week = 4  # Friday

    now = datetime.utcnow()
    now_time = now.time()
    now_weekday = now.weekday()

    if start_of_week <= now_weekday <= end_of_week:
        if start_of_week == now_weekday:
            return now_time >= open_time
        elif end_of_week == now_weekday:
            return now_time < close_time
        else:
            return True
    else:
        return False
def is_market_open(cst, x_security_token): #Note uses api request have not uses yet
    headers_with_token = headers.copy()
    headers_with_token['CST'] = cst
    headers_with_token['X-SECURITY-TOKEN'] = x_security_token
    epic = 'CS.D.EURUSD.MINI.IP'

    request_url = f"{base_url}/markets/{epic}"
    response = requests.get(request_url, headers=headers_with_token)
    if response.status_code == 200:
        data = json.loads(response.text)
        return data['snapshot']['marketStatus'] == 'TRADEABLE'
    else:
        raise ValueError(f"Error {response.status_code}: {response.text}")

def update_csv(filename, data):
    df = pd.DataFrame(data)
    df['datetime'] = pd.to_datetime(df['snapshotTime'], unit='ms')
    df.set_index('datetime', inplace=True)

    if not df.empty:
        try:
            existing_df = pd.read_csv(filename, index_col='datetime', parse_dates=True)
            df = existing_df.append(df)
            df.to_csv(filename)
        except FileNotFoundError:
            df.to_csv(filename)
def main():
    filename = 'eur_usd_5min_data.csv'
    retry_limit = 3
    retry_delay = 60  # Time in seconds between retries

    while True:
        try:
            cst, x_security_token = authenticate()

            if is_market_open_manual():
                try:
                    existing_df = pd.read_csv(filename, index_col='datetime', parse_dates=True)
                    start_date = existing_df.index[-1].strftime('%Y-%m-%d %H:%M:%S')
                except FileNotFoundError:
                    start_date = '2023-03-01 00:00:00'

                end_date = datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')

                data = get_eur_usd_data(cst, x_security_token, start_date, end_date)
                update_csv(filename, data)
                print("Data updated.")
            else:
                print("Market is closed, no data is retrieved.")

            # Sleep for 5 minutes (300 seconds)
            time.sleep(300)

        except (ValueError, requests.exceptions.RequestException) as e:
            print(f"An error occurred: {e}")
            retry_limit -= 1
            if retry_limit > 0:
                print(f"Retrying... {retry_limit} attempts remaining.")
                time.sleep(retry_delay)
            else:
                print("Retry limit reached. Exiting.")
                break

if __name__ == '__main__':
    main()
 

Hi @lukeTradingLearner

Thanks for reaching out. Have a look at this link: https://labs.ig.com/faq

It answers most of your questions if not all. Please let me know if you still need assistance. We are more than happy to help. 

Thanks,

Ofentse

 

[KJM34]

It looks like you are just trying to make a price action / time series type strategy, it could be easier to use ProReal time by enabling in settings, the advantage is you can backtest and visualise the strategy and also run on the ProOrder servers. I use pro real time to test that sort of thing and then jump over to a platform called quantconnect to test a wider range of products and data sets with other brokers, the advantage there is massive data sets (fundamental etc) and the option to use either C# or Python. the downside is no IG markets data or API, I am interested in making an API for IG on quantconnect some time in the future but I have no experience building API's, Quantconnect does offer $5000USD to anyone who produces a new broker API and IG markets has been a popular request.

Edited April 7 by KJM34

[KJM34]

Using, the Oanda brokerage template that already exists in quantconnects Lean engine repository on GitHub and the IG markets API in Labs it wouldn't be to hard to knock something up as they both use the same type of API as far as I can tell. that's about as far as I delved into that idea though. it would still require access to the historical datasets which IG markets probably already has on request by an api.

Edited April 7 by KJM34

[lukeTradingLearner]

Thank you for suggestions literally started making automated strategy's 2 weeks ago. learned a lot by the time I made this post. that script was not a strategy yet it was just the foundation of just getting historic data and checking if markets are open. we limited to 10000 pulls of historic data per week from the ig api so was just optimizing my script to not exceed that limit. im implementing a technical analysis strategy with good risk management on to that data i pulled. but i am starting to get the hang of it and making good progress and my first script will be operational very soon. just finished with back testing then will test it on demo ig account. ig trading api is really good in my opinion and the documentation is good. plus the commission is just the spread. so for me ig is perfect for my approach to trading.