Overall Statistics |
Total Trades 41 Average Win 0.57% Average Loss -0.69% Compounding Annual Return -9.461% Drawdown 7.200% Expectancy -0.365 Net Profit -5.789% Sharpe Ratio -1.271 Probabilistic Sharpe Ratio 2.578% Loss Rate 65% Win Rate 35% Profit-Loss Ratio 0.82 Alpha -0.069 Beta -0.046 Annual Standard Deviation 0.059 Annual Variance 0.004 Information Ratio -0.529 Tracking Error 0.397 Treynor Ratio 1.625 Total Fees $125.60 |
from clr import AddReference AddReference("QuantConnect.Common") AddReference("QuantConnect.Algorithm") AddReference("QuantConnect.Algorithm.Framework") AddReference("QuantConnect.Indicators") from QuantConnect import * from QuantConnect.Indicators import * from QuantConnect.Algorithm import * from QuantConnect.Algorithm.Framework import * from QuantConnect.Algorithm.Framework.Alphas import * from QuantConnect.Data.Consolidators import * from datetime import timedelta class MACDForexAlphaModel(AlphaModel): def __init__(self, fastPeriod = 12, slowPeriod = 26, signalPeriod = 9, trendPeriod = 200, movingAverageType = MovingAverageType.Exponential, resolution = Resolution.Daily, difference = 0): self.fastPeriod = fastPeriod self.slowPeriod = slowPeriod self.signalPeriod = signalPeriod self.trendPeriod = trendPeriod self.movingAverageType = movingAverageType self.resolution = resolution self.insightPeriod = timedelta(days=10) self.bounceThresholdPercent = 0.000 self.differenceThreshold = difference self.symbolData = {} def Update(self, algorithm, data): ''' Determines an insight for each security based on it's current MACD signal Args: algorithm: The algorithm instance data: The new data available Returns: The new insights generated''' insights = [] for key, sd in self.symbolData.items(): if sd.Security.Price == 0: continue #if not sd.MACD.IsReady:# or not sd.Trend.IsReady: # continue direction = InsightDirection.Flat if sd.MACD.Current.Value > sd.MACD.Signal.Current.Value * (1 + self.bounceThresholdPercent):# and sd.Security.Price > sd.Trend.Current.Value:# and sd.MACD.Current.Value < (0 + self.differenceThreshold): direction = InsightDirection.Up elif sd.MACD.Current.Value < sd.MACD.Signal.Current.Value * (1 - self.bounceThresholdPercent):# and sd.Security.Price < sd.Trend.Current.Value:# and sd.MACD.Current.Value > (0 + self.differenceThreshold): direction = InsightDirection.Down if direction == sd.PreviousDirection: continue insight = Insight.Price(sd.Security.Symbol, self.insightPeriod, direction) sd.PreviousDirection = insight.Direction insights.append(insight) return insights def OnSecuritiesChanged(self, algorithm, changes): '''Event fired each time the we add/remove securities from the data feed. This initializes the MACD for each added security and cleans up the indicator for each removed security. Args: algorithm: The algorithm instance that experienced the change in securities changes: The security additions and removals from the algorithm''' for added in changes.AddedSecurities: self.symbolData[added.Symbol] = SymbolData(algorithm, added, self.fastPeriod, self.slowPeriod, self.signalPeriod, self.trendPeriod, self.movingAverageType, self.resolution) for removed in changes.RemovedSecurities: data = self.symbolData.pop(removed.Symbol, None) if data is not None: # clean up our consolidator algorithm.SubscriptionManager.RemoveConsolidator(removed.Symbol, data.Consolidator) class SymbolData: def __init__(self, algorithm, security, fastPeriod, slowPeriod, signalPeriod, trendPeriod, movingAverageType, resolution): self.Security = security self.MACD = MovingAverageConvergenceDivergence(fastPeriod, slowPeriod, signalPeriod, movingAverageType) # self.Trend = ExponentialMovingAverage(trendPeriod) self.Consolidator = algorithm.ResolveConsolidator(security.Symbol, resolution) #resolveconsolidator vs addconsolidator # self.Consolidator = QuoteBarConsolidator(timedelta(days=1)) # Register consolidator to get automatically updated with minute data # algorithm.SubscriptionManager.AddConsolidator(self.Security.Symbol, self.Consolidator) algorithm.RegisterIndicator(security.Symbol, self.MACD, self.Consolidator) # algorithm.RegisterIndicator(security.Symbol, self.Trend, self.Consolidator) self.PreviousDirection = None return history = algorithm.History(security.Symbol, 200, Resolution.Daily) for bar in history.itertuples(): time = bar.Index[1] close = bar.close self.MACD.Update(time, close) # self.Trend.Update(time, close)
from Execution.ImmediateExecutionModel import ImmediateExecutionModel from Portfolio.EqualWeightingPortfolioConstructionModel import EqualWeightingPortfolioConstructionModel from G10CurrencySelectionModel import G10CurrencySelectionModel from MACDForexModel import MACDForexAlphaModel from RiskManagementModule import ModifiedTrailingStopRiskManagementModel class MACDTrendFollower(QCAlgorithm): def Initialize(self): self.SetStartDate(2020, 1, 1) # Set Start Date self.SetCash(100000) # Set Strategy Cash # self.AddEquity("SPY", Resolution.Minute) self.SetExecution(ImmediateExecutionModel()) self.SetPortfolioConstruction(EqualWeightingPortfolioConstructionModel()) self.AddAlpha(MACDForexAlphaModel()) # self.SetRiskManagement(ModifiedTrailingStopRiskManagementModel(0.02, .04)) self.AddUniverseSelection( G10CurrencySelectionModel()) self.SetBrokerageModel(BrokerageName.FxcmBrokerage, AccountType.Margin) self.UniverseSettings.Leverage = 1 self.UniverseSettings.Resolution = Resolution.Daily
from QuantConnect import * from Selection.ManualUniverseSelectionModel import ManualUniverseSelectionModel class G10CurrencySelectionModel(ManualUniverseSelectionModel): def __init__(self): super().__init__([Symbol.Create(x, SecurityType.Forex, Market.FXCM) for x in [ "EURJPY"]]) # super().__init__([Symbol.Create(x, SecurityType.Forex, Market.Oanda) for x in [ "EURUSD", "GBPUSD", "USDJPY", "AUDUSD", "NZDUSD","USDCAD", "USDCHF", "USDNOK", "USDSEK"]])
from clr import AddReference AddReference("System") AddReference("QuantConnect.Common") AddReference("QuantConnect.Algorithm") AddReference("QuantConnect.Algorithm.Framework") from QuantConnect import * from QuantConnect.Algorithm import * from QuantConnect.Algorithm.Framework import * from QuantConnect.Algorithm.Framework.Portfolio import PortfolioTarget from QuantConnect.Algorithm.Framework.Risk import RiskManagementModel class ModifiedTrailingStopRiskManagementModel(RiskManagementModel): '''Provides an implementation of IRiskManagementModel that limits the maximum possible loss measured from the highest unrealized profit''' def __init__(self, maximumDrawdownPercent = 0.01, takeProfitPercent = 0.01): '''Initializes a new instance of the TrailingStopRiskManagementModel class Args: maximumDrawdownPercent: The maximum percentage drawdown allowed for algorithm portfolio compared with the highest unrealized profit, defaults to 5% drawdown''' self.maximumDrawdownPercent = -abs(maximumDrawdownPercent) self.takeProfitPercent = takeProfitPercent self.trailingHighs = dict() def ManageRisk(self, algorithm, targets): '''Manages the algorithm's risk at each time step Args: algorithm: The algorithm instance targets: The current portfolio targets to be assessed for risk''' riskAdjustedTargets = list() for kvp in algorithm.Portfolio: symbol = kvp.Key security = kvp.Value # Remove if not invested if not security.Invested: self.trailingHighs.pop(symbol, None) continue # Add newly invested securities if symbol not in self.trailingHighs: self.trailingHighs[symbol] = security.HoldingsCost # Set to average holding cost continue if security.UnrealizedProfitPercent >= self.takeProfitPercent: algorithm.Debug(f"{algorithm.Time}: 'take profit'") riskAdjustedTargets.append(PortfolioTarget(symbol, 0)) continue bidPrice = algorithm.Securities[symbol].BidPrice askPrice = algorithm.Securities[symbol].AskPrice if security.IsLong: # Check for new highs and update - set to tradebar high algorithm.Debug(f"{algorithm.Time}: {symbol}: 'long'") if self.trailingHighs[symbol] < askPrice: self.trailingHighs[symbol] = askPrice continue # Check for securities past the drawdown limit securityHigh = self.trailingHighs[symbol] drawdown = (bidPrice / securityHigh) - 1 if security.IsShort: # Check for new lows and update - set to tradebar Low algorithm.Debug(f"{algorithm.Time}: {symbol}: 'short'") if self.trailingHighs[symbol] > bidPrice: self.trailingHighs[symbol] = bidPrice continue # Check for securities past the drawdown limit securityHigh = self.trailingHighs[symbol] drawdown = (securityHigh / askPrice) - 1 if drawdown < self.maximumDrawdownPercent: # liquidate algorithm.Debug(f"{algorithm.Time}: loss liquidate") riskAdjustedTargets.append(PortfolioTarget(symbol, 0)) return riskAdjustedTargets