Capítulo 37, Projetos

Projeto Intermediário: relatório de vendas

Três tabelas, uma junção que não pode perder pedido, um contrato que descreve o dado, e relatórios que respondem às perguntas de uma gerência: mensal, por categoria e canal, melhores clientes. Faça depois do capítulo 26.

O problema

Uma loja tem clientes, produtos e pedidos, em três arquivos. A gerência quer saber quanto vendeu por mês, por categoria e por canal, quem são os melhores clientes e qual é a tendência. E há uma armadilha escondida: 12 pedidos citam clientes que não existem (os códigos 61 a 65).

Pergunta	Ferramenta	Capítulo
Quanto vendeu por mês, e como variou?	`groupby`, `pct_change`	13, 18
Receita por categoria e canal	`pivot_table`	22
Quem são os melhores clientes?	`groupby`, ordenação com desempate	12, 23
Qual a tendência dos últimos dias?	`resample`, `rolling`	24
O dado está certo?	Um contrato de validação	33

As decisões que eu tomei

A junção nunca descarta pedido. O montar_vendas usa how="left" e validate="m:1": o validate falha se uma chave se repetir (o que multiplicaria linhas), e o left mantém os pedidos dos clientes que não existem, marcados com cliente_cadastrado=False e o segmento Sem cadastro. O teste que mais importa neste projeto confere que o número de linhas depois da junção é igual ao de antes.

O contrato descreve o dado, e devolve todos os problemas. Um cliente inexistente é um aviso (o relatório o mostra), mas uma quantidade negativa, um canal desconhecido, um desconto de 150% ou um id repetido interrompem a execução.

Os tipos são declarados na leitura (parse_dates, category), em vez de o Pandas adivinhar.

Os relatórios são funções puras sobre a tabela de vendas, e podem ser testados com cinco pedidos escritos à mão, onde cada total é conhecido.

O código

A leitura declara os tipos, e o contrato valida o que chegou:

projetos/vendas/src/vendas/carga.py

from pathlib import Path

import pandas as pd


def ler_tabelas(pasta: Path) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
    """Lê as três tabelas com os tipos declarados, em vez de deixar o Pandas adivinhar."""
    clientes = pd.read_csv(
        pasta / "clientes.csv", parse_dates=["data_cadastro"], dtype={"segmento": "category"}
    )
    produtos = pd.read_csv(pasta / "produtos.csv", dtype={"categoria": "category"})
    pedidos = pd.read_csv(
        pasta / "pedidos.csv", parse_dates=["data_pedido"], dtype={"canal": "category"}
    )
    return clientes, produtos, pedidos

projetos/vendas/src/vendas/contrato.py

import pandas as pd

CANAIS = {"site", "loja", "app"}


class ContratoViolado(ValueError):
    pass


def validar_pedidos(
    pedidos: pd.DataFrame, clientes: pd.DataFrame, produtos: pd.DataFrame
) -> list[str]:
    """Devolve TODOS os problemas de uma vez. Pedido de cliente inexistente é aviso, não erro."""
    problemas = []
    if pedidos["id_pedido"].duplicated().any():
        problemas.append("id_pedido repetido")
    for coluna in ("id_pedido", "id_cliente", "id_produto", "quantidade", "data_pedido"):
        if pedidos[coluna].isna().any():
            problemas.append(f"{coluna} com ausentes")
    if (pedidos["quantidade"] <= 0).any():
        problemas.append("quantidade não positiva")
    if not pedidos["canal"].isin(CANAIS).all():
        problemas.append("canal desconhecido")
    if not pedidos["desconto"].dropna().between(0, 1).all():
        problemas.append("desconto fora de 0 a 1")
    if not pedidos["id_produto"].isin(produtos["id_produto"]).all():
        problemas.append("pedido de produto inexistente")
    if clientes["id_cliente"].duplicated().any():
        problemas.append("id_cliente repetido em clientes")
    return problemas


def exigir_contrato(pedidos: pd.DataFrame, clientes: pd.DataFrame, produtos: pd.DataFrame) -> None:
    problemas = validar_pedidos(pedidos, clientes, produtos)
    if problemas:
        raise ContratoViolado("; ".join(problemas))

O modelo é a parte delicada: duas junções, e nenhuma linha a menos. O comentário da função diz o porquê do validate:

projetos/vendas/src/vendas/modelo.py

import pandas as pd


