Crypto Correlation Breakdown: How Traditional Markets Impact Digital Assets in 2025

pip install yfinance pandas numpy matplotlib seaborn plotly

import yfinance as yf
import pandas as pd
import numpy as np
from datetime import datetime, timedelta

def fetch_market_data(period="1y"):
    """
    Fetch crypto and traditional market data
    Returns: DataFrame with price data for analysis
    """
    
    # Define tickers for analysis
    tickers = {
        'BTC-USD': 'Bitcoin',
        'ETH-USD': 'Ethereum', 
        '^GSPC': 'S&P 500',
        '^IXIC': 'NASDAQ',
        '^DJI': 'Dow Jones',
        'GLD': 'Gold ETF',
        '^TNX': '10-Year Treasury'
    }
    
    # Download price data
    data = {}
    for ticker, name in tickers.items():
        stock = yf.Ticker(ticker)
        hist = stock.history(period=period)
        data[name] = hist['Close']
    
    # Combine into single DataFrame
    df = pd.DataFrame(data)
    
    # Calculate daily returns
    returns_df = df.pct_change().dropna()
    
    return df, returns_df

# Fetch the data
prices_df, returns_df = fetch_market_data("2y")
print("Data successfully loaded!")
print(f"Date range: {prices_df.index[0].date()} to {prices_df.index[-1].date()}")

def calculate_rolling_correlations(returns_df, window=30):
    """
    Calculate rolling correlations between crypto and traditional assets
    Window: Number of days for rolling calculation
    """
    
    correlations = {}
    crypto_assets = ['Bitcoin', 'Ethereum']
    traditional_assets = ['S&P 500', 'NASDAQ', 'Dow Jones', 'Gold ETF']
    
    for crypto in crypto_assets:
        correlations[crypto] = {}
        for traditional in traditional_assets:
            # Calculate rolling correlation
            rolling_corr = returns_df[crypto].rolling(window=window).corr(
                returns_df[traditional]
            )
            correlations[crypto][traditional] = rolling_corr
    
    return correlations

# Calculate 30-day rolling correlations
rolling_corrs = calculate_rolling_correlations(returns_df, window=30)

# Display current correlation levels
print("Current 30-day correlations:")
for crypto in rolling_corrs:
    print(f"\n{crypto}:")
    for traditional, corr_series in rolling_corrs[crypto].items():
        current_corr = corr_series.iloc[-1]
        print(f"  vs {traditional}: {current_corr:.3f}")

import plotly.graph_objects as go
from plotly.subplots import make_subplots

def plot_correlation_trends(rolling_corrs, crypto_asset='Bitcoin'):
    """
    Create interactive correlation trend charts
    """
    
    fig = make_subplots(
        rows=2, cols=2,
        subplot_titles=['vs S&P 500', 'vs NASDAQ', 'vs Dow Jones', 'vs Gold ETF'],
        vertical_spacing=0.08
    )
    
    traditional_assets = ['S&P 500', 'NASDAQ', 'Dow Jones', 'Gold ETF']
    positions = [(1,1), (1,2), (2,1), (2,2)]
    
    for i, traditional in enumerate(traditional_assets):
        row, col = positions[i]
        
        corr_data = rolling_corrs[crypto_asset][traditional].dropna()
        
        fig.add_trace(
            go.Scatter(
                x=corr_data.index,
                y=corr_data.values,
                mode='lines',
                name=f'{crypto_asset} vs {traditional}',
                line=dict(width=2)
            ),
            row=row, col=col
        )
        
        # Add horizontal reference lines
        fig.add_hline(y=0, line_dash="dash", line_color="gray", 
                     row=row, col=col)
        fig.add_hline(y=0.5, line_dash="dot", line_color="orange", 
                     row=row, col=col)
        fig.add_hline(y=-0.5, line_dash="dot", line_color="orange", 
                     row=row, col=col)
    
    fig.update_layout(
        title=f'{crypto_asset} 30-Day Rolling Correlations with Traditional Markets',
        height=600,
        showlegend=False
    )
    
    fig.update_yaxes(range=[-1, 1], title_text="Correlation")
    fig.show()

# Plot Bitcoin correlations
plot_correlation_trends(rolling_corrs, 'Bitcoin')

def correlation_alert_system(returns_df, threshold=0.7):
    """
    Alert system for high correlation periods
    Threshold: Correlation level that triggers alerts
    """
    
    # Calculate current 7-day correlation
    btc_sp500_corr = returns_df['Bitcoin'].rolling(7).corr(
        returns_df['S&P 500']
    ).iloc[-1]
    
    eth_nasdaq_corr = returns_df['Ethereum'].rolling(7).corr(
        returns_df['NASDAQ']
    ).iloc[-1]
    
    alerts = []
    
    if abs(btc_sp500_corr) > threshold:
        direction = "positive" if btc_sp500_corr > 0 else "negative"
        alerts.append(f"🚨 Bitcoin-S&P 500 {direction} correlation: {btc_sp500_corr:.3f}")
    
    if abs(eth_nasdaq_corr) > threshold:
        direction = "positive" if eth_nasdaq_corr > 0 else "negative"
        alerts.append(f"🚨 Ethereum-NASDAQ {direction} correlation: {eth_nasdaq_corr:.3f}")
    
    return alerts

# Check for current alerts
current_alerts = correlation_alert_system(returns_df, threshold=0.6)
for alert in current_alerts:
    print(alert)

from arch import arch_model
import scipy.stats as stats

def calculate_dcc_correlations(returns_df, crypto_col, traditional_col):
    """
    Calculate Dynamic Conditional Correlation using GARCH models
    More accurate for financial time series with volatility clustering
    """
    
    # Fit GARCH models to each series
    crypto_returns = returns_df[crypto_col].dropna()
    traditional_returns = returns_df[traditional_col].dropna()
    
    # Ensure same length
    min_length = min(len(crypto_returns), len(traditional_returns))
    crypto_returns = crypto_returns.iloc[-min_length:]
    traditional_returns = traditional_returns.iloc[-min_length:]
    
    # Fit GARCH(1,1) models
    crypto_garch = arch_model(crypto_returns * 100, vol='Garch', p=1, q=1)
    traditional_garch = arch_model(traditional_returns * 100, vol='Garch', p=1, q=1)
    
    crypto_fit = crypto_garch.fit(disp='off')
    traditional_fit = traditional_garch.fit(disp='off')
    
    # Extract standardized residuals
    crypto_std_resid = crypto_fit.std_resid
    traditional_std_resid = traditional_fit.std_resid
    
    # Calculate dynamic correlation
    window = 30
    dcc_correlations = []
    
    for i in range(window, len(crypto_std_resid)):
        corr = np.corrcoef(
            crypto_std_resid[i-window:i],
            traditional_std_resid[i-window:i]
        )[0,1]
        dcc_correlations.append(corr)
    
    return pd.Series(dcc_correlations, 
                    index=crypto_returns.index[-len(dcc_correlations):])

# Calculate DCC between Bitcoin and S&P 500
dcc_btc_sp500 = calculate_dcc_correlations(returns_df, 'Bitcoin', 'S&P 500')
print(f"Current DCC correlation: {dcc_btc_sp500.iloc[-1]:.3f}")