Confronta il testo

Trova la differenza tra due file di testo

Editor live

Nascondi spazi bianchi

Comprimi invariate

Senza a capo

Layout

Livello di dettaglio

Stili testo

Evidenziazione sintassi

Diffchecker Desktop Il modo più sicuro per usare Diffchecker. Ottieni l'app Diffchecker Desktop: i tuoi diff non lasciano mai il tuo computer!Ottieni Desktop

Untitled Diff

Creato 5 anni faIl diff non scade mai

34 rimozioni

Linee
Totale
Rimosso

Caratteri
Totale
Rimosso

Per continuare a utilizzare questa funzione, aggiorna a Diffchecker Pro Visualizza prezzi

336 linee

121 aggiunte

Linee
Totale
Aggiunto

Caratteri
Totale
Aggiunto

Per continuare a utilizzare questa funzione, aggiorna a Diffchecker Pro Visualizza prezzi

417 linee

##########################################################################

# Inspired by @nitay-rabinovich at QuqntConnect

# https://www.quantconnect.com/forum/discussion/12768/share-kalman-filter-crossovers-for-crypto-and-smart-rollingwindows/p1/comment-38144

##########################################################################

# EMA Crossover In a Crypto Universe

# ---------------------------------------------

# FOR EDUCATIONAL PURPOSES ONLY. DO NOT DEPLOY.

# Entry:

# -------

# Minimum volume threshold traded

# and

# Price > Fast Daily EMA

# and

# Fast Daily EMA > Slow Daily EMA

# Exit:

# ------

# Price < Slow Daily EMA

# or

# Slow Daily EMA < Fast Daily EMA

# Additional Consideration:

# --------------------------

# Max exposure pct: Total % of available capital to trade with at any time

# Max holdings: Total # of positions that can be held simultaneously

# Rebalance Weekly: If false, only rebalance when we add/remove positions

# UseMomWeight: If true, rebalance w/momentum-based weights (top gainers=more weight)

#########################################################################

from SmartRollingWindow import *

class EMACrossoverUniverse(QCAlgorithm):

def Initialize(self):

self.InitAlgoParams()

self.InitAssets()

self.InitUniverse()

self.InitBacktestParams()

self.ScheduleRoutines()

## Set backtest params: dates, cash, etc. Called from Initialize().

## ----------------------------------------------------------------

def InitBacktestParams(self):

self.SetStartDate(2020, 1, 1)

# self.SetEndDate(2019, 2, 1)

self.SetCash(100000)

self.SetBenchmark(Symbol.Create("BTCUSDT", SecurityType.Crypto, Market.Binance))

def InitUniverse(self):

self.UniverseSettings.Resolution = Resolution.Daily

self.symDataDict = { }

self.UniverseTickers = ["SOLUSDT", "ETHUSDT", "BNBUSDT", "ADAUSDT", "BTCUSDT"]

## More test tickers

# self.UniverseTickers = ["ANTUSDT","BATUSDT","BNBUSDT","BNTUSDT",

# "BTCUSDT", "BTGUSDT",

# "DAIUSDT","DASHUSDT","DGBUSDT",

# "EOSUSDT","ETCUSDT",

# "ETHUSDT","FUNUSDT",

# "IOTAUSDT","KNCUSDT","LRCUSDT",

# "LTCUSDT","MKRUSDT",

# "NEOUSDT","OMGUSDT",

# "PNTUSDT","QTUMUSDT","REQUSDT",

# "STORJUSDT","TRXUSDT","UTKUSDT","VETUSDT",

# "XLMUSDT","XMRUSDT",

# "XRPUSDT","XTZUSDT","XVGUSDT","ZECUSDT",

# "ZILUSDT","ZRXUSDT"]

universeSymbols = []

for symbol in self.UniverseTickers:

universeSymbols.append(Symbol.Create(symbol, SecurityType.Crypto, Market.Binance))

self.SetUniverseSelection(ManualUniverseSelectionModel(universeSymbols))

# --------------------

def InitAlgoParams(self):

self.emaSlowPeriod = int(self.GetParameter('emaSlowPeriod'))

self.emaFastPeriod = int(self.GetParameter('emaFastPeriod'))

self.mompPeriod = int(self.GetParameter('mompPeriod')) # used for momentum based weight

self.minimumVolPeriod = int(self.GetParameter('minimumVolPeriod')) # used for volume threshold

self.warmupPeriod = max(self.emaSlowPeriod, self.mompPeriod, self.minimumVolPeriod)

self.useMomWeight = (int(self.GetParameter("useMomWeight")) == 1)

self.maxExposurePct = float(self.GetParameter("maxExposurePct"))/100

self.rebalanceWeekly = (int(self.GetParameter("rebalanceWeekly")) == 1)

self.minimumVolume = int(self.GetParameter("minimumVolume"))

self.maxHoldings = int(self.GetParameter("maxHoldings"))

## Experimental:

## self.maxSecurityDrawDown = float(self.GetParameter("maxSecurityDrawDown"))

# --------------------

def InitAssets(self):

self.symbol = "BTCUSDT"

