Using Quantdle Data

This example shows how to use Quantdle as your data source — downloading historical data, caching it locally as CSV, and running a backtest on it. The strategy itself is a simple MA Crossover, but the focus here is the data workflow.

Difficulty

Beginner

Focus

Data sourcing and caching

Data Provider

Quantdle API

Prerequisites

Quantdle API credentials

Why Use Quantdle?

PyEventBT backtests run on CSV files. You can create these manually, but the QuantdleDataUpdater automates the entire workflow:

• First run: Downloads all data from Quantdle and saves as CSV
• Subsequent runs: Uses cached CSV files — no API calls needed
• Date range extension: Only downloads the missing gap, not the full range
• Format conversion: Automatically converts Quantdle data to PyEventBT’s expected CSV format

No Quantdle Account?

You can still run any PyEventBT strategy by setting csv_dir=None to use the built-in sample dataset, or by providing your own CSV files. Quantdle is one option, not a requirement.

Setup

1. Install the Quantdle client

   pip install quantdle

2. Get your API credentials from quantdle.com — you’ll need an API Key and API Key ID.

Step-by-Step Walkthrough

1. Configure data download

from pyeventbt import QuantdleDataUpdater
from datetime import datetime

QUANTDLE_API_KEY = "your_api_key_here"
QUANTDLE_API_KEY_ID = "your_api_key_id_here"

csv_dir = './data'                      # Where CSVs will be cached
symbols_to_trade = ['EURUSD']
from_date = datetime(2020, 1, 1)
to_date = datetime(2023, 12, 1)

2. Update local CSV cache

updater = QuantdleDataUpdater(
    api_key=QUANTDLE_API_KEY,
    api_key_id=QUANTDLE_API_KEY_ID
)

updater.update_data(
    csv_dir=csv_dir,
    symbols=symbols_to_trade,
    start_date=from_date,
    end_date=to_date,
    timeframe="1min"
)

This is the key step. The updater checks what data already exists in csv_dir and only downloads what’s missing.

3. Define your strategy

The strategy code is identical to any other PyEventBT strategy — the only difference is that csv_dir points to the Quantdle cache instead of None.

strategy = Strategy(logging_level=logging.INFO)

@strategy.custom_signal_engine(
    strategy_id=strategy_id,
    strategy_timeframes=[StrategyTimeframes.ONE_DAY]
)
def ma_crossover_strategy(event: BarEvent, modules: Modules):
    # ... your strategy logic (same as any other strategy)

4. Run backtest with cached data

backtest = strategy.backtest(
    strategy_id=strategy_id,
    initial_capital=100000,
    symbols_to_trade=symbols_to_trade,
    csv_dir=csv_dir,               # Points to the Quantdle cache
    start_date=from_date,
    end_date=to_date,
    ...
)
backtest.plot()

The only change from a normal backtest is csv_dir=csv_dir instead of csv_dir=None.

How the Caching Works

First run

update_data() called
  |-- No CSVs found in ./data/
  |-- Downloads EURUSD 2020-01-01 to 2023-12-01 from Quantdle
  |-- Converts to PyEventBT CSV format
  \-- Saves as ./data/EURUSD.csv

Second run (same date range)

update_data() called
  |-- Found ./data/EURUSD.csv
  |-- Data already covers 2020-01-01 to 2023-12-01
  \-- No download needed — uses cache instantly

Extending the date range

update_data() called with start_date=2019-01-01
  |-- Found ./data/EURUSD.csv (covers 2020-2023)
  |-- Downloads only 2019-01-01 to 2020-01-01 (the gap)
  \-- Prepends to existing CSV

CSV Format

The updater produces CSV files with no header row, in the format: date,time,open,high,low,close,tickvol,volume,spread. Files are named by symbol (e.g., EURUSD.csv).

QuantdleDataUpdater Reference

Constructor

QuantdleDataUpdater(api_key: str, api_key_id: str)

update_data()

updater.update_data(
    csv_dir: str,          # Directory for CSV cache
    symbols: list[str],    # e.g., ['EURUSD', 'GBPUSD']
    start_date: datetime,
    end_date: datetime,
    timeframe: str = "1min"  # Also accepts "5min", "1h", "1d"
)

Parameter	Default	Description
csv_dir	—	Directory where CSV files are stored (created if missing)
symbols	—	List of symbols to download
start_date	—	Start of the data range
end_date	—	End of the data range
timeframe	"1min"	Bar timeframe. Auto-converted to Quantdle format ("M1", "H1", etc.)

Tips

• Keep a dedicated cache directory — don’t mix Quantdle CSVs with other files
• Add the cache to .gitignore — CSV files can be large and are easily re-downloaded
• Extend gradually — when adding years of data, do it incrementally to verify quality
• Multiple symbols — just add them to the symbols list; each gets its own CSV file

