Exchange Rates

Purpose

This worker populates the Exchange Rates fact table with exchange rates for each defined currency pair. The data is used to consistently convert monetary amounts across assets and dates.

Inputs

The worker consumes a dataframe of currency pairs and dates sourced from the following Omora Labs components:

Each row represents a single currency pair to be resolved for a specific date.

Required columns:

Column	Description
date	Date represented as `YYYY-MM-DD`. Links to Dates semantic contract
currency_pair_id	Currency Pair identifier. Links to Currency Pairs semantic contract
base_currency	Base Currency as per ISO 4217, e.g. "EUR", "USD"
quote_currency	Quote Currency as per ISO 4217, e.g. "EUR", "USD"

External Data Source

Exchange rates are fetched from the public open-source currency API provided by @fawazahmed0/currency-api.

The API exposes historical daily snapshots and returns exchange rates indexed by base currency.

Work

For each date, currency_pair combination:

Extract the base and quote currencies.
Request the historical exchange rates for the base currency on the given date.
Select the quote currency rate from the response.
Normalize the result into the internal fact structure.

Each exchange rate is materialized as a single factual observation per date and currency pair.

import polars as pl
import requests


def run_worker(currency_pairs_df: pl.DataFrame) -> pl.DataFrame:
    """
    Fetch exchange rates for given dates and currency pairs.

    Args:
        currency_pairs_df: DataFrame with columns: date, currency_pair_id, base_currency, quote_currency

    Returns:
        DataFrame with exchange rates
    """
    exchange_rates = []
    for row in currency_pairs_df.iter_rows(named=True):
        date = row["date"]
        currency_pair_id = row["currency_pair_id"]
        base = row["base_currency"]
        quote = row["quote_currency"]

        print(f"Fetching exchange rate for date {date} and FX Pair {base}/{quote}")

        response = requests.get(
            f"https://cdn.jsdelivr.net/npm/@fawazahmed0/currency-api@{date}/v1/currencies/{base.lower()}.json"
        )
        rates = response.json()[base.lower()]
        exchange_rates.append(
            {
                "date": date,
                "currency_pair_id": currency_pair_id,
                "value": rates[quote.lower()],
            }
        )

    return pl.DataFrame(exchange_rates)

Orchestrator

A simple orchestrator suffices for handling the inputs and the worker:

from .worker import run_worker


def orchestrator(db) -> None:
    """
    Worker orchestrator
    """
    currency_pairs_df = db.get_currency_pairs_df()
    exchange_rates_df = run_worker(currency_pairs_df)

    db.insert_pl_dataframe(exchange_rates_df, "exchange_rates")

Output

The worker produces a dataframe compatible with the Exchange Rates fact table.

Each record represents the exchange rate for a given currency pair on a specific date, uniquely identified by date, currency_pair_id.