def montar_vendas(
    pedidos: pd.DataFrame, clientes: pd.DataFrame, produtos: pd.DataFrame
) -> pd.DataFrame:
    """Junta as três tabelas SEM descartar pedido: o total antes e depois da junção é o mesmo.

    O validate="m:1" falha se uma chave da direita se repetir (isso multiplicaria linhas).
    Pedidos de clientes que não existem ficam, marcados com cliente_cadastrado=False.
    """
    vendas = pedidos.merge(produtos, on="id_produto", how="left", validate="m:1").merge(
        clientes[["id_cliente", "nome", "segmento"]].rename(columns={"nome": "nome_cliente"}),
        on="id_cliente",
        how="left",
        validate="m:1",
    )
    return vendas.assign(
        cliente_cadastrado=vendas["nome_cliente"].notna(),
        nome_cliente=vendas["nome_cliente"].fillna("Sem cadastro"),
        segmento=vendas["segmento"].astype("object").fillna("Sem cadastro"),
        receita=vendas["quantidade"]
        * vendas["preco_unitario"]
        * (1 - vendas["desconto"].fillna(0)),
        mes=vendas["data_pedido"].dt.to_period("M").astype(str),
    )

Os relatórios são o que a gerência lê. Repare no desempate do top_clientes (receita, depois nome), que o torna reproduzível, e no margins=True do pivô, que acrescenta o total para conferência:

projetos/vendas/src/vendas/relatorios.py

import pandas as pd


def receita_mensal(vendas: pd.DataFrame) -> pd.DataFrame:
    """Receita, pedidos, ticket médio e variação percentual sobre o mês anterior."""
    mensal = (
        vendas.groupby("mes", as_index=False)
        .agg(receita=("receita", "sum"), pedidos=("id_pedido", "count"))
        .sort_values("mes")
        .reset_index(drop=True)
    )
    return mensal.assign(
        ticket_medio=mensal["receita"] / mensal["pedidos"],
        variacao_pct=mensal["receita"].pct_change() * 100,
    )


def receita_por_categoria_e_canal(vendas: pd.DataFrame) -> pd.DataFrame:
    return vendas.pivot_table(
        index="categoria",
        columns="canal",
        values="receita",
        aggfunc="sum",
        margins=True,
        margins_name="Total",
        observed=True,
    )


def top_clientes(vendas: pd.DataFrame, n: int = 5) -> pd.DataFrame:
    """Os n clientes com mais receita, com desempate pelo nome (resultado reproduzível)."""
    return (
        vendas.groupby("nome_cliente", as_index=False)["receita"]
        .sum()
        .sort_values(["receita", "nome_cliente"], ascending=[False, True])
        .head(n)
        .reset_index(drop=True)
    )


def participacao_por_segmento(vendas: pd.DataFrame) -> pd.Series:
    por_segmento = vendas.groupby("segmento")["receita"].sum()
    return (por_segmento / por_segmento.sum()).sort_values(ascending=False)


def media_movel_7_dias(vendas: pd.DataFrame) -> pd.Series:
    """Receita diária (com os dias sem venda valendo zero) e a média dos últimos 7 dias."""
    diaria = vendas.set_index("data_pedido")["receita"].resample("D").sum()
    return diaria.rolling(7).mean()

projetos/vendas/src/vendas/cli.py

import argparse
from pathlib import Path

from vendas.carga import ler_tabelas
from vendas.contrato import exigir_contrato, validar_pedidos
from vendas.modelo import montar_vendas
from vendas.relatorios import (
    media_movel_7_dias,
    participacao_por_segmento,
    receita_mensal,
    receita_por_categoria_e_canal,
    top_clientes,
)


def criar_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(prog="vendas", description="Relatório de vendas")
    parser.add_argument("--pasta", type=Path, default=Path("../../dados"), help="pasta dos CSVs")
    sub = parser.add_subparsers(dest="comando", required=True)
    sub.add_parser("relatorio", help="mostra o relatório")
    exportar = sub.add_parser("exportar", help="grava os resultados em CSV e Parquet")
    exportar.add_argument("--saida", type=Path, required=True)
    return parser


