Selic (CPM/COPOM)

Contratos CPM

CPM — B3 COPOM Digital Option contract data.

Ticker format: CPM{month_code}{year2}{C|P}{strike_6digits} Example: CPMZ25C099500 ^^^ → prefix "CPM" ^ → month code "Z" = December ^^ → year "25" = 2025 ^ → option type "C" = call ^^^^^^ → strike "099500" → 99.500

Strike interpretation (Selic Meta context): strike_float = int(strike_6digits) / 1000 # e.g. 99.500 change_bps = round((strike_float - 100) * 100) # e.g. -50 bps

Month codes (B3 standard futures convention): F=1, G=2, H=3, J=4, K=5, M=6, N=7, Q=8, U=9, V=10, X=11, Z=12

Implementation note

CPM options expire on the first business day AFTER the COPOM meeting ends (ExpiryDate), not on the first business day of the meeting month. This module bypasses the generic B3 pipeline (which applies a 6-char ticker filter and a DaysToExp > 0 filter, both incorrect for CPM) and calls the lower-level price-report helpers directly.

ExpiryDate and MeetingEndDate are resolved by joining against the COPOM calendar (bc.copom.calendar()) on meeting month + year extracted from the ticker. The join is a left join so CPM rows are never dropped even if the calendar has gaps for very recently announced future meetings.

data(date)

Fetch B3 end-of-day CPM contract data for a given trade date.

Parameters

date : DateLike Trade date. Accepts DD-MM-YYYY, YYYY-MM-DD, datetime.date, etc.

Returns

pl.DataFrame Columns: TradeDate : date TickerSymbol : str MeetingEndDate : date actual COPOM meeting end date (from BCB calendar) ExpiryDate : date next business day after MeetingEndDate (= B3 CPM contract settlement date) OptionType : str "call" or "put" StrikeChangeBps: int change in bps vs 100.000 strike SettlementPrice: float B3 official "Preço de Referência" from the CSV endpoint, in points (0–100). This is the price shown on the B3 dashboard ("Probabilidades da Taxa Selic Meta"). Null for older dates (> ~1 month) where the CSV endpoint is unavailable. BDaysToExp : int business days from TradeDate to ExpiryDate

Returns empty DataFrame with correct schema if no CPM data
exists for the requested date (weekend, holiday, etc.).

Examples

import pyield as yd df = yd.selic.cpm.data("29-01-2025") df.is_empty() or set(df.schema.keys()) >= { ... "data_referencia", ... "codigo_negociacao", ... "data_fim_reuniao", ... "data_expiracao", ... "tipo_opcao", ... "variacao_strike_bps", ... "preco_ajuste", ... } True

Source code in pyield/selic/cpm.py

