| Overall Statistics |
|
Total Trades 0 Average Win 0% Average Loss 0% Compounding Annual Return 0% Drawdown 0% Expectancy 0 Net Profit 0% Sharpe Ratio 0 Sortino Ratio 0 Probabilistic Sharpe Ratio 0% Loss Rate 0% Win Rate 0% Profit-Loss Ratio 0 Alpha 0 Beta 0 Annual Standard Deviation 0 Annual Variance 0 Information Ratio -0.436 Tracking Error 0.191 Treynor Ratio 0 Total Fees $0.00 Estimated Strategy Capacity $0 Lowest Capacity Asset Portfolio Turnover 0% |
from AlgorithmImports import *
class ForexCFDAlgorithm(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2016, 1, 1) # Start Date
self.SetCash(50000) # Set Strategy Cash
# Forex symbol
self.symbol = self.AddForex("EURUSD", Resolution.Daily).Symbol
# Indicators
self.daily_ema = self.EMA(self.symbol, 10, Resolution.Daily)
self.atr = self.ATR(self.symbol, 100, Resolution.Daily)
self.month_end_window = RollingWindow[TradeBar](2) # For monthly EMA
# State variables
self.is_long_term_bullish = False
self.batch_positions = []
self.batch_entry_prices = []
self.open_positions = 0
self.basket_size = 5 # Define basket size
self.atr_multiplier_for_space = 1 # Define ATR multiplier for spacing
self.last_basket_position_time = self.Time
self.trailing_stop = 0
# Initialize monthly EMA
self.monthly_ema = self.EMA(self.symbol, 30, Resolution.Daily) # 30 days for a rough monthly EMA
def OnEndOfDay(self, symbol):
if symbol != self.symbol:
return
# Add the daily bar to the rolling window for monthly EMA calculation
self.month_end_window.Add(self.CurrentSlice[symbol])
# Update the monthly EMA at the end of each month
if self.month_end_window.IsReady and self.Time.month != self.month_end_window[0].EndTime.month:
monthly_close = self.month_end_window[0].Close
self.monthly_ema.Update(self.month_end_window[0].EndTime, monthly_close)
# Update long-term bullish/bearish based on monthly EMA
if self.month_end_window.IsReady and self.monthly_ema.IsReady:
self.is_long_term_bullish = self.Securities[self.symbol].Close > self.monthly_ema.Current.Value
def OnData(self, data):
if not data.ContainsKey(self.symbol):
self.Debug(f"No data for {self.symbol.Value}")
return
self.Log("OnData called")
if not self.symbol in data:
self.Log(f"No data for {self.symbol}")
return
if self.is_long_term_bullish and self.open_positions < self.basket_size:
self.Log(f"Trying to enter long position. Open Positions: {self.open_positions}, Basket Size: {self.basket_size}")
self.EnterPosition('Long', data)
elif not self.is_long_term_bullish and self.open_positions < self.basket_size:
self.Log(f"Trying to enter short position. Open Positions: {self.open_positions}, Basket Size: {self.basket_size}")
self.EnterPosition('Short', data)
else:
self.Log(f"Position entry skipped. Open Positions: {self.open_positions}, Basket Size: {self.basket_size}")
def EnterPosition(self, direction, data):
if self.open_positions < self.basket_size:
lot_size = 0.05 # Define lot size
self.Log(f"Entering {direction} position: Lot Size: {lot_size}")
if direction == 'Long':
self.Buy(self.symbol, lot_size)
elif direction == 'Short':
self.Sell(self.symbol, lot_size)
self.batch_positions.append(self.symbol)
self.batch_entry_prices.append((data[self.symbol].Close, lot_size, direction)) # Include direction
self.open_positions += 1
self.last_basket_position_time = self.Time
else:
self.Log("Position entry aborted. Max positions reached.")
def OnEndOfAlgorithm(self):
# Liquidate all positions at the end of the algorithm
self.Liquidate()
### Part 3: Risk Management
def ExecuteTrailingStopLoss(self, data):
atr_value = self.atr.Current.Value
trailing_stop_distance = 5 * atr_value # Example multiplier
for price, size, direction in self.batch_entry_prices: # Include direction in loop
if direction == 'Long':
self.trailing_stop = max(self.trailing_stop, price - trailing_stop_distance)
elif direction == 'Short':
self.trailing_stop = min(self.trailing_stop, price + trailing_stop_distance)
current_price = data[self.symbol].Close
if ((direction == 'Long' and current_price <= self.trailing_stop) or
(direction == 'Short' and current_price >= self.trailing_stop)):
self.Liquidate(self.symbol)
from AlgorithmImports import *
from datetime import timedelta
class ForexCFDAlgorithm(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2016, 1, 1) # Start Date
self.SetCash(50000) # Set Strategy Cash
self.symbol = self.AddForex("EURUSD", Resolution.Daily).Symbol
self.daily_ema = self.EMA(self.symbol, 10, Resolution.Daily)
self.atr = self.ATR(self.symbol, 14, Resolution.Daily)
self.month_end_window = RollingWindow[TradeBar](2)
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.monthly_ema = self.EMA(self.symbol, 30, Resolution.Daily)
def OnEndOfDay(self, symbol):
if symbol != self.symbol:
return
self.month_end_window.Add(self.CurrentSlice[symbol])
if self.month_end_window.IsReady and self.Time.month != self.month_end_window[0].EndTime.month:
monthly_close = self.month_end_window[0].Close
self.monthly_ema.Update(self.month_end_window[0].EndTime, monthly_close)
if self.month_end_window.IsReady and self.monthly_ema.IsReady:
self.is_long_term_bullish = self.Securities[self.symbol].Close > self.monthly_ema.Current.Value
def OnData(self, data):
if not data.ContainsKey(self.symbol):
return
current_price = data[self.symbol].Close
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'
if self.ShouldStartNewBasket(current_price):
self.StartNewBasket()
self.EnterPosition(self.is_long_term_bullish, data)
elif self.open_positions < self.basket_size and self.Time - self.last_basket_position_time > timedelta(minutes=120):
distance = abs(current_price - self.batch_entry_prices[-1][0])
if distance >= self.atr.Current.Value * self.atr_multiplier_for_space:
self.EnterPosition(self.is_long_term_bullish, data)
def EnterPosition(self, is_bullish, data):
lot_size = 0.05
direction = 'Long' if is_bullish else 'Short'
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 True
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 OnEndOfAlgorithm(self):
# Liquidate all positions at the end of the algorithm
self.Liquidate()
### Part 3: Risk Management
def ExecuteTrailingStopLoss(self, data):
atr_value = self.atr.Current.Value
trailing_stop_distance = 5 * atr_value # Example multiplier
for price, size, direction in self.batch_entry_prices: # Include direction in loop
if direction == 'Long':
self.trailing_stop = max(self.trailing_stop, price - trailing_stop_distance)
elif direction == 'Short':
self.trailing_stop = min(self.trailing_stop, price + trailing_stop_distance)
current_price = data[self.symbol].Close
if ((direction == 'Long' and current_price <= self.trailing_stop) or
(direction == 'Short' and current_price >= self.trailing_stop)):
self.Liquidate(self.symbol)
#########Check for order events only - remove after checking########
def OnOrderEvent(self, orderEvent):
if orderEvent.Status == OrderStatus.Filled:
self.Debug(str(orderEvent))from AlgorithmImports import *
from datetime import timedelta
class ForexCFDAlgorithm(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2016, 1, 1) # Start Date
self.SetCash(50000) # Set Strategy Cash
# Forex symbol
self.symbol = self.AddForex("EURUSD", Resolution.Daily).Symbol
# Indicators
self.daily_ema = self.EMA(self.symbol, 10, Resolution.Daily)
self.atr = self.ATR(self.symbol, 100, Resolution.Daily)
# Replace monthly EMA with daily 200 EMA for long-term trend
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 # Define basket size
self.atr_multiplier_for_space = 1 # Define ATR multiplier for spacing
self.last_basket_position_time = self.Time
self.trailing_stop = 0
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):
if not data.ContainsKey(self.symbol):
self.Debug(f"No data for {self.symbol.Value}")
return
self.Log("OnData called")
if not self.symbol in data:
self.Log(f"No data for {self.symbol}")
return
# Update long-term trend based on the 200-day EMA
self.is_long_term_bullish = self.Securities[self.symbol].Price > self.daily_200ema.Current.Value
# Check if the symbol's daily EMA is ready before proceeding
if not self.daily_ema.IsReady:
return
# Logic to handle entry for both bullish and bearish trends
if self.is_long_term_bullish and self.open_positions < self.basket_size:
self.Log(f"Trying to enter long position. Open Positions: {self.open_positions}, Basket Size: {self.basket_size}")
self.EnterPosition('Long', data)
elif not self.is_long_term_bullish and self.open_positions < self.basket_size:
self.Log(f"Trying to enter short position. Open Positions: {self.open_positions}, Basket Size: {self.basket_size}")
self.EnterPosition('Short', data)
else:
self.Log(f"Position entry skipped. Open Positions: {self.open_positions}, Basket Size: {self.basket_size}")
def EnterPosition(self, is_bullish, data):
lot_size = 0.05
direction = 'Long' if is_bullish else 'Short'
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 True
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 OnEndOfAlgorithm(self):
# Liquidate all positions at the end of the algorithm
self.Liquidate()
### Part 3: Risk Management
def ExecuteTrailingStopLoss(self, data):
atr_value = self.atr.Current.Value
trailing_stop_distance = 5 * atr_value # Example multiplier
for price, size, direction in self.batch_entry_prices: # Include direction in loop
if direction == 'Long':
self.trailing_stop = max(self.trailing_stop, price - trailing_stop_distance)
elif direction == 'Short':
self.trailing_stop = min(self.trailing_stop, price + trailing_stop_distance)
current_price = data[self.symbol].Close
if ((direction == 'Long' and current_price <= self.trailing_stop) or
(direction == 'Short' and current_price >= self.trailing_stop)):
self.Liquidate(self.symbol)
from AlgorithmImports import *
from datetime import timedelta
class ForexCFDAlgorithm(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2016, 1, 1) # Start Date
self.SetCash(50000) # Set Strategy Cash
# Forex symbol
self.symbol = self.AddForex("EURUSD", Resolution.Daily).Symbol
# 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'
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):
if not data.ContainsKey(self.symbol):
return
# Update ATR and ensure all indicators are ready
if not self.daily_ema.IsReady or not self.atr.IsReady or not self.long_term_ema.IsReady:
return
current_price = data[self.symbol].Price
# Update last 10 EMA cross state
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'
# Check if conditions are met to start a new basket
if self.ShouldStartNewBasket(current_price):
self.StartNewBasket()
# Logic to handle entry for both bullish and bearish trends
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 = 0.05 # Adjust lot size as needed
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 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):
if not data.ContainsKey(self.symbol):
self.Debug(f"No data for {self.symbol.Value}")
return
self.Log("OnData called")
if not self.symbol in data:
self.Log(f"No data for {self.symbol}")
return
# Update long-term trend based on the 200-day EMA
self.is_long_term_bullish = self.Securities[self.symbol].Price > self.long_term_ema.Current.Value
# Check if the symbol's daily EMA is ready before proceeding
if not self.daily_ema.IsReady:
return
# Logic to handle entry for both bullish and bearish trends
if self.is_long_term_bullish and self.open_positions < self.basket_size:
self.Log(f"Trying to enter long position. Open Positions: {self.open_positions}, Basket Size: {self.basket_size}")
self.EnterPosition('Long', data)
elif not self.is_long_term_bullish and self.open_positions < self.basket_size:
self.Log(f"Trying to enter short position. Open Positions: {self.open_positions}, Basket Size: {self.basket_size}")
self.EnterPosition('Short', data)
else:
self.Log(f"Position entry skipped. Open Positions: {self.open_positions}, Basket Size: {self.basket_size}")
def EnterPosition(self, direction, data):
lot_size = 0.05 # Define lot size for each trade
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 ExecuteTrailingStopLoss(self, data):
atr_value = self.atr.Current.Value
trailing_stop_distance = 5 * atr_value # Example multiplier
for price, size, direction in self.batch_entry_prices:
if direction == 'Long':
self.trailing_stop = max(self.trailing_stop, price - trailing_stop_distance)
elif direction == 'Short':
self.trailing_stop = min(self.trailing_stop, price + trailing_stop_distance)
current_price = data[self.symbol].Close
if ((direction == 'Long' and current_price <= self.trailing_stop) or
(direction == 'Short' and current_price >= self.trailing_stop)):
self.Liquidate(self.symbol)
def OnEndOfAlgorithm(self):
# Liquidate all positions at the end of the algorithm
self.Liquidate()from AlgorithmImports import *
from datetime import timedelta
class ForexCFDAlgorithm(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2016, 1, 1) # Start Date
self.SetCash(50000) # Set Strategy Cash
# Forex symbol
self.symbol = self.AddForex("EURUSD", Resolution.Daily).Symbol
# 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'
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):
if not data.ContainsKey(self.symbol):
return
# Update ATR and ensure all indicators are ready
if not self.daily_ema.IsReady or not self.atr.IsReady or not self.long_term_ema.IsReady:
return
current_price = data[self.symbol].Price
# Update last 10 EMA cross state
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'
# Check if conditions are met to start a new basket
if self.ShouldStartNewBasket(current_price):
self.StartNewBasket()
# Logic to handle entry for both bullish and bearish trends
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 0.05
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 i in range(len(self.batch_entry_prices)):
entry_price, lot_size, direction = self.batch_entry_prices[i]
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.")#from AlgorithmImports import *
#from datetime import timedelta
#class ForexCFDAlgorithm(QCAlgorithm):
# from AlgorithmImports import *
from AlgorithmImports import *
class ForexCFDAlgorithm(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2016, 5, 5) # Start Date
self.SetCash(50000) # Set Strategy Cash
self.symbol = self.AddForex("EURUSD", Resolution.Daily).Symbol
self.Debug("Initialization complete")
def OnData(self, data):
# Manual trigger for a long trade
if self.Time.hour == 10 and not self.Portfolio.Invested:
quantity = 1 # Positive for long position
self.MarketOrder(self.symbol, quantity)
self.Debug(f"Manual long trade executed for {self.symbol.Value}")
#def Initialize(self):
# self.SetStartDate(2016, 1, 1) # Start Date
# self.SetCash(50000) # Set Strategy Cash
# Forex symbol
# self.symbol = self.AddForex("EURUSD", Resolution.Daily).Symbol
# 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'
#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):
# Force open a position for testing
# if not self.Portfolio.Invested:
# self.MarketOrder(self.symbol, 0.05) # Replace 'quantity' with the desired number of units
# self.Debug(f"Forced position opened for {self.symbol.Value}")
#if not data.ContainsKey(self.symbol):
# return
# Update ATR and ensure all indicators are ready
#if not self.daily_ema.IsReady or not self.atr.IsReady or not self.long_term_ema.IsReady:
# return
#current_price = data[self.symbol].Price
# Update last 10 EMA cross state
#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'
# Check if conditions are met to start a new basket
#if self.ShouldStartNewBasket(current_price):
# self.StartNewBasket()
# Execute trailing stop loss logic
#self.ExecuteTrailingStopLoss(data)
# Logic to handle entry for both bullish and bearish trends
#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 0.05
#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 i in range(len(self.batch_entry_prices)):
# entry_price, lot_size, direction = self.batch_entry_prices[i]
# 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.")from AlgorithmImports import *
class ForexCFDAlgorithm(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2020, 1, 1) # Start Date
self.SetCash(100000) # Set Strategy Cash
# Forex symbol
self.symbol = self.AddForex("EURUSD", Resolution.Daily).Symbol
# Indicators
self.daily_ema = self.EMA(self.symbol, 10, Resolution.Daily)
self.atr = self.ATR(self.symbol, 100, Resolution.Daily)
self.month_end_window = RollingWindow[TradeBar](2) # For monthly EMA
# State variables
self.is_long_term_bullish = False
self.batch_positions = []
self.batch_entry_prices = []
self.open_positions = 0
self.basket_size = 5 # Define basket size
self.atr_multiplier_for_space = 2 # Define ATR multiplier for spacing
self.last_basket_position_time = self.Time
self.trailing_stop = 0
# Initialize monthly EMA
self.monthly_ema = self.EMA(self.symbol, 30, Resolution.Daily) # 30 days for a rough monthly EMA
def OnEndOfDay(self, symbol):
if symbol != self.symbol:
return
# Add the daily bar to the rolling window for monthly EMA calculation
self.month_end_window.Add(self.CurrentSlice[symbol])
# Update the monthly EMA at the end of each month
if self.month_end_window.IsReady and self.Time.month != self.month_end_window[0].EndTime.month:
monthly_close = self.month_end_window[0].Close
self.monthly_ema.Update(self.month_end_window[0].EndTime, monthly_close)
def OnData(self, data):
# Ensure data for the symbol is available
if not self.symbol in data:
return
# Update daily EMA and ATR
if not self.daily_ema.Update(data[self.symbol]) or not self.atr.Update(data[self.symbol]):
return
# Update long-term bullish/bearish based on monthly EMA
if self.month_end_window.IsReady and self.monthly_ema.IsReady:
self.is_long_term_bullish = data[self.symbol].Close > self.monthly_ema.Current.Value
# Position entry logic
if (self.Time - self.last_basket_position_time).total_minutes() >= 120:
if self.is_long_term_bullish and data[self.symbol].Close > self.daily_ema.Current.Value:
self.EnterPosition('Long', data)
elif not self.is_long_term_bearish and data[self.symbol].Close < self.daily_ema.Current.Value:
self.EnterPosition('Short', data)
# Risk management
if self.open_positions > 0:
self.ExecuteTrailingStopLoss(data)
def EnterPosition(self, direction, data):
self.Debug(f"Attempting to enter {direction} position")
if self.open_positions < self.basket_size:
if self.open_positions == 0 or (data[self.symbol].Close - self.batch_entry_prices[-1][0] >= self.atr.Current.Value * self.atr_multiplier_for_space):
lot_size = 0.05 # Define lot size
if direction == 'Long':
self.Buy(self.symbol, lot_size)
else:
self.Sell(self.symbol, lot_size)
# Update state variables
self.batch_positions.append(self.symbol)
self.batch_entry_prices.append((data[self.symbol].Close, lot_size))
self.open_positions += 1
self.last_basket_position_time = self.Time
self.Debug(f"Position entered: {direction}, Lot size: {lot_size}, Open positions: {self.open_positions}")
def OnEndOfAlgorithm(self):
# Liquidate all positions at the end of the algorithm
self.Liquidate()
### Part 3: Risk Management
def ExecuteTrailingStopLoss(self, data):
atr_value = self.atr.Current.Value
trailing_stop_distance = 3 * atr_value # Example multiplier
# Update trailing stop for each position
for price, size in self.batch_entry_prices:
if self.Portfolio[self.symbol].IsLong:
self.trailing_stop = max(self.trailing_stop, price - trailing_stop_distance)
elif self.Portfolio[self.symbol].IsShort:
self.trailing_stop = min(self.trailing_stop, price + trailing_stop_distance)
# Check and execute trailing stop
current_price = data[self.symbol].Close
if (self.Portfolio[self.symbol].IsLong and current_price <= self.trailing_stop) or \
(self.Portfolio[self.symbol].IsShort and current_price >= self.trailing_stop):
self.Liquidate(self.symbol)
from AlgorithmImports import *
class ForexCFDAlgorithm(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2020, 1, 1) # Start Date
self.SetCash(100000) # Set Strategy Cash
# Forex symbol
self.symbol = self.AddForex("EURUSD", Resolution.Daily).Symbol
# Indicators
self.daily_ema = self.EMA(self.symbol, 10, Resolution.Daily)
self.atr = self.ATR(self.symbol, 100, Resolution.Daily)
self.month_end_window = RollingWindow[TradeBar](2) # For monthly EMA
# State variables
self.is_long_term_bullish = False
self.batch_positions = []
self.batch_entry_prices = []
self.open_positions = 0
self.basket_size = 5 # Define basket size
self.atr_multiplier_for_space = 2 # Define ATR multiplier for spacing
self.last_basket_position_time = self.Time
self.trailing_stop = 0
def OnEndOfDay(self, symbol):
if symbol != self.symbol:
return
# Add the daily bar to the rolling window for monthly EMA calculation
self.month_end_window.Add(self.CurrentSlice[symbol])
# Update the monthly EMA at the end of each month
if self.month_end_window.IsReady and self.Time.month != self.month_end_window[0].EndTime.month:
monthly_close = self.month_end_window[0].Close
self.monthly_ema.Update(self.month_end_window[0].EndTime, monthly_close)
def OnData(self, data):
# Ensure data for the symbol is available
if not self.symbol in data:
return
# Update daily EMA and ATR
if not self.daily_ema.Update(data[self.symbol]) or not self.atr.Update(data[self.symbol]):
return
# Update long-term bullish/bearish based on monthly EMA
if self.month_end_window.IsReady and self.monthly_ema.IsReady:
self.is_long_term_bullish = data[self.symbol].Close > self.monthly_ema.Current.Value
# Position entry logic
if (self.Time - self.last_basket_position_time).total_minutes() >= 120:
if self.is_long_term_bullish and data[self.symbol].Close > self.daily_ema.Current.Value:
self.EnterPosition('Long', data)
elif not self.is_long_term_bearish and data[self.symbol].Close < self.daily_ema.Current.Value:
self.EnterPosition('Short', data)
def EnterPosition(self, direction, data):
if self.open_positions < self.basket_size:
if self.open_positions == 0 or (data[self.symbol].Close - self.batch_entry_prices[-1][0] >= self.atr.Current.Value * self.atr_multiplier_for_space):
lot_size = 0.05 # Define lot size
if direction == 'Long':
self.Buy(self.symbol, lot_size)
else:
self.Sell(self.symbol, lot_size)
# Update state variables
self.batch_positions.append(self.symbol)
self.batch_entry_prices.append((data[self.symbol].Close, lot_size))
self.open_positions += 1
self.last_basket_position_time = self.Time
def OnEndOfAlgorithm(self):
# Liquidate all positions at the end of the algorithm
self.Liquidate()
### Part 3: Risk Management
def ExecuteTrailingStopLoss(self, data):
atr_value = self.atr.Current.Value
trailing_stop_distance = 3 * atr_value # Example multiplier
# Update trailing stop for each position
for price, size in self.batch_entry_prices:
if self.Portfolio[self.symbol].IsLong:
self.trailing_stop = max(self.trailing_stop, price - trailing_stop_distance)
elif self.Portfolio[self.symbol].IsShort:
self.trailing_stop = min(self.trailing_stop, price + trailing_stop_distance)
# Check and execute trailing stop
current_price = data[self.symbol].Close
if (self.Portfolio[self.symbol].IsLong and current_price <= self.trailing_stop) or \
(self.Portfolio[self.symbol].IsShort and current_price >= self.trailing_stop):
self.Liquidate(self.symbol)
def OnData(self, data):
# Previous OnData content...
# Risk management
if self.open_positions > 0:
self.ExecuteTrailingStopLoss(data)
### Part 4: Monthly EMA Calculation
def OnEndOfDay(self, symbol):
# Previous OnEndOfDay content...
# Update long-term bullish/bearish based on monthly EMA
if self.month_end_window.IsReady and self.monthly_ema.IsReady:
self.is_long_term_bullish = data[self.symbol].Close > self.monthly_ema.Current.Value