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

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)

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

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

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

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

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

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

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

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

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

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

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

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

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

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
• Portfolio Optimization
• Forex Trading
• Risk And Portfolio