| Overall Statistics |
|
Total Trades 1139 Average Win 0.47% Average Loss -0.26% Compounding Annual Return -10.248% Drawdown 29.800% Expectancy -0.068 Net Profit -12.918% Sharpe Ratio -0.334 Probabilistic Sharpe Ratio 4.598% Loss Rate 67% Win Rate 33% Profit-Loss Ratio 1.81 Alpha -0.075 Beta 0.06 Annual Standard Deviation 0.2 Annual Variance 0.04 Information Ratio -0.851 Tracking Error 0.247 Treynor Ratio -1.111 Total Fees $1843278.78 Estimated Strategy Capacity $53000.00 |
# from BankingIndustryStocks import BankingIndustryStocks
from custom_sp500 import QC500UniverseSelectionModel_V2
from datetime import datetime, timedelta
from numpy import sum
from gid_func import *
from collections import deque
class CalibratedParticleAtmosphericScrubbers(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2010, 1, 1) # Set Start Date
self.SetEndDate(2011, 4, 12) # Set End Date
self.SetCash(21000000) # Set Strategy Cash
self.UniverseSettings.Resolution = Resolution.Daily
# self.SetPortfolioConstruction( EqualWeightingPortfolioConstructionModel() )
# self.SetPortfolioConstruction(EqualWeightingPortfolioConstructionModel(self.DateRules.Every(DayOfWeek.Wednesday)))
#https://www.quantconnect.com/forum/discussion/4361/rebalancing-monthly-using-portfolioconstructionmodel-algorithm-framework/p1
#https://github.com/QuantConnect/Lean/blob/master/Algorithm.Python/PortfolioRebalanceOnCustomFuncRegressionAlgorithm.py#L50
self.SetPortfolioConstruction(EqualWeightingPortfolioConstructionModel(self.DateRules.MonthStart()))#https://www.quantconnect.com/docs/algorithm-reference/scheduled-events
self.SetExecution( ImmediateExecutionModel() )
self.SetRiskManagement(MaximumDrawdownPercentPortfolio())
# Add QC500 Universe
self.SetUniverseSelection(QC500UniverseSelectionModel_V2())
# self.SetUniverseSelection(BankingIndustryStocks())
MAX_REDRAW = self.GetParameter("MAX_REDRAW_LENGTH")
BLUE_DEG = self.GetParameter("blue_deg")
GREEN_DEG = self.GetParameter("green_deg")
self.SetAlpha(Grid_indicator("Grid", MAX_REDRAW, BLUE_DEG, GREEN_DEG))
class Grid_indicator(AlphaModel):
def __init__(self, name, MAX_REDRAW, BLUE_DEG, GREEN_DEG):
self.symbolDataBySymbol = {}
self.HIGH_HIST = {}
self.LOW_HIST = {}
self.TIME_HIST = {}
self.CHART_HIGH = {}
self.CHART_LOW = {}
self.SOFT_HIST_HIGH = {}
self.SOFT_HIST_LOW = {}
self.SOFT_CHART_HIGH = {}
self.SOFT_CHART_LOW = {}
self.Name = name
self.Time = datetime.min
self.Value = 0
self.dfdf = {}
self.mean_HIGH = {}
self.STD_four = {}
self.High_value = {}
self.MAX_REDRAW = float(MAX_REDRAW) # Since the Algorithm Parameter is a string we must convert it to a float
self.BLUE_DEG = float(BLUE_DEG)
self.GREEN_DEG = float(GREEN_DEG)
def Update(self, algorithm, data):
insights = []
for symbol, item in self.symbolDataBySymbol.items():
'''
Error checking
past 122 high bars
6 standard diviations of the mean + mean
if high
'''
if data.ContainsKey(item.Symbol) and data[item.Symbol] is not None :
self.SOFT_HIST_HIGH[item.Symbol] = list(self.SOFT_HIST_HIGH[item.Symbol])
self.SOFT_HIST_LOW[item.Symbol] = list(self.SOFT_HIST_LOW[item.Symbol])
self.mean_HIGH[item.Symbol] = round(np.mean(self.SOFT_HIST_HIGH[item.Symbol][-122:]),2)
self.STD_four[item.Symbol] = round((np.std(self.SOFT_HIST_HIGH[item.Symbol][-122:]) * 6),4)
if data[item.Symbol].High > self.mean_HIGH[item.Symbol] + self.STD_four[item.Symbol]:
# print("here")
algorithm.Debug(f"{item.Symbol.Value} SKIP THIS DAY!!!!!!!!!!!!!!! High_value: {data[item.Symbol].High}, mean_{item.Symbol.Value}: {self.mean_HIGH[item.Symbol]} STD_{item.Symbol.Value}_4: {self.STD_four[item.Symbol]}")
self.SOFT_HIST_HIGH[item.Symbol].append(data[item.Symbol].High)
self.SOFT_HIST_LOW[item.Symbol].append(data[item.Symbol].Low)
continue
# return insights
#can redraw if in trade
#close out trade first
if item.BASE_series_index >= self.MAX_REDRAW and not algorithm.Portfolio[item.Symbol].Invested:#
for mod in list(self.symbolDataBySymbol.values()):
if symbol == mod.Symbol and data.ContainsKey(item.Symbol) and data[item.Symbol] is not None:
algorithm.Debug(f"{mod.Symbol.Value} HIT max grid length {self.MAX_REDRAW} for BASE_series_index of {item.BASE_series_index}, HARD RESET HERE")
self.symbolDataBySymbol[item.Symbol].PAUSED = True
data_IN = {'Open' : data[item.Symbol].Open,
'High' : data[item.Symbol].High,
'Low' : data[item.Symbol].Low,
'Close' : data[item.Symbol].Close,
'EndTime': algorithm.Time
}
algorithm.Debug(data_IN)
##This resets the dict with the latest grid for the specific symbol
self.symbolDataBySymbol[item.Symbol] = SymbolData(item.Symbol, data_IN, algorithm, self.Name, self.BLUE_DEG, self.GREEN_DEG)
self.symbolDataBySymbol[item.Symbol].REBUILD_CHECK = data[item.Symbol].High
self.symbolDataBySymbol[item.Symbol].HAS_RESET = True
self.symbolDataBySymbol[item.Symbol].HAS_BEEN_STOPPED_OUT = False
#Set CONFIRMING_GRID to true
#This means the next three bars are used to see if the lows are greater than blue 0
self.symbolDataBySymbol[item.Symbol].CONFIRMING_GRID = True
#This gets the low of the bar that the potential OP is set on
#This is used to compare if any of HIGH's the next 3 bars are less than this VALUE
#IF a high is LOWER then the grid fails
self.symbolDataBySymbol[item.Symbol].LOW_OF_RANGE = data[item.Symbol].Low
item.BASE_series_index = 0
item.series_index = 0
item.UNDERNEATH_LIST = []
'''
crude 5% stop loss
'''
if data.ContainsKey(item.Symbol) and data[item.Symbol] is not None :
if round((algorithm.Portfolio[item.Symbol].UnrealizedProfitPercent*100),3) < -5:
algorithm.EmitInsights(Insight.Price(item.Symbol, timedelta(100000), InsightDirection.Flat))
# algorithm.SetHoldings(item.Symbol, 0, False, f"{item.Symbol.Value} THIS IS WHERE WE WOULD STOP OUT BASED ON RISK_MANAGMENT ALONE")
item.TRADE_INITIATED = True
# return insights
continue
'''
soft reset logic
SOFT reset
- OP set at chart low
'''
if data.ContainsKey(item.Symbol) and data[item.Symbol] is not None:
if item.TRADE_INITIATED and data.ContainsKey(item.Symbol) and data[item.Symbol] is not None:
algorithm.Debug(f"{item.Symbol.Value} LastTradeProfit: ${round(algorithm.Portfolio[item.Symbol].LastTradeProfit,2)}")
item.TRADE_INITIATED = False
#make lists of the arrays
self.SOFT_HIST_HIGH[item.Symbol] = list(self.SOFT_HIST_HIGH[item.Symbol])
self.SOFT_HIST_LOW[item.Symbol] = list(self.SOFT_HIST_LOW[item.Symbol])
self.SOFT_HIST_HIGH[item.Symbol].append(data[item.Symbol].High)
self.SOFT_HIST_LOW[item.Symbol].append(data[item.Symbol].Low)
self.SOFT_CHART_HIGH[item.Symbol] = max(self.SOFT_HIST_HIGH[item.Symbol][-12:])
self.SOFT_CHART_LOW[item.Symbol] = min(self.SOFT_HIST_LOW[item.Symbol][-12:])
self.HIGH_HIST[item.Symbol].append(data[item.Symbol].High)
self.LOW_HIST[item.Symbol].append(data[item.Symbol].Low)
self.TIME_HIST[item.Symbol].append(algorithm.Time)
self.CHART_HIGH[item.Symbol] = max(self.HIGH_HIST[item.Symbol])
self.CHART_LOW[item.Symbol] = min(self.LOW_HIST[item.Symbol])
algorithm.Debug(f"{item.Symbol.Value} 12 day Low: {round(self.SOFT_CHART_LOW[item.Symbol],2)} Current Low: {round(data[item.Symbol].Low,2)} Current Open: {round(data[item.Symbol].Open,2)} blue_zero @ Index: {item.series_index}: {round(item.blue_zero.values[item.series_index][0],2)}")
#Here is where we check if the low of the stock is lower than the 12 day low
#if so then set grid
if data[item.Symbol].Low <= self.SOFT_CHART_LOW[item.Symbol] :
algorithm.Debug(f"12 day LOW for {item.Symbol.Value}: {round(self.SOFT_CHART_LOW[item.Symbol],2)}: is today ")
algorithm.Debug(f"Setting the grid for {item.Symbol.Value}: {round(self.SOFT_CHART_LOW[item.Symbol],2)}")
data_IN = { 'Open' : data[item.Symbol].Open,
'High' : data[item.Symbol].High,
'Low' : data[item.Symbol].Low,
'Close' : data[item.Symbol].Close,
'EndTime': algorithm.Time
}
algorithm.Debug(data_IN)
##This resets the dict with the latest grid for the specific symbol
self.symbolDataBySymbol[item.Symbol] = SymbolData(item.Symbol, data_IN, algorithm, self.Name, self.BLUE_DEG, self.GREEN_DEG)
self.symbolDataBySymbol[item.Symbol].REBUILD_CHECK = data[item.Symbol].High
self.symbolDataBySymbol[item.Symbol].HAS_RESET = True
self.symbolDataBySymbol[item.Symbol].HAS_BEEN_STOPPED_OUT = False
#This gets the low of the bar that the potential OP is set on
#This is used to compare if any of HIGH's the next 3 bars are less than this VALUE
#IF a high is LOWER then the grid fails
self.symbolDataBySymbol[item.Symbol].LOW_OF_RANGE = data[item.Symbol].Low
#Set CONFIRMING_GRID to true
#This means the next three bars are used to see if the lows are greater than blue 0
self.symbolDataBySymbol[item.Symbol].CONFIRMING_GRID = True
#NOW JUST COPIED OVER STUFF FROM MAIN RUN, SO AS TO NOT MISS THE DATA ADDITIONS
self.HIGH_HIST[item.Symbol].append(data[item.Symbol].High)
self.LOW_HIST[item.Symbol].append(data[item.Symbol].Low)
self.TIME_HIST[item.Symbol].append(algorithm.Time)
self.CHART_HIGH[item.Symbol] = max(self.HIGH_HIST[item.Symbol])
self.CHART_LOW[item.Symbol] = min(self.LOW_HIST[item.Symbol])
item.BASE_series_index = 0
item.series_index = 0
item.UNDERNEATH_LIST = []
'''
HERE is where we do the confirming
- 3 bars go by which must be above the BLUE angle 0
'''
if data[item.Symbol].High >= self.symbolDataBySymbol[item.Symbol].LOW_OF_RANGE and self.symbolDataBySymbol[item.Symbol].PRINTED_CONFORMATION == False:
self.symbolDataBySymbol[item.Symbol].CONFIRMED_DAYS += 1
algorithm.Debug(f"{item.Symbol.Value}: High {round(data[item.Symbol].High, 2)} is ABOVE LOW_OF_RANGE {round(self.symbolDataBySymbol[item.Symbol].LOW_OF_RANGE, 2)} PASS: {self.symbolDataBySymbol[item.Symbol].CONFIRMED_DAYS}")
if self.symbolDataBySymbol[item.Symbol].CONFIRMED_DAYS == 2:
self.symbolDataBySymbol[item.Symbol].SECOND_BAR_HIGH = round(data[item.Symbol].High,2)
algorithm.Debug(f"{item.Symbol.Value} GET HIGH {round(data[item.Symbol].High,2)} OF THIS BAR (2nd) TO USE WHEN COMPARING TO 3rd BAR")
if data[item.Symbol].High < self.symbolDataBySymbol[item.Symbol].LOW_OF_RANGE:
algorithm.Debug(f"{item.Symbol.Value}: High {round(data[item.Symbol].High, 2)} is BELOW LOW_OF_RANGE {round(self.symbolDataBySymbol[item.Symbol].LOW_OF_RANGE, 2)} FAIL: REDRAW GRID WITH LOWEST LOW SINCE OP")
self.symbolDataBySymbol[item.Symbol].CONFIRMING_GRID = False
data_IN = {'Open' : data[item.Symbol].Open,
'High' : data[item.Symbol].High,
'Low' : data[item.Symbol].Low,
'Close' : data[item.Symbol].Close,
'EndTime': algorithm.Time
}
algorithm.Debug(data_IN)
##This resets the dict with the latest grid for the specific symbol
self.symbolDataBySymbol[item.Symbol] = SymbolData(item.Symbol, data_IN, algorithm, self.Name, self.BLUE_DEG, self.GREEN_DEG)
self.symbolDataBySymbol[item.Symbol].REBUILD_CHECK = data[item.Symbol].High
self.symbolDataBySymbol[item.Symbol].HAS_RESET = True
self.symbolDataBySymbol[item.Symbol].HAS_BEEN_STOPPED_OUT = False
#This gets the low of the bar that the potential OP is set on
#This is used to compare if any of HIGH's the next 3 bars are less than this VALUE
#IF a high is LOWER then the grid fails
self.symbolDataBySymbol[item.Symbol].LOW_OF_RANGE = data[symbol].Low
#Set CONFIRMING_GRID to true
#This means the next three bars are used to see if the lows are greater than blue 0
self.symbolDataBySymbol[item.Symbol].CONFIRMING_GRID = True
#NOW JUST COPIED OVER STUFF FROM MAIN RUN, SO AS TO NOT MISS THE DATA ADDITIONS
self.HIGH_HIST[item.Symbol].append(data[item.Symbol].High)
self.LOW_HIST[item.Symbol].append(data[item.Symbol].Low)
self.TIME_HIST[item.Symbol].append(algorithm.Time)
self.CHART_HIGH[item.Symbol] = max(self.HIGH_HIST[item.Symbol])
self.CHART_LOW[item.Symbol] = min(self.LOW_HIST[item.Symbol])
item.BASE_series_index = 0
item.series_index = 0
item.UNDERNEATH_LIST = []
if self.symbolDataBySymbol[item.Symbol].CONFIRMED_DAYS >= 3 and self.symbolDataBySymbol[item.Symbol].HAS_BEEN_STOPPED_OUT == False and self.symbolDataBySymbol[item.Symbol].PRINTED_CONFORMATION == False:
algorithm.Debug(f"{item.Symbol.Value} TIME TO COMPARE 3rd BAR {round(data[item.Symbol].High,2)} to 2nd BAR {self.symbolDataBySymbol[item.Symbol].SECOND_BAR_HIGH}")
if round(data[item.Symbol].High,2) > self.symbolDataBySymbol[item.Symbol].SECOND_BAR_HIGH :
algorithm.Debug(f"{item.Symbol.Value} GRID CONFIRMED - OP point {item.Time}")
self.symbolDataBySymbol[item.Symbol].CONFIRMED = True
self.symbolDataBySymbol[item.Symbol].PAUSED = False
self.symbolDataBySymbol[item.Symbol].PRINTED_CONFORMATION = True
else:
algorithm.Debug(f"{item.Symbol.Value} GRID FAILED 3rd bar HIGH was lower than 2nd HIGH - REDRAW GRID WITH LOWEST LOW SINCE OP")
self.symbolDataBySymbol[item.Symbol].CONFIRMING_GRID = False
data_IN = {'Open' : data[item.Symbol].Open,
'High' : data[item.Symbol].High,
'Low' : data[item.Symbol].Low,
'Close' : data[item.Symbol].Close,
'EndTime': algorithm.Time
}
algorithm.Debug(data_IN)
##This resets the dict with the latest grid for the specific symbol
self.symbolDataBySymbol[item.Symbol] = SymbolData(item.Symbol, data_IN, algorithm, self.Name, self.BLUE_DEG, self.GREEN_DEG)
self.symbolDataBySymbol[item.Symbol].REBUILD_CHECK = data[item.Symbol].High
self.symbolDataBySymbol[item.Symbol].HAS_RESET = True
self.symbolDataBySymbol[item.Symbol].HAS_BEEN_STOPPED_OUT = False
#This gets the low of the bar that the potential OP is set on
#This is used to compare if any of HIGH's the next 3 bars are less than this VALUE
#IF a high is LOWER then the grid fails
self.symbolDataBySymbol[item.Symbol].LOW_OF_RANGE = data[item.Symbol].Low
#Set CONFIRMING_GRID to true
#This means the next three bars are used to see if the lows are greater than blue 0
self.symbolDataBySymbol[item.Symbol].CONFIRMING_GRID = True
#NOW JUST COPIED OVER STUFF FROM MAIN RUN, SO AS TO NOT MISS THE DATA ADDITIONS
self.HIGH_HIST[item.Symbol].append(data[item.Symbol].High)
self.LOW_HIST[item.Symbol].append(data[item.Symbol].Low)
self.TIME_HIST[item.Symbol].append(algorithm.Time)
self.CHART_HIGH[item.Symbol] = max(self.HIGH_HIST[item.Symbol])
self.CHART_LOW[item.Symbol] = min(self.LOW_HIST[item.Symbol])
item.BASE_series_index = 0
item.series_index = 0
item.UNDERNEATH_LIST = []
'''
Here is the trading logic
as we have a confirmed grid OP
'''
if data.ContainsKey(item.Symbol) and data[item.Symbol] is not None and self.symbolDataBySymbol[item.Symbol].CONFIRMED and not algorithm.Portfolio[item.Symbol].Invested and self.symbolDataBySymbol[item.Symbol].PAUSED == False and item.TRADE_INITIATED==False:
if data[item.Symbol].Open <= item.blue_zero.values[item.series_index][0] or data[item.Symbol].Low <= item.blue_zero.values[item.series_index][0]:
algorithm.Debug(f"{item.Symbol.Value} - BUY POINT - Enter Long trade - OP {item.Time}")
algorithm.EmitInsights(Insight.Price(item.Symbol, timedelta(100000), InsightDirection.Up))
item.TRADE_INITIATED = True
'''
6 Stoping out due to logic
'''
if data.ContainsKey(item.Symbol) and data[item.Symbol] is not None and algorithm.Portfolio[item.Symbol].Invested and self.symbolDataBySymbol[item.Symbol].PAUSED == False and item.TRADE_INITIATED==False:
if data[item.Symbol].High <= item.blue_pos_two.values[item.series_index][0]:
#2 bars needing to be under blue pos 2 to stop out the position
self.symbolDataBySymbol[item.Symbol].BARS_BELOW_BLUE_ZERO += 1
if self.symbolDataBySymbol[item.Symbol].BARS_BELOW_BLUE_ZERO >=2:
algorithm.Debug(f"{item.Symbol.Value} - SELL POINT - Stopped out logic - Close Long trade - OP {item.Time}")
algorithm.EmitInsights(Insight.Price(item.Symbol, timedelta(100000), InsightDirection.Flat))
item.TRADE_INITIATED = True
self.symbolDataBySymbol[item.Symbol].CONFIRMED = False
self.symbolDataBySymbol[item.Symbol].HAS_BEEN_STOPPED_OUT = True
item.series_index += 1
item.BASE_series_index += 1
return insights
def OnSecuritiesChanged(self, algorithm, changes):
for removed in changes.RemovedSecurities:
self.symbolDataBySymbol.pop(removed.Symbol, None)
algorithm.Debug(f"REMOVED Symbol in OnSecuritiesChanged: {removed.Symbol.Value}")
algorithm.EmitInsights(Insight.Price(removed.Symbol, timedelta(100000), InsightDirection.Flat))
for added in changes.AddedSecurities:
if added.Symbol not in self.symbolDataBySymbol:
algorithm.Debug(f"Added Symbol in OnSecuritiesChanged: {added.Symbol.Value}")
self.dfdf[added.Symbol] = algorithm.History(added.Symbol, 99)
time_out = []
for index, row in self.dfdf[added.Symbol].loc[added.Symbol].iterrows():
time_out.append(index)
#Log the highs and lows of the past 99 days
self.SOFT_HIST_HIGH[added.Symbol] = self.dfdf[added.Symbol].high.values
self.SOFT_HIST_LOW[added.Symbol] = self.dfdf[added.Symbol].low.values
self.df = algorithm.History(added.Symbol, 1)
data_IN = {'Open' : self.df.open[0],
'High' : self.df.high[0],
'Low' : self.df.low[0],
'Close' : self.df.close[0],
'EndTime': algorithm.Time
}
algorithm.Debug(data_IN)
##This resets the dict with the latest grid for the specific symbol
self.symbolDataBySymbol[added.Symbol] = SymbolData(added.Symbol, data_IN, algorithm, self.Name, self.BLUE_DEG, self.GREEN_DEG)
self.symbolDataBySymbol[added.Symbol].REBUILD_CHECK = self.df.high[0]
self.symbolDataBySymbol[added.Symbol].HAS_RESET = True
self.symbolDataBySymbol[added.Symbol].HAS_BEEN_STOPPED_OUT = False
#NOW JUST COPIED OVER STUFF FROM MAIN RUN, SO AS TO NOT MISS THE DATA ADDITIONS
self.HIGH_HIST[added.Symbol] = [self.df.high[0]]
self.LOW_HIST[added.Symbol] = [self.df.low[0]]
self.TIME_HIST[added.Symbol] = [algorithm.Time]
#This gets the low of the bar that the potential OP is set on
#This is used to compare if any of HIGH's the next 3 bars are less than this VALUE
#IF a high is LOWER then the grid fails
self.symbolDataBySymbol[added.Symbol].LOW_OF_RANGE = self.df.low[0]
#Set CONFIRMING_GRID to true
#This means the next three bars are used to see if the lows are greater than blue 0
self.symbolDataBySymbol[added.Symbol].CONFIRMING_GRID = True
self.CHART_HIGH[added.Symbol] = max(self.HIGH_HIST[added.Symbol])
self.CHART_LOW[added.Symbol] = min(self.LOW_HIST[added.Symbol])
class SymbolData:
def __init__(self, symbol, data_IN, algorithm, name, BLUE_DEG, GREEN_DEG):
self.Symbol = symbol
self.Name = name
self.Time = data_IN["EndTime"]
self.Value = 0
self.BLUE_DEG = float(BLUE_DEG)
self.GREEN_DEG = float(GREEN_DEG)
self.PAUSED = True
self.CONFIRMING_GRID = False
self.CONFIRMED_DAYS = 0
self.CONFIRMED = False
self.HAS_BEEN_STOPPED_OUT = False
self.BARS_BELOW_BLUE_ZERO = 0
self.PRINTED_CONFORMATION = False
self.TRADE_INITIATED = False
self.REBUILD_CHECK = 0
self.HAS_RESET = False
self.blue_zero = None
self.blue_neg_one = None
self.blue_neg_two = None
self.blue_neg_three = None
self.blue_neg_four = None
self.blue_neg_five = None
self.blue_neg_six = None
self.blue_neg_severn = None
self.blue_neg_eight = None
self.blue_neg_nine = None
self.blue_neg_ten = None
self.blue_neg_eleven = None
self.blue_neg_twelve = None
self.blue_neg_thirteen = None
self.blue_neg_fourteen = None
self.blue_neg_fifteen = None
self.blue_pos_one = None
self.blue_pos_two = None
self.blue_pos_three = None
self.blue_pos_four = None
self.blue_pos_five = None
self.blue_pos_six = None
self.blue_pos_severn = None
self.blue_pos_eight = None
self.blue_pos_nine = None
self.blue_pos_ten = None
self.blue_pos_eleven = None
self.blue_pos_twelve = None
self.blue_pos_thirteen = None
self.blue_pos_fourteen = None
self.blue_pos_fifteen = None
self.green_zero = None
self.green_neg_one = None
self.green_neg_two = None
self.green_neg_three = None
self.green_neg_four = None
self.green_neg_five = None
self.green_neg_six = None
self.green_neg_severn = None
self.green_neg_eight = None
self.green_neg_nine = None
self.green_neg_ten = None
self.green_neg_eleven = None
self.green_neg_twelve = None
self.green_neg_thirteen = None
self.green_neg_fourteen = None
self.green_neg_fifteen = None
self.green_pos_one = None
self.green_pos_two = None
self.green_pos_three = None
self.green_pos_four = None
self.green_pos_five = None
self.green_pos_six = None
self.green_pos_severn = None
self.green_pos_eight = None
self.green_pos_nine = None
self.green_pos_ten = None
self.green_pos_eleven = None
self.green_pos_twelve = None
self.green_pos_thirteen = None
self.green_pos_fourteen = None
self.green_pos_fifteen = None
self.UNDERNEATH_LIST = []
self.series_index = 0
self.BASE_series_index = 0
self.indicator = self.makeIndicator(symbol, data_IN)
def makeIndicator(self, symbol, data_IN):
self.LOW_series = get_price_series(price_point="Low", interval="1d", data=data_IN, csv_records = 222, blue_deg = self.BLUE_DEG, green_deg = self.GREEN_DEG)
self.blue_zero = self.LOW_series['blue 0']
self.blue_neg_one = self.LOW_series['blue neg1']
self.blue_neg_two = self.LOW_series['blue neg2']
self.blue_neg_three = self.LOW_series['blue neg3']
self.blue_neg_four = self.LOW_series['blue neg4']
self.blue_neg_five = self.LOW_series['blue neg5']
self.blue_neg_six = self.LOW_series['blue neg6']
self.blue_neg_severn = self.LOW_series['blue neg7']
self.blue_neg_eight = self.LOW_series['blue neg8']
self.blue_neg_nine = self.LOW_series['blue neg9']
self.blue_neg_ten = self.LOW_series['blue neg10']
self.blue_neg_eleven = self.LOW_series['blue neg11']
self.blue_neg_twelve = self.LOW_series['blue neg12']
self.blue_neg_thirteen = self.LOW_series['blue neg13']
self.blue_neg_fourteen = self.LOW_series['blue neg14']
self.blue_neg_fifteen = self.LOW_series['blue neg15']
self.blue_pos_one = self.LOW_series['blue pos1']
self.blue_pos_two = self.LOW_series['blue pos2']
self.blue_pos_three = self.LOW_series['blue pos3']
self.blue_pos_four = self.LOW_series['blue pos4']
self.blue_pos_five = self.LOW_series['blue pos5']
self.blue_pos_six = self.LOW_series['blue pos6']
self.blue_pos_severn = self.LOW_series['blue pos7']
self.blue_pos_eight = self.LOW_series['blue pos8']
self.blue_pos_nine = self.LOW_series['blue pos9']
self.blue_pos_ten = self.LOW_series['blue pos10']
self.blue_pos_eleven = self.LOW_series['blue pos11']
self.blue_pos_twelve = self.LOW_series['blue pos12']
self.blue_pos_thirteen = self.LOW_series['blue pos13']
self.blue_pos_fourteen = self.LOW_series['blue pos14']
self.blue_pos_fifteen = self.LOW_series['blue pos15']
self.green_zero = self.LOW_series['green 0']
self.green_neg_one = self.LOW_series['green neg1']
self.green_neg_two = self.LOW_series['green neg2']
self.green_neg_three = self.LOW_series['green neg3']
self.green_neg_four = self.LOW_series['green neg4']
self.green_neg_five = self.LOW_series['green neg5']
self.green_neg_six = self.LOW_series['green neg6']
self.green_neg_severn = self.LOW_series['green neg7']
self.green_neg_eight = self.LOW_series['green neg8']
self.green_neg_nine = self.LOW_series['green neg9']
self.green_neg_ten = self.LOW_series['green neg10']
self.green_neg_eleven = self.LOW_series['green neg11']
self.green_neg_twelve = self.LOW_series['green neg12']
self.green_neg_thirteen = self.LOW_series['green neg13']
self.green_neg_fourteen = self.LOW_series['green neg14']
self.green_neg_fifteen = self.LOW_series['green neg15']
self.green_pos_one = self.LOW_series['green pos1']
self.green_pos_two = self.LOW_series['green pos2']
self.green_pos_three = self.LOW_series['green pos3']
self.green_pos_four = self.LOW_series['green pos4']
self.green_pos_five = self.LOW_series['green pos5']
self.green_pos_six = self.LOW_series['green pos6']
self.green_pos_severn = self.LOW_series['green pos7']
self.green_pos_eight = self.LOW_series['green pos8']
self.green_pos_nine = self.LOW_series['green pos9']
self.green_pos_ten = self.LOW_series['green pos1']
self.green_pos_eleven = self.LOW_series['green pos11']
self.green_pos_twelve = self.LOW_series['green pos12']
self.green_pos_thirteen = self.LOW_series['green pos13']
self.green_pos_fourteen = self.LOW_series['green pos14']
self.green_pos_fifteen = self.LOW_series['green pos15']
self.Value = self.blue_zero.iloc[0].values[0]
#This check does not seem to do anything
return len(self.blue_zero) == 150from Selection.FundamentalUniverseSelectionModel import FundamentalUniverseSelectionModel
class BankingIndustryStocks(FundamentalUniverseSelectionModel):
'''
This module selects the most liquid stocks listed on the Nasdaq Stock Exchange.
'''
def __init__(self, filterFineData = True, universeSettings = None, securityInitializer = None):
'''Initializes a new default instance of the TechnologyUniverseModule'''
super().__init__(filterFineData, universeSettings, securityInitializer)
self.numberOfSymbolsCoarse = 1000
self.numberOfSymbolsFine = 100
self.dollarVolumeBySymbol = {}
self.lastMonth = -1
def SelectCoarse(self, algorithm, coarse):
'''
Performs a coarse selection:
-The stock must have fundamental data
-The stock must have positive previous-day close price
-The stock must have positive volume on the previous trading day
'''
if algorithm.Time.month == self.lastMonth:
return Universe.Unchanged
sortedByDollarVolume = sorted([x for x in coarse if x.HasFundamentalData and x.Volume > 0 and x.Price > 0],
key = lambda x: x.DollarVolume, reverse=True)[:self.numberOfSymbolsCoarse]
self.dollarVolumeBySymbol = {x.Symbol:x.DollarVolume for x in sortedByDollarVolume}
# If no security has met the QC500 criteria, the universe is unchanged.
if len(self.dollarVolumeBySymbol) == 0:
return Universe.Unchanged
return list(self.dollarVolumeBySymbol.keys())
def SelectFine(self, algorithm, fine):
'''
Performs a fine selection for companies in the Morningstar Banking Sector
'''
# Filter stocks and sort on dollar volume
sortedByDollarVolume = sorted([x for x in fine if x.AssetClassification.MorningstarIndustryGroupCode == MorningstarIndustryGroupCode.Banks],
key = lambda x: self.dollarVolumeBySymbol[x.Symbol], reverse=True)
if len(sortedByDollarVolume) == 0:
return Universe.Unchanged
self.lastMonth = algorithm.Time.month
return [x.Symbol for x in sortedByDollarVolume[:self.numberOfSymbolsFine]]#[:5]import matplotlib
import pandas as pd
import numpy as np
import re
from math import *
import os
def get_price_series(price_point: str, interval: str = "1d",
csv_records: int = 125, data=None,
blue_deg: int = 39, green_deg: int = 141.2):
class Line:
slope = 0.0
intercept = 0.0
point = (0, 0)
color = ""
def __init__(self, slope: float, intercept: float, color: str, point: tuple, label: str):
self.slope = slope
self.intercept = intercept
self.label = label
self.point = point
self.color = color
def get_y(self, x: float):
return self.slope * float(x) + self.intercept
lines = []
date_price_series = {}
grid_lines = []
col = price_point
blue_d = 73.2
green_d = 76.1
blue_deg = blue_deg
green_deg = green_deg
const_blue_d = 1.06
const_green_d = 0.67
const_blue_angle = 1.3800000000000001
const_green_angle = 1.1300000000000001
xminratio = -1.2826428983835214
xmaxratio = 117.42191299143596
yminratio = 0.6681857061381054
ymaxratio = 1.0065736915744727
xpx = 2095.351999999999
dpi = 100.0
# We will be using x_y_angle everywhere since it is in radians
blue_angle = radians(blue_deg)
green_angle = radians(green_deg)
frequency_unit = re.findall(r'[A-Za-z]+|\d+', interval)[1]
frequency = re.findall(r'[A-Za-z]+|\d+', interval)[0]
interval = frequency + frequency_unit
if interval != "1d":
matplotlib.rcParams['timezone'] = 'US/Eastern'
else:
matplotlib.rcParams['timezone'] = 'UTC'
global stock
# tmp_data = yf.download(stock, dataStart, dataEnd, interval)
if isinstance(data["High"], float):
leny = abs(data["High"] - data["Low"])
xmin = xminratio #* len(data)
xmax = xmaxratio #* len(data)
ymin = yminratio * data["Low"]
ymax = ymaxratio * data["High"]
xp = 0.0
yp = data[col]
else:
leny = abs(data["High"][0] - data["Low"][0])
xmin = xminratio #* len(data)
xmax = xmaxratio #* len(data)
ymin = yminratio * data["Low"][0]
ymax = ymaxratio * data["High"][0]
xp = 0.0
yp = data[col][int(xp)]
ratio = (ymax - ymin) / (xmax - xmin)
xpx = xpx / abs(xmax - xmin)
point = (xp, yp)
y_line_slope = 1.0 / tan(green_angle * const_green_angle) * ratio
y_line_intercept = yp - y_line_slope * xp
x_line_slope = tan(blue_angle * const_blue_angle) * ratio
x_line_intercept = yp - x_line_slope * xp
lines.append(Line(x_line_slope, x_line_intercept, 'blue', point, "blue 0"))
lines.append(Line(y_line_slope, y_line_intercept, "green", point, "green 0"))
for line in lines:
if line.color == "green":
g_d_inches = green_d / 25.4
inches = g_d_inches / sin(green_angle - pi / 2.0)
xp = inches * dpi / xpx * const_green_d
else:
b_d_inches = blue_d / 25.4
inches = b_d_inches / sin(blue_angle)
xp = inches * dpi / xpx * const_blue_d
for direction in (-1.0, 1.0):
intercept = float(line.intercept)
for x in range(15):
yp = float(intercept)
intercept = yp + abs(line.slope) * xp * direction
point = (xp, yp)
if direction == 1: # going up
pos = "pos" + str(x + 1)
else: # going down
pos = "neg" + str(x + 1)
label = line.color + " " + pos
li = Line(line.slope, intercept, line.color, point, label)
grid_lines.append(li)
all_lines = lines + grid_lines
xp = lines[0].point[0]
for line in all_lines:
new_intercept = xp * float(line.slope) + float(line.intercept)
prices = []
for x in range(csv_records):
price = round(float(line.slope) * x + new_intercept, 8)
prices.append(price)
df = {col: prices}
df = pd.DataFrame(df)
df.index.name = "Date"
date_price_series[line.label] = df
return date_price_series#https://github.com/QuantConnect/Lean/blob/master/Algorithm.Framework/Selection/QC500UniverseSelectionModel.py
from clr import AddReference
AddReference("System")
AddReference("QuantConnect.Common")
AddReference("QuantConnect.Algorithm.Framework")
from QuantConnect.Data.UniverseSelection import *
from Selection.FundamentalUniverseSelectionModel import FundamentalUniverseSelectionModel
from itertools import groupby
from math import ceil
class QC500UniverseSelectionModel_V2(FundamentalUniverseSelectionModel):
'''Defines the QC500 universe as a universe selection model for framework algorithm
For details: https://github.com/QuantConnect/Lean/pull/1663'''
def __init__(self, filterFineData = True, universeSettings = None, securityInitializer = None):
'''Initializes a new default instance of the QC500UniverseSelectionModel'''
super().__init__(filterFineData, universeSettings, securityInitializer)
self.numberOfSymbolsCoarse = 1000
self.numberOfSymbolsFine = 500
self.dollarVolumeBySymbol = {}
self.lastMonth = -1
def SelectCoarse(self, algorithm, coarse):
'''Performs coarse selection for the QC500 constituents.
The stocks must have fundamental data
The stock must have positive previous-day close price
The stock must have positive volume on the previous trading day'''
if algorithm.Time.month == self.lastMonth:
return Universe.Unchanged
sortedByDollarVolume = sorted([x for x in coarse if x.HasFundamentalData and x.Volume > 0 and x.Price > 0],
key = lambda x: x.DollarVolume, reverse=True)[:self.numberOfSymbolsCoarse]
self.dollarVolumeBySymbol = {x.Symbol:x.DollarVolume for x in sortedByDollarVolume}
# If no security has met the QC500 criteria, the universe is unchanged.
# A new selection will be attempted on the next trading day as self.lastMonth is not updated
if len(self.dollarVolumeBySymbol) == 0:
return Universe.Unchanged
# return the symbol objects our sorted collection
return list(self.dollarVolumeBySymbol.keys())
def SelectFine(self, algorithm, fine):
'''Performs fine selection for the QC500 constituents
The company's headquarter must in the U.S.
The stock must be traded on either the NYSE or NASDAQ
At least half a year since its initial public offering
The stock's market cap must be greater than 500 million'''
sortedBySector = sorted([x for x in fine if x.CompanyReference.CountryId == "USA"
and x.CompanyReference.PrimaryExchangeID in ["NYS","NAS"]
and (algorithm.Time - x.SecurityReference.IPODate).days > 180
and x.MarketCap > 5e8],
key = lambda x: x.CompanyReference.IndustryTemplateCode)
count = len(sortedBySector)
# If no security has met the QC500 criteria, the universe is unchanged.
# A new selection will be attempted on the next trading day as self.lastMonth is not updated
if count == 0:
return Universe.Unchanged
# Update self.lastMonth after all QC500 criteria checks passed
self.lastMonth = algorithm.Time.month
percent = self.numberOfSymbolsFine / count
sortedByDollarVolume = []
# select stocks with top dollar volume in every single sector
for code, g in groupby(sortedBySector, lambda x: x.CompanyReference.IndustryTemplateCode):
y = sorted(g, key = lambda x: self.dollarVolumeBySymbol[x.Symbol], reverse = True)
c = ceil(len(y) * percent)
sortedByDollarVolume.extend(y[:c])
sortedByDollarVolume = sorted(sortedByDollarVolume, key = lambda x: self.dollarVolumeBySymbol[x.Symbol], reverse=True)
return [x.Symbol for x in sortedByDollarVolume[:self.numberOfSymbolsFine]][475:]