FX Rate Anomaly Detection Pipeline

Introduction

30 Second Summary

Currency exchange rates shift every day, and most of the time those movements are normal. But occasionally, something unusual happens that signals a market shock, a policy change, or an error in the data.

In this project, you will build a Python pipeline that fetches real foreign exchange rate data, applies two different anomaly detection algorithms, and generates a visual business insights report.

What You'll Build

A modular Python data pipeline that ingests historical FX rates, engineers time-series features, runs statistical and ML-based anomaly detection, and produces an HTML report with charts.

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

• A data ingestion module that pulls 2 years of multi-currency FX rates from a free public API into structured DataFrames.
• Two anomaly detection models (rolling z-score and Isolation Forest) with scored outputs and comparison metrics.
• A generated HTML business insights report with visualizations showing detected anomalies, volatility trends, and method agreement.
• Secret Mission: Add cross-currency correlation analysis that detects when the relationship between currency pairs breaks down.

Are there any prerequisites?

You should be comfortable with basic Python (functions, imports, working with libraries). Familiarity with pandas and numpy will help, but the project walks you through every step.

Before We Start

Before We Start...

Set Up the Project Structure

Your pipeline will have four separate modules, each handling a distinct responsibility: ingesting data, engineering features, detecting anomalies, and generating reports.

In this step, you'll create the project directory, set up a virtual environment to isolate your dependencies, and install all the Python packages the pipeline needs.

In this step, get ready to:

• Create the project directory and virtual environment.
• Install required Python packages.
• Set up the modular folder structure with placeholder files.

Create the project directory and virtual environment

A virtual environment keeps your project's packages isolated from your system Python.

This prevents version conflicts between projects and ensures your pipeline uses exactly the versions you specify.

