Examples

This page provides comprehensive examples of how to use the forex_data package for various common tasks.

Historical Data Examples

Example 1: Basic Data Retrieval

Get historical data for a specific currency pair and date range:

from forex_data import HistoricalManagerDB
from forex_data.config import APPCONFIG_FILE_YAML

# Create manager for NZD/USD
manager = HistoricalManagerDB(
    config=APPCONFIG_FILE_YAML
)

# Get hourly data for 2 months
data = manager.get_data(
    ticker='NZDUSD',
    timeframe='1h',
    start='2009-10-04',
    end='2009-12-03'
)

print(f"Retrieved {len(data)} rows of data")
print(data.head())

Expected Output (with Polars):

Retrieved 1421 rows of data
┌─────────────────────┬─────────┬─────────┬─────────┬─────────┐
│ timestamp           ┆ open    ┆ high    ┆ low     ┆ close   │
│ ---                 ┆ ---     ┆ ---     ┆ ---     ┆ ---     │
│ datetime[ms]        ┆ f32     ┆ f32     ┆ f32     ┆ f32     │
╞═════════════════════╪═════════╪═════════╪═════════╪═════════╡
│ 2009-10-04 17:00:00 ┆ 0.7172  ┆ 0.71735 ┆ 0.7165  ┆ 0.71665 │
│ 2009-10-04 18:00:00 ┆ 0.71655 ┆ 0.7168  ┆ 0.71555 ┆ 0.71635 │
│ 2009-10-04 19:00:00 ┆ 0.7164  ┆ 0.71665 ┆ 0.71385 ┆ 0.71445 │
└─────────────────────┴─────────┴─────────┴─────────┴─────────┘

Example 2: Working with Multiple Timeframes

Aggregate data into different timeframes:

from forex_data import HistoricalManagerDB

manager = HistoricalManagerDB(
    config='appconfig.yaml'
)

# Get 1-minute data
minute_data = manager.get_data(
    ticker='EURUSD',
    timeframe='1m',
    start='2020-01-01 00:00:00',
    end='2020-01-01 23:59:59'
)

# Add and cache a 15-minute timeframe
manager.add_timeframe('15m')

# Get 15-minute data
fifteen_min_data = manager.get_data(
    ticker='EURUSD',
    timeframe='15m',
    start='2020-01-01',
    end='2020-01-07'
)

# Add daily timeframe
manager.add_timeframe('1D')

# Get daily data
daily_data = manager.get_data(
    ticker='EURUSD',
    timeframe='1D',
    start='2020-01-01',
    end='2020-12-31'
)

print(f"1-minute candles: {len(minute_data)}")
print(f"15-minute candles: {len(fifteen_min_data)}")
print(f"Daily candles: {len(daily_data)}")

Example 3: Visualizing Historical Data

Create candlestick charts from historical data:

from forex_data import HistoricalManagerDB

manager = HistoricalManagerDB(
    config='appconfig.yaml'
)

# Plot 5 months of daily data
manager.plot(
    ticker='NZDUSD',
    timeframe='1D',
    start_date='2013-02-02',
    end_date='2013-06-23'
)

This generates an interactive candlestick chart showing price movements.

Example 4: Multi-Year Analysis

Analyze data across multiple years:

from forex_data import HistoricalManagerDB

manager = HistoricalManagerDB(
    config='appconfig.yaml'
)

# Get 5 years of weekly data
weekly_data = manager.get_data(
    ticker='GBPUSD',
    timeframe='1W',
    start='2015-01-01',
    end='2019-12-31'
)

# Calculate some basic statistics
if hasattr(weekly_data, 'describe'):  # Polars
    print(weekly_data.describe())
else:  # Pandas
    print(weekly_data.describe())

# Find highest and lowest weekly close
max_close = weekly_data['close'].max()
min_close = weekly_data['close'].min()

print(f"Highest weekly close: {max_close}")
print(f"Lowest weekly close: {min_close}")

Example 5: Conditional Data Filtering

Filter data based on price conditions directly during retrieval:

from forex_data import (
      HistoricalManagerDB,
      COLUMN_NAME,
      SQL_COMPARISON_OPERATORS,
      SQL_CONDITION_AGGREGATION_MODES
   )

manager = HistoricalManagerDB(config='appconfig.yaml')

# 1. Simple Filter: Get days where Close >= 1.12
high_close_data = manager.get_data(
   ticker='EURUSD',
   timeframe='1D',
      start='2020-06-01',
      end='2020-06-30',
      comparison_column_name=COLUMN_NAME.CLOSE,
      check_level=1.12,
      comparison_operator=SQL_COMPARISON_OPERATORS.GREATER_THAN_OR_EQUAL
   )

print(f"Days with Close >= 1.12: {len(high_close_data)}")

