Build a Currency Exchange Rate Pipeline

Build a Python ETL pipeline that pulls live currency data into SQLite.

Introduction

30 Second Summary

Every time you check a currency converter app before an international trip, you're seeing the end result of a data pipeline. Someone had to pull exchange rates from a source, clean them up, and store them in a database so the app can show you the latest numbers.

In this project, you will build your own version of that pipeline using Python. You will pull real exchange rate data from a live API, clean it with pandas, and store it in a SQLite database you can query for insights.

What You'll Build

A Python ETL pipeline that extracts currency exchange rates from the Frankfurter API, transforms the raw data into a clean structure, and loads it into a SQLite database.

By the end of this project, you'll have:

A working ETL pipeline that pulls real currency data from a live REST API with no API key required.
A SQLite database loaded with 30 days of exchange rates for 5 major currencies.
SQL queries that surface summary statistics like average rates, minimums, and maximums.
Secret Mission: Add trend analysis to identify the most volatile currency over the past month.

Are there any prerequisites?

You should be comfortable with basic Python concepts like variables, functions, loops, and dictionaries. No prior experience with data engineering or databases is needed.

Before We Start

Before We Start...

Set Up Your Python Project

Before you can build your ETL pipeline, you need a clean workspace. A dedicated project folder and virtual environment keep your dependencies separate from everything else on your machine.

In this step, you'll verify your Python version, create a project folder, and install the libraries your pipeline depends on. You'll also create the main script file that will hold all your pipeline code.

In this step, get ready to:

Verify that Python 3.11+ is installed.
Create a project folder with a virtual environment inside it.
Install dependencies and create pipeline.py.

Verify your Python version and create a project folder

Your pipeline will use pandas for data transformation. Version 3.0+ of pandas requires Python 3.11 or higher.

Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
Type Terminal and press Enter to open it.

Check your Python version by running this command:

macOS

python3 --version

Windows

python --version

✔️ I see Python 3.11 or higher

Your Python version is compatible. Move on to the next instruction.

ⓧ I see an older version or an error

Download the latest version of Python from python.org/downloads.
During installation on Windows, make sure to check the Add Python to PATH option.
After installing, close and reopen your terminal, then run the version check again.

Move to your Desktop by running this command:

cd ~/Desktop

Create your project folder and move into it by running these commands:

mkdir currency-etl
cd currency-etl

Confirm you're in the right folder by running:

pwd

You should see a path ending in currency-etl.

Create a virtual environment inside your project by running this command:

macOS

python3 -m venv venv

Windows

python -m venv venv

What is a virtual environment?

A virtual environment is an isolated Python environment inside your project folder. Any packages you install while it's active only exist within this environment.

This prevents version conflicts between different Python projects on your machine.

Activate the virtual environment by running:

macOS

source venv/bin/activate

Windows

.\venv\Scripts\activate

You should see (venv) appear at the start of your terminal prompt. This means the virtual environment is active.

Don't see (venv) in your prompt?

Make sure you're inside the currency-etl folder. Run pwd to check your location.
On macOS, make sure you typed source before the path.
On Windows, if you see an execution policy error, close this terminal, reopen PowerShell as Administrator, and run the following command before trying again:

Set-ExecutionPolicy -ExecutionPolicy RemoteSigned -Scope CurrentUser

Help me activate my Python virtual environment.

Install the pandas and requests packages

Your pipeline needs two external libraries. pandas handles data transformation and analysis. The requests library handles HTTP calls to the exchange rate API.

Install both packages by running this command:

pip install pandas==3.0.2 requests==2.33.1

You should see Successfully installed near the end of the terminal output.

Seeing an install error?

Make sure your virtual environment is active. You should see (venv) at the start of your terminal prompt.
If you see a Permission denied error, you're likely installing outside the virtual environment. Activate it first, then retry the install command.

Help me install pandas and requests in my Python project.

Create pipeline.py and add the import statements

Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
Type Visual Studio Code and press Enter to open it.

Click File in the top menu bar.
Click Open Folder.
Navigate to your Desktop and select the currency-etl folder.
Click Open.

