Technical Intermarket Analysis

---




name: technical-intermarket-analysis
compatibility: opencode
completeness: 95
content-types:
- code
- guidance
- config
- do-dont
description: '"Implements cross-market relationships and asset class correlations
  for risk management and algorithmic trading execution."'
license: MIT
maturity: stable
metadata:
  domain: trading
  output-format: code
  related-skills: ai-regime-classification, fundamentals-trading-plan, technical-cycle-analysis
  role: implementation
  scope: implementation
  triggers: asset, cross-market, relationships, technical intermarket analysis, technical-intermarket-analysis
  archetypes:
  - tactical
  anti_triggers:
  - brainstorming
  - vague ideation
  - no risk management
  response_profile:
    verbosity: low
    directive_strength: high
    abstraction_level: operational
version: "1.0.0"




---




**Role:** Analyze relationships between markets to identify divergences and arbitrage opportunities

**Philosophy:** Markets move in concert; intermarket analysis reveals hidden correlations and regime shifts

## Key Principles

1. **Correlation Analysis**: Asset classes exhibit stable and transient correlations
2. **Relative Strength**: Compare performance across markets
3. **Cross-Asset Arbs**: Exploit pricing discrepancies between related markets
4. **Regime Detection**: Market regimes shift correlation structures
5. **Macro Signals**: Bond yields, FX rates predict equity moves

## Implementation Guidelines

### Structure
- Core logic: technical_analysis/intermarket.py
- Helper functions: technical_analysis/correlation_engine.py
- Tests: tests/test_intermarket.py

### Patterns to Follow
- Use rolling correlations withLookback windows
- Track cross-market ratios and spreads
- Monitor correlation breakdowns as risk signals

## Adherence Checklist
Before completing your task, verify:
- [ ] Rolling correlations update on new data with minimum window
- [ ] Correlation breakdowns trigger alerts
- [ ] Cross-market ratios normalized for volatility
- [ ] Latent factor analysis identifies shared drivers
- [ ] Regime shifts detected using change-point detection


Relative paths in this skill (e.g., scripts/, reference/) are relative to this base directory.

## Python Implementation

