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__]) |