Cross Asset Relationships

---
name: cross-asset-relationships
description: "Cross-asset and quantitative analysis: pair correlations, correlation heatmaps, currency strength, cross-timeframe divergence, intermarket analysis, market breadth, carry trades, swap rates, risk premia, and multi-pair baskets. USE FOR: correlation, currency strength, intermarket, market breadth, carry trade, swap rate, risk premia, basket, pair trading, cross asset, quant."
related_skills:
  - market-intelligence
  - portfolio-optimization
  - forex-trading
  - risk-and-portfolio
tags:
  - trading
  - fundamentals
  - correlation
  - intermarket
  - carry-trade
  - cross-asset
skill_level: advanced
kind: reference
category: trading/market-context
status: active
---
> **Skill:** Cross Asset Relationships  |  **Domain:** trading  |  **Category:** fundamentals  |  **Level:** advanced
> **Tags:** `trading`, `fundamentals`, `correlation`, `intermarket`, `carry-trade`, `cross-asset`



---

## Pair Correlation Engine

# Pair Correlation Engine

## Overview
Analyzes statistical relationships between financial instruments across multiple timeframes
and historical periods. Detects regime shifts, divergences, lead-lag relationships, and
provides actionable correlation intelligence for trading decisions.

## Architecture

```
┌───────────────────────────────────────────────────┐
│            Pair Correlation Engine                 │
├────────────┬─────────────┬────────────────────────┤
│ Correlation│ History vs  │ Regime Detection &     │
│ Matrix     │ Current     │ Divergence Scanner     │
└────────────┴─────────────┴────────────────────────┘
```

---

## 1. Core Correlation Computation

```python
import pandas as pd
import numpy as np
from scipy import stats
from scipy.cluster.hierarchy import linkage, fcluster, dendrogram
from typing import Optional
from datetime import datetime, timedelta

def compute_returns(prices: pd.DataFrame, method: str = "log") -> pd.DataFrame:
    """Convert price DataFrame to returns. Columns = symbols."""
    if method == "log":
        return np.log(prices / prices.shift(1)).dropna()
    return prices.pct_change().dropna()

def correlation_matrix(
    prices: pd.DataFrame,
    method: str = "pearson",
    window: Optional[int] = None
) -> pd.DataFrame:
    """
    Full correlation matrix.
    method: pearson, spearman, kendall
    window: if set, uses last N bars only
    """
    returns = compute_returns(prices)
    if window:
        returns = returns.tail(window)
    return returns.corr(method=method)

def rolling_correlation(
    series_a: pd.Series,
    series_b: pd.Series,
    window: int = 60,
    method: str = "pearson"
) -> pd.Series:
    """Rolling window correlation between two series."""
    ret_a = np.log(series_a / series_a.shift(1)).dropna()
    ret_b = np.log(series_b / series_b.shift(1)).dropna()
    aligned = pd.concat([ret_a, ret_b], axis=1).dropna()
    aligned.columns = ["a", "b"]
    return aligned["a"].rolling(window).corr(aligned["b"])
```

---

## 2. Historical vs Current Correlation (Core Feature)