```python
import numpy as np
import pandas as pd
from typing import Dict, List, Tuple, Optional
from dataclasses import dataclass
from scipy import stats
from scipy.stats import zscore
from statsmodels.tsa.stattools import coint

@dataclass
class CorrelationPair:
    """Correlation relationship between two assets."""
    asset1: str
    asset2: str
    correlation: float
    p_value: float
    lookback_period: int
    trend: str  # 'increasing', 'decreasing', 'stable'

@dataclass
class IntermarketSpread:
    """Spread relationship between markets."""
    asset1: str
    asset2: str
    spread: float
    z_score: float
    mean: float
    std: float

class IntermarketAnalyzer:
    """Analyzes relationships across multiple asset classes."""
    
    def __init__(self, correlation_windows: List[int] = [20, 50, 100]):
        self.windows = correlation_windows
    
    def calculate_rolling_correlation(
        self, returns1: pd.Series, returns2: pd.Series, window: int = 20
    ) -> pd.Series:
        """Calculate rolling correlation between two asset returns."""
        return returns1.rolling(window=window).corr(returns2)
    
    def analyze_correlation_regime(
        self, returns_df: pd.DataFrame, assets: List[str], window: int = 50
    ) -> Dict[str, Dict]:
        """Analyze correlation regime across multiple assets."""
        results = {}
        
        for i, asset1 in enumerate(assets):
            for asset2 in assets[i+1:]:
                returns1 = returns_df[asset1]
                returns2 = returns_df[asset2]
                
                # Short-term correlation
                short_corr = self.calculate_rolling_correlation(returns1, returns2, 20)
                # Long-term correlation
                long_corr = self.calculate_rolling_correlation(returns1, returns2, 100)
                
                # Calculate regime
                short_mean = short_corr.mean()
                short_vol = short_corr.std()
                long_mean = long_corr.mean()
                
                if short_vol > long_mean * 0.5:
                    regime = 'high_volatility'
                elif abs(short_mean) < 0.2:
                    regime = 'weak_correlation'
                else:
                    regime = 'stable'
                
                results[f"{asset1}_{asset2}"] = {
                    'correlation': short_corr.iloc[-1] if len(short_corr) > 0 else 0,
                    'regime': regime,
                    'regime_change': bool(abs(short_corr.iloc[-1] - long_mean) > short_vol)
                }
        
        return results
    
    def calculate_cross_market_ratio(
        self, price1: pd.Series, price2: pd.Series, normalize: bool = True
    ) -> pd.Series:
        """Calculate price ratio between two markets."""
        ratio = price1 / (price2 + 1e-8)
        
        if normalize:
            # Z-score normalize the ratio
            return zscore(ratio)
        
        return ratio
    
    def detect_intermarket_divergence(
        self, prices: Dict[str, pd.Series], base_asset: str
    ) -> List[Dict]:
        """Detect divergences between markets."""
        divergences = []
        base_prices = prices[base_asset]
        
        for asset, prices_df in prices.items():
            if asset == base_asset:
                continue
            
            # Calculate returns correlation over different periods
            base_returns = base_prices.pct_change()
            asset_returns = prices_df.pct_change()
            
            for window in [20, 50, 100]:
                base_rolling = base_returns.rolling(window).mean()
                asset_rolling = asset_returns.rolling(window).mean()
                
                if len(base_rolling) < window:
                    continue
                
                # Check for divergence
                base_z = zscore(base_rolling)[-1]
                asset_z = zscore(asset_rolling)[-1]
                
                if base_z * asset_z < 0 and abs(base_z - asset_z) > 2:
                    divergences.append({
                        'asset': asset,
                        'window': window,
                        'base_z': base_z,
                        'asset_z': asset_z,
                        'strength': abs(base_z - asset_z)
                    })
        
        return divergences
    
    def calculate_cointegration_spread(
        self, price1: pd.Series, price2: pd.Series
    ) -> Tuple[float, float, float]:
        """Calculate cointegration spread for pairs trading."""
        # Engle-Granger two-step method
        # Step 1: Regress price1 on price2
        X = sm.add_constant(price2)
        model = sm.OLS(price1, X).fit()
        
        hedge_ratio = model.params[1]
        spread = price1 - hedge_ratio * price2
        
        # Step 2: Test for stationarity
        from statsmodels.tsa.stattools import adfuller
        adf_result = adfuller(spread)
        
        return spread.mean(), spread.std(), adf_result[1]  # p-value
    
    def intermarket_arb_opportunity(
        self, spreads: Dict[str, pd.Series], z_threshold: float = 2.0
    ) -> List[Dict]:
        """Identify intermarket arbitrage opportunities."""
        opportunities = []
        
        for pair, spread in spreads.items():
            if len(spread) < 20:
                continue
            
            mean = spread.mean()
            std = spread.std()
            current_z = (spread.iloc[-1] - mean) / std if std > 0 else 0
            
            if abs(current_z) > z_threshold:
                direction = 'short' if current_z > 0 else 'long'
                opportunities.append({
                    'pair': pair,
                    'z_score': current_z,
                    'direction': direction,
                    'expected_reversion': abs(current_z) / 2  # Expected z-score decay
                })
        
        return opportunities
    
    def macro_signal_analysis(
        self, market_indices: Dict[str, pd.Series], macro_series: pd.Series
    ) -> Dict[str, float]:
        """Analyze how macro indicators drive market movements."""
        signals = {}
        
        for name, prices in market_indices.items():
            returns = prices.pct_change()
            
            # Calculate sensitivity to macro factor
            combined = pd.concat([returns, macro_series], axis=1).dropna()
            
            if len(combined) < 50:
                continue
            
            X = sm.add_constant(combined.iloc[:, 1])
            y = combined.iloc[:, 0]
            model = sm.OLS(y, X).fit()
            
            signals[name] = model.params[1]  # Beta coefficient
        
        return signals
``````
```
```

---

## Constraints

### MUST DO
- Implement indicator calculations using rolling windows with explicit lookback periods; never use full-history data for online indicators
- Validate signal generation by confirming alignment across multiple independent indicators before acting on a single signal
- Calculate all price-based indicators (SMA, EMA, RSI) on closing prices unless specifically designed for tick data
- Include proper handling of missing/NaN candles in indicator pipelines — forward-fill only within session boundaries
- Log signal generation with the full context window of indicator values that led to each signal

### MUST NOT DO
- Do not use look-ahead bias: never reference future bars or prices when calculating indicators during backtesting
- Avoid recalculating all indicators from scratch on every tick — maintain running state for efficiency
- Never combine indicators with different timeframes without explicit resampling and clear documentation of the alignment logic
- Do not generate signals based on a single indicator crossover; require confirmation from price action or volume
- Avoid hardcoding parameter values (e.g., RSI period = 14) without testing regime-specific optima


## Live References

> Authoritative documentation links for this domain. The model follows markdown links at load time to resolve external references and inline content.

- [Intermarket Analysis: Stocks, Bonds, Commodities (John J. Murphy)](https://www.investopedia.com/articles/investing/072815/intermarket-analysis.asp) — John Murphy's framework for analyzing relationships between stock, bond, commodity, and currency markets
- [Correlation Matrix in Financial Markets (Investopedia)](https://www.investopedia.com/terms/c/correlation-matrix.asp) — Investopedia's guide to using correlation matrices for intermarket relationship analysis
- [SPY vs TLT Correlation (StockCharts)](https://stockcharts.com/free-charting/tech-radar.html) — StockCharts tools for visualizing and analyzing cross-market correlations in real-time
- [Asset Class Correlations (Federal Reserve Economic Data)](https://fred.stlouisfed.org/) — FRED dataset for tracking intermarket relationships including Treasury yields, equity indices, and commodity prices
- [Dow Jones U.S. Composite Index Methodology](https://indexes.dj.com/methodology) — Dow Jones methodology documentation for understanding composite index relationships across sectors