# 2. Complex Filter: Get days where High > 1.145 OR Low < 1.12
volatile_days = manager.get_data(
   ticker='EURUSD',
   timeframe='1D',
      start='2019-01-01',
      end='2019-12-31',
      comparison_column_name=[
         COLUMN_NAME.HIGH,
         COLUMN_NAME.LOW
      ],
      check_level=[1.145, 1.12],
      comparison_operator=[
         SQL_COMPARISON_OPERATORS.GREATER_THAN,
         SQL_COMPARISON_OPERATORS.LESS_THAN
      ],
      aggregation_mode=SQL_CONDITION_AGGREGATION_MODES.OR
   )

print(f"Volatile days found: {len(volatile_days)}")

Real-time Data Examples

Example 5: Get Latest Market Price (Twelve Data)

Retrieve the live top-of-book market quote:

from forex_data import TwelveDataConnector

connector = TwelveDataConnector(plan="free")
latest_price_lf = connector.get_realtime_price(symbol="EUR/USD")
print(latest_price_lf.collect())

Expected Output:

┌────────────────────────────┬─────────┬─────────┬──────────┐
│ timestamp                  ┆ ticker  ┆ price   ┆ timezone │
│ ---                        ┆ ---     ┆ ---     ┆ ---      │
│ datetime[μs]               ┆ str     ┆ f64     ┆ str      │
╞════════════════════════════╪═════════╪═════════╪══════════╡
│ 2026-10-12 16:11:47.711957 ┆ EUR/USD ┆ 1.11762 ┆ UTC      │
└────────────────────────────┴─────────┴─────────┴──────────┘

Example 6: Top-of-Book Quotes (Tiingo)

Retrieve bid/ask spreads from Tiingo FX:

from forex_data import TiingoConnector

connector = TiingoConnector(plan="free")
quote_lf = connector.get_realtime_price(symbol="EUR/USD")
print(quote_lf.collect())

Example 7: Institutional Trendbars (cTrader)

Retrieve broker trendbars via Spotware Open API:

import os
from forex_data import CTraderConnector

connector = CTraderConnector(
    client_id=os.environ["CTRADER_CLIENT_ID"],
    client_secret=os.environ["CTRADER_CLIENT_SECRET"],
    access_token=os.environ["CTRADER_ACCESS_TOKEN"],
    broker_account_id=os.environ["CTRADER_ACCOUNT_ID"],
)
connector.connect()
bars_lf = connector.get_data("EURUSD", "1h", "2026-10-01 00:00:00", "2026-10-05 00:00:00")
print(bars_lf.collect())
connector.close()

Plotting Examples

Example 8: Interactive Historical Candlestick Chart

Render interactive candlestick charts with volume bars:

from forex_data import DataCharts

charts = DataCharts(data_path="~/.test_database", offline=True)
charts.plot_historical(
    ticker="EURUSD",
    timeframe="1D",
    start_date="2024-01-01",
    end_date="2024-06-01",
    chart_type="ohlc"
)

Example 9: Hourly Line Chart with VWMP

Plot closing prices alongside Volume-Weighted Mid-Price:

from forex_data import DataCharts

charts = DataCharts(data_path="~/.test_database", offline=True)
charts.plot_historical(
    ticker="EURUSD",
    timeframe="1h",
    start_date="2024-03-01",
    end_date="2024-03-15",
    chart_type="line"
)

Combined Examples

Example 9: Historical + Real-time Combination

Combine historical data with recent real-time updates:

from forex_data import HistoricalManagerDB, RealtimeManager
import polars as pl  # or pandas as pd

ticker = 'NZDUSD'
config = 'appconfig.yaml'

# Get historical data up to last month
hist_manager = HistoricalManagerDB(config=config)
historical = hist_manager.get_data(
    ticker=ticker,
    timeframe='1D',
    start='2020-01-01',
    end='2023-12-31'
)

# Get recent data
rt_manager = RealtimeManager(config=config)
recent = rt_manager.get_daily_close(
    ticker=ticker,
    day_start='2024-01-01',
    day_end='2024-12-31'
)

# Combine datasets (Polars example)
if isinstance(historical, pl.DataFrame):
    combined = pl.concat([historical, recent])
else:  # Pandas
    combined = pd.concat([historical, recent])

print(f"Total data points: {len(combined)}")
print(f"Date range: {combined['timestamp'].min()} to {combined['timestamp'].max()}")

Example 10: Multi-Currency Analysis

Analyze multiple currency pairs:

from forex_data import HistoricalManagerDB

pairs = ['EURUSD', 'GBPUSD', 'USDJPY', 'AUDUSD']
data_dict = {}

for pair in pairs:
    manager = HistoricalManagerDB(
        config='appconfig.yaml'
    )

    data = manager.get_data(
        ticker=pair,
        timeframe='1D',
        start='2023-01-01',
        end='2023-12-31'
    )

    data_dict[pair] = data
    print(f"{pair}: {len(data)} days of data")

# You can now analyze correlations, etc.
for pair, data in data_dict.items():
    avg_close = data['close'].mean()
    print(f"{pair} average close price: {avg_close:.5f}")

