API IG en Python (trading_ig) : récupérer les données de marché en continu

[Deep]

Bonjour à vous qui prenez le temps de me lire!

je développe depuis un certain temps une learning machine neuronale que j'alimente avec les données de Yahoo finance en 1 minute...j'ai évidemment un temps de retard....j'ai un compte pro ig et je cherche à comprendre comment obtenir ces même données en flux continue sur US500 et Dax, une ame charitable pourrait elle m'aiguiller sur un script python me permettant d'obtenir ces données et sur quelle profondeur? si vous avez des historiques en 1 minute je suis également preneur. Pour les plus investi, je suis en mesure de partager mes avancées si vous m'expliquer comment passer des ordres via l'api IG.
Merci de votre écoute et à plus dans le bus !!!

[Deep]

voici ou j'en suis mais je bloque....merci pour votre aide:

Code : #

# -*- coding: utf-8 -*-
import logging
from trading_ig import IGService, IGStreamService
from lightstreamer.client import Subscription, LightstreamerClient

# Configuration du logger
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def on_item_update(item_update):
    """ Gere les mises a jour de l'abonnement. """
    logger.info(f"{item_update['name']} - {item_update['values']}")

def main():
    # Paramètres de l'API IG
    username = "......"
    password = "..........."
    api_key = ".............."
    acc_type = "LIVE"
    acc_number = "......"

    # Configuration du service IG
    ig_service = IGService(username, password, api_key, acc_type, acc_number=acc_number)
    ig_stream_service = IGStreamService(ig_service)
    ig_stream_service.create_session()

    # Récupération de l'URL Lightstreamer
    ls_endpoint = ig_service.read_session()["lightstreamerEndpoint"]
    ls_url = ls_endpoint.replace("https://", "wss://") + "/lightstreamer"
    logger.info(f"URL Lightstreamer: {ls_url}")

    # Abonnement aux données du marché pour un instrument spécifique
    epic = "IX.D.SPTRD.IFE.IP"
    subscription = Subscription(
        mode="MERGE",
        items=[f"MARKET:{epic}"],
        fields=["BID", "OFFER", "HIGH", "LOW", "MID_OPEN", "CHANGE", "CHANGE_PCT", "UPDATE_TIME", "MARKET_STATE"]
    )
    subscription.addListener(on_item_update)

    # Connexion au serveur Lightstreamer
    ls_client = LightstreamerClient(ls_url, adapter_set="")
    ls_client.connect()
    ls_client.subscribe(subscription)

    # Attendez ici pour les mises à jour
    input("Appuyez sur Entree pour arreter...\n")

    # Se déconnecter
    ls_client.unsubscribe(subscription)
    ls_client.disconnect()
    ig_stream_service.logout()

if __name__ == "__main__":
    main()

[cafeiine2023]

Oh. Je serais très intéressé de récupérer des cours via API, historiques. Mais j'ai butté très vite sur la complexité et le manque de sources/exemples.

[cafeiine2023]

Le peu que j'ai réussi à faire avec les api ig est ici
https://github.com/cafeiine/trading

Dit moi si à un moment donné tu arrives à récupérer des cours.

[Deep]

Ceci devrait répondre à tes besoins:

Code : #

# -*- coding: utf-8 -*-
import logging
import pandas as (mot censuré merci de rester poli)
from trading_ig import IGService

# Configuration du logger
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def main():
    # Paramètres de l'API IG
    username = "XXXX"
    password = "XXXXX"
    api_key = "XXXXXXXXXXXXX"
    acc_type = "LIVE"
    acc_number = "le numero de ton compte"

    # Configuration du service IG
    ig_service = IGService(username, password, api_key, acc_type, acc_number=acc_number)
    ig_service.create_session()

    # EPIC pour lequel obtenir les données historiques
    epic = "IX.D.SPTRD.IFE.IP"

    # Dates de récupération des données historiques
    start_date_str = "2023-12-01 09:00:00"
    end_date_str = "2023-12-01 09:10:00"

    try:
        # Récupérer les données historiques
        historical_data = ig_service.fetch_historical_prices_by_epic_and_date_range(
            epic, '5Min', start_date_str, end_date_str
        )
        
        if historical_data and 'prices' in historical_data:
            prices_df = (mot censuré merci de rester poli).DataFrame(historical_data['prices'])
            logger.info("Donnees historiques recuperees.")

            # Création du nom de fichier CSV
            filename = f"{epic}_{start_date_str.replace(':', '').replace('-', '').replace(' ', '_')}_{end_date_str.replace(':', '').replace('-', '').replace(' ', '_')}.csv"

            # Enregistrer les donnees dans un fichier CSV
            prices_df.to_csv(filename, index=False)
            logger.info(f"Les donnees historiques ont ete sauvegardees dans '{filename}'")
        else:
            logger.warning("Aucune donnee historique trouvee pour la periode specifiee.")

    except Exception as e:
        logger.error(f"Erreur lors de la récuperation des donnees historiques : {e}")