In the Explorer sidebar on the left, click the New File icon (it looks like a page with a plus sign).
Type pipeline.py and press Enter.

Set up the imports your pipeline needs by adding this code to pipeline.py:

# HTTP requests library for calling the exchange rate API
import requests
# Data analysis library for transforming and cleaning data
import pandas as pd
# Built-in module for working with SQLite databases
import sqlite3
# Date utilities for calculating time ranges
from datetime import date, timedelta

What do these imports do?

requests sends HTTP requests to the Frankfurter API to fetch exchange rate data.
pandas (imported as pd) provides DataFrames for transforming and cleaning the raw data.
sqlite3 is Python's built-in module for working with SQLite databases. No separate install needed.
date and timedelta from the datetime module let you calculate date ranges for historical data.

Save pipeline.py by pressing Cmd+S (macOS) or Ctrl+S (Windows).
Switch back to your terminal and confirm the imports work by running this command:

macOS

python3 pipeline.py

Windows

python pipeline.py

You should see no output and no errors. A clean run means all four imports loaded successfully.

Seeing an error when you run the script?

If you see ModuleNotFoundError, your virtual environment might not be active. Check that (venv) appears in your terminal prompt. If not, activate it and rerun the pip install command from the previous substep.
If you see a SyntaxError, double check that each import line matches the code above exactly.

Help me fix an import error in my Python pipeline.

✔️ Awesome, I've got everything!

Great. Double check that you've saved your file before moving on.

ⓧ I'd like to double check the full code

Here is the complete pipeline.py file so far. Make sure yours matches exactly.

import requests
import pandas as pd
import sqlite3
from datetime import date, timedelta

Your workspace is ready with all the tools you need. Next up, you'll write your first function to pull live exchange rate data from the Frankfurter API.

Extract Data from the Frankfurter API

Your Python environment is ready and pipeline.py has its import statements in place. Now it's time to start building the actual pipeline.

Every ETL pipeline begins with the same first stage: extracting raw data from an external source. In this step, you'll write a function that pulls live currency exchange rates from the Frankfurter API, a free REST API that returns real financial data from 55 central banks with no API key or signup required.

In this step, get ready to:

Write an extract_rates() function that calls the Frankfurter API.
Add a temporary test block to run the function.
Verify that live exchange rate data comes back from the API.

Write the extract_rates() function

The Frankfurter API provides exchange rates through a single endpoint at https://api.frankfurter.dev/v2/rates. Your function will accept parameters for the base currency, which currencies to compare against, and an optional date range.

Why Frankfurter over other APIs?

Most financial APIs (like Alpha Vantage or OpenWeatherMap) require you to sign up and manage an API key before you can make a single request. Frankfurter is completely free with no authentication, no daily limits, and real data from 55 central banks.

This means you can focus on learning ETL patterns instead of wrestling with API key management.

In pipeline.py, find the import statements at the top of the file.
Copy the following function below the imports:

def extract_rates(base="USD", quotes=None, start_date=None, end_date=None):
    """Extract exchange rates from the Frankfurter API."""
    # Set the API endpoint and required base currency parameter
    url = "https://api.frankfurter.dev/v2/rates"
    params = [f"base={base}"]

    # Add optional filters for specific currencies and date range
    if quotes:
        params.append(f"quotes={','.join(quotes)}")
    if start_date:
        params.append(f"from={start_date}")
    if end_date:
        params.append(f"to={end_date}")

    # Build the full URL manually (the API requires literal commas, not %2C)
    full_url = f"{url}?{'&'.join(params)}"

    # Send the GET request and parse the JSON response
    response = requests.get(full_url)
    response.raise_for_status()
    data = response.json()

    print(f"Extracted rates (base: {base})")
    return data

What does this code do?

url points to the Frankfurter API's rates endpoint. params is a list that builds each query parameter as a string like base=USD.
The if blocks append optional parameters. quotes filters which currencies to return. start_date and end_date set a date range using the API's from and to parameters.
full_url joins the endpoint with the query string. Building the URL manually keeps commas as literal characters. The API rejects percent-encoded commas (%2C).
response.json() parses the API's JSON response into a Python list of dictionaries that your code can work with.

