Overall Statistics
Total Trades
248
Average Win
0.45%
Average Loss
-0.81%
Compounding Annual Return
42.677%
Drawdown
14.200%
Expectancy
0.177
Net Profit
19.385%
Sharpe Ratio
1.575
Probabilistic Sharpe Ratio
60.211%
Loss Rate
25%
Win Rate
75%
Profit-Loss Ratio
0.56
Alpha
0.363
Beta
0.076
Annual Standard Deviation
0.244
Annual Variance
0.06
Information Ratio
0.371
Tracking Error
0.269
Treynor Ratio
5.041
Total Fees
$262.04
Estimated Strategy Capacity
$760000000.00
#Imports
from itertools import groupby
from datetime import datetime, timedelta
from pytz import utc

from clr import AddReference
AddReference("System")
AddReference("QuantConnect.Common")
AddReference("QuantConnect.Algorithm")
AddReference("QuantConnect.Algorithm.Framework")

from QuantConnect import Resolution, Extensions
from QuantConnect.Algorithm.Framework.Portfolio import *
from QuantConnect.Algorithm.Framework.Risk import *


#Global variables
Zero = int(0)

class TrailingStop(RiskManagementModel):

    def __init__(self):
        '''
        Initialization variables
        
        '''
        
        # Long Position Variables
        self.LongTrail = {}
        self.LongTrailingDrawdown = float(0.03)
        
        # Short Position Variables
        self.ShortTrail = {}
        self.ShortTrailingDrawdown = float(0.03)


    def ManageRisk(self, algorithm, targets):
        '''
        Main risk management handler. Passes algorithm and targets
        
        '''
        
        RiskAdjustedTargets = []
        
        for asset in self.LongTrail:
            if not algorithm.Portfolio[asset].Invested:
                self.LongTrail[asset] = [algorithm.Securities[asset].Price, 0]
            
        for asset in self.ShortTrail:
            if not algorithm.Portfolio[asset].Invested:
                self.ShortTrail[asset] = [algorithm.Securities[asset].Price, 0]
        
        invested = [x.Key for x in algorithm.Portfolio if x.Value.Invested]
    
        if invested:
            for asset in invested:
                
                if algorithm.Portfolio[asset].IsLong:
                    if asset not in self.LongTrail or self.LongTrail[asset][1] == 0:
                        self.LongTrail[asset] = [algorithm.Portfolio[asset].Price, algorithm.Portfolio[asset].Quantity]
                    
                elif algorithm.Portfolio[asset].IsShort:
                    if asset not in self.ShortTrail or self.ShortTrail[asset][1] == 0:
                        self.ShortTrail[asset] = [algorithm.Portfolio[asset].Price, algorithm.Portfolio[asset].Quantity]

                self.TrailingStop(algorithm, asset, RiskAdjustedTargets)
            
        return RiskAdjustedTargets
        
        
    def TrailingStop(self, algorithm, asset, RiskAdjustedTargets):
        '''
        Manages trailing stop for both long and short assets respectively
        
        '''
    
        if algorithm.Portfolio[asset].IsLong:
            
            if algorithm.Portfolio[asset].Price > self.LongTrail[asset][0]:
                self.LongTrail[asset][0] = algorithm.Portfolio[asset].Price
               
            elif algorithm.Portfolio[asset].Price / self.LongTrail[asset][0] < (1-self.LongTrailingDrawdown):
                RiskAdjustedTargets.append(PortfolioTarget(asset, 0))
                algorithm.Debug(f'Long trailing Stop Triggered for {asset}.  Current Price: {algorithm.Portfolio[asset].Price} | Highest Price: {self.LongTrail[asset][0]} | Loss: {algorithm.Portfolio[asset].Price / self.LongTrail[asset][0]} | Date: {algorithm.Time}')
                self.LongTrail.pop(asset)
                
        
        if algorithm.Portfolio[asset].IsShort:
            
            if algorithm.Portfolio[asset].Price < self.ShortTrail[asset][0]:
                self.ShortTrail[asset][0] = algorithm.Portfolio[asset].Price
               
            elif algorithm.Portfolio[asset].Price / self.ShortTrail[asset][0] > 1 / (1-self.ShortTrailingDrawdown):
                RiskAdjustedTargets.append(PortfolioTarget(asset, 0))
                algorithm.Debug(f'Short trailing Stop Triggered for {asset}. Current Price: {algorithm.Portfolio[asset].Price} | Lowest Price: {self.ShortTrail[asset][0]} | Loss: {algorithm.Portfolio[asset].Price / self.ShortTrail[asset][0]} | Date: {algorithm.Time}')
                self.ShortTrail.pop(asset)
        
        return RiskAdjustedTargets
from Execution.ImmediateExecutionModel import ImmediateExecutionModel
from Portfolio.EqualWeightingPortfolioConstructionModel import EqualWeightingPortfolioConstructionModel
from risk import TrailingStop

class EnergeticLightBrownCow(QCAlgorithm):

    def Initialize(self):
        self.SetStartDate(2020, 11, 12)  # Set Start Date
        self.SetCash(100000)  # Set Strategy Cash
        
        self.UniverseSettings.Resolution = Resolution.Daily
        
        tickers = ['SPY', 'TSLA', 'AAPL']
        symbols = [Symbol.Create(ticker, SecurityType.Equity, Market.USA) for ticker in tickers]
        self.SetUniverseSelection(ManualUniverseSelectionModel(symbols))
        
        # Option 1: Just emitting 1 insight with a long duration -- works
        #self.AddAlpha(MyAlpha())
        
        # Option 2: Emitting insights everyday -- doesn't work
        self.AddAlpha(ConstantAlphaModel(InsightType.Price, InsightDirection.Up, timedelta(days=1), 0.025, None)) 
                    
        self.SetPortfolioConstruction(EqualWeightingPortfolioConstructionModel(lambda time: None))
        
        self.AddRiskManagement(TrailingStop())
        
        self.SetExecution(ImmediateExecutionModel())

class MyAlpha(AlphaModel):
    symbols = []
    emitted = False
    
    def Update(self, algorithm, data):
        if self.emitted:
            return []
        
        for symbol in self.symbols:
            if not (data.ContainsKey(symbol) and data[symbol] is not None):
                return []
        
        self.emitted = True
        
        return [Insight.Price(symbol, timedelta(days=100), InsightDirection.Down) for symbol in self.symbols]
        
    def OnSecuritiesChanged(self, algorithm, changes):
        for security in changes.AddedSecurities:
            self.symbols.append(security.Symbol)