Backtesting de Opções com Python e brapi: Covered Call e Cash-Secured Put (2026)

options-backtest/
├── data.py             # Carrega dados de underlying e opções
├── strategies.py       # Implementa as estratégias
├── metrics.py          # Sharpe, drawdown, returns
├── plot.py             # Visualizações
└── run.py              # Pipeline principal

mkdir options-backtest && cd options-backtest
python -m venv .venv && source .venv/bin/activate
pip install requests pandas numpy matplotlib python-dateutil

export BRAPI_TOKEN="seu_token_pro"  # opcional para PETR4

import os
import requests
from datetime import date
import pandas as pd

BRAPI_QUOTE = "https://brapi.dev/api/quote"
BRAPI_OPTIONS = "https://brapi.dev/api/v2/options"
TOKEN = os.getenv("BRAPI_TOKEN", "")
HEADERS = {"Authorization": f"Bearer {TOKEN}"} if TOKEN else {}


def load_underlying_history(ticker: str, range_: str = "2y") -> pd.DataFrame:
    """Carrega OHLCV diário do underlying."""
    resp = requests.get(
        f"{BRAPI_QUOTE}/{ticker}",
        params={"range": range_, "interval": "1d"},
        headers=HEADERS,
        timeout=60,
    )
    resp.raise_for_status()
    data = resp.json()["results"][0].get("historicalDataPrice", [])

    df = pd.DataFrame(data)
    df["date"] = pd.to_datetime(df["date"], unit="s")
    return df.set_index("date").sort_index()