def data(date: DateLike) -> pl.DataFrame:
    """
    Fetch B3 end-of-day CPM contract data for a given trade date.

    Parameters
    ----------
    date : DateLike
        Trade date. Accepts DD-MM-YYYY, YYYY-MM-DD, datetime.date, etc.

    Returns
    -------
    pl.DataFrame
        Columns:
            TradeDate      : date
            TickerSymbol   : str
            MeetingEndDate : date   actual COPOM meeting end date (from BCB calendar)
            ExpiryDate     : date   next business day after MeetingEndDate
                                    (= B3 CPM contract settlement date)
            OptionType     : str    "call" or "put"
            StrikeChangeBps: int    change in bps vs 100.000 strike
            SettlementPrice: float  B3 official "Preço de Referência" from the
                                    CSV endpoint, in points (0–100).  This is
                                    the price shown on the B3 dashboard
                                    ("Probabilidades da Taxa Selic Meta").
                                    Null for older dates (> ~1 month) where
                                    the CSV endpoint is unavailable.
            BDaysToExp     : int    business days from TradeDate to ExpiryDate

        Returns empty DataFrame with correct schema if no CPM data
        exists for the requested date (weekend, holiday, etc.).

    Examples
    --------
    >>> import pyield as yd
    >>> df = yd.selic.cpm.data("29-01-2025")
    >>> df.is_empty() or set(df.schema.keys()) >= {
    ...     "data_referencia",
    ...     "codigo_negociacao",
    ...     "data_fim_reuniao",
    ...     "data_expiracao",
    ...     "tipo_opcao",
    ...     "variacao_strike_bps",
    ...     "preco_ajuste",
    ... }
    True
    """
    trade_date = cv.converter_datas(date) if date is not None else None
    if trade_date is None:
        return _empty_schema()

    if not data_negociacao_valida(trade_date):
        logger.warning(
            "Data %s inválida para CPM. Retornando DataFrame vazio.", trade_date
        )
        return _empty_schema()

    try:
        zip_data = _baixar_zip_url(trade_date, relatorio_completo=False)
        if not zip_data:
            return _empty_schema()
        xml_bytes = _extrair_xml_de_zip(zip_data)
        records = _parsear_xml_registros(xml_bytes, "CPM")
    except Exception:
        logger.exception("CPM: falha ao baixar SPR para %s.", trade_date)
        return _empty_schema()

    if not records:
        return _empty_schema()

    df = _converter_para_df(records)
    df = df.rename(_RENOMEAR_COLUNAS_CPM, strict=False)
    df = df.with_columns(data_referencia=trade_date)

    # Extrai tipo de opção (codigo_negociacao[6]) e variação de strike (codigo_negociacao[7:13])
    # inteiramente com expressões de string Polars — sem loops Python.
    df = df.with_columns(
        tipo_opcao=pl.col("codigo_negociacao")
        .str.slice(6, 1)
        .replace({"C": "call", "P": "put"}),
        variacao_strike_bps=(
            pl.col("codigo_negociacao")
            .str.slice(7, 6)
            .cast(pl.Int64, strict=False)
            .floordiv(10)
            .sub(10_000)
            .cast(pl.Int32)
        ),
        # Mês e ano da reunião extraídos das posições 3 e 4-5 do código de negociação.
        # Usados como chaves de junção com o calendário COPOM.
        _mes_reuniao=(
            pl.col("codigo_negociacao")
            .str.slice(3, 1)
            .replace(_MONTH_CODE_STR)
            .cast(pl.Int32, strict=False)
        ),
        _ano_reuniao=(
            pl.col("codigo_negociacao")
            .str.slice(4, 2)
            .cast(pl.Int32, strict=False)
            .add(2000)
        ),
    )

    # Join with COPOM calendar to get MeetingEndDate and the correct ExpiryDate.
    # Import is deferred to avoid a module-level circular dependency risk.
    from pyield.bc import copom  # noqa: PLC0415

    cal = copom.calendar().select(
        _mes_reuniao=pl.col("EndDate").dt.month().cast(pl.Int32),
        _ano_reuniao=pl.col("EndDate").dt.year().cast(pl.Int32),
        data_fim_reuniao=pl.col("EndDate"),
        data_expiracao=pl.col("ExpiryDate"),
    )

    df = df.join(cal, on=["_mes_reuniao", "_ano_reuniao"], how="left").drop(
        "_mes_reuniao", "_ano_reuniao"
    )

    # dias_uteis: dias úteis de data_referencia até data_expiracao.
    df = df.with_columns(
        dias_uteis=bday.count_expr("data_referencia", "data_expiracao").cast(pl.Int32)
    )

    # preco_ajuste: "Preço de Referência" da B3 via endpoint CSV.
    # O XML SPR para contratos de opções não contém este campo;
    # o XML é usado acima apenas para obter a lista de contratos (códigos de negociação).
    precos_ajuste = _fetch_settlement_prices(trade_date)
    if not precos_ajuste.is_empty():
        df = df.join(precos_ajuste, on="codigo_negociacao", how="left")
    else:
        df = df.with_columns(preco_ajuste=pl.lit(None, dtype=pl.Float64))

    return df.select(
        pl.col("data_referencia"),
        pl.col("codigo_negociacao"),
        pl.col("data_fim_reuniao"),
        pl.col("data_expiracao"),
        pl.col("tipo_opcao"),
        pl.col("variacao_strike_bps"),
        pl.col("preco_ajuste"),
        pl.col("dias_uteis"),
    ).sort("data_expiracao", "variacao_strike_bps")

Probabilidades Implícitas

probabilities — Implied COPOM meeting probabilities from CPM option prices.

The CPM contract is a cash-or-nothing European option. Under risk-neutral pricing, the B3 settlement price in points (0–100) encodes the market-implied probability of each Selic change scenario, discounted by the DI rate to expiry (B3 Pricing Manual §3.5).

Probability conventions