```python
def historical_vs_current(
    prices: pd.DataFrame,
    current_window: int = 30,
    historical_windows: list[int] = [90, 180, 365],
    method: str = "pearson"
) -> dict:
    """
    Compare current correlation regime vs historical norms.
    Returns: {pair: {current, hist_90d, hist_180d, hist_365d, deviation, regime_shift}}
    """
    returns = compute_returns(prices)
    symbols = returns.columns.tolist()
    results = {}

    for i, sym_a in enumerate(symbols):
        for sym_b in symbols[i + 1:]:
            pair_key = f"{sym_a}/{sym_b}"
            pair_data = returns[[sym_a, sym_b]].dropna()

            current_corr = pair_data.tail(current_window).corr().iloc[0, 1]

            hist_corrs = {}
            for w in historical_windows:
                if len(pair_data) >= w:
                    hist_corrs[f"hist_{w}d"] = pair_data.tail(w).corr().iloc[0, 1]
                else:
                    hist_corrs[f"hist_{w}d"] = np.nan

            # Compute deviation from longest available history
            longest_hist = next((hist_corrs[f"hist_{w}d"] for w in sorted(historical_windows, reverse=True)
                                 if not np.isnan(hist_corrs.get(f"hist_{w}d", np.nan))), np.nan)

            deviation = current_corr - longest_hist if not np.isnan(longest_hist) else 0
            regime_shift = abs(deviation) > 0.3  # >0.3 = significant regime change

            results[pair_key] = {
                "current": round(current_corr, 4),
                **{k: round(v, 4) for k, v in hist_corrs.items()},
                "deviation": round(deviation, 4),
                "regime_shift": regime_shift,
                "signal": _interpret_deviation(deviation),
            }

    return results

def _interpret_deviation(dev: float) -> str:
    """Interpret correlation deviation into actionable signal."""
    if abs(dev) < 0.1: return "STABLE — correlations normal"
    if dev > 0.3: return "CONVERGENCE — pairs moving together unusually strongly"
    if dev < -0.3: return "DIVERGENCE — pairs decoupling, watch for mean reversion"
    if dev > 0.1: return "STRENGTHENING — correlation increasing"
    return "WEAKENING — correlation decreasing"
```

---

## 3. Divergence Detection

```python
def detect_divergences(
    prices: pd.DataFrame,
    lookback: int = 60,
    threshold: float = 0.3
) -> list[dict]:
    """
    Find pairs that have recently diverged from their historical correlation.
    These are potential mean-reversion or trend-change signals.
    """
    hvc = historical_vs_current(prices, current_window=lookback // 2)
    divergences = []
    for pair, data in hvc.items():
        if data["regime_shift"]:
            divergences.append({
                "pair": pair,
                "current_corr": data["current"],
                "historical_corr": data.get("hist_365d", data.get("hist_180d", np.nan)),
                "deviation": data["deviation"],
                "type": "convergence_anomaly" if data["deviation"] > 0 else "divergence_anomaly",
                "signal": data["signal"],
                "priority": abs(data["deviation"]),
            })
    return sorted(divergences, key=lambda x: x["priority"], reverse=True)

def spread_analysis(
    price_a: pd.Series,
    price_b: pd.Series,
    window: int = 60
) -> pd.DataFrame:
    """
    Compute normalized spread between two series for pair trading.
    Z-score indicates mean-reversion opportunity.
    """
    # Normalize both series to start at 1.0
    norm_a = price_a / price_a.iloc[0]
    norm_b = price_b / price_b.iloc[0]
    spread = norm_a - norm_b
    z_score = (spread - spread.rolling(window).mean()) / spread.rolling(window).std()
    return pd.DataFrame({
        "spread": spread, "z_score": z_score,
        "upper_band": spread.rolling(window).mean() + 2 * spread.rolling(window).std(),
        "lower_band": spread.rolling(window).mean() - 2 * spread.rolling(window).std(),
    })
```

---

## 4. Lead-Lag Analysis

```python
def lead_lag_analysis(
    series_a: pd.Series,
    series_b: pd.Series,
    max_lag: int = 10
) -> pd.DataFrame:
    """
    Cross-correlation at different lags to detect if one pair leads another.
    Positive lag = A leads B. Negative lag = B leads A.
    """
    ret_a = np.log(series_a / series_a.shift(1)).dropna()
    ret_b = np.log(series_b / series_b.shift(1)).dropna()
    aligned = pd.concat([ret_a, ret_b], axis=1).dropna()
    aligned.columns = ["a", "b"]

    results = []
    for lag in range(-max_lag, max_lag + 1):
        if lag >= 0:
            corr = aligned["a"].iloc[lag:].reset_index(drop=True).corr(aligned["b"].iloc[:len(aligned) - lag].reset_index(drop=True))
        else:
            corr = aligned["a"].iloc[:len(aligned) + lag].reset_index(drop=True).corr(aligned["b"].iloc[-lag:].reset_index(drop=True))
        results.append({"lag": lag, "correlation": round(corr, 4)})

    df = pd.DataFrame(results)
    best = df.loc[df["correlation"].abs().idxmax()]
    df.attrs["best_lag"] = int(best["lag"])
    df.attrs["best_corr"] = float(best["correlation"])
    df.attrs["interpretation"] = (
        f"{'A leads B' if best['lag'] > 0 else 'B leads A' if best['lag'] < 0 else 'Synchronous'} "
        f"by {abs(int(best['lag']))} bars (r={best['correlation']:.4f})"
    )
    return df
```