self.SetBrokerageModel(BrokerageName.Binance, AccountType.Cash)

self.SetAccountCurrency("USDT")

self.AddCrypto(self.symbol, Resolution.Daily)

self.EnableAutomaticIndicatorWarmUp = True

self.SetWarmUp(timedelta(self.warmupPeriod))

self.SelectedSymbolsAndWeights = {}

## Experimental:

## self.AddRiskManagement(MaximumUnrealizedProfitPercentPerSecurity(self.maxSecurityDrawDown))

# ------------------------

## Schedule routines

## ------------------------

def ScheduleRoutines(self):

## TODO:

## Check if rebalancing has happened in the last 7 days,

## If it has, do not rebalance again

if(self.rebalanceWeekly):

self.Schedule.On( self.DateRules.WeekStart(self.symbol),

self.TimeRules.AfterMarketOpen(self.symbol, 31),

self.RebalanceHoldings )

## Check if we are already holding the max # of open positions.

## TODO:

## When we start using limit orders, include pending holdings

## ------------------------------------------------------------

@property

def PortfolioAtCapacity(self):

numHoldings = len([x.Key for x in self.Portfolio if x.Value.Invested])

return numHoldings >= self.maxHoldings

## In the OnData Event handler, check for signals

## TODO:

## Test logic below for pending holdings

# pendingOrders = len( [x for x in self.Transactions.GetOpenOrders()

# if x.Direction == OrderDirection.Buy

# and x.Type == OrderType.Limit ] )

## Check for signals

## ------------------------------------------------

def OnData(self, dataSlice):

## loop through the symbols in the slice

for symbol in dataSlice.Keys:

## if we have this symbol in our data dictioary

if symbol in self.symDataDict:

symbolData = self.symDataDict[symbol]

## Update the symbol with the data slice data

symbolData.OnSymbolData(self.Securities[symbol].Price, dataSlice[symbol])

## If we're invested in this symbol, manage any open positions

if(self.Portfolio[symbolData.symbol.Value].Invested):

symbolData.ManageOpenPositions()

## otherwise, if we're not invested, check for entry signal

else:

## First check if we are at capacity for new positions.

## TODO:

## For Go-Live, note that the portfolio capacity may not be accurate while

## checking it inside this for-loop. It will be accurate after the positions

## have been open. IE: When the orders are actually filled.

if(not self.PortfolioAtCapacity):

if( symbolData.EntrySignalFired() ):

self.OpenNewPosition(symbolData.symbol)

## TODO:

## For Go-Live, call OnNewPositionOpened only after

## the order is actually filled

symbolData.OnNewPositionOpened()

## Logic to rebalance our portfolio of holdings.

## We will either rebalance with equal weighting,

## or assign weights based on momentum.

## ----------------------------------------------

def RebalanceHoldings(self):

## TODO:

try:

## Check if rebalancing has happened in the last 7 days,

## If it has, do not rebalance again

## -----------------------------------------------------

def RebalanceHoldings(self, rebalanceCurrHoldings=False):

# try:

if self.useMomWeight:

momentumSum = sum(self.symDataDict[symbol].momp.Current.Value for symbol in self.SelectedSymbolsAndWeights)

if (momentumSum == 0):

self.useMomWeight = False

for symbol in self.SelectedSymbolsAndWeights:

if self.useMomWeight:

momentumSum = sum(self.symDataDict[symbol].momp.Current.Value for symbol in self.SelectedSymbolsAndWeights)

symbolWeight = round((self.symDataDict[symbol].momp.Current.Value / momentumSum),4)

else:

symbolWeight = round(1/len(self.SelectedSymbolsAndWeights),4)

for symbol in self.SelectedSymbolsAndWeights:

self.SetWeightedHolding(symbol,symbolWeight)

if self.useMomWeight:

return

symbolWeight = round((self.symDataDict[symbol].momp.Current.Value / momentumSum),4)

else:

symbolWeight = round(1/len(self.SelectedSymbolsAndWeights),4)

## Truncate symbolweight decimal places

truncFactor = 10.0 ** 2

symbolWeight = math.trunc(symbolWeight * truncFactor) / truncFactor

self.SelectedSymbolsAndWeights[symbol] = symbolWeight

## TODO: Calculate order qty instead of using % setholdings

## https://github.com/QuantConnect/Lean/blob/master/Algorithm.Python/BasicTemplateCryptoAlgorithm.py

orderMsg = f"{symbol} | {round(symbolWeight*100,2)}% alloc. | price: {round(self.Securities[symbol].Close,2)}"

if(self.Portfolio[symbol].Invested):

orderMsg = f"[Re-Balancing] {orderMsg}"

else:

orderMsg = f"[NEW Addition] {orderMsg}"

self.SetHoldings(symbol, symbolWeight * self.maxExposurePct, tag=orderMsg)

except:

## Allocate the specified weight (pct) of the portfolio value to

self.Debug(f"Failed to rebalance")

## the specified symbol. This weight will first be adjusted to consider

## cost basis, whether the position is already open and has profit.

## We are doing this to solve the problem where re-balancing causes winners

## to reduce in position size.

