| Overall Statistics |
|
Total Trades 0 Average Win 0% Average Loss 0% Compounding Annual Return 0% Drawdown 0% Expectancy 0 Net Profit 0% Sharpe Ratio 0 Probabilistic Sharpe Ratio 0% Loss Rate 0% Win Rate 0% Profit-Loss Ratio 0 Alpha 0 Beta 0 Annual Standard Deviation 0 Annual Variance 0 Information Ratio 1.306 Tracking Error 0.121 Treynor Ratio 0 Total Fees $0.00 Estimated Strategy Capacity $0 Lowest Capacity Asset |
from AlgorithmImports import *
from EnumFunc import *
class NasdaqAlpha(AlphaModel):
def __init__(self, lumberDataClass):
self.insights = []
self.securities = []
self.lumberDataClass = lumberDataClass # Passed the Lumber Data Class
# Handles security changes in from your universe model.
def OnSecuritiesChanged(self, algorithm: QCAlgorithm, changes: SecurityChanges) -> None:
# This code runs everytime there is a contract rollover but does it know which symbol has switched
for changed_event in algorithm.CurrentSlice.SymbolChangedEvents.Values:
# algorithm.Debug(f"Contract rollover from (AM- OnSecuritiesChanged) {changed_event.OldSymbol} to {changed_event.NewSymbol}, Time:{algorithm.Time}")
pass
# Everytime Contract rollsOver (any Lumber or Nasdaq), the following line runs, but ...
# algorithm.Debug(f"Count AM: {changes.Count}, changes:{changes}")
# .. this code only runs Once at the very start not when future contracts roll!
for security in changes.AddedSecurities:
# algorithm.Debug(f"In OnSecuritiesChanged(AM) @ DateTime:{algorithm.Time}, Mapped/ID: {security.Mapped}, Canonical: {security.Mapped.Canonical} \
# Symbol: {security.Symbol}, Value: {security.Mapped.Value}, SecurityType: {getSecurityType(security.Mapped.SecurityType)}")
pass
for security in changes.RemovedSecurities:
pass
def Update(self, algorithm, data):
# https://www.quantconnect.com/docs/v2/writing-algorithms/datasets/quantconnect/us-futures-security-master#05-Data-Point-Attributes
# This Code runs everytime there is a futures rollover
# BEAWARE SymbolChangedEvents may also be linked to other events?
# algorithm.Debug(f"data.SymbolChangedEvents.Keys:{data.SymbolChangedEvents.Keys}")
if data.Keys[0] == self.lumberDataClass.Symbol:
for changed_event in data.SymbolChangedEvents.Values:
algorithm.Debug(f"Contract rollover (AM- Update Method) from {changed_event.OldSymbol} to {changed_event.NewSymbol}")
security = data.Keys[0]
self.securities.append(security.Value)
algorithm.Debug(f"In Update (AM) @ DateTime:{algorithm.Time}, security:{security}, ID:{security.ID},Canonical:{security.Canonical} \
+ Value:{security.Value},Underlying:{security.Underlying}. SecurityType:{getSecurityType(security.SecurityType)}")
insight = None
return self.insights
from AlgorithmImports import *
import math
from collections import deque
from EnumFunc import *
from functools import wraps
# A skeleton class for storing Futures Data - Shared by all Models
class DataClass:
def __init__(self, Algorithm, continuous_contract):
self.algo = Algorithm # This is Our Instance of QCAlgorithm
# self.algo.UniverseSettings.ExtendedMarketHours = True
# https://www.quantconnect.com/docs/v2/writing-algorithms/securities/asset-classes/futures/requesting-data#09-Properties
self.continuous_contract = continuous_contract
# Symbols for securities:
self.Symbol = self.continuous_contract.Symbol
self.tickSize = self.continuous_contract.SymbolProperties.MinimumPriceVariation
# Index, askclose, askhigh, asklow, askopen, asksize, bidclose, bidhigh, bidlow, bidopen, bidsize, close, high, low, open, volume)
# Index, Value:(Timestamp('2022-11-15 00:00:00'), 'LBS Y3IN813DW4QP', Timestamp('2022-10-26 19:00:00'))
self.yesterday = dotdict(dict.fromkeys(['open', 'high', 'low', 'close', 'volume']))
# Make sure to use self.Mapped and not self.Symbol since using BackwardPanama
self.lastBid = dotdict(dict.fromkeys(['time', 'price', 'size']))
self.lastAsk = dotdict(dict.fromkeys(['time','price', 'size']))
self._bestBid = dotdict(dict.fromkeys(['time', 'price', 'size']))
self._bestAsk = dotdict(dict.fromkeys(['time','price', 'size']))
self._trade = dotdict(dict.fromkeys(['time','price', 'size']))
# self.trade = dotdict(dict.fromkeys(['time','price', 'size']))
self._WAP = None
# https://www.quantconnect.com/docs/v2/writing-algorithms/indicators/supported-indicators/average-true-range
# This should be yesterday's
# self.atr = self.algo.ATR(self.Symbol, Resolution.Daily)
# quoteEvent is triggered everytime a new Quote is placed not only when the best quote is changed/placed- I think?
# self.quoteEvent = TickQuoteBarConsolidator(1)
# self.quoteEvent.DataConsolidated += self.quoteEventHandler
# self.algo.SubscriptionManager.AddConsolidator(self.Symbol, self.quoteEvent)
# tradeEvent is triggered everytime a Trade is placed?
# self.tradeEvent = TickConsolidator(1)
# self.tradeEvent.DataConsolidated += self.tradeEventHandler
# self.algo.SubscriptionManager.AddConsolidator(self.Symbol, self.tradeEvent)
# To Update Yesterday: At Midnight the contract Rolls and pulls the correct data for Lumber - The issue is other futures may have different Open Close timings
# self.algo.Schedule.On(self.algo.DateRules.EveryDay(), self.algo.TimeRules.Midnight, self.updateYesterday)
# Update at a specific Time - Symbol agnostic - This is 10 minutes before Lumber Open - Change when daylight savings
# self.algo.Schedule.On(self.algo.DateRules.EveryDay(), self.algo.TimeRules.At(8,59), self.updateYesterday)
# Get Yesterday's Settlement - Make it a Property
# self.Securities[self.lumber.Mapped].Expiry # Get contract's expiry?
# Update Yesterday Price 10 minutes before Market Opens - Can even use 'LBS' in TimeRules.AfterMarketOpen
self.algo.Schedule.On(self.algo.DateRules.EveryDay(self.Symbol), self.algo.TimeRules.AfterMarketOpen(self.Symbol, -10), self.updateYesterday)
# Reset BBO dicts 10 minutes after market close - Not sure if it impacts extendedMarketHours
self.algo.Schedule.On(self.algo.DateRules.EveryDay(self.Symbol), self.algo.TimeRules.BeforeMarketClose(self.Symbol, -10), self.resetTickDict)
# Nasdaq - closes at 4 pm & opens at 8:30 Chicago, Also trades at 11 pm till not sure when?
# OnSecuritiesChanged: Date:2022-12-19 23:00:00, security:MNQ Y6URRFPZ86BL
def quoteEventHandler(self, sender: object, Qbar: QuoteBar) -> None:
if self.algo.CurrentSlice.Ticks.ContainsKey(self.Symbol) and self.algo.CurrentSlice.Ticks[self.Symbol] is not None:
ticks = self.algo.CurrentSlice.Ticks[self.Symbol]
for tick in ticks:
tick_type = getTickType(tick.TickType)
if tick_type == 'Quote':
if int(getattr(self.algo.Securities[self.Mapped], 'AskSize')) != 0:
for key,prop in zip(list(self._bestAsk.keys()),['LocalTime','AskPrice', 'AskSize']):
setattr(self._bestAsk,str(key),getattr(self.algo.Securities[self.Mapped], prop))
if int(getattr(self.algo.Securities[self.Mapped], 'BidSize')) != 0:
for key,prop in zip(list(self._bestBid.keys()),['LocalTime','BidPrice', 'BidSize']):
setattr(self._bestBid,str(key),getattr(self.algo.Securities[self.Mapped], prop))
# Code to store Last Bid & Ask - switching it off for the time being
# if int(getattr(tick, 'AskPrice')) != 0:
# # In DateTime - microseconds missing leading zeros in our dict versus reported
# for key,prop in zip(list(self.lastAsk.keys()),['EndTime','AskPrice', 'AskSize']):
# setattr(self.lastAsk,str(key),getattr(tick, prop))
# if int(getattr(tick, 'BidPrice')) != 0:
# for key,prop in zip(list(self.lastBid.keys()),['EndTime','BidPrice', 'BidSize']):
# setattr(self.lastBid,str(key),getattr(tick, prop))
def tradeEventHandler(self, sender: object, Tbar: TradeBar) -> None:
if self.algo.CurrentSlice.Ticks.ContainsKey(self.Symbol) and self.algo.CurrentSlice.Ticks[self.Symbol] is not None:
ticks = self.algo.CurrentSlice.Ticks[self.Symbol]
for tick in ticks:
tick_type = getTickType(tick.TickType)
if tick_type == 'Trade':
self.algo.tradeCount += 1
if int(getattr(tick, 'Quantity')) != 0:
for key,prop in zip(list(self.trade.keys()),['EndTime','Price','Quantity']):
setattr(self.trade,str(key),getattr(tick, prop))
# Resetting Best Bid/Ask at MarketClose
def resetTickDict(self):
# Resetting After Market Close
# self.lastAsk = self.lastAsk.fromkeys(self.lastAsk, None)
# self.lastBid = self.lastBid.fromkeys(self.lastBid, None)
self._bestAsk = self._bestAsk.fromkeys(self._bestAsk, None)
self._bestBid = self._bestBid.fromkeys(self._bestBid, None)
self._trade = self._trade.fromkeys(self._trade, None)
self._WAP = None
# self.algo.Debug(f"Ask Reset:{self.Symbol} :{self.algo.Time}:{self._bestAsk}")
# self.algo.Debug(f"Bid Reset :{self.Symbol} :{self.algo.Time}:{self._bestBid}")
# self.algo.Debug(f"Trade Reset :{self.Symbol} :{self.algo.Time}:{self._trade}")
def updateYesterday(self):
# self.algo.Debug(f"Updated Yesterday price for:{self.Symbol} @ {self.algo.Time}")
# Stores Contracts Yesterday's OHCLV
# Note that Volume doesn't match with reported in our Database (atleast for Lumber)
if self.Mapped is not None:
# self.algo.Debug(f"ctrctHistory Symbol:{self.Mapped},{self.continuous_contract}")
ctrctHistory = self.algo.History(self.Mapped, 1, Resolution.Daily )
for bar in ctrctHistory.itertuples():
for property in ['open', 'high', 'low', 'close', 'volume']:
if not math.isnan(float(getattr(bar, property))): # Required since updateYesterday is called even on Non trading days
self.yesterday[str(property)] = getattr(bar, property)
# for prop in dir(bar):
# value = getattr(bar, prop)
# self.algo.Debug(f"prop:{prop}, value:{value}")
# Checking if we get same data
# if self.algo.CurrentSlice[self.Mapped].High is not None:
# self.algo.Debug(f"{self.algo.CurrentSlice[self.Mapped].High}")
# RealOnly RealTime
@property
def Mapped(self):
return getattr(self.continuous_contract, 'Mapped')
@property
def Canonical(self):
return getattr(self.Symbol, 'Canonical')
# @property
def Underlying(self):
return getattr(self, 'Underlying')
@property
def bestBid(self):
# Not using this since somehow when we club 2 futures, this wasn't updating correctly or giving None
# Also using getattr(self.algo.Securities[self.Mapped], prop)) instead of
# if self.algo.CurrentSlice.Ticks.ContainsKey(self.Symbol) and self.algo.CurrentSlice.Ticks[self.Symbol] is not None:
# ticks = self.algo.CurrentSlice.Ticks[self.Symbol]
# for tick in ticks:
# tick_type = getTickType(tick.TickType)
# if tick_type == 'Quote' and int(getattr(self.algo.Securities[self.Mapped], 'BidSize')) != 0:
# for key,prop in zip(list(self._bestBid.keys()),['LocalTime','BidPrice', 'BidSize']):
# setattr(self._bestBid,str(key),getattr(self.algo.Securities[self.Mapped], prop))
# Only updating if Bid price and Size changed so as to keep the original Time of last update
if int(getattr(self.algo.Securities[self.Mapped], 'BidSize')) != 0 and not self.alreadyUpdated('BidPrice', 'BidSize', self._bestBid):
for key,prop in zip(list(self._bestBid.keys()),['LocalTime','BidPrice', 'BidSize']):
setattr(self._bestBid,str(key),getattr(self.algo.Securities[self.Mapped], prop))
return self._bestBid
@property
def bestAsk(self):
# Only updating if Ask price and Size changed so as to keep the original Time of last update
if int(getattr(self.algo.Securities[self.Mapped], 'AskSize')) != 0 and not self.alreadyUpdated('AskPrice', 'AskSize', self._bestAsk):
for key,prop in zip(list(self._bestBid.keys()),['LocalTime','AskPrice', 'AskSize']):
setattr(self._bestAsk,str(key),getattr(self.algo.Securities[self.Mapped], prop))
return self._bestAsk
# This is not entirely correct as someone may have canceled and others may have added a bid with net being affect 0
def alreadyUpdated(self, price, size, dictionary):
return getattr(self.algo.Securities[self.Mapped], price) == dictionary.price and getattr(self.algo.Securities[self.Mapped], size) == dictionary.size
@property
def trade(self):
# TO DO: Need to make it work like Quotes - Ask/Bid - getattr(self.algo.Securities[self.Mapped], prop))
# Ideally this should use self.Mapped but it doesn't take that.
if self.algo.CurrentSlice.Ticks.ContainsKey(self.Symbol) and self.algo.CurrentSlice.Ticks[self.Symbol] is not None:
ticks = self.algo.CurrentSlice.Ticks[self.Symbol]
for tick in ticks:
tick_type = getTickType(tick.TickType)
if tick_type == 'Trade' and int(getattr(tick, 'Quantity')) != 0:
for key,prop in zip(list(self._trade.keys()),['Time','Price','Quantity']):
setattr(self._trade,str(key),getattr(tick, prop))
# setattr(self._trade,str(key),getattr(self.algo.Securities[self.Mapped], prop))
# https://www.quantconnect.com/docs/v2/writing-algorithms/historical-data/history-requests
# end_time = self.algo.Time
# start_time = end_time - timedelta(seconds=1)
# ticks = self.algo.History[Tick](self.Mapped, start_time, end_time, Resolution.Tick)
# count = 0
# # self.algo.Debug(f"type of ticks:{type(Ticks)}")
# for tick in ticks:
# if getTickType(tick.TickType) == 'Trade':
# self.algo.Debug(f"tick{count}:{tick}")
# count +=1
# if getTickType(tick.TickType) == 'Trade':
# for prop in dir(tick):
# try:
# value = getattr(tick, prop)
# self.algo.Debug(f"Tprop:{prop}, Tvalue:{value}")
# except:
# pass
# tick.set_Symbol(self.Mapped)
# tick_type = getTickType(tick.TickType)
# and int(getattr(tick, 'Quantity')) != 0
# if tick_type == 'Trade':
# count +=1
# dic = {k:getattr(tick, k) for k in ['EndTime','Price','Quantity'] if getattr(tick, k) is not None}
# # self.algo.Debug(f"Tickcount:{count}, time:{self.algo.Time} {dic}")
# for key,prop in zip(list(self._trade.keys()),['EndTime','Price','Quantity']):
# setattr(self._trade,str(key),getattr(tick, prop))
return self._trade
@property
def WAP(self):
if None not in (self.bestAsk.size,self.bestBid.size):
self._WAP = round(((self.bestBid.price * self.bestAsk.size) + (self.bestAsk.price * self.bestBid.size))/(self.bestBid.size + self.bestAsk.size),1)
return self._WAP
# @property
# def Bid(self):
# return self._Bid
# @Bid.setter
# def Bid(self, bid):
# # if bid == 0 or bid is None:
# # self._Bid = 0 # Pull this from other rolling?
# # pass
# # self._Bid = bid
# # self.WAP = None # Invalidate previously Calculated WAP
# # return self._Bid
# pass
# Links to documentation pertaining
# Time Modeling
# https://www.quantconnect.com/docs/v2/writing-algorithms/key-concepts/time-modeling/timeslices#03-Properties
# Futures - Handling Data
# https://www.quantconnect.com/docs/v2/writing-algorithms/securities/asset-classes/futures/handling-data
# To get the current Slice object, define an OnData method or use the CurrentSlice property of your algorithm (outside of the OnData method).
# If the data object doesn't contain any market data but it contains auxiliary data, the slice.ContainsKey(symbol) method can return true while slice[symbol] returns None.
class dotdict(dict):
"""dot.notation access to dictionary attributes"""
__getattr__ = dict.get
__setattr__ = dict.__setitem__
__delattr__ = dict.__delitem__
# region imports
from AlgorithmImports import *
from AlphaModel import *
from DataClass import *
import math
class NasdaqStrategy(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2022,12,19)
self.SetEndDate(2023,1,5)
self.SetCash(100000)
self.SetTimeZone(TimeZones.Chicago) # If TimeZone is Chicago - Algo Time and Data End Time are same at 1800 hours
self.SetWarmUp(timedelta(days=1))
self.Data = {}
# FutureSymbols = {'Lumber':Futures.Forestry.RandomLengthLumber,'Nasdaq':Futures.Indices.NASDAQ100EMini}
FutureSymbols = {'Lumber':Futures.Forestry.RandomLengthLumber}
# For Nasdaq we need DataMappingMode.OpenInterest or LastTradingDay - How to change?
for key, value in FutureSymbols.items():
dataMappingMode_ = DataMappingMode.FirstDayMonth if key == 'Lumber' \
else DataMappingMode.OpenInterest if key == 'Nasdaq' else DataMappingMode.LastTradingDay
# dataMappingMode_ = DataMappingMode.FirstDayMonth if key == 'Lumber' \
# else DataMappingMode.LastTradingDay
# BackwardsPanamaCanal
future = self.AddFuture(value, Resolution.Tick,dataMappingMode = dataMappingMode_, contractDepthOffset=0,
dataNormalizationMode = DataNormalizationMode.Raw, extendedMarketHours=True, fillDataForward = True)
self.Data[key] = DataClass(self, future) # Initiating DataClass for each Future & Passing our instance of QCAlgorithm Class
# NOT SURE IF WORKING CORRECTLY
# To trade the contract in the same time step you subscribe to the contract - Since GetLastKnowPrice makes a history request, it may slow the algorithm down.
# self.SetSecurityInitializer(lambda x: x.SetMarketPrice(self.GetLastKnownPrice(x)))
# Set alpha model - This also gives you access to the AlphaModel Instance
# self.alpha = NasdaqAlpha(self.Data['Lumber'])
# self.SetAlpha(self.alpha)
# Printing Yesterday data
# self.Schedule.On(self.DateRules.EveryDay(self.Data['Lumber'].Symbol), self.TimeRules.AfterMarketOpen(self.Data['Lumber'].Symbol, -1), self.beforeLumberOpen)
self.Schedule.On(self.DateRules.EveryDay(), self.TimeRules.At(8,55), self.beforeLumberOpen)
# self.TimeRules.BeforeMarketClose(symbol: Symbol, minutesBeforeClose: float = 0, extendedMarketOpen: bool = False)
self.tradeCount = 0
# Check Entry Condition only 30 seconds prior to Market Open
self.entryTimeStart = self.Time.replace(hour=9, minute=0, second=0, microsecond=55000)
self.entryTimeEnd = self.Time.replace(hour=9, minute=1, second=0, microsecond=0)
def beforeLumberOpen(self):
if self.IsWarmingUp: return
# Works - Get Yesterdays' data keeping in mind Rolled Over Contracts
for symbol in self.Data.keys():
# self.Debug(f"Yester Time:{self.Time}, {self.Data[symbol].Mapped}.yesterday:{self.Data[symbol].yesterday}")
pass
# Works
# for security in self.ActiveSecurities.Values:
# self.Debug(f"self.Time:{self.Time} ActiveSecurity:{security}")
def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
# Only runs at the start of program not when Contract rolls
for security in changes.AddedSecurities:
if len(changes.AddedSecurities) == 0: return
self.Debug(f"In Main: OnSecuritiesChanged: Date:{self.Time}, security:{changes.AddedSecurities[0].Mapped}")
def OnEndOfDay(self, symbol):
# self.Debug(f"OnEndOfDay self.Time:{self.Time}, Symbol:{self.Data['Lumber'].Mapped.Value}")
pass
def OnData(self, data):
# Useless
# self.Debug("Hello")
# for chain in data.FutureChains:
# contracts = list(chain.Value)
# self.Debug(f"Last contracts: {contracts}")
# for contract in contracts:
# self.Debug(f"contract:{contract}, Last Price: {contract.LastPrice}")
# https://www.quantconnect.com/docs/v2/writing-algorithms/historical-data/warm-up-periods#03-Warm-Up-Vs-Reality
# In OnData: Don't run if the indicators aren't ready
if self.IsWarmingUp: return
# Below code important to compare Ticks with whats reported from DataClass
tickTradeProps = ['LastPrice','Quantity']
tickQuoteProps = ['BidPrice','BidSize','AskPrice','AskSize']
# tickOIProps = ['Value'] # Not Getting Data for OpenInterest
# for k in data.Ticks.Keys:
# self.Debug(f"keys:{k}")
# self.Debug(f"Underlying:{k.Underlying}")
# if data.Ticks.ContainsKey(k) and data.Ticks[k] is not None:
# self.Debug("yes")
for security in data.Keys:
# if self.Time < self.entryTimeStart or self.Time > self.entryTimeEnd: return
# pass
# self.Debug(f"security:{security}")
# for prop in dir(security):
# value = getattr(security, prop)
# self.Debug(f"prop:{prop}, value:{value}")
if security == self.Data['Lumber'].Mapped:
self.Debug(f"security Matched:{security}")
if data.Ticks.ContainsKey(security) and data.Ticks[security] is not None:
ticks = data.Ticks[security]
for tick in ticks:
previous_time = self.Time
tick_type = getTickType(tick.TickType)
if tick_type == 'Trade' and int(getattr(tick, 'Quantity')) != 0:
# This is for Trade Data
dic = {k:getattr(tick, k) for k in tickTradeProps if getattr(tick, k) is not None}
self.Debug(f"TradeTick:{self.Time} {dic}")
# self.Debug(f"self.Data['Lumber'].Underlying:{self.Data['Lumber'].Underlying} ")
# elif tick_type == 'Quote':
# This if for Last Bid & Ask
# dic = {k:getattr(tick, k) for k in tickQuoteProps if getattr(tick, k) is not None}
# self.Debug(f"*QuoteTick*: {self.Time}, security:{security}, {dic}")
# This if for Best Bid & Ask
# dic2 = {k:getattr(self.Securities[security], k) for k in tickQuoteProps}
# self.Debug(f"+QuoteTick2+:{self.Time}, security:{security}, {dic2}")
for symbol in self.Data.keys():
if self.Time < self.entryTimeStart or self.Time > self.entryTimeEnd: return
# self.Debug(f"SYMBOL:{symbol}")
# self.Debug(f"++BestBid:{symbol}:{self.Time}:{self.Data[symbol].bestBid}")
# self.Debug(f"^^BestAsk:{symbol}:{self.Time}:{self.Data[symbol].bestAsk}")
self.Debug(f"**trade:{symbol}:{self.Time}:{self.Data[symbol].trade}")
# self.Debug(f"**WAP:{symbol}:{self.Time}:{self.Data[symbol].WAP}")
# self.Debug(f"**ATR:{symbol}:{self.Time}:{self.Data[symbol].atr.Current.Value}")
# self.Debug(f"**LastAsk:{self.Data[symbol].lastAsk}")
# self.Debug(f"**LastBid:{self.Data[symbol].lastBid}")
# self.Debug(f" BestBid == LastBid:{self.Data[symbol].bestBid.price == self.Data[symbol].lastBid.price}")
# self.Debug(f" BestAsk == LastAsk:{self.Data[symbol].bestAsk.price == self.Data[symbol].lastAsk.price}")
# Just to check if FuncSecuritySeeder gets us GetLastKnownPrices
# for security in self.ActiveSecurities.Values:
# self.Debug(f"Todays Close:{self.Securities[security.Symbol].Close}, Date:{self.Time}, security:{security}")
# if security == 'LBS Y54QLJOCEN7L':
# self.Debug(f"Mapped:{self.Data['Lumber'].Mapped}")
# self.Debug(f"security == self.Data['Lumber'].Mapped:{security.Symbol == self.Data['Lumber'].Mapped}")
# if security.Symbol == self.Data['Lumber'].Mapped:
# self.Debug(f"Todays Close:{self.Securities[security.Symbol].Close}, Date:{self.Time}, security:{security}")
# self.Debug("Yes")
# b = self.Securities[self.Data['Lumber'].Mapped]
# for prop in dir(b):
# value = getattr(b, prop)
# self.Debug(f"prop11:{prop}, value:{value}")
# WORKS - Remove from here since updated several times in a day
# self.Debug(f"Todays ____Close:{self.Securities[self.Data['Lumber'].Mapped].Price}, Date:{self.Time}")
if data.Ticks.ContainsKey(self.Data['Lumber'].Mapped):
if data.Ticks[self.Data['Lumber'].Symbol] is not None:
self.Debug(f"**MAIN** Date:{self.Time}, MappedValue:{self.Data['Lumber'].Mapped.Value}, MappedID:{self.Data['Lumber'].Mapped.ID}, ")
# b = data.Bars[self.Data['Lumber'].Mapped]
# for prop in dir(b):
# value = getattr(b, prop)
# self.algo.Debug(f"prop:{prop}, value:{value}")
# Works
# if data.Ticks.ContainsKey(self.Data['Lumber'].Symbol):
# if data.Ticks[self.Data['Lumber'].Symbol] is not None:
# self.Debug(f"**MAIN** Date:{self.Time}, MappedValue:{self.Data['Lumber'].Mapped.Value}, MappedID:{self.Data['Lumber'].Mapped.ID}, ")
# https://www.quantconnect.com/docs/v2/writing-algorithms/datasets/quantconnect/us-futures-security-master#05-Data-Point-Attributes
for symbol in self.Data.keys():
if data.SymbolChangedEvents.ContainsKey(self.Data[symbol].Symbol):
symbolChangedEvent = data.SymbolChangedEvents[self.Data[symbol].Symbol]
self.Debug(f"In MAIN Symbol changed: {symbolChangedEvent.OldSymbol} -> {symbolChangedEvent.NewSymbol} \
EndTime:{symbolChangedEvent.EndTime} DataType:{getDataType(symbolChangedEvent.DataType)}, Expiry: {self.Securities[self.Data[symbol].Mapped].Expiry}")
# if data[self.Data['Lumber'].Mapped] is not None:
# self.Debug(f"data123Map: {data[self.Data['Lumber'].Mapped]}")
# When your algorithm stops executing, LEAN calls the OnEndOfAlgorithm method.
def OnEndOfAlgorithm(self) -> None:
# self.Debug(f"self.Alpha.securities:{self.alpha.securities}")
# self.Debug(f"self.count:{self.count}")
self.Debug(f"self.TradeCount:{self.tradeCount}")
def OnWarmUpFinished(self) -> None:
self.Debug(f"Algorithm Ready@{self.Time}")
pass
from AlgorithmImports import *
import math
from collections import deque
from EnumFunc import *
from functools import wraps
from IndicatorClass import *
# A skeleton class for storing Futures Data - Shared by all Models
class DataClass:
def __init__(self, Algorithm, continuous_contract):
self.algo = Algorithm # This is Our Instance of QCAlgorithm
self.continuous_contract = continuous_contract
self.Symbol = self.continuous_contract.Symbol
self.tickSize = self.continuous_contract.SymbolProperties.MinimumPriceVariation
self.yesterday = dotdict(dict.fromkeys(['open', 'high', 'low', 'close', 'volume']))
self._bestBid = dotdict(dict.fromkeys(['time', 'price', 'size']))
self._bestAsk = dotdict(dict.fromkeys(['time','price', 'size']))
self._trade = dotdict(dict.fromkeys(['time','price', 'size']))
self._WAP = None
self.atrPeriod = 2
# Cannot use Automatic Indicator for continuous Contract since it will only be updated at the beggining of next bar of Daily Resolution
# Same issue with using Daily consolidator (Registered or Manual) since all update at the start of next daily bar
# self.algo.WarmUpIndicator(self.Symbol, self._AutoATR, timedelta(days=self.atrPeriod+1))
# tradeEvent is triggered everytime a Trade is placed?
# self.tradeEvent = TickConsolidator(timedelta(days=1))
# # self.algo.RegisterIndicator(self.Symbol, self._AutoATR, self.tradeEvent,)
# # self.algo.RegisterIndicator(self.Symbol, self._AutoATR, Resolution.Daily, TradeBar)
# self.tradeEvent.DataConsolidated += self.tradeEventHandler
# self.algo.SubscriptionManager.AddConsolidator(self.Symbol, self.tradeEvent)
# self._AutoATR = self.algo.ATR(self.Symbol, self.atrPeriod, MovingAverageType.Simple, Resolution.Daily)
# self.algo.WarmUpIndicator(self.Symbol, self._AutoATR, timedelta(days=self.atrPeriod+1)) # Since this Indicator depends on Previous close, we do Period Plus 1
# Attaching an event with Auto ATR
# self._AutoATR.Updated += self.update_event_handler
# self.beforeMarketOpenIndicatorsMapped = {}
# self.beforeMarketOpenIndicatorsContinuous = {}
self.dailyIndicators = {
"ATR":IndicatorClass(algorithm=Algorithm,indicator=AverageTrueRange(self.atrPeriod, MovingAverageType.Simple),symbol=self.continuous_contract.Mapped),
} # beforeMarketOpenIndicators - Mapped as well as Continuous Indicators?
# Update Yesterday Price 10 minutes before Market Opens - Can even use 'LBS' in TimeRules.AfterMarketOpen
self.algo.Schedule.On(self.algo.DateRules.EveryDay(self.Symbol), self.algo.TimeRules.AfterMarketOpen(self.Symbol, -10), self.updateIndicators)
# Reset BBO dicts 10 minutes after market close - Not sure if it impacts extendedMarketHours
self.algo.Schedule.On(self.algo.DateRules.EveryDay(self.Symbol), self.algo.TimeRules.BeforeMarketClose(self.Symbol, -10), self.resetTickDict)
# def tradeEventHandler(self, sender: object, Tbar: TradeBar) -> None:
# tradebar = TradeBar(float(Tbar.Open), float(Tbar.High), float(Tbar.Low), float(Tbar.Close), float(Tbar.Volume))
# self._AutoATR.Update(tradebar)
def Update(self, data):
if data.SymbolChangedEvents.ContainsKey(self.Symbol):
for key,indicator in enumerate(self.dailyIndicators):
# if indicator is for mapped contract need to pass the current mapped contract
if self.Mapped:
self.dailyIndicators[indicator].resetIndicator(mapped=self.Mapped)
else:
self.dailyIndicators[indicator].resetIndicator(mapped=self.Symbol)
@property
def ATR(self):
# self.algo.Debug(f"self._ATR.IsReady:{self._ATR.IsReady} at {self.algo.Time}")
# self.algo.Debug(f"# of samples{self._ATR.Samples}")
return self.dailyIndicators["ATR"].value
# Resetting Best Bid/Ask at MarketClose
def resetTickDict(self):
self._bestAsk = self._bestAsk.fromkeys(self._bestAsk, None)
self._bestBid = self._bestBid.fromkeys(self._bestBid, None)
self._trade = self._trade.fromkeys(self._trade, None)
self._WAP = None
def businessDay(self,dt):
return 1 if getattr(self.algo.TradingCalendar.GetTradingDay(dt),'BusinessDay') else 0
def startBDate(self,end_dt, lag):
count = 0
start_dt = end_dt
while count!= lag:
start_dt -= timedelta(days=1)
count += self.businessDay(start_dt)
return start_dt
def updateIndicators(self):
historyATR = None
# Note that Volume doesn't match with reported in our Database (atleast for Lumber)
if self.Mapped is not None:
historyATR = self.algo.History[TradeBar](self.Mapped, 1, Resolution.Daily)
# https://www.quantconnect.com/docs/v2/writing-algorithms/historical-data/rolling-window#09-Cast-to-Other-Types
historyATR = self.algo.PandasConverter.GetDataFrame[TradeBar](list(historyATR))
for bar in historyATR.itertuples():
for property in ['open', 'high', 'low', 'close', 'volume']:
if not math.isnan(float(getattr(bar, property))): # Required since updateIndicators is called even on Non trading days
self.yesterday[str(property)] = getattr(bar, property)
# # ATR Update - # All Manual Indicators Updated Manually Here
# if not self._ATR.IsReady: # Runs (1) At start (2) When contract Rolls
# historyATR = self.algo.History[TradeBar](self.Mapped, self.atrPeriod+1, Resolution.Daily)
# historyATR = self.algo.PandasConverter.GetDataFrame[TradeBar](list(historyATR))
# # firstdate = historyATR.index.values[0][1]
# # self.algo.Debug(f"firstdate:{firstdate.date()}")
# for bar in historyATR.itertuples():
# tradebar = TradeBar(bar.Index[1], self.Mapped, float(bar.open), float(bar.high), float(bar.low), float(bar.close), float(bar.volume))
# # self.algo.Debug(f"Bar1:{tradebar} @ {bar.Index[1]}")
# self._ATR.Update(tradebar)
# # self.algo.Debug(f"1stManual:{tradebar},{bar.Index[1]}")
# else: # Indicator is ready so just updating it for the day.
# historyATR = self.algo.History[TradeBar](self.Mapped, 1, Resolution.Daily)
# historyATR = self.algo.PandasConverter.GetDataFrame[TradeBar](list(historyATR))
# # firstdate = historyATR.index.values[0][1]
# for bar in historyATR.itertuples():
# tradebar = TradeBar(bar.Index[1], self.Mapped, float(bar.open), float(bar.high), float(bar.low), float(bar.close), float(bar.volume))
# self._ATR.Update(tradebar)
# # self.algo.Debug(f"2ndManual:{tradebar},{bar.Index[1]}")
# # ATR Update - # All Manual Indicators Updated Manually Here
# if not self._AutoATR.IsReady: # Runs (1) At start OnLY
# # self.algo.Debug(f"start_1date:{self.startBDate(self.algo.Time,self.atrPeriod+1).date()}")
# # self.algo.Debug(f"firstdate1:{firstdate}")
# historyATR = self.algo.History(tickers=[self.Symbol],
# start=self.startBDate(self.algo.Time,self.atrPeriod+1).date(), # Using this since if we simply use startdate minus timedelta(self.atrperiod), it doesn't keep holidays in count
# end=self.algo.Time,
# resolution=Resolution.Daily,
# fillForward=True,
# extendedMarket=True,
# dataMappingMode=DataMappingMode.FirstDayMonth,
# dataNormalizationMode=DataNormalizationMode.Raw,
# contractDepthOffset=0)
# for bar in historyATR.itertuples():
# tradebar = TradeBar(bar.Index[2], bar.Index[1], float(bar.open), float(bar.high), float(bar.low), float(bar.close), float(bar.volume))
# self._AutoATR.Update(tradebar)
# # self.algo.Debug(f"1stAuto:{tradebar},{bar.Index[2]}")
# else: # Indicator is ready so just updating it for the day.
# # self.algo.Debug(f"start_2date:{self.startBDate(self.algo.Time,1).date()}")
# # self.algo.Debug(f"firstdate2:{firstdate}")
# historyATR = self.algo.History(tickers=[self.Symbol],
# start=self.startBDate(self.algo.Time,1).date(), # Using this since if we simply use startdate minus timedelta(self.atrperiod), it doesn't keep holidays in count
# end=self.algo.Time,
# resolution=Resolution.Daily,
# fillForward=True,
# extendedMarket=True,
# dataMappingMode=DataMappingMode.FirstDayMonth,
# dataNormalizationMode=DataNormalizationMode.Raw,
# contractDepthOffset=0)
# for bar in historyATR.itertuples():
# # self.algo.Debug(f"2ndAuto:{tradebar},{bar.Index[2]}")
# tradebar = TradeBar(bar.Index[2], bar.Index[1], float(bar.open), float(bar.high), float(bar.low), float(bar.close), float(bar.volume))
# self._AutoATR.Update(tradebar)
# RealOnly RealTime
@property
def Mapped(self):
return getattr(self.continuous_contract, 'Mapped')
@property
def Canonical(self):
return getattr(self.Symbol, 'Canonical')
@property
def bestBid(self):
# Only updating if Bid price and Size changed so as to keep the original Time of last update
if int(getattr(self.algo.Securities[self.Mapped], 'BidSize')) != 0 and not self.alreadyUpdated('BidPrice', 'BidSize', self._bestBid):
for key,prop in zip(list(self._bestBid.keys()),['LocalTime','BidPrice', 'BidSize']):
setattr(self._bestBid,str(key),getattr(self.algo.Securities[self.Mapped], prop))
return self._bestBid
@property
def bestAsk(self):
# Only updating if Ask price and Size changed so as to keep the original Time of last update
if int(getattr(self.algo.Securities[self.Mapped], 'AskSize')) != 0 and not self.alreadyUpdated('AskPrice', 'AskSize', self._bestAsk):
for key,prop in zip(list(self._bestBid.keys()),['LocalTime','AskPrice', 'AskSize']):
setattr(self._bestAsk,str(key),getattr(self.algo.Securities[self.Mapped], prop))
return self._bestAsk
# This is not entirely correct as someone may have canceled and others may have added a bid with net being affect 0
def alreadyUpdated(self, price, size, dictionary):
return getattr(self.algo.Securities[self.Mapped], price) == dictionary.price and getattr(self.algo.Securities[self.Mapped], size) == dictionary.size
# TO DO: Incorrect since when we BackwardsPanamaCanal for DataNormalizationMode, trade prices for older contracts will be adjusted unless we use self.Mapped just like Quotes - Ask/Bid - getattr(self.algo.Securities[self.Mapped], prop))
# However, self.Securities option doesn't take self.Mapped while ticks don't seem to have a property that matches last traded price. LastPrice gives BestBid/Ask
@property
def trade(self):
if self.algo.CurrentSlice.Ticks.ContainsKey(self.Symbol) and self.algo.CurrentSlice.Ticks[self.Symbol] is not None:
ticks = self.algo.CurrentSlice.Ticks[self.Symbol]
for tick in ticks:
tick_type = getTickType(tick.TickType)
if tick_type == 'Trade' and int(getattr(tick, 'Quantity')) != 0:
for key,prop in zip(list(self._trade.keys()),['Time','Price','Quantity']):
setattr(self._trade,str(key),getattr(tick, prop))
# setattr(self._trade,str(key),getattr(self.algo.Securities[self.Mapped], prop))
# end_time = self.algo.Time
# start_time = end_time - timedelta(seconds=1)
# self.History[Tick](self.Mapped, start_time,end_time, Resolution.Tick)
return self._trade
@property
def WAP(self):
if None not in (self.bestAsk.size,self.bestBid.size):
self._WAP = round(((self.bestBid.price * self.bestAsk.size) + (self.bestAsk.price * self.bestBid.size))/(self.bestBid.size + self.bestAsk.size),1)
return self._WAP
class dotdict(dict):
"""dot.notation access to dictionary attributes"""
__getattr__ = dict.get
__setattr__ = dict.__setitem__
__delattr__ = dict.__delitem__
from AlgorithmImports import *
# https://github.com/QuantConnect/Lean/blob/master/Common/Orders/OrderTypes.cs#L87
def get_order_status_name(index):
return {
0: 'New',
1: 'Submitted',
2: 'PartiallyFilled',
3: 'Filled',
4: 'None',
5: 'Canceled',
6: 'None',
7: 'Invalid',
8: 'CancelPending',
9: 'UpdateSubmitted '
}[index]
def get_order_direction_name(index):
return {
0: 'Buy',
1: 'Sell',
2: 'Hold',
}[index]
def get_order_type_name(index):
return {
0: 'Market',
1: 'Limit',
2: 'StopMarket',
3: 'StopLimit',
4: 'MarketOnOpen',
5: 'MarketOnClose',
6: 'OptionExercise',
7: 'LimitIfTouched'
}[index]
# This dictionary is Used to detrmine the Upside of Lumber Prices versus Yesterday's Settle:
# From there we assume, TP,
def get_upside(index):
return {
'1.5%':10,
'2%':12,
'3%':15
}[index]
# https://www.quantconnect.com/docs/v2/writing-algorithms/securities/key-concepts
def getSecurityType(index):
return {
0: 'Base',
1: 'Equity',
2: 'Option',
3: 'Commodity',
4: 'Forex',
5: 'Future',
6: 'Cfd',
7: 'Crypto',
8: 'FutureOption',
9: 'Index',
10: 'IndexOption',
'CryptoFuture': 'CryptoFuture'
}[index]
def getDataType(index):
return {
0: 'Base',
1: 'TradeBar',
2: 'Tick',
3: 'Auxiliary', # Data associated with an instrument
4: 'QuoteBar',
5: 'OptionChain',
6: 'FuturesChain'
}[index]
def getTickType(index):
return {
0: 'Trade',
1: 'Quote',
2: 'OpenInterest' # Open Interest type tick object (for options, futures)
}[index]
from AlgorithmImports import *
from functools import wraps
""""
Description: class to handle all the indicators
"""
class IndicatorClass:
def __init__(self,algorithm,indicator,symbol,update_on_stream=False,update_on=TradeBar,resolution=Resolution.Daily):
"""
Parimeters:
algorithm:QCAlgorithm Instance
Indicator:QCAlgorithm.Indicator Instance
symbol:str (symbol of security )
update_on_stream:(bool) default false if true will take updates from streamed data.
update_on:QCAlgorithm.Datatype(type of data to update the indicator used for historical updates)
resolution:QcAlgorithm.Resolution default:Resolution.daily
"""
self.algo=algorithm
self.indicator=indicator
self.symbol=symbol
self.period=indicator.WarmUpPeriod
self.updateOn=update_on
self.resolution=resolution
self.stream=update_on_stream
self.indicator.Updated += self.updateHandler
#updates indicator using historical data
def updateHistorically(self):
self.algo.Debug(f"fetching history for {self.symbol}")
history= self.algo.History[self.updateOn](self.symbol, self.period,self.resolution)
for bar in history:
self.algo.Debug(f" bar is {bar}")
self.indicator.Update(bar)
# resets the indicator and changes the symbol of indicator if mapped
def resetIndicator(self,mapped=None):
if mapped:
self.indicator.Reset()
self.symbol=mapped
else:
self.indicator.Reset()
#handler function for indicator update events.
def updateHandler(self,indicator: object, indicator_data_point: IndicatorDataPoint):
if indicator.IsReady:
pass
# self.algo.Debug(f"Indicator value for symbol{self.symbol}: {indicator_data_point.Value} @ {self.algo.Time}")
#method to update indicators with stream data from onData,Update or consildator methods
def update(self,data):
self.indicator.Update(data)
#value of indicator
@property
def value(self):
if not self.stream:
self.updateHistorically()
return self.indicator.Current.Value
else:
if not self.indicator.IsReady:
self.updateHistorically()
self.indicator.Current.Value
#region imports
from AlgorithmImports import *
#endregion
# NEED HELP
# 1. Done: Get Yesterday's Close Price for Lumber Contracts - 2 ways to acheive.
# 1a. Use history fuction everyday - Works Fine
# 2a. Use Identity Indicator - Its partially correct when using Resolution.Daily but incorrect On days when the contract rolls.
# i.e. it still shows yesterday's close for previous contract on RollOverDays. Also would require a day or 2 of warmuP
# Secondly, if I use Resolution.Tick in Lumber Contract definition, it shows 2 days ago prices when called before market Open
# 3. Find trading days when both Nasdaq and Lumber were open (keeping in mind yesterday's data) and only trade when both are available atleast for backtesting
# https://www.quantconnect.com/docs/v2/writing-algorithms/securities/asset-classes/futures/market-hours/cme/mnq
# https://www.quantconnect.com/docs/v2/writing-algorithms/securities/asset-classes/futures/market-hours/cme/lbs
# 4. Nasdaq data not available on first day since it gets added at 2300 hours while Lumber gets added at 00 hours.
# 5. Didn't check if Nasdaq prices are fine or if Nasdaq yesterday contract prices match the rolled over contract
# 6. Whats the difference between getting Bids & Asks from ticks like below - Is the first one Time&Sales (for quotes) while second one gives Best Bids/Asks?
# ticks = self.algo.CurrentSlice.Ticks[self.Symbol]
# for tick in ticks:
# tick.AskPrice
# versus getting bids/asks from self.Securities[security].AskPrice
# 7. # Questions on quoteEvent (TickQuoteBarConsolidator):
# 1. Does this tell you total number of best bids/asks at any moment in time or are these just seperate orders and their corresponding sizes?
# 2. Quotes are for mapped rolled contracts?
# 3. Also these are not all quotes as placed but only Top Bids/Asks. So for example, they are only generated if someone has topped a previous bid/offer?
# Some other references:
# 1. Tick Data related
# https://www.quantconnect.com/docs/v2/writing-algorithms/securities/asset-classes/futures/handling-data#07-Ticks
# LastPrice - Alias for "Value" - the last sale for this asset. Same as Price
# IsFillForward - True if this is a fill forward piece of data
# 'Time','EndTime' - for Trade Tick is same as self.Time but for time use this property only
# 'Symbol' - Not available but has other sub properties
# ,'SaleCondition','Suspicious' - Not relevant
# tickTradeProps = ['Price','Quantity']
# tickQuoteProps = ['BidPrice','BidSize','AskPrice','AskSize']
# tickOIProps = ['Value'] # Not Getting Data for OpenInterest
# Other Notes to self:
# 1. Since Resolution is for Ticks, we won't be getting Trade & Quote Bars
# 2. 'LocalTime': Local time for this market
# 3. Properties of self.Securities - https://www.quantconnect.com/docs/v2/writing-algorithms/securities/properties#02-Security-Properties
# 4. Nasdaq - closes at 4 pm & opens at 8:30 Chicago, Also trades at 11 pm till not sure when?
# OnSecuritiesChanged: Date:2022-12-19 23:00:00, security:MNQ Y6URRFPZ86BL
# 5. # Links to documentation pertaining
# Time Modeling
# https://www.quantconnect.com/docs/v2/writing-algorithms/key-concepts/time-modeling/timeslices#03-Properties
# Futures - Handling Data
# https://www.quantconnect.com/docs/v2/writing-algorithms/securities/asset-classes/futures/handling-data
# To get the current Slice object, define an OnData method or use the CurrentSlice property of your algorithm (outside of the OnData method).
# If the data object doesn't contain any market data but it contains auxiliary data, the slice.ContainsKey(symbol) method can return true while slice[symbol] returns None.
# Indicator Help
# In QuantConnect/Lean, we have shortcut methods for indicators, they belong to the QCAlgorithm class (use self) and name are upper-cased. These helper methods create a new instance of a indicator object and hook it up to a data consolidator so that the indicator is automatically updated by the engine.
# Since these methods create a new instance, we just should only to call it once (normally in Initialize) and assign it to a class variable to be accessed throughout the algorithm.
# 1. AverageTrueRange:
# https://www.quantconnect.com/forum/discussion/7972/using-atr-and-other-039-complex-039-indicators-with-history/p1
# https://www.quantconnect.com/forum/discussion/11457/warmup-atr-with-history
# https://www.quantconnect.com/docs/v2/writing-algorithms/indicators/supported-indicators/average-true-range
# 2. Aplying Indicators to the continuous contract prices:
# we don't need to reset the indicator when the contract rolls over.
# The indicator will just continue to reflect the prices of the continuous contract.
# 3. Aplying Indicators to Specific Mapped contract prices:
# reset the indicator if we were applying the indicator to a specific contract
# and then we switch to a new contract after the rollover.
# 4. Common Mistakes - Creating Automatic Indicators in a Dynamic Universe: - You can't currently remove the consolidator that LEAN creates to update automatic indicators.
# If you add consolidators to a dynamic universe, the consolidators build up over time and slow down your algorithm.
# To avoid issues, if you algorithm has a dynamic universe, use manual indicators.
# However, I think since expired contracts expire (say 15 days after we roll over), the consolidators won't have data to fill so won't be as slow?
# https://www.quantconnect.com/docs/v2/writing-algorithms/indicators/manual-indicators#05-Automatic-Updates
# 5. Automatic Updates for Manual Indicators: To configure automatic updates, create a consolidator and then call the RegisterIndicator method.
# If your algorithm has a dynamic universe, save a reference to the consolidator so you can remove it when the universe removes the security.
# If you register an indicator for automatic updates, don't call the indicator's Update method or else the indicator will receive double updates.
# 6. Looks like if you need an Indicator before market open, then may have to use manual indicators??
# Since "Once your algorithm reaches the EndTime of a data point, LEAN sends the data to your OnData method.
# For bar data, this is the beginning of the next period. "
# How about if we use consolidated data in Automatic Indicators - consolidated from much lower time frame
# Not sure but it may also be that using the above solution, the indicator keeps getting updated through the day
# The consolidators can update your indicators at each time step or with aggregated bars. By default, LEAN updates data point indicators with the close price of the consolidated bars, but you can change it to a custom data field.
# region imports
from AlgorithmImports import *
from AlphaModel import *
from DataClass import *
import math
class NasdaqStrategy(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2022,12,23)
# self.SetStartDate(2023,1,3)
self.SetEndDate(2023,1,5)
self.SetCash(100000)
self.SetTimeZone(TimeZones.Chicago) # If TimeZone is Chicago - Algo Time and Data End Time are same at 1800 hours
# self.SetWarmUp(timedelta(days=2)) # What is this for ?
self.Data = {}
# FutureSymbols = {'Lumber':Futures.Forestry.RandomLengthLumber,'Nasdaq':Futures.Indices.NASDAQ100EMini}
FutureSymbols = {'Lumber':Futures.Forestry.RandomLengthLumber}
#dict to hold futures
self.futures={}
for key, value in FutureSymbols.items():
dataMappingMode_ = DataMappingMode.FirstDayMonth if key == 'Lumber' \
else DataMappingMode.OpenInterest if key == 'Nasdaq' else DataMappingMode.LastTradingDay
# Use BackwardsPanamaCanal for DataNormalizationMode when "Trade" PROPERTY is able to map to Mapped contract
future = self.AddFuture(value, Resolution.Tick,dataMappingMode = dataMappingMode_, contractDepthOffset=0,
dataNormalizationMode = DataNormalizationMode.Raw, extendedMarketHours=True, fillDataForward = True)
self.futures[key]=future
# # Set alpha model - This also gives you access to the AlphaModel Instance
# self.alpha = NasdaqAlpha(self.Data['Lumber'])
# self.SetAlpha(self.alpha)
# Printing Indicator's data
self.Schedule.On(self.DateRules.EveryDay(self.futures['Lumber'].Symbol), self.TimeRules.AfterMarketOpen(self.futures['Lumber'].Symbol, -1), self.beforeLumberOpen)
# self.Schedule.On(self.DateRules.EveryDay(), self.TimeRules.At(8,55), self.beforeLumberOpen)
# Temporary
self.entryTimeStart = self.Time.replace(hour=8, minute=59, second=0, microsecond=55000)
self.entryTimeEnd = self.Time.replace(hour=9, minute=3, second=0, microsecond=0)
def beforeLumberOpen(self):
# if self.IsWarmingUp: return
# Works - Get Indicator's data - For mapped Indicators, keeping in mind Rolled Over Contracts
for symbol in self.Data.keys():
pass
self.Debug(f"Yester Time:{self.Time}, {self.Data[symbol].Mapped}.yesterday:{self.Data[symbol].yesterday}")
# self.Debug(f"Calculated Yesterday's High-Low:{self.Time}, {round(float(self.Data[symbol].yesterday.high) - float(self.Data[symbol].yesterday.low),1)}")
# self.Debug(f"___ATR Yesterday:{round(self.Data[symbol]._ATR.TrueRange.Current.Value,1)}, ATR {self.Data[symbol].atrPeriod} days:{round(self.Data[symbol]._ATR.Current.Value,1)}")
self.Debug(f"ATR property:{self.Data[symbol].ATR}, {self.Time}")
# Works
# for security in self.ActiveSecurities.Values:
# self.Debug(f"self.Time:{self.Time} ActiveSecurity:{security}")
def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
# Only runs at the start of program not when Contract rolls
for security in changes.AddedSecurities:
if len(changes.AddedSecurities) == 0: return
if str(security) == str(self.futures["Lumber"].Symbol):
self.Data["Lumber"] = DataClass(self,self.futures["Lumber"]) # Initiating DataClass for each Future & Passing our instance of QCAlgorithm Class
def OnEndOfDay(self, symbol):
# self.Debug(f"OnEndOfDay self.Time:{self.Time}, Symbol:{self.Data['Lumber'].Mapped.Value}")
pass
def OnData(self, data):
# https://www.quantconnect.com/docs/v2/writing-algorithms/historical-data/warm-up-periods#03-Warm-Up-Vs-Reality
if self.IsWarmingUp: return
# Below code important to compare Ticks with whats reported from DataClass
tickTradeProps = ['LastPrice','Quantity']
tickQuoteProps = ['BidPrice','BidSize','AskPrice','AskSize']
# tickOIProps = ['Value'] # Not Getting Data for OpenInterest
for security in data.Keys:
# if self.Time < self.entryTimeStart or self.Time > self.entryTimeEnd: return # temporary
if data.Ticks.ContainsKey(security) and data.Ticks[security] is not None:
ticks = data.Ticks[security]
for tick in ticks:
tick_type = getTickType(tick.TickType)
if tick_type == 'Trade' and int(getattr(tick, 'Quantity')) != 0:
# For Trade Data
dic = {k:getattr(tick, k) for k in tickTradeProps if getattr(tick, k) is not None}
# self.Debug(f"TradeTick:{self.Time} {dic}")
elif tick_type == 'Quote':
# For Best Bid & Ask
dic2 = {k:getattr(self.Securities[security], k) for k in tickQuoteProps}
# self.Debug(f"QuoteTick:{self.Time}, security:{security}, {dic2}")
for symbol in self.Data.keys():
self.Data[symbol].Update(data)
# if self.Time < self.entryTimeStart or self.Time > self.entryTimeEnd: return # temporary
# self.Debug(f"++BestBid:{symbol}:{self.Time}:{self.Data[symbol].bestBid}")
# self.Debug(f"^^BestAsk:{symbol}:{self.Time}:{self.Data[symbol].bestAsk}")
# self.Debug(f"**trade:{symbol}:{self.Time}:{self.Data[symbol].trade}")
# self.Debug(f"**WAP:{symbol}:{self.Time}:{self.Data[symbol].WAP}")
# WORKS - Updated several times in a day - Doesn't give last traded price - gives bid or ask
# self.Debug(f"Todays ____Close:{self.Securities[self.Data['Lumber'].Mapped].Price}, Date:{self.Time}")
# Works
# if data.Ticks.ContainsKey(self.Data['Lumber'].Symbol):
# if data.Ticks[self.Data['Lumber'].Symbol] is not None:
# self.Debug(f"**MAIN** Date:{self.Time}, MappedValue:{self.Data['Lumber'].Mapped.Value}, MappedID:{self.Data['Lumber'].Mapped.ID}, ")
# https://www.quantconnect.com/docs/v2/writing-algorithms/datasets/quantconnect/us-futures-security-master#05-Data-Point-Attributes
for symbol in self.Data.keys():
if data.SymbolChangedEvents.ContainsKey(self.Data[symbol].Symbol):
symbolChangedEvent = data.SymbolChangedEvents[self.Data[symbol].Symbol]
self.Debug(f"In MAIN Symbol changed: {symbolChangedEvent.OldSymbol} -> {symbolChangedEvent.NewSymbol} \
EndTime:{symbolChangedEvent.EndTime} DataType:{getDataType(symbolChangedEvent.DataType)}, Expiry: {self.Securities[self.Data[symbol].Mapped].Expiry}")
# When your algorithm stops executing, LEAN calls the OnEndOfAlgorithm method.
def OnEndOfAlgorithm(self) -> None:
# self.Debug(f"self.Alpha.securities:{self.alpha.securities}")
pass
def OnWarmUpFinished(self) -> None:
self.Debug(f"Algorithm Ready@{self.Time}")
pass