Skip to content

quantex.execution

quantex.execution — Execution utilities.

This module implements ImmediateFillSimulator, a zero-latency execution engine used by the Quantex backtesting framework.

ImmediateFillSimulator converts strategy-generated quantex.models.Order objects directly into quantex.models.Fill objects at the supplied execution price, then updates the provided quantex.models.Portfolio so metrics such as positions, cash, and equity remain consistent during a backtest.

Key characteristics
  • Immediate fills – no market latency, orders are filled in the same timestamp they are received.
  • Configurable commission – per-fill commission (default 0.0) to approximate transaction costs. Slippage is currently assumed to be zero but can be incorporated in future extensions.

This concise implementation serves as a reference template; more sophisticated execution models (e.g., with latency, partial fills, or order books) can inherit from or replace this class in future versions.

ImmediateFillSimulator

Immediate (zero-latency) execution simulator.

Converts incoming quantex.models.Order objects into quantex.models.Fill objects instantly at the supplied execution price. Commission and slippage are currently fixed to zero but the constructor exposes a commission argument so users can override the default behaviour in future iterations.

Source code in src/quantex/execution.py 
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
class ImmediateFillSimulator:
    """Immediate (zero-latency) execution simulator.

    Converts incoming `quantex.models.Order` objects into
    `quantex.models.Fill` objects *instantly* at the supplied execution
    price. Commission and slippage are currently fixed to *zero* but the
    constructor exposes a *commission* argument so users can override the
    default behaviour in future iterations.
    """

    def __init__(
        self,
        portfolio: Portfolio,
        commission: float = 0.0,
        min_holding_period: pd.Timedelta | None = None,
    ) -> None:
        """Initializes the ImmediateFillSimulator.

        Args:
            portfolio: The portfolio to update upon fill.
            commission: The commission to apply to each fill.
        """
        self.portfolio = portfolio
        self.commission = commission
        self.min_holding_period = min_holding_period

    def execute(
        self, order: Order, execution_price: float, timestamp: datetime
    ) -> Fill:
        """Converts order into a Fill and updates the portfolio.

        Args:
            order: The order generated by the strategy.
            execution_price: The price used for the fill.
            timestamp: Timestamp for the resulting `quantex.models.Fill`.

        Returns:
            The created Fill object.
        """

        # --- Optional minimum holding period enforcement --------------
        if (
            self.min_holding_period is not None
            and order.symbol in self.portfolio.positions
        ):
            pos = self.portfolio.positions[order.symbol]
            if pos.open_timestamp is not None:
                holding_time = timestamp - pos.open_timestamp
                # Closing or reducing existing exposure before min period?
                if (
                    (order.side == "sell" and pos.position > 0)
                    or (order.side == "buy" and pos.position < 0)
                ) and holding_time < self.min_holding_period:
                    raise ValueError(
                        "Order violates minimum holding period: "
                        f"held for {holding_time}, required {self.min_holding_period}."
                    )

        if order.side == "buy":
            required_cash = order.quantity * execution_price + self.commission
            if required_cash > self.portfolio.cash:
                raise ValueError(
                    "Insufficient cash to execute buy order: "
                    f"required={required_cash:.2f}, available={self.portfolio.cash:.2f}"
                )

        signed_qty = order.quantity if order.side == "buy" else -order.quantity

        fill = Fill(
            order_id=order.id,
            symbol=order.symbol,
            quantity=signed_qty,
            price=execution_price,
            timestamp=timestamp,
            commission=self.commission,
        )

        # Immediately reflect the fill in the portfolio.
        self.portfolio.process_fill(fill)
        return fill

__init__(portfolio, commission=0.0, min_holding_period=None)

Initializes the ImmediateFillSimulator.

Parameters:

Name Type Description Default
portfolio Portfolio

The portfolio to update upon fill.

 required
commission float

The commission to apply to each fill.

 0.0
Source code in src/quantex/execution.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def __init__(
    self,
    portfolio: Portfolio,
    commission: float = 0.0,
    min_holding_period: pd.Timedelta | None = None,
) -> None:
    """Initializes the ImmediateFillSimulator.

    Args:
        portfolio: The portfolio to update upon fill.
        commission: The commission to apply to each fill.
    """
    self.portfolio = portfolio
    self.commission = commission
    self.min_holding_period = min_holding_period

execute(order, execution_price, timestamp)

Converts order into a Fill and updates the portfolio.

Parameters:

Name Type Description Default
order Order

The order generated by the strategy.

 required
execution_price float

The price used for the fill.

 required
timestamp datetime

Timestamp for the resulting quantex.models.Fill.

 required

Returns:

Type Description
Fill