## --------–--------–--------–--------–--------–--------–--------–--------–

def SetWeightedHolding(self,symbol,symbolWeight):

## Calculate the basis (the denominator) for rebalancing weights

## This is the sum of costs basis, plus uninvested cash

if( self.Portfolio.Invested ):

# numHoldings = len([x.Key for x in self.Portfolio if x.Value.Invested])

totalCostBasis = sum( [x.Value.HoldingsCost for x in self.Portfolio if x.Value.Invested] )

else:

totalCostBasis = 0.0

## it's okay if this includes cash reserved for pending orders

## because we have alread considered those orders in the symbolsAndWeights list

cashAvailable = self.Portfolio.CashBook["USDT"].Amount

weightingBasis = totalCostBasis + cashAvailable

amtToInvest = weightingBasis * symbolWeight

## if already invested, our adjusted weight needs to account for

## the profits gained, so we adjust the 'amt to invest' based on

## unrealized profit pct of the position.

if(self.Portfolio[symbol].Invested):

profitPct = self.Portfolio[symbol].UnrealizedProfitPercent

adjustedAmtToInvest = amtToInvest * (1 + profitPct)

adjustedWeight = adjustedAmtToInvest / self.Portfolio.TotalPortfolioValue

else:

adjustedWeight = amtToInvest / self.Portfolio.TotalPortfolioValue

symbolWeight = self.GetTruncatedValue(symbolWeight,3)

adjustedWeight = self.GetTruncatedValue(adjustedWeight,3)

self.SelectedSymbolsAndWeights[symbol] = adjustedWeight

## TODO: Calculate order qty instead of using % setholdings

## https://github.com/QuantConnect/Lean/blob/master/Algorithm.Python/BasicTemplateCryptoAlgorithm.py

orderMsg = f"{symbol} | {round(symbolWeight*100,2)}% alloc. ({round(adjustedWeight*100,2)}% adjusted) "

if(self.Portfolio[symbol].Invested):

orderMsg = f"[Re-Balancing] {orderMsg}"

else:

orderMsg = f"[NEW Addition] {orderMsg}"

self.SetHoldings(symbol, adjustedWeight * self.maxExposurePct, tag=orderMsg)

## Adding the symbol to our dictionary will ensure

## that it gets processed in the rebalancing routine

## -------------------------------------------------

def OpenNewPosition(self, symbol):

self.SelectedSymbolsAndWeights[symbol] = 0

self.RebalanceHoldings()

## Removing the symbol from our dictionary will ensure

## that it wont get processed in the rebalancing routine

## -----------------------------------------------------

def ExitPosition(self, symbol, exitMsg=""):

profitPct = round(self.Securities[symbol].Holdings.UnrealizedProfitPercent,2)

self.Liquidate(symbol, tag=f"SELL {symbol.Value} ({profitPct}% profit) [{exitMsg}]")

self.SelectedSymbolsAndWeights.pop(symbol)

## TODO:

## Before go-live, wait until liquidation has happened before rebalancing

## Perhaps Call RebalanceHoldings after an order event has occured.

self.RebalanceHoldings()

return

## Create new symboldata object and add to our dictionary

## ------------------------------------------------------

def OnSecuritiesChanged(self, changes):

for security in changes.AddedSecurities:

symbol = security.Symbol

if( symbol in self.UniverseTickers and \

symbol not in self.symDataDict.keys()):

self.symDataDict[symbol] = SymbolData(symbol, self)

def GetTruncatedValue(self, value, decPlaces):

truncFactor = 10.0 ** decPlaces

return math.trunc(value * truncFactor) / truncFactor

##################################

# SymbolData Class

##################################

class SymbolData():

def __init__(self, theSymbol, algo):

## Algo / Symbol / Price reference

self.algo = algo

self.symbol = theSymbol

self.lastPrice = 0

self.price = 0

## Initialize indicators

self.InitIndicators()

## ----------------------------------------

def InitIndicators(self):

self.indicators = { 'EMA_FAST' : self.algo.EMA(self.symbol,self.algo.emaFastPeriod,Resolution.Daily),

'EMA_SLOW' : self.algo.EMA(self.symbol,self.algo.emaSlowPeriod,Resolution.Daily),

'30DAY_VOL' : IndicatorExtensions.Times( self.algo.SMA(self.symbol,self.algo.minimumVolPeriod, Resolution.Daily, Field.Volume),

self.algo.SMA(self.symbol,self.algo.minimumVolPeriod, Resolution.Daily, Field.Close)),

'MOMP' : self.algo.MOMP(self.symbol,self.algo.mompPeriod,Resolution.Daily)}

## for easy reference from main algo

self.momp = self.indicators['MOMP']

for key, indicator in self.indicators.items():

self.algo.WarmUpIndicator(self.symbol, indicator, Resolution.Minute)

self.emaFastWindow = SmartRollingWindow("float", 2)

self.emaSlowWindow = SmartRollingWindow("float", 2)

self.lastPriceWindow = SmartRollingWindow("float", 2)

## ----------------------------------------

def OnSymbolData(self, lastKnownPrice, tradeBar):

self.lastPrice = lastKnownPrice

