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()