Bloom Filter em Python Puro: Como Verificar se 10 Milhões de E-mails São Spam Sem Armazenar Nenhum Deles (30 Linhas de Código)

Você tem uma lista negra com 10 milhões de e-mails de spam. Cada consulta precisa responder em microssegundos. Armazenar tudo em memória? Esquece. Consultar um banco de dados? Lento demais. A solução está numa estrutura de dados de 1970 que usa 12 MB para representar o que ocuparia 400 MB — e erra de propósito.

O Problema que Ninguém Te Conta

Todo sistema de filtragem esbarra na mesma parede: como verificar se algo está num conjunto gigante sem carregar o conjunto inteiro na memória?

A resposta ingênua é um set() do Python. Funciona. Até você precisar armazenar 10 milhões de strings e perceber que o processo consome 2 GB de RAM. A resposta corporativa é Redis ou um banco. Funciona também. Até você descobrir que cada lookup custa 2ms e seu endpoint precisa responder em 500μs.

O Bloom Filter resolve os dois problemas com uma troca elegante: aceita falsos positivos em troca de economia absurda de espaço.

O que é um Bloom Filter (Sem Enrolação)

É um array de bits + várias funções hash. Só isso.

Para inserir um elemento:
1. Passe o elemento por K funções hash diferentes
2. Cada hash aponta uma posição no array de bits
3. Marque todas essas posições como 1

Para consultar um elemento:
1. Passe o elemento pelas mesmas K funções
2. Se todas as posições forem 1 → “provavelmente está”
3. Se qualquer posição for 0 → “definitivamente não está”

O truque: falsos positivos acontecem porque colisões de hash podem marcar posições que coincidem com outros elementos. Mas falsos negativos são impossíveis. Se uma posição é 0, o elemento nunca foi inserido.

Implementação em 30 Linhas (Python Puro, Sem Dependências)

import hashlib
import math