def main(argv: list[str] | None = None) -> int:
    args = criar_parser().parse_args(argv)
    try:
        clientes, produtos, pedidos = ler_tabelas(args.pasta)
        exigir_contrato(pedidos, clientes, produtos)
    except (OSError, ValueError, KeyError) as erro:
        print(f"Erro: {erro}")
        return 1
    vendas = montar_vendas(pedidos, clientes, produtos)
    sem_cadastro = int((~vendas["cliente_cadastrado"]).sum())
    if args.comando == "relatorio":
        total = float(vendas["receita"].sum())
        print(f"pedidos: {len(vendas)} | receita total: {total:,.2f}")
        print(f"pedidos de clientes sem cadastro (mantidos): {sem_cadastro}")
        print("\nReceita mensal")
        print(receita_mensal(vendas).round(1).head(4).to_string(index=False))
        print("\nReceita por categoria e canal")
        print(receita_por_categoria_e_canal(vendas).round(0).to_string())
        print("\nTop 3 clientes")
        print(top_clientes(vendas, 3).round(0).to_string(index=False))
        print("\nParticipação por segmento")
        print((participacao_por_segmento(vendas) * 100).round(1).to_string())
        print(f"\nmédia móvel de 7 dias, último dia: {media_movel_7_dias(vendas).iloc[-1]:.1f}")
        print("\nproblemas no contrato:", validar_pedidos(pedidos, clientes, produtos) or "nenhum")
    else:
        args.saida.mkdir(parents=True, exist_ok=True)
        mensal = receita_mensal(vendas)
        mensal.to_csv(args.saida / "receita_mensal.csv", index=False)
        vendas.to_parquet(args.saida / "vendas.parquet", index=False)
        print(f"{len(mensal)} meses e {len(vendas)} vendas gravados em {args.saida}")
    return 0

Os testes

Cinco pedidos, dois clientes, dois produtos: tudo se confere de cabeça. O pedido 2, por exemplo, tem 2 unidades de um produto de 50 reais com 10% de desconto, então vale 90. O pedido 3 tem desconto ausente, e vale o preço cheio. O pedido 5 é do cliente 99, que não existe:

projetos/vendas/tests/conftest.py

import pandas as pd
import pytest


@pytest.fixture
def clientes() -> pd.DataFrame:
    return pd.DataFrame(
        {
            "id_cliente": [1, 2],
            "nome": ["Alfa", "Beta"],
            "cidade": ["São Paulo", "Curitiba"],
            "segmento": ["Varejo", "Online"],
            "data_cadastro": pd.to_datetime(["2023-01-01", "2023-06-01"]),
        }
    )


@pytest.fixture
def produtos() -> pd.DataFrame:
    return pd.DataFrame(
        {
            "id_produto": [10, 20],
            "produto": ["A", "B"],
            "categoria": ["Casa", "Esporte"],
            "preco_unitario": [100.0, 50.0],
        }
    )


@pytest.fixture
def pedidos() -> pd.DataFrame:
    """Cinco pedidos com resposta conhecida; o último é de um cliente que não existe."""
    return pd.DataFrame(
        {
            "id_pedido": [1, 2, 3, 4, 5],
            "id_cliente": [1, 1, 2, 2, 99],
            "id_produto": [10, 20, 10, 20, 10],
            "quantidade": [1, 2, 1, 4, 1],
            "data_pedido": pd.to_datetime(
                ["2025-01-10", "2025-01-20", "2025-02-05", "2025-02-15", "2025-02-20"]
            ),
            "desconto": [0.0, 0.1, None, 0.5, 0.0],
            "canal": ["site", "loja", "app", "site", "site"],
        }
    )

projetos/vendas/tests/test_modelo.py

import pandas as pd
import pytest

from vendas.modelo import montar_vendas


def test_a_juncao_nao_descarta_pedido(pedidos, clientes, produtos) -> None:
    vendas = montar_vendas(pedidos, clientes, produtos)
    assert len(vendas) == len(pedidos)


def test_pedido_de_cliente_inexistente_fica_marcado(pedidos, clientes, produtos) -> None:
    vendas = montar_vendas(pedidos, clientes, produtos).set_index("id_pedido")
    assert not vendas.loc[5, "cliente_cadastrado"]
    assert vendas.loc[5, "segmento"] == "Sem cadastro"
    assert vendas.loc[1, "segmento"] == "Varejo"