RawProb = SettlementPrice * DiscountExp / 100

  Direct risk-neutral probability per B3 Manual §3.5 (inverted):

      p_n(K) = PR_n * exp(+n * r_n) / N

  where
      PR_n  = SettlementPrice  (B3 "Preço de Referência", points 0–100)
      N     = 100              (fixed notional)
      n     = BDaysToExp / 252 (time in years, business-day convention)
      r_n   = ln(1 + DI1Rate)  (continuously-compounded DI1 rate to expiry)
      DI1Rate = flat-forward interpolated DI1 rate from TradeDate to ExpiryDate

  SettlementPrice in cpm.data() is the B3 official "Preço de Referência"
  from the B3 CSV endpoint — the price shown on the B3 dashboard
  ("Probabilidades da Taxa Selic Meta") and the output of B3's
  P1/P2/P3/P4 methodology.  It may be null for dates older than ~1 month
  where the CSV endpoint is unavailable.

  May not sum to 1.0 per meeting due to bid-ask spreads or B3 P1/P2 pricing.
  When BDaysToExp == 0 (meeting-day itself), DiscountExp == 1.0 exactly and
  RawProb reduces to SettlementPrice / 100.

Prob = RawProb / sum(RawProb) within ExpiryDate group Normalized so each meeting sums to exactly 1.0. This is the B3 P3 pricing adjustment. Use this for scenario analysis and charts.

CumProb = cumulative sum of Prob, sorted by StrikeChangeBps ascending.

Notes on null SettlementPrice

CPM contracts are sometimes listed without a settlement price (no B3 official pricing for that strike on that date, or CSV endpoint unavailable for older dates). Strikes with null SettlementPrice are excluded from the probability output because:

1. Their contribution to the normalized distribution is undefined.
2. Polars group_by().agg(sum()) returns 0.0 (not null) for all-null groups, which would break the invariant Prob.sum() == 1.0 per meeting.

As a consequence, a meeting where ALL listed strikes have null prices (e.g. CPMH25 on the January 2025 COPOM day) will not appear in the output. MeetingRank is therefore assigned over the priced meetings only and is always a consecutive sequence [1, 2, …, n].

Notes on DI1 fallback

If DI1 data is unavailable for the trade date (network error, holiday, etc.), DI1Rate falls back to 0.0 and DiscountExp to 1.0, so RawProb reduces to SettlementPrice / 100 — equivalent to the old (incorrect) formula. This degradation is logged as a warning but never raises an exception.

all_meetings(date, option_type='call')

Implied COPOM probabilities for every meeting with CPM contracts trading on date.

Only strikes with a non-null SettlementPrice are included. Meetings where all listed strikes have null prices are excluded entirely (see module-level notes).

Parameters

date : DateLike Trade date. option_type : {"call", "put"} Which side to use. Default "call" (the liquid side in practice).

Returns

pl.DataFrame Columns: TradeDate : Date MeetingEndDate : Date ExpiryDate : Date MeetingRank : Int32 1 = nearest meeting with priced contracts StrikeChangeBps: Int32 sorted ascending within each meeting BDaysToExp : Int32 business days from TradeDate to ExpiryDate SettlementPrice: Float64 B3 "Preço de Referência" in points (0–100) DI1Rate : Float64 flat-forward DI1 rate to ExpiryDate DiscountExp : Float64 exp(BDaysToExp/252 * ln(1+DI1Rate)) RawProb : Float64 SettlementPrice * DiscountExp / 100 Prob : Float64 normalized, sums to 1.0 per meeting CumProb : Float64 cumulative Prob ascending by strike

Sorted by (MeetingRank, StrikeChangeBps).
Returns empty DataFrame with correct schema on missing data.

Examples

import pyield as yd import polars as pl df = yd.selic.probabilities.all_meetings("29-01-2025") df.is_empty() or df["ranking_reuniao"].min() == 1 True sums = df.group_by("data_expiracao").agg(pl.col("prob").sum()) df.is_empty() or (sums["prob"] - 1.0).abs().max() < 1e-9 True

Source code in pyield/selic/probabilities.py

