# 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("""
""")
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
)