File size: 8,214 Bytes
099d8d9 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 |
#!/usr/bin/env python3
"""
Alignment and Divergence Analysis Test
Test the new alignment/divergence analyzer with real FRED data
"""
import os
import sys
import pandas as pd
import numpy as np
from datetime import datetime
# Add src to path
sys.path.append(os.path.join(os.path.dirname(__file__), 'src'))
from src.core.enhanced_fred_client import EnhancedFREDClient
from src.analysis.alignment_divergence_analyzer import AlignmentDivergenceAnalyzer
def test_alignment_divergence_analysis():
"""Test the new alignment and divergence analysis"""
# Use the provided API key
api_key = "acf8bbec7efe3b6dfa6ae083e7152314"
print("=== ALIGNMENT & DIVERGENCE ANALYSIS TEST ===")
print("Using Spearman correlation for long-term alignment detection")
print("Using Z-score analysis for sudden deviation detection")
print()
try:
# Initialize FRED client
client = EnhancedFREDClient(api_key)
# Fetch economic data (last 5 years for better trend analysis)
end_date = datetime.now()
start_date = end_date.replace(year=end_date.year - 5)
print("1. Fetching economic data...")
data = client.fetch_economic_data(
start_date=start_date.strftime('%Y-%m-%d'),
end_date=end_date.strftime('%Y-%m-%d')
)
if data.empty:
print("❌ No data fetched")
return
print(f"✅ Fetched {len(data)} observations across {len(data.columns)} indicators")
print(f" Date range: {data.index.min()} to {data.index.max()}")
print(f" Indicators: {list(data.columns)}")
print()
# Initialize alignment analyzer
analyzer = AlignmentDivergenceAnalyzer(data)
# 2. Analyze long-term alignment using Spearman correlation
print("2. Analyzing long-term alignment (Spearman correlation)...")
alignment_results = analyzer.analyze_long_term_alignment(
window_sizes=[12, 24, 48], # 1, 2, 4 years for quarterly data
min_periods=8
)
print("✅ Long-term alignment analysis completed")
print(f" Analyzed {len(alignment_results['rolling_correlations'])} indicator pairs")
# Show alignment summary
summary = alignment_results['alignment_summary']
print(f" Increasing alignment pairs: {len(summary['increasing_alignment'])}")
print(f" Decreasing alignment pairs: {len(summary['decreasing_alignment'])}")
print(f" Stable alignment pairs: {len(summary['stable_alignment'])}")
print(f" Strong trends: {len(summary['strong_trends'])}")
print()
# Show some specific alignment trends
if summary['increasing_alignment']:
print("🔺 Examples of increasing alignment:")
for pair in summary['increasing_alignment'][:3]:
print(f" - {pair}")
print()
if summary['decreasing_alignment']:
print("🔻 Examples of decreasing alignment:")
for pair in summary['decreasing_alignment'][:3]:
print(f" - {pair}")
print()
# 3. Detect sudden deviations using Z-score analysis
print("3. Detecting sudden deviations (Z-score analysis)...")
deviation_results = analyzer.detect_sudden_deviations(
z_threshold=2.0, # Flag deviations beyond 2 standard deviations
window_size=12, # 3-year rolling window for quarterly data
min_periods=6
)
print("✅ Sudden deviation detection completed")
# Show deviation summary
dev_summary = deviation_results['deviation_summary']
print(f" Total deviations detected: {dev_summary['total_deviations']}")
print(f" Indicators with deviations: {len(dev_summary['indicators_with_deviations'])}")
print(f" Extreme events: {dev_summary['extreme_events_count']}")
print()
# Show most volatile indicators
if dev_summary['most_volatile_indicators']:
print("📈 Most volatile indicators:")
for item in dev_summary['most_volatile_indicators'][:5]:
print(f" - {item['indicator']}: {item['volatility']:.4f} volatility")
print()
# Show extreme events
extreme_events = deviation_results['extreme_events']
if extreme_events:
print("🚨 Recent extreme events (Z-score > 3.0):")
for indicator, events in extreme_events.items():
if events['events']:
extreme_events_list = [e for e in events['events'] if abs(e['z_score']) > 3.0]
if extreme_events_list:
latest = extreme_events_list[0]
print(f" - {indicator}: {latest['date'].strftime('%Y-%m-%d')} "
f"(Z-score: {latest['z_score']:.2f}, Growth: {latest['growth_rate']:.2f}%)")
print()
# 4. Generate insights report
print("4. Generating comprehensive insights report...")
insights_report = analyzer.generate_insights_report()
print("✅ Insights report generated")
print()
# Save insights to file
with open('alignment_divergence_insights.txt', 'w') as f:
f.write(insights_report)
print("📄 Insights report saved to 'alignment_divergence_insights.txt'")
print()
# 5. Create visualization
print("5. Creating alignment analysis visualization...")
analyzer.plot_alignment_analysis(save_path='alignment_analysis_plot.png')
print("📊 Visualization saved to 'alignment_analysis_plot.png'")
print()
# 6. Detailed analysis examples
print("6. Detailed analysis examples:")
print()
# Show specific correlation trends
if alignment_results['trend_analysis']:
print("📊 Correlation Trend Examples:")
for pair_name, trends in list(alignment_results['trend_analysis'].items())[:3]:
print(f" {pair_name}:")
for window_name, trend_info in trends.items():
if trend_info['trend'] != 'insufficient_data':
print(f" {window_name}: {trend_info['trend']} ({trend_info['strength']})")
print(f" Slope: {trend_info['slope']:.4f}, R²: {trend_info['r_squared']:.3f}")
print()
# Show specific deviation patterns
if deviation_results['z_scores']:
print("⚠️ Deviation Pattern Examples:")
for indicator, z_scores in list(deviation_results['z_scores'].items())[:3]:
deviations = deviation_results['deviations'][indicator]
if not deviations.empty:
print(f" {indicator}:")
print(f" Total deviations: {len(deviations)}")
print(f" Max Z-score: {deviations.abs().max():.2f}")
print(f" Mean Z-score: {deviations.abs().mean():.2f}")
print(f" Recent deviations: {len(deviations[deviations.index > '2023-01-01'])}")
print()
print("=== ANALYSIS COMPLETED SUCCESSFULLY ===")
print("✅ Spearman correlation analysis for long-term alignment")
print("✅ Z-score analysis for sudden deviation detection")
print("✅ Comprehensive insights and visualizations generated")
print()
print("Key findings:")
print("- Long-term alignment patterns identified using rolling Spearman correlation")
print("- Sudden deviations flagged using Z-score analysis")
print("- Extreme events detected and categorized")
print("- Volatility patterns analyzed across indicators")
except Exception as e:
print(f"❌ Error during alignment/divergence analysis: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
test_alignment_divergence_analysis() |