MA Crossover

A classic trend-following strategy that goes long when a fast SMA crosses above a slow SMA, and short when it crosses below. This is the simplest complete strategy you can build with PyEventBT — a great starting point.

Difficulty

Beginner

Timeframe

Daily (D1)

Order Type

Market Orders

Concepts Covered

Signal generation, position management, indicator calculation, backtesting

Strategy Logic

1. Calculate a fast SMA (10-period) and a slow SMA (30-period) on daily closes
2. If fast SMA > slow SMA → we want to be long
3. If fast SMA < slow SMA → we want to be short
4. Before opening a new position, close any opposite position first
5. Only hold one position at a time

Step-by-Step Walkthrough

1. Imports and setup

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

Every PyEventBT strategy needs these core imports:

• Strategy — the main class you instantiate and configure
• BarEvent / Modules — passed to your signal engine on every bar
• SignalEvent — the object you return to generate a trade signal
• StrategyTimeframes — enum of supported timeframes (ONE_MIN, ONE_HOUR, ONE_DAY, etc.)
• SignalType / OrderType — BUY/SELL and MARKET/LIMIT/STOP constants

2. Configuration

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

signal_timeframe = StrategyTimeframes.ONE_DAY
strategy_timeframes = [signal_timeframe]

symbols_to_trade = ['EURUSD']
starting_capital = 100000

fast_ma_period = 10
slow_ma_period = 30

The strategy_timeframes list tells PyEventBT which bar timeframes to feed into your signal engine. Here we only need daily bars.

3. Signal engine decorator

@strategy.custom_signal_engine(
    strategy_id=strategy_id,
    strategy_timeframes=strategy_timeframes
)
def ma_crossover_strategy(event: BarEvent, modules: Modules):

The @strategy.custom_signal_engine decorator registers your function as the signal generator. PyEventBT calls it on every new bar for the timeframes you specify. The function receives:

• event — the bar that just completed (symbol, datetime, timeframe, OHLCV data)
• modules — access to DATA_PROVIDER, PORTFOLIO, EXECUTION_ENGINE, and TRADING_CONTEXT

4. Fetching data and computing indicators

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  # Not enough history yet

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)

Data Format

get_latest_bars() returns a Polars DataFrame. Use .select('close').to_numpy().flatten() to extract a NumPy array that you can pass directly to any indicator.

5. Position management

open_positions = modules.PORTFOLIO \
    .get_number_of_strategy_open_positions_by_symbol(symbol)

This returns a dict with keys 'LONG', 'SHORT', and 'TOTAL'. Always check your current positions before generating a signal — this prevents duplicate entries and ensures you close the opposite side before flipping direction.

# Go long: close any short, then buy
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

6. Signal generation

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'),
))

Signal Timing

In backtesting, set time_generated to the next bar’s open using event.datetime + signal_timeframe.to_timedelta(). This avoids look-ahead bias — the signal is generated at bar close but executed at the next bar’s open price.

7. Sizing, risk, and running the backtest

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

backtest = strategy.backtest(
    strategy_id=strategy_id,
    initial_capital=starting_capital,
    symbols_to_trade=symbols_to_trade,
    csv_dir=None,  # Uses built-in sample data
    ...
)
backtest.plot()

Set csv_dir=None to use PyEventBT’s built-in sample dataset — perfect for testing. When you have your own data, point it to a directory containing CSV files named by symbol (e.g., EURUSD.csv).

Adapt This Strategy

Here are a few ideas to experiment with:

Variation	What to change
Different MA periods	Change fast_ma_period and slow_ma_period
Use EMA instead of SMA	Replace SMA import with EMA from pyeventbt.indicators
Add a stop-loss	Set the sl parameter in SignalEvent to a non-zero Decimal
Trade multiple symbols	Add symbols to symbols_to_trade list
Shorter timeframe	Change signal_timeframe to StrategyTimeframes.ONE_HOUR

Going Live

The same strategy code works for live trading — just replace the strategy.backtest(...) call:

from pyeventbt import Mt5PlatformConfig

mt5_config = Mt5PlatformConfig(
    path="C:\\Program Files\\MetaTrader 5\\terminal64.exe",
    login=12345,
    password="your_password",
    server="YourBroker-Demo",
    timeout=60000,
    portable=False
)

strategy.run_live(
    mt5_configuration=mt5_config,
    strategy_id=strategy_id,
    initial_capital=100000,
    symbols_to_trade=symbols_to_trade,
    heartbeat=0.1
)

Live Trading Requirements

Live trading requires MetaTrader 5 installed and running on Windows. Always test on a demo account first.

Full Code

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

from pyeventbt import (
    Strategy,
    BarEvent,
    SignalEvent,
    Modules,
    StrategyTimeframes,
    PassthroughRiskConfig,
    MinSizingConfig
)
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")

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

signal_timeframe = StrategyTimeframes.ONE_DAY
strategy_timeframes = [signal_timeframe]

symbols_to_trade = ['EURUSD']
starting_capital = 100000

# Strategy Parameters
fast_ma_period = 10
slow_ma_period = 30


# ── Signal Engine ────────────────────────────────────────────────
@strategy.custom_signal_engine(
    strategy_id=strategy_id,
    strategy_timeframes=strategy_timeframes
)
def ma_crossover_strategy(event: BarEvent, modules: Modules):
    """
    Stay long while fast MA > slow MA.
    Stay short while fast MA < slow MA.
    Always maintain at most one open position.
    """

    if event.timeframe != signal_timeframe:
        return

    symbol = event.symbol
    signal_events = []

    # Get enough bars to compute the slow MA
    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  # Not enough history yet

    # Calculate indicators
    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)

    current_fast = fast_ma[-1]
    current_slow = slow_ma[-1]

    # Determine desired position
    if current_fast > current_slow:
        desired_position = "LONG"
    elif current_fast < current_slow:
        desired_position = "SHORT"
    else:
        return  # MAs are equal — do nothing

    # Check current positions
    open_positions = modules.PORTFOLIO \
        .get_number_of_strategy_open_positions_by_symbol(symbol)

    signal_type = None

    # Go long: close any short, then buy
    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

    # Go short: close any long, then sell
    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  # Already in the desired position

    # Set execution time for the 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)

    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


# ── Sizing & Risk ────────────────────────────────────────────────
strategy.configure_predefined_sizing_engine(MinSizingConfig())
strategy.configure_predefined_risk_engine(PassthroughRiskConfig())


# ── Run Backtest ─────────────────────────────────────────────────
backtest = strategy.backtest(
    strategy_id=strategy_id,
    initial_capital=starting_capital,
    symbols_to_trade=symbols_to_trade,
    csv_dir=None,          # None uses built-in sample data
    backtest_name=strategy_id,
    start_date=datetime(2020, 1, 1),
    end_date=datetime(2023, 12, 1),
    export_backtest_parquet=False,
    account_currency='USD'
)

backtest.plot()