Real Time Currency Arbitrage

rt_curr_arbitrage.py

import requests
from typing import Tuple, List
from math import log

currencies = ('PLN', 'EUR', 'USD', 'RUB', 'INR', 'MXN')
api_key = 'fe69bd3037f74f699cf2bf8070b44374'

def get_rates(currencies: Tuple[str, ...], api_key: str) -> List[List[float]]:
    rates = []
    response = requests.get(f'https://openexchangerates.org/api/latest.json?app_id={api_key}')
    data = response.json()['rates']
    rates_to_usd = {currency: data.get(currency, 0) for currency in currencies}

    for currency in currencies:
        rates.append([rates_to_usd[currency] / rates_to_usd[curr] if rates_to_usd[curr] != 0 else 0 for curr in currencies])
    return rates

def negate_logarithm_convertor(graph: Tuple[Tuple[float]]) -> List[List[float]]:
    ''' log of each rate in graph and negate it'''
    result = [[-log(edge) if edge != 0 else float('inf') for edge in row] for row in graph]
    return result

def arbitrage(currency_tuple: tuple, rates_matrix: Tuple[Tuple[float, ...]]):
    ''' Calculates arbitrage situations and prints out the details of this calculations'''

    trans_graph = negate_logarithm_convertor(rates_matrix)

    n = len(trans_graph)
    min_dist = [float('inf')] * n

    pre = [-1] * n

    for source in range(n):
        min_dist[source] = 0

        for _ in range(n-1):
            for source_curr in range(n):
                for dest_curr in range(n):
                    if source_curr == dest_curr or trans_graph[source_curr][dest_curr] == float('inf'):
                        continue
                    if min_dist[dest_curr] > min_dist[source_curr] + trans_graph[source_curr][dest_curr]:
                        min_dist[dest_curr] = min_dist[source_curr] + trans_graph[source_curr][dest_curr]
                        pre[dest_curr] = source_curr

        for source_curr in range(n):
            for dest_curr in range(n):
                if source_curr == dest_curr or trans_graph[source_curr][dest_curr] == float('inf'):
                    continue
                if min_dist[dest_curr] > min_dist[source_curr] + trans_graph[source_curr][dest_curr]:
                    print_cycle = [dest_curr, source_curr]
                    while pre[source_curr] not in print_cycle:
                        source_curr = pre[source_curr]
                        print_cycle.append(source_curr)
                    print_cycle.append(pre[source_curr])
                    if print_cycle[0] == print_cycle[-1]:
                        print(f"Arbitrage Opportunity: \n{' --> '.join([currencies[p] for p in print_cycle[::-1]])}")

if __name__ == "__main__":
    rates = get_rates(currencies, api_key)
    arbitrage(currencies, rates)

@anilpai API key 🫣