self.UpdateRollingWindows()

self.PlotCharts()

## ----------------------------------------

def UpdateRollingWindows(self):

self.emaFastWindow.Add(self.indicators['EMA_FAST'].Current.Value)

self.emaSlowWindow.Add(self.indicators['EMA_SLOW'].Current.Value)

self.lastPriceWindow.Add(self.lastPrice)

## ----------------------------------------

def IsReady(self):

return (self.indicators['EMA_FAST'].IsReady and self.indicators['EMA_SLOW'].IsReady \

and self.indicators['30DAY_VOL'].IsReady)

## ----------------------------------------

def MinimumVolTraded(self):

if( self.indicators['30DAY_VOL'].IsReady ):

dollarVolume = self.indicators['30DAY_VOL'].Current.Value

if( dollarVolume >= self.algo.minimumVolume ):

return True

return False

## ----------------------------------------

def EntrySignalFired(self):

if( self.IsReady() ):

if( self.MinimumVolTraded() ):

if( self.emaFastWindow.isAbove(self.emaSlowWindow) and \

self.lastPriceWindow.isAbove(self.emaFastWindow) ):

return True

return False

## ----------------------------------------

def ExitSignalFired(self):

if( self.IsReady() ):

if ( self.lastPriceWindow.isBelow(self.emaSlowWindow) or \

self.emaSlowWindow.isAbove(self.emaFastWindow) ):

return True

return False

## Logic to run immediately after a new position is opened.

## ---------------------------------------------------------

def OnNewPositionOpened(self):

# self.algo.Log(f"[BOUGHT {self.symbol.Value}] @ ${self.lastPrice:.2f}")

return

## Manage open positions if any. ie: close them, update stops, add to them, etc

## Called periodically, eg: from a scheduled routine

## TODO:

## Consilder also liquidating if volume or liquidity thresholds arent met

## -----------------------------------------------------------------------------

def ManageOpenPositions(self):

## if( not self.MinimumVolTraded() ):

## self.ExitPosition(exitMsg="No longer liquid")

## self.ExitPosition(exitMsg="Trading volume below threshold")

if(self.ExitSignalFired()):

self.ExitPosition(exitMsg="Exit Signal Fired")

# ----------------------------------------

def ExitPosition(self, exitMsg):

# self.algo.Log(f"[SELL {self.symbol.Value}] @ ${self.lastPrice:.2f}")

self.algo.ExitPosition(self.symbol, exitMsg)

# ----------------------------------------

def PlotCharts(self):

## To Plot charts, comment out the below

# self.algo.Plot(f"{self.symbol}-charts", "Price", self.lastPriceWindow[0])

# self.algo.Plot(f"{self.symbol}-charts", "EMA Fast", self.indicators['EMA_FAST'].Current.Value)

# self.algo.Plot(f"{self.symbol}-charts", "EMA Slow", self.indicators['EMA_SLOW'].Current.Value)

return

Diff salvati

Testo originale

Apri file

##########################################################################
# Inspired by @nitay-rabinovich at QuqntConnect
# https://www.quantconnect.com/forum/discussion/12768/share-kalman-filter-crossovers-for-crypto-and-smart-rollingwindows/p1/comment-38144
##########################################################################
#
# EMA Crossover In a Crypto Universe 
# ---------------------------------------------
# FOR EDUCATIONAL PURPOSES ONLY. DO NOT DEPLOY.
#
#
# Entry:
# -------
# Minimum volume threshold traded 
#         and
# Price > Fast Daily EMA 
#         and
# Fast Daily EMA > Slow Daily EMA
#
# Exit:
# ------
# Price < Slow Daily EMA 
#       or 
# Slow Daily EMA < Fast Daily EMA
#
# Additional Consideration:
# --------------------------
# Max exposure pct: Total % of available capital to trade with at any time
# Max holdings: Total # of positions that can be held simultaneously
# Rebalance Weekly: If false,  only rebalance when we add/remove positions  
# UseMomWeight: If true, rebalance w/momentum-based weights (top gainers=more weight)
#
#########################################################################

from SmartRollingWindow import *
class EMACrossoverUniverse(QCAlgorithm):
    
    ## 
    def Initialize(self):
        self.InitAlgoParams()
        self.InitAssets()
        self.InitUniverse()
        self.InitBacktestParams()        
        self.ScheduleRoutines()

## Set backtest params: dates, cash, etc. Called from Initialize().
    ## ----------------------------------------------------------------
    def InitBacktestParams(self):
        self.SetStartDate(2020, 1, 1)
        # self.SetEndDate(2019, 2, 1)  
        self.SetCash(100000)
        self.SetBenchmark(Symbol.Create("BTCUSDT", SecurityType.Crypto, Market.Binance))

def InitUniverse(self):
        self.UniverseSettings.Resolution = Resolution.Daily
        self.symDataDict = { }

self.UniverseTickers = ["SOLUSDT", "ETHUSDT", "BNBUSDT", "ADAUSDT", "BTCUSDT"]
        
        universeSymbols = []
        for symbol in self.UniverseTickers:

