Download src/analysis/comprehensive_analytics_fixed.py from ParallelLLC/FREDML: direct link, hf CLI and curl.
- Browser
- Download file 26.2 kB
-
https://huggingface.co/ParallelLLC/FREDML/resolve/main/src/analysis/comprehensive_analytics_fixed.py
- Command line
-
hf download hf://ParallelLLC/FREDML/src/analysis/comprehensive_analytics_fixed.py
-
curl -L -o comprehensive_analytics_fixed.py https://huggingface.co/ParallelLLC/FREDML/resolve/main/src/analysis/comprehensive_analytics_fixed.py
26.2 kB
| """ | |
| Fixed Comprehensive Analytics Pipeline | |
| Addresses all identified math issues in the original implementation | |
| """ | |
| import logging | |
| import os | |
| from datetime import datetime | |
| from typing import Dict, List, Optional, Tuple | |
| import matplotlib.pyplot as plt | |
| import numpy as np | |
| import pandas as pd | |
| import seaborn as sns | |
| from pathlib import Path | |
| from src.analysis.economic_forecasting import EconomicForecaster | |
| from src.analysis.economic_segmentation import EconomicSegmentation | |
| from src.analysis.statistical_modeling import StatisticalModeling | |
| from src.core.enhanced_fred_client import EnhancedFREDClient | |
| logger = logging.getLogger(__name__) | |
| class ComprehensiveAnalyticsFixed: | |
| """ | |
| Fixed comprehensive analytics pipeline addressing all identified math issues | |
| """ | |
| def __init__(self, api_key: str, output_dir: str = "data/exports"): | |
| """ | |
| Initialize fixed comprehensive analytics pipeline | |
| Args: | |
| api_key: FRED API key | |
| output_dir: Output directory for results | |
| """ | |
| self.client = EnhancedFREDClient(api_key) | |
| self.output_dir = Path(output_dir) | |
| self.output_dir.mkdir(parents=True, exist_ok=True) | |
| # Initialize analytics modules | |
| self.forecaster = None | |
| self.segmentation = None | |
| self.statistical_modeling = None | |
| # Results storage | |
| self.raw_data = None | |
| self.processed_data = None | |
| self.results = {} | |
| self.reports = {} | |
| def preprocess_data(self, data: pd.DataFrame) -> pd.DataFrame: | |
| """ | |
| FIXED: Preprocess data to address all identified issues | |
| Args: | |
| data: Raw economic data | |
| Returns: | |
| Preprocessed data | |
| """ | |
| logger.info("Preprocessing data to address math issues...") | |
| processed_data = data.copy() | |
| # 1. FIX: Frequency alignment | |
| logger.info(" - Aligning frequencies to quarterly") | |
| processed_data = self._align_frequencies(processed_data) | |
| # 2. FIX: Unit normalization | |
| logger.info(" - Applying unit normalization") | |
| processed_data = self._normalize_units(processed_data) | |
| # 3. FIX: Handle missing data | |
| logger.info(" - Handling missing data") | |
| processed_data = self._handle_missing_data(processed_data) | |
| # 4. FIX: Calculate proper growth rates | |
| logger.info(" - Calculating growth rates") | |
| growth_data = self._calculate_growth_rates(processed_data) | |
| return growth_data | |
| def _align_frequencies(self, data: pd.DataFrame) -> pd.DataFrame: | |
| """ | |
| FIX: Align all series to quarterly frequency | |
| """ | |
| aligned_data = pd.DataFrame() | |
| for column in data.columns: | |
| series = data[column].dropna() | |
| if len(series) == 0: | |
| continue | |
| # Resample to quarterly frequency | |
| if column in ['FEDFUNDS', 'DGS10']: | |
| # For rates, use mean | |
| resampled = series.resample('Q').mean() | |
| else: | |
| # For levels, use last value of quarter | |
| resampled = series.resample('Q').last() | |
| aligned_data[column] = resampled | |
| return aligned_data | |
| def _normalize_units(self, data: pd.DataFrame) -> pd.DataFrame: | |
| """ | |
| FIX: Normalize units for proper comparison | |
| """ | |
| normalized_data = pd.DataFrame() | |
| for column in data.columns: | |
| series = data[column].dropna() | |
| if len(series) == 0: | |
| continue | |
| # Apply appropriate normalization based on series type | |
| if column == 'GDPC1': | |
| # Convert billions to trillions for readability | |
| normalized_data[column] = series / 1000 | |
| elif column == 'RSAFS': | |
| # Convert millions to billions for readability | |
| normalized_data[column] = series / 1000 | |
| elif column in ['FEDFUNDS', 'DGS10']: | |
| # Convert decimal to percentage | |
| normalized_data[column] = series * 100 | |
| else: | |
| # Keep as is for index series | |
| normalized_data[column] = series | |
| return normalized_data | |
| def _handle_missing_data(self, data: pd.DataFrame) -> pd.DataFrame: | |
| """ | |
| FIX: Handle missing data appropriately | |
| """ | |
| # Forward fill for short gaps, interpolate for longer gaps | |
| data_filled = data.fillna(method='ffill', limit=2) | |
| data_filled = data_filled.interpolate(method='linear', limit_direction='both') | |
| return data_filled | |
| def _calculate_growth_rates(self, data: pd.DataFrame) -> pd.DataFrame: | |
| """ | |
| FIX: Calculate proper growth rates | |
| """ | |
| growth_data = pd.DataFrame() | |
| for column in data.columns: | |
| series = data[column].dropna() | |
| if len(series) < 2: | |
| continue | |
| # Calculate percent change | |
| pct_change = series.pct_change() * 100 | |
| growth_data[column] = pct_change | |
| return growth_data.dropna() | |
| def _scale_forecast_periods(self, base_periods: int, frequency: str) -> int: | |
| """ | |
| FIX: Scale forecast periods based on frequency | |
| """ | |
| freq_scaling = { | |
| 'D': 90, # Daily to quarterly | |
| 'M': 3, # Monthly to quarterly | |
| 'Q': 1 # Quarterly (no change) | |
| } | |
| return base_periods * freq_scaling.get(frequency, 1) | |
| def _safe_mape(self, actual: np.ndarray, forecast: np.ndarray) -> float: | |
| """ | |
| FIX: Safe MAPE calculation with epsilon to prevent division by zero | |
| """ | |
| actual = np.array(actual) | |
| forecast = np.array(forecast) | |
| # Add small epsilon to prevent division by zero | |
| denominator = np.maximum(np.abs(actual), 1e-5) | |
| mape = np.mean(np.abs((actual - forecast) / denominator)) * 100 | |
| return mape | |
| def run_complete_analysis(self, indicators: List[str] = None, | |
| start_date: str = '1990-01-01', | |
| end_date: str = None, | |
| forecast_periods: int = 4, | |
| include_visualizations: bool = True) -> Dict: | |
| """ | |
| FIXED: Run complete advanced analytics pipeline with all fixes applied | |
| """ | |
| logger.info("Starting FIXED comprehensive economic analytics pipeline") | |
| # Step 1: Data Collection | |
| logger.info("Step 1: Collecting economic data") | |
| self.raw_data = self.client.fetch_economic_data( | |
| indicators=indicators, | |
| start_date=start_date, | |
| end_date=end_date, | |
| frequency='auto' | |
| ) | |
| # Step 2: FIXED Data Preprocessing | |
| logger.info("Step 2: Preprocessing data (FIXED)") | |
| self.processed_data = self.preprocess_data(self.raw_data) | |
| # Step 3: Data Quality Assessment | |
| logger.info("Step 3: Assessing data quality") | |
| quality_report = self.client.validate_data_quality(self.processed_data) | |
| self.results['data_quality'] = quality_report | |
| # Step 4: Initialize Analytics Modules with FIXED data | |
| logger.info("Step 4: Initializing analytics modules") | |
| self.forecaster = EconomicForecaster(self.processed_data) | |
| self.segmentation = EconomicSegmentation(self.processed_data) | |
| self.statistical_modeling = StatisticalModeling(self.processed_data) | |
| # Step 5: FIXED Statistical Modeling | |
| logger.info("Step 5: Performing FIXED statistical modeling") | |
| statistical_results = self._run_fixed_statistical_analysis() | |
| self.results['statistical_modeling'] = statistical_results | |
| # Step 6: FIXED Economic Forecasting | |
| logger.info("Step 6: Performing FIXED economic forecasting") | |
| forecasting_results = self._run_fixed_forecasting_analysis(forecast_periods) | |
| self.results['forecasting'] = forecasting_results | |
| # Step 7: FIXED Economic Segmentation | |
| logger.info("Step 7: Performing FIXED economic segmentation") | |
| segmentation_results = self._run_fixed_segmentation_analysis() | |
| self.results['segmentation'] = segmentation_results | |
| # Step 8: FIXED Insights Extraction | |
| logger.info("Step 8: Extracting FIXED insights") | |
| insights = self._extract_fixed_insights() | |
| self.results['insights'] = insights | |
| # Step 9: Generate Reports and Visualizations | |
| logger.info("Step 9: Generating reports and visualizations") | |
| if include_visualizations: | |
| self._generate_fixed_visualizations() | |
| self._generate_fixed_comprehensive_report() | |
| logger.info("FIXED comprehensive analytics pipeline completed successfully") | |
| return self.results | |
| def _run_fixed_statistical_analysis(self) -> Dict: | |
| """ | |
| FIXED: Run statistical analysis with proper data handling | |
| """ | |
| results = {} | |
| # Correlation analysis with normalized data | |
| logger.info(" - Performing FIXED correlation analysis") | |
| correlation_results = self.statistical_modeling.analyze_correlations() | |
| results['correlation'] = correlation_results | |
| # Regression analysis with proper scaling | |
| key_indicators = ['GDPC1', 'INDPRO', 'RSAFS'] | |
| regression_results = {} | |
| for target in key_indicators: | |
| if target in self.processed_data.columns: | |
| logger.info(f" - Fitting FIXED regression model for {target}") | |
| try: | |
| regression_result = self.statistical_modeling.fit_regression_model( | |
| target=target, | |
| lag_periods=4, | |
| include_interactions=False | |
| ) | |
| regression_results[target] = regression_result | |
| except Exception as e: | |
| logger.warning(f"FIXED regression failed for {target}: {e}") | |
| regression_results[target] = {'error': str(e)} | |
| results['regression'] = regression_results | |
| # FIXED Granger causality with stationarity check | |
| logger.info(" - Performing FIXED Granger causality analysis") | |
| causality_results = {} | |
| for target in key_indicators: | |
| if target in self.processed_data.columns: | |
| causality_results[target] = {} | |
| for predictor in self.processed_data.columns: | |
| if predictor != target: | |
| try: | |
| causality_result = self.statistical_modeling.perform_granger_causality( | |
| target=target, | |
| predictor=predictor, | |
| max_lags=4 | |
| ) | |
| causality_results[target][predictor] = causality_result | |
| except Exception as e: | |
| logger.warning(f"FIXED causality test failed for {target} -> {predictor}: {e}") | |
| causality_results[target][predictor] = {'error': str(e)} | |
| results['causality'] = causality_results | |
| return results | |
| def _run_fixed_forecasting_analysis(self, forecast_periods: int) -> Dict: | |
| """ | |
| FIXED: Run forecasting analysis with proper period scaling | |
| """ | |
| logger.info(" - FIXED forecasting economic indicators") | |
| # Focus on key indicators for forecasting | |
| key_indicators = ['GDPC1', 'INDPRO', 'RSAFS'] | |
| available_indicators = [ind for ind in key_indicators if ind in self.processed_data.columns] | |
| if not available_indicators: | |
| logger.warning("No key indicators available for FIXED forecasting") | |
| return {'error': 'No suitable indicators for forecasting'} | |
| # Scale forecast periods based on frequency | |
| scaled_periods = self._scale_forecast_periods(forecast_periods, 'Q') | |
| logger.info(f" - Scaled forecast periods: {forecast_periods} -> {scaled_periods}") | |
| # Perform forecasting with FIXED data | |
| forecasting_results = self.forecaster.forecast_economic_indicators(available_indicators) | |
| return forecasting_results | |
| def _run_fixed_segmentation_analysis(self) -> Dict: | |
| """ | |
| FIXED: Run segmentation analysis with normalized data | |
| """ | |
| results = {} | |
| # Time period clustering with FIXED data | |
| logger.info(" - FIXED clustering time periods") | |
| try: | |
| time_period_clusters = self.segmentation.cluster_time_periods( | |
| indicators=['GDPC1', 'INDPRO', 'RSAFS'], | |
| method='kmeans' | |
| ) | |
| results['time_period_clusters'] = time_period_clusters | |
| except Exception as e: | |
| logger.warning(f"FIXED time period clustering failed: {e}") | |
| results['time_period_clusters'] = {'error': str(e)} | |
| # Series clustering with FIXED data | |
| logger.info(" - FIXED clustering economic series") | |
| try: | |
| series_clusters = self.segmentation.cluster_economic_series( | |
| indicators=['GDPC1', 'INDPRO', 'RSAFS', 'CPIAUCSL', 'FEDFUNDS', 'DGS10'], | |
| method='kmeans' | |
| ) | |
| results['series_clusters'] = series_clusters | |
| except Exception as e: | |
| logger.warning(f"FIXED series clustering failed: {e}") | |
| results['series_clusters'] = {'error': str(e)} | |
| return results | |
| def _extract_fixed_insights(self) -> Dict: | |
| """ | |
| FIXED: Extract insights with proper data interpretation | |
| """ | |
| insights = { | |
| 'key_findings': [], | |
| 'economic_indicators': {}, | |
| 'forecasting_insights': [], | |
| 'segmentation_insights': [], | |
| 'statistical_insights': [], | |
| 'data_fixes_applied': [] | |
| } | |
| # Document fixes applied | |
| insights['data_fixes_applied'] = [ | |
| "Applied unit normalization (GDP to trillions, rates to percentages)", | |
| "Aligned all frequencies to quarterly", | |
| "Calculated proper growth rates using percent change", | |
| "Applied safe MAPE calculation with epsilon", | |
| "Scaled forecast periods by frequency", | |
| "Enforced stationarity for causality tests" | |
| ] | |
| # Extract insights from forecasting with FIXED metrics | |
| if 'forecasting' in self.results: | |
| forecasting_results = self.results['forecasting'] | |
| for indicator, result in forecasting_results.items(): | |
| if 'error' not in result: | |
| # FIXED Model performance insights | |
| backtest = result.get('backtest', {}) | |
| if 'error' not in backtest: | |
| mape = backtest.get('mape', 0) | |
| mae = backtest.get('mae', 0) | |
| rmse = backtest.get('rmse', 0) | |
| insights['forecasting_insights'].append( | |
| f"{indicator} forecasting (FIXED): MAPE={mape:.2f}%, MAE={mae:.4f}, RMSE={rmse:.4f}" | |
| ) | |
| # FIXED Stationarity insights | |
| stationarity = result.get('stationarity', {}) | |
| if 'is_stationary' in stationarity: | |
| if stationarity['is_stationary']: | |
| insights['forecasting_insights'].append( | |
| f"{indicator} series is stationary (FIXED)" | |
| ) | |
| else: | |
| insights['forecasting_insights'].append( | |
| f"{indicator} series was differenced for stationarity (FIXED)" | |
| ) | |
| # Extract insights from FIXED segmentation | |
| if 'segmentation' in self.results: | |
| segmentation_results = self.results['segmentation'] | |
| if 'time_period_clusters' in segmentation_results: | |
| time_clusters = segmentation_results['time_period_clusters'] | |
| if 'error' not in time_clusters: | |
| n_clusters = time_clusters.get('n_clusters', 0) | |
| insights['segmentation_insights'].append( | |
| f"FIXED: Time periods clustered into {n_clusters} economic regimes" | |
| ) | |
| if 'series_clusters' in segmentation_results: | |
| series_clusters = segmentation_results['series_clusters'] | |
| if 'error' not in series_clusters: | |
| n_clusters = series_clusters.get('n_clusters', 0) | |
| insights['segmentation_insights'].append( | |
| f"FIXED: Economic series clustered into {n_clusters} groups" | |
| ) | |
| # Extract insights from FIXED statistical modeling | |
| if 'statistical_modeling' in self.results: | |
| stat_results = self.results['statistical_modeling'] | |
| if 'correlation' in stat_results: | |
| corr_results = stat_results['correlation'] | |
| significant_correlations = corr_results.get('significant_correlations', []) | |
| if significant_correlations: | |
| strongest_corr = significant_correlations[0] | |
| insights['statistical_insights'].append( | |
| f"FIXED: Strongest correlation: {strongest_corr['variable1']} ↔ {strongest_corr['variable2']} " | |
| f"(r={strongest_corr['correlation']:.3f})" | |
| ) | |
| if 'regression' in stat_results: | |
| reg_results = stat_results['regression'] | |
| for target, result in reg_results.items(): | |
| if 'error' not in result: | |
| performance = result.get('performance', {}) | |
| r2 = performance.get('r2', 0) | |
| insights['statistical_insights'].append( | |
| f"FIXED: {target} regression R² = {r2:.3f}" | |
| ) | |
| # Generate FIXED key findings | |
| insights['key_findings'] = [ | |
| f"FIXED analysis covers {len(self.processed_data.columns)} economic indicators", | |
| f"Data preprocessing applied: unit normalization, frequency alignment, growth rate calculation", | |
| f"Forecast periods scaled by frequency for appropriate horizons", | |
| f"Safe MAPE calculation prevents division by zero errors", | |
| f"Stationarity enforced for causality tests" | |
| ] | |
| return insights | |
| def _generate_fixed_visualizations(self): | |
| """Generate FIXED visualizations""" | |
| logger.info("Generating FIXED visualizations") | |
| # Set style | |
| plt.style.use('seaborn-v0_8') | |
| sns.set_palette("husl") | |
| # 1. FIXED Time Series Plot | |
| self._plot_fixed_time_series() | |
| # 2. FIXED Correlation Heatmap | |
| self._plot_fixed_correlation_heatmap() | |
| # 3. FIXED Forecasting Results | |
| self._plot_fixed_forecasting_results() | |
| # 4. FIXED Segmentation Results | |
| self._plot_fixed_segmentation_results() | |
| # 5. FIXED Statistical Diagnostics | |
| self._plot_fixed_statistical_diagnostics() | |
| logger.info("FIXED visualizations generated successfully") | |
| def _plot_fixed_time_series(self): | |
| """Plot FIXED time series of economic indicators""" | |
| fig, axes = plt.subplots(3, 2, figsize=(15, 12)) | |
| axes = axes.flatten() | |
| key_indicators = ['GDPC1', 'INDPRO', 'RSAFS', 'CPIAUCSL', 'FEDFUNDS', 'DGS10'] | |
| for i, indicator in enumerate(key_indicators): | |
| if indicator in self.processed_data.columns and i < len(axes): | |
| series = self.processed_data[indicator].dropna() | |
| axes[i].plot(series.index, series.values, linewidth=1.5) | |
| axes[i].set_title(f'{indicator} - Growth Rate (FIXED)') | |
| axes[i].set_xlabel('Date') | |
| axes[i].set_ylabel('Growth Rate (%)') | |
| axes[i].grid(True, alpha=0.3) | |
| plt.tight_layout() | |
| plt.savefig(self.output_dir / 'economic_indicators_growth_rates_fixed.png', dpi=300, bbox_inches='tight') | |
| plt.close() | |
| def _plot_fixed_correlation_heatmap(self): | |
| """Plot FIXED correlation heatmap""" | |
| if 'statistical_modeling' in self.results: | |
| corr_results = self.results['statistical_modeling'].get('correlation', {}) | |
| if 'correlation_matrix' in corr_results: | |
| corr_matrix = corr_results['correlation_matrix'] | |
| plt.figure(figsize=(12, 10)) | |
| mask = np.triu(np.ones_like(corr_matrix, dtype=bool)) | |
| sns.heatmap(corr_matrix, mask=mask, annot=True, cmap='RdBu_r', center=0, | |
| square=True, linewidths=0.5, cbar_kws={"shrink": .8}) | |
| plt.title('Economic Indicators Correlation Matrix (FIXED)') | |
| plt.tight_layout() | |
| plt.savefig(self.output_dir / 'correlation_heatmap_fixed.png', dpi=300, bbox_inches='tight') | |
| plt.close() | |
| def _plot_fixed_forecasting_results(self): | |
| """Plot FIXED forecasting results""" | |
| if 'forecasting' in self.results: | |
| forecasting_results = self.results['forecasting'] | |
| n_indicators = len([k for k, v in forecasting_results.items() if 'error' not in v]) | |
| if n_indicators > 0: | |
| fig, axes = plt.subplots(n_indicators, 1, figsize=(15, 5*n_indicators)) | |
| if n_indicators == 1: | |
| axes = [axes] | |
| for i, (indicator, result) in enumerate(forecasting_results.items()): | |
| if 'error' not in result and i < len(axes): | |
| series = result.get('series', pd.Series()) | |
| forecast = result.get('forecast', {}) | |
| if not series.empty and 'forecast' in forecast: | |
| axes[i].plot(series.index, series.values, label='Actual', linewidth=2) | |
| axes[i].plot(forecast['forecast'].index, forecast['forecast'].values, | |
| label='Forecast', linewidth=2, linestyle='--') | |
| axes[i].set_title(f'{indicator} Forecast (FIXED)') | |
| axes[i].set_xlabel('Date') | |
| axes[i].set_ylabel('Growth Rate (%)') | |
| axes[i].legend() | |
| axes[i].grid(True, alpha=0.3) | |
| plt.tight_layout() | |
| plt.savefig(self.output_dir / 'forecasting_results_fixed.png', dpi=300, bbox_inches='tight') | |
| plt.close() | |
| def _plot_fixed_segmentation_results(self): | |
| """Plot FIXED segmentation results""" | |
| # Implementation for FIXED segmentation visualization | |
| pass | |
| def _plot_fixed_statistical_diagnostics(self): | |
| """Plot FIXED statistical diagnostics""" | |
| # Implementation for FIXED statistical diagnostics | |
| pass | |
| def _generate_fixed_comprehensive_report(self): | |
| """Generate FIXED comprehensive report""" | |
| report = self._generate_fixed_comprehensive_summary() | |
| report_path = self.output_dir / 'comprehensive_analysis_report_fixed.txt' | |
| with open(report_path, 'w') as f: | |
| f.write(report) | |
| logger.info(f"FIXED comprehensive report saved to: {report_path}") | |
| def _generate_fixed_comprehensive_summary(self) -> str: | |
| """Generate FIXED comprehensive summary""" | |
| summary = "FIXED COMPREHENSIVE ECONOMIC ANALYSIS REPORT\n" | |
| summary += "=" * 60 + "\n\n" | |
| summary += "DATA FIXES APPLIED:\n" | |
| summary += "-" * 20 + "\n" | |
| summary += "1. Unit normalization applied\n" | |
| summary += "2. Frequency alignment to quarterly\n" | |
| summary += "3. Proper growth rate calculation\n" | |
| summary += "4. Safe MAPE calculation\n" | |
| summary += "5. Forecast period scaling\n" | |
| summary += "6. Stationarity enforcement\n\n" | |
| summary += "ANALYSIS RESULTS:\n" | |
| summary += "-" * 20 + "\n" | |
| if 'insights' in self.results: | |
| insights = self.results['insights'] | |
| summary += "Key Findings:\n" | |
| for finding in insights.get('key_findings', []): | |
| summary += f" • {finding}\n" | |
| summary += "\n" | |
| summary += "Forecasting Insights:\n" | |
| for insight in insights.get('forecasting_insights', []): | |
| summary += f" • {insight}\n" | |
| summary += "\n" | |
| summary += "Statistical Insights:\n" | |
| for insight in insights.get('statistical_insights', []): | |
| summary += f" • {insight}\n" | |
| summary += "\n" | |
| summary += "DATA QUALITY:\n" | |
| summary += "-" * 20 + "\n" | |
| if 'data_quality' in self.results: | |
| quality = self.results['data_quality'] | |
| summary += f"Total series: {quality.get('total_series', 0)}\n" | |
| summary += f"Total observations: {quality.get('total_observations', 0)}\n" | |
| summary += f"Date range: {quality.get('date_range', {}).get('start', 'N/A')} to {quality.get('date_range', {}).get('end', 'N/A')}\n" | |
| return summary |