# app.py import gradio as gr import pandas as pd import plotly.graph_objects as go from plotly.subplots import make_subplots from datetime import datetime from geo_macro import UnifiedMarketDataDownloader, FRED_API_KEY from feature_engineering import MarketRegimeDetector # ==================== PROFESSIONAL COLOR SCHEME ==================== COLORS = { 'crisis': '#DC2626', # Red 'recession': '#F59E0B', # Amber 'stagflation': '#8B5CF6', # Purple 'expansion': '#10B981', # Green 'transition': '#6B7280', # Gray 'primary': '#2563EB', # Blue 'secondary': '#64748B', # Slate } REGIME_CONFIG = { 'FINANCIAL_CRISIS': {'color': COLORS['crisis'], 'icon': '🚨'}, 'RECESSION_WARNING': {'color': COLORS['recession'], 'icon': 'âš ī¸'}, 'STAGFLATION': {'color': COLORS['stagflation'], 'icon': '📉'}, 'EXPANSION': {'color': COLORS['expansion'], 'icon': '📈'}, 'TRANSITION': {'color': COLORS['transition'], 'icon': '🔄'}, } # ==================== DATA CACHING ==================== _cached_df = None _cached_dates = (None, None) def get_data(start_date: str, end_date: str): """Fetch market data with caching""" global _cached_df, _cached_dates if _cached_df is not None and _cached_dates == (start_date, end_date): return _cached_df.copy() print(f"đŸ“Ĩ Downloading data from {start_date} to {end_date}...") downloader = UnifiedMarketDataDownloader(fred_api_key=FRED_API_KEY) df = downloader.download_all_data(start_date=start_date, end_date=end_date) _cached_df = df.copy() _cached_dates = (start_date, end_date) return df # ==================== VISUALIZATION FUNCTIONS ==================== def create_summary_card(latest): """Professional HTML summary card with key metrics""" regime = str(latest['regime']) config = REGIME_CONFIG.get(regime, REGIME_CONFIG['TRANSITION']) confidence = latest.get('regime_confidence', 0) html = f"""

{config['icon']} Market Regime Analysis

