Strategy templates

Canonical patterns: copy, adapt, submit. All extend bagtester.Strategy.

SMA crossover

python
from bagtester import Strategy

class SmaCross(Strategy):
    def __init__(self, fast=20, slow=50):
        self.fast = fast
        self.slow = slow
        self.closes = []

    def on_init(self, ctx):
        ctx.subscribe("BTCUSDT")  # symbol passed via submit_strategy is auto-subscribed

    def on_bar(self, ctx, bar):
        self.closes.append(bar.close)
        if len(self.closes) < self.slow:
            return
        fast = sum(self.closes[-self.fast:]) / self.fast
        slow = sum(self.closes[-self.slow:]) / self.slow
        pos = ctx.position(bar.symbol)
        if fast > slow and not pos:
            ctx.buy(bar.symbol, ctx.equity * 0.95 / bar.close)
        elif fast < slow and pos:
            ctx.close(bar.symbol)

RSI mean-reversion

python
from bagtester import Strategy
import pandas as pd

class RsiReversion(Strategy):
    def __init__(self, length=14, oversold=30, overbought=70):
        self.length = length
        self.oversold = oversold
        self.overbought = overbought
        self.closes = []

    def on_bar(self, ctx, bar):
        self.closes.append(bar.close)
        if len(self.closes) < self.length + 1:
            return
        s = pd.Series(self.closes[-(self.length + 1):])
        delta = s.diff().dropna()
        gain = delta.clip(lower=0).mean()
        loss = (-delta.clip(upper=0)).mean()
        rs = gain / loss if loss > 0 else float("inf")
        rsi = 100 - 100 / (1 + rs)
        pos = ctx.position(bar.symbol)
        if rsi < self.oversold and not pos:
            ctx.buy(bar.symbol, ctx.equity * 0.95 / bar.close)
        elif rsi > self.overbought and pos:
            ctx.close(bar.symbol)

Donchian breakout

python
from bagtester import Strategy
from collections import deque

class DonchianBreakout(Strategy):
    def __init__(self, lookback=55):
        self.lookback = lookback
        self.window = deque(maxlen=lookback)

    def on_bar(self, ctx, bar):
        self.window.append((bar.high, bar.low))
        if len(self.window) < self.lookback:
            return
        upper = max(h for h, _ in self.window)
        lower = min(l for _, l in self.window)
        pos = ctx.position(bar.symbol)
        if bar.close >= upper and not pos:
            ctx.buy(bar.symbol, ctx.equity * 0.95 / bar.close)
        elif bar.close <= lower and pos:
            ctx.close(bar.symbol)

Cross-symbol pairs trade

python
# Note: multi-symbol backtests are a Quant-tier feature. Single-symbol shown for reference.
from bagtester import Strategy
import numpy as np

class ZScoreReversion(Strategy):
    def __init__(self, window=200, entry_z=2.0, exit_z=0.5):
        self.window = window
        self.entry_z = entry_z
        self.exit_z = exit_z
        self.history = []

    def on_bar(self, ctx, bar):
        self.history.append(bar.close)
        if len(self.history) < self.window:
            return
        recent = np.array(self.history[-self.window:])
        z = (bar.close - recent.mean()) / recent.std()
        pos = ctx.position(bar.symbol)
        if z < -self.entry_z and not pos:
            ctx.buy(bar.symbol, ctx.equity * 0.5 / bar.close)
        elif z > self.entry_z and not pos:
            ctx.sell(bar.symbol, ctx.equity * 0.5 / bar.close)
        elif abs(z) < self.exit_z and pos:
            ctx.close(bar.symbol)

Optimizing parameters

Once you have a strategy with __init__(self, **kwargs) accepting parameters, you can sweep them via optimize_strategy:

json
{
  "method": "tools/call",
  "params": {
    "name": "optimize_strategy",
    "arguments": {
      "code": "...",
      "symbol": "BTCUSDT",
      "from_date": "2024-01-01",
      "to_date": "2024-12-31",
      "param_grid": {
        "fast": [10, 20, 50],
        "slow": [100, 200]
      },
      "primary_metric": "sharpe_ratio"
    }
  }
}

Available imports

The runtime ships with:

No pip install at runtime. No network egress. Wall-clock cap of 4 hours per job.