class BloomFilter:
    def __init__(self, expected_items: int, false_positive_rate: float = 0.01):
        """
        Inicializa o filtro com tamanho ótimo calculado matematicamente.
        
        Args:
            expected_items: Número esperado de elementos
            false_positive_rate: Taxa aceitável de falsos positivos (0.01 = 1%)
        """
        self.size = self._optimal_size(expected_items, false_positive_rate)
        self.hash_count = self._optimal_hash_count(self.size, expected_items)
        self.bit_array = bytearray((self.size + 7) // 8)  # 8 bits por byte
        self.count = 0
    
    def _optimal_size(self, n: int, p: float) -> int:
        """Calcula o tamanho ótimo do array: m = -(n * ln(p)) / (ln(2))^2"""
        return int(-n  math.log(p) / (math.log(2) * 2))
    
    def _optimal_hash_count(self, m: int, n: int) -> int:
        """Calcula o número ótimo de funções hash: k = (m/n) * ln(2)"""
        return max(1, int((m / n) * math.log(2)))
    
    def _get_bit(self, index: int) -> bool:
        """Lê um bit específico do array."""
        byte_index = index // 8
        bit_index = index % 8
        return bool(self.bit_array[byte_index] & (1 << bit_index))
    
    def _set_bit(self, index: int):
        """Define um bit específico como 1."""
        byte_index = index // 8
        bit_index = index % 8
        self.bit_array[byte_index] |= (1 << bit_index)
    
    def _get_hashes(self, item: str) -> list[int]:
        """Gera K hashes usando double hashing (Kirsch-Mitzenmacher)."""
        h1 = int(hashlib.md5(item.encode()).hexdigest(), 16)
        h2 = int(hashlib.sha1(item.encode()).hexdigest(), 16)
        return [(h1 + i * h2) % self.size for i in range(self.hash_count)]
    
    def add(self, item: str):
        """Adiciona um elemento ao filtro."""
        for pos in self._get_hashes(item):
            self._set_bit(pos)
        self.count += 1
    
    def __contains__(self, item: str) -> bool:
        """Verifica se um elemento provavelmente está no filtro."""
        return all(self._get_bit(pos) for pos in self._get_hashes(item))

Testando na Prática: Filtro de Spam

Vamos criar um filtro com 1 milhão de e-mails de spam e testar a precisão:

import time

# Cria o filtro para 1M de itens com 1% de falsos positivos
spam_filter = BloomFilter(expected_items=1_000_000, false_positive_rate=0.01)

print(f"Tamanho do array: {spam_filter.size:,} bits")
print(f"Memória usada: {len(spam_filter.bit_array) / 1024 / 1024:.2f} MB")
print(f"Funções hash: {spam_filter.hash_count}")

# Popula com e-mails fictícios
start = time.perf_counter()
for i in range(1_000_000):
    spam_filter.add(f"spam{i}@malicious.com")
populate_time = time.perf_counter() - start

print(f"\nTempo de inserção: {populate_time:.3f}s")
print(f"Velocidade: {1_000_000 / populate_time:,.0f} itens/segundo")

# Testa falsos positivos
false_positives = 0
test_size = 100_000

start = time.perf_counter()
for i in range(test_size):
    if f"legit{i}@good.com" in spam_filter:
        false_positives += 1
query_time = time.perf_counter() - start

fp_rate = false_positives / test_size
print(f"\nTeste de {test_size:,} e-mails legítimos:")
print(f"Falsos positivos: {false_positives} ({fp_rate:.4%})")
print(f"Taxa real vs esperada: {fp_rate / 0.01:.2f}x")
print(f"Tempo de consulta: {query_time:.3f}s ({query_time/test_size*1e6:.2f}μs por consulta)")

# Testa falsos negativos (deve ser ZERO)
false_negatives = sum(1 for i in range(1000) if f"spam{i}@malicious.com" not in spam_filter)
print(f"\nFalsos negativos: {false_negatives} (deve ser 0)")

Resultado típico:

Tamanho do array: 9.585.059 bits
Memória usada: 1.14 MB
Funções hash: 7

Tempo de inserção: 2.847s
Velocidade: 351.246 itens/segundo

Teste de 100.000 e-mails legítimos:
Falsos positivos: 987 (0.9870%)
Taxa real vs esperada: 0.99x
Tempo de consulta: 0.234s (2.34μs por consulta)

Falsos negativos: 0 (deve ser 0)

1.14 MB para representar 1 milhão de elementos. Um set() equivalente consumiria ~80 MB.

O Perrengue do Olivetto: Quando 0.01% Virou 90%

Em 2024, implementei um Bloom Filter para deduplicar URLs crawleadas. Configurei false_positive_rate=0.0001 (0.01%) porque “quanto menor, melhor”.

O problema? Não ajustei o expected_items para o volume real. O filtro foi dimensionado para 100K URLs, mas o crawler processou 5M. Resultado: a taxa de falsos positivos saltou para 90%. Metade das URLs novas eram descartadas como duplicatas.

O debug foi humilhante: adicionei logging de URLs rejeitadas e percebi que 9 em cada 10 eram legítimas. A lição: o expected_items é tão crítico quanto o false_positive_rate. Se você subestimar o volume, a matemática desmorona.

A correção: implementei um filtro dinâmico que dobra de tamanho quando atinge 70% da capacidade. Sim, perde-se a imutabilidade, mas ganha-se resiliência.

A Matemática Por Trás da Mágica

O tamanho ótimo do array é calculado por:

m = -(n × ln(p)) / (ln(2))²

Onde:
– m = tamanho do array em bits
– n = número esperado de elementos
– p = taxa de falsos positivos desejada

Para 1M de itens com 1% de erro:
– m = -(1.000.000 × ln(0.01)) / (ln(2))²
– m = -(1.000.000 × -4.605) / 0.480
– m = 9.585.058 bits ≈ 1.14 MB

O número ótimo de funções hash é:

k = (m/n) × ln(2)

Para o exemplo:
– k = (9.585.058 / 1.000.000) × 0.693
– k ≈ 6.64 → arredonda para 7

Menos funções = mais colisões. Mais funções = array satura rápido. O ótimo é o equilíbrio.

Casos de Uso Reais (Além do Spam)

1. Cache Busting
Antes de consultar o banco, verifique se a chave existe. Se o Bloom Filter disser “não”, economize uma query.

user_cache = BloomFilter(expected_items=500_000, false_positive_rate=0.001)

def get_user(user_id: str) -> dict:
    if user_id not in user_cache:
        return None  # Definitivamente não existe, nem consulta o DB
    
    # Pode existir (falso positivo), consulta o DB
    return db.query("SELECT * FROM users WHERE id = ?", user_id)

2. Deduplicação de Logs
Filtre eventos repetidos antes de gravar no disco.

seen_events = BloomFilter(expected_items=10_000_000, false_positive_rate=0.001)

def process_log(event_hash: str):
    if event_hash in seen_events:
        return  # Provavelmente duplicado, descarta
    
    seen_events.add(event_hash)
    write_to_disk(event_hash)

3. Rate Limiting Distribuído
Marque IPs que já excederam o limite sem armazenar timestamps.

blocked_ips = BloomFilter(expected_items=100_000, false_positive_rate=0.0001)

def should_block(ip: str) -> bool:
    if ip in blocked_ips:
        return True  # Provavelmente já bloqueado
    
    if count_requests(ip) > LIMIT:
        blocked_ips.add(ip)
        return True
    
    return False

Limitações (Sim, Existem)

1. Não deleta elementos
Uma vez marcado, um bit não pode ser desmarcado sem afetar outros elementos. Solução: Counting Bloom Filter (usa contadores em vez de bits), mas quadruplica a memória.

2. Falsos positivos são permanentes
Se o filtro disser “provavelmente está”, você precisa de uma fonte de verdade secundária para confirmar.

3. Não lista elementos
O filtro não armazena os dados, só a “impressão digital”. Não dá para iterar sobre os elementos.

4. Dimensionamento é crítico
Subestimar expected_items destrói a precisão. Superestimar desperdiça memória.

Comparação: Bloom Filter vs Alternativas

Estrutura	Memória (1M itens)	Lookup	Falso Positivo	Falso Negativo
set() Python	~80 MB	O(1)	0%	0%
Redis SET	~100 MB	~2ms	0%	0%
Bloom Filter (1%)	1.14 MB	2μs	1%	0%
Bloom Filter (0.1%)	1.71 MB	3μs	0.1%	0%

Persistência: Salvando e Carregando o Filtro

import pickle

def save_filter(filter_obj: BloomFilter, filepath: str):
    """Serializa o filtro para disco."""
    with open(filepath, 'wb') as f:
        pickle.dump({
            'size': filter_obj.size,
            'hash_count': filter_obj.hash_count,
            'bit_array': filter_obj.bit_array,
            'count': filter_obj.count
        }, f)

def load_filter(filepath: str) -> BloomFilter:
    """Carrega um filtro salvo."""
    with open(filepath, 'rb') as f:
        data = pickle.load(f)
    
    filter_obj = BloomFilter.__new__(BloomFilter)
    filter_obj.size = data['size']
    filter_obj.hash_count = data['hash_count']
    filter_obj.bit_array = data['bit_array']
    filter_obj.count = data['count']
    return filter_obj

# Uso
spam_filter = BloomFilter(expected_items=1_000_000)
spam_filter.add("spam@example.com")
save_filter(spam_filter, "spam_filter.pkl")

# Depois, em outro processo
loaded_filter = load_filter("spam_filter.pkl")
assert "spam@example.com" in loaded_filter

A Pergunta que Fica

Você tem algum sistema que sofre com lookups em conjuntos gigantes? Um filtro de spam, deduplicação de logs, ou cache de banco? Onde um Bloom Filter economizaria memória e tempo no seu stack?

Testa a implementação acima com seus dados reais. Mede a taxa de falsos positivos. Compara com o set() puro. E me conta nos comentários: vale a troca de precisão por performance?

O código está pronto pra copiar. A matemática está explicada. A decisão é sua.

—

Box: A Fórmula Mágica

Para calcular rapidamente o tamanho do filtro:

def bloom_filter_size(n: int, p: float) -> tuple[int, int]:
    """
    Retorna (tamanho em bits, memória em MB) para n itens e taxa p.
    
    Exemplo: bloom_filter_size(1_000_000, 0.01) → (9585059, 1.14)
    """
    m = int(-n  math.log(p) / (math.log(2) * 2))
    mb = (m / 8) / 1024 / 1024
    return m, round(mb, 2)

# Tabela rápida
for n in [100_000, 1_000_000, 10_000_000]:
    for p in [0.01, 0.001, 0.0001]:
        bits, mb = bloom_filter_size(n, p)
        print(f"{n:>10,} itens, {p:.4%} erro → {mb:>6.2f} MB")

100.000 itens, 1.0000% erro →   0.11 MB
   100.000 itens, 0.1000% erro →   0.17 MB
   100.000 itens, 0.0100% erro →   0.23 MB
 1.000.000 itens, 1.0000% erro →   1.14 MB
 1.000.000 itens, 0.1000% erro →   1.71 MB
 1.000.000 itens, 0.0100% erro →   2.29 MB
10.000.000 itens, 1.0000% erro →  11.43 MB
10.000.000 itens, 0.1000% erro →  17.14 MB
10.000.000 itens, 0.0100% erro →  22.85 MB

10 milhões de elementos com 0.01% de erro cabem em 23 MB. Tenta fazer isso com um set().