Skip to content

quantex.strategy

Strategy

Bases: ABC

Base class for all trading strategies.

This class owns a quantex.models.Portfolio instance which keeps track of cash and Position objects. For convenience and backward compatibility, the underlying positions mapping is exposed directly so that existing strategy implementations that reference self.positions[<symbol>] continue to work unchanged.

Source code in src/quantex/strategy.py 
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 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
119
120
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
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
class Strategy(ABC):
    """Base class for all trading strategies.

    This class owns a `quantex.models.Portfolio` instance which keeps
    track of cash and `Position` objects. For convenience and backward
    compatibility, the underlying `positions` mapping is exposed directly so
    that existing strategy implementations that reference
    `self.positions[<symbol>]` continue to work unchanged.
    """

    # The engine (or outer loop) advances *all* data sources by calling
    # :pycode{_increment_index} on them **and** on the strategy to allow for
    # strategy-level book-keeping.
    index: int = 0
    timestamp: datetime | None = None

    def __init__(
        self,
        data_sources: Mapping[str, DataSource] | None = None,
        symbols: list[str] | None = None,
        *,
        initial_cash: float = 0.0,
    ) -> None:
        """Initializes a new strategy instance.

        Args:
            data_sources: Mapping from a source name to a concrete
                `quantex.sources.DataSource` implementation.
            symbols: List of tradable symbols to initialize `Position` objects
                for. If None, positions are created lazily.
            initial_cash: Starting cash for the internal `Portfolio`.
        """

        # Store references to market data sources (may be empty – engine injects)
        self.data_sources = data_sources or {}
        self.timestamp = None

        # Maintain a portfolio to aggregate cash & PnL
        self.portfolio: Portfolio = Portfolio(cash=initial_cash)

        # Expose positions dict for backward compatibility
        self.positions = self.portfolio.positions

        # Register provided symbols (if any)
        self.symbols = symbols or []
        for sym in self.symbols:
            # Pre-create empty Position objects so that strategy code can rely
            # on their existence.
            if sym not in self.positions:
                self.positions[sym] = Position(sym)

        # Queue of orders submitted during the current bar – cleared each step
        self._pending_orders: list[Order] = []
        self.signals = pd.DataFrame()

        # --- Internals set by the engine each bar ---
        self._price_row: list[float] | None = None
        self._symbols: list[str] | None = None
        self._symbol_idx: dict[str, int] | None = None

    # ------------------------------------------------------------------
    # Convenience Properties
    # ------------------------------------------------------------------

    @property
    def cash(self) -> float:
        """Current available cash in the underlying Portfolio.

        Example:
            ```python
            if self.cash > 10_000:
                self.buy("AAPL", 5)
            ```

        Returns:
            float: The cash balance that can be deployed for new positions.
        """

        return self.portfolio.cash

    def _increment_index(self) -> None:
        """Advances the internal bar pointer by one.

        This should be called by the backtesting engine after all logic
        for the current bar has executed.
        """
        self.index += 1

    @abstractmethod
    def run(self):
        """Executes the strategy logic for the current bar.

        Concrete strategies must override this method. It should inspect
        the available data sources and make trading decisions.
        """
        raise NotImplementedError("Subclasses must implement this method.")

    def buy(
        self, symbol: str, quantity: float, limit_price: float | None = None
    ) -> None:
        """Creates and submits a buy order using the current strategy timestamp.

        Args:
            symbol: The symbol to buy.
            quantity: The quantity to buy. Must be positive.
            limit_price: If provided, a limit order is created. Otherwise, a
                market order is created.
        """
        if self.timestamp is None:
            raise RuntimeError("Cannot place order: strategy timestamp is not set.")

        order = Order(
            id=f"buy-{symbol}-{self.timestamp}",
            symbol=symbol,
            side="buy",
            quantity=quantity,
            order_type="limit" if limit_price else "market",
            limit_price=limit_price,
            timestamp=self.timestamp,
        )
        self.submit_order(order)

    def sell(
        self, symbol: str, quantity: float, limit_price: float | None = None
    ) -> None:
        """Creates and submits a sell order using the current strategy timestamp.

        Args:
            symbol: The symbol to sell.
            quantity: The quantity to sell. Must be positive.
            limit_price: If provided, a limit order is created. Otherwise, a
                market order is created.
        """
        if self.timestamp is None:
            raise RuntimeError("Cannot place order: strategy timestamp is not set.")

        order = Order(
            id=f"sell-{symbol}-{self.timestamp}",
            symbol=symbol,
            side="sell",
            quantity=quantity,
            order_type="limit" if limit_price else "market",
            limit_price=limit_price,
            timestamp=self.timestamp,
        )
        self.submit_order(order)

    def close_position(self, symbol: str) -> None:
        """Creates and submits an order to close the entire position for a symbol.

        This is a helper method that checks if a position is open for the
        given symbol and, if so, creates a market order to close it at the
        current strategy timestamp.

        Args:
            symbol: The symbol of the position to close.
        """
        if self.timestamp is None:
            raise RuntimeError("Cannot place order: strategy timestamp is not set.")

        position = self.positions.get(symbol)
        if not position or position.is_closed:
            return

        if position.is_long:
            side = "sell"
            quantity = position.position
        else:  # is_short
            side = "buy"
            quantity = abs(position.position)

        order = Order(
            id=f"close-{symbol}-{self.timestamp}",
            symbol=symbol,
            side=side,
            quantity=quantity,
            timestamp=self.timestamp,
        )
        self.submit_order(order)

    def submit_order(self, order: Order) -> None:
        """Queues an order to be executed by the engine.

        Strategies should call this method to simulate realistic order routing.

        Args:
            order: The `Order` to be submitted.
        """

        self._pending_orders.append(order)

    def _pop_pending_orders(self) -> list[Order]:
        """Returns and clears the list of queued orders. For internal use."""
        orders, self._pending_orders = self._pending_orders, []
        return orders

    def get_price(self, symbol: str) -> float:
        """Returns the latest price for *symbol*.

        This accesses the numpy price row injected by the engine and is
        therefore O(1) without any pandas overhead. Raises ``KeyError`` if the
        symbol is not part of the backtest universe.
        """

        if self._price_row is None or self._symbol_idx is None:
            raise RuntimeError("Market data not yet initialised for this bar.")
        idx = self._symbol_idx.get(symbol)
        if idx is None:
            raise KeyError(symbol)
        return float(self._price_row[idx])

    @property
    def prices(self) -> dict[str, float]:
        """Lazy dict of symbol→price for the current bar (lightweight)."""

        if self._price_row is None or self._symbols is None:
            raise RuntimeError("Market data not yet initialised for this bar.")
        return {sym: float(price) for sym, price in zip(self._symbols, self._price_row)}

    def _get_price_history_df(self) -> pd.DataFrame:
        """Internal helper to fetch the full, forward-filled price DataFrame.

        Returns:
            pd.DataFrame: Price data for *all* symbols in the backtest universe
                (columns) indexed by the global event timeline. Values are
                forward-filled by the engine ensuring there are no missing
                timestamps – ideal for multi-asset lookback computations.

        Raises:
            RuntimeError: If the strategy is not attached to an *EventBus* or
                if the price DataFrame has not been initialised yet (i.e. the
                first bar has not been processed).
        """

        event_bus = getattr(self, "event_bus", None)
        if event_bus is None:
            raise RuntimeError(
                "Strategy is not attached to an EventBus; price history unavailable."
            )

        price_df: pd.DataFrame | None = (
            event_bus._price_df
        )  # pylint: disable=protected-access
        if price_df is None:
            raise RuntimeError(
                "Price history not yet initialised – wait until the first bar has been processed."
            )

        return price_df

    @property
    def price_history(self) -> pd.DataFrame:  # noqa: D401 – property describing data
        """Price history up to *and including* the current bar.

        Example:
            > history = self.price_history
            > latest_btc = history["BTC"].iloc[-1]
        """

        df = self._get_price_history_df()
        # ``self.index`` corresponds to the *current* bar position
        return df.iloc[: self.index + 1]

    def get_lookback_prices(self, lookback_period: int) -> pd.DataFrame:
        """Returns an *aligned* lookback window for all symbols.

        Args:
            lookback_period: Number of bars (inclusive of the current bar) to
                return.

        Returns:
            pd.DataFrame: DataFrame with the last *lookback_period* rows from
            :pyattr:`price_history`. If there are fewer than *lookback_period*
            observations available (e.g. at the beginning of a backtest), the
            entire available history is returned instead. The returned frame
            is guaranteed to be free of missing timestamps across symbols due
            to the forward-filling performed by the engine.
        """

        history = self.price_history
        if len(history) < lookback_period:
            return history.copy()
        return history.iloc[-lookback_period:].copy()

    # Called by EventBus – should be considered *private*
    def _update_market_data(self, price_row, symbols, symbol_idx):
        self._price_row = price_row
        self._symbols = symbols
        self._symbol_idx = symbol_idx