---

## 5. Cluster Analysis — Group Correlated Pairs

```python
def cluster_pairs(
    prices: pd.DataFrame,
    n_clusters: int = 4,
    method: str = "ward"
) -> dict:
    """
    Hierarchical clustering of instruments by correlation.
    Returns cluster assignments and cluster statistics.
    """
    corr = correlation_matrix(prices)
    distance = np.sqrt(2 * (1 - corr))
    np.fill_diagonal(distance.values, 0)

    condensed = distance.values[np.triu_indices_from(distance.values, k=1)]
    Z = linkage(condensed, method=method)
    labels = fcluster(Z, t=n_clusters, criterion="maxclust")

    clusters = {}
    for sym, cluster_id in zip(corr.columns, labels):
        cid = int(cluster_id)
        if cid not in clusters:
            clusters[cid] = []
        clusters[cid].append(sym)

    return {
        "n_clusters": n_clusters,
        "clusters": clusters,
        "linkage": Z,
        "assignments": dict(zip(corr.columns.tolist(), [int(l) for l in labels])),
    }

def find_hedge_pairs(
    prices: pd.DataFrame,
    target_symbol: str,
    min_negative_corr: float = -0.5
) -> list[dict]:
    """Find instruments negatively correlated to target — potential hedges."""
    corr = correlation_matrix(prices)
    if target_symbol not in corr.columns:
        return []
    target_corr = corr[target_symbol].drop(target_symbol).sort_values()
    hedges = []
    for sym, c in target_corr.items():
        if c <= min_negative_corr:
            hedges.append({"symbol": sym, "correlation": round(c, 4), "hedge_quality": "strong" if c < -0.7 else "moderate"})
    return hedges
```

---

## 6. Correlation Report Generator

```python
def full_correlation_report(
    prices: pd.DataFrame,
    current_window: int = 30
) -> dict:
    """
    Complete correlation intelligence report.
    Pipe this to trading-brain for decision-making.
    """
    return {
        "timestamp": datetime.utcnow().isoformat(),
        "symbols_analyzed": prices.columns.tolist(),
        "n_bars": len(prices),
        "current_matrix": correlation_matrix(prices, window=current_window).to_dict(),
        "historical_matrix": correlation_matrix(prices).to_dict(),
        "historical_vs_current": historical_vs_current(prices, current_window),
        "divergences": detect_divergences(prices),
        "clusters": cluster_pairs(prices),
    }
```

---

## Usage Conventions

1. **Always use log returns** for correlation — more statistically stable than simple returns
2. **Check sample size** — need minimum 30 bars for meaningful correlation
3. **Multi-timeframe** — compute correlations on H1, H4, and D1 for robust signals
4. **Regime shifts > 0.3** are significant and should trigger alerts
5. **Lead-lag > 2 bars** with |r| > 0.3 is a potential predictive signal
6. **Correlation ≠ causation** — always note this in reports


---

## Correlation Heatmap Visualizer

# Correlation Heatmap Visualizer

