Strategy Module

Execution

signalflow.strategy.runner.backtest_runner.BacktestRunner dataclass

BacktestRunner(strategy_id: str = 'backtest', broker: Any = None, entry_rules: list[EntryRule] = list(), exit_rules: list[ExitRule] = list(), metrics: list[StrategyMetric] = list(), initial_capital: float = 10000.0, pair_col: str = 'pair', ts_col: str = 'timestamp', price_col: str = 'close', data_key: str = 'spot')

Bases: StrategyRunner

Runs backtests over historical data.

Execution flow per bar

1. Mark prices on all positions
2. Compute metrics
3. Check and execute exits
4. Check and execute entries

This order ensures

• Metrics reflect current market state
• Exits are processed before entries (can close and re-enter same bar)
• No look-ahead bias

metrics_df property

metrics_df: DataFrame

Get metrics history as a DataFrame.

trades property

trades: list[Trade]

Get all trades from the backtest.

trades_df property

trades_df: DataFrame

Get trades as a DataFrame.

get_results

get_results() -> dict[str, Any]

Get backtest results summary.

Source code in src/signalflow/strategy/runner/backtest_runner.py

def get_results(self) -> dict[str, Any]:
    """Get backtest results summary."""
    trades_df = self.trades_df
    metrics_df = self.metrics_df

    results = {
        'total_trades': len(self._trades),
        'metrics_df': metrics_df,
        'trades_df': trades_df,
    }

    if metrics_df.height > 0 and 'total_return' in metrics_df.columns:
        results['final_return'] = metrics_df.select('total_return').tail(1).item()
        results['final_equity'] = metrics_df.select('equity').tail(1).item()

    if trades_df.height > 0:
        entry_trades = trades_df.filter(pl.col('meta').struct.field('type') == 'entry')
        exit_trades = trades_df.filter(pl.col('meta').struct.field('type') == 'exit')
        results['entry_count'] = entry_trades.height
        results['exit_count'] = exit_trades.height

    return results

run

run(raw_data: RawData, signals: Signals, state: StrategyState | None = None) -> StrategyState

Run backtest over the entire dataset.

Parameters:

Name	Type	Description	Default
raw_data	RawData	Historical OHLCV data	required
signals	Signals	Pre-computed signals for the period	required
state	StrategyState | None	Optional initial state (for continuing backtests)	None

Returns:

Type	Description
StrategyState	Final strategy state

Source code in src/signalflow/strategy/runner/backtest_runner.py

def run(
    self,
    raw_data: RawData,
    signals: Signals,
    state: StrategyState | None = None
) -> StrategyState:
    """
    Run backtest over the entire dataset.

    Args:
        raw_data: Historical OHLCV data
        signals: Pre-computed signals for the period
        state: Optional initial state (for continuing backtests)

    Returns:
        Final strategy state
    """
    if state is None:
        state = StrategyState(
            strategy_id=self.strategy_id,
        )
        state.portfolio.cash = self.initial_capital

    self._trades = []
    self._metrics_history = []

    # Get data
    df = raw_data.get(self.data_key)
    if df.height == 0:
        logger.warning("No data to backtest")
        return state

    timestamps = df.select(self.ts_col).unique().sort(self.ts_col).get_column(self.ts_col)

    signals_df = signals.value if signals else pl.DataFrame()

    logger.info(f"Starting backtest: {len(timestamps)} bars, {signals_df.height} signals")

    for ts in tqdm(timestamps, desc="Processing bars"):
        state = self._process_bar(
            ts=ts,
            raw_df=df,
            signals_df=signals_df,
            state=state
        )

    logger.info(
        f"Backtest complete: {len(self._trades)} trades, "
        f"{len(state.portfolio.open_positions())} open positions"
    )

    return state

Exit Rules

signalflow.strategy.component.exit.tp_sl.TakeProfitStopLossExit dataclass

TakeProfitStopLossExit(take_profit_pct: float = 0.02, stop_loss_pct: float = 0.01, use_position_levels: bool = False)

Bases: ExitRule

Exit rule based on take-profit and stop-loss levels.

Can use fixed percentages or dynamic levels from position meta.

Metrics

signalflow.strategy.component.metric

__all__ module-attribute

__all__ = ['TotalReturnMetric', 'BalanceAllocationMetric', 'DrawdownMetric', 'WinRateMetric', 'SharpeRatioMetric']

BalanceAllocationMetric dataclass

BalanceAllocationMetric(initial_capital: float = 10000.0)

Bases: StrategyMetric

DrawdownMetric dataclass

DrawdownMetric(_peak_equity: float = 0.0, _max_drawdown: float = 0.0)

Bases: StrategyMetric

SharpeRatioMetric dataclass

SharpeRatioMetric(initial_capital: float = 10000.0, window_size: int = 100, risk_free_rate: float = 0.0, _returns_history: list[float] = None)

Bases: StrategyMetric

TotalReturnMetric dataclass

TotalReturnMetric(initial_capital: float = 10000.0)

Bases: StrategyMetric

Computes total return metrics for the portfolio.

WinRateMetric dataclass

WinRateMetric()

Bases: StrategyMetric