Revolut API

IMPORTANT PLEASE READ: in the example Workflows in this topic we will be trading with real money and other financial assets like cryptocurrency. You are free to use,share, extend, modify etc. the code examples herein BUT IN NO CIRCUMSTANCES I or the company KNIME will be responsible for any financial losses or gains you may incur.

This is about trading which is by nature risky.

Also. Please be careful when you share any code NOT to share any real access keys, passwords etc.

In the example Workflows presented here I purposely use small amounts of money.

In this post I will try to show you how to access and operate on Cryptocurrency assets using KNIME. In this concrete example I will be using a Revolut  X account.  I suppose other trading platforms will have similar APIs and the code shown here can be adapted to these, but I had to choose one .

The final goal will be to build an agent that can make autonomous trading decisions, but with limits configurable by you!

To access the data from Revolut we will be using a Python Scripting node. On this page you can find additional information on how to to make Python work together with KNIME.

The first thing we will need to do is get an API key and an Ed25519 key pair for Revolut. The documentation on how to do this can be found here 

The two necessary pieces we will need in our Python node and got from the above are:

• the API key (a string)

• the file containing the private key (in this example called private.pem)

Below is a screenshot of the first workflow we will build. Note that the Variable Creator and Variable to Table Row nodes are just there to create some 'dummy' input for the Python script node

And below is the configuration of the Python Script node: For obvious reasons I have changed the API key for a placeholder.

Note that in this example,  the private.pem file would be located in the same folder as where our workflow resides. 

import knime.scripting.io as knio

import requests

import base64

import time

from pathlib import Path

from nacl.signing import SigningKey

from cryptography.hazmat.primitives import serialization

from cryptography.hazmat.backends import default_backend

# 1. Load your Private Key

pem_data = Path("private.pem").read_bytes()

private_key_obj = serialization.load_pem_private_key(

   pem_data,

   password=None,

   backend=default_backend()

)

# Extract raw bytes for PyNaCl

raw_private = private_key_obj.private_bytes(

   encoding=serialization.Encoding.Raw,

   format=serialization.PrivateFormat.Raw,

   encryption_algorithm=serialization.NoEncryption()

)

# 2. Prepare the message

timestamp = str(int(time.time() * 1000))

method = "GET"

path = "/api/1.0/balances"

query = ""  # No query for balances endpoint

body = ""   # Empty for GET

# Concatenate without separators

message = f"{timestamp}{method}{path}{query}{body}".encode('utf-8')

# 3. Sign and Encode

signing_key = SigningKey(raw_private)

signed = signing_key.sign(message)

signature = base64.b64encode(signed.signature).decode()

url = "https://revx.revolut.com/api/1.0/balances"

headers = {

   'Accept': 'application/json',

   'X-Revx-Timestamp': timestamp,

   'X-Revx-Signature': signature,

   'X-Revx-API-Key': '<here is where you put your API key>'

}

response = requests.get(url, headers=headers)

print(response.content)

# Optionally, parse the response as JSON and output as a KNIME table

import pandas as pd

try:

   data = response.json()

   # If the response is a list of balances

   if isinstance(data, list):

       df = pd.DataFrame(data)

   # If the response is a dict with a key containing the balances

   elif isinstance(data, dict):

       # Try to find a likely key

       for key in ['balances', 'data', 'results']:

           if key in data and isinstance(data[key], list):

               df = pd.DataFrame(data[key])

               break

       else:

           # Fallback: just wrap the dict

           df = pd.DataFrame([data])

   else:

       df = pd.DataFrame([{"response": str(data)}])

except Exception as e:

   df = pd.DataFrame([{"error": str(e), "response_text": response.text}])

knio.output_tables[0] = knio.Table.from_pandas(df)

Below a screenshot of what I get when running the workflow. Seems I have about 0,000376 BTC and 3,83 € in this account.

This is still a very simple example, but I will build on this later on.  On this page we will see another simple example of a workflow showing the history of orders we have placed.