```python
import pandas as pd, numpy as np, matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
import io, base64

class CorrelationHeatmapVisualizer:
    @staticmethod
    def render(prices: pd.DataFrame, method: str = "pearson", window: int = None, save_path: str = None) -> str:
        returns = np.log(prices / prices.shift(1)).dropna()
        if window: returns = returns.tail(window)
        corr = returns.corr(method=method)
        fig, ax = plt.subplots(figsize=(12, 10), facecolor="#131722")
        ax.set_facecolor("#131722")
        im = ax.imshow(corr.values, cmap="RdYlGn", vmin=-1, vmax=1, aspect="auto")
        ax.set_xticks(range(len(corr.columns))); ax.set_yticks(range(len(corr.columns)))
        ax.set_xticklabels(corr.columns, rotation=45, ha="right", fontsize=8, color="#e0e0e0")
        ax.set_yticklabels(corr.columns, fontsize=8, color="#e0e0e0")
        for i in range(len(corr)):
            for j in range(len(corr)):
                color = "white" if abs(corr.values[i, j]) > 0.5 else "gray"
                ax.text(j, i, f"{corr.values[i,j]:.2f}", ha="center", va="center", fontsize=7, color=color)
        plt.colorbar(im, label="Correlation")
        ax.set_title(f"Correlation Matrix ({method}, {'all data' if not window else f'last {window} bars'})", color="#e0e0e0")
        plt.tight_layout()
        buf = io.BytesIO()
        fig.savefig(buf, format="png", dpi=150, facecolor="#131722")
        plt.close(fig)
        buf.seek(0)
        if save_path:
            with open(save_path, "wb") as f: f.write(buf.read()); buf.seek(0)
        return base64.b64encode(buf.read()).decode("utf-8")
```


---

## Currency Strength Meter

# Currency Strength Meter

```python
import pandas as pd, numpy as np

MAJOR_PAIRS = {
    "USD": ["EURUSD", "GBPUSD", "USDJPY", "USDCHF", "AUDUSD", "USDCAD", "NZDUSD"],
    "EUR": ["EURUSD", "EURJPY", "EURGBP", "EURAUD", "EURCHF", "EURCAD", "EURNZD"],
    "GBP": ["GBPUSD", "EURGBP", "GBPJPY", "GBPAUD", "GBPCAD", "GBPCHF", "GBPNZD"],
    "JPY": ["USDJPY", "EURJPY", "GBPJPY", "AUDJPY", "CADJPY", "CHFJPY", "NZDJPY"],
    "AUD": ["AUDUSD", "EURAUD", "GBPAUD", "AUDJPY", "AUDCAD", "AUDCHF", "AUDNZD"],
    "NZD": ["NZDUSD", "EURNZD", "GBPNZD", "NZDJPY", "AUDNZD", "NZDCAD", "NZDCHF"],
    "CAD": ["USDCAD", "EURCAD", "GBPCAD", "CADJPY", "AUDCAD", "NZDCAD", "CADCHF"],
    "CHF": ["USDCHF", "EURCHF", "GBPCHF", "CHFJPY", "AUDCHF", "NZDCHF", "CADCHF"],
}

class CurrencyStrengthMeter:

    @staticmethod
    def compute(pair_returns: dict, period: int = 20) -> dict:
        """
        Compute relative strength for each currency from pair returns.
        pair_returns: {"EURUSD": pd.Series, "GBPUSD": pd.Series, ...}
        """
        strengths = {}
        for currency in ["USD", "EUR", "GBP", "JPY", "AUD", "NZD", "CAD", "CHF"]:
            scores = []
            for pair in MAJOR_PAIRS.get(currency, []):
                if pair not in pair_returns: continue
                ret = pair_returns[pair].tail(period).sum()
                # If currency is base, positive return = currency strong
                if pair.startswith(currency):
                    scores.append(ret)
                else:
                    scores.append(-ret)  # Currency is quote — inverse
            strengths[currency] = round(np.mean(scores) * 100, 3) if scores else 0

        ranked = sorted(strengths.items(), key=lambda x: x[1], reverse=True)
        strongest = ranked[0]
        weakest = ranked[-1]

        return {
            "strengths": dict(ranked),
            "strongest": {"currency": strongest[0], "score": strongest[1]},
            "weakest": {"currency": weakest[0], "score": weakest[1]},
            "best_trade": f"LONG {strongest[0]} / SHORT {weakest[0]} = {strongest[0]}{weakest[0]}",
            "ranking": [r[0] for r in ranked],
            "period": period,
            "spread": round(strongest[1] - weakest[1], 3),
            "note": "Trade strongest vs weakest for maximum directional edge.",
        }

    @staticmethod
    def divergence_alert(current: dict, previous: dict) -> list[str]:
        """Detect currency strength ranking changes — potential trend shifts."""
        alerts = []
        curr_rank = current["ranking"]
        prev_rank = previous["ranking"]
        for currency in curr_rank:
            curr_pos = curr_rank.index(currency)
            prev_pos = prev_rank.index(currency)
            shift = prev_pos - curr_pos
            if abs(shift) >= 3:
                direction = "STRENGTHENING" if shift > 0 else "WEAKENING"
                alerts.append(f"{currency} {direction} rapidly (moved {abs(shift)} ranks)")
        return alerts
```


