Chapter 37, Projects

Intermediate Project: sales report

Three tables, a join that cannot lose orders, a contract that describes the data, and reports that answer the questions of management: monthly, by category and channel, best customers. Do this after chapter 26.

The problem

A store has customers, products and orders, in three files. Management wants to know how much was sold per month, by category and by channel, who the best customers are, and what the trend is. And there is a hidden trap: 12 orders mention customers that do not exist (codes 61 to 65).

Question	Tool	Chapter
How much was sold per month, and how did it vary?	`groupby`, `pct_change`	13, 18
Revenue by category and channel	`pivot_table`	22
Who are the best customers?	`groupby`, sorting with tiebreaker	12, 23
What is the trend of the last days?	`resample`, `rolling`	24
Is the data correct?	A validation contract	33

The decisions I made

The join never discards orders. The montar_vendas uses how="left" and validate="m:1": the validate fails if a key repeats (which would multiply rows), and the left keeps the orders from non-existent customers, marked with cliente_cadastrado=False and the segment Sem cadastro. The test that matters most in this project checks that the number of rows after the join is equal to that before.

The contract describes the data and returns all the issues. A non-existent customer is a warning (the report shows it), but a negative quantity, an unknown channel, a 150% discount, or a repeated id interrupt the execution.

The types are declared in the read (parse_dates, category), instead of Pandas guessing.

The reports are pure functions over the sales table, and can be tested with five manually written orders, where each total is known.

The code

The read declares the types, and the contract validates what arrived:

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))

The model is the delicate part: two joins, and no fewer rows. The function comment explains why the 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),
    )

The reports are what management reads. Note the tiebreaker of the top_clientes (revenue, then name), which makes it reproducible, and the margins=True of the pivot, which adds the total for verification:

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

The tests

Five orders, two customers, two products: everything checks out in your head. Order 2, for example, has 2 units of a product costing 50 reais with a 10% discount, so it is worth 90. Order 3 has no discount and is worth the full price. Order 5 is from customer 99, who does not exist:

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

The contract test corrupts the data in five ways at the same time and checks that all five appear together. And the last one runs on the real files and checks the numbers that chapter 21 discovered manually: 800 orders and 12 without registration.

Run

Terminal

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

Output

................                                                         [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

Reading the result

No orders were lost. The 800 orders are in the report, and the total of the pivot (970,851) is equal to the overall total. An inner join would have delivered 788 orders and a lower total, without any errors.
Orders without registration account for 1.8% of revenue, and appear as their own segment. This allows management to decide if it is a registration problem that deserves attention, rather than the revenue simply being lower without explanation.
Revenue fluctuates greatly from month to month (from -28.9% to +45.9% in the first four months). In synthetic data, this is just noise. In real data, such a variation would require an investigation before any conclusion.
The "site" channel dominates in all categories, accounting for between 44% and 50% of the revenue of each (and the "app" is the smallest in all).

Challenges

Retention. Calculate, for each month, how many customers made their first purchase and how many returned. Tip: the cumcount from chapter 23.
Exchange. Use the merge_asof from chapter 31 with a quarterly exchange table to give the revenue in dollars for each order.
Excel. Add the command exportar in .xlsx, with one tab per report (chapter 26).
Date contract. Add to the contract the rule "no order prior to the customer's registration date", with a test.