Edwin Salguero
Enhanced FRED ML with improved Reports & Insights page, fixed alignment analysis, and comprehensive analytics improvements
2469150
| #!/usr/bin/env python3 | |
| """ | |
| Comprehensive analytics testing module for FRED ML | |
| Consolidates functionality from multiple test files into enterprise-grade test suite | |
| """ | |
| import sys | |
| import os | |
| import pytest | |
| import pandas as pd | |
| import numpy as np | |
| from datetime import datetime, timedelta | |
| # Add project root to path | |
| sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../..'))) | |
| class TestAnalyticsImports: | |
| """Test analytics module imports and basic functionality""" | |
| def test_imports(self): | |
| """Test if all required modules can be imported""" | |
| try: | |
| from src.core.enhanced_fred_client import EnhancedFREDClient | |
| from src.analysis.comprehensive_analytics import ComprehensiveAnalytics | |
| from src.analysis.economic_forecasting import EconomicForecaster | |
| from src.analysis.economic_segmentation import EconomicSegmentation | |
| from src.analysis.statistical_modeling import StatisticalModeling | |
| assert True | |
| except ImportError as e: | |
| pytest.fail(f"Import test failed: {e}") | |
| def test_fred_client_structure(self): | |
| """Test FRED client functionality""" | |
| try: | |
| from src.core.enhanced_fred_client import EnhancedFREDClient | |
| client = EnhancedFREDClient("test_key") | |
| # Test basic functionality - check for the correct method names | |
| assert hasattr(client, 'fetch_economic_data') | |
| assert hasattr(client, 'fetch_quarterly_data') | |
| except Exception as e: | |
| pytest.fail(f"FRED Client test failed: {e}") | |
| def test_analytics_structure(self): | |
| """Test analytics module structure""" | |
| try: | |
| from src.analysis.comprehensive_analytics import ComprehensiveAnalytics | |
| analytics = ComprehensiveAnalytics("test_key") | |
| required_methods = [ | |
| 'run_complete_analysis', | |
| '_run_statistical_analysis', | |
| '_run_forecasting_analysis', | |
| '_run_segmentation_analysis', | |
| '_extract_insights' | |
| ] | |
| for method in required_methods: | |
| assert hasattr(analytics, method), f"Missing method: {method}" | |
| except Exception as e: | |
| pytest.fail(f"Analytics structure test failed: {e}") | |
| class TestMathematicalFixes: | |
| """Test mathematical fixes and data processing""" | |
| def setup_method(self): | |
| """Set up test data""" | |
| self.dates = pd.date_range('2020-01-01', periods=100, freq='ME') | |
| self.test_data = pd.DataFrame({ | |
| 'GDPC1': np.random.normal(22000, 1000, 100), # Billions | |
| 'INDPRO': np.random.normal(100, 5, 100), # Index | |
| 'CPIAUCSL': np.random.normal(250, 10, 100), # Index | |
| 'FEDFUNDS': np.random.normal(2, 0.5, 100), # Percent | |
| 'PAYEMS': np.random.normal(150000, 5000, 100) # Thousands | |
| }, index=self.dates) | |
| def test_mathematical_fixes_import(self): | |
| """Test mathematical fixes module import""" | |
| try: | |
| from src.analysis.mathematical_fixes import MathematicalFixes | |
| fixes = MathematicalFixes() | |
| assert fixes is not None | |
| except ImportError as e: | |
| pytest.fail(f"Mathematical fixes import failed: {e}") | |
| def test_unit_normalization(self): | |
| """Test unit normalization functionality""" | |
| try: | |
| from src.analysis.mathematical_fixes import MathematicalFixes | |
| fixes = MathematicalFixes() | |
| normalized_data = fixes.normalize_units(self.test_data) | |
| assert normalized_data.shape == self.test_data.shape | |
| assert not normalized_data.isnull().all().all() | |
| except Exception as e: | |
| pytest.fail(f"Unit normalization test failed: {e}") | |
| def test_frequency_alignment(self): | |
| """Test frequency alignment functionality""" | |
| try: | |
| from src.analysis.mathematical_fixes import MathematicalFixes | |
| fixes = MathematicalFixes() | |
| aligned_data = fixes.align_frequencies(self.test_data, target_freq='QE') | |
| # The aligned data might be longer due to interpolation | |
| assert len(aligned_data) > 0 | |
| assert not aligned_data.isnull().all().all() | |
| except Exception as e: | |
| pytest.fail(f"Frequency alignment test failed: {e}") | |
| def test_growth_rate_calculation(self): | |
| """Test growth rate calculation""" | |
| try: | |
| from src.analysis.mathematical_fixes import MathematicalFixes | |
| fixes = MathematicalFixes() | |
| growth_data = fixes.calculate_growth_rates(self.test_data, method='pct_change') | |
| assert growth_data.shape == self.test_data.shape | |
| # Growth rates should have some NaN values (first row) | |
| assert growth_data.isnull().sum().sum() > 0 | |
| except Exception as e: | |
| pytest.fail(f"Growth rate calculation test failed: {e}") | |
| def test_comprehensive_fixes(self): | |
| """Test comprehensive fixes application""" | |
| try: | |
| from src.analysis.mathematical_fixes import MathematicalFixes | |
| fixes = MathematicalFixes() | |
| fixed_data, fix_info = fixes.apply_comprehensive_fixes( | |
| self.test_data, | |
| target_freq='QE', | |
| growth_method='pct_change', | |
| normalize_units=True | |
| ) | |
| assert fixed_data is not None | |
| assert isinstance(fix_info, dict) | |
| # Check for any of the expected keys in fix_info | |
| expected_keys = ['fixes_applied', 'frequency_alignment', 'growth_calculation', 'unit_normalization'] | |
| assert any(key in fix_info for key in expected_keys) | |
| except Exception as e: | |
| pytest.fail(f"Comprehensive fixes test failed: {e}") | |
| class TestConfiguration: | |
| """Test configuration and environment setup""" | |
| def test_config_loading(self): | |
| """Test configuration loading""" | |
| try: | |
| # Test if config can be loaded | |
| import os | |
| fred_key = os.getenv('FRED_API_KEY', 'test_key') | |
| assert fred_key is not None | |
| assert len(fred_key) > 0 | |
| except Exception as e: | |
| pytest.fail(f"Configuration test failed: {e}") | |
| def test_config_import(self): | |
| """Test config.settings import""" | |
| try: | |
| from config.settings import Config | |
| assert Config is not None | |
| except ImportError: | |
| # Config import might fail in test environment, which is OK | |
| pass | |
| class TestAppFunctionality: | |
| """Test application functionality and health checks""" | |
| def test_app_health_check(self): | |
| """Test app health check functionality""" | |
| # This would test the actual app if running | |
| # For now, just test the function exists | |
| def mock_health_check(): | |
| return True | |
| assert mock_health_check() is True | |
| def test_fred_api_integration(self): | |
| """Test FRED API integration""" | |
| try: | |
| import requests | |
| # Test with a mock API call | |
| api_key = "test_key" | |
| test_url = f"https://api.stlouisfed.org/fred/series?series_id=GDP&api_key={api_key}&file_type=json" | |
| # This would fail with test key, but we're testing the structure | |
| assert "api.stlouisfed.org" in test_url | |
| assert "GDP" in test_url | |
| except Exception as e: | |
| pytest.fail(f"FRED API integration test failed: {e}") | |
| class TestDataValidation: | |
| """Test data validation and quality checks""" | |
| def test_data_structure_validation(self): | |
| """Test data structure validation""" | |
| test_data = pd.DataFrame({ | |
| 'GDPC1': [22000, 22100, 22200], | |
| 'INDPRO': [100, 101, 102], | |
| 'CPIAUCSL': [250, 251, 252] | |
| }) | |
| # Basic validation | |
| assert not test_data.empty | |
| assert len(test_data.columns) > 0 | |
| assert len(test_data) > 0 | |
| assert not test_data.isnull().all().all() | |
| def test_data_type_validation(self): | |
| """Test data type validation""" | |
| test_data = pd.DataFrame({ | |
| 'GDPC1': [22000, 22100, 22200], | |
| 'INDPRO': [100, 101, 102], | |
| 'CPIAUCSL': [250, 251, 252] | |
| }) | |
| # Check numeric types | |
| for col in test_data.columns: | |
| assert pd.api.types.is_numeric_dtype(test_data[col]) | |
| if __name__ == "__main__": | |
| pytest.main([__file__]) |