The created Fill object.
Source code in src/quantex/execution.py 
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def execute(
    self, order: Order, execution_price: float, timestamp: datetime
) -> Fill:
    """Converts order into a Fill and updates the portfolio.

    Args:
        order: The order generated by the strategy.
        execution_price: The price used for the fill.
        timestamp: Timestamp for the resulting `quantex.models.Fill`.

    Returns:
        The created Fill object.
    """

    # --- Optional minimum holding period enforcement --------------
    if (
        self.min_holding_period is not None
        and order.symbol in self.portfolio.positions
    ):
        pos = self.portfolio.positions[order.symbol]
        if pos.open_timestamp is not None:
            holding_time = timestamp - pos.open_timestamp
            # Closing or reducing existing exposure before min period?
            if (
                (order.side == "sell" and pos.position > 0)
                or (order.side == "buy" and pos.position < 0)
            ) and holding_time < self.min_holding_period:
                raise ValueError(
                    "Order violates minimum holding period: "
                    f"held for {holding_time}, required {self.min_holding_period}."
                )

    if order.side == "buy":
        required_cash = order.quantity * execution_price + self.commission
        if required_cash > self.portfolio.cash:
            raise ValueError(
                "Insufficient cash to execute buy order: "
                f"required={required_cash:.2f}, available={self.portfolio.cash:.2f}"
            )

    signed_qty = order.quantity if order.side == "buy" else -order.quantity

    fill = Fill(
        order_id=order.id,
        symbol=order.symbol,
        quantity=signed_qty,
        price=execution_price,
        timestamp=timestamp,
        commission=self.commission,
    )

    # Immediately reflect the fill in the portfolio.
    self.portfolio.process_fill(fill)
    return fill

NextBarSimulator

Bases: ImmediateFillSimulator

Execution simulator that fills market orders at the next bar.

Orders received during bar t are queued and executed at the beginning of bar t+1 using that bar's open price.

The EventBus calls :pycode{flush_pending()} before it injects the current bar's market snapshot into the strategy. This ensures that fills appear in the portfolio before the strategy logic for the new bar runs – matching the behaviour of many off-the-shelf backtesting libraries where orders are assumed to execute on the next bar's open.

Source code in src/quantex/execution.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
class NextBarSimulator(ImmediateFillSimulator):
    """Execution simulator that fills market orders at the *next* bar.

    Orders received during bar *t* are **queued** and executed at the **beginning
    of bar *t+1*** using that bar's **open** price.

    The EventBus calls :pycode{flush_pending()} before it injects the current
    bar's market snapshot into the strategy. This ensures that fills appear in
    the portfolio *before* the strategy logic for the new bar runs – matching
    the behaviour of many off-the-shelf backtesting libraries where orders are
    assumed to execute on the next bar's open.
    """

    def __init__(
        self,
        portfolio: Portfolio,
        commission: float = 0.0,
        min_holding_period: pd.Timedelta | None = None,
        *,
        fill_at: str = "open",
    ) -> None:
        if fill_at not in {"open", "close"}:
            raise ValueError("fill_at must be 'open' or 'close'")

        super().__init__(portfolio, commission, min_holding_period)
        self._pending_orders: list[Order] = []
        self._fill_at = fill_at
        self.event_bus: EventBus | None = None

    # ------------------------------------------------------------------
    # Public API – called by EventBus for *new* orders in bar t
    # ------------------------------------------------------------------
    def execute(
        self, order: Order, execution_price: float, timestamp: datetime
    ) -> None:  # type: ignore[override]
        """Queues *order* for execution on the following bar.

        Unlike :class:`ImmediateFillSimulator`, this method **does not** return
        a :class:`quantex.models.Fill` immediately. The fill will be generated
        by :meth:`flush_pending` when the next timestamp is processed.
        """

        # Store the original *Order* object – we intentionally ignore the
        # *execution_price* passed in for the current bar because the trade
        # should fill at the next bar's price.
        self._pending_orders.append(order)
        # Returning *None* signals the EventBus that nothing was filled yet.
        return None  # type: ignore[return-value]

    # ------------------------------------------------------------------
    # Internal API – invoked by EventBus at the start of bar t+1
    # ------------------------------------------------------------------
    def flush_pending(
        self,
        timestamp: datetime,
        price_row: Sequence[float],
        symbol_idx: dict[str, int],
    ) -> list[Fill]:
        """Converts *all* queued orders into fills at *timestamp*.

        Args:
            timestamp: The timestamp that will be assigned to each fill
                (i.e. the *current* bar processed by the EventBus).
            price_row: Numpy-converted prices for the current bar **aligned**
                to ``symbol_idx`` – obtained directly from the EventBus.
            symbol_idx: Mapping symbol→column index allowing O(1) look-ups.

        Returns:
            List of created :class:`quantex.models.Fill` objects.
        """

        fills: list[Fill] = []

        # We iterate over a *copy* of the list so we can clear it safely even
        # if an exception is raised inside ``super().execute``.
        pending, self._pending_orders = self._pending_orders, []

        for order in pending:
            # Prefer the *open* price for the current bar. We fetch it from
            # the underlying DataSource via the EventBus reference.
            event_bus: EventBus | None = self.event_bus
            ds: BacktestingDataSource | None = None
            if event_bus is not None:
                ds = event_bus.data_sources.get(order.symbol)

            if ds is not None:
                raw = ds.get_raw_data()
                if timestamp not in raw.index:
                    # In case of missing timestamp fallback to close via price_row
                    ds = None
                else:
                    if self._fill_at == "open":
                        if "open" in raw.columns:
                            execution_price = float(raw.at[timestamp, "open"])
                        else:
                            execution_price = float(raw.at[timestamp, "close"])
                    else:  # fill_at == "close"
                        execution_price = float(raw.at[timestamp, "close"])
            else:
                # Fallback to price_row close price when symbol missing or ds None
                idx = symbol_idx.get(order.symbol)
                if idx is None:
                    continue
                execution_price = float(price_row[idx])

            # If price moved unfavourably we may not have enough cash for the
            # original quantity (sized with yesterday's close). Adjust down.
            if order.side == "buy":
                max_affordable = math.floor(self.portfolio.cash / execution_price)
                if max_affordable == 0:
                    # Skip order entirely – cannot afford even 1 share
                    continue
                if max_affordable < order.quantity:
                    order.quantity = max_affordable  # type: ignore[assignment]

            fill = super().execute(order, execution_price, timestamp)
            fills.append(fill)

        return fills