• Open your terminal in VS Code by pressing Ctrl+` (macOS and Windows).
• Move to your Desktop by running this command:

cd ~/Desktop

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

mkdir fx-anomaly-pipeline
cd fx-anomaly-pipeline

• Confirm you're in the new folder by running:

ls

The output should be empty since the folder is brand new.

• Create and activate your virtual environment by running these commands:

macOS

python3 -m venv venv
source venv/bin/activate

Windows

python -m venv venv
venv\Scripts\activate

You should see (venv) appear at the beginning of your terminal prompt.

What does (venv) mean?

The (venv) prefix means your virtual environment is active.

Any packages you install with pip while this is active go into the venv/ folder, not your global Python installation.

Don't see (venv) in your prompt?

• Make sure you ran both commands (create then activate). The first command creates the environment. The second activates it.
• On Windows, if you get a permissions error, run Set-ExecutionPolicy -ExecutionPolicy RemoteSigned -Scope CurrentUser in PowerShell first, then try activating again.
• On macOS, if you see "python3 not found", confirm Python is installed by running python3 --version.

Help me troubleshoot my virtual environment activation

Install the required packages

The pipeline depends on six Python packages. Pinning exact versions in a requirements.txt file ensures anyone who runs your project gets the same behavior.

• Create a file called requirements.txt in the root of your fx-anomaly-pipeline folder and paste in the following content:

pandas==2.2.3
numpy==2.1.3
scikit-learn==1.8.0
statsmodels==0.14.6
matplotlib==3.9.3
requests==2.32.3

What do these packages do?

• pandas and numpy handle data manipulation and numerical computation.
• scikit-learn provides the Isolation Forest algorithm for ML-based anomaly detection.
• statsmodels supports statistical time-series analysis.
• matplotlib creates the charts for your HTML report.
• requests fetches live FX rate data from the Frankfurter API.

• Save requirements.txt.
• Install all the packages by running this command in your terminal:

pip install -r requirements.txt

You should see pip downloading and installing each package. The final lines will confirm successful installation of all six packages and their dependencies.

Seeing install errors?

• Confirm your virtual environment is active. You should see (venv) in your prompt. If not, re-run the activate command from the previous substep.
• If you see a "No matching distribution" error, double-check that each line in requirements.txt matches the versions above exactly. A single typo can break the install.
• If you see a permissions error, make sure you are inside the virtual environment and not trying to install globally.

Help me fix my pip install errors

✔️ Awesome, I've got everything!

Great. Double check you've saved your file.

ⓧ I'd like to double check the full code

Here is the complete requirements.txt file:

pandas==2.2.3
numpy==2.1.3
scikit-learn==1.8.0
statsmodels==0.14.6
matplotlib==3.9.3
requests==2.32.3

Set up the modular folder structure

Production data pipelines separate concerns into distinct modules. Your pipeline will follow this pattern with a src/ package containing one file per responsibility.

• Create the src/ package directory and all placeholder files by running these commands:

macOS

mkdir src
touch src/__init__.py
touch src/ingest.py
touch src/features.py
touch src/detect.py
touch src/report.py
touch run_pipeline.py

Windows

mkdir src
New-Item src\__init__.py
New-Item src\ingest.py
New-Item src\features.py
New-Item src\detect.py
New-Item src\report.py
New-Item run_pipeline.py

What is __init__.py for?

The __init__.py file tells Python that the src/ folder is an importable package.

Without it, Python would not be able to run from src.ingest import ... in your pipeline runner. The file can stay empty.

• Confirm all the files exist by running:

ls src/
ls run_pipeline.py

You should see __init__.py, ingest.py, features.py, detect.py, and report.py inside src/, plus run_pipeline.py in the project root.

Missing a file?

• Check you ran the commands from inside the fx-anomaly-pipeline directory. Run pwd to confirm your current location.
• Make sure __init__.py has two underscores on each side. A single underscore will not work.

Help me fix my project folder structure

• Open the fx-anomaly-pipeline folder in VS Code by clicking File in the top menu bar.
• Click Open Folder.
• Navigate to your Desktop, select the fx-anomaly-pipeline folder, and click Open.

In the VS Code Explorer sidebar, you should see your full project tree: requirements.txt, run_pipeline.py, and the src/ folder with all five module files.

Your project skeleton is ready. Next up, you'll write the data ingestion module that pulls live FX rate data from the Frankfurter API.

Build the Data Ingestion Module

Your project structure is ready. Now you need data to detect anomalies in.

In this step, you will write the data ingestion module that pulls two years of real foreign exchange rate data from the Frankfurter API. This free API provides historical rates published by the European Central Bank, covering major currency pairs without requiring an API key.

In this step, get ready to:

• Write the fetch function to pull EUR/USD, EUR/GBP, EUR/JPY, and EUR/CHF rates from the Frankfurter API.
• Parse the JSON response into a pandas DataFrame with a DatetimeIndex.
• Add data validation that handles missing dates from weekends and holidays.

Set up the module imports and API configuration

Every module starts with its imports and configuration constants. The ingestion module needs requests to call the API, pandas to structure the data, and Python's built-in datetime to compute date ranges.

• Open src/ingest.py in your editor.
• Replace the empty file contents with the following code:

"""FX rate data ingestion from the Frankfurter API."""

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


API_BASE = "https://api.frankfurter.dev/v2/rates"
CURRENCY_PAIRS = ["USD", "GBP", "JPY", "CHF"]

What does this code do?

API_BASE stores the Frankfurter API's time-series endpoint URL. All rate requests go through this single endpoint.

CURRENCY_PAIRS lists the target currencies you want rates for. The base currency is always EUR because the Frankfurter API uses European Central Bank data, which naturally references EUR.

You won't see a change yet. The next chunk uses these imports to build the data fetching function.

Write the function to fetch FX rates

The Frankfurter API accepts a date range and returns all daily rates in a single JSON response. Your function needs to compute a 2-year lookback window, make the request, and reshape the response into a clean DataFrame.

• Below the CURRENCY_PAIRS line in src/ingest.py, add the fetch_fx_rates function:

def fetch_fx_rates(years_back: int = 2) -> pd.DataFrame:
    """Fetch historical FX rates from Frankfurter API.

    Returns a DataFrame with DatetimeIndex and columns for each currency pair.
    Base currency is EUR.
    """
    end_date = date.today()
    start_date = end_date - timedelta(days=years_back * 365)

    params = {
        "base": "EUR",
        "quotes": ",".join(CURRENCY_PAIRS),
        "from": start_date.isoformat(),
        "to": end_date.isoformat(),
    }

    print(f"Fetching FX rates from {start_date} to {end_date}...")
    response = requests.get(API_BASE, params=params)
    response.raise_for_status()

    data = response.json()
    rates = data["rates"]

    df = pd.DataFrame.from_dict(rates, orient="index")
    df.index = pd.to_datetime(df.index)
    df.index.name = "date"
    df = df.sort_index()

    # Rename columns for clarity
    df.columns = [f"EUR_{col}" for col in df.columns]

    return df

What does this code do?

• date.today() and timedelta compute a 2-year date range ending today.
• The params dictionary tells the API you want EUR-based rates for USD, GBP, JPY, and CHF across that date range.
• response.raise_for_status() throws an error if the API returns a non-200 status code, so failures surface immediately.
• pd.DataFrame.from_dict(rates, orient="index") converts the JSON dictionary (keyed by date strings) into a DataFrame where each row is a trading day.
• Columns get renamed to EUR_USD, EUR_GBP, etc. to make it clear these are EUR-based exchange rates.

• Save src/ingest.py.
• Test the function by running this command from your project root:

python -c "from src.ingest import fetch_fx_rates; print(fetch_fx_rates().head())"

You should see a line starting with Fetching FX rates from... followed by a table showing the first five rows of exchange rate data with columns like EUR_USD and EUR_GBP.

Seeing a connection error?

• Make sure you have an active internet connection. The Frankfurter API is a live web service.
• Check that API_BASE is exactly https://api.frankfurter.dev/v2/rates with no trailing slash.
• If you see ModuleNotFoundError: No module named 'src', make sure you are running the command from the project root folder (the one containing the src/ directory).

Help me debug my Frankfurter API connection error

Why the Frankfurter API over other FX data sources?

The Frankfurter API is completely free with no rate limits, requires no API key or signup, and provides 20+ years of ECB-published rates. Other options like ExchangeRate-API cap you at 1,500 requests per month. Open Exchange Rates requires a key and limits you to 1,000 monthly requests on the free tier.

For a portfolio project where you want reproducible results without credential management, Frankfurter is the clear choice.

Add data validation for missing dates

The API only returns data for trading days. Weekends and holidays create gaps in the date index. Your anomaly detection models need a continuous time series, so you need to fill those gaps.

• Below the fetch_fx_rates function in src/ingest.py, add the validate_data function:

def validate_data(df: pd.DataFrame) -> pd.DataFrame:
    """Validate and clean the ingested data.

    Checks for missing values and fills gaps from weekends/holidays
    using forward-fill.
    """
    print(f"Raw data: {len(df)} trading days, {df.isnull().sum().sum()} missing values")

    # Reindex to full business day range to identify gaps
    full_range = pd.bdate_range(start=df.index.min(), end=df.index.max())
    df = df.reindex(full_range)
    df.index.name = "date"

    # Forward-fill missing values (weekends/holidays)
    missing_before = df.isnull().sum().sum()
    df = df.ffill()
    missing_after = df.isnull().sum().sum()

    print(f"Filled {missing_before - missing_after} missing values via forward-fill")
    print(f"Final dataset: {len(df)} rows, {len(df.columns)} currency pairs")

    return df

What does this code do?

• pd.bdate_range() generates a complete sequence of business days between the first and last date in your data. This reveals any days the API skipped.
• df.reindex(full_range) expands the DataFrame to include every business day, filling new rows with NaN where data was missing.
• df.ffill() carries the last known rate forward into gaps. This is standard practice for financial data because the previous day's closing rate is the best estimate when markets are closed.

• Save src/ingest.py.
• Test both functions together by running this command:

python -c "from src.ingest import fetch_fx_rates, validate_data; df = validate_data(fetch_fx_rates()); print(df.shape)"

You should see output showing the raw data count, how many values were forward-filled, and the final dataset shape. The shape should be approximately (520, 4) representing around 520 business days across 4 currency pairs.

Seeing a KeyError or AttributeError?

• A KeyError: 'rates' means the API response format changed or the request failed silently. Print data.keys() to see what the response actually contains.
• Make sure validate_data is defined below fetch_fx_rates in the same file. Python reads files top to bottom.
• Check that you used quotes (not symbols) in the params dictionary. The Frankfurter API v2 uses quotes while v1 used symbols.

Help me debug my data validation function

✔️ Awesome, I've got everything!

Great. Double check you have saved src/ingest.py before moving on.

ⓧ I'd like to double check the full code

Here is the complete src/ingest.py file. Compare your code against this:

"""FX rate data ingestion from the Frankfurter API."""

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


API_BASE = "https://api.frankfurter.dev/v2/rates"
CURRENCY_PAIRS = ["USD", "GBP", "JPY", "CHF"]


def fetch_fx_rates(years_back: int = 2) -> pd.DataFrame:
    """Fetch historical FX rates from Frankfurter API.

    Returns a DataFrame with DatetimeIndex and columns for each currency pair.
    Base currency is EUR.
    """
    end_date = date.today()
    start_date = end_date - timedelta(days=years_back * 365)

    params = {
        "base": "EUR",
        "quotes": ",".join(CURRENCY_PAIRS),
        "from": start_date.isoformat(),
        "to": end_date.isoformat(),
    }

    print(f"Fetching FX rates from {start_date} to {end_date}...")
    response = requests.get(API_BASE, params=params)
    response.raise_for_status()

    data = response.json()
    rates = data["rates"]

    df = pd.DataFrame.from_dict(rates, orient="index")
    df.index = pd.to_datetime(df.index)
    df.index.name = "date"
    df = df.sort_index()

    # Rename columns for clarity
    df.columns = [f"EUR_{col}" for col in df.columns]

    return df


def validate_data(df: pd.DataFrame) -> pd.DataFrame:
    """Validate and clean the ingested data.

    Checks for missing values and fills gaps from weekends/holidays
    using forward-fill.
    """
    print(f"Raw data: {len(df)} trading days, {df.isnull().sum().sum()} missing values")

    # Reindex to full business day range to identify gaps
    full_range = pd.bdate_range(start=df.index.min(), end=df.index.max())
    df = df.reindex(full_range)
    df.index.name = "date"

    # Forward-fill missing values (weekends/holidays)
    missing_before = df.isnull().sum().sum()
    df = df.ffill()
    missing_after = df.isnull().sum().sum()

    print(f"Filled {missing_before - missing_after} missing values via forward-fill")
    print(f"Final dataset: {len(df)} rows, {len(df.columns)} currency pairs")

    return df

Your ingestion module is pulling real financial data from the web and cleaning it into a consistent time series. Next up, you will engineer features from this raw data that your anomaly detection models can use.

Engineer Time-Series Features

Your ingestion module pulls raw exchange rates from the API. Raw rates alone are not enough for anomaly detection though.

Anomaly detection algorithms need derived signals like how fast a rate is moving, how volatile it has been recently, and how today compares to the recent average. In this step, you will build a feature engineering module that transforms raw rates into a rich feature matrix ready for both statistical and ML-based detection.

In this step, get ready to:

• Compute rolling statistics (7-day and 30-day mean and standard deviation) for each currency pair.
• Calculate daily returns, annualized volatility, lag features, and day-of-week indicators.
• Normalize features with RobustScaler and prepare separate inputs for z-score and Isolation Forest detection.

Create the features module with rolling statistics

Rolling statistics capture how a rate behaves relative to its recent history. A 7-day rolling mean shows short-term trend, while a 30-day rolling mean captures the medium-term baseline.

• Create a new file called src/features.py inside your src/ folder.
• Add the imports and the rolling statistics function by pasting this code:

"""Time-series feature engineering for FX rate data."""

import pandas as pd
import numpy as np
from sklearn.preprocessing import RobustScaler


def compute_rolling_stats(df: pd.DataFrame) -> pd.DataFrame:
    """Compute rolling mean and standard deviation for each currency pair."""
    features = df.copy()

    for col in df.columns:
        features[f"{col}_roll7_mean"] = df[col].rolling(window=7, min_periods=7).mean()
        features[f"{col}_roll30_mean"] = df[col].rolling(window=30, min_periods=30).mean()
        features[f"{col}_roll7_std"] = df[col].rolling(window=7, min_periods=7).std()
        features[f"{col}_roll30_std"] = df[col].rolling(window=30, min_periods=30).std()

    return features

What does this code do?

The function loops through each currency pair column and computes four new columns per pair:

• A 7-day and 30-day rolling mean, which smooth out daily noise to reveal trends.
• A 7-day and 30-day rolling standard deviation, which measure how spread out recent values are.
• The min_periods parameter ensures the window must be completely full before computing a value. This avoids unreliable statistics from partial windows at the start of the data.

• Save src/features.py.

You will not see visible output yet. The next chunks add more features before you run anything.

Add returns, volatility, and lag features

Daily returns show how much a rate moved as a percentage. Annualized volatility tells you how "jumpy" a currency has been over the past month. Lag features give the model memory of recent movements.

• In src/features.py, add the following function below compute_rolling_stats:

def compute_returns_and_volatility(df: pd.DataFrame) -> pd.DataFrame:
    """Compute daily returns and rolling volatility."""
    features = df.copy()

    for col in [c for c in df.columns if not c.startswith("EUR_")]:
        continue

    for col in [c for c in df.columns if c.startswith("EUR_")]:
        # Daily percentage returns
        features[f"{col}_return"] = df[col].pct_change()

        # Rolling 20-day annualized volatility
        daily_returns = df[col].pct_change()
        features[f"{col}_volatility_20d"] = (
            daily_returns.rolling(window=20, min_periods=20).std() * np.sqrt(252)
        )

        # Lag features
        features[f"{col}_return_lag1"] = features[f"{col}_return"].shift(1)
        features[f"{col}_return_lag5"] = features[f"{col}_return"].shift(5)

    # Day of week indicator
    features["day_of_week"] = df.index.dayofweek

    return features

What does this code do?

• pct_change() computes the fractional change between consecutive rows, giving you the daily return for each pair.
• The annualized volatility multiplies the 20-day rolling standard deviation of returns by sqrt(252). There are roughly 252 trading days in a year, so this scales daily volatility to an annual figure.
• Lag features (_return_lag1 and _return_lag5) shift the return column by 1 and 5 days. This gives the Isolation Forest model information about recent momentum.
• The day_of_week column encodes Monday as 0 through Friday as 4. Some currency pairs exhibit day-of-week effects in volatility.

• Save src/features.py.

You will not see output yet. The next function prepares these features for the detection models.

Normalize features and build the orchestrator

The Isolation Forest model works best when features are on a comparable scale. Standard normalization (mean and standard deviation) is sensitive to the very outliers you are trying to detect.

Why RobustScaler instead of StandardScaler?

RobustScaler uses the median and interquartile range (IQR) instead of mean and standard deviation. This makes it resistant to outliers in the training data. Since the whole point of this pipeline is to find outliers, using a scaler that is not distorted by them gives cleaner results.

• In src/features.py, add the prepare_detection_features function below compute_returns_and_volatility:

def prepare_detection_features(df: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame]:
    """Prepare feature matrix for anomaly detection models.

    Returns:
        features_scaled: Scaled feature matrix for Isolation Forest
        z_score_data: DataFrame with columns needed for z-score detection
    """
    # Features for Isolation Forest
    feature_cols = [c for c in df.columns if any(
        x in c for x in ["_return", "_volatility", "_roll7_std", "_roll30_std", "_lag"]
    )]
    feature_cols.append("day_of_week")

    # Drop rows with NaN from rolling windows
    features_raw = df[feature_cols].dropna()

    # Scale with RobustScaler (resistant to outliers)
    scaler = RobustScaler()
    features_scaled = pd.DataFrame(
        scaler.fit_transform(features_raw),
        index=features_raw.index,
        columns=features_raw.columns,
    )

    # Z-score data: raw rates and rolling stats
    z_score_cols = [c for c in df.columns if c.startswith("EUR_") and "_" not in c[4:]]
    z_score_roll_cols = [c for c in df.columns if "_roll30_mean" in c or "_roll30_std" in c]
    z_score_data = df[z_score_cols + z_score_roll_cols].dropna()

    return features_scaled, z_score_data

What does this code do?

This function prepares two separate datasets, one for each detection method:

• features_scaled selects return, volatility, standard deviation, lag, and day-of-week columns, then scales them with RobustScaler. This feeds into the Isolation Forest model.
• z_score_data keeps the raw EUR rates alongside their 30-day rolling mean and standard deviation. The z-score method only needs these to compute how far each day deviates from the rolling baseline.
• Both datasets call .dropna() to remove the initial rows where rolling windows have not yet filled.

• Now add the orchestrator function that chains all three feature steps together. Paste this below prepare_detection_features:

def build_features(df: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
    """Full feature engineering pipeline.

    Returns:
        full_features: Complete DataFrame with all features
        features_scaled: Scaled features for Isolation Forest
        z_score_data: Data for z-score detection
    """
    print("Engineering features...")
    features = compute_rolling_stats(df)
    features = compute_returns_and_volatility(features)
    features_scaled, z_score_data = prepare_detection_features(features)

    print(f"Feature matrix: {features_scaled.shape[0]} rows, {features_scaled.shape[1]} features")

    return features, features_scaled, z_score_data

What does this code do?

build_features is the single entry point other modules will call. It runs rolling stats, then returns and volatility, then prepares the two detection-ready DataFrames. It also prints progress so you can see the pipeline running.

• Save src/features.py.
• Verify the module loads cleanly by running this in your terminal:

python -c "from src.features import build_features; print('features module OK')"

You should see features module OK printed with no errors.

Seeing an ImportError?

• Make sure you are running the command from your project root directory (the folder that contains the src/ folder), not from inside src/ itself.
• Confirm src/__init__.py exists. Python needs this file to treat src/ as a package.
• Check that your virtual environment is activated. You should see its name in your terminal prompt.

Why can't Python find my src.features module?

✔️ Awesome, I've got everything!

Great. Double check you have saved src/features.py and the import test printed successfully.

ⓧ I'd like to double check the full code

Here is the complete src/features.py file. Compare yours against it:

"""Time-series feature engineering for FX rate data."""

import pandas as pd
import numpy as np
from sklearn.preprocessing import RobustScaler


def compute_rolling_stats(df: pd.DataFrame) -> pd.DataFrame:
    """Compute rolling mean and standard deviation for each currency pair."""
    features = df.copy()

    for col in df.columns:
        features[f"{col}_roll7_mean"] = df[col].rolling(window=7, min_periods=7).mean()
        features[f"{col}_roll30_mean"] = df[col].rolling(window=30, min_periods=30).mean()
        features[f"{col}_roll7_std"] = df[col].rolling(window=7, min_periods=7).std()
        features[f"{col}_roll30_std"] = df[col].rolling(window=30, min_periods=30).std()

    return features


def compute_returns_and_volatility(df: pd.DataFrame) -> pd.DataFrame:
    """Compute daily returns and rolling volatility."""
    features = df.copy()

    for col in [c for c in df.columns if not c.startswith("EUR_")]:
        continue

    for col in [c for c in df.columns if c.startswith("EUR_")]:
        # Daily percentage returns
        features[f"{col}_return"] = df[col].pct_change()

        # Rolling 20-day annualized volatility
        daily_returns = df[col].pct_change()
        features[f"{col}_volatility_20d"] = (
            daily_returns.rolling(window=20, min_periods=20).std() * np.sqrt(252)
        )

        # Lag features
        features[f"{col}_return_lag1"] = features[f"{col}_return"].shift(1)
        features[f"{col}_return_lag5"] = features[f"{col}_return"].shift(5)

    # Day of week indicator
    features["day_of_week"] = df.index.dayofweek

    return features


def prepare_detection_features(df: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame]:
    """Prepare feature matrix for anomaly detection models.

    Returns:
        features_scaled: Scaled feature matrix for Isolation Forest
        z_score_data: DataFrame with columns needed for z-score detection
    """
    # Features for Isolation Forest
    feature_cols = [c for c in df.columns if any(
        x in c for x in ["_return", "_volatility", "_roll7_std", "_roll30_std", "_lag"]
    )]
    feature_cols.append("day_of_week")

    # Drop rows with NaN from rolling windows
    features_raw = df[feature_cols].dropna()

    # Scale with RobustScaler (resistant to outliers)
    scaler = RobustScaler()
    features_scaled = pd.DataFrame(
        scaler.fit_transform(features_raw),
        index=features_raw.index,
        columns=features_raw.columns,
    )

    # Z-score data: raw rates and rolling stats
    z_score_cols = [c for c in df.columns if c.startswith("EUR_") and "_" not in c[4:]]
    z_score_roll_cols = [c for c in df.columns if "_roll30_mean" in c or "_roll30_std" in c]
    z_score_data = df[z_score_cols + z_score_roll_cols].dropna()

    return features_scaled, z_score_data


def build_features(df: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
    """Full feature engineering pipeline.

    Returns:
        full_features: Complete DataFrame with all features
        features_scaled: Scaled features for Isolation Forest
        z_score_data: Data for z-score detection
    """
    print("Engineering features...")
    features = compute_rolling_stats(df)
    features = compute_returns_and_volatility(features)
    features_scaled, z_score_data = prepare_detection_features(features)

    print(f"Feature matrix: {features_scaled.shape[0]} rows, {features_scaled.shape[1]} features")

    return features, features_scaled, z_score_data

Your feature engineering module is complete. Next up, you will build the anomaly detection module that consumes these features and flags unusual rate movements.

Implement Anomaly Detection

Your features module now produces two outputs: a scaled feature matrix for machine learning and a z-score dataset with rolling statistics.

In this step, you will write the detection module that consumes those outputs and flags unusual rate movements. You will implement two complementary approaches and compare their results.

In this step, get ready to:

• Write a rolling z-score anomaly detector that flags rates beyond 3 standard deviations from their 30-day mean.
• Write an Isolation Forest detector that learns multivariate patterns from the scaled features.
• Compare both methods and compute agreement metrics.

Write the z-score detection method

A rolling z-score measures how far today's rate sits from its recent average, expressed in standard deviations. Any day where the rate deviates beyond 3 sigma is flagged as anomalous.

• Create a new file called src/detect.py inside your src/ folder.
• Add the module docstring, imports, and the z-score detection function by pasting this code:

"""Anomaly detection methods for FX rate data."""

import pandas as pd
import numpy as np
from sklearn.ensemble import IsolationForest


def detect_zscore_anomalies(z_score_data: pd.DataFrame, threshold: float = 3.0) -> pd.DataFrame:
    """Detect anomalies using rolling z-score method.

    Flags data points where the rate deviates more than `threshold`
    standard deviations from the 30-day rolling mean.
    """
    results = pd.DataFrame(index=z_score_data.index)

    rate_cols = [c for c in z_score_data.columns if c.startswith("EUR_") and "_roll" not in c]

    for col in rate_cols:
        mean_col = f"{col}_roll30_mean"
        std_col = f"{col}_roll30_std"

        if mean_col in z_score_data.columns and std_col in z_score_data.columns:
            z_scores = (z_score_data[col] - z_score_data[mean_col]) / z_score_data[std_col]
            results[f"{col}_zscore"] = z_scores
            results[f"{col}_zscore_anomaly"] = (z_scores.abs() > threshold).astype(int)

    # Aggregate: flag as anomaly if ANY pair is anomalous
    anomaly_cols = [c for c in results.columns if c.endswith("_zscore_anomaly")]
    results["zscore_anomaly"] = results[anomaly_cols].max(axis=1)
    results["zscore_max_score"] = results[
        [c for c in results.columns if c.endswith("_zscore") and not c.endswith("_anomaly")]
    ].abs().max(axis=1)

    return results

What does this code do?

For each currency pair, the function computes a z-score: (current rate - 30-day rolling mean) / 30-day rolling std. Any day where the absolute z-score exceeds 3.0 gets a binary anomaly flag of 1.

The aggregate columns at the bottom combine all four pairs. zscore_anomaly is 1 if any pair is anomalous that day. zscore_max_score stores the highest absolute z-score across all pairs, giving a continuous severity measure.

• Save src/detect.py.

Why a 3-sigma threshold?

In a normal distribution, 99.7% of values fall within 3 standard deviations of the mean. For FX rates, anything outside that band represents a genuinely unusual movement worth investigating.

A lower threshold (like 2 sigma) would catch more events but include routine fluctuations. The 3-sigma default keeps false positives low for relatively stable currency pairs.

Write the Isolation Forest detection method

The z-score method looks at each currency pair individually. Isolation Forest takes a different approach by examining all features simultaneously to find data points that are easy to isolate from the rest.

• In src/detect.py, add the Isolation Forest detection function below the detect_zscore_anomalies function:

def detect_isolation_forest_anomalies(
    features_scaled: pd.DataFrame,
    n_estimators: int = 200,
    contamination: float = 0.01,
) -> pd.DataFrame:
    """Detect anomalies using Isolation Forest.

    Returns DataFrame with anomaly labels and continuous scores.
    """
    model = IsolationForest(
        n_estimators=n_estimators,
        contamination=contamination,
        random_state=42,
    )

    predictions = model.fit_predict(features_scaled)
    scores = model.decision_function(features_scaled)

    results = pd.DataFrame(index=features_scaled.index)
    results["iforest_anomaly"] = (predictions == -1).astype(int)
    results["iforest_score"] = scores

    return results

What does this code do?

IsolationForest builds 200 random trees (n_estimators=200) and isolates outliers by measuring how few splits are needed to separate a data point. Points that are easy to isolate are anomalies.

contamination=0.01 tells the model to expect roughly 1% of the data to be anomalous. fit_predict returns -1 for outliers and 1 for inliers. decision_function provides a continuous score where more negative values indicate stronger anomalies.

• Save src/detect.py.

Why two methods instead of one?

Z-scores are interpretable and work well for single-variable outliers. Isolation Forest captures complex multivariate patterns that a univariate method misses.

When both methods agree on a flag, you can be more confident the anomaly is real. This comparison is exactly what hiring managers look for in portfolio projects.

Seeing an import error for IsolationForest?

Make sure your virtual environment is activated and scikit-learn is installed. The import path is from sklearn.ensemble import IsolationForest (not sklearn.tree or sklearn.forest).

Help me fix my IsolationForest import in detect.py

Compare detection results and wire up the orchestrator

Now that both detection methods exist, you need a way to compare their outputs. The comparison function calculates how often the two methods agree and where they diverge.

• In src/detect.py, add the compare_methods function below detect_isolation_forest_anomalies:

def compare_methods(zscore_results: pd.DataFrame, iforest_results: pd.DataFrame) -> dict:
    """Compare anomaly detection results from both methods."""
    # Align indices
    common_idx = zscore_results.index.intersection(iforest_results.index)
    zs = zscore_results.loc[common_idx, "zscore_anomaly"]
    if_res = iforest_results.loc[common_idx, "iforest_anomaly"]

    total_days = len(common_idx)
    zscore_count = zs.sum()
    iforest_count = if_res.sum()
    both_count = ((zs == 1) & (if_res == 1)).sum()
    either_count = ((zs == 1) | (if_res == 1)).sum()

    agreement = (zs == if_res).sum() / total_days * 100

    comparison = {
        "total_days": total_days,
        "zscore_anomalies": int(zscore_count),
        "iforest_anomalies": int(iforest_count),
        "both_methods": int(both_count),
        "either_method": int(either_count),
        "agreement_pct": round(agreement, 1),
    }

    return comparison

What does this code do?

The function first aligns both result sets to the same date range using index.intersection(). This handles cases where the two methods produce results for slightly different date ranges (due to different NaN handling).

It then counts how many days each method flagged, how many both flagged simultaneously, and the overall agreement percentage. Days flagged by both methods are your highest-confidence anomalies.

• Save src/detect.py.
• Now add the run_detection orchestrator function at the bottom of the file:

def run_detection(
    features_scaled: pd.DataFrame, z_score_data: pd.DataFrame
) -> tuple[pd.DataFrame, pd.DataFrame, dict]:
    """Run both detection methods and compare results.

    Returns:
        zscore_results: Z-score anomaly results
        iforest_results: Isolation Forest anomaly results
        comparison: Dict with comparison metrics
    """
    print("Running z-score anomaly detection...")
    zscore_results = detect_zscore_anomalies(z_score_data)
    print(f"  Z-score anomalies found: {zscore_results['zscore_anomaly'].sum()}")

    print("Running Isolation Forest anomaly detection...")
    iforest_results = detect_isolation_forest_anomalies(features_scaled)
    print(f"  Isolation Forest anomalies found: {iforest_results['iforest_anomaly'].sum()}")

    comparison = compare_methods(zscore_results, iforest_results)
    print(f"  Method agreement: {comparison['agreement_pct']}%")

    return zscore_results, iforest_results, comparison

What does this code do?

run_detection is the public entry point for this module. It calls both detection methods in sequence, prints progress with counts, runs the comparison, and returns all three results.

This follows the same orchestrator pattern as build_features in the features module. Each module exposes one top-level function that the pipeline runner will call.

• Save src/detect.py.

Not sure where to place each function?

The functions should appear in this exact order in your file: detect_zscore_anomalies, then detect_isolation_forest_anomalies, then compare_methods, then run_detection. The orchestrator must be last because it calls the other three.

Help me check the structure of my detect.py file

✔️ Awesome, I've got everything!

Great. Double check you have saved src/detect.py with all four functions.

ⓧ I'd like to double check the full code

Here is the complete src/detect.py file for comparison:

"""Anomaly detection methods for FX rate data."""

import pandas as pd
import numpy as np
from sklearn.ensemble import IsolationForest


def detect_zscore_anomalies(z_score_data: pd.DataFrame, threshold: float = 3.0) -> pd.DataFrame:
    """Detect anomalies using rolling z-score method.

    Flags data points where the rate deviates more than `threshold`
    standard deviations from the 30-day rolling mean.
    """
    results = pd.DataFrame(index=z_score_data.index)

    rate_cols = [c for c in z_score_data.columns if c.startswith("EUR_") and "_roll" not in c]

    for col in rate_cols:
        mean_col = f"{col}_roll30_mean"
        std_col = f"{col}_roll30_std"

        if mean_col in z_score_data.columns and std_col in z_score_data.columns:
            z_scores = (z_score_data[col] - z_score_data[mean_col]) / z_score_data[std_col]
            results[f"{col}_zscore"] = z_scores
            results[f"{col}_zscore_anomaly"] = (z_scores.abs() > threshold).astype(int)

    # Aggregate: flag as anomaly if ANY pair is anomalous
    anomaly_cols = [c for c in results.columns if c.endswith("_zscore_anomaly")]
    results["zscore_anomaly"] = results[anomaly_cols].max(axis=1)
    results["zscore_max_score"] = results[
        [c for c in results.columns if c.endswith("_zscore") and not c.endswith("_anomaly")]
    ].abs().max(axis=1)

    return results


def detect_isolation_forest_anomalies(
    features_scaled: pd.DataFrame,
    n_estimators: int = 200,
    contamination: float = 0.01,
) -> pd.DataFrame:
    """Detect anomalies using Isolation Forest.

    Returns DataFrame with anomaly labels and continuous scores.
    """
    model = IsolationForest(
        n_estimators=n_estimators,
        contamination=contamination,
        random_state=42,
    )

    predictions = model.fit_predict(features_scaled)
    scores = model.decision_function(features_scaled)

    results = pd.DataFrame(index=features_scaled.index)
    results["iforest_anomaly"] = (predictions == -1).astype(int)
    results["iforest_score"] = scores

    return results


def compare_methods(zscore_results: pd.DataFrame, iforest_results: pd.DataFrame) -> dict:
    """Compare anomaly detection results from both methods."""
    # Align indices
    common_idx = zscore_results.index.intersection(iforest_results.index)
    zs = zscore_results.loc[common_idx, "zscore_anomaly"]
    if_res = iforest_results.loc[common_idx, "iforest_anomaly"]

    total_days = len(common_idx)
    zscore_count = zs.sum()
    iforest_count = if_res.sum()
    both_count = ((zs == 1) & (if_res == 1)).sum()
    either_count = ((zs == 1) | (if_res == 1)).sum()

    agreement = (zs == if_res).sum() / total_days * 100

    comparison = {
        "total_days": total_days,
        "zscore_anomalies": int(zscore_count),
        "iforest_anomalies": int(iforest_count),
        "both_methods": int(both_count),
        "either_method": int(either_count),
        "agreement_pct": round(agreement, 1),
    }

    return comparison


def run_detection(
    features_scaled: pd.DataFrame, z_score_data: pd.DataFrame
) -> tuple[pd.DataFrame, pd.DataFrame, dict]:
    """Run both detection methods and compare results.

    Returns:
        zscore_results: Z-score anomaly results
        iforest_results: Isolation Forest anomaly results
        comparison: Dict with comparison metrics
    """
    print("Running z-score anomaly detection...")
    zscore_results = detect_zscore_anomalies(z_score_data)
    print(f"  Z-score anomalies found: {zscore_results['zscore_anomaly'].sum()}")

    print("Running Isolation Forest anomaly detection...")
    iforest_results = detect_isolation_forest_anomalies(features_scaled)
    print(f"  Isolation Forest anomalies found: {iforest_results['iforest_anomaly'].sum()}")

    comparison = compare_methods(zscore_results, iforest_results)
    print(f"  Method agreement: {comparison['agreement_pct']}%")

    return zscore_results, iforest_results, comparison

Your detection module is ready to flag anomalies from two different angles. Next up, you will build the report module that turns these raw detections into visual business insights.

Generate the Business Insights Report

Your pipeline can now detect anomalies using two different methods. The detection results are stored as DataFrames with scores and binary labels, but raw numbers alone are hard to act on.

In this step, you'll build the reporting module that transforms those detection results into a professional HTML report with matplotlib charts and a key findings summary. This is what turns a data pipeline into a business deliverable.

In this step, get ready to:

• Create three matplotlib visualizations: rate time series with anomalies, rolling volatility, and a method comparison bar chart.
• Build an HTML report generator that embeds charts and summary metrics.
• Wire up an orchestrator function that produces the full report to an output directory.

Create the report module and rate time series chart

The first visualization plots each currency pair's exchange rate over time, with anomaly points highlighted in different colors per detection method. This gives a quick visual sense of where the models disagree.

• Create a new file called src/report.py in your src/ folder.
• Add the imports, output directory constant, and the first chart function by pasting this code:

"""Business insights report generation."""

import pandas as pd
import numpy as np
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
from pathlib import Path


OUTPUT_DIR = Path("output")


def plot_rate_with_anomalies(
    rates_df: pd.DataFrame,
    zscore_results: pd.DataFrame,
    iforest_results: pd.DataFrame,
    output_path: Path,
) -> str:
    """Plot FX rates with anomaly points highlighted."""
    fig, axes = plt.subplots(2, 2, figsize=(14, 10))
    axes = axes.flatten()

    rate_cols = [c for c in rates_df.columns if c.startswith("EUR_") and "_" not in c[4:]]
    common_idx = rates_df.index.intersection(zscore_results.index).intersection(iforest_results.index)

    for i, col in enumerate(rate_cols[:4]):
        ax = axes[i]
        ax.plot(rates_df.loc[common_idx].index, rates_df.loc[common_idx, col], linewidth=0.8, color="steelblue", label="Rate")

        # Z-score anomalies
        zs_anom = zscore_results.loc[common_idx]
        zs_mask = zs_anom["zscore_anomaly"] == 1
        if zs_mask.any():
            ax.scatter(
                rates_df.loc[common_idx][zs_mask].index,
                rates_df.loc[common_idx][zs_mask][col],
                color="red", s=20, zorder=5, label="Z-score", alpha=0.7
            )

        # Isolation Forest anomalies
        if_anom = iforest_results.loc[common_idx]
        if_mask = if_anom["iforest_anomaly"] == 1
        if if_mask.any():
            ax.scatter(
                rates_df.loc[common_idx][if_mask].index,
                rates_df.loc[common_idx][if_mask][col],
                color="orange", s=20, zorder=5, marker="^", label="IForest", alpha=0.7
            )

        ax.set_title(col.replace("_", "/"), fontsize=11)
        ax.xaxis.set_major_formatter(mdates.DateFormatter("%b %Y"))
        ax.tick_params(axis="x", rotation=45)
        ax.legend(fontsize=8)

    plt.tight_layout()
    chart_path = output_path / "rates_anomalies.png"
    plt.savefig(chart_path, dpi=120, bbox_inches="tight")
    plt.close()
    return str(chart_path)

What does this code do?

The imports set the matplotlib backend to "Agg" (non-interactive) so the pipeline can generate PNGs without needing a display window. This is standard practice for scripts that run headless.

• OUTPUT_DIR defines where all generated charts and the final HTML report will be saved.
• plot_rate_with_anomalies creates a 2x2 grid of subplots, one per currency pair. Red dots mark z-score anomalies. Orange triangles mark Isolation Forest anomalies.
• The function intersects indices from all three DataFrames to ensure only dates present in all results are plotted.

• Save src/report.py.
• You won't see a visual change yet. The next chunks will add the remaining chart functions before we run the full pipeline in Step 6.

Add the volatility and comparison charts

The second chart plots rolling volatility over time so you can spot which periods had the most turbulent markets. The third chart is a bar chart comparing how many anomalies each method found uniquely versus jointly.

• In src/report.py, add the volatility chart function below plot_rate_with_anomalies:

def plot_volatility(full_features: pd.DataFrame, output_path: Path) -> str:
    """Plot rolling volatility for each currency pair."""
    fig, ax = plt.subplots(figsize=(12, 5))

    vol_cols = [c for c in full_features.columns if "_volatility_20d" in c]
    for col in vol_cols:
        label = col.replace("_volatility_20d", "").replace("_", "/")
        ax.plot(full_features.index, full_features[col], linewidth=0.9, label=label)

    ax.set_title("20-Day Annualized Volatility", fontsize=13)
    ax.set_ylabel("Volatility")
    ax.legend()
    ax.xaxis.set_major_formatter(mdates.DateFormatter("%b %Y"))
    ax.tick_params(axis="x", rotation=45)

    plt.tight_layout()
    chart_path = output_path / "volatility.png"
    plt.savefig(chart_path, dpi=120, bbox_inches="tight")
    plt.close()
    return str(chart_path)

What does this code do?

This function filters the full features DataFrame for columns ending in _volatility_20d and plots each currency pair's 20-day annualized volatility on a single shared axis. Overlaying all four currencies makes it easy to compare which pair was most volatile at any given time.

• Now add the comparison bar chart function directly below plot_volatility:

def plot_comparison(comparison: dict, output_path: Path) -> str:
    """Plot method comparison bar chart."""
    fig, ax = plt.subplots(figsize=(8, 5))

    categories = ["Z-Score Only", "IForest Only", "Both Methods"]
    zs_only = comparison["zscore_anomalies"] - comparison["both_methods"]
    if_only = comparison["iforest_anomalies"] - comparison["both_methods"]
    values = [zs_only, if_only, comparison["both_methods"]]
    colors = ["#e74c3c", "#f39c12", "#27ae60"]

    ax.bar(categories, values, color=colors, edgecolor="black", linewidth=0.5)
    ax.set_title("Anomaly Detection Method Comparison", fontsize=13)
    ax.set_ylabel("Number of Anomalous Days")

    for i, v in enumerate(values):
        ax.text(i, v + 0.5, str(v), ha="center", fontsize=11)

    plt.tight_layout()
    chart_path = output_path / "comparison.png"
    plt.savefig(chart_path, dpi=120, bbox_inches="tight")
    plt.close()
    return str(chart_path)

What does this code do?

This function computes how many anomalies were found exclusively by each method versus by both. It then renders a color-coded bar chart: red for z-score-only detections, orange for Isolation Forest-only, and green for days flagged by both methods (high-confidence anomalies).

• Save src/report.py.
• You won't see output yet. The next chunk adds the report orchestrator that calls all three chart functions.

Seeing a syntax error on save?

• Check that plot_volatility and plot_comparison are at the top-level indentation (no extra indent nesting them inside the previous function).
• Confirm the f-string quotes in ax.set_title() use regular double quotes, not smart quotes.
• help me fix a syntax error in my report.py chart functions

Build the HTML report generator

The final function ties everything together. It creates the output directory, generates all three charts, and builds a styled HTML page with metric boxes, embedded images, and a key findings summary.

• Add the generate_report function below plot_comparison in src/report.py:

def generate_report(
    rates_df: pd.DataFrame,
    full_features: pd.DataFrame,
    zscore_results: pd.DataFrame,
    iforest_results: pd.DataFrame,
    comparison: dict,
) -> str:
    """Generate the full HTML business insights report."""
    OUTPUT_DIR.mkdir(exist_ok=True)

    print("Generating visualizations...")
    rates_chart = plot_rate_with_anomalies(rates_df, zscore_results, iforest_results, OUTPUT_DIR)
    vol_chart = plot_volatility(full_features, OUTPUT_DIR)
    comp_chart = plot_comparison(comparison, OUTPUT_DIR)

    print("Building HTML report...")
    html = f"""<!DOCTYPE html>
<html>
<head>
    <title>FX Rate Anomaly Detection Report</title>
    <style>
        body {{ font-family: Arial, sans-serif; max-width: 1100px; margin: 0 auto; padding: 20px; background: #f9f9f9; }}
        h1 {{ color: #2c3e50; border-bottom: 3px solid #3498db; padding-bottom: 10px; }}
        h2 {{ color: #34495e; margin-top: 30px; }}
        .metric-box {{ display: inline-block; background: white; border: 1px solid #ddd; border-radius: 8px; padding: 15px 25px; margin: 10px; text-align: center; box-shadow: 0 2px 4px rgba(0,0,0,0.1); }}
        .metric-value {{ font-size: 28px; font-weight: bold; color: #2980b9; }}
        .metric-label {{ font-size: 12px; color: #7f8c8d; margin-top: 5px; }}
        img {{ max-width: 100%; border: 1px solid #ddd; border-radius: 4px; margin: 15px 0; }}
        .summary {{ background: white; padding: 20px; border-radius: 8px; border-left: 4px solid #3498db; margin: 20px 0; }}
    </style>
</head>
<body>
    <h1>FX Rate Anomaly Detection Report</h1>
    <p>Pipeline run covering {comparison['total_days']} trading days across 4 EUR currency pairs.</p>

    <div>
        <div class="metric-box">
            <div class="metric-value">{comparison['zscore_anomalies']}</div>
            <div class="metric-label">Z-Score Anomalies</div>
        </div>
        <div class="metric-box">
            <div class="metric-value">{comparison['iforest_anomalies']}</div>
            <div class="metric-label">Isolation Forest Anomalies</div>
        </div>
        <div class="metric-box">
            <div class="metric-value">{comparison['both_methods']}</div>
            <div class="metric-label">Detected by Both</div>
        </div>
        <div class="metric-box">
            <div class="metric-value">{comparison['agreement_pct']}%</div>
            <div class="metric-label">Method Agreement</div>
        </div>
    </div>

    <h2>Exchange Rates with Anomalies</h2>
    <img src="rates_anomalies.png" alt="FX rates with anomalies highlighted">

    <h2>Rolling Volatility Analysis</h2>
    <img src="volatility.png" alt="20-day annualized volatility">

    <h2>Detection Method Comparison</h2>
    <img src="comparison.png" alt="Method comparison">

    <div class="summary">
        <h2>Key Findings</h2>
        <ul>
            <li>Z-score method detected <strong>{comparison['zscore_anomalies']}</strong> anomalous days (statistical outliers beyond 3 sigma)</li>
            <li>Isolation Forest detected <strong>{comparison['iforest_anomalies']}</strong> anomalous days (multivariate ML-based detection)</li>
            <li><strong>{comparison['both_methods']}</strong> days were flagged by both methods (high-confidence anomalies)</li>
            <li>Overall method agreement: <strong>{comparison['agreement_pct']}%</strong> of trading days received the same classification</li>
        </ul>
    </div>
</body>
</html>"""

    report_path = OUTPUT_DIR / "report.html"
    report_path.write_text(html)
    print(f"Report saved to: {report_path}")
    return str(report_path)

What does this code do?

This orchestrator function does three things in sequence:

• Creates the output/ directory if it does not already exist.
• Calls all three chart functions to save PNGs into that directory.
• Builds an f-string HTML template with inline CSS styling, metric boxes showing the comparison dict values, embedded chart images via relative src paths, and a key findings summary section.

The double curly braces ({{ and }}) are Python f-string escaping so that literal braces appear in the CSS output.

Why an HTML report instead of a Jupyter notebook?

A standalone HTML report demonstrates production thinking. Notebooks are great for exploration, but a generated HTML artifact shows the pipeline produces a deliverable that can be shared with non-technical stakeholders or embedded in a dashboard.

• Save src/report.py.
• Confirm the file has no syntax errors by running:

python -c "import src.report; print('report.py loaded OK')"

You should see report.py loaded OK printed with no errors.

Seeing an ImportError or SyntaxError?

• Make sure you are running the command from your project root directory (the folder containing src/).
• Check that src/__init__.py exists. Without it, Python cannot import from the src package.
• Verify the f-string HTML template has matching triple-quote delimiters at both ends.
• help me debug my report.py import error

✔️ Awesome, I've got everything!

Great. Double check you have saved src/report.py and that the import check passed.

ⓧ I'd like to double check the full code

Here is the complete src/report.py file:

"""Business insights report generation."""

import pandas as pd
import numpy as np
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
from pathlib import Path


OUTPUT_DIR = Path("output")


def plot_rate_with_anomalies(
    rates_df: pd.DataFrame,
    zscore_results: pd.DataFrame,
    iforest_results: pd.DataFrame,
    output_path: Path,
) -> str:
    """Plot FX rates with anomaly points highlighted."""
    fig, axes = plt.subplots(2, 2, figsize=(14, 10))
    axes = axes.flatten()

    rate_cols = [c for c in rates_df.columns if c.startswith("EUR_") and "_" not in c[4:]]
    common_idx = rates_df.index.intersection(zscore_results.index).intersection(iforest_results.index)

    for i, col in enumerate(rate_cols[:4]):
        ax = axes[i]
        ax.plot(rates_df.loc[common_idx].index, rates_df.loc[common_idx, col], linewidth=0.8, color="steelblue", label="Rate")

        # Z-score anomalies
        zs_anom = zscore_results.loc[common_idx]
        zs_mask = zs_anom["zscore_anomaly"] == 1
        if zs_mask.any():
            ax.scatter(
                rates_df.loc[common_idx][zs_mask].index,
                rates_df.loc[common_idx][zs_mask][col],
                color="red", s=20, zorder=5, label="Z-score", alpha=0.7
            )

        # Isolation Forest anomalies
        if_anom = iforest_results.loc[common_idx]
        if_mask = if_anom["iforest_anomaly"] == 1
        if if_mask.any():
            ax.scatter(
                rates_df.loc[common_idx][if_mask].index,
                rates_df.loc[common_idx][if_mask][col],
                color="orange", s=20, zorder=5, marker="^", label="IForest", alpha=0.7
            )

        ax.set_title(col.replace("_", "/"), fontsize=11)
        ax.xaxis.set_major_formatter(mdates.DateFormatter("%b %Y"))
        ax.tick_params(axis="x", rotation=45)
        ax.legend(fontsize=8)

    plt.tight_layout()
    chart_path = output_path / "rates_anomalies.png"
    plt.savefig(chart_path, dpi=120, bbox_inches="tight")
    plt.close()
    return str(chart_path)


def plot_volatility(full_features: pd.DataFrame, output_path: Path) -> str:
    """Plot rolling volatility for each currency pair."""
    fig, ax = plt.subplots(figsize=(12, 5))

    vol_cols = [c for c in full_features.columns if "_volatility_20d" in c]
    for col in vol_cols:
        label = col.replace("_volatility_20d", "").replace("_", "/")
        ax.plot(full_features.index, full_features[col], linewidth=0.9, label=label)

    ax.set_title("20-Day Annualized Volatility", fontsize=13)
    ax.set_ylabel("Volatility")
    ax.legend()
    ax.xaxis.set_major_formatter(mdates.DateFormatter("%b %Y"))
    ax.tick_params(axis="x", rotation=45)

    plt.tight_layout()
    chart_path = output_path / "volatility.png"
    plt.savefig(chart_path, dpi=120, bbox_inches="tight")
    plt.close()
    return str(chart_path)


def plot_comparison(comparison: dict, output_path: Path) -> str:
    """Plot method comparison bar chart."""
    fig, ax = plt.subplots(figsize=(8, 5))

    categories = ["Z-Score Only", "IForest Only", "Both Methods"]
    zs_only = comparison["zscore_anomalies"] - comparison["both_methods"]
    if_only = comparison["iforest_anomalies"] - comparison["both_methods"]
    values = [zs_only, if_only, comparison["both_methods"]]
    colors = ["#e74c3c", "#f39c12", "#27ae60"]

    ax.bar(categories, values, color=colors, edgecolor="black", linewidth=0.5)
    ax.set_title("Anomaly Detection Method Comparison", fontsize=13)
    ax.set_ylabel("Number of Anomalous Days")

    for i, v in enumerate(values):
        ax.text(i, v + 0.5, str(v), ha="center", fontsize=11)

    plt.tight_layout()
    chart_path = output_path / "comparison.png"
    plt.savefig(chart_path, dpi=120, bbox_inches="tight")
    plt.close()
    return str(chart_path)


def generate_report(
    rates_df: pd.DataFrame,
    full_features: pd.DataFrame,
    zscore_results: pd.DataFrame,
    iforest_results: pd.DataFrame,
    comparison: dict,
) -> str:
    """Generate the full HTML business insights report."""
    OUTPUT_DIR.mkdir(exist_ok=True)

    print("Generating visualizations...")
    rates_chart = plot_rate_with_anomalies(rates_df, zscore_results, iforest_results, OUTPUT_DIR)
    vol_chart = plot_volatility(full_features, OUTPUT_DIR)
    comp_chart = plot_comparison(comparison, OUTPUT_DIR)

    print("Building HTML report...")
    html = f"""<!DOCTYPE html>
<html>
<head>
    <title>FX Rate Anomaly Detection Report</title>
    <style>
        body {{ font-family: Arial, sans-serif; max-width: 1100px; margin: 0 auto; padding: 20px; background: #f9f9f9; }}
        h1 {{ color: #2c3e50; border-bottom: 3px solid #3498db; padding-bottom: 10px; }}
        h2 {{ color: #34495e; margin-top: 30px; }}
        .metric-box {{ display: inline-block; background: white; border: 1px solid #ddd; border-radius: 8px; padding: 15px 25px; margin: 10px; text-align: center; box-shadow: 0 2px 4px rgba(0,0,0,0.1); }}
        .metric-value {{ font-size: 28px; font-weight: bold; color: #2980b9; }}
        .metric-label {{ font-size: 12px; color: #7f8c8d; margin-top: 5px; }}
        img {{ max-width: 100%; border: 1px solid #ddd; border-radius: 4px; margin: 15px 0; }}
        .summary {{ background: white; padding: 20px; border-radius: 8px; border-left: 4px solid #3498db; margin: 20px 0; }}
    </style>
</head>
<body>
    <h1>FX Rate Anomaly Detection Report</h1>
    <p>Pipeline run covering {comparison['total_days']} trading days across 4 EUR currency pairs.</p>

    <div>
        <div class="metric-box">
            <div class="metric-value">{comparison['zscore_anomalies']}</div>
            <div class="metric-label">Z-Score Anomalies</div>
        </div>
        <div class="metric-box">
            <div class="metric-value">{comparison['iforest_anomalies']}</div>
            <div class="metric-label">Isolation Forest Anomalies</div>
        </div>
        <div class="metric-box">
            <div class="metric-value">{comparison['both_methods']}</div>
            <div class="metric-label">Detected by Both</div>
        </div>
        <div class="metric-box">
            <div class="metric-value">{comparison['agreement_pct']}%</div>
            <div class="metric-label">Method Agreement</div>
        </div>
    </div>

    <h2>Exchange Rates with Anomalies</h2>
    <img src="rates_anomalies.png" alt="FX rates with anomalies highlighted">

    <h2>Rolling Volatility Analysis</h2>
    <img src="volatility.png" alt="20-day annualized volatility">

    <h2>Detection Method Comparison</h2>
    <img src="comparison.png" alt="Method comparison">

    <div class="summary">
        <h2>Key Findings</h2>
        <ul>
            <li>Z-score method detected <strong>{comparison['zscore_anomalies']}</strong> anomalous days (statistical outliers beyond 3 sigma)</li>
            <li>Isolation Forest detected <strong>{comparison['iforest_anomalies']}</strong> anomalous days (multivariate ML-based detection)</li>
            <li><strong>{comparison['both_methods']}</strong> days were flagged by both methods (high-confidence anomalies)</li>
            <li>Overall method agreement: <strong>{comparison['agreement_pct']}%</strong> of trading days received the same classification</li>
        </ul>
    </div>
</body>
</html>"""

    report_path = OUTPUT_DIR / "report.html"
    report_path.write_text(html)
    print(f"Report saved to: {report_path}")
    return str(report_path)

Your reporting module is ready. Next up, you'll wire all four modules together into a single CLI pipeline runner that executes the full workflow end-to-end.

Wire Up the CLI Pipeline Runner

You have four modules that each handle one piece of the pipeline: ingestion, feature engineering, detection, and reporting.

Now you need a single entry point that chains them together in sequence. In this step, you will write run_pipeline.py to orchestrate the full pipeline with structured logging so you can see each stage complete in real time.

In this step, get ready to:

• Write the CLI pipeline runner that imports and chains all four modules.
• Add timestamped logging so each pipeline stage is visible at runtime.
• Run the full pipeline and verify the HTML report generates correctly.

Write the pipeline runner script

Production data pipelines need an orchestrator that calls each stage in order and logs progress. This makes debugging easier because you can see exactly which stage failed and how long each one took.

• Create a new file called run_pipeline.py in your project root (the same folder that contains requirements.txt and the src/ folder).
• Add the imports and logging configuration by pasting this code at the top of the file:

"""FX Rate Anomaly Detection Pipeline - Main Entry Point."""

import time
import logging

from src.ingest import fetch_fx_rates, validate_data
from src.features import build_features
from src.detect import run_detection
from src.report import generate_report


logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%H:%M:%S",
)
logger = logging.getLogger(__name__)

What does this code do?

The imports pull in the main function from each of your four src/ modules. Each function represents one pipeline stage.

logging.basicConfig sets up Python's built-in logging with timestamps in HH:MM:SS format. This lets you see exactly when each stage starts without cluttering the output with full dates.

• Below the logging setup, add the main() function and the entry point guard by pasting this code:

def main():
    start_time = time.time()
    logger.info("Starting FX Rate Anomaly Detection Pipeline")
    logger.info("=" * 50)

    # Step 1: Data Ingestion
    logger.info("STEP 1: Data Ingestion")
    raw_data = fetch_fx_rates(years_back=2)
    rates_df = validate_data(raw_data)

    # Step 2: Feature Engineering
    logger.info("STEP 2: Feature Engineering")
    full_features, features_scaled, z_score_data = build_features(rates_df)

    # Step 3: Anomaly Detection
    logger.info("STEP 3: Anomaly Detection")
    zscore_results, iforest_results, comparison = run_detection(features_scaled, z_score_data)

    # Step 4: Report Generation
    logger.info("STEP 4: Report Generation")
    report_path = generate_report(
        rates_df, full_features, zscore_results, iforest_results, comparison
    )

    elapsed = time.time() - start_time
    logger.info("=" * 50)
    logger.info(f"Pipeline complete in {elapsed:.1f}s")
    logger.info(f"Open {report_path} in your browser to view results")


if __name__ == "__main__":
    main()

What does this code do?

• main() chains the four pipeline stages in order. Each stage's output becomes the next stage's input.
• time.time() records the start time so you can measure total pipeline duration at the end.
• logger.info() prints timestamped messages marking each stage. This makes it easy to identify which stage is slow or failing.
• The if __name__ == "__main__" guard ensures main() only runs when you execute the script directly, not when it is imported as a module.

• Save run_pipeline.py.

Seeing a red underline on the imports?

If your editor shows import errors for the src modules, make sure you have the src/__init__.py file in place. That empty file tells Python to treat src/ as a package.

Also confirm that run_pipeline.py is in the project root (next to the src/ folder), not inside src/ itself.

Why are my src imports failing?

Run the full pipeline

The pipeline will fetch live FX data from the Frankfurter API, process it through all four stages, and produce your HTML report. This takes around 10-30 seconds depending on your network speed.

• Make sure your virtual environment is active (you should see the environment name in your terminal prompt).
• Run the pipeline by executing this command:

python run_pipeline.py

You should see timestamped log output for each stage, ending with a line like:

12:34:56 [INFO] Pipeline complete in 15.3s
12:34:56 [INFO] Open output/report.html in your browser to view results

What should I see?

The output shows each stage executing in order: data ingestion (fetching from the API and validating), feature engineering (computing rolling stats), anomaly detection (running both methods), and report generation (saving charts and HTML).

The total time depends on network speed for the API call and CPU speed for the Isolation Forest model fitting.

• Confirm the output files exist by running:

ls output/

• You should see rates_anomalies.png, volatility.png, comparison.png, and report.html listed.
• Open output/report.html in your browser to view the final report.

You should see a styled HTML page with metric boxes at the top (showing anomaly counts and agreement percentage), three embedded chart images, and a key findings summary at the bottom.

Pipeline failing with an error?

• ModuleNotFoundError: Make sure your virtual environment is activated and you installed all packages from requirements.txt.
• ConnectionError or timeout: The Frankfurter API may be temporarily unreachable. Check your internet connection and try again in a few seconds.
• FileNotFoundError: Make sure you are running the command from your project root directory (the folder containing both run_pipeline.py and the src/ folder).

Help me debug my pipeline error.

✔️ Awesome, I've got everything!

Great. Double check you have saved run_pipeline.py and that your output/ folder contains all four generated files.

ⓧ I'd like to double check the full code

Here is the complete run_pipeline.py file for comparison:

"""FX Rate Anomaly Detection Pipeline - Main Entry Point."""

import time
import logging

from src.ingest import fetch_fx_rates, validate_data
from src.features import build_features
from src.detect import run_detection
from src.report import generate_report


logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%H:%M:%S",
)
logger = logging.getLogger(__name__)


def main():
    start_time = time.time()
    logger.info("Starting FX Rate Anomaly Detection Pipeline")
    logger.info("=" * 50)

    # Step 1: Data Ingestion
    logger.info("STEP 1: Data Ingestion")
    raw_data = fetch_fx_rates(years_back=2)
    rates_df = validate_data(raw_data)

    # Step 2: Feature Engineering
    logger.info("STEP 2: Feature Engineering")
    full_features, features_scaled, z_score_data = build_features(rates_df)

    # Step 3: Anomaly Detection
    logger.info("STEP 3: Anomaly Detection")
    zscore_results, iforest_results, comparison = run_detection(features_scaled, z_score_data)

    # Step 4: Report Generation
    logger.info("STEP 4: Report Generation")
    report_path = generate_report(
        rates_df, full_features, zscore_results, iforest_results, comparison
    )

    elapsed = time.time() - start_time
    logger.info("=" * 50)
    logger.info(f"Pipeline complete in {elapsed:.1f}s")
    logger.info(f"Open {report_path} in your browser to view results")


if __name__ == "__main__":
    main()

Your end-to-end pipeline is working. You have a single command that takes raw API data through ingestion, feature engineering, anomaly detection, and report generation. Next up, the Secret Mission challenges you to add cross-currency correlation analysis.

Secret mission

Cross-Currency Correlation Breakdown Detection

Currency pairs don't move independently. When the historical relationship between pairs breaks down, it signals market stress or structural shifts. In this mission, you'll add correlation analysis that detects these breakdown events and visualizes them in your report.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally and uses a free API with no quotas, 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:

• Python virtual environment (venv/ folder).
• Project directory with src/ package (ingest.py, features.py, detect.py, report.py).
• Pipeline entry point: run_pipeline.py.
• Generated output: output/ folder with HTML report and PNG chart images.
• Installed Python packages: pandas, numpy, scikit-learn, statsmodels, matplotlib, requests.

Keep everything running

No action needed. Choose this if you're still actively building or plan to extend the pipeline with the Secret Mission.

• Your virtual environment stays in the project folder and uses minimal disk space.
• The Frankfurter API is free with no account, so nothing accrues charges.
• You can re-run the pipeline anytime with python run_pipeline.py.

Pause - I'll come back to this later

Deactivate the virtual environment to free up your terminal session. Your files and packages stay intact for next time.

• Run deactivate in your terminal to exit the virtual environment.
• When you return, navigate back to your project folder and reactivate with source venv/bin/activate (macOS) or venv\Scripts\activate (Windows).
• All packages and generated output will still be there.

Delete - I don't want to use this again

Remove all project resources and start fresh. This frees up disk space from the packages and generated files.

• Deactivate the virtual environment first by running deactivate.
• Navigate to your Desktop (or wherever you created the project).
• Delete the entire project folder. On macOS/Linux: rm -rf fx-anomaly-pipeline. On Windows: right-click the folder in File Explorer and select Delete.
• No accounts or cloud resources need to be closed.

That's a Wrap!

That's a Wrap!

Nice work! You just built a production-style FX anomaly detection pipeline from raw API data through to a visual business insights report.

Here's what you learned:

• Built a data ingestion pipeline that fetches 2 years of real FX rate data from the Frankfurter API and engineers time-series features including rolling statistics, returns, and volatility.
• Implemented and compared two anomaly detection approaches: a statistical rolling z-score method and an ML-based Isolation Forest model with scored outputs and agreement metrics.
• Generated a professional HTML business insights report with embedded matplotlib charts showing rate time series, volatility trends, and a detection method comparison.
• Secret Mission: Added cross-currency correlation analysis to detect structural market shifts when correlation breakdowns occur.

Ready to quiz yourself?