cash property

Current available cash in the underlying Portfolio.

Example 
if self.cash > 10_000:
    self.buy("AAPL", 5)

Returns:

Name Type Description
float float

The cash balance that can be deployed for new positions.

price_history property

Price history up to and including the current bar.

Example

history = self.price_history latest_btc = history["BTC"].iloc[-1]

prices property

Lazy dict of symbol→price for the current bar (lightweight).

__init__(data_sources=None, symbols=None, *, initial_cash=0.0)

Initializes a new strategy instance.

Parameters:

Name Type Description Default
data_sources Mapping[str, DataSource] | None

Mapping from a source name to a concrete quantex.sources.DataSource implementation.

 None
symbols list[str] | None

List of tradable symbols to initialize Position objects for. If None, positions are created lazily.

 None
initial_cash float

Starting cash for the internal Portfolio.

 0.0
Source code in src/quantex/strategy.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
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
def __init__(
    self,
    data_sources: Mapping[str, DataSource] | None = None,
    symbols: list[str] | None = None,
    *,
    initial_cash: float = 0.0,
) -> None:
    """Initializes a new strategy instance.

    Args:
        data_sources: Mapping from a source name to a concrete
            `quantex.sources.DataSource` implementation.
        symbols: List of tradable symbols to initialize `Position` objects
            for. If None, positions are created lazily.
        initial_cash: Starting cash for the internal `Portfolio`.
    """

    # Store references to market data sources (may be empty – engine injects)
    self.data_sources = data_sources or {}
    self.timestamp = None

    # Maintain a portfolio to aggregate cash & PnL
    self.portfolio: Portfolio = Portfolio(cash=initial_cash)

    # Expose positions dict for backward compatibility
    self.positions = self.portfolio.positions

    # Register provided symbols (if any)
    self.symbols = symbols or []
    for sym in self.symbols:
        # Pre-create empty Position objects so that strategy code can rely
        # on their existence.
        if sym not in self.positions:
            self.positions[sym] = Position(sym)

    # Queue of orders submitted during the current bar – cleared each step
    self._pending_orders: list[Order] = []
    self.signals = pd.DataFrame()

    # --- Internals set by the engine each bar ---
    self._price_row: list[float] | None = None
    self._symbols: list[str] | None = None
    self._symbol_idx: dict[str, int] | None = None