Save pipeline.py.

You won't see any output yet. The next substep adds a test block so you can actually run the function.

Add a test block to call extract_rates()

Python's if __name__ == "__main__": guard runs code only when you execute the file directly. This is a handy pattern for testing functions as you build them.

At the bottom of pipeline.py, below the extract_rates() function, add this test block:

if __name__ == "__main__":
    # Call extract_rates with default USD base and print the result
    raw = extract_rates()
    print(raw)

What does this code do?

extract_rates() is called with no arguments, so it uses the default base currency of USD and fetches the latest rates for all available currencies.
print(raw) displays the full JSON response so you can inspect exactly what the API returns.

Save pipeline.py.

Run the script and check the API response

Run your script by entering this command in your terminal:

python pipeline.py

You should see two things printed in your terminal: a confirmation message saying Extracted rates (base: USD) followed by a dictionary of currency exchange rates.

What should I see?

The output is a Python list of dictionaries. Each dictionary has four keys: date, base, quote, and rate. You'll see one entry per currency, showing its exchange rate relative to USD.

🙋‍♀️ Seeing an error?

If you see ModuleNotFoundError: No module named 'requests', your virtual environment is not activated. Run source venv/bin/activate (macOS) or venv\Scripts\activate (Windows), then try again.
If you see ConnectionError, check your internet connection. The Frankfurter API requires an active connection to fetch live data.
If you see SyntaxError, double check that your function definition and if __name__ block match the code above exactly. Pay attention to indentation and quotation marks.
Help me debug an error when running pipeline.py

✔️ Awesome, I've got everything!

Great. Double check you've saved pipeline.py before moving on.

ⓧ I'd like to double check the full code

Here is what your complete pipeline.py should look like at this point.

import requests
import pandas as pd
import sqlite3
from datetime import date, timedelta


def extract_rates(base="USD", quotes=None, start_date=None, end_date=None):
    """Extract exchange rates from the Frankfurter API."""
    url = "https://api.frankfurter.dev/v2/rates"
    params = [f"base={base}"]
    if quotes:
        params.append(f"quotes={','.join(quotes)}")
    if start_date:
        params.append(f"from={start_date}")
    if end_date:
        params.append(f"to={end_date}")

    full_url = f"{url}?{'&'.join(params)}"

    response = requests.get(full_url)
    response.raise_for_status()
    data = response.json()

    print(f"Extracted rates (base: {base})")
    return data


if __name__ == "__main__":
    raw = extract_rates()
    print(raw)

Your extraction function is pulling live data from the Frankfurter API. Next, you'll transform this raw JSON into a clean, structured DataFrame using pandas.

Transform the Raw Data with Pandas

Your extract_rates() function is pulling live exchange rate data from the Frankfurter API. The raw response is a JSON blob with nested dictionaries, which isn't easy to analyze or query directly.

The "T" in ETL stands for Transform. In this step, you'll write a function that reshapes messy JSON into a clean, structured pandas DataFrame with proper data types and sorting.

In this step, get ready to:

Write a transform_rates() function that flattens nested JSON into a DataFrame.
Update the test block to pass extracted data through the transform function.
Run the script and verify the output is a clean, sorted DataFrame.

Write the transform_rates() function

The Frankfurter API v2 returns a flat list of rate records. Each record is a dictionary with date, base, quote, and rate fields.

Your function will reshape this list into a proper DataFrame with renamed columns, correct data types, and a calculated inverse rate column.

In pipeline.py, find the blank line after the extract_rates() function (just above if __name__ == "__main__":).
Add the transform_rates() function by copying the following code above the if __name__ block:

def transform_rates(raw_data):
    """Transform raw API JSON into a clean pandas DataFrame."""
    # Create DataFrame directly from the flat array of rate records
    df = pd.DataFrame(raw_data)

    # Rename 'quote' column to 'currency' for clarity
    df = df.rename(columns={"quote": "currency"})

    # Convert data types for proper sorting and analysis
    df["date"] = pd.to_datetime(df["date"])
    df["rate"] = pd.to_numeric(df["rate"], errors="coerce")
    df = df.dropna(subset=["rate"])
    df = df.sort_values(["currency", "date"]).reset_index(drop=True)
    # Calculate how much 1 unit of quote currency buys in base currency
    df["inverse_rate"] = (1 / df["rate"]).round(4)

    print(f"Transformed {len(df)} records")
    print(f"Currencies: {', '.join(sorted(df['currency'].unique()))}")
    return df

What does this code do?

pd.DataFrame(raw_data) converts the list of dictionaries directly into a DataFrame. Each dictionary becomes one row.
df.rename(columns={"quote": "currency"}) renames the quote column to currency for clearer readability in your analysis.
pd.to_datetime() converts date strings to proper datetime objects. pd.to_numeric(errors="coerce") converts rates to numbers, turning any bad values into NaN so dropna() can remove them.
inverse_rate shows how much 1 unit of the quote currency buys in the base currency. For example, if 1 USD = 0.87 EUR, then 1 EUR = 1.149 USD.

Save pipeline.py.

You won't see a change yet. The next substep connects this function to your test block.

Seeing a red underline on pd.DataFrame?

Make sure you pasted the function above the if __name__ block, not inside it. The function should be at the top level of the file (no indentation on the def keyword).
Check that your import pandas as pd line is still at the top of the file. If it was accidentally deleted, the pd reference won't resolve.
Why is my transform_rates function showing errors in VS Code?

Update the test block to use transform_rates()

The test block at the bottom of your file currently prints raw JSON. You need to route that data through your new transform function instead.

In pipeline.py, scroll to the if __name__ == "__main__": block at the bottom of the file.
Find this code:

if __name__ == "__main__":
    raw = extract_rates()
    print(raw)

Replace print(raw) with two new lines that pass the data through transform_rates() and print the resulting DataFrame, so the block looks like this:

if __name__ == "__main__":
    raw = extract_rates()
    df = transform_rates(raw)
    print(df)

Save pipeline.py.

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before moving on.

ⓧ I'd like to double check the full code

Here is the full pipeline.py file as it should look right now.

import requests
import pandas as pd
import sqlite3
from datetime import date, timedelta


def extract_rates(base="USD", quotes=None, start_date=None, end_date=None):
    """Extract exchange rates from the Frankfurter API."""
    url = "https://api.frankfurter.dev/v2/rates"
    params = {"base": base}
    if quotes:
        params["quotes"] = ",".join(quotes)
    if start_date:
        params["from"] = start_date
    if end_date:
        params["to"] = end_date

    response = requests.get(url, params=params)
    response.raise_for_status()
    data = response.json()

    print(f"Extracted rates (base: {base})")
    return data


def transform_rates(raw_data):
    """Transform raw API JSON into a clean pandas DataFrame."""
    df = pd.DataFrame(raw_data)
    df = df.rename(columns={"quote": "currency"})
    df["date"] = pd.to_datetime(df["date"])
    df["rate"] = pd.to_numeric(df["rate"], errors="coerce")
    df = df.dropna(subset=["rate"])
    df = df.sort_values(["currency", "date"]).reset_index(drop=True)
    df["inverse_rate"] = (1 / df["rate"]).round(4)

    print(f"Transformed {len(df)} records")
    print(f"Currencies: {', '.join(sorted(df['currency'].unique()))}")
    return df


if __name__ == "__main__":
    raw = extract_rates()
    df = transform_rates(raw)
    print(df)

Run the script and verify the output

Run the updated script by entering this command in your terminal:

python pipeline.py

You should see output that starts with the extraction message, followed by a transform summary and a printed DataFrame.

What should I see?

The output should include a line like Transformed X records and a list of currencies. Below that, you'll see a table with columns for date, base, currency, rate, and inverse_rate.

The dates should be in datetime format (like 2026-04-30), rates should be decimal numbers, and the rows should be sorted alphabetically by currency.

🙋‍♀️ Seeing an AttributeError?