if __name__ == "__main__":
    main()

[cafeiine2023]

Merci beaucoup. Je regarde çà.

[Stochastic]

XLIT investing

https://fr.readkong.com/page/xl-investing-tools-xlit-1183498

Capture d'écran 2023-12-03 191308.png (100.88 Kio) Vu 17759 fois

n'est plus en service !...

[cafeiine2023]

script lancé et adopté. Merci Deep !

[cafeiine2023]

Subtilité 1 : Pour avoir également les DATES en sortie CSV, il faut faire :

Code : #

prices_df.to_csv(filename, index=True)

et non

Code : #

prices_df.to_csv(filename, index=False)

Va comprendre... Tant que ça marche....

Subtilité 2 : les prix ne sont pas strictement identiques à ceux dans PRT.
(1) Il faut prendre le Bid Ask et prendre la moyenne
(2) Il subsite une légère différence, ça se joue entre 0pts et 1pts.

Je m'amuse.

[cafeiine2023]

ERROR:__main__:Erreur lors de la récuperation des donnees historiques : error.public-api.exceeded-account-historical-data-allowance

Fin de la partie pour aujourd'hui

[Deep]

Par contre je bloque avec le streaming....j'arrive à me connecter mais ma souscription ne semble pas correcte, si tu as des pistes je suis preneur

[Deep]

Voici ou j'en suis avec le streaming......

Code : #

# -*- coding: utf-8 -*-
import logging
from trading_ig import IGService, IGStreamService
from lightstreamer.client import Subscription, LightstreamerClient, SubscriptionListener, ItemUpdate

# Configuration du logger
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class MarketListener(SubscriptionListener):
    def onItemUpdate(self, update: ItemUpdate):
        """ Gere les mises a jour de l'abonnement. """
        logger.info(
            f"{update.getItemName()} - "
            f"UTM: {update.getValue('UTM')}, "
            f"Open: {update.getValue('LTP_OPEN')}, "
            f"High: {update.getValue('LTP_HIGH')}, "
            f"Low: {update.getValue('LTP_LOW')}, "
            f"Close: {update.getValue('LTP_CLOSE')}, "
            f"Tick Count: {update.getValue('ballots_TICK_COUNT')}"
        )

def main():
    # Paramètres de l'API IG
    username = "xxxx"
    password = "xxxxx"
    api_key = "9cfc10a59caxxxxxxxxxxxxxx"
    acc_type = "LIVE"
    acc_number = "xxx"

    # Configuration du service IG
    ig_service = IGService(username, password, api_key, acc_type, acc_number=acc_number)
    ig_stream_service = IGStreamService(ig_service)
    ig_stream_service.create_session()

    # Récupération de l'URL Lightstreamer
    ls_endpoint = ig_service.read_session()["lightstreamerEndpoint"]
    ls_url = ls_endpoint  # Utilisez l'URL de base sans transformation

    print("URL Lightstreamer:", ls_url)  # Pour vérifier l'URL

    # Abonnement aux données du marché
    epic = "IX.D.SPTRD.IFE.IP"
    market_subscription = Subscription(
        mode="MERGE",
        items=[f"CHART:{epic}:1MINUTE"],  
        fields=["UTM", "LTP_OPEN", "LTP_HIGH", "LTP_LOW", "LTP_CLOSE", "ballots_TICK_COUNT"]  
    )
    market_subscription.addListener(MarketListener())

    # Connexion au serveur Lightstreamer
    ls_client = LightstreamerClient(ls_url, adapterSet="DEFAULT")
    ls_client.connect()
    ls_client.subscribe(market_subscription)

    # Attendez ici pour les mises à jour
    input("Appuyez sur Entree pour arreter...\n")

    # Se déconnecter
    ls_client.unsubscribe(market_subscription)
    ls_client.disconnect()
    ig_stream_service.logout()