buy(symbol, quantity, limit_price=None)

Creates and submits a buy order using the current strategy timestamp.

Parameters:

Name Type Description Default
symbol str

The symbol to buy.

 required
quantity float

The quantity to buy. Must be positive.

 required
limit_price float | None

If provided, a limit order is created. Otherwise, a market order is created.

 None
Source code in src/quantex/strategy.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def buy(
    self, symbol: str, quantity: float, limit_price: float | None = None
) -> None:
    """Creates and submits a buy order using the current strategy timestamp.

    Args:
        symbol: The symbol to buy.
        quantity: The quantity to buy. Must be positive.
        limit_price: If provided, a limit order is created. Otherwise, a
            market order is created.
    """
    if self.timestamp is None:
        raise RuntimeError("Cannot place order: strategy timestamp is not set.")

    order = Order(
        id=f"buy-{symbol}-{self.timestamp}",
        symbol=symbol,
        side="buy",
        quantity=quantity,
        order_type="limit" if limit_price else "market",
        limit_price=limit_price,
        timestamp=self.timestamp,
    )
    self.submit_order(order)

close_position(symbol)

Creates and submits an order to close the entire position for a symbol.

This is a helper method that checks if a position is open for the given symbol and, if so, creates a market order to close it at the current strategy timestamp.

Parameters:

Name Type Description Default
symbol str

The symbol of the position to close.

 required
Source code in src/quantex/strategy.py 
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
def close_position(self, symbol: str) -> None:
    """Creates and submits an order to close the entire position for a symbol.

    This is a helper method that checks if a position is open for the
    given symbol and, if so, creates a market order to close it at the
    current strategy timestamp.

    Args:
        symbol: The symbol of the position to close.
    """
    if self.timestamp is None:
        raise RuntimeError("Cannot place order: strategy timestamp is not set.")

    position = self.positions.get(symbol)
    if not position or position.is_closed:
        return

    if position.is_long:
        side = "sell"
        quantity = position.position
    else:  # is_short
        side = "buy"
        quantity = abs(position.position)

    order = Order(
        id=f"close-{symbol}-{self.timestamp}",
        symbol=symbol,
        side=side,
        quantity=quantity,
        timestamp=self.timestamp,
    )
    self.submit_order(order)

