turtle_quant_1.trading.engine

turtle_quant_1.trading.engine

Trading engine for executing trading signals.

TradingEngine

TradingEngine(
    strategy_engine: BaseStrategyEngine,
    portfolio: Portfolio,
    max_lookback_days: int,
    tick_interval: CandleUnit = "4H",
    event_interval: CandleUnit = "30M",
)

Trading engine that processes ticks and executes trading signals.

Nomenclature: - Signal: A single trading signal that is independent of time. - Tick: A single trigger of the trading engine, typically defined by when the host environment gets spooled up and triggered. - Event: A single candle at the interval of CANDLE_UNIT. This will be at a smaller interval than a tick, i.e. between every tick, multiple events will have passed. - "curr..." variables: The variable refers to the latest value at the time of the tick, not at the event.

Initialise the trading engine.

Parameters:

Name	Type	Description	Default
strategy_engine	BaseStrategyEngine	The strategy engine used to generate signals.	required
portfolio	Portfolio	The portfolio to trade against.	required
max_lookback_days	int	Days of history fed to the strategy on each tick.	required
tick_interval	CandleUnit	Duration of one tick window as a candle-unit string (e.g. "4H"). Used together with CANDLE_UNIT to determine how many candle events are replayed per tick.	'4H'
event_interval	CandleUnit	Duration of one event window as a candle-unit string (e.g. "1H"). Used together with CANDLE_UNIT to determine how many candle events are replayed per event.	'30M'

Source code in src/turtle_quant_1/trading/engine.py

def __init__(
    self,
    strategy_engine: BaseStrategyEngine,
    portfolio: Portfolio,
    max_lookback_days: int,
    tick_interval: CandleUnit = "4H",
    event_interval: CandleUnit = "30M",
):
    """Initialise the trading engine.

    Args:
        strategy_engine: The strategy engine used to generate signals.
        portfolio: The portfolio to trade against.
        max_lookback_days: Days of history fed to the strategy on each tick.
        tick_interval: Duration of one tick window as a candle-unit string
            (e.g. "4H"). Used together with CANDLE_UNIT to determine how
            many candle events are replayed per tick.
        event_interval: Duration of one event window as a candle-unit string
            (e.g. "1H"). Used together with CANDLE_UNIT to determine how
            many candle events are replayed per event.
    """
    self.strategy_engine = strategy_engine
    self.portfolio = portfolio
    self.max_lookback_days = max_lookback_days
    self.tick_interval = tick_interval
    self.event_interval = event_interval

    self.signal_history_list: list[dict] = []
    self.total_signals: int = 0
    self.total_transactions: int = 0

handle_tick

handle_tick(symbol: str, curr_data: DataFrame) -> None

Process a tick for symbol by replaying every event in the tick window.

Rather than acting on the latest candle alone, iterates over every candle within the most recent tick window of curr_data, simulating an event-driven system where each arriving candle is assessed individually. The window size is determined by tick_interval set at construction time.

Parameters:

Name	Type	Description	Default
symbol	str	Symbol being processed.	required
curr_data	DataFrame	All candles available up to the current tick timestamp.	required

Source code in src/turtle_quant_1/trading/engine.py

def handle_tick(
    self,
    symbol: str,
    curr_data: pd.DataFrame,
) -> None:
    """Process a tick for *symbol* by replaying every event in the tick window.

    Rather than acting on the latest candle alone, iterates over every candle
    within the most recent tick window of *curr_data*, simulating an
    event-driven system where each arriving candle is assessed individually.
    The window size is determined by *tick_interval* set at construction time.

    Args:
        symbol: Symbol being processed.
        curr_data: All candles available up to the current tick timestamp.
    """
    # Get the number of events between the last tick and the current tick.
    n_units_per_event = convert_units(1, self.event_interval, CANDLE_UNIT)
    n_events_per_tick = convert_units(1, self.tick_interval, self.event_interval)
    n_units_per_tick = n_events_per_tick * n_units_per_event
    last_tick_index = max(0, len(curr_data) - n_units_per_tick)

    signals: list[Signal] = []
    for i in range(last_tick_index, len(curr_data), n_units_per_event):
        signal = self._handle_event(
            symbol=symbol, curr_data=curr_data, event_index=i
        )
        signals.append(signal)

    # NOTE: Use the latest price at the time of the tick (not the event),
    # since the buy/sell price will be executed on the tick itself, i.e. delayed.
    curr_price = curr_data.iloc[-1]["Close"]
    # NOTE: Use the timestamp of the tick, since the signal is based on the tick.
    curr_timestamp = curr_data.iloc[-1]["datetime"]

    b_signals = [signal for signal in signals if signal.action == SignalAction.BUY]
    h_signals = [signal for signal in signals if signal.action == SignalAction.HOLD]
    s_signals = [signal for signal in signals if signal.action == SignalAction.SELL]
    if len(s_signals) > 0:
        signal = s_signals[0]
    elif len(b_signals) > 0:
        signal = b_signals[0]
    elif len(h_signals) > 0:
        signal = h_signals[0]
    else:
        raise ValueError(f"No signal found for {symbol} at {curr_timestamp}")

    transaction_executed = self._handle_signal(
        symbol=symbol, signal=signal, price=curr_price, timestamp=curr_timestamp
    )
    if transaction_executed:
        self.total_transactions += 1

    self.portfolio.check_take_profit_triggers({symbol: curr_price})
    self.portfolio.check_stop_loss_triggers({symbol: curr_price})

    self.signal_history_list.append(
        {
            "datetime": curr_timestamp,
            "symbol": symbol,
            "action": signal.action.value,
            "score": signal.score,
            "strategies": signal.strategies,
            "executed": transaction_executed,
        }
    )