universeSymbols.append(Symbol.Create(symbol, SecurityType.Crypto, Market.Binance))
            
        self.SetUniverseSelection(ManualUniverseSelectionModel(universeSymbols))

# --------------------
    def InitAlgoParams(self):
        
        self.emaSlowPeriod    = int(self.GetParameter('emaSlowPeriod'))
        self.emaFastPeriod    = int(self.GetParameter('emaFastPeriod'))
        self.mompPeriod       = int(self.GetParameter('mompPeriod'))       # used for momentum based weight
        self.minimumVolPeriod = int(self.GetParameter('minimumVolPeriod')) # used for volume threshold
        self.warmupPeriod     = max(self.emaSlowPeriod, self.mompPeriod, self.minimumVolPeriod)
        
        self.useMomWeight     = (int(self.GetParameter("useMomWeight")) == 1)                 
        self.maxExposurePct   = float(self.GetParameter("maxExposurePct"))/100
        self.rebalanceWeekly  = (int(self.GetParameter("rebalanceWeekly")) == 1)

self.minimumVolume    = int(self.GetParameter("minimumVolume")) 
        self.maxHoldings      = int(self.GetParameter("maxHoldings")) 
        
        ## Experimental:
        ## self.maxSecurityDrawDown  = float(self.GetParameter("maxSecurityDrawDown"))
    
    # --------------------
    def InitAssets(self):    
        self.symbol = "BTCUSDT" 
        self.SetBrokerageModel(BrokerageName.Binance, AccountType.Cash)
        self.SetAccountCurrency("USDT")
        self.AddCrypto(self.symbol, Resolution.Daily)

self.EnableAutomaticIndicatorWarmUp = True
        self.SetWarmUp(timedelta(self.warmupPeriod))
        self.SelectedSymbolsAndWeights = {}
        
        ## Experimental:
        ## self.AddRiskManagement(MaximumUnrealizedProfitPercentPerSecurity(self.maxSecurityDrawDown))
    
    # ------------------------
    def ScheduleRoutines(self):
        if(self.rebalanceWeekly):
            self.Schedule.On( self.DateRules.WeekStart(self.symbol),
                              self.TimeRules.AfterMarketOpen(self.symbol, 31),
                              self.RebalanceHoldings )

## Check if we are already holding the max # of open positions.
    ## ------------------------------------------------------------
    @property
    def PortfolioAtCapacity(self):
        numHoldings = len([x.Key for x in self.Portfolio if x.Value.Invested])
        return numHoldings >= self.maxHoldings
        
        ## In the OnData Event handler, check for  signals
    ## ------------------------------------------------
    def OnData(self, dataSlice):

## loop through the symbols in the slice
        for symbol in dataSlice.Keys:

## if we have this symbol in our data dictioary
            if symbol in self.symDataDict:

symbolData = self.symDataDict[symbol]

## Update the symbol with the data slice data
                symbolData.OnSymbolData(self.Securities[symbol].Price, dataSlice[symbol])
                
                ## If we're invested in this symbol, manage any open positions
                if(self.Portfolio[symbolData.symbol.Value].Invested):
                    symbolData.ManageOpenPositions()
                
                ## otherwise, if we're not invested, check for entry signal
                else:

## First check if we are at capacity for new positions. 
                    ##
                    ## TODO: 
                    ## For Go-Live, note that the portfolio capacity may not be accurate while
                    ## checking it inside this for-loop. It will be accurate after the positions 
                    ## have been open. IE: When the orders are actually filled.                    
                    if(not self.PortfolioAtCapacity):            
                        if( symbolData.EntrySignalFired() ):
                            self.OpenNewPosition(symbolData.symbol)
                    
                            ## TODO: 
                            ## For Go-Live, call OnNewPositionOpened only after 
                            ## the order is actually filled
                            symbolData.OnNewPositionOpened()
       
    ## Logic to rebalance our portfolio of holdings. 
    ## We will either rebalance with equal weighting, 
    ## or assign weights based on momentum. 
    ## ----------------------------------------------
    def RebalanceHoldings(self):
        try:
            
            if self.useMomWeight:
                momentumSum = sum(self.symDataDict[symbol].momp.Current.Value for symbol in self.SelectedSymbolsAndWeights)
            
            for symbol in self.SelectedSymbolsAndWeights:

if self.useMomWeight:
                    symbolWeight = round((self.symDataDict[symbol].momp.Current.Value / momentumSum),4) 
                else:
                    symbolWeight = round(1/len(self.SelectedSymbolsAndWeights),4)
                    
                ## Truncate symbolweight decimal places
                truncFactor  = 10.0 ** 2
                symbolWeight = math.trunc(symbolWeight * truncFactor) / truncFactor
                
                self.SelectedSymbolsAndWeights[symbol] = symbolWeight

## TODO: Calculate order qty instead of using % setholdings
                ## https://github.com/QuantConnect/Lean/blob/master/Algorithm.Python/BasicTemplateCryptoAlgorithm.py
                orderMsg = f"{symbol} | {round(symbolWeight*100,2)}% alloc. | price: {round(self.Securities[symbol].Close,2)}"
                
                if(self.Portfolio[symbol].Invested): 
                    orderMsg = f"[Re-Balancing] {orderMsg}"
                else:
                    orderMsg = f"[NEW Addition] {orderMsg}"
                    
                self.SetHoldings(symbol, symbolWeight * self.maxExposurePct, tag=orderMsg)

