Skip to main content

Event-driven backtesting and live trading framework for Python and MetaTrader 5

Project description

pyeventbt

PyPI version Python versions License: Apache 2.0 Code style: black

PyEventBT is an institutional-grade event-driven backtesting and live trading framework built with Python for the MetaTrader 5 platform.

It provides a complete mock of the MT5 API for an easy transition between backtesting and live trading, allowing traders to easily develop multi-rule, multi-timeframe and multi-instrument strategies.

Whether you're building simple moving average crossovers or complex multi-rule and multi-timeframe strategies, PyEventBT provides the tools you need to develop, test, and deploy with confidence.

Its modular architecture allows you to design your own signal sources, position sizing logic and risk management overlay as independent and interchangeable blocks.

Why PyEventBT?

If you're looking to connect Python and MetaTrader 5 for algorithmic trading, PyEventBT gives you a complete event-driven backtesting and live trading framework without writing a single line of MQL5.

Its core design principle is one codebase, two modes: the framework provides a full mock of the MT5 API, so the same strategy code you use for backtesting runs unchanged in live trading. This eliminates the common problem of backtest-to-live divergence and lets you focus on strategy logic rather than infrastructure.

  • Native MT5 integration — Connect Python to MetaTrader 5 for live order execution, position management, and real-time data
  • Realistic backtesting — Event-driven processing ensures your strategy only sees data available at that moment, preventing look-ahead bias by design
  • Modular engines — Plug in your own signal generation, position sizing, and risk management logic as independent, interchangeable components
  • Multi-everything — Handle multiple timeframes, instruments, and trading rules in a single strategy
  • Python ecosystem — Built on Polars, NumPy, and Pandas — use any Python library in your strategies

Key Concepts

PyEventBT processes market data through an event pipeline. You only write the signal logic — the framework handles everything else:

BarEvent → Your Strategy → SignalEvent → Sizing Engine → Risk Engine → OrderEvent → Execution Engine → FillEvent → Portfolio
Component What it does
Signal Engine Your strategy logic — receives market data, returns trading signals
Sizing Engine Calculates position size (min lot, fixed, or risk-percentage based)
Risk Engine Validates orders against your risk rules before execution
Execution Engine Routes orders to the simulated broker (backtest) or MT5 (live)
Portfolio Tracks positions, balance, equity, and P&L
Data Provider Serves historical bars (CSV) or real-time data (MT5)

Installation

pip install pyeventbt

Quick Start

This is a summarized guide. For the full documentation, please visit pyeventbt.com.

1. Define your Strategy

Create a strategy by decorating a function with @strategy.custom_signal_engine.

from pyeventbt import (
    Strategy,
    BarEvent,
    SignalEvent,
    Modules,
    StrategyTimeframes,
    PassthroughRiskConfig,
    MinSizingConfig
)
from pyeventbt.events.events import OrderType, SignalType
from pyeventbt.strategy.core.account_currencies import AccountCurrencies
from pyeventbt.indicators import SMA

from datetime import datetime
from decimal import Decimal
import logging

logger = logging.getLogger("pyeventbt")

# Strategy Configuration
strategy_id = "123456"
strategy = Strategy(logging_level=logging.INFO)

# Timeframes
signal_timeframe = StrategyTimeframes.ONE_DAY

strategy_timeframes = [signal_timeframe]

# Trading Configuration
symbols_to_trade = ['EURUSD']
starting_capital = 100000

# Strategy Parameters
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):
    """
    Moving Average Dominance Strategy:
    - Stay long while fast MA is above slow MA
    - Stay short while fast MA is below slow MA
    - Flat (or hold current) when both averages equal
    - Always maintain at most one open position
    """
    
    if event.timeframe != signal_timeframe:
        return
    
    symbol = event.symbol

    signal_events = []
    
    # Get bars for MA calculation
    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
    
    # Calculate moving averages
    close_prices = bars.select('close').to_numpy().flatten()
    fast_ma_values = SMA.compute(close_prices, fast_ma_period)
    slow_ma_values = SMA.compute(close_prices, slow_ma_period)
    
    current_fast_ma = fast_ma_values[-1]
    current_slow_ma = slow_ma_values[-1]
    
    # Determine desired position state
    if current_fast_ma > current_slow_ma:
        desired_position = "LONG"
    elif current_fast_ma < current_slow_ma:
        desired_position = "SHORT"
    else:
        return

    # Check current positions (at current bar time - no lookahead)
    open_positions = modules.PORTFOLIO.get_number_of_strategy_open_positions_by_symbol(symbol)
    
    signal_type = None
    
    
    # Signal generation
    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 == None:
        return
    
    # Time for signal generation (for NEXT bar)
    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)
    
    # Generate signals based on desired position
    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(str(0.0)),
        tp=Decimal(str(0.0)),
    ))
    
    return signal_events

2. Configure and Run Backtest