Advanced Examples

Example 11: Custom Data Processing Pipeline

Build a custom processing pipeline:

from forex_data import HistoricalManagerDB
import polars as pl

manager = HistoricalManagerDB(
    config='appconfig.yaml'
)

# Get raw data
data = manager.get_data(
    ticker='EURUSD',
    timeframe='1h',
    start='2023-01-01',
    end='2023-12-31'
)

# Add custom calculations (Polars example)
if isinstance(data, pl.DataFrame):
    processed = data.with_columns([
        # Calculate price range
        (pl.col('high') - pl.col('low')).alias('range'),

        # Calculate percentage change
        ((pl.col('close') - pl.col('open')) / pl.col('open') * 100).alias('pct_change'),

        # Moving average (7-hour)
        pl.col('close').rolling_mean(window_size=7).alias('ma_7')
    ])

print(processed.head())

Example 12: Error Handling and Validation

Robust data retrieval with error handling:

from forex_data import HistoricalManagerDB
from loguru import logger

def get_forex_data_safely(ticker, timeframe, start, end):
    """Safely retrieve forex data with error handling."""
    try:
        manager = HistoricalManagerDB(
            config='appconfig.yaml'
        )

        data = manager.get_data(
            ticker=ticker,
            timeframe=timeframe,
            start=start,
            end=end
        )

        # Validate data
        if len(data) == 0:
            logger.warning(f"No data retrieved for {ticker}")
            return None

        logger.info(f"Successfully retrieved {len(data)} rows for {ticker}")
        return data

    except FileNotFoundError:
        logger.error("Configuration file not found")
        return None
    except Exception as e:
        logger.error(f"Error retrieving data: {e}")
        return None

# Use the function
data = get_forex_data_safely(
    ticker='EURUSD',
    timeframe='1D',
    start='2023-01-01',
    end='2023-12-31'
)

if data is not None:
    print("Data retrieved successfully")

Example 13: Performance Optimization

Optimize for large datasets:

from forex_data import HistoricalManagerDB

# Defaults to polars_lazy engine and parquet storage automatically
manager = HistoricalManagerDB(
    config='appconfig.yaml'
)

# Get large dataset
import time
start_time = time.time()

data = manager.get_data(
    ticker='EURUSD',
    timeframe='1m',
    start='2020-01-01',
    end='2023-12-31'
)

elapsed = time.time() - start_time
print(f"Retrieved {len(data)} rows in {elapsed:.2f} seconds")
print(f"Rate: {len(data)/elapsed:.0f} rows/second")

Example 14: Export to Different Formats

Export data for use in other tools:

from forex_data import HistoricalManagerDB

manager = HistoricalManagerDB(
    config='appconfig.yaml'
)

data = manager.get_data(
    ticker='GBPUSD',
    timeframe='1D',
    start='2023-01-01',
    end='2023-12-31'
)

# Export to CSV
if hasattr(data, 'write_csv'):  # Polars
    data.write_csv('gbpusd_2023.csv')
else:  # Pandas
    data.to_csv('gbpusd_2023.csv', index=False)

# Export to Parquet
if hasattr(data, 'write_parquet'):  # Polars
    data.write_parquet('gbpusd_2023.parquet')
else:  # Pandas
    data.to_parquet('gbpusd_2023.parquet', index=False)

print("Data exported successfully")

Testing Examples

Example 15: Running the Test Suite

The package includes comprehensive tests. Run them with:

uv run pytest

Run specific test files:

uv run pytest tests/test_historical_manager_db.py
uv run pytest tests/test_realtime_twelvedata_connector.py

Run with verbose output:

uv run pytest -v

Tips and Best Practices

  1. Built-in Polars Lazy Performance

    The engine automatically defaults to polars_lazy with Parquet storage for optimal memory usage and query pushdown. You do not need to configure ENGINE or DATA_TYPE.

  2. Cache Timeframes

    # Add common timeframes once
manager.add_timeframe('15m')
manager.add_timeframe('1h')
manager.add_timeframe('1D')

  3. Handle Time Zones

    # Forex data is typically in UTC
# Be explicit about timezones in your date strings
data = manager.get_data(
    timeframe='1h',
    start='2023-01-01 00:00:00',  # UTC
    end='2023-01-31 23:59:59'
)

  4. Monitor API Rate Limits

    from time import sleep

# Add delays between API calls
for pair in ['EURUSD', 'GBPUSD', 'USDJPY']:
    rt_manager = RealtimeManager(config='appconfig.yaml')
    data = rt_manager.get_daily_close(ticker=pair, recent_days_window=10)
    sleep(1)  # Respect rate limits

  5. Validate Data Quality

    # Check for missing data
expected_rows = 365  # For daily data in a year
if len(data) < expected_rows:
    print(f"Warning: Expected {expected_rows} rows, got {len(data)}")

Next Steps