except:
            self.Debug(f"Failed to rebalance")

## Adding the symbol to our dictionary will ensure
    ## that it gets processed in the rebalancing routine
    ## -------------------------------------------------
    def OpenNewPosition(self, symbol):
        self.SelectedSymbolsAndWeights[symbol] = 0
        self.RebalanceHoldings()    
                    
    ## Removing the symbol from our dictionary will ensure
    ## that it wont get processed in the rebalancing routine
    ## -----------------------------------------------------
    def ExitPosition(self, symbol, exitMsg=""):
        profitPct   = round(self.Securities[symbol].Holdings.UnrealizedProfitPercent,2)
        self.Liquidate(symbol, tag=f"SELL {symbol.Value} ({profitPct}% profit) [{exitMsg}]")
        self.SelectedSymbolsAndWeights.pop(symbol)
        self.RebalanceHoldings()
        return

## Create new symboldata object and add to our dictionary
    ## ------------------------------------------------------
    def OnSecuritiesChanged(self, changes):
        for security in changes.AddedSecurities:
            symbol  = security.Symbol
            if( symbol in self.UniverseTickers and \
                symbol not in self.symDataDict.keys()):
                self.symDataDict[symbol] = SymbolData(symbol, self)

##################################
# SymbolData Class
##################################
class SymbolData():
    
    def __init__(self, theSymbol, algo):

## Algo / Symbol / Price reference
        self.algo       = algo
        self.symbol     = theSymbol
        self.lastPrice  = 0
        self.price      = 0

## Initialize indicators
        self.InitIndicators()
            
            
    ## ----------------------------------------
    def InitIndicators(self):

self.indicators = { 'EMA_FAST'  : self.algo.EMA(self.symbol,self.algo.emaFastPeriod,Resolution.Daily),
                            'EMA_SLOW'  : self.algo.EMA(self.symbol,self.algo.emaSlowPeriod,Resolution.Daily),
                            '30DAY_VOL' : IndicatorExtensions.Times( self.algo.SMA(self.symbol,self.algo.minimumVolPeriod, Resolution.Daily, Field.Volume), 
                                                                     self.algo.SMA(self.symbol,self.algo.minimumVolPeriod, Resolution.Daily, Field.Close)),
                            'MOMP'      : self.algo.MOMP(self.symbol,self.algo.mompPeriod,Resolution.Daily)}
        
        ## for easy reference from main algo
        self.momp = self.indicators['MOMP']
                            
        for key, indicator in self.indicators.items():
            self.algo.WarmUpIndicator(self.symbol, indicator, Resolution.Minute)

self.emaFastWindow  = SmartRollingWindow("float", 2)
        self.emaSlowWindow  = SmartRollingWindow("float", 2)
        self.lastPriceWindow    = SmartRollingWindow("float", 2)
        
    ## ----------------------------------------
    def OnSymbolData(self, lastKnownPrice, tradeBar):
        self.lastPrice = lastKnownPrice
        self.UpdateRollingWindows()
        self.PlotCharts()
        
    ## ----------------------------------------
    def UpdateRollingWindows(self):
        self.emaFastWindow.Add(self.indicators['EMA_FAST'].Current.Value)
        self.emaSlowWindow.Add(self.indicators['EMA_SLOW'].Current.Value)
        self.lastPriceWindow.Add(self.lastPrice)
        
    ## ----------------------------------------
    def IsReady(self):
        return (self.indicators['EMA_FAST'].IsReady and self.indicators['EMA_SLOW'].IsReady \
            and self.indicators['30DAY_VOL'].IsReady)

## ----------------------------------------
    def MinimumVolTraded(self):
        if( self.indicators['30DAY_VOL'].IsReady ):
            dollarVolume = self.indicators['30DAY_VOL'].Current.Value
            if( dollarVolume >= self.algo.minimumVolume ):
                return True
                
        return False
        
    ## ----------------------------------------
    def EntrySignalFired(self):
        if( self.IsReady() ):
            if( self.MinimumVolTraded() ):
                if( self.emaFastWindow.isAbove(self.emaSlowWindow) and \
                    self.lastPriceWindow.isAbove(self.emaFastWindow) ):
                    return True
        
        return False

## ----------------------------------------
    def ExitSignalFired(self):
        if( self.IsReady() ):
            if ( self.lastPriceWindow.isBelow(self.emaSlowWindow) or \
                 self.emaSlowWindow.isAbove(self.emaFastWindow) ):
                return True
        
        return False

## Logic to run immediately after a new position is opened.
    ## ---------------------------------------------------------
    def OnNewPositionOpened(self):            
        # self.algo.Log(f"[BOUGHT {self.symbol.Value}] @ ${self.lastPrice:.2f}")    
        return
    
    ## Manage open positions if any. ie: close them, update stops, add to them, etc
    ## Called periodically, eg: from a scheduled routine
    ##
    ## TODO:
    ## Consilder also liquidating if volume or liquidity thresholds arent met
    ## -----------------------------------------------------------------------------
    def ManageOpenPositions(self):