def list_expirations(underlying: str, include_expired: bool = True) -> list[str]:
    """Vencimentos disponíveis (incluindo passados, para backtest)."""
    resp = requests.get(
        f"{BRAPI_OPTIONS}/expirations",
        params={"underlying": underlying, "includeExpired": str(include_expired).lower()},
        headers=HEADERS,
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json()["expirations"]


def get_chain_at_date(
    underlying: str,
    expiration_date: str,
    date_: str,
    side: str | None = None,
) -> list[dict]:
    """Cadeia de opções em uma data específica do passado."""
    params = {
        "underlying": underlying,
        "expirationDate": expiration_date,
        "date": date_,
    }
    if side:
        params["side"] = side
    resp = requests.get(f"{BRAPI_OPTIONS}/chain", params=params, headers=HEADERS, timeout=30)
    resp.raise_for_status()
    return resp.json().get("series", [])


def get_option_history(symbol: str, expiration_date: str) -> pd.DataFrame:
    """OHLCV diário de uma série de opção."""
    resp = requests.get(
        f"{BRAPI_OPTIONS}/historical",
        params={"symbol": symbol, "expirationDate": expiration_date},
        headers=HEADERS,
        timeout=60,
    )
    resp.raise_for_status()
    payload = resp.json().get("option", {})
    history = payload.get("history", [])

    df = pd.DataFrame(history)
    if df.empty:
        return df
    df["date"] = pd.to_datetime(df["date"], unit="s")
    return df.set_index("date").sort_index()

from dateutil.relativedelta import relativedelta


def monthly_open_dates(start: date, end: date) -> list[date]:
    """Lista o primeiro dia útil de cada mês no intervalo."""
    out: list[date] = []
    cursor = date(start.year, start.month, 1)
    while cursor <= end:
        # Avança até o primeiro dia útil do mês
        d = cursor
        while d.weekday() >= 5:  # 5=sat, 6=sun
            d += relativedelta(days=+1)
        out.append(d)
        cursor = (cursor + relativedelta(months=+1)).replace(day=1)
    return out


def pick_expiration_for_month(
    expirations: list[str],
    open_date: date,
) -> str | None:
    """Pega o próximo vencimento após a open_date."""
    for exp_str in sorted(expirations):
        exp = date.fromisoformat(exp_str)
        if exp > open_date:
            return exp_str
    return None

from dataclasses import dataclass, field
import pandas as pd
from data import (
    load_underlying_history,
    list_expirations,
    get_chain_at_date,
    monthly_open_dates,
    pick_expiration_for_month,
)
from datetime import date


@dataclass
class Trade:
    open_date: date
    close_date: date
    underlying_open_price: float
    strike: float
    premium_received: float
    underlying_close_price: float
    exercised: bool
    pnl: float                 # PnL da posição em opção (prêmio − recompra ou exercício)
    underlying_pnl: float      # Variação do underlying no período
    total_pnl: float           # Soma


@dataclass
class CoveredCallConfig:
    otm_pct: float = 0.05      # 5% acima do spot
    contract_size: int = 100   # 1 contrato = 100 ações
    min_trades_filter: int = 50  # Filtro mínimo de liquidez


def run_covered_call(
    underlying: str,
    start: date,
    end: date,
    config: CoveredCallConfig,
) -> list[Trade]:
    underlying_df = load_underlying_history(underlying, range_="5y")
    expirations = list_expirations(underlying)

    trades: list[Trade] = []
    for open_date in monthly_open_dates(start, end):
        # Spot na abertura
        try:
            spot = float(underlying_df.loc[pd.Timestamp(open_date)]["close"])
        except KeyError:
            # Pega o pregão mais próximo
            spot = float(underlying_df.loc[underlying_df.index >= pd.Timestamp(open_date)].iloc[0]["close"])

        # Vencimento alvo
        exp = pick_expiration_for_month(expirations, open_date)
        if exp is None:
            continue

        # Cadeia naquela data
        chain = get_chain_at_date(
            underlying=underlying,
            expiration_date=exp,
            date_=open_date.isoformat(),
            side="call",
        )

        # Filtra strikes ≥ spot×(1+otm_pct), pega o menor (mais próximo do dinheiro mas ainda OTM)
        target_strike = spot * (1 + config.otm_pct)
        candidates = [
            s for s in chain
            if s["strike"] >= target_strike
            and s["trades"] >= config.min_trades_filter
            and s.get("close") is not None
        ]
        if not candidates:
            continue

        chosen = min(candidates, key=lambda s: s["strike"])

        # Preço de venda do prêmio: usa close do open_date (proxy aceitável de execução EOD)
        premium = float(chosen["close"])

        # Spot no vencimento
        exp_date = date.fromisoformat(exp)
        try:
            close_spot = float(underlying_df.loc[underlying_df.index >= pd.Timestamp(exp_date)].iloc[0]["close"])
        except IndexError:
            continue

        strike = float(chosen["strike"])
        exercised = close_spot > strike

        # PnL da call vendida (você lucra se ela vira pó; perde se a paga acima do prêmio recebido no exercício)
        # Simplificação: se exercido, paga max(close_spot − strike, 0) a quem comprou
        if exercised:
            option_pnl = (premium - (close_spot - strike)) * config.contract_size
            # Underlying pnl: vc vendeu no strike (não no close_spot), então capture só até strike
            underlying_pnl = (strike - spot) * config.contract_size
        else:
            option_pnl = premium * config.contract_size
            underlying_pnl = (close_spot - spot) * config.contract_size

        trades.append(Trade(
            open_date=open_date,
            close_date=exp_date,
            underlying_open_price=spot,
            strike=strike,
            premium_received=premium,
            underlying_close_price=close_spot,
            exercised=exercised,
            pnl=option_pnl,
            underlying_pnl=underlying_pnl,
            total_pnl=option_pnl + underlying_pnl,
        ))

    return trades

@dataclass
class CashSecuredPutConfig:
    otm_pct: float = 0.05
    contract_size: int = 100
    min_trades_filter: int = 50
    # Yield do caixa colocado em CDB / SELIC equivalente
    cash_yield_annual: float = 0.10  # 10% a.a.


def run_cash_secured_put(
    underlying: str,
    start: date,
    end: date,
    config: CashSecuredPutConfig,
) -> list[Trade]:
    underlying_df = load_underlying_history(underlying, range_="5y")
    expirations = list_expirations(underlying)

    trades: list[Trade] = []
    for open_date in monthly_open_dates(start, end):
        try:
            spot = float(underlying_df.loc[pd.Timestamp(open_date)]["close"])
        except KeyError:
            spot = float(underlying_df.loc[underlying_df.index >= pd.Timestamp(open_date)].iloc[0]["close"])

        exp = pick_expiration_for_month(expirations, open_date)
        if exp is None:
            continue

        chain = get_chain_at_date(
            underlying=underlying,
            expiration_date=exp,
            date_=open_date.isoformat(),
            side="put",
        )

        target_strike = spot * (1 - config.otm_pct)
        # Pega a put OTM mais próxima do spot (strike ≤ target_strike, maior strike primeiro)
        candidates = [
            s for s in chain
            if s["strike"] <= target_strike
            and s["trades"] >= config.min_trades_filter
            and s.get("close") is not None
        ]
        if not candidates:
            continue

        chosen = max(candidates, key=lambda s: s["strike"])
        premium = float(chosen["close"])
        strike = float(chosen["strike"])

        exp_date = date.fromisoformat(exp)
        try:
            close_spot = float(underlying_df.loc[underlying_df.index >= pd.Timestamp(exp_date)].iloc[0]["close"])
        except IndexError:
            continue

        exercised = close_spot < strike

        # Caixa rende no período
        days = (exp_date - open_date).days
        cash_pnl = strike * config.contract_size * config.cash_yield_annual * days / 365

        if exercised:
            # Comprou no strike, mark-to-market no close_spot
            option_pnl = (premium - (strike - close_spot)) * config.contract_size
        else:
            option_pnl = premium * config.contract_size

        trades.append(Trade(
            open_date=open_date,
            close_date=exp_date,
            underlying_open_price=spot,
            strike=strike,
            premium_received=premium,
            underlying_close_price=close_spot,
            exercised=exercised,
            pnl=option_pnl,
            underlying_pnl=cash_pnl,  # neste caso "underlying" é o caixa rendendo
            total_pnl=option_pnl + cash_pnl,
        ))

    return trades

import numpy as np
import pandas as pd
from strategies import Trade


def trades_to_returns(trades: list[Trade], capital_per_trade: float) -> pd.Series:
    """Converte trades em série temporal de retornos mensais (% sobre capital alocado)."""
    df = pd.DataFrame([
        {"date": t.close_date, "pnl": t.total_pnl}
        for t in trades
    ]).set_index("date").sort_index()

    df["return"] = df["pnl"] / capital_per_trade
    return df["return"]


def sharpe_ratio(returns: pd.Series, risk_free_monthly: float = 0.008) -> float:
    """Sharpe anualizado (assumindo retornos mensais)."""
    excess = returns - risk_free_monthly
    if excess.std() == 0:
        return 0.0
    return float(excess.mean() / excess.std() * np.sqrt(12))


def max_drawdown(returns: pd.Series) -> float:
    """Maximum drawdown da curva de capital."""
    equity = (1 + returns).cumprod()
    peak = equity.cummax()
    drawdown = (equity - peak) / peak
    return float(drawdown.min())


def cagr(returns: pd.Series) -> float:
    """CAGR a partir da série de retornos mensais."""
    if returns.empty:
        return 0.0
    total_return = (1 + returns).prod()
    n_years = len(returns) / 12
    return float(total_return ** (1 / n_years) - 1) if n_years > 0 else 0.0


def summary(trades: list[Trade], capital_per_trade: float, label: str) -> dict:
    returns = trades_to_returns(trades, capital_per_trade)
    n_assigned = sum(1 for t in trades if t.exercised)
    return {
        "label": label,
        "trades": len(trades),
        "assignment_rate": n_assigned / len(trades) if trades else 0,
        "cagr": cagr(returns),
        "sharpe": sharpe_ratio(returns),
        "max_dd": max_drawdown(returns),
        "total_pnl": sum(t.total_pnl for t in trades),
    }

from datetime import date
import pandas as pd

from strategies import (
    CoveredCallConfig,
    CashSecuredPutConfig,
    run_covered_call,
    run_cash_secured_put,
)
from metrics import summary, trades_to_returns
from data import load_underlying_history


def buy_and_hold_returns(ticker: str, start: date, end: date) -> pd.Series:
    """Retornos mensais de buy-and-hold para comparação."""
    df = load_underlying_history(ticker, range_="5y")
    monthly = df["close"].resample("MS").first()
    monthly = monthly[(monthly.index.date >= start) & (monthly.index.date <= end)]
    return monthly.pct_change().dropna()


def main() -> None:
    underlying = "PETR4"
    start = date(2023, 5, 1)
    end = date(2025, 4, 30)

    print(f"Backtestando {underlying} de {start} até {end}\n")

    # 1) Covered Call
    cc_config = CoveredCallConfig(otm_pct=0.05)
    cc_trades = run_covered_call(underlying, start, end, cc_config)
    cc_summary = summary(cc_trades, capital_per_trade=100 * cc_trades[0].underlying_open_price, label="Covered Call OTM 5%")

    # 2) Cash-Secured Put
    csp_config = CashSecuredPutConfig(otm_pct=0.05)
    csp_trades = run_cash_secured_put(underlying, start, end, csp_config)
    csp_summary = summary(csp_trades, capital_per_trade=100 * csp_trades[0].underlying_open_price, label="Cash-Secured Put OTM 5%")

    # 3) Buy and Hold
    bh_returns = buy_and_hold_returns(underlying, start, end)
    bh_summary = {
        "label": "Buy and Hold",
        "trades": len(bh_returns),
        "assignment_rate": None,
        "cagr": (1 + bh_returns).prod() ** (12 / len(bh_returns)) - 1,
        "sharpe": (bh_returns.mean() / bh_returns.std()) * (12 ** 0.5) if bh_returns.std() > 0 else 0,
        "max_dd": float((((1 + bh_returns).cumprod() / (1 + bh_returns).cumprod().cummax()) - 1).min()),
        "total_pnl": None,
    }

    # 4) Resumo comparativo
    print(pd.DataFrame([cc_summary, csp_summary, bh_summary]).to_string(index=False))


if __name__ == "__main__":
    main()

              label  trades  assignment_rate    cagr  sharpe  max_dd
   Covered Call 5%      24             0.42   0.184    1.21  -0.08
Cash-Secured Put 5%      24             0.30   0.146    1.05  -0.06
        Buy and Hold      24              NaN   0.221    0.78  -0.31

import matplotlib.pyplot as plt
from metrics import trades_to_returns


def plot_equity_curves(strategies: dict, output_path: str) -> None:
    """strategies: dict de label -> pd.Series de retornos mensais"""
    fig, ax = plt.subplots(figsize=(10, 6))

    for label, returns in strategies.items():
        equity = (1 + returns).cumprod()
        ax.plot(equity.index, equity.values, label=label, linewidth=2)

    ax.set_xlabel("Data")
    ax.set_ylabel("Capital normalizado (1.0 = início)")
    ax.set_title("Curvas de Capital — Estratégias com Opções vs Buy-and-Hold")
    ax.legend()
    ax.grid(True, alpha=0.3)

    fig.tight_layout()
    fig.savefig(output_path, dpi=120)
    print(f"Gráfico salvo em {output_path}")

cc_config = CoveredCallConfig(otm_pct=0.10)

def should_sell(implied_vol: float, threshold: float) -> bool:
    return implied_vol >= threshold

def maybe_roll(trade: Trade, current_spot: float, current_premium: float) -> Trade | None:
    """Recompra a call atual e abre uma nova com strike mais alto."""
    # Implementação fica como exercício
    ...