| Overall Statistics |
|
Total Trades 10 Average Win 0% Average Loss 0.00% Compounding Annual Return 0.000% Drawdown 0.000% Expectancy -1 Net Profit 0.000% Sharpe Ratio -148720.835 Sortino Ratio -11213.437 Probabilistic Sharpe Ratio 0.000% Loss Rate 100% Win Rate 0% Profit-Loss Ratio 0 Alpha -0.01 Beta -0 Annual Standard Deviation 0 Annual Variance 0 Information Ratio -0.548 Tracking Error 0.155 Treynor Ratio 771346.915 Total Fees $0.00 Estimated Strategy Capacity $400000.00 Lowest Capacity Asset EURUSD 8G Portfolio Turnover 0.00% |
from AlgorithmImports import *
from datetime import timedelta
class ForexCFDAlgorithm(QCAlgorithm):
def Initialize(self):
# Example of adding a security and an indicator
self.symbol = self.AddForex("EURUSD", Resolution.Daily).Symbol
self.ema = self.EMA("EURUSD", 200, Resolution.Daily)
# Warm-up period based on the longest indicator
self.SetWarmUp(200, Resolution.Daily)
self.SetStartDate(2016, 5, 5) # Start Date
self.SetCash(50000) # Set Strategy Cash
# Indicators
self.daily_ema = self.EMA(self.symbol, 10, Resolution.Daily)
self.atr = self.ATR(self.symbol, 100, Resolution.Daily)
self.long_term_ema = self.EMA(self.symbol, 200, Resolution.Daily)
# State variables
self.is_long_term_bullish = False
self.batch_positions = []
self.batch_entry_prices = []
self.open_positions = 0
self.basket_size = 5
self.atr_multiplier_for_space = 1
self.last_basket_position_time = self.Time
self.trailing_stop = 0
self.last_10ema_cross = 'none'
self.previous_day = self.Time.date()
def OnEndOfDay(self, symbol):
if symbol != self.symbol:
return
# Update long-term bullish/bearish based on daily 200 EMA
if self.long_term_ema.IsReady:
self.is_long_term_bullish = self.Securities[self.symbol].Close > self.long_term_ema.Current.Value
def OnData(self, data: Slice) -> None:
# Check if the algorithm is still warming up
if self.IsWarmingUp:
return
# Check if we have data for the symbol
if not data.ContainsKey(self.symbol):
return
# Update ATR and EMA indicators
if not (self.daily_ema.IsReady and self.atr.IsReady and self.long_term_ema.IsReady):
return
# Retrieve current price
current_price = self.Securities[self.symbol].Price
# Update the state for the last 10 EMA cross
if current_price > self.daily_ema.Current.Value:
self.last_10ema_cross = 'above'
elif current_price < self.daily_ema.Current.Value:
self.last_10ema_cross = 'below'
# Execute trailing stop loss logic
self.ExecuteTrailingStopLoss(data)
# Check if conditions are met to start a new basket of positions
if self.ShouldStartNewBasket(current_price):
self.StartNewBasket()
# Trading logic for both bullish and bearish market conditions
if self.is_long_term_bullish and self.open_positions < self.basket_size:
if self.CanEnterNewPosition(current_price, 'Long'):
self.EnterPosition('Long', data)
elif not self.is_long_term_bullish and self.open_positions < self.basket_size:
if self.CanEnterNewPosition(current_price, 'Short'):
self.EnterPosition('Short', data)
def EnterPosition(self, direction, data):
lot_size = self.CalculateLotSize() # Calculate lot size based on risk management
self.batch_positions.append(self.symbol)
self.batch_entry_prices.append((data[self.symbol].Close, lot_size, direction))
self.open_positions += 1
self.last_basket_position_time = self.Time
if direction == 'Long':
self.Buy(self.symbol, lot_size)
else:
self.Sell(self.symbol, lot_size)
def ShouldStartNewBasket(self, current_price):
if self.last_10ema_cross == 'none':
return False
if self.is_long_term_bullish and self.last_10ema_cross == 'below' and current_price > self.daily_ema.Current.Value:
return True
if not self.is_long_term_bullish and self.last_10ema_cross == 'above' and current_price < self.daily_ema.Current.Value:
return True
return False
def StartNewBasket(self):
self.batch_positions = []
self.batch_entry_prices = []
self.open_positions = 0
self.last_basket_position_time = self.Time
def CanEnterNewPosition(self, current_price, direction):
if len(self.batch_entry_prices) == 0:
return True
last_entry_price, _, _ = self.batch_entry_prices[-1]
elapsed_time = self.Time - self.last_basket_position_time
# Check time delay and price distance
if elapsed_time >= timedelta(minutes=120):
atr_value = self.atr.Current.Value
pip_distance = self.atr_multiplier_for_space * atr_value
if direction == 'Long' and current_price - last_entry_price >= pip_distance:
return True
elif direction == 'Short' and last_entry_price - current_price >= pip_distance:
return True
return False
def CalculateLotSize(self):
# Implement lot size calculation logic based on account value and risk management
# Placeholder value
return 1
def ExecuteTrailingStopLoss(self, data):
if not self.atr.IsReady or len(self.batch_entry_prices) == 0:
return
atr_value = self.atr.Current.Value
trailing_stop_distance = 0.5 * atr_value # Trailing by 0.5x ATR
# Calculate new trailing stop for each position
for entry_price, lot_size, direction in self.batch_entry_prices:
if direction == 'Long':
# For long positions, trailing stop is below the entry price
new_trailing_stop = entry_price + trailing_stop_distance
if new_trailing_stop > self.trailing_stop:
self.trailing_stop = new_trailing_stop
elif direction == 'Short':
# For short positions, trailing stop is above the entry price
new_trailing_stop = entry_price - trailing_stop_distance
if self.trailing_stop == 0 or new_trailing_stop < self.trailing_stop:
self.trailing_stop = new_trailing_stop
# Check if current price has hit the trailing stop
current_price = data[self.symbol].Price
if ((direction == 'Long' and current_price <= self.trailing_stop) or
(direction == 'Short' and current_price >= self.trailing_stop)):
self.Liquidate(self.symbol)
self.Debug("Trailing stop hit. Liquidating positions.")