A Nova MAM é uma biblioteca de alta performance para Modelagem de Atribuição de Marketing (Marketing Attribution Models) e análise de jornadas de conversão de clientes. Ela foi completamente migrada e otimizada utilizando Polars, resultando em processamentos ordens de grandeza mais rápidos e uso extremamente eficiente de memória RAM, mantendo estrita paridade matemática com as regras de negócio legadas.
A biblioteca requer Python 3.12+ e as dependências listadas no pyproject.toml.
Para instalar em modo de desenvolvimento em seu ambiente virtual:
# Certifique-se de ativar o seu ambiente virtual
source .venv/bin/activate
# Instale o pacote e suas dependências
pip install -e .[dev]
nova_mam/
├── mam/
│ ├── __init__.py
│ ├── core.py # Orquestrador unificado da API (Classe principal MAM)
│ ├── preprocessing.py # Pipeline de ingestão polars para os 3 formatos
│ ├── analysis.py # Journey Analysis Toolbox (JAToolbox)
│ ├── reporting.py # Gerador de relatórios HTML (Jinja2/Plotly) e JSON BI
│ ├── results.py # Wrapper unificado de retorno dos resultados
│ └── models/
│ ├── __init__.py
│ ├── base.py # Classe base para modelos
│ ├── heuristics.py # Modelos: Last Click, First Click, Linear, Position Based, Time Decay
│ ├── markov.py # Modelo Algorítmico baseado em Cadeias de Markov estável
│ └── shapley.py # Modelo Algorítmico baseado em Teoria dos Jogos Cooperativos (Shapley)
O orquestrador principal mam.core.MAM detecta automaticamente ou configura a ingestão dos dados com base no parâmetro format_type.
format_type="session")Cada linha representa um ponto de contato individual no tempo.
import polars as pl
from mam.core import MAM
df = pl.DataFrame({
"user_id": ["u1", "u1", "u1", "u2", "u2"],
"datetime": ["2026-01-01 10:00:00", "2026-01-01 11:00:00", "2026-01-01 12:00:00", "2026-01-01 09:00:00", "2026-01-01 15:00:00"],
"channel": ["Google_Search", "Meta_Ads", "Direct", "Organic_Search", "Email"],
"conversion": [False, False, True, False, True]
})
mam = MAM(
df=df,
format_type="session",
channels_colname="channel",
journey_with_conv_colname="conversion",
datetime_colname="datetime",
user_id_colname="user_id"
)
format_type="journey")Cada linha representa uma jornada completa de um usuário, com os canais e tempos encadeados por um separador (ex: " > ").
df = pl.DataFrame({
"journey_id": ["j1", "j2"],
"journey": ["Google_Search > Meta_Ads > Direct", "Organic_Search > Email"],
"conversion": [True, True],
"time_till_end": ["72.0 > 36.0 > 0.0", "120.0 > 0.0"] # tempos decrescentes (horas)
})
mam = MAM(
df=df,
format_type="journey",
channels_colname="journey",
journey_with_conv_colname="conversion",
time_till_conv_colname="time_till_end"
)
format_type="grouped_journey")Caminhos agregados sem carimbo temporal, contendo uma coluna de peso/frequência (occurrences).
df = pl.DataFrame({
"journey": ["Google_Search > Meta_Ads", "Organic_Search > Direct"],
"conversion": [True, False],
"occurrences": [150, 430]
})
mam = MAM(
df=df,
format_type="grouped_journey",
channels_colname="journey",
journey_with_conv_colname="conversion",
occurrences_colname="occurrences"
)
A JAToolbox oferece ferramentas estatísticas poderosas para análise exploratória de touchpoints e caminhos de conversão, totalmente integrada com Polars (e com suporte automático a Pandas).
Há duas formas principais de utilizar a JAToolbox:
MAM (Recomendado)Se você já instanciou o orquestrador MAM, você pode acessar a JAToolbox pré-configurada diretamente pela propriedade .jatoolbox:
from mam import MAM
# 1. Instancia o orquestrador MAM normalmente
mam = MAM(df=df, format_type="journey", channels_colname="journey", journey_with_conv_colname="conversion")
# 2. Acessa a JAToolbox já pré-configurada com os dados unificados
jatoolbox = mam.jatoolbox
Você também pode instanciar a JAToolbox de forma avulsa. Ela aceita DataFrames do Polars ou Pandas (convertendo de Pandas para Polars sob o capô automaticamente):
from mam import JAToolbox
# Inicializando com um DataFrame que já possui listas de canais
jatoolbox = JAToolbox(df=df_unificado, channels_col="channels")
Se os seus dados estiverem brutos (não unificados), a JAToolbox é capaz de realizar o pré-processamento interno automático no próprio construtor:
from mam import JAToolbox
# Inicializando e pré-processando dados brutos do Formato 2 (journey) de forma transparente
jatoolbox = JAToolbox(
df=df_bruto,
format_type="journey",
channels_col="jornada_original",
journey_with_conv_colname="conversao_original",
time_till_conv_colname="tempo_para_conversao"
)
Uma vez obtida ou instanciada a sua JAToolbox, você pode realizar diversas análises exploratórias:
# 1. Obter o tamanho das jornadas (quantidade de touchpoints por linha)
tamanhos = jatoolbox.get_size()
# 2. Obter a contagem ponderada (volume real) de aparições por canal
contagens = jatoolbox.get_tps_counts()
# 3. Obter transições ponto a ponto (matriz de transição detalhada)
transicoes = jatoolbox.get_transitions(count=True, norm=True)
# 4. Obter a distribuição de canais por posição da jornada (estágios/etapas de conversão)
estagios_df = jatoolbox.channels_by_tp(max_journey_size=5)
Cada modelo de atribuição é instanciado e executado através de métodos simples que retornam um objeto AttributionResult.
# Last Click (Atribui 100% ao último ponto de contato)
res_last = mam.run_last_click()
# First Click (Atribui 100% ao primeiro ponto de contato)
res_first = mam.run_first_click()
# Linear (Distribui o valor igualmente entre todos os canais)
res_linear = mam.run_linear()
# Position Based (Atribui 40% ao primeiro, 40% ao último e divide 20% no meio)
res_position = mam.run_position_based()
# Time Decay (Decaimento contínuo baseado em meia-vida exponencial - requer dados temporais)
res_decay = mam.run_time_decay(half_life_hours=168.0)
Calcula a probabilidade de conversão e atribui pesos a cada canal usando a taxa de efeito de remoção (Removal Effect).
res_markov = mam.run_markov(transition_to_same_state=False)
# Metadados adicionais do modelo
matriz_transicao = res_markov.metadata["transition_matrix"]
efeito_remocao = res_markov.metadata["removal_effect"]
Aplica conceitos de Teoria dos Jogos Cooperativos para avaliar a contribuição marginal de cada canal nas combinações de jornadas. Ele conta com normalização por jornada, evitando a inflação artificial em caminhos com touchpoints repetidos.
res_shapley = mam.run_shapley(max_size=4, value_column="conv_rate")
# Tabela de conversão agrupada de coalizões gerada
conv_table = res.metadata["conv_table"]
Você pode facilmente gerar relatórios ricos e interativos em HTML, além de exportar dados consolidados em formato JSON ideal para ferramentas de Business Intelligence (BI):
# Gera um One-Page Report em HTML interativo com Plotly e exporta os dados limpos para BI em JSON
mam.generate_report(
models=["last_click", "first_click", "linear", "position_based", "time_decay", "markov", "shapley"],
output_html_path="relatorio_atribuicao.html",
output_json_path="exportacao_bi.json"
)
A biblioteca é exaustivamente testada por meio de testes unitários integrados e de um validador de migração ponta a ponta.
pytest
O script validate_migration.py executa o código legado de forma assíncrona ao lado da Nova MAM em um dataset sintético de grande escala, provando a estrita paridade matemática das heurísticas e calculando o ganho de velocidade (speedup):
python validate_migration.py
Projeto licenciado nos termos internos da DP6. Todas as contribuições devem aderir às convenções descritas no manifesto técnico e passar em 100% da suíte de testes automáticos livres de deprecation warnings.