---

## Cross Timeframe Divergence Scanner

# Cross Timeframe Divergence Scanner

```python
import pandas as pd, numpy as np
class CrossTFDivergenceScanner:
    @staticmethod
    def scan(data_by_tf: dict) -> dict:
        def _rsi(close, p=14):
            d = close.diff()
            g = d.where(d > 0, 0).rolling(p).mean()
            l = (-d.where(d < 0, 0)).rolling(p).mean()
            return (100 - 100 / (1 + g / l.replace(0, np.nan))).iloc[-1]
        tf_signals = {}
        for tf, df in data_by_tf.items():
            if df.empty or len(df) < 50: continue
            rsi = _rsi(df["close"])
            ema50 = df["close"].ewm(span=50).mean().iloc[-1]
            above_ema = df["close"].iloc[-1] > ema50
            tf_signals[tf] = {"rsi": round(rsi, 1), "above_ema50": above_ema,
                             "bias": "BULLISH" if rsi > 50 and above_ema else "BEARISH" if rsi < 50 and not above_ema else "MIXED"}
        biases = [s["bias"] for s in tf_signals.values()]
        all_agree = len(set(biases)) == 1
        conflicts = []
        tfs = list(tf_signals.keys())
        for i in range(len(tfs)):
            for j in range(i+1, len(tfs)):
                if tf_signals[tfs[i]]["bias"] != tf_signals[tfs[j]]["bias"]:
                    conflicts.append(f"{tfs[i]}({tf_signals[tfs[i]]['bias']}) vs {tfs[j]}({tf_signals[tfs[j]]['bias']})")
        return {
            "tf_signals": tf_signals, "all_aligned": all_agree, "conflicts": conflicts,
            "n_conflicts": len(conflicts),
            "action": "CLEAR — all TFs agree" if all_agree else f"CAUTION — {len(conflicts)} TF conflict(s): {', '.join(conflicts[:3])}",
        }
```


---

## Intermarket Divergence Trader

# Intermarket Divergence Trader

```python
import pandas as pd, numpy as np

KNOWN_RELATIONSHIPS = {
    "XAUUSD_DXY": {"correlation": -0.80, "logic": "Gold and USD are inversely correlated"},
    "USDCAD_OIL": {"correlation": -0.70, "logic": "CAD strengthens when oil rises"},
    "USDJPY_SPX": {"correlation": 0.65, "logic": "Risk-on: stocks up, JPY weakens"},
    "EURUSD_DE10Y_US10Y": {"correlation": 0.60, "logic": "EUR follows yield differential"},
    "AUDUSD_COPPER": {"correlation": 0.70, "logic": "AUD follows commodity demand"},
}

class IntermarketDivergenceTrader:
    @staticmethod
    def detect_divergence(series_a: pd.Series, series_b: pd.Series, expected_corr: float,
                           window: int = 30) -> dict:
        norm_a = series_a / series_a.iloc[0]
        norm_b = series_b / series_b.iloc[0]
        recent_corr = norm_a.tail(window).pct_change().corr(norm_b.tail(window).pct_change())
        diverging = abs(recent_corr - expected_corr) > 0.3
        a_move = (norm_a.iloc[-1] / norm_a.iloc[-window] - 1) * 100
        b_move = (norm_b.iloc[-1] / norm_b.iloc[-window] - 1) * 100
        return {
            "expected_correlation": expected_corr,
            "current_correlation": round(recent_corr, 3),
            "diverging": diverging,
            "a_move_pct": round(a_move, 2), "b_move_pct": round(b_move, 2),
            "signal": f"CONVERGENCE TRADE — expect reversion to normal relationship" if diverging else "NO DIVERGENCE",
            "trade": f"If A moved more, fade A. If B moved more, fade B." if diverging else "No trade.",
        }
```


