> ## Documentation Index
> Fetch the complete documentation index at: https://docs.datasnap.cloud/llms.txt
> Use this file to discover all available pages before exploring further.
# Análise de Dados com IA
> Como usar ferramentas de IA para analisar seus dados no DataSnap de forma mais eficiente
## 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:
"Mostre as vendas mensais dos últimos 6 meses"
"Identifique valores atípicos nos dados de pagamento"
"Agrupe clientes por comportamento de compra"
"Compare performance atual vs período anterior"
### Exemplos práticos
#### Análise de vendas com IA
```python theme={null}
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
```python theme={null}
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
```python theme={null}
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
```python theme={null}
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
```python theme={null}
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
```python theme={null}
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
```
```python theme={null}
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
Implemente soluções personalizadas com IA
Configure automação inteligente