adona-robot — market data by capacityA forex data API in Python
From the API key to candles and live prices in Python, with two packages, requests and websockets: the quickstart's own code, run against the API.
- What you need
- Python and two packages: pip install requests "websockets>=13". The API key comes from your account page; keep it in the ADONA_API_KEY environment variable on your server, never in a web page or a repository.
- Candles by REST
- history.py exchanges the key for a token that lives 12 hours, then reads the latest 1 000 one-minute EURUSD candles, oldest first, ending with the bar still forming. Prices are strings, exactly as served: convert them at the last moment, for instance into a DataFrame. For more than a page, follow next_before, up to 5 000 bars a page.
- Live prices
- stream.py takes a single-use ticket with the token and opens the WebSocket offering two subprotocols, adona.data.v1 and the ticket. It subscribes to the live bid and ask and to one-minute candles, and prints every frame until you stop it.
- When the API says no
- Both scripts stop when an HTTP call is refused and print the status and the API's answer. Branch on the code, the part of the detail field before the first colon: a wrong key answers INVALID_CUSTOMER_KEY, an ended trial CUSTOMER_TRIAL_EXPIRED. On the socket, a refusal is an error frame, printed like the others, or a close with code 1008 and a reason. Every code, and what to do about it, is on the reference page.
- What it costs
- Billed in slices of 1 000 bars, rounded up per response: the 1 000 candles of history.py are 1 slice, and the recent bars each candle channel first sends on subscribe are billed the same way. A page that carries the bar still forming, like history.py's, counts as following the symbol live: past 80% of its open hours in a month, that symbol adds CHF 150. Plans from CHF 500 a month; the trial: 7 days, on EURUSD · XAUUSD.
Facts as of 03.10.2026.
The code: history, then the live stream
# pip install requests
import os
import requests
API = "https://api.adona-robot.com"
# your server exchanges the key for a token
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"]
# the token reads candles
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"]
# a ticket for the stream, from the token
ticket = post("/v1/ws-ticket", token)["ticket"]
# the stream: live quotes and one-minute bars
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())
The quickstart: from the key to the live stream, step by step