def test_receita_com_resposta_conhecida(pedidos, clientes, produtos) -> None:
    vendas = montar_vendas(pedidos, clientes, produtos).set_index("id_pedido")
    assert vendas.loc[1, "receita"] == 100.0
    assert vendas.loc[2, "receita"] == pytest.approx(2 * 50 * 0.9)
    assert vendas.loc[3, "receita"] == 100.0
    assert vendas.loc[4, "receita"] == pytest.approx(4 * 50 * 0.5)


def test_chave_repetida_em_clientes_e_recusada(pedidos, clientes, produtos) -> None:
    repetidos = pd.concat([clientes, clientes.head(1)])
    with pytest.raises(pd.errors.MergeError):
        montar_vendas(pedidos, repetidos, produtos)

projetos/vendas/tests/test_relatorios.py

import pandas as pd
import pytest

from vendas.modelo import montar_vendas
from vendas.relatorios import (
    media_movel_7_dias,
    participacao_por_segmento,
    receita_mensal,
    receita_por_categoria_e_canal,
    top_clientes,
)


@pytest.fixture
def vendas(pedidos, clientes, produtos) -> pd.DataFrame:
    return montar_vendas(pedidos, clientes, produtos)


def test_receita_mensal_soma_o_total(vendas) -> None:
    mensal = receita_mensal(vendas)
    assert mensal["receita"].sum() == pytest.approx(vendas["receita"].sum())
    assert mensal["mes"].tolist() == ["2025-01", "2025-02"]
    assert mensal["pedidos"].tolist() == [2, 3]


def test_variacao_do_primeiro_mes_e_ausente(vendas) -> None:
    mensal = receita_mensal(vendas)
    assert pd.isna(mensal["variacao_pct"].iloc[0])
    esperado = (mensal["receita"].iloc[1] / mensal["receita"].iloc[0] - 1) * 100
    assert mensal["variacao_pct"].iloc[1] == pytest.approx(esperado)


def test_pivo_fecha_com_o_total(vendas) -> None:
    pivo = receita_por_categoria_e_canal(vendas)
    assert pivo.loc["Total", "Total"] == pytest.approx(vendas["receita"].sum())


def test_top_clientes_inclui_quem_nao_tem_cadastro(vendas) -> None:
    top = top_clientes(vendas, 10)
    assert "Sem cadastro" in top["nome_cliente"].tolist()
    assert top["receita"].sum() == pytest.approx(vendas["receita"].sum())
    assert top["receita"].is_monotonic_decreasing


def test_participacao_soma_1(vendas) -> None:
    assert participacao_por_segmento(vendas).sum() == pytest.approx(1.0)


def test_media_movel_comeca_depois_de_7_dias(vendas) -> None:
    serie = media_movel_7_dias(vendas)
    assert serie.iloc[:6].isna().all() and serie.iloc[6:].notna().all()

projetos/vendas/tests/test_contrato_e_cli.py

from pathlib import Path

import pandas as pd
import pytest

from vendas.cli import main
from vendas.contrato import ContratoViolado, exigir_contrato, validar_pedidos


def test_dados_corretos_nao_tem_problemas(pedidos, clientes, produtos) -> None:
    assert validar_pedidos(pedidos, clientes, produtos) == []
    exigir_contrato(pedidos, clientes, produtos)


def test_todos_os_problemas_aparecem_de_uma_vez(pedidos, clientes, produtos) -> None:
    ruins = pedidos.copy()
    ruins.loc[0, "quantidade"] = -1
    ruins.loc[1, "canal"] = "telefone"
    ruins.loc[2, "desconto"] = 2.0
    ruins.loc[3, "id_pedido"] = ruins.loc[4, "id_pedido"]
    ruins.loc[4, "id_produto"] = 777
    problemas = validar_pedidos(ruins, clientes, produtos)
    assert len(problemas) == 5
    with pytest.raises(ContratoViolado):
        exigir_contrato(ruins, clientes, produtos)


def gravar(pasta: Path, pedidos, clientes, produtos) -> None:
    pedidos.to_csv(pasta / "pedidos.csv", index=False)
    clientes.to_csv(pasta / "clientes.csv", index=False)
    produtos.to_csv(pasta / "produtos.csv", index=False)


def test_relatorio_de_ponta_a_ponta(tmp_path: Path, pedidos, clientes, produtos, capsys) -> None:
    gravar(tmp_path, pedidos, clientes, produtos)
    assert main(["--pasta", str(tmp_path), "relatorio"]) == 0
    saida = capsys.readouterr().out
    assert "pedidos: 5" in saida
    assert "sem cadastro (mantidos): 1" in saida
    assert "problemas no contrato: nenhum" in saida