{latest.name.strftime('%b %d, %Y') if hasattr(latest.name, 'strftime') else 'Latest'}
Current Regime
{regime.replace('_', ' ')}
Confidence: {confidence:.0%}
Recession Risk
{latest.get('recession_probability', 0):.0%}
Crisis Risk
{latest.get('financial_crisis_risk', 0):.0%}
Stagflation Risk
{latest.get('stagflation_risk', 0):.0%}
Expansion Probability
{latest.get('expansion_probability', 0):.0%}
â„šī¸
Methodology: Empirically validated indicators from 50+ years of market history. Leading indicators provide 6-18 month predictive signals.
""" return html def create_regime_probabilities_chart(latest): """Horizontal bar chart for regime probabilities""" probs = { 'Expansion': latest.get('expansion_probability', 0), 'Stagflation': latest.get('stagflation_risk', 0), 'Recession': latest.get('recession_probability', 0), 'Crisis': latest.get('financial_crisis_risk', 0), } colors = [COLORS['expansion'], COLORS['stagflation'], COLORS['recession'], COLORS['crisis']] fig = go.Figure(go.Bar( y=list(probs.keys()), x=list(probs.values()), orientation='h', marker=dict( color=colors, line=dict(color='white', width=2) ), text=[f"{v:.0%}" for v in probs.values()], textposition='outside', textfont=dict(size=14, color='#1F2937', weight=600), hovertemplate='%{y}
Probability: %{x:.1%}' )) fig.update_layout( title=dict( text="Regime Probability Analysis", font=dict(size=18, color='#1F2937'), x=0.5, xanchor='center' ), xaxis=dict( title="Probability", tickformat='.0%', range=[0, 1], gridcolor='#E5E7EB', showgrid=True ), yaxis=dict( title="", tickfont=dict(size=13, color='#1F2937') ), height=350, plot_bgcolor='white', paper_bgcolor='white', margin=dict(t=60, b=50, l=120, r=100), font=dict(family="Inter, Arial, sans-serif") ) return fig def create_leading_indicators_dashboard(latest): """Multi-panel dashboard for key leading indicators""" fig = make_subplots( rows=2, cols=2, subplot_titles=( 'Yield Curve Spread', 'Credit Stress Index', 'Copper/Gold Ratio', 'Consumer Rotation' ), specs=[[{'type': 'indicator'}, {'type': 'indicator'}], [{'type': 'indicator'}, {'type': 'indicator'}]], vertical_spacing=0.25, horizontal_spacing=0.15 ) # Yield Curve spread = latest.get('yield_curve_spread', 0) spread_color = COLORS['crisis'] if spread < -0.15 else COLORS['expansion'] fig.add_trace(go.Indicator( mode="number+delta+gauge", value=spread, delta={'reference': 0, 'valueformat': '.2f'}, gauge={ 'axis': {'range': [-1.5, 1.5]}, 'bar': {'color': spread_color}, 'threshold': { 'line': {'color': COLORS['crisis'], 'width': 3}, 'thickness': 0.75, 'value': -0.15 }, 'steps': [ {'range': [-1.5, -0.15], 'color': '#FEE2E2'}, {'range': [-0.15, 0], 'color': '#FEF3C7'}, {'range': [0, 1.5], 'color': '#D1FAE5'} ] }, number={'suffix': '%', 'font': {'size': 28}}, domain={'row': 0, 'column': 0} ), row=1, col=1) # Credit Stress credit_stress = latest.get('credit_spread_proxy', 0) fig.add_trace(go.Indicator( mode="number+gauge", value=credit_stress * 100, gauge={ 'axis': {'range': [0, 10]}, 'bar': {'color': COLORS['recession']}, 'threshold': { 'line': {'color': COLORS['crisis'], 'width': 3}, 'thickness': 0.75, 'value': 5 }, 'steps': [ {'range': [0, 3], 'color': '#D1FAE5'}, {'range': [3, 5], 'color': '#FEF3C7'}, {'range': [5, 10], 'color': '#FEE2E2'} ] }, number={'suffix': '', 'font': {'size': 28}}, domain={'row': 0, 'column': 1} ), row=1, col=2) # Copper/Gold cu_au = latest.get('copper_gold_ratio', 0) cu_au_color = COLORS['crisis'] if cu_au < 0.002 else COLORS['expansion'] fig.add_trace(go.Indicator( mode="number+gauge", value=cu_au * 1000, gauge={ 'axis': {'range': [0, 5]}, 'bar': {'color': cu_au_color}, 'threshold': { 'line': {'color': COLORS['crisis'], 'width': 3}, 'thickness': 0.75, 'value': 2 }, 'steps': [ {'range': [0, 2], 'color': '#FEE2E2'}, {'range': [2, 3], 'color': '#FEF3C7'}, {'range': [3, 5], 'color': '#D1FAE5'} ] }, number={'suffix': ' ×10âģÂŗ', 'font': {'size': 24}}, domain={'row': 1, 'column': 0} ), row=2, col=1) # Consumer Rotation rotation = latest.get('consumer_rotation_ratio', 0) rotation_color = COLORS['recession'] if rotation < 1.5 else COLORS['expansion'] fig.add_trace(go.Indicator( mode="number+gauge", value=rotation, gauge={ 'axis': {'range': [1, 3]}, 'bar': {'color': rotation_color}, 'threshold': { 'line': {'color': COLORS['recession'], 'width': 3}, 'thickness': 0.75, 'value': 1.5 }, 'steps': [ {'range': [1, 1.5], 'color': '#FEE2E2'}, {'range': [1.5, 2], 'color': '#FEF3C7'}, {'range': [2, 3], 'color': '#D1FAE5'} ] }, number={'font': {'size': 28}}, domain={'row': 1, 'column': 1} ), row=2, col=2) fig.update_layout( height=600, showlegend=False, paper_bgcolor='white', font=dict(family="Inter, Arial, sans-serif", color='#1F2937'), margin=dict(t=80, b=40, l=40, r=40) ) return fig def create_regime_timeline(features): """Enhanced timeline showing regime history""" tail = features[['regime', 'regime_confidence']].tail(252).copy() if tail.empty: return go.Figure() tail['date'] = tail.index tail['color'] = tail['regime'].map(lambda x: REGIME_CONFIG.get(x, REGIME_CONFIG['TRANSITION'])['color']) fig = go.Figure() # Add scatter with color coding for regime, config in REGIME_CONFIG.items(): mask = tail['regime'] == regime if mask.any(): fig.add_trace(go.Scatter( x=tail[mask]['date'], y=tail[mask]['regime_confidence'], mode='markers', name=regime.replace('_', ' ').title(), marker=dict( color=config['color'], size=10, line=dict(color='white', width=1.5), symbol='circle' ), hovertemplate=( f'{regime.replace("_", " ")}
' + 'Date: %{x|%Y-%m-%d}
' + 'Confidence: %{y:.0%}' ) )) fig.update_layout( title=dict( text="12-Month Regime History", font=dict(size=18, color='#1F2937'), x=0.5, xanchor='center' ), xaxis=dict( gridcolor='#E5E7EB', showgrid=True ), yaxis=dict( title="Regime Confidence", tickformat='.0%', gridcolor='#E5E7EB', showgrid=True, range=[0, 1] ), height=400, plot_bgcolor='white', paper_bgcolor='white', margin=dict(t=60, b=50, l=70, r=40), legend=dict( orientation="h", yanchor="bottom", y=-0.35, xanchor="center", x=0.5, font=dict(size=11) ), font=dict(family="Inter, Arial, sans-serif"), hovermode='closest' ) return fig def create_cross_asset_signals(features): """Multi-line chart showing key cross-asset signals""" tail = features.tail(252) fig = go.Figure() signals = [ ('yield_curve_spread', 'Yield Curve', COLORS['primary']), ('copper_gold_zscore', 'Copper/Gold Z-Score', COLORS['expansion']), ('credit_spread_proxy', 'Credit Spread', COLORS['recession']), ('consumer_confidence_zscore', 'Consumer Confidence', COLORS['stagflation']), ] for col, name, color in signals: if col in tail.columns: fig.add_trace(go.Scatter( x=tail.index, y=tail[col], mode='lines', name=name, line=dict(color=color, width=2), hovertemplate=f'{name}
Date: %{{x|%Y-%m-%d}}
Value: %{{y:.2f}}' )) fig.update_layout( title=dict( text="Cross-Asset Leading Indicators", font=dict(size=18, color='#1F2937'), x=0.5, xanchor='center' ), xaxis=dict( gridcolor='#E5E7EB', showgrid=True ), yaxis=dict( title="Normalized Value", gridcolor='#E5E7EB', showgrid=True, zeroline=True, zerolinecolor='#9CA3AF', zerolinewidth=2 ), height=400, plot_bgcolor='white', paper_bgcolor='white', margin=dict(t=60, b=50, l=70, r=40), legend=dict( orientation="h", yanchor="bottom", y=-0.3, xanchor="center", x=0.5 ), font=dict(family="Inter, Arial, sans-serif"), hovermode='x unified' ) return fig # ==================== MAIN PIPELINE ==================== def run_pipeline(days_back: int = 1825): """Execute the full analysis pipeline""" try: today = pd.Timestamp.today() start_date = (today - pd.Timedelta(days=days_back)).strftime('%Y-%m-%d') end_date = today.strftime('%Y-%m-%d') # Fetch data df = get_data(start_date, end_date) if len(df) < 300: error_html = """

