# Backtest Zipline

> Backtest trading strategies with zipline-reloaded backed by bcolz columnar storage. Use when the user wants to ingest a custom data bundle from Polygon (or any source), run a zipline algorithm against it, and surface results (equity curve, Sharpe, drawdown) as JSON for a web UI.

- Skill: `traderspost/backtest-zipline` (Agent Skill)
- Install (CLI): `npx skillmds@latest add traderspost/backtest-zipline`
- Raw SKILL.md: https://api.skillmd.com/api/skills/traderspost/backtest-zipline/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- Author: traderspost (https://skillmd.com/u/traderspost)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/traderspost/backtest-zipline

---


# backtest-zipline

Set up a working backtesting pipeline using [zipline-reloaded](https://github.com/stefan-jansen/zipline-reloaded) for the algorithm engine and [bcolz-zipline](https://github.com/blosc/bcolz) 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`:

```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`](https://github.com/astral-sh/uv) to install — it handles the legacy numpy pin without fuss:

```bash
uv sync
```

## 1. Register a custom bundle (Polygon daily bars)

Create `bundles/polygon_daily.py`:

```python
"""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/...`):

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

## 2. Run an algorithm programmatically

```python
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

```python
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:

```python
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.