Backtest Zipline

backtest-zipline

Set up a working backtesting pipeline using zipline-reloaded for the algorithm engine and bcolz-zipline for fast columnar storage of OHLC bundles.

When to use

Trigger this skill when the user wants to:

• Backtest a Python trading strategy against historical OHLC data
• Build a custom data bundle from a non-Quandl source (Polygon, Binance, IEX, CSV)
• Run zipline programmatically (not via the zipline CLI) and receive results in-process
• Convert backtest output into a JSON shape a React frontend can render

Prereqs

zipline-reloaded requires Python 3.11 (3.12 and 3.13 are not supported yet — numpy ABI breaks the C extensions). Pin it in pyproject.toml:

requires-python = ">=3.11,<3.12"
dependencies = [
  "zipline-reloaded>=3.0.4",
  "bcolz-zipline>=1.2.6",
  "pandas>=2.2",
  "numpy>=1.26,<2.0",
]

Use uv to install — it handles the legacy numpy pin without fuss:

uv sync

1. Register a custom bundle (Polygon daily bars)

Create bundles/polygon_daily.py:

"""Ingest Polygon daily bars into a zipline bcolz bundle."""
from __future__ import annotations
import os
import pandas as pd
import httpx
from zipline.data.bundles import register


def polygon_bundle(symbols: list[str]):
    def ingest(environ, asset_db_writer, minute_bar_writer, daily_bar_writer,
               adjustment_writer, calendar, start_session, end_session,
               cache, show_progress, output_dir):
        sid = 0
        metadata = []
        ohlc = []
        api_key = os.environ["MASSIVE_API_KEY"]
        for symbol in symbols:
            url = (f"https://api.massive.com/v2/aggs/ticker/{symbol}"
                   f"/range/1/day/{start_session.date()}/{end_session.date()}"
                   f"?adjusted=true&sort=asc&limit=50000&apiKey={api_key}")
            results = httpx.get(url, timeout=30).json().get("results", [])
            if not results:
                continue
            df = pd.DataFrame(results).rename(
                columns={"t": "ts", "o": "open", "h": "high", "l": "low", "c": "close", "v": "volume"}
            )
            df["date"] = pd.to_datetime(df["ts"], unit="ms", utc=True).dt.tz_convert(None).dt.normalize()
            df = df.set_index("date")[["open", "high", "low", "close", "volume"]]
            ohlc.append((sid, df))
            metadata.append({
                "sid": sid, "symbol": symbol,
                "start_date": df.index[0], "end_date": df.index[-1],
                "first_traded": df.index[0], "auto_close_date": df.index[-1] + pd.Timedelta(days=1),
                "exchange": "NYSE",
            })
            sid += 1

        daily_bar_writer.write(ohlc, show_progress=show_progress)
        asset_db_writer.write(pd.DataFrame(metadata).set_index("sid"))
        adjustment_writer.write()  # no splits/divs

    return ingest


# Register at import time — call ingest() from your app:
register("polygon-daily", polygon_bundle(["SPY", "QQQ", "AAPL"]), calendar_name="NYSE")

Trigger ingestion (creates the bcolz files under ~/.zipline/data/polygon-daily/...):

from zipline.data.bundles import ingest
ingest("polygon-daily")

2. Run an algorithm programmatically

from datetime import datetime, timezone
import pandas as pd
from zipline import run_algorithm
from zipline.api import order_target_percent, symbol, record


def initialize(context):
    context.asset = symbol("SPY")
    context.fast = 20
    context.slow = 50


def handle_data(context, data):
    prices = data.history(context.asset, "close", context.slow + 1, "1d")
    fast = prices[-context.fast:].mean()
    slow = prices.mean()
    if fast > slow:
        order_target_percent(context.asset, 1.0)
    else:
        order_target_percent(context.asset, 0.0)
    record(price=prices.iloc[-1], fast=fast, slow=slow)


def run(start: str, end: str, capital: float) -> dict:
    result = run_algorithm(
        start=pd.Timestamp(start, tz=timezone.utc),
        end=pd.Timestamp(end, tz=timezone.utc),
        initialize=initialize,
        handle_data=handle_data,
        capital_base=capital,
        bundle="polygon-daily",
        data_frequency="daily",
    )
    return _to_payload(result, capital)


def _to_payload(result: pd.DataFrame, capital: float) -> dict:
    """Shape zipline's perf DataFrame for a web UI."""
    equity_curve = [
        {"time": int(ts.timestamp()), "value": float(row["portfolio_value"])}
        for ts, row in result.iterrows()
    ]
    final = float(result["portfolio_value"].iloc[-1])
    returns = result["returns"]
    sharpe = float((returns.mean() / returns.std()) * (252 ** 0.5)) if returns.std() else 0.0
    rolling_max = result["portfolio_value"].cummax()
    drawdown = float(((result["portfolio_value"] - rolling_max) / rolling_max).min())
    return {
        "initialCapital": capital,
        "finalEquity": final,
        "totalReturn": final / capital - 1.0,
        "equityCurve": equity_curve,
        "stats": {"sharpe": sharpe, "maxDrawdown": drawdown, "volatility": float(returns.std() * (252 ** 0.5))},
        "trades": [],  # populate from result["transactions"] if needed
    }

3. Wire to FastAPI

from fastapi import APIRouter
from pydantic import BaseModel

router = APIRouter()


class Req(BaseModel):
    start: str
    end: str
    capital: float = 100_000


@router.post("/api/backtest")
async def backtest(req: Req):
    from .algo import run
    return run(req.start, req.end, req.capital)

Common gotchas

• pandas.tseries.offsets errors — usually means the bundle's calendar bounds don't overlap with start/end. Re-ingest covering the requested range.
• KeyError: 'SPY' when calling symbol(...) — bundle didn't ingest that ticker, or you forgot to call register() before run_algorithm.
• bcolz import fails on macOS arm64 — install bcolz-zipline (the fork), NOT the upstream bcolz. The latter is dead and won't build.
• Equity curve looks flat — zipline holds the position once it's opened and record() data is independent of orders. Inspect result["transactions"] to verify trades actually fired.

Lightweight alternative

For an MVP or quick prototype, you don't need zipline at all — a pure Python loop over Polygon daily bars works fine and is much faster to spin up. command-dash's apps/api/app/backtest/runner.py does exactly this:

for bar in bars:
    signal = strategy.signal_for_bar(bar, state)  # 'buy' | 'sell' | None
    # … apply to cash/position …

Use zipline when you need: minute data, slippage models, multi-asset portfolios, leverage/margin, or accurate corporate action handling. Otherwise, the loop is enough.