if __name__ == "__main__":
    main()

[Deep]

finalement j'y suis arrivé, voici un script qui charge les données en stream pour du 1 minute...il y a une classe qui épure le flux de facon à ne pas surcharger le csv en ne gardant que la temporalité une minute.......
Attention il faut absolument installer: python -m pip install lightstreamer-client-lib

Code : #

# -*- coding: utf-8 -*-
import logging
import sys
import pandas as (mot censuré merci de rester poli)
from trading_ig import IGService, IGStreamService
from lightstreamer.client import Subscription, LightstreamerClient, SubscriptionListener, ItemUpdate, ConsoleLoggerProvider, ConsoleLogLevel

# Configuration du logger standard
logger = logging.getLogger(__name__)
logging.basicConfig(stream=sys.stdout, level=logging.INFO, format="%(message)s")

# Configuration du logger pour Lightstreamer
loggerProvider = ConsoleLoggerProvider(ConsoleLogLevel.INFO)
LightstreamerClient.setLoggerProvider(loggerProvider)

class ChartListener(SubscriptionListener):
    last_written_minute = None  # Variable pour suivre la dernière minute écrite

    def onItemUpdate(self, update: ItemUpdate):
        logger.info(f"Data received for {update.getItemName()}:")
        data = {}
        try:
            fields = ["UTM", "BID_OPEN", "BID_HIGH", "BID_LOW", "BID_CLOSE", "ballots_END", "ballots_TICK_COUNT"]
            for field in fields:
                value = update.getValue(field)
                if value is not None:
                    data[field] = value
                    logger.info(f"{field}: {value}")

            # Convertir UTM en temps et vérifier si c'est une nouvelle minute
            current_time = (mot censuré merci de rester poli).to_datetime(int(data["UTM"]), unit='ms')
            current_minute = current_time.floor('T')
            if current_minute != ChartListener.last_written_minute:
                # Ajout de la colonne de temps formatée
                data['Timestamp'] = current_time.strftime('%Y-%m-%d %H:%M:%S')
                # Écriture des données dans un fichier CSV
                (mot censuré merci de rester poli).DataFrame([data]).to_csv("streaming_data.csv", mode='a', header=False, index=True)
                ChartListener.last_written_minute = current_minute
        except Exception as e:
            logger.error(f"Error in onItemUpdate: {e}")

def main():
    # Paramètres de l'API IG et connexion
    username = 
    password = 
    api_key =
    acc_type = 
    acc_number = 

    ig_service = IGService(username, password, api_key, acc_type, acc_number=acc_number)
    ig_service.create_session()

    # Récupération des tokens CST et X-SECURITY-TOKEN
    response = ig_service.session
    cst_token = response.headers['CST']
    x_sec_token = response.headers['X-SECURITY-TOKEN']

    # Récupération de l'URL Lightstreamer
    ls_endpoint = ig_service.read_session()["lightstreamerEndpoint"]

    # Configuration du service IG Streaming
    ls_client = LightstreamerClient(ls_endpoint, adapterSet="DEFAULT")
    ls_client.connectionDetails.setUser(acc_number)
    ls_client.connectionDetails.setPassword(f"CST-{cst_token}|XST-{x_sec_token}")

    # Création d'un fichier CSV avec en-tête
    (mot censuré merci de rester poli).DataFrame(columns=["UTM", "BID_OPEN", "BID_HIGH", "BID_LOW", "BID_CLOSE", "ballots_END", "ballots_TICK_COUNT"]).to_csv("streaming_data.csv", index=False)

    # Configuration de la souscription
    epic = "IX.D.SPTRD.IFE.IP"
    scale = "1MINUTE"
    fields = ["UTM", "BID_OPEN", "BID_HIGH", "BID_LOW", "BID_CLOSE", "ballots_END", "ballots_TICK_COUNT"]
    items = [f"CHART:{epic}:{scale}"]

    chart_subscription = Subscription(
        mode="MERGE",
        items=items,
        fields=fields
    )
    chart_subscription.addListener(ChartListener())

    # Connexion au serveur Lightstreamer
    try:
        ls_client.connect()
        logger.info("Connected to Lightstreamer server")
    except Exception as e:
        logger.error(f"Error connecting to Lightstreamer server: {e}")
        return

    # Souscription
    try:
        ls_client.subscribe(chart_subscription)
        logger.info("Subscription request sent")
    except Exception as e:
        logger.error(f"Error during subscription: {e}")

    # Boucle principale
    try:
        while True:
            pass
    except KeyboardInterrupt:
        logger.info("Interrupted by user, shutting down")
    except Exception as e:
        logger.error(f"Error in main loop: {e}")

    # Déconnexion
    try:
        ls_client.unsubscribe(chart_subscription)
        ls_client.disconnect()
        ig_service.logout()
        logger.info("Disconnected and cleaned up")
    except Exception as e:
        logger.error(f"Error during disconnection and cleanup: {e}")

