Carry Trade คืออะไรและทำงานอย่างไร
Carry Trade เป็นกลยุทธ์การลงทุนที่ยืมเงินในสกุลที่มีอัตราดอกเบี้ยต่ำ แล้วนำไปลงทุนในสกุลที่มีอัตราดอกเบี้ยสูงกว่า กำไรมาจากส่วนต่างของดอกเบี้ย (interest rate differential) เช่น ยืม Japanese Yen (JPY) ที่ดอกเบี้ย 0.1% ไปลงทุนใน Australian Dollar (AUD) ที่ดอกเบี้ย 4.35% ได้ส่วนต่าง ~4.25% ต่อปี
องค์ประกอบของ Carry Trade ได้แก่ Funding Currency สกุลเงินที่ยืม (ดอกเบี้ยต่ำ เช่น JPY, CHF, EUR), Target Currency สกุลเงินที่ลงทุน (ดอกเบี้ยสูง เช่น AUD, NZD, MXN, BRL), Carry ส่วนต่างดอกเบี้ยที่ได้รับ และ Exchange Rate Risk ความเสี่ยงจากการเปลี่ยนแปลงอัตราแลกเปลี่ยน
Carry Trade ทำกำไรได้ดีเมื่อ interest rate differentials กว้าง, exchange rates เสถียรหรือ target currency แข็งค่า, volatility ต่ำ (low VIX) และ global risk sentiment เป็นบวก (risk-on environment) แต่มีความเสี่ยงสูงเมื่อ exchange rates เปลี่ยนแปลงรุนแรง (carry trade unwinding) ซึ่งมักเกิดในช่วง financial crisis
ตัวอย่างที่มีชื่อเสียงคือ JPY Carry Trade ในช่วง 2004-2007 นักลงทุนยืม JPY ราคาถูกไปลงทุนทั่วโลก เมื่อ Lehman Crisis 2008 เกิด carry trade unwinding ทำให้ JPY แข็งค่าอย่างรวดเร็ว 30%+ ภายในไม่กี่เดือน ทำให้นักลงทุนขาดทุนมหาศาล
วิเคราะห์ Interest Rate Differentials
เครื่องมือวิเคราะห์อัตราดอกเบี้ย
#!/usr/bin/env python3
# interest_rate_analysis.py — Interest Rate Differential Analysis
import json
import logging
from datetime import datetime
from typing import Dict, List, Tuple
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("carry_trade")
# Central Bank Interest Rates (as of analysis date)
INTEREST_RATES = {
"JPY": {"rate": 0.10, "bank": "Bank of Japan", "trend": "rising"},
"CHF": {"rate": 1.50, "bank": "Swiss National Bank", "trend": "stable"},
"EUR": {"rate": 4.50, "bank": "European Central Bank", "trend": "stable"},
"USD": {"rate": 5.50, "bank": "Federal Reserve", "trend": "stable"},
"GBP": {"rate": 5.25, "bank": "Bank of England", "trend": "stable"},
"AUD": {"rate": 4.35, "bank": "Reserve Bank of Australia", "trend": "stable"},
"NZD": {"rate": 5.50, "bank": "Reserve Bank of NZ", "trend": "stable"},
"CAD": {"rate": 5.00, "bank": "Bank of Canada", "trend": "falling"},
"MXN": {"rate": 11.25, "bank": "Banxico", "trend": "stable"},
"BRL": {"rate": 10.50, "bank": "BCB", "trend": "falling"},
"TRY": {"rate": 50.00, "bank": "CBRT", "trend": "stable"},
"ZAR": {"rate": 8.25, "bank": "SARB", "trend": "stable"},
"THB": {"rate": 2.50, "bank": "Bank of Thailand", "trend": "stable"},
}
class CarryTradeAnalyzer:
def __init__(self, rates=None):
self.rates = rates or INTEREST_RATES
def get_differential(self, funding: str, target: str) -> float:
if funding not in self.rates or target not in self.rates:
raise ValueError(f"Unknown currency: {funding} or {target}")
return self.rates[target]["rate"] - self.rates[funding]["rate"]
def find_best_pairs(self, min_differential: float = 2.0) -> List[Dict]:
pairs = []
currencies = list(self.rates.keys())
for funding in currencies:
for target in currencies:
if funding == target:
continue
diff = self.get_differential(funding, target)
if diff >= min_differential:
pairs.append({
"funding": funding,
"target": target,
"pair": f"{target}/{funding}",
"differential_pct": round(diff, 2),
"funding_rate": self.rates[funding]["rate"],
"target_rate": self.rates[target]["rate"],
"funding_trend": self.rates[funding]["trend"],
"target_trend": self.rates[target]["trend"],
})
pairs.sort(key=lambda x: x["differential_pct"], reverse=True)
return pairs
def calculate_carry_return(self, funding, target, amount,
holding_days=365, leverage=1):
diff = self.get_differential(funding, target)
daily_carry = diff / 365
total_carry_pct = daily_carry * holding_days
notional = amount * leverage
gross_return = notional * (total_carry_pct / 100)
# Estimate transaction costs
spread_cost = notional * 0.001 # 10 pips
swap_cost = notional * 0.0002 * holding_days # daily swap
net_return = gross_return - spread_cost - swap_cost
return {
"funding": funding,
"target": target,
"amount": amount,
"leverage": leverage,
"notional": notional,
"differential_pct": round(diff, 2),
"holding_days": holding_days,
"gross_return": round(gross_return, 2),
"spread_cost": round(spread_cost, 2),
"swap_cost": round(swap_cost, 2),
"net_return": round(net_return, 2),
"roi_pct": round(net_return / amount * 100, 2),
}
def risk_assessment(self, funding, target):
diff = self.get_differential(funding, target)
# Risk factors
high_yield_risk = self.rates[target]["rate"] > 10
trend_divergence = (
self.rates[funding]["trend"] == "rising" and
self.rates[target]["trend"] == "falling"
)
risk_score = 0
risks = []
if high_yield_risk:
risk_score += 3
risks.append("High yield currency (>10%) carries devaluation risk")
if trend_divergence:
risk_score += 2
risks.append("Interest rate trends diverging (narrowing differential)")
if diff > 8:
risk_score += 2
risks.append("Very high differential often indicates instability")
if self.rates[target]["rate"] > 20:
risk_score += 3
risks.append("Extreme interest rates indicate economic distress")
risk_level = "low" if risk_score <= 2 else "medium" if risk_score <= 4 else "high"
return {
"pair": f"{target}/{funding}",
"risk_level": risk_level,
"risk_score": risk_score,
"risks": risks,
"differential": round(diff, 2),
}
analyzer = CarryTradeAnalyzer()
# Find best pairs
best = analyzer.find_best_pairs(min_differential=3.0)
print("=== Top Carry Trade Pairs ===")
for p in best[:10]:
print(f" {p['pair']}: {p['differential_pct']}% ({p['funding_rate']}% -> {p['target_rate']}%)")
# Calculate returns
ret = analyzer.calculate_carry_return("JPY", "MXN", 100000, holding_days=365, leverage=3)
print(f"\nReturn: {ret['net_return']:,.0f} ({ret['roi_pct']}% ROI)")
# Risk assessment
risk = analyzer.risk_assessment("JPY", "TRY")
print(f"\nRisk: {risk['risk_level']} (score: {risk['risk_score']})")
for r in risk["risks"]:
print(f" - {r}")สร้าง Carry Trade Analysis Tool
เครื่องมือวิเคราะห์ carry trade แบบครบวงจร
#!/usr/bin/env python3
# carry_trade_tool.py — Carry Trade Analysis Dashboard
import json
import logging
from datetime import datetime, timedelta
from typing import Dict, List
from dataclasses import dataclass, field
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("carry_tool")
@dataclass
class Position:
pair: str
funding_currency: str
target_currency: str
amount: float
leverage: float
entry_rate: float
entry_date: str
differential_pct: float
class CarryTradePortfolio:
def __init__(self):
self.positions: List[Position] = []
self.history = []
def add_position(self, pair, funding, target, amount,
leverage, entry_rate, differential):
position = Position(
pair=pair,
funding_currency=funding,
target_currency=target,
amount=amount,
leverage=leverage,
entry_rate=entry_rate,
entry_date=datetime.utcnow().isoformat(),
differential_pct=differential,
)
self.positions.append(position)
logger.info(f"Added position: {pair} {amount:,.0f} @ {entry_rate} (x{leverage})")
return position
def calculate_pnl(self, current_rates: Dict[str, float]):
results = []
total_pnl = 0
for pos in self.positions:
current_rate = current_rates.get(pos.pair, pos.entry_rate)
# Exchange rate P&L
rate_change_pct = (current_rate - pos.entry_rate) / pos.entry_rate * 100
notional = pos.amount * pos.leverage
fx_pnl = notional * (rate_change_pct / 100)
# Carry P&L (simplified daily)
entry_dt = datetime.fromisoformat(pos.entry_date)
days_held = max((datetime.utcnow() - entry_dt).days, 1)
carry_pnl = notional * (pos.differential_pct / 365 * days_held / 100)
total_position_pnl = fx_pnl + carry_pnl
total_pnl += total_position_pnl
results.append({
"pair": pos.pair,
"notional": round(notional),
"days_held": days_held,
"entry_rate": pos.entry_rate,
"current_rate": current_rate,
"fx_pnl": round(fx_pnl, 2),
"carry_pnl": round(carry_pnl, 2),
"total_pnl": round(total_position_pnl, 2),
"roi_pct": round(total_position_pnl / pos.amount * 100, 2),
})
return {
"positions": results,
"total_pnl": round(total_pnl, 2),
"position_count": len(self.positions),
"timestamp": datetime.utcnow().isoformat(),
}
def risk_metrics(self, current_rates: Dict[str, float]):
total_notional = sum(p.amount * p.leverage for p in self.positions)
total_equity = sum(p.amount for p in self.positions)
# Maximum adverse move calculation
worst_case_moves = {
"JPY": 0.15, # 15% appreciation possible
"CHF": 0.10,
"USD": 0.08,
}
max_loss = 0
for pos in self.positions:
adverse_move = worst_case_moves.get(pos.funding_currency, 0.10)
position_loss = pos.amount * pos.leverage * adverse_move
max_loss += position_loss
return {
"total_equity": round(total_equity),
"total_notional": round(total_notional),
"effective_leverage": round(total_notional / max(total_equity, 1), 1),
"max_drawdown_estimate": round(max_loss),
"max_drawdown_pct": round(max_loss / max(total_equity, 1) * 100, 1),
"weighted_carry": round(
sum(p.differential_pct * p.amount * p.leverage for p in self.positions) /
max(total_notional, 1), 2
),
}
portfolio = CarryTradePortfolio()
portfolio.add_position("MXN/JPY", "JPY", "MXN", 50000, 3, 8.50, 11.15)
portfolio.add_position("AUD/JPY", "JPY", "AUD", 30000, 2, 97.50, 4.25)
portfolio.add_position("NZD/CHF", "CHF", "NZD", 20000, 2, 0.5350, 4.00)
pnl = portfolio.calculate_pnl({"MXN/JPY": 8.55, "AUD/JPY": 98.00, "NZD/CHF": 0.5380})
print(json.dumps(pnl, indent=2))
risk = portfolio.risk_metrics({"MXN/JPY": 8.55, "AUD/JPY": 98.00, "NZD/CHF": 0.5380})
print(json.dumps(risk, indent=2))Risk Management สำหรับ Carry Trade
จัดการความเสี่ยงของ carry trade
#!/usr/bin/env python3
# risk_management.py — Carry Trade Risk Management
import json
import math
import logging
from typing import Dict, List
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("risk")
class CarryTradeRiskManager:
def __init__(self, max_portfolio_risk_pct=10, max_single_position_pct=30):
self.max_portfolio_risk = max_portfolio_risk_pct
self.max_single_position = max_single_position_pct
def calculate_position_size(self, equity, risk_per_trade_pct,
stop_loss_pips, pip_value):
risk_amount = equity * (risk_per_trade_pct / 100)
position_size = risk_amount / (stop_loss_pips * pip_value)
return round(position_size)
def calculate_stop_loss(self, entry_rate, risk_pct, direction="long"):
if direction == "long":
stop = entry_rate * (1 - risk_pct / 100)
else:
stop = entry_rate * (1 + risk_pct / 100)
return round(stop, 5)
def calculate_var(self, notional, daily_volatility_pct,
confidence=0.95, days=1):
z_scores = {0.90: 1.282, 0.95: 1.645, 0.99: 2.326}
z = z_scores.get(confidence, 1.645)
var = notional * (daily_volatility_pct / 100) * z * math.sqrt(days)
return round(var, 2)
def portfolio_check(self, equity, positions):
total_notional = sum(p.get("notional", 0) for p in positions)
total_leverage = total_notional / max(equity, 1)
checks = {
"max_leverage": {
"limit": 5,
"actual": round(total_leverage, 1),
"passed": total_leverage <= 5,
},
"single_position_limit": {
"limit": f"{self.max_single_position}%",
"passed": True,
},
"correlation_risk": {
"warning": False,
"message": "",
},
}
# Check single position concentration
for p in positions:
pct = p.get("notional", 0) / max(total_notional, 1) * 100
if pct > self.max_single_position:
checks["single_position_limit"]["passed"] = False
checks["single_position_limit"]["actual"] = f"{p['pair']}: {pct:.0f}%"
# Check correlation risk (JPY funding concentration)
jpy_funded = sum(
p.get("notional", 0) for p in positions
if p.get("funding") == "JPY"
)
jpy_pct = jpy_funded / max(total_notional, 1) * 100
if jpy_pct > 70:
checks["correlation_risk"]["warning"] = True
checks["correlation_risk"]["message"] = (
f"JPY funding concentration: {jpy_pct:.0f}% — "
"JPY strengthening would impact all positions"
)
all_passed = all(
c.get("passed", True)
for c in checks.values()
if isinstance(c.get("passed"), bool)
)
return {
"equity": equity,
"total_notional": round(total_notional),
"leverage": round(total_leverage, 1),
"checks": checks,
"overall": "PASS" if all_passed else "FAIL",
}
def generate_risk_report(self, equity, positions):
report = {
"timestamp": __import__("datetime").datetime.utcnow().isoformat(),
"portfolio_check": self.portfolio_check(equity, positions),
"position_vars": [],
}
volatilities = {
"MXN/JPY": 12.0, "AUD/JPY": 10.5, "NZD/JPY": 11.0,
"USD/JPY": 8.0, "TRY/JPY": 25.0, "ZAR/JPY": 15.0,
}
total_var = 0
for p in positions:
vol = volatilities.get(p["pair"], 10.0)
var_95 = self.calculate_var(p["notional"], vol / math.sqrt(252), 0.95, 1)
total_var += var_95
report["position_vars"].append({
"pair": p["pair"],
"notional": p["notional"],
"daily_var_95": var_95,
"annual_volatility": vol,
})
report["total_daily_var_95"] = round(total_var)
report["var_as_pct_equity"] = round(total_var / max(equity, 1) * 100, 2)
return report
rm = CarryTradeRiskManager()
positions = [
{"pair": "MXN/JPY", "funding": "JPY", "notional": 150000},
{"pair": "AUD/JPY", "funding": "JPY", "notional": 60000},
{"pair": "NZD/CHF", "funding": "CHF", "notional": 40000},
]
report = rm.generate_risk_report(100000, positions)
print(json.dumps(report, indent=2))Backtesting Carry Trade Strategies
ทดสอบกลยุทธ์ย้อนหลัง
#!/usr/bin/env python3
# backtest.py — Carry Trade Backtesting Framework
import json
import random
import math
import logging
from datetime import datetime, timedelta
from typing import List, Dict
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("backtest")
class CarryTradeBacktest:
def __init__(self, initial_equity=100000):
self.initial_equity = initial_equity
self.equity = initial_equity
self.trades = []
self.equity_curve = [initial_equity]
def simulate_price_series(self, start_price, days,
daily_drift=0.0, daily_vol=0.005):
prices = [start_price]
for _ in range(days):
change = random.gauss(daily_drift, daily_vol)
prices.append(prices[-1] * (1 + change))
return prices
def run_backtest(self, pair, funding_rate, target_rate,
start_price, days=365, leverage=2,
stop_loss_pct=5, take_profit_pct=15,
daily_vol=0.005):
logger.info(f"Backtesting {pair} for {days} days (x{leverage})")
prices = self.simulate_price_series(
start_price, days,
daily_drift=0.00005, # slight upward bias
daily_vol=daily_vol,
)
daily_carry = (target_rate - funding_rate) / 365 / 100
position_size = self.equity * leverage
cumulative_carry = 0
peak_equity = self.equity
max_drawdown = 0
for day in range(1, len(prices)):
# Daily carry income
carry_income = position_size * daily_carry
cumulative_carry += carry_income
# Exchange rate P&L
rate_change = (prices[day] - prices[day - 1]) / prices[day - 1]
fx_pnl = position_size * rate_change
# Update equity
self.equity += carry_income + fx_pnl
self.equity_curve.append(round(self.equity, 2))
# Track drawdown
if self.equity > peak_equity:
peak_equity = self.equity
drawdown = (peak_equity - self.equity) / peak_equity * 100
max_drawdown = max(max_drawdown, drawdown)
# Check stop loss / take profit
total_return_pct = (self.equity - self.initial_equity) / self.initial_equity * 100
if total_return_pct <= -stop_loss_pct:
self.trades.append({
"day": day, "exit": "stop_loss",
"return_pct": round(total_return_pct, 2),
})
break
if total_return_pct >= take_profit_pct:
self.trades.append({
"day": day, "exit": "take_profit",
"return_pct": round(total_return_pct, 2),
})
break
# Results
total_return = self.equity - self.initial_equity
total_return_pct = total_return / self.initial_equity * 100
# Sharpe ratio (simplified)
daily_returns = []
for i in range(1, len(self.equity_curve)):
daily_returns.append(
(self.equity_curve[i] - self.equity_curve[i-1]) / self.equity_curve[i-1]
)
avg_daily = sum(daily_returns) / max(len(daily_returns), 1)
std_daily = math.sqrt(
sum((r - avg_daily) ** 2 for r in daily_returns) / max(len(daily_returns) - 1, 1)
)
sharpe = (avg_daily / max(std_daily, 0.0001)) * math.sqrt(252)
results = {
"pair": pair,
"days": len(self.equity_curve) - 1,
"initial_equity": self.initial_equity,
"final_equity": round(self.equity, 2),
"total_return": round(total_return, 2),
"total_return_pct": round(total_return_pct, 2),
"cumulative_carry": round(cumulative_carry, 2),
"max_drawdown_pct": round(max_drawdown, 2),
"sharpe_ratio": round(sharpe, 2),
"trades": self.trades,
}
logger.info(f"Result: {total_return_pct:.1f}% return, {max_drawdown:.1f}% max DD, Sharpe {sharpe:.2f}")
return results
# Run backtest
bt = CarryTradeBacktest(initial_equity=100000)
results = bt.run_backtest(
pair="AUD/JPY",
funding_rate=0.10,
target_rate=4.35,
start_price=97.50,
days=365,
leverage=2,
stop_loss_pct=8,
take_profit_pct=20,
daily_vol=0.006,
)
print(json.dumps(results, indent=2))Automated Monitoring และ Alerts
Monitor carry trade conditions อัตโนมัติ
# === Carry Trade Monitoring Setup ===
# 1. Rate Monitoring Cron Job
# ===================================
# crontab -e
# 0 */4 * * * /opt/carry_trade/check_rates.py >> /var/log/carry_monitor.log 2>&1
# 0 8 * * 1 /opt/carry_trade/weekly_report.py >> /var/log/carry_report.log 2>&1
# 2. Alert Conditions
# ===================================
# - Interest rate change announcement
# - Exchange rate moves > 2% in single day
# - VIX spike above 25 (risk-off signal)
# - Central bank meeting dates approaching
# - Carry differential narrowing below threshold
# 3. Prometheus Metrics
# ===================================
# carry_trade_differential{pair="AUD/JPY"} 4.25
# carry_trade_position_pnl{pair="AUD/JPY"} 1250.50
# carry_trade_portfolio_equity 102500
# carry_trade_max_drawdown_pct 3.2
# carry_trade_vix_level 18.5
# 4. Grafana Dashboard Queries
# ===================================
# Interest Rate Differentials over time
# carry_trade_differential
# Portfolio P&L
# carry_trade_portfolio_equity - 100000
# Risk metrics
# carry_trade_max_drawdown_pct
# 5. Alert Rules
# ===================================
# alerting_rules.yml
# groups:
# - name: carry_trade
# rules:
# - alert: LargeFXMove
# expr: abs(rate(fx_rate_change[1d])) > 0.02
# labels:
# severity: warning
# annotations:
# summary: "{{ $labels.pair }} moved >2% today"
#
# - alert: DifferentialNarrowing
# expr: carry_trade_differential < 2.0
# for: 1h
# labels:
# severity: critical
# annotations:
# summary: "{{ $labels.pair }} carry below 2%"
#
# - alert: DrawdownExceeded
# expr: carry_trade_max_drawdown_pct > 8
# labels:
# severity: critical
# annotations:
# summary: "Portfolio drawdown exceeded 8%"
#
# - alert: VIXSpike
# expr: carry_trade_vix_level > 25
# labels:
# severity: warning
# annotations:
# summary: "VIX above 25 - risk-off signal"
# 6. Economic Calendar Integration
# ===================================
# Key events to monitor:
# - FOMC meetings (8x/year)
# - BOJ meetings (8x/year)
# - ECB meetings (8x/year)
# - NFP (monthly)
# - CPI releases (monthly per country)
# - GDP releases (quarterly)
# Script to check upcoming events
# import requests
# resp = requests.get("https://api.tradingeconomics.com/calendar",
# params={"c": "api_key"})
# events = resp.json()
# for e in events:
# if e["importance"] == 3: # high importance
# print(f"{e['date']}: {e['country']} - {e['event']}")
echo "Carry trade monitoring configured"FAQ คำถามที่พบบ่อย
Q: Carry trade เหมาะกับนักลงทุนทุกู้คืนไหม?
A: ไม่เหมาะกับทุกู้คืน Carry trade ต้องการ ความเข้าใจ forex market และ interest rate dynamics, capital เพียงพอสำหรับรองรับ drawdowns (แนะนำเริ่มต้นขั้นต่ำ $10,000), ความอดทน เพราะ carry สะสมช้า (daily basis), การจัดการความเสี่ยงที่ดี (stop loss, position sizing) และความสามารถ monitor positions สม่ำเสมอ เหมาะสำหรับ intermediate-advanced traders ที่มี holding period ระยะกลาง-ยาว (weeks to months)
Q: Carry trade unwinding คืออะไร?
A: เมื่อนักลงทุนจำนวนมากปิด carry trade positions พร้อมกัน (ขาย target currency ซื้อ funding currency คืน) ทำให้ funding currency แข็งค่าอย่างรวดเร็ว มักเกิดเมื่อ risk-off events เช่น financial crisis, geopolitical tensions, unexpected rate changes ตัวอย่าง JPY carry unwind ปี 2008 JPY แข็งค่า 30%+ ใน 3 เดือน ทำให้ positions ที่กำไรจาก carry กลายเป็นขาดทุนหนักจาก FX loss
Q: ควรใช้ leverage เท่าไหร?
A: สำหรับ carry trade แนะนำ leverage ไม่เกิน 3x สำหรับ G10 currencies (major pairs) และไม่เกิน 2x สำหรับ emerging market currencies leverage สูงเพิ่มทั้ง carry income และ FX risk หลักคือ daily carry income ต้องมากพอชดเชย potential FX losses ใช้ position sizing ตาม risk per trade (1-2% ของ equity) ไม่ใช่ maximize leverage
Q: สัญญาณอะไรบอกว่าควรปิด carry trade?
A: สัญญาณที่ควรพิจารณาปิด ได้แก่ central bank ของ funding currency เริ่มขึ้นดอกเบี้ย (differential ลดลง), VIX พุ่งสูงกว่า 25-30 (risk-off), target currency เริ่มอ่อนค่าเร็ว (trend reversal), economic crisis ใน target country, carry differential ลดลงต่ำกว่า 2% (ไม่คุ้มความเสี่ยง) และ political instability ที่กระทบ monetary policy