âš ī¸ Insufficient Data

Not enough data points for reliable analysis. Please increase the lookback window to at least 1000 days.

""" return error_html, None, None, None, None, None # Build features print("Building regime features...") detector = MarketRegimeDetector(df) features = detector.build_all_features() # Get latest data point with valid regime latest = features.dropna(subset=['regime']).iloc[-1] # Create visualizations summary_html = create_summary_card(latest) prob_chart = create_regime_probabilities_chart(latest) indicators_dash = create_leading_indicators_dashboard(latest) timeline = create_regime_timeline(features) cross_asset = create_cross_asset_signals(features) # Detailed JSON output json_output = { "📊 Current Status": { "Regime": str(latest['regime']), "Confidence": f"{latest.get('regime_confidence', 0):.1%}", "Date": latest.name.strftime('%Y-%m-%d') if hasattr(latest.name, 'strftime') else 'N/A' }, "đŸŽ¯ Regime Probabilities": { "Recession": f"{latest.get('recession_probability', 0):.1%}", "Financial Crisis": f"{latest.get('financial_crisis_risk', 0):.1%}", "Stagflation": f"{latest.get('stagflation_risk', 0):.1%}", "Expansion": f"{latest.get('expansion_probability', 0):.1%}" }, "📈 Leading Indicators": { "Yield Curve Spread": f"{latest.get('yield_curve_spread', 0):.2f}%", "Yield Curve Inverted": bool(latest.get('yield_curve_inverted', 0)), "Copper/Gold Ratio": f"{latest.get('copper_gold_ratio', 0):.4f}", "Consumer Rotation": f"{latest.get('consumer_rotation_ratio', 0):.2f}", "Credit Stress": bool(latest.get('credit_stress', 0)) }, "đŸŒĄī¸ Market Health": { "VIX Level": f"{latest.get('vix_level', 0):.1f}", "S&P 500 3M Return": f"{latest.get('sp500_return_3m', 0):.1%}", "Dollar Strength": f"{latest.get('dollar_strength', 0):.1f}", "Inflation YoY": f"{latest.get('inflation_yoy', 0):.1f}%", "Unemployment Rate": f"{latest.get('unemployment_rate', 0):.1f}%" } } return summary_html, json_output, prob_chart, indicators_dash, timeline, cross_asset except Exception as e: import traceback error_detail = traceback.format_exc() error_html = f"""

