Análise de Dados com IA

IA para análise de dados

As ferramentas de IA podem acelerar significativamente a análise e processamento 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.com.br"
    
    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

Processamento automático 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

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

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

Começando

Personalização

Escrevendo conteúdo

Ferramentas de IA

IA para análise de dados

Casos de uso com IA

Geração automática de consultas

Análise de tendências

Detecção de anomalias

Segmentação de clientes

Análise temporal

Exemplos práticos

Análise de vendas com IA

Detecção de anomalias

Sistema automático de alertas

Geração automática de dashboards

Dashboard inteligente

Análise preditiva

Previsão de tendências

Automação inteligente

Processamento automático com IA

Boas práticas com IA

Otimização de custos

Próximos passos

Integração avançada

Automação de workflows

Começando

Personalização

Escrevendo conteúdo

Ferramentas de IA

​IA para análise de dados

​Casos de uso com IA

​Geração automática de consultas

Análise de tendências

Detecção de anomalias

Segmentação de clientes

Análise temporal

​Exemplos práticos

​Análise de vendas com IA

​Detecção de anomalias

​Sistema automático de alertas

​Geração automática de dashboards

​Dashboard inteligente

​Análise preditiva

​Previsão de tendências

​Automação inteligente

​Processamento automático com IA

​Boas práticas com IA

​Otimização de custos

​Próximos passos

Integração avançada

Automação de workflows

IA para análise de dados

Casos de uso com IA

Geração automática de consultas

Exemplos práticos

Análise de vendas com IA

Detecção de anomalias

Sistema automático de alertas

Geração automática de dashboards

Dashboard inteligente

Análise preditiva

Previsão de tendências

Automação inteligente

Processamento automático com IA

Boas práticas com IA

Otimização de custos

Próximos passos