# Configure Strategy
strategy.configure_predefined_sizing_engine(MinSizingConfig())
strategy.configure_predefined_risk_engine(PassthroughRiskConfig())

# Backtest Configuration
from_date = datetime(year=2020, month=1, day=1)
to_date = datetime(year=2023, month=12, day=1)
# csv_dir = './data' # Change it with your own path to the CSV data
csv_dir = None # If you don't have CSV data, you can set this to None

# Launch Backtest
backtest = strategy.backtest(
    strategy_id=strategy_id,
    initial_capital=starting_capital,
    symbols_to_trade=symbols_to_trade,
    csv_dir=csv_dir,
    backtest_name=strategy_id,
    start_date=from_date,
    end_date=to_date,
    export_backtest_parquet=False,
    account_currency=AccountCurrencies.USD
)

backtest.plot()

Example Strategy

Here is a complete example of a Bollinger Bands breakout strategy:

from pyeventbt import (
    Strategy,
    BarEvent,
    SignalEvent,
    Modules,
    StrategyTimeframes,
    PassthroughRiskConfig,
    MinSizingConfig
)
from pyeventbt.events.events import OrderType, SignalType
from pyeventbt.indicators.indicators import BollingerBands
from pyeventbt.strategy.core.account_currencies import AccountCurrencies

from datetime import datetime, time
from decimal import Decimal
import logging
import numpy as np

logger = logging.getLogger("pyeventbt")

# Strategy Configuration
strategy_id = "1234"
strategy = Strategy(logging_level=logging.INFO)

# Timeframes
signal_timeframe = StrategyTimeframes.ONE_HOUR
daily_timeframe = StrategyTimeframes.ONE_DAY

strategy_timeframes = [signal_timeframe, daily_timeframe]

# Trading Configuration
symbols_to_trade = ['EURUSD']
starting_capital = 100000

# Strategy Parameters
bb_period = 20
bb_std_dev = 2.5
close_hour = 21
close_minute = 0
order_placement_hour = 8
order_placement_minute = 0

# Daily tracking
orders_placed_today: dict[str, bool] = {symbol: False for symbol in symbols_to_trade}
current_trading_date: dict[str, datetime|None] = {symbol: None for symbol in symbols_to_trade}


@strategy.custom_signal_engine(strategy_id=strategy_id, strategy_timeframes=strategy_timeframes)
def bbands_breakout(event: BarEvent, modules: Modules):
    """
    Bollinger Bands Breakout Strategy:
    - Breakout levels: Upper and Lower Bollinger Bands
    - Exit: Close all at 21:00
    """
    
    symbol = event.symbol
    signal_events = []
    
    # Get current time and date
    current_time = event.datetime.time()
    current_date = event.datetime.date()
    
    # Reset daily tracking if new day
    if current_trading_date[symbol] != current_date:
        current_trading_date[symbol] = current_date
        orders_placed_today[symbol] = False
    
    # Get positions and orders
    open_positions = modules.PORTFOLIO.get_number_of_strategy_open_positions_by_symbol(symbol)
    pending_orders = modules.PORTFOLIO.get_number_of_strategy_pending_orders_by_symbol(symbol)
    
    # Close positions and cancel orders at close time
    if current_time >= time(close_hour, close_minute):
        if open_positions['TOTAL'] > 0:
            logger.info(f"{event.datetime} - Closing all positions for {symbol}")
            modules.EXECUTION_ENGINE.close_all_strategy_positions()
        
        if pending_orders['TOTAL'] > 0:
            logger.info(f"{event.datetime} - Cancelling all pending orders for {symbol}")
            modules.EXECUTION_ENGINE.cancel_all_strategy_pending_orders()
        
        return
    
    # Place orders at order placement time
    if (current_time >= time(order_placement_hour, order_placement_minute) and 
        not orders_placed_today[symbol] and 
        pending_orders['TOTAL'] == 0 and
        event.timeframe == signal_timeframe):
        
        # Get bars for Bollinger Bands calculation
        bars_needed = bb_period + 10
        indicator_bars = modules.DATA_PROVIDER.get_latest_bars(symbol, signal_timeframe, bars_needed)
        
        if indicator_bars is None or indicator_bars.height < bars_needed:
            return
        
        # Calculate Bollinger Bands
        close = indicator_bars.select('close').to_numpy().flatten()
        upper, middle, lower = BollingerBands.compute(close, bb_period, bb_std_dev)
        
        current_upper = upper[-1]
        current_lower = lower[-1]
        
        if np.isnan(current_upper) or np.isnan(current_lower):
            return
        
        upper_breakout = Decimal(str(current_upper))
        lower_breakout = Decimal(str(current_lower))
        
        # Time for signal generation
        if modules.TRADING_CONTEXT == "BACKTEST":
            time_generated = event.datetime + signal_timeframe.to_timedelta()
        else:
            time_generated = datetime.now()
        
        # Place BUY STOP order
        signal_events.append(SignalEvent(
            symbol=symbol,
            time_generated=time_generated,
            strategy_id=strategy_id,
            signal_type=SignalType.BUY,
            order_type=OrderType.STOP,
            order_price=upper_breakout,
            sl=Decimal(str(0.0)),
            tp=Decimal(str(0.0)),
        ))
        
        # Place SELL STOP order
        signal_events.append(SignalEvent(
            symbol=symbol,
            time_generated=time_generated,
            strategy_id=strategy_id,
            signal_type=SignalType.SELL,
            order_type=OrderType.STOP,
            order_price=lower_breakout,
            sl=Decimal(str(0.0)),
            tp=Decimal(str(0.0)),
        ))
        
        orders_placed_today[symbol] = True
    
    return signal_events


