CPM

def contratos(data: DateLike) -> pl.DataFrame:
    """Busca os contratos CPM negociados em uma data.

    Fonte: B3. A lista de contratos vem do boletim de preços (SPR) e o preço de
    ajuste vem do "Preço de Referência" do boletim de derivativos consolidados,
    o mesmo exibido no painel "Probabilidades da Taxa Selic Meta" da B3. As
    datas da reunião e de expiração vêm do calendário do COPOM.

    Args:
        data: Data de negociação.

    Returns:
        DataFrame Polars ordenado por ``data_expiracao`` e
        ``variacao_strike_bps``. Retorna DataFrame vazio, com o schema abaixo,
        se não houver contratos CPM na data.

    Output Columns:
        * data_referencia (Date): data de negociação.
        * codigo_negociacao (String): código do contrato (ex.: ``CPMZ25C099500``).
        * data_fim_reuniao (Date): último dia da reunião do COPOM.
        * data_expiracao (Date): primeiro dia útil após ``data_fim_reuniao``,
            data de liquidação do contrato.
        * tipo_opcao (String): ``"call"`` ou ``"put"``.
        * variacao_strike_bps (Int32): variação da Selic Meta, em pontos-base,
            associada ao strike.
        * preco_ajuste (Float64): Preço de Referência da B3, em pontos (0–100).
        * dias_uteis (Int32): dias úteis de ``data_referencia`` até
            ``data_expiracao``.

    Notes:
        A B3 só publica o Preço de Referência de datas recentes (cerca de um
        mês). Para datas mais antigas, ``preco_ajuste`` é nulo.

    Examples:
        >>> df = yd.cpm.contratos("29-01-2025")  # doctest: +SKIP
    """
    data_negociacao = converter_datas(data)
    df = boletim.buscar(data_negociacao, prefixo_ticker="CPM")
    if df.is_empty():
        return _df_vazio()

    df = df.rename(_RENOMEAR_COLUNAS_CPM, strict=False).with_columns(
        data_referencia=data_negociacao,
        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)
        ),
        _mes_reuniao=pl.col("codigo_negociacao")
        .str.slice(3, 1)
        .replace_strict(_MAPA_MESES, default=None, return_dtype=pl.Int32),
        _ano_reuniao=(
            pl.col("codigo_negociacao")
            .str.slice(4, 2)
            .cast(pl.Int32, strict=False)
            .add(2000)
        ),
    )

    cal = copom.calendario().select(
        _mes_reuniao=pl.col("data_decisao").dt.month().cast(pl.Int32),
        _ano_reuniao=pl.col("data_decisao").dt.year().cast(pl.Int32),
        data_fim_reuniao=pl.col("data_decisao"),
        data_expiracao=pl.col("data_efetividade"),
    )

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

    # O SPR não traz o Preço de Referência das opções; ele vem do CSV consolidado.
    precos = _parsear_precos_referencia(_buscar_csv(data_negociacao))

    return (
        df.join(precos, on="codigo_negociacao", how="left")
        .with_columns(
            dias_uteis=du.contar_expr("data_referencia", "data_expiracao").cast(
                pl.Int32
            )
        )
        .select(_df_vazio().columns)
        .sort("data_expiracao", "variacao_strike_bps")
    )

def probabilidades(
    data: DateLike,
    tipo_opcao: Literal["call", "put"] = "call",
) -> pl.DataFrame:
    """Calcula as probabilidades implícitas das reuniões do COPOM.

    Fonte: B3, a partir dos preços dos contratos CPM (``yd.cpm.contratos``),
    descontados pela curva de DI1. Inclui todas as reuniões com contratos
    negociados na data. Somente strikes com ``preco_ajuste`` não nulo são
    incluídos; reuniões em que todos os strikes têm preço nulo são excluídas.

    Args:
        data: Data de negociação.
        tipo_opcao: Tipo de opção, ``"call"`` ou ``"put"``. O padrão é
            ``"call"``, que na prática é o lado mais líquido.

    Returns:
        DataFrame Polars ordenado por ``ranking_reuniao`` e
        ``variacao_strike_bps``. Retorna DataFrame vazio, com o schema abaixo,
        se não houver dados.

    Output Columns:
        * data_referencia (Date): data de negociação.
        * data_fim_reuniao (Date): último dia da reunião do COPOM.
        * data_expiracao (Date): data de expiração do contrato CPM.
        * ranking_reuniao (Int32): 1 para a reunião mais próxima com preços,
            2 para a seguinte, e assim por diante.
        * variacao_strike_bps (Int32): variação do strike em pontos-base,
            crescente dentro de cada reunião.
        * dias_uteis (Int32): dias úteis até a expiração.
        * preco_ajuste (Float64): Preço de Referência da B3 em pontos (0–100).
        * taxa_di1 (Float64): taxa DI1 interpolada até a expiração.
        * fator_desconto (Float64): fator aplicado na inversão da fórmula da
            B3, ``exp(dias_uteis / 252 * ln(1 + taxa_di1))``.
        * prob_bruta (Float64): probabilidade antes da normalização,
            ``preco_ajuste * fator_desconto / 100``.
        * prob (Float64): probabilidade normalizada,
            ``prob_bruta / soma(prob_bruta)`` por ``data_expiracao``, com soma
            1,0 por reunião.
        * prob_acumulada (Float64): probabilidade acumulada em ordem crescente
            de strike.

    Notes:
        Para uma única reunião, filtre o resultado, por exemplo
        ``df.filter(pl.col("ranking_reuniao") == 1)`` para a mais próxima.

    Examples:
        >>> df = yd.cpm.probabilidades("29-01-2025")  # doctest: +SKIP
        >>> proxima = df.filter(pl.col("ranking_reuniao") == 1)  # doctest: +SKIP
    """
    df = (
        contratos(data)
        .filter(
            pl.col("tipo_opcao") == tipo_opcao,
            pl.col("preco_ajuste").is_not_null(),
        )
        .pipe(_adicionar_ranking_reuniao)
    )
    if df.is_empty():
        return _df_vazio()

    return (
        df.pipe(_adicionar_probabilidades)
        .select(_df_vazio().columns)
        .sort("ranking_reuniao", "variacao_strike_bps")
    )