AI-Powered Data Analysis: Automating Insights with Python

Published on December 15, 2024 • 19 min read

Transform your data analysis workflow with AI-powered automation. This comprehensive guide shows you how to build intelligent data analysis systems that can automatically discover insights, generate reports, and make data-driven recommendations.

Table of Contents

1. Introduction to AI-Powered Data Analysis
2. Setting Up Your Environment
3. Automated Data Exploration
4. Intelligent Pattern Recognition
5. Automated Report Generation
6. Real-Time Analytics
7. Production Deployment
8. Best Practices

Introduction to AI-Powered Data Analysis {#introduction}

AI-powered data analysis combines traditional statistical methods with machine learning to automatically discover insights, detect anomalies, and generate actionable recommendations from your data.

Key Benefits

• Automated Insights: Discover patterns without manual exploration
• Scalability: Analyze large datasets efficiently
• Real-time Processing: Get insights as data arrives
• Consistency: Eliminate human bias in analysis

Setting Up Your Environment {#setup}

Installation and Dependencies

1pip install pandas numpy scikit-learn matplotlib seaborn
2pip install plotly dash streamlit
3pip install openai langchain
4pip install prophet statsmodels
5pip install jupyter notebook

Basic Setup

1import pandas as pd
2import numpy as np
3import matplotlib.pyplot as plt
4import seaborn as sns
5from sklearn.ensemble import IsolationForest
6from sklearn.cluster import KMeans
7from sklearn.preprocessing import StandardScaler
8import openai
9from datetime import datetime, timedelta
10import warnings
11warnings.filterwarnings('ignore')
12
13# Configure plotting
14plt.style.use('seaborn-v0_8')
15sns.set_palette("husl")
16
17# OpenAI configuration
18openai.api_key = "your_openai_api_key"
19
20print("Environment setup complete!")

Automated Data Exploration {#exploration}

Intelligent Data Profiling

1class DataProfiler:
2    def __init__(self, df):
3        self.df = df
4        self.profile = {}
5        
6    def generate_profile(self):
7        """Generate comprehensive data profile"""
8        self.profile = {
9            'shape': self.df.shape,
10            'columns': list(self.df.columns),
11            'dtypes': self.df.dtypes.to_dict(),
12            'missing_values': self.df.isnull().sum().to_dict(),
13            'duplicates': self.df.duplicated().sum(),
14            'memory_usage': self.df.memory_usage(deep=True).sum(),
15            'numeric_summary': self._numeric_summary(),
16            'categorical_summary': self._categorical_summary(),
17            'correlations': self._correlation_analysis(),
18            'outliers': self._detect_outliers()
19        }
20        return self.profile
21    
22    def _numeric_summary(self):
23        """Analyze numeric columns"""
24        numeric_cols = self.df.select_dtypes(include=[np.number]).columns
25        summary = {}
26        
27        for col in numeric_cols:
28            summary[col] = {
29                'mean': self.df[col].mean(),
30                'median': self.df[col].median(),
31                'std': self.df[col].std(),
32                'min': self.df[col].min(),
33                'max': self.df[col].max(),
34                'skewness': self.df[col].skew(),
35                'kurtosis': self.df[col].kurtosis()
36            }
37        
38        return summary
39    
40    def _categorical_summary(self):
41        """Analyze categorical columns"""
42        categorical_cols = self.df.select_dtypes(include=['object']).columns
43        summary = {}
44        
45        for col in categorical_cols:
46            summary[col] = {
47                'unique_count': self.df[col].nunique(),
48                'most_frequent': self.df[col].mode().iloc[0] if not self.df[col].mode().empty else None,
49                'frequency': self.df[col].value_counts().head().to_dict()
50            }
51        
52        return summary
53    
54    def _correlation_analysis(self):
55        """Analyze correlations between numeric variables"""
56        numeric_df = self.df.select_dtypes(include=[np.number])
57        if numeric_df.shape[1] > 1:
58            corr_matrix = numeric_df.corr()
59            # Find strong correlations (> 0.7 or < -0.7)
60            strong_corr = []
61            for i in range(len(corr_matrix.columns)):
62                for j in range(i+1, len(corr_matrix.columns)):
63                    corr_val = corr_matrix.iloc[i, j]
64                    if abs(corr_val) > 0.7:
65                        strong_corr.append({
66                            'var1': corr_matrix.columns[i],
67                            'var2': corr_matrix.columns[j],
68                            'correlation': corr_val
69                        })
70            return strong_corr
71        return []
72    
73    def _detect_outliers(self):
74        """Detect outliers using IQR method"""
75        numeric_cols = self.df.select_dtypes(include=[np.number]).columns
76        outliers = {}
77        
78        for col in numeric_cols:
79            Q1 = self.df[col].quantile(0.25)
80            Q3 = self.df[col].quantile(0.75)
81            IQR = Q3 - Q1
82            lower_bound = Q1 - 1.5 * IQR
83            upper_bound = Q3 + 1.5 * IQR
84            
85            outlier_count = ((self.df[col] < lower_bound) | (self.df[col] > upper_bound)).sum()
86            outliers[col] = {
87                'count': outlier_count,
88                'percentage': (outlier_count / len(self.df)) * 100
89            }
90        
91        return outliers
92
93# Usage
94df = pd.read_csv('your_data.csv')
95profiler = DataProfiler(df)
96profile = profiler.generate_profile()
97print(f"Dataset shape: {profile['shape']}")

AI-Powered Data Insights

1class AIInsightGenerator:
2    def __init__(self, openai_api_key):
3        openai.api_key = openai_api_key
4    
5    def generate_insights(self, data_profile):
6        """Generate AI-powered insights from data profile"""
7        prompt = self._create_insight_prompt(data_profile)
8        
9        response = openai.ChatCompletion.create(
10            model="gpt-4",
11            messages=[
12                {"role": "system", "content": "You are a data scientist expert. Analyze the data profile and provide actionable insights."},
13                {"role": "user", "content": prompt}
14            ],
15            max_tokens=1000,
16            temperature=0.3
17        )
18        
19        return response.choices[0].message.content
20    
21    def _create_insight_prompt(self, profile):
22        """Create prompt for AI insight generation"""
23        prompt = f"""
24        Analyze this dataset profile and provide key insights:
25        
26        Dataset Overview:
27        - Shape: {profile['shape']}
28        - Columns: {len(profile['columns'])}
29        - Missing values: {sum(profile['missing_values'].values())}
30        - Duplicates: {profile['duplicates']}
31        
32        Numeric Variables Summary:
33        {self._format_numeric_summary(profile['numeric_summary'])}
34        
35        Strong Correlations:
36        {self._format_correlations(profile['correlations'])}
37        
38        Outliers:
39        {self._format_outliers(profile['outliers'])}
40        
41        Please provide:
42        1. Key findings and patterns
43        2. Data quality issues
44        3. Recommended next steps
45        4. Potential business insights
46        """
47        
48        return prompt
49    
50    def _format_numeric_summary(self, summary):
51        """Format numeric summary for prompt"""
52        formatted = []
53        for col, stats in summary.items():
54            formatted.append(f"- {col}: mean={stats['mean']:.2f}, std={stats['std']:.2f}, skew={stats['skewness']:.2f}")
55        return "\n".join(formatted)
56    
57    def _format_correlations(self, correlations):
58        """Format correlations for prompt"""
59        if not correlations:
60            return "No strong correlations found"
61        
62        formatted = []
63        for corr in correlations:
64            formatted.append(f"- {corr['var1']} ↔ {corr['var2']}: {corr['correlation']:.3f}")
65        return "\n".join(formatted)
66    
67    def _format_outliers(self, outliers):
68        """Format outliers for prompt"""
69        formatted = []
70        for col, stats in outliers.items():
71            if stats['count'] > 0:
72                formatted.append(f"- {col}: {stats['count']} outliers ({stats['percentage']:.1f}%)")
73        return "\n".join(formatted) if formatted else "No significant outliers detected"
74
75# Usage
76insight_generator = AIInsightGenerator("your_openai_api_key")
77insights = insight_generator.generate_insights(profile)
78print(insights)

Intelligent Pattern Recognition {#patterns}

Automated Clustering Analysis

1class AutoClusterAnalyzer:
2    def __init__(self, df):
3        self.df = df
4        self.scaler = StandardScaler()
5        
6    def find_optimal_clusters(self, max_clusters=10):
7        """Find optimal number of clusters using elbow method"""
8        numeric_df = self.df.select_dtypes(include=[np.number])
9        scaled_data = self.scaler.fit_transform(numeric_df)
10        
11        inertias = []
12        silhouette_scores = []
13        
14        for k in range(2, max_clusters + 1):
15            kmeans = KMeans(n_clusters=k, random_state=42)
16            kmeans.fit(scaled_data)
17            inertias.append(kmeans.inertia_)
18            
19            from sklearn.metrics import silhouette_score
20            score = silhouette_score(scaled_data, kmeans.labels_)
21            silhouette_scores.append(score)
22        
23        # Find elbow point
24        optimal_k = self._find_elbow_point(inertias) + 2
25        
26        return {
27            'optimal_clusters': optimal_k,
28            'inertias': inertias,
29            'silhouette_scores': silhouette_scores
30        }
31    
32    def _find_elbow_point(self, inertias):
33        """Find elbow point in inertia curve"""
34        # Simple elbow detection using second derivative
35        second_derivatives = []
36        for i in range(1, len(inertias) - 1):
37            second_deriv = inertias[i-1] - 2*inertias[i] + inertias[i+1]
38            second_derivatives.append(second_deriv)
39        
40        return np.argmax(second_derivatives) + 1
41    
42    def perform_clustering(self, n_clusters):
43        """Perform clustering with specified number of clusters"""
44        numeric_df = self.df.select_dtypes(include=[np.number])
45        scaled_data = self.scaler.fit_transform(numeric_df)
46        
47        kmeans = KMeans(n_clusters=n_clusters, random_state=42)
48        clusters = kmeans.fit_predict(scaled_data)
49        
50        # Add cluster labels to dataframe
51        result_df = self.df.copy()
52        result_df['cluster'] = clusters
53        
54        # Analyze clusters
55        cluster_analysis = self._analyze_clusters(result_df, numeric_df.columns)
56        
57        return {
58            'clustered_data': result_df,
59            'cluster_centers': kmeans.cluster_centers_,
60            'analysis': cluster_analysis
61        }
62    
63    def _analyze_clusters(self, df, numeric_cols):
64        """Analyze characteristics of each cluster"""
65        analysis = {}
66        
67        for cluster_id in df['cluster'].unique():
68            cluster_data = df[df['cluster'] == cluster_id]
69            
70            analysis[cluster_id] = {
71                'size': len(cluster_data),
72                'percentage': (len(cluster_data) / len(df)) * 100,
73                'characteristics': {}
74            }
75            
76            # Analyze numeric characteristics
77            for col in numeric_cols:
78                analysis[cluster_id]['characteristics'][col] = {
79                    'mean': cluster_data[col].mean(),
80                    'median': cluster_data[col].median(),
81                    'std': cluster_data[col].std()
82                }
83        
84        return analysis
85
86# Usage
87cluster_analyzer = AutoClusterAnalyzer(df)
88optimal_clusters = cluster_analyzer.find_optimal_clusters()
89clustering_result = cluster_analyzer.perform_clustering(optimal_clusters['optimal_clusters'])

Anomaly Detection System

1class AnomalyDetector:
2    def __init__(self, df):
3        self.df = df
4        self.models = {}
5        
6    def detect_anomalies(self, contamination=0.1):
7        """Detect anomalies using multiple methods"""
8        numeric_df = self.df.select_dtypes(include=[np.number])
9        
10        results = {}
11        
12        # Isolation Forest
13        iso_forest = IsolationForest(contamination=contamination, random_state=42)
14        iso_anomalies = iso_forest.fit_predict(numeric_df)
15        results['isolation_forest'] = iso_anomalies
16        
17        # Statistical outliers (Z-score)
18        z_scores = np.abs((numeric_df - numeric_df.mean()) / numeric_df.std())
19        z_anomalies = (z_scores > 3).any(axis=1).astype(int)
20        z_anomalies = np.where(z_anomalies == 1, -1, 1)  # Convert to same format
21        results['z_score'] = z_anomalies
22        
23        # IQR method
24        iqr_anomalies = self._iqr_anomalies(numeric_df)
25        results['iqr'] = iqr_anomalies
26        
27        # Ensemble approach
28        ensemble_scores = (
29            (results['isolation_forest'] == -1).astype(int) +
30            (results['z_score'] == -1).astype(int) +
31            (results['iqr'] == -1).astype(int)
32        )
33        
34        # Consider anomaly if detected by at least 2 methods
35        ensemble_anomalies = np.where(ensemble_scores >= 2, -1, 1)
36        results['ensemble'] = ensemble_anomalies
37        
38        return results
39    
40    def _iqr_anomalies(self, df):
41        """Detect anomalies using IQR method"""
42        anomalies = np.ones(len(df))
43        
44        for col in df.columns:
45            Q1 = df[col].quantile(0.25)
46            Q3 = df[col].quantile(0.75)
47            IQR = Q3 - Q1
48            lower_bound = Q1 - 1.5 * IQR
49            upper_bound = Q3 + 1.5 * IQR
50            
51            col_anomalies = (df[col] < lower_bound) | (df[col] > upper_bound)
52            anomalies[col_anomalies] = -1
53        
54        return anomalies
55    
56    def analyze_anomalies(self, anomaly_results, method='ensemble'):
57        """Analyze detected anomalies"""
58        anomaly_indices = np.where(anomaly_results[method] == -1)[0]
59        anomaly_data = self.df.iloc[anomaly_indices]
60        
61        analysis = {
62            'count': len(anomaly_indices),
63            'percentage': (len(anomaly_indices) / len(self.df)) * 100,
64            'indices': anomaly_indices.tolist(),
65            'summary': anomaly_data.describe().to_dict()
66        }
67        
68        return analysis
69
70# Usage
71anomaly_detector = AnomalyDetector(df)
72anomalies = anomaly_detector.detect_anomalies()
73anomaly_analysis = anomaly_detector.analyze_anomalies(anomalies)

Automated Report Generation {#reports}

AI-Powered Report Generator

1class AutoReportGenerator:
2    def __init__(self, openai_api_key):
3        openai.api_key = openai_api_key
4        
5    def generate_comprehensive_report(self, df, analysis_results):
6        """Generate comprehensive analysis report"""
7        report_sections = {
8            'executive_summary': self._generate_executive_summary(df, analysis_results),
9            'data_overview': self._generate_data_overview(df),
10            'key_findings': self._generate_key_findings(analysis_results),
11            'recommendations': self._generate_recommendations(analysis_results),
12            'technical_details': self._generate_technical_details(analysis_results)
13        }
14        
15        # Combine into full report
16        full_report = self._compile_report(report_sections)
17        
18        return full_report
19    
20    def _generate_executive_summary(self, df, results):
21        """Generate executive summary using AI"""
22        prompt = f"""
23        Create an executive summary for a data analysis report:
24        
25        Dataset: {df.shape[0]} rows, {df.shape[1]} columns
26        Clusters found: {len(results.get('clustering', {}).get('analysis', {}))}
27        Anomalies detected: {results.get('anomalies', {}).get('count', 0)}
28        
29        Write a concise executive summary highlighting the most important findings.
30        """
31        
32        response = openai.ChatCompletion.create(
33            model="gpt-4",
34            messages=[{"role": "user", "content": prompt}],
35            max_tokens=300
36        )
37        
38        return response.choices[0].message.content
39    
40    def _generate_data_overview(self, df):
41        """Generate data overview section"""
42        overview = f"""
43        ## Data Overview
44        
45        - **Dataset Size**: {df.shape[0]:,} rows × {df.shape[1]} columns
46        - **Memory Usage**: {df.memory_usage(deep=True).sum() / 1024**2:.2f} MB
47        - **Missing Values**: {df.isnull().sum().sum():,} ({(df.isnull().sum().sum() / df.size) * 100:.2f}%)
48        - **Duplicate Rows**: {df.duplicated().sum():,}
49        
50        ### Column Types
51        """
52        
53        dtype_counts = df.dtypes.value_counts()
54        for dtype, count in dtype_counts.items():
55            overview += f"- **{dtype}**: {count} columns\n"
56        
57        return overview
58    
59    def _generate_key_findings(self, results):
60        """Generate key findings section"""
61        findings = "## Key Findings\n\n"
62        
63        # Clustering findings
64        if 'clustering' in results:
65            cluster_count = len(results['clustering']['analysis'])
66            findings += f"### Clustering Analysis\n"
67            findings += f"- Identified {cluster_count} distinct data segments\n"
68            
69            for cluster_id, info in results['clustering']['analysis'].items():
70                findings += f"- Cluster {cluster_id}: {info['size']} records ({info['percentage']:.1f}%)\n"
71        
72        # Anomaly findings
73        if 'anomalies' in results:
74            anomaly_count = results['anomalies']['count']
75            anomaly_pct = results['anomalies']['percentage']
76            findings += f"\n### Anomaly Detection\n"
77            findings += f"- Detected {anomaly_count} anomalous records ({anomaly_pct:.2f}%)\n"
78        
79        return findings
80    
81    def _generate_recommendations(self, results):
82        """Generate recommendations using AI"""
83        prompt = f"""
84        Based on this data analysis, provide actionable recommendations:
85        
86        Analysis Results:
87        - Clusters: {len(results.get('clustering', {}).get('analysis', {}))}
88        - Anomalies: {results.get('anomalies', {}).get('count', 0)}
89        
90        Provide 3-5 specific, actionable recommendations for business stakeholders.
91        """
92        
93        response = openai.ChatCompletion.create(
94            model="gpt-4",
95            messages=[{"role": "user", "content": prompt}],
96            max_tokens=400
97        )
98        
99        return f"## Recommendations\n\n{response.choices[0].message.content}"
100    
101    def _generate_technical_details(self, results):
102        """Generate technical details section"""
103        details = "## Technical Details\n\n"
104        
105        if 'clustering' in results:
106            details += "### Clustering Methodology\n"
107            details += "- Algorithm: K-Means clustering\n"
108            details += "- Preprocessing: StandardScaler normalization\n"
109            details += "- Optimization: Elbow method for optimal K\n\n"
110        
111        if 'anomalies' in results:
112            details += "### Anomaly Detection Methodology\n"
113            details += "- Isolation Forest\n"
114            details += "- Statistical Z-score (threshold: 3)\n"
115            details += "- Interquartile Range (IQR) method\n"
116            details += "- Ensemble approach (majority voting)\n\n"
117        
118        return details
119    
120    def _compile_report(self, sections):
121        """Compile all sections into final report"""
122        report = f"""
123# Automated Data Analysis Report
124*Generated on {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}*
125
126{sections['executive_summary']}
127
128{sections['data_overview']}
129
130{sections['key_findings']}
131
132{sections['recommendations']}
133
134{sections['technical_details']}
135
136---
137*This report was generated automatically using AI-powered data analysis tools.*
138        """
139        
140        return report
141
142# Usage
143report_generator = AutoReportGenerator("your_openai_api_key")
144analysis_results = {
145    'clustering': clustering_result,
146    'anomalies': anomaly_analysis
147}
148report = report_generator.generate_comprehensive_report(df, analysis_results)
149print(report)

Real-Time Analytics {#realtime}

Streaming Data Processor

1import asyncio
2from datetime import datetime
3import json
4
5class RealTimeAnalyzer:
6    def __init__(self, window_size=100):
7        self.window_size = window_size
8        self.data_buffer = []
9        self.anomaly_detector = None
10        self.alerts = []
11        
12    async def process_stream(self, data_stream):
13        """Process streaming data in real-time"""
14        async for data_point in data_stream:
15            await self.process_data_point(data_point)
16    
17    async def process_data_point(self, data_point):
18        """Process individual data point"""
19        # Add to buffer
20        self.data_buffer.append(data_point)
21        
22        # Maintain window size
23        if len(self.data_buffer) > self.window_size:
24            self.data_buffer.pop(0)
25        
26        # Analyze if we have enough data
27        if len(self.data_buffer) >= 10:
28            await self.analyze_current_window()
29    
30    async def analyze_current_window(self):
31        """Analyze current data window"""
32        df = pd.DataFrame(self.data_buffer)
33        
34        # Quick anomaly detection
35        if self.anomaly_detector is None:
36            self.anomaly_detector = IsolationForest(contamination=0.1)
37            self.anomaly_detector.fit(df.select_dtypes(include=[np.number]))
38        
39        # Check latest point for anomalies
40        latest_point = df.tail(1).select_dtypes(include=[np.number])
41        if not latest_point.empty:
42            is_anomaly = self.anomaly_detector.predict(latest_point)[0] == -1
43            
44            if is_anomaly:
45                await self.trigger_alert(df.tail(1).iloc[0])
46    
47    async def trigger_alert(self, anomalous_point):
48        """Trigger alert for anomalous data"""
49        alert = {
50            'timestamp': datetime.now().isoformat(),
51            'type': 'anomaly',
52            'data': anomalous_point.to_dict(),
53            'severity': 'high'
54        }
55        
56        self.alerts.append(alert)
57        print(f"🚨 ALERT: Anomaly detected at {alert['timestamp']}")
58        
59        # In production, send to monitoring system
60        await self.send_alert(alert)
61    
62    async def send_alert(self, alert):
63        """Send alert to monitoring system"""
64        # Simulate sending alert
65        await asyncio.sleep(0.1)
66        print(f"Alert sent: {json.dumps(alert, indent=2)}")
67
68# Usage example
69async def simulate_data_stream():
70    """Simulate streaming data"""
71    for i in range(1000):
72        # Normal data
73        if i % 50 != 0:
74            yield {
75                'timestamp': datetime.now().isoformat(),
76                'value1': np.random.normal(100, 10),
77                'value2': np.random.normal(50, 5),
78                'category': np.random.choice(['A', 'B', 'C'])
79            }
80        else:
81            # Anomalous data
82            yield {
83                'timestamp': datetime.now().isoformat(),
84                'value1': np.random.normal(200, 10),  # Anomalous
85                'value2': np.random.normal(50, 5),
86                'category': np.random.choice(['A', 'B', 'C'])
87            }
88        
89        await asyncio.sleep(0.1)
90
91# Run real-time analysis
92analyzer = RealTimeAnalyzer()
93# asyncio.run(analyzer.process_stream(simulate_data_stream()))

Production Deployment {#deployment}

Streamlit Dashboard

1import streamlit as st
2import plotly.express as px
3import plotly.graph_objects as go
4
5def create_dashboard():
6    """Create interactive dashboard"""
7    st.set_page_config(
8        page_title="AI Data Analysis Dashboard",
9        page_icon="📊",
10        layout="wide"
11    )
12    
13    st.title("🤖 AI-Powered Data Analysis Dashboard")
14    
15    # Sidebar for file upload
16    st.sidebar.header("Data Upload")
17    uploaded_file = st.sidebar.file_uploader(
18        "Choose a CSV file",
19        type="csv"
20    )
21    
22    if uploaded_file is not None:
23        # Load data
24        df = pd.read_csv(uploaded_file)
25        
26        # Data overview
27        st.header("📋 Data Overview")
28        col1, col2, col3, col4 = st.columns(4)
29        
30        with col1:
31            st.metric("Rows", f"{df.shape[0]:,}")
32        with col2:
33            st.metric("Columns", df.shape[1])
34        with col3:
35            st.metric("Missing Values", f"{df.isnull().sum().sum():,}")
36        with col4:
37            st.metric("Memory Usage", f"{df.memory_usage(deep=True).sum() / 1024**2:.1f} MB")
38        
39        # Data profiling
40        if st.button("🔍 Generate AI Insights"):
41            with st.spinner("Analyzing data..."):
42                profiler = DataProfiler(df)
43                profile = profiler.generate_profile()
44                
45                insight_generator = AIInsightGenerator(st.secrets["openai_api_key"])
46                insights = insight_generator.generate_insights(profile)
47                
48                st.header("🧠 AI-Generated Insights")
49                st.write(insights)
50        
51        # Clustering analysis
52        st.header("🎯 Clustering Analysis")
53        if st.button("Perform Clustering"):
54            with st.spinner("Finding optimal clusters..."):
55                cluster_analyzer = AutoClusterAnalyzer(df)
56                optimal_clusters = cluster_analyzer.find_optimal_clusters()
57                clustering_result = cluster_analyzer.perform_clustering(
58                    optimal_clusters['optimal_clusters']
59                )
60                
61                # Plot clusters
62                numeric_cols = df.select_dtypes(include=[np.number]).columns
63                if len(numeric_cols) >= 2:
64                    fig = px.scatter(
65                        clustering_result['clustered_data'],
66                        x=numeric_cols[0],
67                        y=numeric_cols[1],
68                        color='cluster',
69                        title="Cluster Visualization"
70                    )
71                    st.plotly_chart(fig, use_container_width=True)
72        
73        # Anomaly detection
74        st.header("🚨 Anomaly Detection")
75        if st.button("Detect Anomalies"):
76            with st.spinner("Detecting anomalies..."):
77                anomaly_detector = AnomalyDetector(df)
78                anomalies = anomaly_detector.detect_anomalies()
79                anomaly_analysis = anomaly_detector.analyze_anomalies(anomalies)
80                
81                st.metric("Anomalies Detected", f"{anomaly_analysis['count']} ({anomaly_analysis['percentage']:.2f}%)")
82                
83                # Show anomalous records
84                if anomaly_analysis['count'] > 0:
85                    anomaly_indices = anomaly_analysis['indices']
86                    st.subheader("Anomalous Records")
87                    st.dataframe(df.iloc[anomaly_indices])
88
89if __name__ == "__main__":
90    create_dashboard()

Best Practices

Performance Optimization

1. Data Preprocessing

   • Use efficient data types
   • Handle missing values appropriately
   • Normalize/standardize features

2. Memory Management

   • Process data in chunks for large datasets
   • Use generators for streaming data
   • Clean up unused variables

3. Model Selection

   • Choose appropriate algorithms for data size
   • Use ensemble methods for robustness
   • Implement early stopping

Error Handling and Monitoring

1import logging
2from functools import wraps
3
4def error_handler(func):
5    """Decorator for error handling"""
6    @wraps(func)
7    def wrapper(*args, **kwargs):
8        try:
9            return func(*args, **kwargs)
10        except Exception as e:
11            logging.error(f"Error in {func.__name__}: {str(e)}")
12            return None
13    return wrapper
14
15class AnalysisMonitor:
16    def __init__(self):
17        self.metrics = {
18            'analyses_performed': 0,
19            'errors_encountered': 0,
20            'average_processing_time': 0
21        }
22    
23    def log_analysis(self, processing_time, success=True):
24        """Log analysis metrics"""
25        self.metrics['analyses_performed'] += 1
26        
27        if not success:
28            self.metrics['errors_encountered'] += 1
29        
30        # Update average processing time
31        current_avg = self.metrics['average_processing_time']
32        count = self.metrics['analyses_performed']
33        self.metrics['average_processing_time'] = (
34            (current_avg * (count - 1) + processing_time) / count
35        )

Conclusion

AI-powered data analysis transforms how we extract insights from data. By automating exploration, pattern recognition, and report generation, you can focus on strategic decision-making rather than manual analysis.

Key takeaways:

• Automate repetitive analysis tasks
• Use AI to generate insights and recommendations
• Implement real-time monitoring for critical metrics
• Build interactive dashboards for stakeholders
• Monitor and optimize performance continuously

---

This guide provides a comprehensive foundation for building AI-powered data analysis systems. Start with basic automation and gradually add more sophisticated AI capabilities as your needs evolve.