| Overall Statistics |
|
Total Orders 301 Average Win 1.44% Average Loss -0.73% Compounding Annual Return 0.289% Drawdown 12.700% Expectancy 0.103 Start Equity 25000 End Equity 26359.92 Net Profit 5.440% Sharpe Ratio -0.409 Sortino Ratio -0.497 Probabilistic Sharpe Ratio 0.000% Loss Rate 63% Win Rate 37% Profit-Loss Ratio 1.98 Alpha -0.012 Beta -0.067 Annual Standard Deviation 0.041 Annual Variance 0.002 Information Ratio -0.489 Tracking Error 0.178 Treynor Ratio 0.25 Total Fees $0.00 Estimated Strategy Capacity $1900000.00 Lowest Capacity Asset USDZAR 8G Portfolio Turnover 1.08% Drawdown Recovery 1348 |
# region imports
from AlgorithmImports import *
# endregion
"""
FX CARRY + MOMENTUM + RATE-CHANGE STRATEGY
This algorithm implements a more dynamic foreign-exchange carry model while
preserving the original working structure:
FX symbol string -> Nasdaq Data Link rate symbol
The model 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 original pure carry model ranked pairs only by the level of the linked rate.
That can result in very low turnover because central-bank-rate series are often
slow-moving.
This version uses a composite score with three components:
1. Carry level:
The current value of the linked interest-rate proxy.
Higher rate = higher score.
2. FX momentum:
The recent price momentum of the FX pair.
Higher momentum = higher score.
3. Rate change:
The change in the linked rate proxy over a lookback window.
Rising rates = higher score.
The composite score is:
composite score =
carry_weight * carry rank
+ momentum_weight * momentum rank
+ rate_change_weight * rate-change rank
The strategy rebalances monthly.
Portfolio construction:
- Rank all FX pairs by composite score.
- Go long the top-ranked group.
- Go short the bottom-ranked group.
- Set all other pairs to zero.
- Use equal weights across selected long and short positions.
- Cap total gross exposure.
- Avoid trading if the score gap between the strongest and weakest names is too
small.
- Avoid unnecessary orders if the target weight change is too small.
The model intentionally has no OnData method. All calculations, logs, plots, and
trades happen only at the monthly rebalance. This keeps the algorithm simple and
avoids excessive plotting or repeated order activity.
Important data note:
The Nasdaq Data Link / BCB series are used as rate proxies. They may not perfectly
represent tradable FX carry. They can be stale, stepwise, or not fully comparable
across countries. For that reason, the algorithm logs the complete monthly
ranking so the user can audit the selected currencies.
"""
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. 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 safer History calls.
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
# ------------------------------------------------------------
# 3. STRATEGY PARAMETERS
# ------------------------------------------------------------
self.top_count = 2
self.bottom_count = 2
self.gross_exposure = 1.00
self.carry_weight = 0.50
self.momentum_weight = 0.35
self.rate_change_weight = 0.15
self.momentum_lookback_days = 126
self.rate_change_lookback_days = 126
# If the best score and worst score are too close, the model holds cash.
self.minimum_score_gap = 1.00
# Avoid submitting orders for very small changes.
self.minimum_weight_change = 0.05
# ------------------------------------------------------------
# 4. DIAGNOSTIC STATE
# ------------------------------------------------------------
self.previous_targets = {}
self.rebalance_count = 0
for symbol in self.symbols.keys():
self.previous_targets[symbol] = 0.0
# ------------------------------------------------------------
# 5. MONTHLY REBALANCE ONLY
# ------------------------------------------------------------
self.Schedule.On(
self.DateRules.MonthStart("USDEUR"),
self.TimeRules.BeforeMarketClose("USDEUR"),
self.Rebalance
)
def Rebalance(self):
# ------------------------------------------------------------
# 1. 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
)
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,
"rate_change": rate_change
}
)
minimum_required = self.top_count + self.bottom_count
if len(data_table) < minimum_required:
return
# ------------------------------------------------------------
# 2. CREATE RANKS
# ------------------------------------------------------------
# Higher rate is better.
carry_sorted = sorted(
data_table,
key=lambda x: x["rate"]
)
for rank, item in enumerate(carry_sorted):
item["carry_rank"] = rank
# Higher FX momentum is better.
momentum_sorted = sorted(
data_table,
key=lambda x: x["momentum"]
)
for rank, item in enumerate(momentum_sorted):
item["momentum_rank"] = rank
# More positive rate change is better.
rate_change_sorted = sorted(
data_table,
key=lambda x: x["rate_change"]
)
for rank, item in enumerate(rate_change_sorted):
item["rate_change_rank"] = rank
# ------------------------------------------------------------
# 3. 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
# ------------------------------------------------------------
# 4. BUILD TARGET WEIGHTS
# ------------------------------------------------------------
target_weights = {}
for item in data_table:
target_weights[item["fx_symbol"]] = 0.0
if score_gap >= self.minimum_score_gap:
number_of_positions = self.top_count + self.bottom_count
leg_weight = self.gross_exposure / number_of_positions
# Short the weakest composite scores.
for item in bottom_group:
target_weights[item["fx_symbol"]] = -leg_weight
# Long the strongest composite scores.
for item in top_group:
target_weights[item["fx_symbol"]] = leg_weight
else:
self.Debug(
"No trade on "
+ str(self.Time.date())
+ " because score gap is too small: "
+ str(round(score_gap, 4))
)
# ------------------------------------------------------------
# 5. APPLY TARGETS ONLY WHEN CHANGE IS MEANINGFUL
# ------------------------------------------------------------
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
# ------------------------------------------------------------
# 6. TURNOVER CALCULATION
# ------------------------------------------------------------
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
self.rebalance_count += 1
# ------------------------------------------------------------
# 7. LOG FULL MONTHLY RANKING
# ------------------------------------------------------------
self.Debug("========== MONTHLY FX CARRY COMPOSITE RANKING ==========")
self.Debug("Date: " + str(self.Time.date()))
for item in ranked:
self.Debug(
str(item["fx_symbol"])
+ " | Rate="
+ str(round(item["rate"], 6))
+ " | FX Momentum="
+ str(round(item["momentum"], 6))
+ " | Rate Change="
+ str(round(item["rate_change"], 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))
)
# ------------------------------------------------------------
# 8. MONTHLY PLOTS ONLY
# ------------------------------------------------------------
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",
"Score Gap",
score_gap
)
self.Plot(
"Carry Diagnostics",
"Highest Composite Score",
highest_score
)
self.Plot(
"Carry Diagnostics",
"Lowest Composite Score",
lowest_score
)
self.Plot(
"Carry Selection",
"Highest Rate",
max(x["rate"] for x in data_table)
)
self.Plot(
"Carry Selection",
"Lowest Rate",
min(x["rate"] for x in data_table)
)
self.Plot(
"Carry Selection",
"Rate Spread",
max(x["rate"] for x in data_table)
- min(x["rate"] for x in data_table)
)
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)
)
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
)
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 GetRateChange(self, rate_symbol, lookback_days):
values = self.GetHistoryCloseValues(
rate_symbol,
lookback_days
)
if len(values) < 2:
return 0.0
start_value = values[0]
end_value = values[-1]
return end_value - start_value
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