❌ Error

{str(e)}

Show technical details
{error_detail}
""" return error_html, {"Error": str(e)}, None, None, None, None # ==================== GRADIO UI ==================== custom_css = """ @import url('https://fonts.googleapis.com/css2?family=Inter:wght@400;600;700;800&display=swap'); .gradio-container { font-family: 'Inter', -apple-system, BlinkMacSystemFont, sans-serif !important; max-width: 1600px !important; margin: auto !important; } .header-banner { background: linear-gradient(135deg, #2563EB 0%, #1E40AF 100%); color: white; padding: 40px 30px; border-radius: 12px; margin-bottom: 30px; box-shadow: 0 10px 25px rgba(37, 99, 235, 0.2); } .header-banner h1 { margin: 0; color: white; font-size: 36px; font-weight: 800; letter-spacing: -0.5px; } .header-banner p { margin: 12px 0 0 0; color: white; font-size: 16px; opacity: 0.95; font-weight: 500; } .btn-primary { background: linear-gradient(135deg, #2563EB 0%, #1E40AF 100%) !important; border: none !important; font-weight: 700 !important; font-size: 15px !important; padding: 12px 24px !important; border-radius: 8px !important; box-shadow: 0 4px 12px rgba(37, 99, 235, 0.3) !important; transition: all 0.2s !important; } .btn-primary:hover { transform: translateY(-2px) !important; box-shadow: 0 6px 16px rgba(37, 99, 235, 0.4) !important; } """ with gr.Blocks(css=custom_css, title="Geopolitics Risk Analysis", theme=gr.themes.Soft()) as demo: gr.HTML("""

Geopolitics Risk Analysis

Market Regime Detector

""") with gr.Row(): with gr.Column(scale=3): days = gr.Slider( 365, 3000, value=1825, step=90, label="📅 Lookback Window (days)", info="Minimum 1000 days recommended for stable regime detection" ) with gr.Column(scale=1): run_btn = gr.Button( "🔄 Update Analysis", variant="primary", size="lg" ) gr.Markdown("---") with gr.Row(): with gr.Column(scale=2): summary_html = gr.HTML(label="Executive Summary") with gr.Column(scale=1): json_output = gr.JSON(label="📋 Detailed Metrics", show_label=True) gr.Markdown("---") gr.Markdown("## 📊 Regime Probability Analysis") with gr.Row(): prob_chart = gr.Plot(label="Regime Probabilities") indicators_dash = gr.Plot(label="Leading Indicators Dashboard") gr.Markdown("---") gr.Markdown("## 📈 Historical Analysis & Cross-Asset Signals") with gr.Row(): timeline_plot = gr.Plot(label="12-Month Regime Timeline") cross_asset_plot = gr.Plot(label="Cross-Asset Leading Indicators") gr.Markdown("---") gr.HTML("""

📚 Methodology & Data Sources

Leading Indicators (6-18 month predictive): Yield curve inversion, credit spreads (HYG/TLT), copper/gold ratio, consumer rotation (XLY/XLP). These signals have preceded major recessions since 1970s.

Historical Validation: All thresholds derived from documented episodes including 2000 dot-com crash, 2008 GFC, 2020 COVID recession, and 2022 inflation surge.

Data Sources: Yahoo Finance (equity/commodity prices), FRED Economic Data (macro indicators), updated daily. Framework based on peer-reviewed research and central bank methodologies.

""") # Event handlers run_btn.click( run_pipeline, inputs=[days], outputs=[summary_html, json_output, prob_chart, indicators_dash, timeline_plot, cross_asset_plot] ) # Auto-run on load demo.load( run_pipeline, inputs=[days], outputs=[summary_html, json_output, prob_chart, indicators_dash, timeline_plot, cross_asset_plot] ) if __name__ == "__main__": demo.launch( share=False, server_name="0.0.0.0", server_port=7860, show_error=True )