if __name__ == "__main__":
    main()

[cafeiine2023]

Joli ! Félicitations !

[DarkPoule]

Salut Deep, j'utilise beaucoup l'API stream et rest, on peut en parler si tu veux.

[cafeiine2023]

Suite demande d'un camarade, voici l'utilitaire simplifié pour récupérer les cotations sous IG, C'est une variante de code de Deep (grand merci Deep).

Téléchargement des cotations sous IG

Code : #

--------------------------------------------------------------------------------
                   DOWNLOAD COTATIONS FROM IG
--------------------------------------------------------------------------------

1/ install (windows/debian)
2/ configure 'download.py'
3/ Launch 'download.py'


--------------------------------------------------------------------------------
install on windows

1/ install python
https://www.python.org/downloads/windows/
- A la première fenêtre d'installation cochez les 2 cases : PATH et l'autre 

2/ pip : pandas, multidict

pip install pandas

pip install multidict

pip install trading_ig

--------------------------------------------------------------------------------
install on debian 

su root
apt-get install sudo
/sbin/usermod -aG sudo debian

# python
sudo apt update
sudo apt install python3 python3-pip
sudo python3 --version

# pandas
sudo apt install python3-pandas
sudo apt install python3-multidict
# trading_ig
# https://www.makeuseof.com/fix-pip-error-externally-managed-environment-linux/
cd /usr/lib/python3.11/
sudo rm EXTERNALLY-MANAGED
sudo pip install trading_ig

--------------------------------------------------------------------------------
configure  'download.py' 

1/ Generate API_KEY on IG
Authenticate > Configuration > Paramètres de trading > Clés de l'API WEB

2/ configure  'download.py'
 
# Paramètres de l'API IG
username = "my-ig-username"
password = "my-ig-password"
api_key = "my-ig-api-key--------------------------"
acc_type = "LIVE"
acc_number = "my-ig-account-number"

# Be carefull, there is a limitation per day for downloading


--------------------------------------------------------------------------------
launch  'download.py' 

1/ launch // python3 download.py ${INDICE} "${PERIOD}" "${DATE1}" "${DATE2}"

python download.py IX.D.NASDAQ.IFE.IP "30min" "2023-12-01 09:00:00" "2023-12-01 09:30:00"


2/ Get output

IX.D.NASDAQ.IFE.IP_20231201_090000_20231201_093000.csv


--------------------------------------------------------------------------------
Annnexes


https://www.andlil.com/forum/api-ig-et-recuperation-des-donnees-de-marches-t57246.html

./download/file.php?id=284680

(2.16 Kio) Téléchargé 256 fois

download/file.php?id=284680


Nommage des indices sous IG
Pour récupérer le nommage des indices à télécharger, récupérer un autre outil, voir https://github.com/cafeiine/trading/tree/main , fichier "IG_Connect_v0.6_public.xlsb", onglet "10-search-epic". Il y a des exemples, et vous pouvez en récupérer avec des recherches par nom.

[ChristelleP]

merci Cafeine

[cafeiine2023]

Limites de l'API IG,

[ChristelleP]

--up--