IA para análise de dados
As ferramentas de IA podem acelerar significativamente a análise de dados no DataSnap, desde a criação de consultas até insights automáticos.Combine a poder do DataSnap com IA para descobrir padrões, automatizar análises e gerar relatórios inteligentes.
Casos de uso com IA
Geração automática de consultas
Use IA para converter perguntas em linguagem natural para consultas DataSnap:Análise de tendências
“Mostre as vendas mensais dos últimos 6 meses”
Detecção de anomalias
“Identifique valores atípicos nos dados de pagamento”
Segmentação de clientes
“Agrupe clientes por comportamento de compra”
Análise temporal
“Compare performance atual vs período anterior”
Exemplos práticos
Análise de vendas com IA
import requests
import openai # ou anthropic para Claude
from dataclasses import dataclass
from typing import Dict, Any
@dataclass
class DataSnapAI:
datasnap_token: str
openai_key: str
schema_slug: str
base_url: str = "https://api.datasnap.cloud"
def __post_init__(self):
openai.api_key = self.openai_key
self.headers = {"Authorization": f"Bearer {self.datasnap_token}"}
def natural_language_to_query(self, question: str, context: str = "") -> Dict[str, Any]:
"""Converte pergunta em linguagem natural para consulta DataSnap"""
prompt = f"""
Converta esta pergunta para uma consulta DataSnap JSON:
Pergunta: {question}
Contexto dos dados: {context}
A consulta deve usar o formato:
{{
"select": ["campo1", "funcao(campo2) as alias"],
"where": [{{"field": "campo", "op": "operador", "value": "valor"}}],
"group_by": ["campo"],
"order_by": [{{"field": "campo", "direction": "asc|desc"}}],
"limit": numero
}}
Responda apenas com o JSON válido:
"""
response = openai.ChatCompletion.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}],
temperature=0.1
)
query_text = response.choices[0].message.content.strip()
return json.loads(query_text)
def execute_ai_query(self, question: str, context: str = "") -> Dict[str, Any]:
"""Executa consulta gerada por IA"""
query = self.natural_language_to_query(question, context)
response = requests.post(
f"{self.base_url}/api/v1/schemas/{self.schema_slug}/query",
headers=self.headers,
json=query
)
return {
'question': question,
'generated_query': query,
'results': response.json()
}
def analyze_results(self, results: Dict[str, Any]) -> str:
"""Gera análise textual dos resultados usando IA"""
prompt = f"""
Analise estes dados e forneça insights em português:
Dados: {json.dumps(results['results'], ensure_ascii=False)}
Pergunta original: {results['question']}
Forneça:
1. Resumo dos principais achados
2. Insights interessantes
3. Recomendações baseadas nos dados
4. Possíveis próximos passos para análise
"""
response = openai.ChatCompletion.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}],
temperature=0.3
)
return response.choices[0].message.content
# Exemplo de uso
ai = DataSnapAI(
datasnap_token="seu_token",
openai_key="sua_chave_openai",
schema_slug="vendas"
)
# Análise automática
result = ai.execute_ai_query(
"Quais produtos venderam mais nos últimos 3 meses?",
"Dados de vendas com campos: produto, data_venda, quantidade, valor"
)
analysis = ai.analyze_results(result)
print(analysis)
Detecção de anomalias
Sistema automático de alertas
import numpy as np
from sklearn.ensemble import IsolationForest
import pandas as pd
class DataSnapAnomalyDetector:
def __init__(self, datasnap_client):
self.client = datasnap_client
self.models = {}
def train_anomaly_detector(self, schema_slug: str, fields: list,
contamination: float = 0.1):
"""Treina detector de anomalias nos dados históricos"""
# Buscar dados históricos
query = {
"select": fields + ["created_at"],
"order_by": [{"field": "created_at", "direction": "desc"}],
"limit": 10000
}
response = self.client.query(schema_slug, query)
df = pd.DataFrame(response['data'])
# Preparar dados numéricos
numeric_data = df.select_dtypes(include=[np.number])
# Treinar modelo
model = IsolationForest(contamination=contamination, random_state=42)
model.fit(numeric_data)
self.models[schema_slug] = {
'model': model,
'fields': list(numeric_data.columns),
'stats': {
'mean': numeric_data.mean().to_dict(),
'std': numeric_data.std().to_dict()
}
}
return f"Modelo treinado para {schema_slug} com {len(numeric_data)} registros"
def detect_anomalies(self, schema_slug: str, recent_hours: int = 24):
"""Detecta anomalias em dados recentes"""
if schema_slug not in self.models:
return "Modelo não treinado para este schema"
model_info = self.models[schema_slug]
# Buscar dados recentes
query = {
"select": model_info['fields'] + ["id", "created_at"],
"where": [
{
"field": "created_at",
"op": ">=",
"value": f"NOW() - INTERVAL {recent_hours} HOUR"
}
]
}
response = self.client.query(schema_slug, query)
if not response['data']:
return "Nenhum dado recente encontrado"
df = pd.DataFrame(response['data'])
numeric_data = df[model_info['fields']]
# Detectar anomalias
anomaly_scores = model_info['model'].decision_function(numeric_data)
is_anomaly = model_info['model'].predict(numeric_data) == -1
anomalies = df[is_anomaly].copy()
anomalies['anomaly_score'] = anomaly_scores[is_anomaly]
return {
'total_records': len(df),
'anomalies_count': len(anomalies),
'anomalies': anomalies.to_dict('records')
}
def generate_anomaly_report(self, schema_slug: str) -> str:
"""Gera relatório de anomalias usando IA"""
anomalies = self.detect_anomalies(schema_slug)
if isinstance(anomalies, str):
return anomalies
prompt = f"""
Analise este relatório de anomalias e forneça insights:
Schema: {schema_slug}
Total de registros analisados: {anomalies['total_records']}
Anomalias detectadas: {anomalies['anomalies_count']}
Detalhes das anomalias:
{json.dumps(anomalies['anomalies'][:5], ensure_ascii=False, indent=2)}
Forneça:
1. Possíveis causas das anomalias
2. Impacto no negócio
3. Ações recomendadas
4. Alertas que devem ser configurados
"""
response = openai.ChatCompletion.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}],
temperature=0.2
)
return response.choices[0].message.content
# Uso
detector = DataSnapAnomalyDetector(datasnap_client)
detector.train_anomaly_detector("vendas", ["valor", "quantidade"])
report = detector.generate_anomaly_report("vendas")
Geração automática de dashboards
Dashboard inteligente
import plotly.graph_objects as go
import plotly.express as px
from datetime import datetime, timedelta
class IntelligentDashboard:
def __init__(self, datasnap_client, ai_client):
self.datasnap = datasnap_client
self.ai = ai_client
def auto_generate_dashboard(self, schema_slug: str) -> Dict[str, Any]:
"""Gera dashboard automaticamente baseado nos dados"""
# Descobrir estrutura dos dados
schema_info = self.discover_schema_structure(schema_slug)
# Gerar visualizações recomendadas
recommendations = self.get_visualization_recommendations(schema_info)
# Criar gráficos
charts = []
for rec in recommendations:
chart_data = self.create_chart(schema_slug, rec)
if chart_data:
charts.append(chart_data)
return {
'schema': schema_slug,
'generated_at': datetime.now().isoformat(),
'charts': charts,
'insights': self.generate_insights(schema_slug, charts)
}
def discover_schema_structure(self, schema_slug: str) -> Dict[str, Any]:
"""Descobre estrutura e características dos dados"""
# Amostra dos dados
query = {
"select": ["*"],
"limit": 100
}
response = self.datasnap.query(schema_slug, query)
df = pd.DataFrame(response['data'])
structure = {
'total_fields': len(df.columns),
'numeric_fields': list(df.select_dtypes(include=[np.number]).columns),
'date_fields': [col for col in df.columns if 'data' in col.lower() or 'date' in col.lower()],
'categorical_fields': list(df.select_dtypes(include=['object']).columns),
'row_count_estimate': len(df) * 100 # Estimativa baseada na amostra
}
return structure
def get_visualization_recommendations(self, schema_info: Dict[str, Any]) -> list:
"""Usa IA para recomendar visualizações"""
prompt = f"""
Baseado nesta estrutura de dados, recomende as melhores visualizações:
Estrutura:
- Campos numéricos: {schema_info['numeric_fields']}
- Campos de data: {schema_info['date_fields']}
- Campos categóricos: {schema_info['categorical_fields']}
- Estimativa de registros: {schema_info['row_count_estimate']}
Para cada visualização recomendada, forneça:
- Tipo de gráfico (line, bar, pie, scatter, heatmap)
- Campos para X e Y
- Agregação necessária (sum, avg, count)
- Título sugerido
Formato da resposta (JSON):
[
{{
"type": "line",
"x_field": "data",
"y_field": "valor",
"aggregation": "sum",
"title": "Evolução das vendas"
}}
]
"""
response = openai.ChatCompletion.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}],
temperature=0.1
)
return json.loads(response.choices[0].message.content)
def create_chart(self, schema_slug: str, recommendation: Dict[str, Any]):
"""Cria gráfico baseado na recomendação"""
try:
# Montar query baseada na recomendação
query = self.build_query_from_recommendation(recommendation)
# Executar query
response = self.datasnap.query(schema_slug, query)
df = pd.DataFrame(response['data'])
# Criar visualização
chart_type = recommendation['type']
if chart_type == 'line':
fig = px.line(df, x=recommendation['x_field'], y=recommendation['y_field'],
title=recommendation['title'])
elif chart_type == 'bar':
fig = px.bar(df, x=recommendation['x_field'], y=recommendation['y_field'],
title=recommendation['title'])
elif chart_type == 'pie':
fig = px.pie(df, names=recommendation['x_field'], values=recommendation['y_field'],
title=recommendation['title'])
return {
'recommendation': recommendation,
'chart_html': fig.to_html(),
'data': df.to_dict('records')
}
except Exception as e:
print(f"Erro ao criar gráfico: {e}")
return None
# Uso
dashboard_gen = IntelligentDashboard(datasnap_client, ai_client)
dashboard = dashboard_gen.auto_generate_dashboard("vendas")
Análise preditiva
Previsão de tendências
from sklearn.linear_model import LinearRegression
from sklearn.preprocessing import PolynomialFeatures
import matplotlib.pyplot as plt
class PredictiveAnalytics:
def __init__(self, datasnap_client):
self.client = datasnap_client
def forecast_metric(self, schema_slug: str, date_field: str,
metric_field: str, days_ahead: int = 30):
"""Previsão de métrica baseada em dados históricos"""
# Buscar dados históricos
query = {
"select": [
f"date({date_field}) as date",
f"sum({metric_field}) as value"
],
"where": [
{
"field": date_field,
"op": ">=",
"value": (datetime.now() - timedelta(days=90)).strftime('%Y-%m-%d')
}
],
"group_by": [f"date({date_field})"],
"order_by": [{"field": "date", "direction": "asc"}]
}
response = self.client.query(schema_slug, query)
df = pd.DataFrame(response['data'])
# Preparar dados para ML
df['date'] = pd.to_datetime(df['date'])
df['day_number'] = (df['date'] - df['date'].min()).dt.days
# Treinar modelo
X = df[['day_number']].values
y = df['value'].values
# Usar regressão polinomial para capturar tendências
poly_features = PolynomialFeatures(degree=2)
X_poly = poly_features.fit_transform(X)
model = LinearRegression()
model.fit(X_poly, y)
# Fazer previsões
last_day = df['day_number'].max()
future_days = np.array([[last_day + i] for i in range(1, days_ahead + 1)])
future_days_poly = poly_features.transform(future_days)
predictions = model.predict(future_days_poly)
future_dates = [df['date'].max() + timedelta(days=i) for i in range(1, days_ahead + 1)]
return {
'historical_data': df.to_dict('records'),
'predictions': [
{'date': date.strftime('%Y-%m-%d'), 'predicted_value': pred}
for date, pred in zip(future_dates, predictions)
],
'model_score': model.score(X_poly, y)
}
def generate_forecast_insights(self, forecast_data: Dict[str, Any]) -> str:
"""Gera insights sobre a previsão usando IA"""
historical = forecast_data['historical_data']
predictions = forecast_data['predictions']
prompt = f"""
Analise esta previsão e forneça insights:
Dados históricos (últimas 5 entradas):
{json.dumps(historical[-5:], ensure_ascii=False, indent=2)}
Previsões:
{json.dumps(predictions[:7], ensure_ascii=False, indent=2)}
Score do modelo: {forecast_data['model_score']:.3f}
Forneça:
1. Tendência identificada (crescimento, queda, estável)
2. Confiabilidade da previsão
3. Fatores que podem impactar a previsão
4. Recomendações estratégicas baseadas na tendência
5. Alertas ou pontos de atenção
"""
response = openai.ChatCompletion.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}],
temperature=0.2
)
return response.choices[0].message.content
Automação inteligente
Análise automática com IA
class IntelligentProcessor:
def __init__(self, datasnap_client, ai_client):
self.datasnap = datasnap_client
self.ai = ai_client
def smart_data_classification(self, schema_slug: str, text_field: str):
"""Classifica automaticamente dados textuais"""
# Buscar amostra dos dados
query = {
"select": ["id", text_field],
"where": [{"field": text_field, "op": "is not null"}],
"limit": 100
}
response = self.datasnap.query(schema_slug, query)
classifications = []
for record in response['data']:
text = record[text_field]
prompt = f"""
Classifique este texto em uma das categorias:
[positivo, negativo, neutro, urgente, informativo, comercial, técnico]
Texto: "{text}"
Responda apenas com a categoria:
"""
classification = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": prompt}],
temperature=0.1,
max_tokens=10
).choices[0].message.content.strip()
classifications.append({
'id': record['id'],
'text': text,
'classification': classification
})
return classifications
def auto_tag_content(self, schema_slug: str, content_field: str):
"""Adiciona tags automáticas ao conteúdo"""
query = {
"select": ["id", content_field],
"limit": 50
}
response = self.datasnap.query(schema_slug, query)
tagged_content = []
for record in response['data']:
content = record[content_field]
prompt = f"""
Gere até 5 tags relevantes para este conteúdo.
Use apenas palavras-chave importantes, sem artigos ou preposições.
Conteúdo: "{content}"
Tags (separadas por vírgula):
"""
tags = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": prompt}],
temperature=0.3,
max_tokens=50
).choices[0].message.content.strip()
tagged_content.append({
'id': record['id'],
'content': content,
'auto_tags': tags.split(', ')
})
return tagged_content
Boas práticas com IA
Otimização de custos
Cache de respostas
Cache de respostas
import hashlib
from functools import lru_cache
class CachedAI:
def __init__(self):
self.cache = {}
def cached_completion(self, prompt: str, **kwargs):
# Criar hash do prompt para cache
prompt_hash = hashlib.md5(prompt.encode()).hexdigest()
if prompt_hash in self.cache:
return self.cache[prompt_hash]
# Fazer requisição apenas se não estiver em cache
response = openai.ChatCompletion.create(
messages=[{"role": "user", "content": prompt}],
**kwargs
)
self.cache[prompt_hash] = response
return response
Batch processing
Batch processing
def batch_analyze_data(records: list, batch_size: int = 10):
"""Processa dados em lotes para otimizar custos"""
results = []
for i in range(0, len(records), batch_size):
batch = records[i:i+batch_size]
# Combinar múltiplos registros em um prompt
combined_prompt = "Analise estes registros:\n\n"
for j, record in enumerate(batch):
combined_prompt += f"Registro {j+1}: {record}\n"
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": combined_prompt}],
temperature=0.1
)
results.extend(process_batch_response(response, len(batch)))
return results
Próximos passos
Integração avançada
Implemente soluções personalizadas com IA
Automação de workflows
Configure automação inteligente