Une API de données forex en Python

De la clé d'API aux bougies et aux prix en direct en Python, avec deux paquets, requests et websockets : le code même des premiers pas, exécuté contre l'API.

Ce qu'il vous faut

Python et deux paquets : pip install requests "websockets>=13". La clé d'API vient de votre page de compte ; gardez-la dans la variable d'environnement ADONA_API_KEY sur votre serveur, jamais dans une page web ni dans un dépôt.

Les bougies par REST

history.py échange la clé contre un jeton valable 12 heures, puis lit les 1 000 dernières bougies d'une minute d'EURUSD, de la plus ancienne à la plus récente, en se terminant par la bougie en cours. Les prix sont des chaînes, tels que servis : convertissez-les au dernier moment, par exemple dans un DataFrame. Au-delà d'une page, suivez next_before, jusqu'à 5 000 bougies par page.

Les prix en direct

stream.py prend un ticket à usage unique avec le jeton et ouvre le WebSocket en proposant deux sous-protocoles, adona.data.v1 et le ticket. Il s'abonne à l'achat et la vente en direct et aux bougies d'une minute, et affiche chaque trame jusqu'à ce que vous l'arrêtiez.

Quand l'API dit non

Les deux scripts s'arrêtent quand un appel HTTP est refusé et affichent le statut et la réponse de l'API. Testez le code, la partie du champ detail avant le premier deux-points : une mauvaise clé répond INVALID_CUSTOMER_KEY, un essai terminé CUSTOMER_TRIAL_EXPIRED. Sur le socket, un refus est une trame d'erreur, affichée comme les autres, ou une fermeture avec le code 1008 et une raison. Chaque code, et ce qu'il faut faire, est sur la page de référence.

Ce que cela coûte

Facturé en tranches de 1 000 bougies, arrondies par réponse : les 1 000 bougies de history.py font 1 tranche, et les bougies récentes que chaque canal de bougies envoie d'abord à l'abonnement sont facturées de la même façon. Une page qui porte la bougie en cours, comme celle de history.py, compte comme un suivi en direct du symbole : au-delà de 80 % de ses heures d'ouverture dans le mois, ce symbole ajoute CHF 150. Vendu à la capacité, dès CHF 500 par mois ; l'essai : 7 jours, sur EURUSD · XAUUSD.

Faits à la date du 03.10.2026.

# pip install requests
import os

import requests

API = "https://api.adona-robot.com"

# votre serveur échange la clé contre un jeton
r = requests.post(
    f"{API}/v1/tokens",
    headers={"Authorization": f"Bearer {os.environ['ADONA_API_KEY']}"},
    json={"end_user_id": "u-42"},
    timeout=30,
)
if not r.ok:
    raise SystemExit(f"refused: {r.status_code} {r.text}")
token = r.json()["access_token"]

# le jeton lit les bougies
r = requests.get(
    f"{API}/v1/candles",
    headers={"Authorization": f"Bearer {token}"},
    params={"symbol": "EURUSD", "timeframe": "M1", "limit": 1000},
    timeout=30,
)
if not r.ok:
    raise SystemExit(f"refused: {r.status_code} {r.text}")
page = r.json()
print(len(page["candles"]), "bars, newest:", page["candles"][-1] if page["candles"] else None)

# pip install requests "websockets>=13"
import asyncio
import json
import os

import requests
from websockets.asyncio.client import connect

API = "https://api.adona-robot.com"
STREAM = "wss://api.adona-robot.com/v1/stream"


def post(path, bearer, body=None):
    r = requests.post(API + path, headers={"Authorization": f"Bearer {bearer}"}, json=body, timeout=30)
    if not r.ok:
        raise SystemExit(f"refused: {r.status_code} {r.text}")
    return r.json()


async def main():
    token = post("/v1/tokens", os.environ["ADONA_API_KEY"], {"end_user_id": "u-42"})["access_token"]
    # un ticket pour le flux, à partir du jeton
    ticket = post("/v1/ws-ticket", token)["ticket"]
    # le flux : prix en direct et barres d'une minute
    async with connect(STREAM, subprotocols=["adona.data.v1", f"ticket.{ticket}"]) as ws:
        await ws.send(json.dumps({
            "type": "subscribe", "request_id": "sub",
            "payload": {"channels": ["quote:EURUSD", "candle:EURUSD:M1"]},
        }))
        async for message in ws:
            print(json.loads(message))


asyncio.run(main())