🏗️ Oscar Niemeyer - Agente de Visualização de Dados
Implementação em src/agents/oscar_niemeyer.py (38KB, 18 métodos). Testes: 70.6% (12/17 passing - core features OK). NOVIDADE Sprint 6: Network Graphs + Choropleth Maps! 🗺️
📋 Visão Geral
Oscar Niemeyer é o agente especializado em agregação inteligente de dados governamentais e geração de metadados otimizados para visualização frontend. Transforma dados brutos em insights visuais compreensíveis através de dashboards, gráficos interativos e mapas.
Inspiração Cultural
Oscar Niemeyer (1907-2012)
- Títulos: Arquiteto modernista brasileiro, criador de Brasília
- Obras: Congresso Nacional, Catedral de Brasília, Museu de Arte Contemporânea de Niterói
- Filosofia: "A vida é um sopro. A arquitetura deve ser uma homenagem à beleza"
- Legado: Transforma conceitos abstratos em formas visuais elegantes e funcionais
- Conexão: Como Niemeyer transformava ideias em arquitetura, o agente transforma dados em visualizações elegantes
🎯 Missão
Preparar dados governamentais para visualização através de agregação multidimensional, otimização para performance frontend, e geração de metadados compatíveis com bibliotecas modernas (Plotly, Chart.js, D3.js). Especialista em transformar milhões de registros em visualizações compreensíveis.
🧠 Capacidades Principais
✅ Agregação Multidimensional (OLAP)
- 9 tipos de agregação: SUM, COUNT, AVG, MEDIAN, MIN, MAX, PERCENTILE, STDDEV, VARIANCE
- Operações OLAP: Slice, Dice, Drill-down, Roll-up
- Pivot tables multidimensionais com subtotals
- Hierarquias: Município → Microrregião → Estado → Região → País
✅ Otimização para Visualização
- Downsampling: LTTB (Largest Triangle Three Buckets) para datasets >10k pontos
- Binning strategies: Equal-width, equal-frequency, custom bins
- Outlier detection: IQR method, Z-score method
- Normalization: Min-max scaling, Z-score standardization, log transformation
✅ Análise Temporal
- STL Decomposition: Trend, seasonal, residual
- Moving averages: SMA, EMA, WMA
- Autocorrelation: ACF, PACF analysis
- Change point detection: CUSUM, Bayesian
✅ Visualizações Suportadas (10 tipos)
- Line Chart, Bar Chart, Pie Chart, Scatter Plot, Heatmap
- Treemap, Sankey, Gauge, Maps (geographic), Tables
- NEW Sprint 6: Network Graphs (fraude), Choropleth Maps (Brasil)
✅ Geoespacial
- Choropleth maps: Estados e municípios brasileiros com GeoJSON
- Hexbin aggregation: Binning hexagonal para mapas
- Clustering: DBSCAN para pontos geográficos
- Regional boundaries: Spatial joins, agregação por polígonos
✅ Network Graphs (Sprint 6) 🔥
- Fraud networks: Visualização de relacionamentos suspeitos
- Community detection: Algoritmo Louvain para identificar anéis de fraude
- Force-directed layout: Spring algorithm NetworkX
- Interactive: Plotly JSON pronto para frontend
📊 Estruturas de Dados
VisualizationType (10 tipos)
class VisualizationType(Enum):
LINE_CHART = "line_chart" # Séries temporais
BAR_CHART = "bar_chart" # Comparações categóricas
PIE_CHART = "pie_chart" # Proporções
SCATTER_PLOT = "scatter_plot" # Correlações
HEATMAP = "heatmap" # Matriz 2D
TREEMAP = "treemap" # Hierarquias
SANKEY = "sankey" # Fluxos
GAUGE = "gauge" # KPIs
MAP = "map" # Mapas geográficos
TABLE = "table" # Tabelas estruturadas
AggregationType (9 tipos)
class AggregationType(Enum):
SUM = "sum" # Soma total
COUNT = "count" # Contagem
AVERAGE = "average" # Média aritmética
MEDIAN = "median" # Mediana
MIN = "min" # Valor mínimo
MAX = "max" # Valor máximo
PERCENTILE = "percentile" # Percentis (25, 50, 75, 95, 99)
STDDEV = "stddev" # Desvio padrão
VARIANCE = "variance" # Variância
TimeGranularity (7 níveis)
class TimeGranularity(Enum):
MINUTE = "minute" # Minuto a minuto
HOUR = "hour" # Por hora
DAY = "day" # Diário
WEEK = "week" # Semanal
MONTH = "month" # Mensal
QUARTER = "quarter" # Trimestral
YEAR = "year" # Anual
DataAggregationResult
@dataclass
class DataAggregationResult:
"""Resultado de agregação multidimensional."""
aggregation_id: str
data_type: str # Tipo dos dados originais
aggregation_type: AggregationType # Tipo de agregação aplicada
time_granularity: Optional[TimeGranularity] # Granularidade temporal
dimensions: List[str] # Dimensões (ex: ['state', 'category'])
metrics: Dict[str, float] # Métricas agregadas
data_points: List[Dict[str, Any]] # Dados agregados
metadata: Dict[str, Any] # Metadados adicionais
timestamp: datetime # Timestamp da agregação
Exemplo:
DataAggregationResult(
aggregation_id="agg_20251013_001",
data_type="government_spending",
aggregation_type=AggregationType.SUM,
time_granularity=TimeGranularity.MONTH,
dimensions=["state", "category"],
metrics={
"total_overall": 1_500_000_000,
"avg_per_state": 55_555_555,
"max_month": 120_000_000
},
data_points=[
{"state": "SP", "month": "2025-01", "total": 50_000_000},
{"state": "RJ", "month": "2025-01", "total": 35_000_000}
],
metadata={"source": "Portal da Transparência"},
timestamp=datetime.now()
)
VisualizationMetadata
@dataclass
class VisualizationMetadata:
"""Metadados para renderização frontend."""
visualization_id: str
title: str # Título do gráfico
subtitle: Optional[str] # Subtítulo
visualization_type: VisualizationType # Tipo de visualização
x_axis: Dict[str, Any] # Config eixo X
y_axis: Dict[str, Any] # Config eixo Y
series: List[Dict[str, Any]] # Séries de dados
filters: Dict[str, Any] # Filtros aplicáveis
options: Dict[str, Any] # Opções do chart
data_url: str # URL para buscar dados
timestamp: datetime
Exemplo:
VisualizationMetadata(
visualization_id="viz_20251013_001",
title="Evolução de Despesas Públicas por Estado",
subtitle="Últimos 12 meses",
visualization_type=VisualizationType.LINE_CHART,
x_axis={
"label": "Mês",
"type": "datetime",
"format": "%b %Y"
},
y_axis={
"label": "Total de Despesas (R$)",
"type": "linear",
"format": ",.0f"
},
series=[
{"name": "SP", "data": [...], "color": "#1f77b4"},
{"name": "RJ", "data": [...], "color": "#ff7f0e"}
],
options={
"legend": {"position": "top"},
"tooltip": {"enabled": True},
"responsive": True
},
data_url="/api/v1/data/spending/states",
timestamp=datetime.now()
)
🔬 Algoritmos Implementados
1. LTTB Downsampling
Largest Triangle Three Buckets - Redução inteligente de pontos preservando forma visual.
def downsample_lttb(data: List[Dict], target_points: int) -> List[Dict]:
"""
Reduz dataset grande para visualização sem perder forma.
Aplicado automaticamente para datasets > 10k pontos.
Target típico: 1000-2000 pontos para performance frontend.
"""
# Divide em buckets
bucket_size = (len(data) - 2) / (target_points - 2)
# Mantém primeiro e último pontos
sampled = [data[0]]
# Para cada bucket, seleciona ponto que forma maior triângulo
for i in range(1, target_points - 1):
# Calcula área do triângulo para cada ponto no bucket
# Seleciona ponto com maior área
pass
sampled.append(data[-1])
return sampled
Vantagens:
- Preserva picos e vales visuais
- Performance: O(n) linear
- Ideal para line charts com muitos dados
2. STL Decomposition (Seasonal-Trend Loess)
Decompõe séries temporais em Trend + Seasonal + Residual.
from statsmodels.tsa.seasonal import seasonal_decompose
decomposition = seasonal_decompose(timeseries, model='additive', period=12)
trend = decomposition.trend # Tendência de longo prazo
seasonal = decomposition.seasonal # Padrão sazonal repetitivo
residual = decomposition.resid # Ruído e anomalias
Aplicações:
- Identificar tendências de gastos (crescente/decrescente)
- Detectar sazonalidade (fim de ano fiscal)
- Anomalias aparecem no residual
3. Louvain Community Detection (Sprint 6)
Detecta anéis de fraude em network graphs via modularidade.
import networkx as nx
from networkx.algorithms import community
# Criar grafo de relacionamentos
G = nx.Graph()
G.add_edges_from(relationships)
# Detectar comunidades
communities = community.greedy_modularity_communities(G)
# Resultado: grupos altamente conectados (possíveis fraudes)
for i, comm in enumerate(communities):
print(f"Community {i}: {len(comm)} entities")
Output: Fraud rings identificados automaticamente.
💻 Exemplos de Uso
Exemplo 1: Agregar Despesas Mensais por Estado
from src.agents.oscar_niemeyer import OscarNiemeyerAgent, AggregationType, TimeGranularity
oscar = OscarNiemeyerAgent()
await oscar.initialize()
# Dados brutos (DataFrame ou dict)
message = AgentMessage(
content="Agregar despesas mensais por estado",
data={
"raw_data": expenses_dataframe, # Colunas: date, state, value
"aggregation": AggregationType.SUM,
"dimensions": ["state"],
"time_dimension": "date",
"time_granularity": TimeGranularity.MONTH
}
)
response = await oscar.process(message, context)
# Resultado agregado
print(response.data["aggregated"])
# {
# "dimensions": ["state"],
# "time_granularity": "MONTH",
# "data_points": [
# {"state": "SP", "month": "2025-01", "total": 50_000_000},
# {"state": "RJ", "month": "2025-01", "total": 35_000_000}
# ],
# "metrics": {
# "total_overall": 1_500_000_000,
# "avg_per_state": 55_555_555
# }
# }
Exemplo 2: Network Graph de Fraude (Sprint 6) 🔥
# ==========================================
# NETWORK GRAPH PARA VISUALIZAÇÃO DE FRAUDE
# ==========================================
# Caso de uso: Oxóssi detectou relacionamentos suspeitos
# Oscar transforma em grafo interativo com community detection
message = AgentMessage(
sender="oxossi", # Agente que detectou a fraude
recipient="OscarNiemeyerAgent",
action="network_graph", # Ação: criar grafo de relacionamentos
payload={
# ==========================================
# ENTIDADES (NODOS DO GRAFO)
# ==========================================
# Cada entidade é um nodo no grafo
# Score: 0-1 (suspeita de envolvimento em fraude)
"entities": [
{
"id": "supplier_001", # ID único da entidade
"name": "Empresa ABC Ltda",
"type": "empresa", # Tipo: empresa, servidor, intermediário
"score": 0.85 # 85% de suspeita (COR VERMELHA no grafo)
},
{
"id": "official_042",
"name": "João Silva",
"type": "servidor", # Servidor público
"score": 0.72 # 72% de suspeita (COR LARANJA)
},
{
"id": "supplier_015",
"name": "Fornecedor XYZ",
"type": "empresa",
"score": 0.45 # 45% de suspeita (COR AMARELA)
}
],
# ==========================================
# RELACIONAMENTOS (ARESTAS DO GRAFO)
# ==========================================
# Cada relationship é uma aresta conectando dois nodos
# Strength: 0-1 (força do relacionamento suspeito)
"relationships": [
{
"source": "supplier_001", # Origem: Empresa ABC
"target": "official_042", # Destino: Servidor João Silva
"type": "contracts_with", # Tipo: contratos firmados
"strength": 0.9 # 90% - RELAÇÃO FORTE (aresta grossa)
},
{
"source": "supplier_015", # Origem: Fornecedor XYZ
"target": "official_042", # Destino: mesmo servidor
"type": "same_address", # RED FLAG: mesmo endereço (empresa fantasma?)
"strength": 0.8 # 80% - RELAÇÃO FORTE
}
],
# Threshold de exibição: só mostra entidades/relações > 0.7
# Filtra ruído e foca em relacionamentos mais suspeitos
"threshold": 0.7
}
)
# ==========================================
# PROCESSAR E GERAR VISUALIZAÇÃO
# ==========================================
# Oscar:
# 1. Cria grafo com NetworkX
# 2. Aplica algoritmo Louvain para detectar comunidades (fraud rings)
# 3. Usa force-directed layout (spring algorithm) para posicionar nodos
# 4. Gera JSON compatível com Plotly para renderização frontend
response = await oscar.process(message, context)
# ==========================================
# METADATA DO GRAFO (ANÁLISE ESTRUTURAL)
# ==========================================
print(response.result["metadata"])
# {
# "communities": 2, # Louvain detectou 2 comunidades (anéis de fraude)
# "nodes": 3, # 3 entidades no grafo (2 empresas + 1 servidor)
# "edges": 2, # 2 relacionamentos entre eles
# "threshold_applied": 0.7 # Threshold usado para filtrar
# }
# ==========================================
# VISUALIZAÇÃO PLOTLY (PRONTA PARA FRONTEND)
# ==========================================
# JSON compatível com Plotly Network Graph:
# - Nodos coloridos por score (verde→amarelo→vermelho)
# - Arestas com espessura proporcional à strength
# - Tooltip com detalhes ao passar mouse
# - Comunidades destacadas com cores diferentes
viz_json = response.result["visualization"]
# Frontend carrega diretamente:
# Plotly.newPlot('graph-div', JSON.parse(viz_json))
# Resultado: Grafo interativo mostrando anel de fraude
Neste exemplo, o grafo revela um possível anel de fraude:
- João Silva (servidor) está conectado a 2 empresas diferentes
- Empresa ABC tem alta suspeita (0.85) e relação forte com João (0.9)
- Fornecedor XYZ compartilha mesmo endereço com João (RED FLAG!)
- 2 comunidades detectadas = João e suas empresas formam grupo isolado
Ação sugerida: Investigação aprofundada do servidor João Silva e das empresas relacionadas.
Exemplo 3: Mapa Choropleth do Brasil (Sprint 6) 🗺️
message = AgentMessage(
sender="lampiao",
recipient="OscarNiemeyerAgent",
action="choropleth_map",
payload={
"data": [
{"state_code": "11", "value": 25000, "name": "Rondônia"},
{"state_code": "12", "value": 18000, "name": "Acre"},
{"state_code": "35", "value": 320000, "name": "São Paulo"},
{"state_code": "33", "value": 185000, "name": "Rio de Janeiro"}
# ... all 27 states
],
"color_column": "value",
"location_column": "state_code"
}
)
response = await oscar.process(message, context)
# Estatísticas do mapa
print(response.result["metadata"]["statistics"])
# {
# "min": 18000,
# "max": 320000,
# "mean": 85000,
# "median": 72000,
# "std_dev": 68000
# }
# Choropleth pronto para visualização
viz = response.result["visualization"]
# Plotly choropleth com boundaries brasileiros (GeoJSON automático)
Exemplo 4: Downsampling de Dataset Grande
# Dataset com 500k pontos - muito pesado para frontend
message = AgentMessage(
content="Otimizar série temporal para visualização",
data={
"raw_data": timeseries_500k_points,
"visualization_type": VisualizationType.LINE_CHART,
"target_points": 1000 # LTTB downsample
}
)
response = await oscar.process(message, context)
# Dados otimizados
print(len(response.data["optimized_data"]))
# Output: 1000 (reduzido de 500k, mas mantendo forma visual)