---

## Market Breadth Analyzer

# Market Breadth Analyzer

```python
import pandas as pd, numpy as np

class MarketBreadthAnalyzer:
    @staticmethod
    def scan_breadth(pairs_data: dict, trend_threshold: float = 25) -> dict:
        trending_up = []; trending_down = []; ranging = []
        for sym, df in pairs_data.items():
            if df.empty or len(df) < 50: continue
            close = df["close"]
            ema50 = close.ewm(span=50).mean().iloc[-1]
            plus_dm = df["high"].diff().clip(lower=0).rolling(14).mean()
            minus_dm = (-df["low"].diff()).clip(lower=0).rolling(14).mean()
            atr = (df["high"] - df["low"]).rolling(14).mean()
            dx = abs(plus_dm - minus_dm) / (plus_dm + minus_dm + 1e-10) * 100
            adx = dx.rolling(14).mean().iloc[-1]
            if adx > trend_threshold and close.iloc[-1] > ema50: trending_up.append(sym)
            elif adx > trend_threshold and close.iloc[-1] < ema50: trending_down.append(sym)
            else: ranging.append(sym)
        total = len(trending_up) + len(trending_down) + len(ranging)
        return {
            "trending_up": trending_up, "trending_down": trending_down, "ranging": ranging,
            "breadth_score": round((len(trending_up) - len(trending_down)) / max(total, 1) * 100, 1),
            "pct_trending": round((len(trending_up) + len(trending_down)) / max(total, 1) * 100, 1),
            "market_mode": "TRENDING" if len(trending_up) + len(trending_down) > len(ranging) else "RANGE-BOUND",
            "bias": "RISK-ON" if len(trending_up) > len(trending_down) * 1.5 else "RISK-OFF" if len(trending_down) > len(trending_up) * 1.5 else "MIXED",
        }
```


---

## Carry Trade Calculator

# Carry Trade Calculator