## if( not self.MinimumVolTraded() ):
        ##     self.ExitPosition(exitMsg="No longer liquid")
        
        if(self.ExitSignalFired()):        
            self.ExitPosition(exitMsg="Exit Signal Fired")
        
        
    # ----------------------------------------
    def ExitPosition(self, exitMsg): 
        # self.algo.Log(f"[SELL {self.symbol.Value}] @ ${self.lastPrice:.2f}") 
        self.algo.ExitPosition(self.symbol, exitMsg)

# ----------------------------------------
    def PlotCharts(self):    
    
        ## To Plot charts, comment out the below
        
        # self.algo.Plot(f"{self.symbol}-charts", "Price", self.lastPriceWindow[0])        
        # self.algo.Plot(f"{self.symbol}-charts", "EMA Fast", self.indicators['EMA_FAST'].Current.Value)            
        # self.algo.Plot(f"{self.symbol}-charts", "EMA Slow", self.indicators['EMA_SLOW'].Current.Value)            
        
        return

Testo modificato

Apri file

def InitUniverse(self):
        self.UniverseSettings.Resolution = Resolution.Daily
        self.symDataDict = { }

self.UniverseTickers = ["SOLUSDT", "ETHUSDT", "BNBUSDT", "ADAUSDT", "BTCUSDT"]

## More test tickers
        ## 
        # self.UniverseTickers = ["ANTUSDT","BATUSDT","BNBUSDT","BNTUSDT",
        #                         "BTCUSDT", "BTGUSDT",
        #                         "DAIUSDT","DASHUSDT","DGBUSDT",
        #                         "EOSUSDT","ETCUSDT",
        #                         "ETHUSDT","FUNUSDT",
        #                         "IOTAUSDT","KNCUSDT","LRCUSDT",
        #                         "LTCUSDT","MKRUSDT",
        #                         "NEOUSDT","OMGUSDT",
        #                         "PNTUSDT","QTUMUSDT","REQUSDT",
        #                         "STORJUSDT","TRXUSDT","UTKUSDT","VETUSDT",
        #                         "XLMUSDT","XMRUSDT",
        #                         "XRPUSDT","XTZUSDT","XVGUSDT","ZECUSDT",
        #                         "ZILUSDT","ZRXUSDT"]
        
        universeSymbols = []
        for symbol in self.UniverseTickers:

universeSymbols.append(Symbol.Create(symbol, SecurityType.Crypto, Market.Binance))
            
        self.SetUniverseSelection(ManualUniverseSelectionModel(universeSymbols))

self.EnableAutomaticIndicatorWarmUp = True
        self.SetWarmUp(timedelta(self.warmupPeriod))
        self.SelectedSymbolsAndWeights = {}
        
        ## Experimental:
        ## self.AddRiskManagement(MaximumUnrealizedProfitPercentPerSecurity(self.maxSecurityDrawDown))
    
    ## Schedule routines
    ## ------------------------
    def ScheduleRoutines(self):
        
        ## TODO: 
        ## Check if rebalancing has happened in the last 7 days, 
        ## If it has, do not rebalance again 
        if(self.rebalanceWeekly):
            self.Schedule.On( self.DateRules.WeekStart(self.symbol),
                              self.TimeRules.AfterMarketOpen(self.symbol, 31),
                              self.RebalanceHoldings )

##
    ## Check if we are already holding the max # of open positions.
    ## TODO:
    ## When we start using limit orders, include pending holdings
    ## ------------------------------------------------------------
    @property
    def PortfolioAtCapacity(self):
        numHoldings = len([x.Key for x in self.Portfolio if x.Value.Invested])
        return numHoldings >= self.maxHoldings
    
    ## TODO:
    ## Test logic below for pending holdings
    # pendingOrders  = len( [x for x in self.Transactions.GetOpenOrders()
    #                         if x.Direction == OrderDirection.Buy
    #                             and x.Type == OrderType.Limit ] )
    
                             
    ## Check for  signals
    ## ------------------------------------------------
    def OnData(self, dataSlice):

## loop through the symbols in the slice
        for symbol in dataSlice.Keys:

## if we have this symbol in our data dictioary
            if symbol in self.symDataDict:

symbolData = self.symDataDict[symbol]

