| 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
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
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.
"""
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
- Always use log returns for correlation — more statistically stable than simple returns
- Check sample size — need minimum 30 bars for meaningful correlation
- Multi-timeframe — compute correlations on H1, H4, and D1 for robust signals
- Regime shifts > 0.3 are significant and should trigger alerts
- Lead-lag > 2 bars with |r| > 0.3 is a potential predictive signal
- 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 pair.startswith(currency):
scores.append(ret)
else:
scores.append(-ret)
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,
}
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
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)
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