```python
import pandas as pd
import numpy as np

CENTRAL_BANK_RATES = {
    "USD": 5.50, "EUR": 4.50, "GBP": 5.25, "JPY": 0.10,
    "AUD": 4.35, "NZD": 5.50, "CAD": 5.00, "CHF": 1.75,
}  # UPDATE via web_search before using

class CarryTradeCalculator:

    @staticmethod
    def rate_differential(base: str, quote: str) -> dict:
        r_base = CENTRAL_BANK_RATES.get(base, 0)
        r_quote = CENTRAL_BANK_RATES.get(quote, 0)
        diff = r_base - r_quote
        return {
            "pair": f"{base}{quote}",
            "base_rate": r_base, "quote_rate": r_quote,
            "differential": round(diff, 2),
            "carry_direction": "LONG" if diff > 0 else "SHORT" if diff < 0 else "NEUTRAL",
            "annual_carry_pct": round(abs(diff), 2),
            "daily_carry_pct": round(abs(diff) / 365, 4),
        }

    @staticmethod
    def swap_income(lot_size: float, daily_swap_pips: float, hold_days: int) -> dict:
        total_swap = daily_swap_pips * lot_size * 10 * hold_days  # $10 per pip per lot
        return {
            "daily_swap_usd": round(daily_swap_pips * lot_size * 10, 2),
            "total_swap_usd": round(total_swap, 2),
            "hold_days": hold_days,
            "annualized_pct": round(total_swap / (lot_size * 100000) * 365 / max(hold_days, 1) * 100, 2),
        }

    @staticmethod
    def rank_carry_pairs() -> list[dict]:
        """Rank all major pairs by carry attractiveness."""
        pairs = []
        currencies = list(CENTRAL_BANK_RATES.keys())
        for i, base in enumerate(currencies):
            for quote in currencies[i+1:]:
                diff = CarryTradeCalculator.rate_differential(base, quote)
                pairs.append(diff)
                # Reverse pair
                diff_rev = CarryTradeCalculator.rate_differential(quote, base)
                pairs.append(diff_rev)
        return sorted(pairs, key=lambda x: abs(x["differential"]), reverse=True)[:20]

    @staticmethod
    def carry_risk_assessment(pair: str, annual_vol: float, carry_pct: float) -> dict:
        """Carry-to-risk ratio: is the carry worth the volatility?"""
        ratio = carry_pct / max(annual_vol, 0.01)
        return {
            "pair": pair,
            "carry_pct": carry_pct,
            "annual_vol_pct": annual_vol,
            "carry_to_risk": round(ratio, 3),
            "verdict": "ATTRACTIVE" if ratio > 0.5 else "MARGINAL" if ratio > 0.25 else "NOT WORTH IT",
        }
```


---

## Swap Rate Optimizer

# Swap Rate Optimizer

```python
class SwapOptimizer:
    @staticmethod
    def optimize(swap_data: dict, risk_tolerance: float = 0.5) -> dict:
        """swap_data: {pair: {long_swap: X, short_swap: Y}}"""
        positive_swaps = []
        for pair, swaps in swap_data.items():
            if swaps.get("long_swap", 0) > 0:
                positive_swaps.append({"pair": pair, "direction": "long", "daily_swap": swaps["long_swap"]})
            if swaps.get("short_swap", 0) > 0:
                positive_swaps.append({"pair": pair, "direction": "short", "daily_swap": swaps["short_swap"]})
        ranked = sorted(positive_swaps, key=lambda x: x["daily_swap"], reverse=True)
        return {
            "best_swaps": ranked[:10],
            "triple_swap_day": "Wednesday (covers weekend — 3x swap charged)",
            "strategy": "Hold positive-swap positions through Wednesday close for 3x income",
            "warning": "Swap income is secondary to directional P&L. Never hold a losing trade just for swap.",
        }
```


---

## Risk Premia Harvester

# Risk Premia Harvester