If you see AttributeError: 'list' object has no attribute 'get', make sure your transform_rates() function uses pd.DataFrame(raw_data) directly, not raw_data.get("rates", {}). The Frankfurter API v2 returns a flat list, not a nested dictionary.
If the DataFrame is empty, make sure the test block passes raw (the result of extract_rates()) into transform_rates(raw). If you passed the function without the variable, it won't have any data to parse.
If you see an IndentationError, check that the def transform_rates line starts with zero indentation (at the left edge of the file), and that everything inside the function is indented by 4 spaces.
My pipeline.py script is failing when I try to transform the exchange rate data. Help me debug.

Your data is clean, typed, and structured. Next up, you'll load this DataFrame into a SQLite database and query it with SQL.

Load into SQLite and Query with SQL

Your DataFrame is clean, sorted, and ready to use. But there is a problem. The moment your script finishes running, that data disappears from memory.

In this step, you will load the transformed data into a SQLite database so it persists on disk. Then you will write SQL queries to pull insights like average rates and min/max values directly from the database.

In this step, get ready to:

Write a function that loads your DataFrame into a SQLite database.
Write a function that queries the database with SQL to surface summary statistics.
Run the full extract, transform, load, and query flow end to end.

Write the load function

The "L" in ETL stands for Load. This is where your cleaned data moves from a temporary DataFrame into a permanent home on disk.

Why SQLite?

SQLite is built into Python's standard library, so there is nothing extra to install. It stores your entire database in a single .db file.

It also uses standard SQL, so everything you learn here transfers directly to larger databases like PostgreSQL or MySQL.

In pipeline.py, find the end of your transform_rates() function (the line that says return df).
Add the following load_to_database() function below it, leaving two blank lines between the functions:

def load_to_database(df, db_path="exchange_rates.db", table_name="rates"):
    """Load a DataFrame into a SQLite database."""
    # Open a connection to the database file (creates it if it doesn't exist)
    conn = sqlite3.connect(db_path)
    # Write the DataFrame to a SQL table, replacing any existing data
    df.to_sql(table_name, conn, if_exists="replace", index=False)

    # Verify the load by counting rows in the table
    row_count = conn.execute(f"SELECT COUNT(*) FROM {table_name}").fetchone()[0]
    print(f"Loaded {row_count} records into '{table_name}' table")
    conn.close()

What does this code do?

sqlite3.connect(db_path) opens a connection to exchange_rates.db. If the file does not exist yet, SQLite creates it automatically.
df.to_sql() writes every row in your DataFrame straight into a SQL table called rates. The if_exists="replace" flag drops and recreates the table each time, so you always start fresh.
The SELECT COUNT(*) query confirms how many rows landed in the table. This is a quick sanity check that the load worked.

Save pipeline.py.
You will not see a change yet. The next substep adds the query function, and then you will run the full flow.

Write the query function

Loading data into a database is only useful if you can ask questions of it. Your next function runs two SQL queries that surface real insights from the exchange rate data.

In pipeline.py, find the end of your load_to_database() function (the line that says conn.close()).
Add the following query_database() function below it, leaving two blank lines between the functions:

def query_database(db_path="exchange_rates.db"):
    """Run sample queries against the loaded data."""
    conn = sqlite3.connect(db_path)

    # Query 1: Get the most recent rate for each currency
    print("\n--- Latest rate for each currency ---")
    result = pd.read_sql("""
        SELECT currency, date, rate, inverse_rate
        FROM rates
        WHERE date = (SELECT MAX(date) FROM rates)
        ORDER BY currency
    """, conn)
    print(result.to_string(index=False))

    # Query 2: Calculate summary statistics per currency
    print("\n--- Summary statistics per currency ---")
    result = pd.read_sql("""
        SELECT currency,
               ROUND(AVG(rate), 4) AS avg_rate,
               ROUND(MIN(rate), 4) AS min_rate,
               ROUND(MAX(rate), 4) AS max_rate
        FROM rates
        GROUP BY currency
        ORDER BY currency
    """, conn)
    print(result.to_string(index=False))

    conn.close()

What do these queries do?