Troubleshooting

Error	Solution
quantdle package is required	Run pip install quantdle
No data received from Quantdle	Check API credentials, symbol name, and internet connection
Error downloading data	Verify the date range and symbol are available on Quantdle

Full Code

Copy this into a .py file and run it directly.

from pyeventbt import (
    Strategy,
    BarEvent,
    SignalEvent,
    Modules,
    StrategyTimeframes,
    PassthroughRiskConfig,
    MinSizingConfig,
    QuantdleDataUpdater,
)
from pyeventbt.events.events import OrderType, SignalType
from pyeventbt.indicators import SMA

from datetime import datetime
from decimal import Decimal
import logging

logger = logging.getLogger("pyeventbt")


# ── Step 1: Configure Data Download ─────────────────────────────

QUANTDLE_API_KEY = "your_api_key_here"
QUANTDLE_API_KEY_ID = "your_api_key_id_here"

csv_dir = './data'                      # Where CSVs will be cached
symbols_to_trade = ['EURUSD']
from_date = datetime(2020, 1, 1)
to_date = datetime(2023, 12, 1)


# ── Step 2: Update Local CSV Cache ──────────────────────────────

updater = QuantdleDataUpdater(
    api_key=QUANTDLE_API_KEY,
    api_key_id=QUANTDLE_API_KEY_ID
)

updater.update_data(
    csv_dir=csv_dir,
    symbols=symbols_to_trade,
    start_date=from_date,
    end_date=to_date,
    timeframe="1min"
)

print("CSV cache updated. Running backtest...")


# ── Step 3: Define Strategy (MA Crossover) ───────────────────────

strategy_id = "quantdle_ma"
strategy = Strategy(logging_level=logging.INFO)

signal_timeframe = StrategyTimeframes.ONE_DAY
strategy_timeframes = [signal_timeframe]

fast_ma_period = 10
slow_ma_period = 30


@strategy.custom_signal_engine(
    strategy_id=strategy_id,
    strategy_timeframes=strategy_timeframes
)
def ma_crossover_strategy(event: BarEvent, modules: Modules):
    if event.timeframe != signal_timeframe:
        return

    symbol = event.symbol
    signal_events = []

    bars_needed = slow_ma_period + 10
    bars = modules.DATA_PROVIDER.get_latest_bars(
        symbol, signal_timeframe, bars_needed
    )

    if bars is None or bars.height < bars_needed:
        return

    close_prices = bars.select('close').to_numpy().flatten()
    fast_ma = SMA.compute(close_prices, fast_ma_period)
    slow_ma = SMA.compute(close_prices, slow_ma_period)

    if fast_ma[-1] > slow_ma[-1]:
        desired_position = "LONG"
    elif fast_ma[-1] < slow_ma[-1]:
        desired_position = "SHORT"
    else:
        return

    open_positions = modules.PORTFOLIO \
        .get_number_of_strategy_open_positions_by_symbol(symbol)

    signal_type = None

    if open_positions['LONG'] == 0 and desired_position == "LONG":
        if open_positions['SHORT'] > 0:
            modules.EXECUTION_ENGINE \
                .close_strategy_short_positions_by_symbol(symbol)
        signal_type = SignalType.BUY

    if open_positions['SHORT'] == 0 and desired_position == "SHORT":
        if open_positions['LONG'] > 0:
            modules.EXECUTION_ENGINE \
                .close_strategy_long_positions_by_symbol(symbol)
        signal_type = SignalType.SELL

    if signal_type is None:
        return

    if modules.TRADING_CONTEXT == "BACKTEST":
        time_generated = event.datetime + signal_timeframe.to_timedelta()
    else:
        time_generated = datetime.now()

    last_tick = modules.DATA_PROVIDER.get_latest_tick(symbol)

    signal_events.append(SignalEvent(
        symbol=symbol,
        time_generated=time_generated,
        strategy_id=strategy_id,
        signal_type=signal_type,
        order_type=OrderType.MARKET,
        order_price=(
            last_tick['ask'] if signal_type == SignalType.BUY
            else last_tick['bid']
        ),
        sl=Decimal('0.0'),
        tp=Decimal('0.0'),
    ))

    return signal_events


# ── Step 4: Run Backtest with Cached Data ────────────────────────

strategy.configure_predefined_sizing_engine(MinSizingConfig())
strategy.configure_predefined_risk_engine(PassthroughRiskConfig())

backtest = strategy.backtest(
    strategy_id=strategy_id,
    initial_capital=100000,
    symbols_to_trade=symbols_to_trade,
    csv_dir=csv_dir,               # Points to the Quantdle cache
    backtest_name=strategy_id,
    start_date=from_date,
    end_date=to_date,
    export_backtest_parquet=False,
    account_currency='USD'
)

backtest.plot()