```python
import pandas as pd, numpy as np

class RiskPremiaHarvester:
    @staticmethod
    def momentum_factor(returns_by_pair: pd.DataFrame, lookback: int = 60) -> dict:
        """Long top momentum, short bottom momentum. Classic cross-sectional momentum."""
        mom = returns_by_pair.tail(lookback).sum()
        ranked = mom.rank(pct=True)
        longs = ranked[ranked > 0.7].index.tolist()
        shorts = ranked[ranked < 0.3].index.tolist()
        return {"factor": "momentum", "longs": longs, "shorts": shorts, "lookback": lookback,
                "spread_return": round((mom[longs].mean() - mom[shorts].mean()) * 100, 2) if longs and shorts else 0}

    @staticmethod
    def value_factor(pairs: dict) -> dict:
        """Value = trade towards PPP or REER fair value. Long undervalued, short overvalued."""
        longs = [p for p, v in pairs.items() if v.get("undervalued")]
        shorts = [p for p, v in pairs.items() if v.get("overvalued")]
        return {"factor": "value", "longs": longs, "shorts": shorts,
                "note": "PPP-based value reverts over 1-3 year horizons. Very slow factor."}

    @staticmethod
    def carry_factor(rate_diffs: dict) -> dict:
        """Long high-yielding, short low-yielding currencies."""
        sorted_pairs = sorted(rate_diffs.items(), key=lambda x: x[1], reverse=True)
        longs = [p for p, r in sorted_pairs[:3]]
        shorts = [p for p, r in sorted_pairs[-3:]]
        return {"factor": "carry", "longs": longs, "shorts": shorts,
                "carry_spread": round(sorted_pairs[0][1] - sorted_pairs[-1][1], 2)}

    @staticmethod
    def volatility_factor(vol_by_pair: dict) -> dict:
        """Sell high IV, buy low IV. Variance risk premium harvesting."""
        sorted_pairs = sorted(vol_by_pair.items(), key=lambda x: x[1].get("iv_rank", 50), reverse=True)
        sells = [p for p, v in sorted_pairs[:3] if v.get("iv_rank", 50) > 70]
        buys = [p for p, v in sorted_pairs[-3:] if v.get("iv_rank", 50) < 30]
        return {"factor": "volatility", "sell_vol": sells, "buy_vol": buys}

    @staticmethod
    def combined_portfolio(momentum: dict, carry: dict, vol: dict) -> dict:
        """Equal-weight across factors for diversified risk premia portfolio."""
        all_longs = set(momentum.get("longs", []) + carry.get("longs", []))
        all_shorts = set(momentum.get("shorts", []) + vol.get("sell_vol", []))
        return {
            "portfolio_longs": list(all_longs),
            "portfolio_shorts": list(all_shorts),
            "n_factors": 3,
            "rebalance": "Monthly",
            "note": "Factor portfolios earn ~2-5% annually each. Combined = ~6-12% with lower vol than any single factor.",
        }
```


---

## Multi Pair Basket Trader

# Multi-Pair Basket Trader

```python
import numpy as np

class BasketTrader:
    CURRENCY_BASKETS = {
        "USD_LONG": {"EURUSD": "sell", "GBPUSD": "sell", "AUDUSD": "sell", "NZDUSD": "sell", "USDCAD": "buy", "USDJPY": "buy", "USDCHF": "buy"},
        "USD_SHORT": {"EURUSD": "buy", "GBPUSD": "buy", "AUDUSD": "buy", "NZDUSD": "buy", "USDCAD": "sell", "USDJPY": "sell", "USDCHF": "sell"},
        "EUR_LONG": {"EURUSD": "buy", "EURJPY": "buy", "EURGBP": "buy", "EURAUD": "buy"},
        "RISK_ON": {"AUDUSD": "buy", "NZDUSD": "buy", "USDJPY": "buy", "USDCHF": "sell"},
        "RISK_OFF": {"USDJPY": "sell", "USDCHF": "buy", "XAUUSD": "buy", "AUDUSD": "sell"},
    }

    @staticmethod
    def generate_basket_orders(basket_name: str, total_risk_pct: float = 2.0, account_balance: float = 10000) -> dict:
        basket = BasketTrader.CURRENCY_BASKETS.get(basket_name)
        if not basket: return {"error": f"Unknown basket: {basket_name}"}
        n_pairs = len(basket)
        risk_per_pair = total_risk_pct / n_pairs
        return {
            "basket": basket_name,
            "orders": [{"pair": p, "direction": d, "risk_pct": round(risk_per_pair, 2)} for p, d in basket.items()],
            "total_risk": total_risk_pct,
            "n_pairs": n_pairs,
            "risk_per_pair": round(risk_per_pair, 2),
            "advantage": "Diversified execution — single currency view, spread across pairs to reduce pair-specific noise",
        }
```

---

## CointegrationEngine & HMMRegimeDetector (Enhancement)

Ctrl click to launch VS Code Native REPL

---

## Related Skills

- [Market Intelligence](../market-intelligence.md)
- [Portfolio Optimization](../portfolio-optimization.md)
- [Forex Trading](../forex-trading.md)
- [Risk And Portfolio](../risk-and-portfolio.md)