execute(order, execution_price, timestamp)

Queues order for execution on the following bar.

Unlike :class:ImmediateFillSimulator, this method does not return a :class:quantex.models.Fill immediately. The fill will be generated by :meth:flush_pending when the next timestamp is processed.

Source code in src/quantex/execution.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
def execute(
    self, order: Order, execution_price: float, timestamp: datetime
) -> None:  # type: ignore[override]
    """Queues *order* for execution on the following bar.

    Unlike :class:`ImmediateFillSimulator`, this method **does not** return
    a :class:`quantex.models.Fill` immediately. The fill will be generated
    by :meth:`flush_pending` when the next timestamp is processed.
    """

    # Store the original *Order* object – we intentionally ignore the
    # *execution_price* passed in for the current bar because the trade
    # should fill at the next bar's price.
    self._pending_orders.append(order)
    # Returning *None* signals the EventBus that nothing was filled yet.
    return None  # type: ignore[return-value]

flush_pending(timestamp, price_row, symbol_idx)

Converts all queued orders into fills at timestamp.

Parameters:

Name Type Description Default
timestamp datetime

The timestamp that will be assigned to each fill (i.e. the current bar processed by the EventBus).

 required
price_row Sequence[float]

Numpy-converted prices for the current bar aligned to symbol_idx – obtained directly from the EventBus.

 required
symbol_idx dict[str, int]

Mapping symbol→column index allowing O(1) look-ups.

 required

Returns:

Type Description
list[Fill]

List of created :class:quantex.models.Fill objects.
Source code in src/quantex/execution.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
def flush_pending(
    self,
    timestamp: datetime,
    price_row: Sequence[float],
    symbol_idx: dict[str, int],
) -> list[Fill]:
    """Converts *all* queued orders into fills at *timestamp*.

    Args:
        timestamp: The timestamp that will be assigned to each fill
            (i.e. the *current* bar processed by the EventBus).
        price_row: Numpy-converted prices for the current bar **aligned**
            to ``symbol_idx`` – obtained directly from the EventBus.
        symbol_idx: Mapping symbol→column index allowing O(1) look-ups.

    Returns:
        List of created :class:`quantex.models.Fill` objects.
    """

    fills: list[Fill] = []

    # We iterate over a *copy* of the list so we can clear it safely even
    # if an exception is raised inside ``super().execute``.
    pending, self._pending_orders = self._pending_orders, []

    for order in pending:
        # Prefer the *open* price for the current bar. We fetch it from
        # the underlying DataSource via the EventBus reference.
        event_bus: EventBus | None = self.event_bus
        ds: BacktestingDataSource | None = None
        if event_bus is not None:
            ds = event_bus.data_sources.get(order.symbol)

        if ds is not None:
            raw = ds.get_raw_data()
            if timestamp not in raw.index:
                # In case of missing timestamp fallback to close via price_row
                ds = None
            else:
                if self._fill_at == "open":
                    if "open" in raw.columns:
                        execution_price = float(raw.at[timestamp, "open"])
                    else:
                        execution_price = float(raw.at[timestamp, "close"])
                else:  # fill_at == "close"
                    execution_price = float(raw.at[timestamp, "close"])
        else:
            # Fallback to price_row close price when symbol missing or ds None
            idx = symbol_idx.get(order.symbol)
            if idx is None:
                continue
            execution_price = float(price_row[idx])

        # If price moved unfavourably we may not have enough cash for the
        # original quantity (sized with yesterday's close). Adjust down.
        if order.side == "buy":
            max_affordable = math.floor(self.portfolio.cash / execution_price)
            if max_affordable == 0:
                # Skip order entirely – cannot afford even 1 share
                continue
            if max_affordable < order.quantity:
                order.quantity = max_affordable  # type: ignore[assignment]

        fill = super().execute(order, execution_price, timestamp)
        fills.append(fill)

    return fills