The first query uses a subquery (SELECT MAX(date)) to find the most recent date in the table, then pulls the rate for each currency on that date.
The second query groups all rows by currency and calculates the average, minimum, and maximum rates. These summary statistics show how each currency performed over the time period.
pd.read_sql() runs the SQL and returns the results as a pandas DataFrame, making it easy to print neatly with to_string(index=False).

Save pipeline.py.
You will not see a change yet. The next substep updates the test block to call both new functions.

Update the test block and run the pipeline

Your two new functions are ready. Now you need to wire them into the test block so the script runs the full extract, transform, load, and query flow.

In pipeline.py, scroll down to the if __name__ == "__main__": block at the bottom of your file.
Find these lines:

if __name__ == "__main__":
    raw = extract_rates()
    df = transform_rates(raw)
    print(df)

Replace print(df) with calls to load_to_database(df) and query_database(), so the block looks like this:

if __name__ == "__main__":
    raw = extract_rates()
    df = transform_rates(raw)
    load_to_database(df)
    query_database()

Save pipeline.py.
Run the script by entering this command in your terminal:

python pipeline.py

You should see output that starts with the extraction and transformation messages from earlier. After those, you should see a new line like Loaded X records into 'rates' table, followed by two formatted tables showing the latest rates and summary statistics.

What should I see?

The "Latest rate for each currency" table shows each currency with its rate and inverse rate for today's date. The "Summary statistics per currency" table shows the average, minimum, and maximum rate for each currency.

Since you are fetching only today's rates (not a date range yet), the average, min, and max values will be the same for each currency. That is expected. You will add historical data in the next step.

🙋‍♀️ Seeing an error when you run the script?

NameError: name 'load_to_database' is not defined means Python cannot find the function. Make sure load_to_database() is defined above the if __name__ block, not inside another function.
OperationalError: no such table: rates means query_database() ran before load_to_database(). Check that load_to_database(df) comes before query_database() in your test block.
IndentationError means one of your lines has inconsistent spacing. Make sure all lines inside the if __name__ block are indented with 4 spaces.

Still stuck? Help me debug my pipeline.py script

✔️ Awesome, I've got everything!

Great. Double check you have saved pipeline.py.

ⓧ I'd like to double check the full code

Imports: requests, pandas, sqlite3, date, timedelta.
The extract_rates() function (from Step 2).
The transform_rates() function (from Step 3).
The load_to_database() function (from this step).
The query_database() function (from this step).
The if __name__ == "__main__": block calling load_to_database(df) and query_database().

Your pipeline.py should contain these sections in order, from top to bottom:

You now have a database sitting on your disk with real financial data inside it. Next up, you will assemble the complete pipeline with historical data spanning the last 30 days, giving those summary statistics some real variation to work with.

Assemble the Complete Pipeline

You now have four standalone functions that extract, transform, load, and query exchange rate data. Right now, your test block calls each one individually with default settings.

In this step, you will tie everything together into a single run_pipeline() function that fetches 30 days of historical data for 5 major currencies and runs the full ETL workflow end-to-end.

In this step, get ready to:

Write a run_pipeline() function that orchestrates the full Extract, Transform, Load, Query workflow.
Swap the test block for a clean entry point that calls run_pipeline().
Run the final pipeline and verify 30 days of exchange rate data flows through to SQL query results.

Write the run_pipeline() function

A production pipeline doesn't rely on someone manually calling each function in the right order. The run_pipeline() function will handle the orchestration, configure the date range and target currencies, and print progress logs at each stage.

In pipeline.py, find the if __name__ == "__main__": block at the bottom of your file.
Add the following run_pipeline() function directly above that block:

def run_pipeline():
    """Run the complete ETL pipeline."""
    # Print a header banner
    print("=" * 50)
    print("  CURRENCY EXCHANGE RATE ETL PIPELINE")
    print("=" * 50)

    # Configure the currencies and date range (last 30 days)
    quotes = ["EUR", "GBP", "JPY", "CAD", "AUD"]
    today = date.today()
    start = today - timedelta(days=30)

    # EXTRACT: fetch rates from the API
    print("\n[EXTRACT] Fetching exchange rates...")
    raw_data = extract_rates(
        base="USD",
        quotes=quotes,
        start_date=start.isoformat(),
        end_date=today.isoformat(),
    )

    # TRANSFORM: clean and reshape the raw JSON
    print("\n[TRANSFORM] Cleaning and reshaping data...")
    df = transform_rates(raw_data)
    print(df.head(10).to_string(index=False))

    # LOAD: save the DataFrame to SQLite
    print("\n[LOAD] Saving to SQLite database...")
    load_to_database(df)

    # QUERY: run SQL queries against the loaded data
    print("\n[QUERY] Verifying loaded data...")
    query_database()

    # Print a completion banner
    print("\n" + "=" * 50)
    print("  PIPELINE COMPLETE!")
    print("=" * 50)

What does this code do?

This function is the single entry point for your entire pipeline. Instead of calling each function manually, it handles the full flow:

quotes defines the 5 target currencies: EUR, GBP, JPY, CAD, and AUD.
date.today() and timedelta(days=30) calculate a 30-day window ending today.
.isoformat() converts dates to strings like 2026-04-01 that the API expects.
df.head(10) prints a preview of the first 10 rows so you can spot-check the transformed data.
The [EXTRACT], [TRANSFORM], [LOAD], [QUERY] labels make it easy to see which stage the pipeline is in while it runs.

Save pipeline.py.

You won't see a change yet because the test block at the bottom still calls the old functions directly. The next substep swaps it out.

Update the main block to call run_pipeline()

The temporary test block at the bottom of your file was great for developing each function step by step. Now that run_pipeline() handles everything, you can replace it with a single call.

In pipeline.py, scroll to the bottom of the file and find this block:

if __name__ == "__main__":
    raw = extract_rates()
    df = transform_rates(raw)
    load_to_database(df)
    query_database()

Replace the four lines inside the block with a single call to run_pipeline(), so it looks like this:

if __name__ == "__main__":
    run_pipeline()

Save pipeline.py.

Run the complete pipeline

This is the moment everything comes together. One command triggers the full Extract, Transform, Load, Query workflow against 30 days of real exchange rate data.

Run the pipeline by executing this command in your terminal:

python pipeline.py

You should see output that flows through several stages. Look for these sections in your terminal:

A CURRENCY EXCHANGE RATE ETL PIPELINE banner at the top.
An [EXTRACT] message confirming rates were fetched.
A [TRANSFORM] section showing approximately 150 records transformed across 5 currencies, followed by a 10-row preview.
A [LOAD] confirmation with the number of records loaded into the rates table.
A [QUERY] section showing the latest rate for each currency and summary statistics (average, min, max).
A PIPELINE COMPLETE! banner at the bottom.

Why ~150 records instead of 150 exactly?

The Frankfurter API returns exchange rates for trading days only. Weekends and holidays have no data, so a 30-day window typically produces around 20-22 trading days.

Multiply those trading days by 5 currencies and you get roughly 100-110 records. The exact number depends on which 30-day window you happen to be in.

🙋‍♀️ Seeing an error when running the pipeline?

ConnectionError or Timeout means the API is unreachable. Check your internet connection and try again in a few seconds.
NameError: name 'run_pipeline' is not defined means the function was placed inside or after the __main__ block. Make sure run_pipeline() is defined above the if __name__ line.
ModuleNotFoundError for pandas or requests means your virtual environment is not activated. Activate it and try again.

Help me debug my pipeline error.

✔️ Awesome, I've got everything!

Great. Double check you have saved pipeline.py.

ⓧ I'd like to double check the full code

Imports: requests, pandas, sqlite3, date, timedelta.
The extract_rates() function (from Step 2).
The transform_rates() function (from Step 3).
The load_to_database() function (from Step 4).
The query_database() function (from Step 4).
The run_pipeline() function (from this step).
The if __name__ == "__main__": block with a single call to run_pipeline().

Your pipeline.py should contain these sections in order, from top to bottom:

Secret mission

Add Trend Analysis to Your Pipeline