# Configure Strategy
strategy.configure_predefined_sizing_engine(MinSizingConfig())
strategy.configure_predefined_risk_engine(PassthroughRiskConfig())

# Backtest Configuration
from_date = datetime(year=2020, month=1, day=1)
to_date = datetime(year=2023, month=12, day=1)
# csv_dir = '/Users/marticastany/Desktop/long_data' # Change it with your own path to the CSV data
csv_dir = None # If you don't have CSV data, you can set this to None

# Launch Backtest
backtest = strategy.backtest(
    strategy_id=strategy_id,
    initial_capital=starting_capital,
    symbols_to_trade=symbols_to_trade,
    csv_dir=csv_dir,
    backtest_name=strategy_id,
    start_date=from_date,
    end_date=to_date,
    export_backtest_csv=True,
    export_backtest_parquet=False,
    account_currency=AccountCurrencies.USD
)

print("Backtest finished")
backtest.plot()

How PyEventBT Compares

The Python ecosystem has several excellent backtesting frameworks, each with different strengths. Here's how they differ:

PyEventBT Backtrader Zipline VectorBT
Architecture Event-driven Event-driven Event-driven Vectorized
MetaTrader 5 integration Native
Live trading Yes (via MT5) Community brokers
Same code backtest & live Yes (full MT5 API mock) Partial
Language Python only Python only Python only Python only
Look-ahead bias prevention By design By design By design Manual
Multi-timeframe Yes Yes Limited Yes
Pending orders (Limit/Stop) Yes Yes Limited

When to choose PyEventBT: You trade on MetaTrader 5 and want to develop, backtest, and deploy strategies entirely in Python — using the same code and the same MT5 API in both backtesting and production, with no MQL5 required.

When to choose another framework: If you don't use MetaTrader 5 as your broker, or if you need vectorized backtesting for high-frequency tick-level research, other frameworks may be a better fit.

Documentation

Full documentation available at pyeventbt.com.

The documentation includes:

  • Complete API reference
  • Detailed examples and tutorials
  • Import patterns and best practices
  • Advanced configuration options
  • Live trading setup guides

For LLM-friendly documentation, see llms.txt and llms-full.txt.

Features

  • Event-driven architecture for realistic backtesting
  • Built-in technical indicators (ATR, SMA, RSI, and more)
  • Seamless transition from backtest to live trading
  • Comprehensive performance metrics and visualization
  • Flexible risk and position sizing engines
  • MetaTrader 5 integration for live trading

License

Apache 2.0

Author

Made with ❤️ for the Community by Martí Castany

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

pyeventbt-0.0.8.tar.gz (10.3 MB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

pyeventbt-0.0.8-py3-none-any.whl (10.8 MB view details)

Uploaded Python 3

File details

Details for the file pyeventbt-0.0.8.tar.gz.

File metadata

  • Download URL: pyeventbt-0.0.8.tar.gz
  • Upload date:
  • Size: 10.3 MB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: poetry/2.3.4 CPython/3.12.13 Linux/6.17.0-1010-azure

File hashes

Hashes for pyeventbt-0.0.8.tar.gz
Algorithm Hash digest
SHA256 452e42d85c64fa3d1b004598f60ec9be815a8be1142b2bf1a4babaf48cf03707
MD5 c48798523858c1e52f3f84c7933859cd
BLAKE2b-256 125b7893af877e469151459fb86a1245277ae49edc6cbd60a2ceb373ebbe6802

See more details on using hashes here.

File details

Details for the file pyeventbt-0.0.8-py3-none-any.whl.

File metadata

  • Download URL: pyeventbt-0.0.8-py3-none-any.whl
  • Upload date:
  • Size: 10.8 MB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: poetry/2.3.4 CPython/3.12.13 Linux/6.17.0-1010-azure

File hashes

Hashes for pyeventbt-0.0.8-py3-none-any.whl
Algorithm Hash digest
SHA256 4455a65de9d69a706ba0e109746af5cb0a44a844d571635ba113e7ea7b500410
MD5 fcb5cd467fe09fc12059cada0d86964c
BLAKE2b-256 8a72de4d767cb9c30ef4e2a356eb087e339d20a62c55d3c5a570a0323202464e

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page