get_lookback_prices(lookback_period)

Returns an aligned lookback window for all symbols.

Parameters:

Name Type Description Default
lookback_period int

Number of bars (inclusive of the current bar) to return.

 required

Returns:

Type Description
DataFrame

pd.DataFrame: DataFrame with the last lookback_period rows from
DataFrame

pyattr:price_history. If there are fewer than lookback_period
DataFrame

observations available (e.g. at the beginning of a backtest), the
DataFrame

entire available history is returned instead. The returned frame
DataFrame

is guaranteed to be free of missing timestamps across symbols due
DataFrame

to the forward-filling performed by the engine.
Source code in src/quantex/strategy.py 
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
def get_lookback_prices(self, lookback_period: int) -> pd.DataFrame:
    """Returns an *aligned* lookback window for all symbols.

    Args:
        lookback_period: Number of bars (inclusive of the current bar) to
            return.

    Returns:
        pd.DataFrame: DataFrame with the last *lookback_period* rows from
        :pyattr:`price_history`. If there are fewer than *lookback_period*
        observations available (e.g. at the beginning of a backtest), the
        entire available history is returned instead. The returned frame
        is guaranteed to be free of missing timestamps across symbols due
        to the forward-filling performed by the engine.
    """

    history = self.price_history
    if len(history) < lookback_period:
        return history.copy()
    return history.iloc[-lookback_period:].copy()

get_price(symbol)

Returns the latest price for symbol.

This accesses the numpy price row injected by the engine and is therefore O(1) without any pandas overhead. Raises KeyError if the symbol is not part of the backtest universe.

Source code in src/quantex/strategy.py 
205
206
207
208
209
210
211
212
213
214
215
216
217
218
def get_price(self, symbol: str) -> float:
    """Returns the latest price for *symbol*.

    This accesses the numpy price row injected by the engine and is
    therefore O(1) without any pandas overhead. Raises ``KeyError`` if the
    symbol is not part of the backtest universe.
    """

    if self._price_row is None or self._symbol_idx is None:
        raise RuntimeError("Market data not yet initialised for this bar.")
    idx = self._symbol_idx.get(symbol)
    if idx is None:
        raise KeyError(symbol)
    return float(self._price_row[idx])

run() abstractmethod

Executes the strategy logic for the current bar.

Concrete strategies must override this method. It should inspect the available data sources and make trading decisions.

Source code in src/quantex/strategy.py 
 97
 98
 99
100
101
102
103
104
@abstractmethod
def run(self):
    """Executes the strategy logic for the current bar.

    Concrete strategies must override this method. It should inspect
    the available data sources and make trading decisions.
    """
    raise NotImplementedError("Subclasses must implement this method.")

sell(symbol, quantity, limit_price=None)

Creates and submits a sell order using the current strategy timestamp.

Parameters:

Name Type Description Default
symbol str

The symbol to sell.

 required
quantity float

The quantity to sell. Must be positive.

 required
limit_price float | None

If provided, a limit order is created. Otherwise, a market order is created.

 None
Source code in src/quantex/strategy.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def sell(
    self, symbol: str, quantity: float, limit_price: float | None = None
) -> None:
    """Creates and submits a sell order using the current strategy timestamp.

    Args:
        symbol: The symbol to sell.
        quantity: The quantity to sell. Must be positive.
        limit_price: If provided, a limit order is created. Otherwise, a
            market order is created.
    """
    if self.timestamp is None:
        raise RuntimeError("Cannot place order: strategy timestamp is not set.")

    order = Order(
        id=f"sell-{symbol}-{self.timestamp}",
        symbol=symbol,
        side="sell",
        quantity=quantity,
        order_type="limit" if limit_price else "market",
        limit_price=limit_price,
        timestamp=self.timestamp,
    )
    self.submit_order(order)

submit_order(order)

Queues an order to be executed by the engine.

Strategies should call this method to simulate realistic order routing.

Parameters:

Name Type Description Default
order Order

The Order to be submitted.

 required
Source code in src/quantex/strategy.py 
189
190
191
192
193
194
195
196
197
198
def submit_order(self, order: Order) -> None:
    """Queues an order to be executed by the engine.

    Strategies should call this method to simulate realistic order routing.

    Args:
        order: The `Order` to be submitted.
    """

    self._pending_orders.append(order)