Your pipeline extracts, transforms, loads, and queries. But a real data pipeline does more than move data around. It produces insights. In this Secret Mission, you'll add trend analysis that calculates daily percentage changes and identifies the most volatile currency over the past month.

Before you go

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

A pipeline.py script containing your ETL functions.
An exchange_rates.db SQLite database file with exchange rate data.
A Python virtual environment inside your project folder.
The pandas and requests packages installed in your virtual environment.

Keep everything running

No action needed. Choose this if you plan to keep building on your pipeline or want to experiment with different currencies and date ranges.

Your project folder, virtual environment, and database file stay on your Desktop.
You can re-run pipeline.py anytime to refresh the data with the latest exchange rates.

Pause - I'll come back to this later

Since this is a local project with no running processes, there is nothing to pause. Your files will be waiting for you when you come back.

Make sure you have saved pipeline.py before closing VS Code.
When you return, reactivate your virtual environment before running the script.

Delete - I don't want to use this again

Remove all project files from your computer to start fresh.

Delete the entire project folder from your Desktop. This removes pipeline.py, exchange_rates.db, and the virtual environment all at once.
No accounts or cloud resources were created, so there is nothing else to clean up.

Mission accomplished

That's a Wrap!

Nicely done! You just built your first data engineering pipeline from scratch.

You pulled real financial data from a live API, cleaned it, stored it in a database, and queried it with SQL. That is the ETL workflow that data engineers use every single day.

Here's what you learned:

Extracted real exchange rate data from a live REST API using the requests library and transformed nested JSON into a clean pandas DataFrame.
Loaded cleaned data into a SQLite database using to_sql() and ran SQL queries to surface summary statistics like average, minimum, and maximum rates.
Built a reusable run_pipeline() function that automates the entire Extract, Transform, Load workflow end to end.
Secret Mission: Added trend analysis to calculate daily percentage changes and identify the most volatile currency over the past month.

Ready to quiz yourself?

Build a Currency Exchange Rate Pipeline

Introduction

30 Second Summary

What You'll Build

Are there any prerequisites?

Before We Start

Before We Start...

Set Up Your Python Project

In this step, get ready to:

Verify your Python version and create a project folder

macOS

Windows

✔️ I see Python 3.11 or higher

ⓧ I see an older version or an error

macOS

Windows

What is a virtual environment?

macOS

Windows

Don't see (venv) in your prompt?

Install the pandas and requests packages

Seeing an install error?

Create pipeline.py and add the import statements

What do these imports do?

macOS

Windows

Seeing an error when you run the script?

✔️ Awesome, I've got everything!

ⓧ I'd like to double check the full code

Extract Data from the Frankfurter API

In this step, get ready to:

Write the extract_rates() function

Why Frankfurter over other APIs?

What does this code do?

Add a test block to call extract_rates()

What does this code do?

Run the script and check the API response

What should I see?

✔️ Awesome, I've got everything!

ⓧ I'd like to double check the full code

Transform the Raw Data with Pandas

In this step, get ready to:

Write the transform_rates() function

What does this code do?

Seeing a red underline on pd.DataFrame?

Update the test block to use transform_rates()

✔️ Awesome, I've got everything!

ⓧ I'd like to double check the full code

Run the script and verify the output

What should I see?

Load into SQLite and Query with SQL

In this step, get ready to:

Write the load function

Why SQLite?

What does this code do?

Write the query function

What do these queries do?

Update the test block and run the pipeline

What should I see?

✔️ Awesome, I've got everything!

ⓧ I'd like to double check the full code

Assemble the Complete Pipeline

In this step, get ready to:

Write the run_pipeline() function

What does this code do?

Update the __main__ block to call run_pipeline()

Run the complete pipeline

Why ~150 records instead of 150 exactly?

✔️ Awesome, I've got everything!

ⓧ I'd like to double check the full code

Secret mission

Add Trend Analysis to Your Pipeline

Before you go

Clean Up Your Resources

Keep everything running

Pause - I'll come back to this later

Delete - I don't want to use this again

Mission accomplished

That's a Wrap!

Update the main block to call run_pipeline()