def test_exportar_grava_os_arquivos(tmp_path: Path, pedidos, clientes, produtos) -> None:
    gravar(tmp_path, pedidos, clientes, produtos)
    destino = tmp_path / "saida"
    assert main(["--pasta", str(tmp_path), "exportar", "--saida", str(destino)]) == 0
    assert len(pd.read_csv(destino / "receita_mensal.csv")) == 2
    assert len(pd.read_parquet(destino / "vendas.parquet")) == 5


def test_contrato_violado_devolve_codigo_1(
    tmp_path: Path, pedidos, clientes, produtos, capsys
) -> None:
    ruins = pedidos.assign(quantidade=-1)
    gravar(tmp_path, ruins, clientes, produtos)
    assert main(["--pasta", str(tmp_path), "relatorio"]) == 1
    assert "quantidade não positiva" in capsys.readouterr().out


def test_com_os_dados_reais_do_livro(capsys) -> None:
    pasta = Path(__file__).resolve().parents[3] / "dados"
    if not (pasta / "pedidos.csv").exists():
        pytest.skip("os dados do livro não estão ao lado do projeto")
    assert main(["--pasta", str(pasta), "relatorio"]) == 0
    saida = capsys.readouterr().out
    assert "pedidos: 800" in saida and "sem cadastro (mantidos): 12" in saida

O teste do contrato estraga o dado de cinco maneiras ao mesmo tempo e confere que as cinco aparecem juntas. E o último roda sobre os arquivos reais e confere os números que o capítulo 21 descobriu à mão: 800 pedidos e 12 sem cadastro.

Rodar

Terminal

cd projetos/vendas
uv sync
uv run pytest
uv run vendas --pasta ../../dados relatorio

Saída

................                                                         [100%]
16 passed

O relatório (execução real)

pedidos: 800 | receita total: 970,851.10
pedidos de clientes sem cadastro (mantidos): 12

Receita mensal
    mes  receita  pedidos  ticket_medio  variacao_pct
2025-01  82156.6       67        1226.2           NaN
2025-02  63464.0       46        1379.7         -22.8
2025-03  92596.7       62        1493.5          45.9
2025-04  65837.6       63        1045.0         -28.9

Receita por categoria e canal
canal             app      loja      site     Total
categoria                                          
Casa          53988.0   77803.0  105830.0  237621.0
Esporte       51362.0   95554.0  124000.0  270915.0
Informática   43765.0   79425.0  124154.0  247345.0
Papelaria     43095.0   69274.0  102601.0  214970.0
Total        192210.0  322056.0  456585.0  970851.0

Top 3 clientes
  nome_cliente  receita
 Gama Comércio  27931.0
Ômega Comércio  27439.0
 Teta Comércio  24552.0

Participação por segmento
segmento
Varejo          42.5
Online          39.3
Atacado         16.4
Sem cadastro     1.8

média móvel de 7 dias, último dia: 2261.2

problemas no contrato: nenhum

Lendo o resultado

Nenhum pedido foi perdido. Os 800 pedidos estão no relatório, e o total do pivô (970.851) é igual ao total geral. Uma junção interna teria entregue 788 pedidos e um total menor, sem erro nenhum.
Os pedidos sem cadastro valem 1,8% da receita, e aparecem como um segmento próprio. Isso permite à gerência decidir se é um problema de cadastro que merece atenção, em vez de a receita simplesmente ser menor sem explicação.
A receita oscila muito de um mês para o outro (de -28,9% a +45,9% nos quatro primeiros meses). Em dados sintéticos, isso é só ruído. Em dados reais, uma variação assim pediria uma investigação antes de qualquer conclusão.
O canal "site" domina em todas as categorias, com entre 44% e 50% da receita de cada uma (e o "app" é o menor em todas).

Desafios

Retenção. Calcule, para cada mês, quantos clientes compraram pela primeira vez e quantos voltaram. Dica: o cumcount do capítulo 23.
Câmbio. Use o merge_asof do capítulo 31 com uma tabela de câmbio trimestral para dar a receita em dólar de cada pedido.
Excel. Acrescente o comando exportar em .xlsx, com uma aba por relatório (capítulo 26).
Contrato de datas. Acrescente ao contrato a regra "nenhum pedido anterior à data de cadastro do cliente", com um teste.