## First check if we are at capacity for new positions. 
                    ##
                    ## TODO: 
                    ## For Go-Live, note that the portfolio capacity may not be accurate while
                    ## checking it inside this for-loop. It will be accurate after the positions 
                    ## have been open. IE: When the orders are actually filled.                    
                    if(not self.PortfolioAtCapacity):            
                        if( symbolData.EntrySignalFired() ):
                            self.OpenNewPosition(symbolData.symbol)
                    
                            ## TODO: 
                            ## For Go-Live, call OnNewPositionOpened only after 
                            ## the order is actually filled
                            symbolData.OnNewPositionOpened()
      
       
    ## Logic to rebalance our portfolio of holdings. 
    ## We will either rebalance with equal weighting, 
    ## or assign weights based on momentum. 
    ##
    ## TODO: 
    ## Check if rebalancing has happened in the last 7 days, 
    ## If it has, do not rebalance again    
    ## -----------------------------------------------------
    def RebalanceHoldings(self, rebalanceCurrHoldings=False):
        # try:
            
        if self.useMomWeight:
            momentumSum = sum(self.symDataDict[symbol].momp.Current.Value for symbol in self.SelectedSymbolsAndWeights)
            if (momentumSum == 0):
                self.useMomWeight = False
            
                
        for symbol in self.SelectedSymbolsAndWeights:
            if self.useMomWeight:
                symbolWeight = round((self.symDataDict[symbol].momp.Current.Value / momentumSum),4) 
            else:
                symbolWeight = round(1/len(self.SelectedSymbolsAndWeights),4)
            
            self.SetWeightedHolding(symbol,symbolWeight)

return

## Allocate the specified weight (pct) of the portfolio value to 
    ## the specified symbol. This weight will first be adjusted to consider
    ## cost basis, whether the position is already open and has profit.
    ## We are doing this to solve the problem where re-balancing causes winners
    ## to reduce in position size.
    ## --------–--------–--------–--------–--------–--------–--------–--------–
    def SetWeightedHolding(self,symbol,symbolWeight):

## Calculate the basis (the denominator) for rebalancing weights
        ## This is the sum of costs basis, plus uninvested cash

if( self.Portfolio.Invested ):
            # numHoldings = len([x.Key for x in self.Portfolio if x.Value.Invested])
            totalCostBasis = sum( [x.Value.HoldingsCost for x in self.Portfolio if x.Value.Invested] )
        else: 
            totalCostBasis = 0.0

## it's okay if this includes cash reserved for pending orders
        ## because we have alread considered those orders in the symbolsAndWeights list
        cashAvailable  = self.Portfolio.CashBook["USDT"].Amount
        weightingBasis = totalCostBasis + cashAvailable
        amtToInvest    = weightingBasis * symbolWeight

## if already invested, our adjusted weight needs to account for
        ## the profits gained, so we adjust the 'amt to invest' based on 
        ## unrealized profit pct of the position.

if(self.Portfolio[symbol].Invested):
            profitPct           = self.Portfolio[symbol].UnrealizedProfitPercent
            adjustedAmtToInvest = amtToInvest * (1 + profitPct)
            adjustedWeight      = adjustedAmtToInvest / self.Portfolio.TotalPortfolioValue
            
        else:
            adjustedWeight = amtToInvest / self.Portfolio.TotalPortfolioValue
            
        symbolWeight   = self.GetTruncatedValue(symbolWeight,3)
        adjustedWeight = self.GetTruncatedValue(adjustedWeight,3)

self.SelectedSymbolsAndWeights[symbol] = adjustedWeight

## TODO: Calculate order qty instead of using % setholdings
        ## https://github.com/QuantConnect/Lean/blob/master/Algorithm.Python/BasicTemplateCryptoAlgorithm.py
        orderMsg = f"{symbol} | {round(symbolWeight*100,2)}% alloc. ({round(adjustedWeight*100,2)}% adjusted) "
        
        if(self.Portfolio[symbol].Invested): 
            orderMsg = f"[Re-Balancing] {orderMsg}"
        else:
            orderMsg = f"[NEW Addition] {orderMsg}"
            
        self.SetHoldings(symbol, adjustedWeight * self.maxExposurePct, tag=orderMsg)

def GetTruncatedValue(self, value, decPlaces):
        truncFactor  = 10.0 ** decPlaces
        return math.trunc(value * truncFactor) / truncFactor
    
##################################
# SymbolData Class
##################################
class SymbolData():
    
    def __init__(self, theSymbol, algo):

## Algo / Symbol / Price reference
        self.algo       = algo
        self.symbol     = theSymbol
        self.lastPrice  = 0
        self.price      = 0

## Initialize indicators
        self.InitIndicators()
            
            
    ## ----------------------------------------
    def InitIndicators(self):

## if( not self.MinimumVolTraded() ):
        ##     self.ExitPosition(exitMsg="Trading volume below threshold")
        
        if(self.ExitSignalFired()):        
            self.ExitPosition(exitMsg="Exit Signal Fired")
        
        
    # ----------------------------------------
    def ExitPosition(self, exitMsg): 
        # self.algo.Log(f"[SELL {self.symbol.Value}] @ ${self.lastPrice:.2f}") 
        self.algo.ExitPosition(self.symbol, exitMsg)

# ----------------------------------------
    def PlotCharts(self):    
    
        ## To Plot charts, comment out the below
        # self.algo.Plot(f"{self.symbol}-charts", "Price", self.lastPriceWindow[0])        
        # self.algo.Plot(f"{self.symbol}-charts", "EMA Fast", self.indicators['EMA_FAST'].Current.Value)            
        # self.algo.Plot(f"{self.symbol}-charts", "EMA Slow", self.indicators['EMA_SLOW'].Current.Value)            
        
        return