Muitos algoritmos precisam comparar cada elemento com o próximo: calcular diferenças, detectar mudanças, validar ordem, medir distâncias, encontrar intervalos e analisar séries temporais. itertools.pairwise() transforma um iterável em pares sobrepostos, produzindo (a, b), depois (b, c), depois (c, d), sem carregar toda a fonte na memória.
Neste guia, você aprenderá a usar pairwise com listas, geradores e arquivos, calcular deltas, detectar transições, validar sequências, trabalhar com timestamps, lidar com iteráveis curtos e distinguir pares sobrepostos de lotes não sobrepostos.
Primeiro exemplo
from itertools import pairwise
valores = [10, 15, 12, 20]
for anterior, atual in pairwise(valores):
print(anterior, atual)
O resultado é (10, 15), (15, 12) e (12, 20). Cada elemento intermediário participa de dois pares: como segundo valor de um par e primeiro do próximo.
Iteráveis curtos
Uma fonte vazia ou com apenas um elemento produz zero pares. Não ocorre erro.
list(pairwise([])) # []
list(pairwise([42])) # []
list(pairwise([1, 2])) # [(1, 2)]
Se o algoritmo exige pelo menos dois elementos, valide antes ou detecte a ausência do primeiro par.
Consumo preguiçoso
def contador():
numero = 0
while True:
yield numero
numero += 1
pares = pairwise(contador())
print(next(pares))
print(next(pares))
A função mantém apenas o valor anterior necessário para formar o próximo par. Isso permite trabalhar com streams grandes ou infinitos.
Calculando diferenças
temperaturas = [20.0, 21.5, 19.0, 23.0]
deltas = [atual - anterior for anterior, atual in pairwise(temperaturas)]
O resultado possui um elemento a menos que a entrada. Essa diferença de comprimento é natural, pois não existe valor anterior para o primeiro item.
Taxas de variação
amostras = [(0.0, 10.0), (2.0, 14.0), (5.0, 20.0)]
taxas = []
for (t1, v1), (t2, v2) in pairwise(amostras):
taxas.append((v2 - v1) / (t2 - t1))
Valide timestamps repetidos para evitar divisão por zero. Também confirme que as amostras estão ordenadas pelo tempo.
Detectando mudanças de estado
estados = ["novo", "novo", "pago", "enviado", "enviado"]
transicoes = [
(antes, depois)
for antes, depois in pairwise(estados)
if antes != depois
]
Esse padrão é útil em auditorias, máquinas de estado, logs e jornadas de usuário. Para preservar o índice da mudança, combine com enumerate().
Validando ordem crescente
def esta_ordenado(valores):
return all(a <= b for a, b in pairwise(valores))
Para listas vazias ou unitárias, all() devolve True. Isso segue a lógica de que não existe par fora de ordem. Caso o domínio exija pelo menos dois valores, acrescente uma regra separada.
Detectando duplicatas consecutivas
duplicadas = [a for a, b in pairwise(valores) if a == b]
Isso encontra apenas repetições adjacentes. Para duplicatas em qualquer posição, use um conjunto, Counter ou ordenação adequada.
Intervalos e lacunas
datas = [1, 2, 5, 6, 10]
lacunas = [(a, b) for a, b in pairwise(datas) if b - a > 1]
Em calendários, IDs sequenciais ou amostras, o padrão identifica buracos. Defina se valores repetidos e regressões também devem ser sinalizados.
Distâncias em trajetórias
from math import hypot
pontos = [(0, 0), (3, 4), (6, 4)]
total = sum(
hypot(x2 - x1, y2 - y1)
for (x1, y1), (x2, y2) in pairwise(pontos)
)
A soma mede o comprimento da trajetória, não a distância direta entre início e fim.
Lendo linhas consecutivas
with open("eventos.log", encoding="utf-8") as arquivo:
for linha_anterior, linha_atual in pairwise(arquivo):
comparar(linha_anterior, linha_atual)
O arquivo é consumido de forma incremental. Não guarde referências aos pares se o objetivo for manter memória constante.
Pairwise e enumerate
for indice, (anterior, atual) in enumerate(pairwise(valores), start=1):
print(f"transição {indice - 1}->{indice}: {anterior} para {atual}")
O índice representa a posição do segundo elemento. Ajuste a mensagem conforme a convenção do domínio.
Pairwise e zip
Antes da função oficial, era comum escrever:
pares = zip(valores, valores[1:])
Essa forma funciona com sequências e cria um slice. Para iteradores genéricos, usava-se tee(). pairwise é mais direto e evita boilerplate.
Implementação conceitual
def pairwise_manual(iterable):
iterator = iter(iterable)
try:
anterior = next(iterator)
except StopIteration:
return
for atual in iterator:
yield anterior, atual
anterior = atual
A função oficial segue essa lógica e mantém apenas uma referência anterior.
Diferença para batched
from itertools import batched, pairwise
list(pairwise([1, 2, 3, 4]))
# [(1, 2), (2, 3), (3, 4)]
list(batched([1, 2, 3, 4], 2))
# [(1, 2), (3, 4)]
pairwise cria uma janela de tamanho dois com passo um. batched cria blocos não sobrepostos.
Janelas maiores
pairwise é específico para dois elementos. Para janelas de tamanho três ou mais, use um recipe com deque ou uma biblioteca apropriada.
from collections import deque
def sliding_window(iterable, n):
iterator = iter(iterable)
janela = deque([], maxlen=n)
for item in iterator:
janela.append(item)
if len(janela) == n:
yield tuple(janela)
Séries temporais irregulares
Ao comparar amostras temporais, não suponha intervalos constantes. Calcule o delta de tempo para cada par. Também trate fuso horário, dados fora de ordem e registros duplicados.
Valores ausentes
Se a série contém None, decida se o par deve ser ignorado, imputado ou marcado como inválido.
for a, b in pairwise(valores):
if a is None or b is None:
continue
processar(b - a)
NaN
Em dados numéricos, NaN se propaga e não é igual a si mesmo. Use funções específicas para detectá-lo antes de comparar mudanças ou ordem.
Objetos mutáveis
pairwise guarda uma referência ao elemento anterior. Se a fonte reutiliza e modifica o mesmo objeto a cada yield, ambos os componentes podem apontar para o mesmo objeto mutado. Geradores devem produzir snapshots independentes quando isso importar.
Exceções na fonte
Se o iterável lança uma exceção, pairwise a propaga. O último valor anterior permanece apenas dentro do iterador e será liberado quando ele for descartado.
Async iterators
pairwise aceita iteráveis síncronos. Para fontes assíncronas, implemente um helper com async for:
async def async_pairwise(source):
iterator = aiter(source)
try:
anterior = await anext(iterator)
except StopAsyncIteration:
return
async for atual in iterator:
yield anterior, atual
anterior = atual
Encontrando máximas mudanças
maior = max(
((abs(b - a), a, b) for a, b in pairwise(valores)),
default=None,
)
Use default porque uma entrada com menos de dois elementos não produz candidatos.
Segmentando por mudanças
pairwise ajuda a localizar fronteiras, mas para produzir grupos completos por chave, itertools.groupby() costuma ser mais simples. Use pairwise quando a transição em si é a informação principal.
Erros comuns
- Esperar pares não sobrepostos: use batched para isso.
- Esquecer que o resultado é menor: existem n-1 pares.
- Consumir o iterador duas vezes: a fonte pode ser de passagem única.
- Ignorar timestamps repetidos: taxas podem dividir por zero.
- Confundir duplicata adjacente com global: pairwise vê apenas vizinhos.
- Usar com objetos reutilizados: referências mutáveis podem surpreender.
Exemplo completo: auditoria de eventos
from itertools import pairwise
ORDEM = {
"criado": 0,
"pago": 1,
"separado": 2,
"enviado": 3,
"entregue": 4,
}
def auditar(eventos):
problemas = []
for anterior, atual in pairwise(eventos):
if atual.timestamp < anterior.timestamp:
problemas.append("timestamp regressivo")
if ORDEM[atual.estado] < ORDEM[anterior.estado]:
problemas.append(
f"transição inválida: {anterior.estado} -> {atual.estado}"
)
return problemas
O exemplo valida cronologia e progressão de estado em uma única passagem e memória constante.
Conclusão
itertools.pairwise() expressa de forma clara algoritmos baseados em vizinhos consecutivos. Ele é preguiçoso, funciona com qualquer iterável síncrono e elimina slices ou combinações manuais com tee.
A documentação oficial de itertools.pairwise define a função. Use-a para deltas, transições, lacunas e validações, lembrando que pares se sobrepõem e que entradas curtas produzem nenhum resultado.