def all_meetings(
    date: DateLike,
    option_type: str = "call",
) -> pl.DataFrame:
    """
    Implied COPOM probabilities for every meeting with CPM contracts
    trading on `date`.

    Only strikes with a non-null SettlementPrice are included.  Meetings
    where all listed strikes have null prices are excluded entirely (see
    module-level notes).

    Parameters
    ----------
    date : DateLike
        Trade date.
    option_type : {"call", "put"}
        Which side to use. Default "call" (the liquid side in practice).

    Returns
    -------
    pl.DataFrame
        Columns:
            TradeDate      : Date
            MeetingEndDate : Date
            ExpiryDate     : Date
            MeetingRank    : Int32   1 = nearest meeting with priced contracts
            StrikeChangeBps: Int32   sorted ascending within each meeting
            BDaysToExp     : Int32   business days from TradeDate to ExpiryDate
            SettlementPrice: Float64 B3 "Preço de Referência" in points (0–100)
            DI1Rate        : Float64 flat-forward DI1 rate to ExpiryDate
            DiscountExp    : Float64 exp(BDaysToExp/252 * ln(1+DI1Rate))
            RawProb        : Float64 SettlementPrice * DiscountExp / 100
            Prob           : Float64 normalized, sums to 1.0 per meeting
            CumProb        : Float64 cumulative Prob ascending by strike

        Sorted by (MeetingRank, StrikeChangeBps).
        Returns empty DataFrame with correct schema on missing data.

    Examples
    --------
    >>> import pyield as yd
    >>> import polars as pl
    >>> df = yd.selic.probabilities.all_meetings("29-01-2025")
    >>> df.is_empty() or df["ranking_reuniao"].min() == 1
    True
    >>> sums = df.group_by("data_expiracao").agg(pl.col("prob").sum())
    >>> df.is_empty() or (sums["prob"] - 1.0).abs().max() < 1e-9
    True
    """
    raw = cpm.data(date)
    if raw.is_empty():
        return _empty_schema()

    df = (
        raw.filter(pl.col("tipo_opcao") == option_type)
        # Excluir strikes sem preço de ajuste — ver docstring do módulo.
        .filter(pl.col("preco_ajuste").is_not_null())
        .pipe(_add_meeting_rank)
        .pipe(_add_probabilities)
        .select(
            "data_referencia",
            "data_fim_reuniao",
            "data_expiracao",
            "ranking_reuniao",
            "variacao_strike_bps",
            "dias_uteis",
            "preco_ajuste",
            "taxa_di1",
            "fator_desconto",
            "prob_bruta",
            "prob",
            "prob_acumulada",
        )
        .sort(["ranking_reuniao", "variacao_strike_bps"])
    )

    return df if not df.is_empty() else _empty_schema()

meeting(date, expiration=None, option_type='call')

Implied COPOM probabilities for a single meeting.

Parameters

date : DateLike Trade date. expiration : DateLike | None ExpiryDate of the target meeting (B3 contract expiry date, i.e. next business day after the meeting end). If None, the nearest meeting with priced contracts is used. option_type : {"call", "put"} Which side to use. Default "call".

Returns

pl.DataFrame Same schema as all_meetings(), filtered to the single meeting. MeetingRank is always 1 in this output (relative to the selected meeting — do not confuse with rank across all meetings).

Examples

import pyield as yd df = yd.selic.probabilities.meeting("29-01-2025") df.is_empty() or abs(df["prob"].sum() - 1.0) < 1e-9 True df.is_empty() or df["prob_acumulada"].tail(1).item() == 1.0 True

Source code in pyield/selic/probabilities.py

def meeting(
    date: DateLike,
    expiration: DateLike | None = None,
    option_type: str = "call",
) -> pl.DataFrame:
    """
    Implied COPOM probabilities for a single meeting.

    Parameters
    ----------
    date : DateLike
        Trade date.
    expiration : DateLike | None
        ExpiryDate of the target meeting (B3 contract expiry date,
        i.e. next business day after the meeting end).
        If None, the nearest meeting with priced contracts is used.
    option_type : {"call", "put"}
        Which side to use. Default "call".

    Returns
    -------
    pl.DataFrame
        Same schema as all_meetings(), filtered to the single meeting.
        MeetingRank is always 1 in this output (relative to the
        selected meeting — do not confuse with rank across all meetings).

    Examples
    --------
    >>> import pyield as yd
    >>> df = yd.selic.probabilities.meeting("29-01-2025")
    >>> df.is_empty() or abs(df["prob"].sum() - 1.0) < 1e-9
    True
    >>> df.is_empty() or df["prob_acumulada"].tail(1).item() == 1.0
    True
    """
    df = all_meetings(date, option_type=option_type)
    if df.is_empty():
        return _empty_schema()

    if expiration is None:
        target_expiry = df.filter(pl.col("ranking_reuniao") == 1)["data_expiracao"][0]
    else:
        target_expiry = converter_datas(expiration)

    return df.filter(pl.col("data_expiracao") == target_expiry).with_columns(
        ranking_reuniao=pl.lit(1, dtype=pl.Int32)
    )