Backtest

Overall Statistics
Total Orders 957 Average Win 0.75% Average Loss -0.68% Compounding Annual Return 10.139% Drawdown 25.000% Expectancy 0.122 Start Equity 1000000 End Equity 1434577.72 Net Profit 43.458% Sharpe Ratio 0.331 Sortino Ratio 0.286 Probabilistic Sharpe Ratio 12.571% Loss Rate 46% Win Rate 54% Profit-Loss Ratio 1.09 Alpha 0.017 Beta 0.488 Annual Standard Deviation 0.157 Annual Variance 0.024 Information Ratio -0.129 Tracking Error 0.158 Treynor Ratio 0.106 Total Fees $22743.32 Estimated Strategy Capacity $29000000.00 Lowest Capacity Asset INTC R735QTJ8XC9X Portfolio Turnover 22.79%

# region imports
import openai
import pytz
import requests
import json
from datetime import timedelta
from AlgorithmImports import *
# endregion

class JumpingTanManatee(QCAlgorithm):

    def Initialize(self):
        self.SetCash(1000000)
        self.SetStartDate(2021,1,1)
        self.rebalanceTime = datetime.min  
        self.UniverseSettings.Resolution = Resolution.Daily
        self.SetPortfolioConstruction(EqualWeightingPortfolioConstructionModel(portfolioBias=PortfolioBias.Long))
        self.AddRiskManagement(NullRiskManagementModel())
        self.SetExecution(ImmediateExecutionModel())
        self.lastDateSaved = date.min
        self.SetWarmUp(60)
        self.Settings.FreePortfolioValuePercentage = 0.10 #cash buffer to 10%


        self.index = self.AddEquity("IWV", Resolution.Daily)
        self.AddUniverse(self.Universe.ETF(self.index.Symbol, self.UniverseSettings, self.Filter))

    def Filter(self, constituents: List[ETFConstituentData]) -> List[Symbol]:
        if self.Time < self.rebalanceTime:
            return Universe.Unchanged
        self.rebalanceTime = self.Time + timedelta(days=30)

        active_symbols   = [s.Symbol for s in self.ActiveSecurities.Values]
        incoming_symbols = sorted([c  for c in constituents if c.Weight], key=lambda c: c.Weight, reverse=True)[:50]
        incoming_symbols = [c.Symbol for c in incoming_symbols]
        grouped =  sorted(list(set(active_symbols)) + list(incoming_symbols))
        return list(set(active_symbols + incoming_symbols))

    def OnSecuritiesChanged(self, changes):
        for security in changes.AddedSecurities: 
            security.STD = self.STD(security.Symbol, 22)
            security.ATR = self.ATR(security.Symbol, 22)
            security.EMA = self.EMA(security.Symbol, 22)
            identity = self.Identity(security.Symbol)
            identity.Window.Size = 15
            security.History = identity

    def OnData(self, slice: Slice) -> None: 
        if not self.index.History.Window.IsReady:
            return       

        trade_securities = []
        signals = {'symbol': '', 'sentiment': [] }

        for security in self.ActiveSecurities.Values: 
            if not security.History.Window.IsReady:
                continue
    
            if security.Symbol == self.index.Symbol or not hasattr(security, 'STD'):
                continue

            threshold   = 2.5
            std = threshold*security.STD.Current.Value
            avg = security.EMA.Current.Value

            if self.index.History[5] is not None and security.History[5] is not None and security.History[5].Value != 0:
                factor = self.index.History[5].Value/security.History[5].Value
                scaled_stockprice = factor*security.Close
                divergence = 1 - scaled_stockprice/self.index.Close
                divergence = abs(divergence)

                if divergence > 0.03 and divergence < 0.20:
                    if (security.Close < (-std+avg)):
                        trade_securities.append(Insight.Price(security.Symbol, Resolution.Daily, 2, InsightDirection.Up))  
        
        if self.IsWarmingUp:
            return

        self.EmitInsights(trade_securities)