Spaces:
Running
Running
| """ | |
| ML VISUALIZER - Matplotlib/Seaborn Chart Generation | |
| ==================================================== | |
| Creates machine learning visualizations as base64-encoded PNG images | |
| that can be embedded directly in the response. | |
| Chart Types: | |
| 1. Forecast Plot with Confidence Band | |
| 2. Feature Importance Bar Chart | |
| 3. Correlation Heatmap | |
| 4. Residual Analysis Plot | |
| 5. Distribution Plot | |
| 6. Time Series Decomposition | |
| All charts use a premium dark theme for consistency! | |
| """ | |
| import numpy as np | |
| import pandas as pd | |
| import base64 | |
| import io | |
| import logging | |
| from typing import Dict, Any, List, Optional, Tuple | |
| logger = logging.getLogger(__name__) | |
| class MLVisualizer: | |
| """ | |
| 📊 ML VISUALIZER - Create Beautiful ML Charts | |
| Generates matplotlib/seaborn charts embedded as base64 PNG. | |
| All charts are dark-theme compatible and publication-ready. | |
| """ | |
| def __init__(self): | |
| self._setup_style() | |
| def _setup_style(self): | |
| """Setup matplotlib style for premium dark theme""" | |
| try: | |
| import matplotlib | |
| matplotlib.use('Agg') # Non-interactive backend | |
| import matplotlib.pyplot as plt | |
| import seaborn as sns | |
| # Premium dark theme | |
| plt.style.use('dark_background') | |
| # Custom color palette | |
| self.colors = { | |
| 'primary': '#6366f1', # Indigo | |
| 'secondary': '#22d3ee', # Cyan | |
| 'accent': '#f472b6', # Pink | |
| 'success': '#22c55e', # Green | |
| 'warning': '#f59e0b', # Amber | |
| 'error': '#ef4444', # Red | |
| 'text': '#f1f5f9', # Light gray | |
| 'bg': '#0f172a', # Dark blue | |
| 'grid': '#334155' # Slate | |
| } | |
| # Seaborn palette | |
| self.palette = [ | |
| self.colors['primary'], | |
| self.colors['secondary'], | |
| self.colors['accent'], | |
| self.colors['success'], | |
| self.colors['warning'] | |
| ] | |
| sns.set_palette(self.palette) | |
| self.plt = plt | |
| self.sns = sns | |
| self.available = True | |
| except ImportError as e: | |
| logger.warning(f"Matplotlib/Seaborn not available: {e}") | |
| self.available = False | |
| def _fig_to_base64(self, fig) -> str: | |
| """Convert matplotlib figure to base64 PNG""" | |
| try: | |
| buf = io.BytesIO() | |
| fig.savefig(buf, format='png', dpi=120, bbox_inches='tight', | |
| facecolor=self.colors['bg'], edgecolor='none') | |
| buf.seek(0) | |
| img_base64 = base64.b64encode(buf.read()).decode('utf-8') | |
| self.plt.close(fig) | |
| return f"data:image/png;base64,{img_base64}" | |
| except Exception as e: | |
| logger.error(f"Error converting figure to base64: {e}") | |
| return "" | |
| def create_forecast_plot( | |
| self, | |
| historical: List[float], | |
| predictions: List[float], | |
| lower_bound: List[float], | |
| upper_bound: List[float], | |
| title: str = "Forecast" | |
| ) -> Dict[str, Any]: | |
| """ | |
| Create forecast plot with confidence band | |
| Returns: | |
| Dict with 'image' (base64) and 'type' | |
| """ | |
| if not self.available: | |
| return {'type': 'ml_chart', 'image': None, 'error': 'Visualization libraries not available'} | |
| try: | |
| fig, ax = self.plt.subplots(figsize=(10, 6)) | |
| # Historical data | |
| hist_x = list(range(len(historical))) | |
| ax.plot(hist_x, historical, | |
| color=self.colors['primary'], linewidth=2.5, | |
| marker='o', markersize=6, label='Historical') | |
| # Forecast | |
| if predictions: | |
| pred_x = list(range(len(historical), len(historical) + len(predictions))) | |
| ax.plot(pred_x, predictions, | |
| color=self.colors['accent'], linewidth=2.5, | |
| marker='s', markersize=8, label='Forecast', linestyle='--') | |
| # Confidence band | |
| if lower_bound and upper_bound: | |
| ax.fill_between(pred_x, lower_bound, upper_bound, | |
| color=self.colors['accent'], alpha=0.2, | |
| label='95% Confidence') | |
| # Styling | |
| ax.set_facecolor(self.colors['bg']) | |
| ax.set_xlabel('Period', color=self.colors['text'], fontsize=12) | |
| ax.set_ylabel('Value', color=self.colors['text'], fontsize=12) | |
| ax.set_title(title, color=self.colors['text'], fontsize=14, fontweight='bold') | |
| ax.legend(loc='upper left', framealpha=0.8) | |
| ax.grid(True, alpha=0.3, color=self.colors['grid']) | |
| ax.tick_params(colors=self.colors['text']) | |
| return { | |
| 'type': 'ml_forecast', | |
| 'image': self._fig_to_base64(fig), | |
| 'title': title | |
| } | |
| except Exception as e: | |
| logger.error(f"Error creating forecast plot: {e}") | |
| return {'type': 'ml_chart', 'image': None, 'error': str(e)} | |
| def create_feature_importance( | |
| self, | |
| importance: Dict[str, float], | |
| title: str = "Feature Importance" | |
| ) -> Dict[str, Any]: | |
| """Create horizontal bar chart of feature importance""" | |
| if not self.available or not importance: | |
| return {'type': 'ml_chart', 'image': None} | |
| try: | |
| # Sort by importance | |
| sorted_items = sorted(importance.items(), key=lambda x: x[1], reverse=True)[:10] | |
| features = [item[0] for item in sorted_items] | |
| values = [item[1] for item in sorted_items] | |
| fig, ax = self.plt.subplots(figsize=(8, max(4, len(features) * 0.5))) | |
| # Create gradient colors based on importance | |
| colors = [self.colors['primary'] if v > 0.2 else self.colors['secondary'] | |
| for v in values] | |
| bars = ax.barh(features[::-1], values[::-1], color=colors[::-1]) | |
| # Add value labels | |
| for bar, val in zip(bars, values[::-1]): | |
| ax.text(bar.get_width() + 0.01, bar.get_y() + bar.get_height()/2, | |
| f'{val:.1%}', va='center', color=self.colors['text'], fontsize=10) | |
| ax.set_facecolor(self.colors['bg']) | |
| ax.set_xlabel('Importance', color=self.colors['text'], fontsize=12) | |
| ax.set_title(title, color=self.colors['text'], fontsize=14, fontweight='bold') | |
| ax.tick_params(colors=self.colors['text']) | |
| ax.grid(True, axis='x', alpha=0.3, color=self.colors['grid']) | |
| return { | |
| 'type': 'ml_importance', | |
| 'image': self._fig_to_base64(fig), | |
| 'title': title | |
| } | |
| except Exception as e: | |
| logger.error(f"Error creating feature importance: {e}") | |
| return {'type': 'ml_chart', 'image': None, 'error': str(e)} | |
| def create_correlation_heatmap( | |
| self, | |
| df: pd.DataFrame, | |
| title: str = "Correlation Matrix" | |
| ) -> Dict[str, Any]: | |
| """Create correlation heatmap for numeric columns""" | |
| if not self.available: | |
| return {'type': 'ml_chart', 'image': None} | |
| try: | |
| # Get numeric columns only | |
| numeric_df = df.select_dtypes(include=[np.number]) | |
| if numeric_df.shape[1] < 2: | |
| return {'type': 'ml_chart', 'image': None, 'error': 'Need at least 2 numeric columns'} | |
| # Limit columns for readability | |
| if numeric_df.shape[1] > 10: | |
| numeric_df = numeric_df.iloc[:, :10] | |
| corr = numeric_df.corr() | |
| fig, ax = self.plt.subplots(figsize=(10, 8)) | |
| # Create heatmap | |
| mask = np.triu(np.ones_like(corr, dtype=bool)) | |
| cmap = self.sns.diverging_palette(220, 20, as_cmap=True) | |
| self.sns.heatmap(corr, mask=mask, cmap=cmap, center=0, | |
| square=True, linewidths=0.5, | |
| annot=True, fmt='.2f', annot_kws={'size': 9}, | |
| cbar_kws={'shrink': 0.8}) | |
| ax.set_title(title, color=self.colors['text'], fontsize=14, fontweight='bold') | |
| ax.tick_params(colors=self.colors['text']) | |
| return { | |
| 'type': 'ml_correlation', | |
| 'image': self._fig_to_base64(fig), | |
| 'title': title | |
| } | |
| except Exception as e: | |
| logger.error(f"Error creating correlation heatmap: {e}") | |
| return {'type': 'ml_chart', 'image': None, 'error': str(e)} | |
| def create_distribution_plot( | |
| self, | |
| data: List[float], | |
| column_name: str = "Value", | |
| title: str = None | |
| ) -> Dict[str, Any]: | |
| """Create distribution plot with histogram and KDE""" | |
| if not self.available: | |
| return {'type': 'ml_chart', 'image': None} | |
| try: | |
| fig, ax = self.plt.subplots(figsize=(10, 6)) | |
| # Histogram with KDE | |
| self.sns.histplot(data, kde=True, ax=ax, | |
| color=self.colors['primary'], | |
| edgecolor=self.colors['text'], | |
| alpha=0.7, line_kws={'linewidth': 2}) | |
| # Add mean and std lines | |
| mean_val = np.mean(data) | |
| std_val = np.std(data) | |
| ax.axvline(mean_val, color=self.colors['accent'], linestyle='--', | |
| linewidth=2, label=f'Mean: {mean_val:.2f}') | |
| ax.axvline(mean_val - std_val, color=self.colors['warning'], linestyle=':', | |
| linewidth=1.5, alpha=0.7) | |
| ax.axvline(mean_val + std_val, color=self.colors['warning'], linestyle=':', | |
| linewidth=1.5, alpha=0.7, label=f'±1 Std: {std_val:.2f}') | |
| ax.set_facecolor(self.colors['bg']) | |
| ax.set_xlabel(column_name, color=self.colors['text'], fontsize=12) | |
| ax.set_ylabel('Count', color=self.colors['text'], fontsize=12) | |
| ax.set_title(title or f'Distribution of {column_name}', | |
| color=self.colors['text'], fontsize=14, fontweight='bold') | |
| ax.legend(loc='upper right', framealpha=0.8) | |
| ax.tick_params(colors=self.colors['text']) | |
| ax.grid(True, alpha=0.3, color=self.colors['grid']) | |
| return { | |
| 'type': 'ml_distribution', | |
| 'image': self._fig_to_base64(fig), | |
| 'title': title or f'Distribution of {column_name}' | |
| } | |
| except Exception as e: | |
| logger.error(f"Error creating distribution plot: {e}") | |
| return {'type': 'ml_chart', 'image': None, 'error': str(e)} | |
| def create_residual_plot( | |
| self, | |
| actual: List[float], | |
| predicted: List[float], | |
| title: str = "Residual Analysis" | |
| ) -> Dict[str, Any]: | |
| """Create residual analysis plot for model validation""" | |
| if not self.available: | |
| return {'type': 'ml_chart', 'image': None} | |
| try: | |
| residuals = np.array(actual) - np.array(predicted) | |
| fig, axes = self.plt.subplots(1, 2, figsize=(14, 5)) | |
| # Residuals vs Predicted | |
| ax1 = axes[0] | |
| ax1.scatter(predicted, residuals, color=self.colors['primary'], | |
| alpha=0.6, s=50, edgecolors='white', linewidth=0.5) | |
| ax1.axhline(y=0, color=self.colors['accent'], linestyle='--', linewidth=2) | |
| ax1.set_xlabel('Predicted', color=self.colors['text'], fontsize=11) | |
| ax1.set_ylabel('Residuals', color=self.colors['text'], fontsize=11) | |
| ax1.set_title('Residuals vs Predicted', color=self.colors['text'], fontsize=12) | |
| ax1.set_facecolor(self.colors['bg']) | |
| ax1.tick_params(colors=self.colors['text']) | |
| ax1.grid(True, alpha=0.3, color=self.colors['grid']) | |
| # Residuals distribution | |
| ax2 = axes[1] | |
| self.sns.histplot(residuals, kde=True, ax=ax2, | |
| color=self.colors['secondary'], alpha=0.7) | |
| ax2.axvline(0, color=self.colors['accent'], linestyle='--', linewidth=2) | |
| ax2.set_xlabel('Residual Value', color=self.colors['text'], fontsize=11) | |
| ax2.set_ylabel('Count', color=self.colors['text'], fontsize=11) | |
| ax2.set_title('Residual Distribution', color=self.colors['text'], fontsize=12) | |
| ax2.set_facecolor(self.colors['bg']) | |
| ax2.tick_params(colors=self.colors['text']) | |
| ax2.grid(True, alpha=0.3, color=self.colors['grid']) | |
| fig.suptitle(title, color=self.colors['text'], fontsize=14, fontweight='bold', y=1.02) | |
| self.plt.tight_layout() | |
| return { | |
| 'type': 'ml_residual', | |
| 'image': self._fig_to_base64(fig), | |
| 'title': title | |
| } | |
| except Exception as e: | |
| logger.error(f"Error creating residual plot: {e}") | |
| return {'type': 'ml_chart', 'image': None, 'error': str(e)} | |
| def create_prediction_summary( | |
| self, | |
| historical: List[float], | |
| predictions: List[float], | |
| lower_bound: List[float], | |
| upper_bound: List[float], | |
| importance: Dict[str, float] = None, | |
| title: str = "ML Prediction Summary" | |
| ) -> Dict[str, Any]: | |
| """Create comprehensive prediction visualization with multiple subplots""" | |
| if not self.available: | |
| return {'type': 'ml_chart', 'image': None} | |
| try: | |
| has_importance = importance and len(importance) > 1 | |
| if has_importance: | |
| fig = self.plt.figure(figsize=(14, 8)) | |
| gs = fig.add_gridspec(2, 2, height_ratios=[1.5, 1], hspace=0.3, wspace=0.3) | |
| ax_forecast = fig.add_subplot(gs[0, :]) | |
| ax_importance = fig.add_subplot(gs[1, 0]) | |
| ax_dist = fig.add_subplot(gs[1, 1]) | |
| else: | |
| fig, (ax_forecast, ax_dist) = self.plt.subplots(1, 2, figsize=(14, 5)) | |
| # Forecast plot | |
| hist_x = list(range(len(historical))) | |
| ax_forecast.plot(hist_x, historical, | |
| color=self.colors['primary'], linewidth=2.5, | |
| marker='o', markersize=5, label='Historical') | |
| if predictions: | |
| pred_x = list(range(len(historical), len(historical) + len(predictions))) | |
| ax_forecast.plot(pred_x, predictions, | |
| color=self.colors['accent'], linewidth=2.5, | |
| marker='s', markersize=7, label='Forecast', linestyle='--') | |
| if lower_bound and upper_bound: | |
| ax_forecast.fill_between(pred_x, lower_bound, upper_bound, | |
| color=self.colors['accent'], alpha=0.2, | |
| label='95% CI') | |
| ax_forecast.set_facecolor(self.colors['bg']) | |
| ax_forecast.set_xlabel('Period', color=self.colors['text']) | |
| ax_forecast.set_ylabel('Value', color=self.colors['text']) | |
| ax_forecast.set_title('Forecast with Confidence Interval', | |
| color=self.colors['text'], fontsize=12, fontweight='bold') | |
| ax_forecast.legend(loc='upper left', framealpha=0.8) | |
| ax_forecast.grid(True, alpha=0.3, color=self.colors['grid']) | |
| ax_forecast.tick_params(colors=self.colors['text']) | |
| # Feature importance (if available) | |
| if has_importance: | |
| sorted_imp = sorted(importance.items(), key=lambda x: x[1], reverse=True)[:6] | |
| features = [item[0] for item in sorted_imp] | |
| values = [item[1] for item in sorted_imp] | |
| ax_importance.barh(features[::-1], values[::-1], | |
| color=self.colors['secondary']) | |
| ax_importance.set_facecolor(self.colors['bg']) | |
| ax_importance.set_xlabel('Importance', color=self.colors['text']) | |
| ax_importance.set_title('Key Drivers', color=self.colors['text'], | |
| fontsize=12, fontweight='bold') | |
| ax_importance.tick_params(colors=self.colors['text']) | |
| ax_importance.grid(True, axis='x', alpha=0.3, color=self.colors['grid']) | |
| # Distribution of historical data | |
| self.sns.histplot(historical, kde=True, ax=ax_dist, | |
| color=self.colors['primary'], alpha=0.7) | |
| ax_dist.axvline(np.mean(historical), color=self.colors['accent'], | |
| linestyle='--', linewidth=2, label=f'Mean: {np.mean(historical):.1f}') | |
| ax_dist.set_facecolor(self.colors['bg']) | |
| ax_dist.set_xlabel('Value', color=self.colors['text']) | |
| ax_dist.set_title('Value Distribution', color=self.colors['text'], | |
| fontsize=12, fontweight='bold') | |
| ax_dist.legend(loc='upper right', framealpha=0.8) | |
| ax_dist.tick_params(colors=self.colors['text']) | |
| ax_dist.grid(True, alpha=0.3, color=self.colors['grid']) | |
| fig.suptitle(title, color=self.colors['text'], fontsize=14, fontweight='bold', y=1.01) | |
| self.plt.tight_layout() | |
| return { | |
| 'type': 'ml_summary', | |
| 'image': self._fig_to_base64(fig), | |
| 'title': title | |
| } | |
| except Exception as e: | |
| logger.error(f"Error creating prediction summary: {e}") | |
| return {'type': 'ml_chart', 'image': None, 'error': str(e)} | |
| # Convenience functions | |
| def create_ml_visualization( | |
| viz_type: str, | |
| **kwargs | |
| ) -> Dict[str, Any]: | |
| """Create any ML visualization by type""" | |
| visualizer = MLVisualizer() | |
| if viz_type == 'forecast': | |
| return visualizer.create_forecast_plot(**kwargs) | |
| elif viz_type == 'importance': | |
| return visualizer.create_feature_importance(**kwargs) | |
| elif viz_type == 'correlation': | |
| return visualizer.create_correlation_heatmap(**kwargs) | |
| elif viz_type == 'distribution': | |
| return visualizer.create_distribution_plot(**kwargs) | |
| elif viz_type == 'residual': | |
| return visualizer.create_residual_plot(**kwargs) | |
| elif viz_type == 'summary': | |
| return visualizer.create_prediction_summary(**kwargs) | |
| else: | |
| return {'type': 'error', 'image': None, 'error': f'Unknown viz type: {viz_type}'} | |