Created with Highcharts 12.1.2EquityJan 2024Jan…Feb 2024Mar 2024Apr 2024May 2024Jun 2024Jul 2024Aug 2024Sep 2024Oct 2024Nov 2024Dec 2024Jan 2025Feb 2025Mar 202550k100k150k200k-40-20000.250.5-101202.5M5M0250k500k
Overall Statistics
Total Orders
194
Average Win
1.12%
Average Loss
-0.27%
Compounding Annual Return
37.530%
Drawdown
24.400%
Expectancy
1.578
Start Equity
100000
End Equity
144143.37
Net Profit
44.143%
Sharpe Ratio
1.198
Sortino Ratio
1.758
Probabilistic Sharpe Ratio
55.368%
Loss Rate
50%
Win Rate
50%
Profit-Loss Ratio
4.16
Alpha
0.228
Beta
0.881
Annual Standard Deviation
0.333
Annual Variance
0.111
Information Ratio
0.652
Tracking Error
0.314
Treynor Ratio
0.453
Total Fees
$1605.05
Estimated Strategy Capacity
$1500000.00
Lowest Capacity Asset
USDCUSD 10B
Portfolio Turnover
1.49%
# region imports
from AlgorithmImports import *
from QuantConnect.DataSource import *
from QuantConnect.Data.UniverseSelection import *
# endregion


class CryptoMomentum(QCAlgorithm):
    def Initialize(self):
        self.SetStartDate(2024, 1,1)  # Set Start Date
        self.SetEndDate(2025, 3, 1)
        self.SetCash(100000)  # Set Strategy Cash

        self.SetBenchmark("SPY")
        # SET UNIVERSE BASED ON MARKET CAP - top 25 (update weekly)
        # Configure brokerage and security initializer
        self.SetBrokerageModel(BrokerageName.Kraken, AccountType.Cash)
        self.SetSecurityInitializer(BrokerageModelSecurityInitializer(self.BrokerageModel, FuncSecuritySeeder(self.GetLastKnownPrice)))

        # Universe settings 
        self.UniverseSettings.Resolution = Resolution.Daily
        self.UniverseSettings.Asynchronous = True
        # The asynchronous setting is a bool that defines whether or not LEAN can run universe selection asynchronously
        # utilizing concurrent execution to increase the speed of your algorithm.

        # Add crypto universe with a selection filter
        self.AddUniverse(CryptoUniverse.Kraken(self.UniverseSelectionFilter))
        # Indicator dictionaries 
        self.rsi = {}
        self.ema_short = {}
        self.ema_long = {}
        self.macd = {}
        self.entry_prices = {}

    def UniverseSelectionFilter(self, universe_day: List[CryptoUniverse]) -> List[Symbol]:
        # only want top 10 traded based on volume - to ensure liquidty for momentum algo 
        return [crypto.Symbol for crypto in sorted(universe_day, key=lambda x: x.VolumeInUsd, reverse=True)[:10]]

    def OnData(self, data):
        # Warm Up is a great way to prepare your algorithm and its indicators for trading.
        # if self.IsWarmingUp:
        #     return
        
        # Sell securities that are no longer in the universe
        for symbol in list(self.entry_prices.keys()):
            if symbol not in self.ActiveSecurities.Keys:
                self.Debug(f"Selling {symbol} as it is no longer in the top 25.")
                self.Liquidate(symbol)
                del self.entry_prices[symbol]

    
        for symbol in self.ActiveSecurities.Keys:
            # no data, move on
            if not data.Bars.ContainsKey(symbol):
                continue

            # Retrieve indicator values
            current_price = data.Bars[symbol].Close
            # RSI 
            rsi_value = self.rsi[symbol].Current.Value 
            # EMA
            ema_short_value = self.ema_short[symbol].Current.Value 
            ema_long_value = self.ema_long[symbol].Current.Value 
            # MACD 
            macd = self.macd[symbol]
            # The MACD line represents the distance between a shorter moving average and a longer moving average
            macd_line = macd.Current.Value
            #  9-day EMA of the MACD series
            signal_line = macd.Signal.Current.Value
            histogram = macd_line - signal_line

            # Exit Condition
            if self.entry_prices.get(symbol) is not None:  # Already holding
                if (
                    ema_short_value < ema_long_value  # Bearish crossover
                    or macd_line < signal_line and histogram < 0
# macd less than signal and negative - bear trend 
                ):
                    self.Debug(f"Selling {symbol}")
                    self.Liquidate(symbol)
                    self.entry_prices[symbol] = None

            # Entry Condition
            if (
                self.entry_prices.get(symbol) is None and  # Not already holding
                ema_short_value > ema_long_value and  # Bullish crossover
                rsi_value is not None and 55 <= rsi_value <= 85 and  # RSI in range
                macd_line > signal_line and histogram > 0
# macd more than signal and positive - bull trend 
            ):
                self.Debug(f"Buying {symbol}")
                self.SetHoldings(symbol, 0.15)  # Allocate 15% of the portfolio
                self.entry_prices[symbol] = current_price

    def OnSecuritiesChanged(self, changes: SecurityChanges):
            for security in changes.AddedSecurities:
                symbol = security.Symbol

                # Initialize indicators for the symbol
                self.rsi[symbol] = self.RSI(symbol, 14, MovingAverageType.Simple, Resolution.Daily)
                self.ema_short[symbol] = self.EMA(symbol, 12, Resolution.Daily)
                self.ema_long[symbol] = self.EMA(symbol, 26, Resolution.Daily)
                self.macd[symbol] = self.MACD(symbol, 12, 26, 9, MovingAverageType.Exponential, Resolution.Daily)