| Overall Statistics |
|
Total Orders 398 Average Win 1.07% Average Loss -0.53% Compounding Annual Return 0.175% Drawdown 13.300% Expectancy 0.077 Start Equity 25000 End Equity 25814.19 Net Profit 3.257% Sharpe Ratio -0.44 Sortino Ratio -0.525 Probabilistic Sharpe Ratio 0.000% Loss Rate 64% Win Rate 36% Profit-Loss Ratio 2.03 Alpha -0.013 Beta -0.059 Annual Standard Deviation 0.04 Annual Variance 0.002 Information Ratio -0.497 Tracking Error 0.177 Treynor Ratio 0.297 Total Fees $0.00 Estimated Strategy Capacity $1800000.00 Lowest Capacity Asset USDZAR 8G Portfolio Turnover 1.30% Drawdown Recovery 1367 |
# region imports
from AlgorithmImports import *
# endregion
import numpy as np
"""
ROBUST PARAMETERIZED FX CARRY MODEL
This algorithm implements a monthly FX carry strategy that can be calibrated from
the QuantConnect parameter panel.
The strategy trades a universe of USD-based FX pairs:
USDEUR, USDZAR, USDAUD, USDJPY, USDTRY,
USDINR, USDCNY, USDMXN, USDCAD
Each FX pair is linked to an external interest-rate proxy from Nasdaq Data Link.
The model ranks each FX pair using a composite score built from three components:
1. Carry level
The current value of the linked interest-rate proxy.
Higher rate proxies receive better carry ranks.
2. FX momentum
The recent price momentum of the FX pair.
Stronger FX momentum receives a better momentum rank.
3. Rate change
The recent change in the linked interest-rate proxy.
Rising rate proxies receive a better rate-change rank.
The composite score is:
score =
carry_weight * carry_rank
+ momentum_weight * momentum_rank
+ rate_change_weight * rate_change_rank
The strategy rebalances once per month.
Portfolio construction:
- Long the top-ranked FX pairs.
- Short the bottom-ranked FX pairs.
- Hold no position in middle-ranked pairs.
- Use inverse-volatility sizing inside the long and short baskets.
- Cap the maximum weight of any single FX pair.
- Cap total gross exposure.
- Avoid trading when the score gap is too small.
- Avoid small unnecessary order updates.
Important implementation point:
There is no OnData method. The model only calculates, logs, plots, and trades
at the monthly rebalance. This keeps the algorithm stable and close to the
original working carry example.
Parameter examples for QuantConnect:
top_count = 2
bottom_count = 2
target_gross_exposure = 1.00
max_single_pair_weight = 0.35
carry_weight = 0.50
momentum_weight = 0.35
rate_change_weight = 0.15
momentum_lookback_days = 126
volatility_lookback_days = 63
rate_change_lookback_days = 126
minimum_score_gap = 1.00
minimum_weight_change = 0.05
max_drawdown_from_peak = 0.30
"""
class ForexCarryTradeAlgorithm(QCAlgorithm):
def Initialize(self):
# ------------------------------------------------------------
# 1. BACKTEST SETTINGS
# ------------------------------------------------------------
self.SetStartDate(2008, 1, 1)
self.SetEndDate(2026, 5, 5)
self.initial_cash = 25000
self.SetCash(self.initial_cash)
# ------------------------------------------------------------
# 2. CALIBRATABLE PARAMETERS
# ------------------------------------------------------------
self.top_count = self.GetIntParameter("top_count", 2)
self.bottom_count = self.GetIntParameter("bottom_count", 2)
self.target_gross_exposure = self.GetFloatParameter(
"target_gross_exposure",
1.00
)
self.max_single_pair_weight = self.GetFloatParameter(
"max_single_pair_weight",
0.35
)
self.carry_weight = self.GetFloatParameter(
"carry_weight",
0.50
)
self.momentum_weight = self.GetFloatParameter(
"momentum_weight",
0.35
)
self.rate_change_weight = self.GetFloatParameter(
"rate_change_weight",
0.15
)
self.momentum_lookback_days = self.GetIntParameter(
"momentum_lookback_days",
126
)
self.volatility_lookback_days = self.GetIntParameter(
"volatility_lookback_days",
63
)
self.rate_change_lookback_days = self.GetIntParameter(
"rate_change_lookback_days",
126
)
self.minimum_score_gap = self.GetFloatParameter(
"minimum_score_gap",
1.00
)
self.minimum_weight_change = self.GetFloatParameter(
"minimum_weight_change",
0.05
)
self.max_drawdown_from_peak = self.GetFloatParameter(
"max_drawdown_from_peak",
0.30
)
# ------------------------------------------------------------
# 3. PARAMETER SAFETY CHECKS
# ------------------------------------------------------------
if self.top_count < 1:
self.top_count = 1
if self.bottom_count < 1:
self.bottom_count = 1
if self.target_gross_exposure < 0:
self.target_gross_exposure = 0.0
if self.max_single_pair_weight <= 0:
self.max_single_pair_weight = 0.10
if self.minimum_weight_change < 0:
self.minimum_weight_change = 0.0
# ------------------------------------------------------------
# 4. FX PAIRS AND RATE DATA
# ------------------------------------------------------------
self.rate_symbol_by_ticker = {
"USDEUR": "BCB/17900", # Euro Area
"USDZAR": "BCB/17906", # South Africa
"USDAUD": "BCB/17880", # Australia
"USDJPY": "BCB/17903", # Japan
"USDTRY": "BCB/17907", # Turkey
"USDINR": "BCB/17901", # India
"USDCNY": "BCB/17899", # China
"USDMXN": "BCB/17904", # Mexico
"USDCAD": "BCB/17881" # Canada
}
# Original working structure:
# string FX symbol -> rate data symbol
self.symbols = {}
# Additional map for robust History calls:
# string FX symbol -> QuantConnect Symbol object
self.fx_symbol_objects = {}
for ticker, rate_symbol in self.rate_symbol_by_ticker.items():
forex_symbol = self.AddForex(
ticker,
Resolution.Daily,
Market.Oanda
).Symbol
data_symbol = self.AddData(
NasdaqDataLink,
rate_symbol,
Resolution.Daily,
TimeZones.Utc,
True
).Symbol
self.symbols[str(forex_symbol)] = data_symbol
self.fx_symbol_objects[str(forex_symbol)] = forex_symbol
# ------------------------------------------------------------
# 5. DIAGNOSTIC STATE
# ------------------------------------------------------------
self.previous_targets = {}
self.rebalance_count = 0
self.portfolio_peak = self.initial_cash
for symbol in self.symbols.keys():
self.previous_targets[symbol] = 0.0
# ------------------------------------------------------------
# 6. MONTHLY REBALANCE ONLY
# ------------------------------------------------------------
self.Schedule.On(
self.DateRules.MonthStart("USDEUR"),
self.TimeRules.BeforeMarketClose("USDEUR"),
self.Rebalance
)
self.Debug(
"Parameters | "
+ "top_count="
+ str(self.top_count)
+ " bottom_count="
+ str(self.bottom_count)
+ " target_gross_exposure="
+ str(self.target_gross_exposure)
+ " max_single_pair_weight="
+ str(self.max_single_pair_weight)
+ " carry_weight="
+ str(self.carry_weight)
+ " momentum_weight="
+ str(self.momentum_weight)
+ " rate_change_weight="
+ str(self.rate_change_weight)
+ " momentum_lookback_days="
+ str(self.momentum_lookback_days)
+ " volatility_lookback_days="
+ str(self.volatility_lookback_days)
+ " rate_change_lookback_days="
+ str(self.rate_change_lookback_days)
+ " minimum_score_gap="
+ str(self.minimum_score_gap)
+ " minimum_weight_change="
+ str(self.minimum_weight_change)
+ " max_drawdown_from_peak="
+ str(self.max_drawdown_from_peak)
)
def Rebalance(self):
# ------------------------------------------------------------
# 1. MONTHLY DRAWDOWN GUARD
# ------------------------------------------------------------
self.portfolio_peak = max(
self.portfolio_peak,
self.Portfolio.TotalPortfolioValue
)
portfolio_drawdown = 0.0
if self.portfolio_peak > 0:
portfolio_drawdown = (
self.Portfolio.TotalPortfolioValue
/ self.portfolio_peak
- 1.0
)
if portfolio_drawdown <= -self.max_drawdown_from_peak:
self.Debug(
"Drawdown guard triggered on "
+ str(self.Time.date())
+ ". Moving to cash. Drawdown="
+ str(round(portfolio_drawdown, 4))
)
for symbol in self.symbols.keys():
self.SetHoldings(symbol, 0.0)
self.previous_targets[symbol] = 0.0
self.portfolio_peak = self.Portfolio.TotalPortfolioValue
self.PlotMonthlyDiagnostics(
data_table=[],
ranked=[],
target_weights=self.previous_targets,
monthly_turnover=0.0,
score_gap=0.0,
portfolio_drawdown=portfolio_drawdown
)
return
# ------------------------------------------------------------
# 2. BUILD DATA TABLE
# ------------------------------------------------------------
data_table = []
for fx_symbol_string, rate_symbol in self.symbols.items():
rate_value = self.Securities[rate_symbol].Price
fx_momentum = self.GetMomentum(
fx_symbol_string,
self.momentum_lookback_days
)
fx_volatility = self.GetVolatility(
fx_symbol_string,
self.volatility_lookback_days
)
rate_change = self.GetRateChange(
rate_symbol,
self.rate_change_lookback_days
)
data_table.append(
{
"fx_symbol": fx_symbol_string,
"rate_symbol": rate_symbol,
"rate": rate_value,
"momentum": fx_momentum,
"volatility": fx_volatility,
"rate_change": rate_change
}
)
minimum_required = self.top_count + self.bottom_count
if len(data_table) < minimum_required:
self.Debug(
"Not enough symbols for ranking on "
+ str(self.Time.date())
)
return
# ------------------------------------------------------------
# 3. RANK SIGNAL COMPONENTS
# ------------------------------------------------------------
self.AssignRank(
data_table,
source_key="rate",
target_key="carry_rank"
)
self.AssignRank(
data_table,
source_key="momentum",
target_key="momentum_rank"
)
self.AssignRank(
data_table,
source_key="rate_change",
target_key="rate_change_rank"
)
# ------------------------------------------------------------
# 4. COMPOSITE SCORE
# ------------------------------------------------------------
for item in data_table:
item["score"] = (
self.carry_weight * item["carry_rank"]
+ self.momentum_weight * item["momentum_rank"]
+ self.rate_change_weight * item["rate_change_rank"]
)
ranked = sorted(
data_table,
key=lambda x: x["score"]
)
bottom_group = ranked[:self.bottom_count]
top_group = ranked[-self.top_count:]
lowest_score = bottom_group[0]["score"]
highest_score = top_group[-1]["score"]
score_gap = highest_score - lowest_score
# ------------------------------------------------------------
# 5. BUILD TARGET WEIGHTS
# ------------------------------------------------------------
target_weights = {}
for item in data_table:
target_weights[item["fx_symbol"]] = 0.0
if score_gap >= self.minimum_score_gap:
long_budget = self.target_gross_exposure / 2.0
short_budget = self.target_gross_exposure / 2.0
long_weights = self.GetInverseVolatilityWeights(
top_group,
long_budget
)
short_weights = self.GetInverseVolatilityWeights(
bottom_group,
short_budget
)
for symbol, weight in long_weights.items():
target_weights[symbol] = weight
for symbol, weight in short_weights.items():
target_weights[symbol] = -weight
else:
self.Debug(
"No trade on "
+ str(self.Time.date())
+ ". Score gap too small: "
+ str(round(score_gap, 4))
)
# ------------------------------------------------------------
# 6. WEIGHT CAPS AND GROSS EXPOSURE CONTROL
# ------------------------------------------------------------
target_weights = self.ApplyWeightCaps(target_weights)
# ------------------------------------------------------------
# 7. MONTHLY TURNOVER BEFORE UPDATING PREVIOUS TARGETS
# ------------------------------------------------------------
monthly_turnover = 0.0
for symbol, target_weight in target_weights.items():
previous_weight = self.previous_targets.get(symbol, 0.0)
monthly_turnover += abs(target_weight - previous_weight)
monthly_turnover = monthly_turnover / 2.0
# ------------------------------------------------------------
# 8. APPLY TARGETS
# ------------------------------------------------------------
for symbol, target_weight in target_weights.items():
previous_weight = self.previous_targets.get(symbol, 0.0)
weight_change = abs(target_weight - previous_weight)
if weight_change >= self.minimum_weight_change:
self.SetHoldings(
symbol,
target_weight
)
self.previous_targets[symbol] = target_weight
self.rebalance_count += 1
# ------------------------------------------------------------
# 9. LOG FULL MONTHLY RANKING
# ------------------------------------------------------------
self.Debug("========== MONTHLY PARAMETERIZED FX CARRY RANKING ==========")
self.Debug("Date: " + str(self.Time.date()))
for item in ranked:
self.Debug(
str(item["fx_symbol"])
+ " | Rate="
+ str(round(item["rate"], 6))
+ " | Momentum="
+ str(round(item["momentum"], 6))
+ " | RateChange="
+ str(round(item["rate_change"], 6))
+ " | Vol="
+ str(round(item["volatility"], 6))
+ " | Score="
+ str(round(item["score"], 4))
)
self.Debug(
"Long group: "
+ str([x["fx_symbol"] for x in top_group])
+ " | Short group: "
+ str([x["fx_symbol"] for x in bottom_group])
+ " | Score gap="
+ str(round(score_gap, 4))
+ " | Turnover="
+ str(round(monthly_turnover, 4))
)
# ------------------------------------------------------------
# 10. MONTHLY PLOTS
# ------------------------------------------------------------
self.PlotMonthlyDiagnostics(
data_table=data_table,
ranked=ranked,
target_weights=target_weights,
monthly_turnover=monthly_turnover,
score_gap=score_gap,
portfolio_drawdown=portfolio_drawdown
)
# ------------------------------------------------------------
# PARAMETER HELPERS
# ------------------------------------------------------------
def GetIntParameter(self, name, default_value):
value = self.GetParameter(name)
if value is None or value == "":
return default_value
return int(value)
def GetFloatParameter(self, name, default_value):
value = self.GetParameter(name)
if value is None or value == "":
return default_value
return float(value)
# ------------------------------------------------------------
# RANKING HELPERS
# ------------------------------------------------------------
def AssignRank(self, data_table, source_key, target_key):
sorted_items = sorted(
data_table,
key=lambda x: x[source_key]
)
for rank, item in enumerate(sorted_items):
item[target_key] = rank
# ------------------------------------------------------------
# PORTFOLIO CONSTRUCTION HELPERS
# ------------------------------------------------------------
def GetInverseVolatilityWeights(self, group, budget):
raw_weights = {}
total_inverse_vol = 0.0
for item in group:
volatility = item["volatility"]
if volatility <= 0:
volatility = 0.0001
inverse_vol = 1.0 / volatility
raw_weights[item["fx_symbol"]] = inverse_vol
total_inverse_vol += inverse_vol
final_weights = {}
if total_inverse_vol <= 0:
equal_weight = budget / len(group)
for item in group:
final_weights[item["fx_symbol"]] = equal_weight
return final_weights
for symbol, inverse_vol in raw_weights.items():
final_weights[symbol] = (
budget
* inverse_vol
/ total_inverse_vol
)
return final_weights
def ApplyWeightCaps(self, target_weights):
capped_weights = {}
for symbol, weight in target_weights.items():
if weight > self.max_single_pair_weight:
weight = self.max_single_pair_weight
if weight < -self.max_single_pair_weight:
weight = -self.max_single_pair_weight
capped_weights[symbol] = weight
gross_target = sum(
abs(weight)
for weight in capped_weights.values()
)
if gross_target > self.target_gross_exposure and gross_target > 0:
scale = self.target_gross_exposure / gross_target
for symbol in capped_weights:
capped_weights[symbol] = capped_weights[symbol] * scale
return capped_weights
# ------------------------------------------------------------
# SIGNAL HELPERS
# ------------------------------------------------------------
def GetMomentum(self, fx_symbol_string, lookback_days):
fx_symbol = self.fx_symbol_objects[fx_symbol_string]
values = self.GetHistoryCloseValues(
fx_symbol,
lookback_days
)
if len(values) < 2:
return 0.0
start_price = values[0]
end_price = values[-1]
if start_price <= 0:
return 0.0
return end_price / start_price - 1.0
def GetVolatility(self, fx_symbol_string, lookback_days):
fx_symbol = self.fx_symbol_objects[fx_symbol_string]
values = self.GetHistoryCloseValues(
fx_symbol,
lookback_days
)
if len(values) < 3:
return 0.01
returns = []
for i in range(1, len(values)):
if values[i - 1] <= 0:
continue
returns.append(
values[i] / values[i - 1] - 1.0
)
if len(returns) < 2:
return 0.01
return float(np.std(returns))
def GetRateChange(self, rate_symbol, lookback_days):
values = self.GetHistoryCloseValues(
rate_symbol,
lookback_days
)
if len(values) < 2:
return 0.0
return values[-1] - values[0]
def GetHistoryCloseValues(self, symbol, lookback_days):
history = self.History(
symbol,
lookback_days,
Resolution.Daily
)
values = []
# Case 1: pandas dataframe
if hasattr(history, "empty"):
if history.empty:
return values
if "close" in history.columns:
for value in history["close"]:
values.append(float(value))
elif "value" in history.columns:
for value in history["value"]:
values.append(float(value))
return values
# Case 2: QC enumerable
for bar in history:
if hasattr(bar, "Close"):
values.append(float(bar.Close))
elif hasattr(bar, "Value"):
values.append(float(bar.Value))
return values
# ------------------------------------------------------------
# DIAGNOSTIC PLOTS
# ------------------------------------------------------------
def PlotMonthlyDiagnostics(
self,
data_table,
ranked,
target_weights,
monthly_turnover,
score_gap,
portfolio_drawdown
):
self.Plot(
"Strategy Equity",
"Portfolio Value",
self.Portfolio.TotalPortfolioValue
)
self.Plot(
"Strategy Equity",
"Cash Benchmark",
self.initial_cash
)
self.Plot(
"Carry Diagnostics",
"Rebalance Count",
self.rebalance_count
)
self.Plot(
"Carry Diagnostics",
"Monthly Turnover",
monthly_turnover
)
self.Plot(
"Carry Diagnostics",
"Score Gap",
score_gap
)
self.Plot(
"Risk Diagnostics",
"Portfolio Drawdown",
portfolio_drawdown
)
self.Plot(
"Risk Diagnostics",
"Drawdown Limit",
-self.max_drawdown_from_peak
)
gross_exposure = sum(
abs(weight)
for weight in target_weights.values()
)
active_positions = sum(
1
for weight in target_weights.values()
if abs(weight) > 0
)
self.Plot(
"Portfolio State",
"Target Gross Exposure",
gross_exposure
)
self.Plot(
"Portfolio State",
"Target Active Positions",
active_positions
)
if len(data_table) > 0:
highest_rate = max(x["rate"] for x in data_table)
lowest_rate = min(x["rate"] for x in data_table)
self.Plot(
"Carry Selection",
"Highest Rate",
highest_rate
)
self.Plot(
"Carry Selection",
"Lowest Rate",
lowest_rate
)
self.Plot(
"Carry Selection",
"Rate Spread",
highest_rate - lowest_rate
)
self.Plot(
"Signal Diagnostics",
"Highest Momentum",
max(x["momentum"] for x in data_table)
)
self.Plot(
"Signal Diagnostics",
"Lowest Momentum",
min(x["momentum"] for x in data_table)
)
self.Plot(
"Signal Diagnostics",
"Highest Rate Change",
max(x["rate_change"] for x in data_table)
)
self.Plot(
"Signal Diagnostics",
"Lowest Rate Change",
min(x["rate_change"] for x in data_table)
)