| 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.436 Tracking Error 0.14 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
# 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
self.Indicators = {} #
# self._ATR = self.algo.ATR(self.Symbol, self.atrPeriod, MovingAverageType.Simple, Resolution.Daily)
# self.Reset()
# Manual Indicator
self._ATR = AverageTrueRange(self.atrPeriod, MovingAverageType.Simple)
# The time period is relative to the data time zone, not the algorithm time zone.
# https://www.quantconnect.com/docs/v2/writing-algorithms/consolidating-data/consolidator-types/time-period-consolidators#04-Consolidate-Trade-Ticks
self.tradeEvent = TickConsolidator(timedelta(days=1)) # This is TradeBar Consolidator for Ticks
# self.algo.RegisterIndicator(self.Symbol, self._ATR, self.consolidator) # For Automatic Updates - See Notes
# self.tradeEvent.DataConsolidated += (lambda _, bar : self._ATR.Update(bar))
# self.algo.SubscriptionManager.AddConsolidator(self.Symbol, self.tradeEvent)
# Manually Updating Manual Indicator
self.tradeEvent.DataConsolidated += self.tradeEventHandler
self.algo.SubscriptionManager.AddConsolidator(self.Symbol, self.tradeEvent)
# Attaching an event with
self._ATR.Updated += self.update_event_handler
# 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)
def tradeEventHandler(self, sender: object, Tbar: TradeBar) -> None:
self._ATR.Update(Tbar)
# self.algo.Debug(f"Updated Consolidated Tbar For Atr,{self.algo.Time} Open:{float(Tbar.Open)}, High:{float(Tbar.High)}, Low {float(Tbar.Low)}, Close:{float(Tbar.Close)}, Price:{float(Tbar.Price)}")
def update_event_handler(self, indicator: object, indicator_data_point: IndicatorDataPoint) -> None:
if indicator.IsReady:
self.algo.Debug(f"Indicator Value: {indicator_data_point.Value} @ {self.algo.Time}")
def Update(self, data):
if data.SymbolChangedEvents.ContainsKey(self.Symbol):
# https://www.quantconnect.com/docs/v2/writing-algorithms/indicators/key-concepts#09-Reset-Indicators
self._ATR.Reset() # This will load History if the contract switched
self.algo.SubscriptionManager.AddConsolidator(self.Symbol, self.tradeEvent)
# After Updaing check if self._ATR.IsReady
# if data.TradeBars.ContainsKey(self.symbol):
# self._ATR.Update(input)
# Won't work since we won't be getting bars - only ticks, I think
# if data.Bars.ContainsKey(self.Symbol):
# self.algo.Debug(f"worked")
# self._ATR.update(data.Bars[self.Symbol])
# Should be moved to Indicator Class
# ToDo: Issues - Make it work for self.Symbol - for now it only works for self.Mapped
@property
def ATR(self):
# self.algo.Debug(f"self._ATR.IsReady:{self._ATR.IsReady} at {self.algo.Time}")
self.algo.Debug(f"self._ATR.IsReady:{self._ATR.IsReady} at {self.algo.Time}")
if not self._ATR.IsReady:
self._ATR.Reset()
# Load History if Indicator not ready yet
historyATR = self.algo.History[TradeBar](self.Mapped,7, 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)[::-1])
for bar in historyATR.itertuples():
self.algo.Debug(f"bar.date:{bar}")
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"# of samples{self._ATR.Samples}")
return self._ATR.Current.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
# Can even combine with ATR since both use history
def updateYesterday(self):
# Note that Volume doesn't match with reported in our Database (atleast for Lumber)
if self.Mapped is not None:
# ctrctHistory = self.algo.History(self.Mapped, 2, 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)
historyATR = self.algo.History[TradeBar](self.Mapped, self.atrPeriod, 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 updateYesterday is called even on Non trading days
self.yesterday[str(property)] = getattr(bar, property)
# Happens every morning
# if not self._ATR.IsReady:
# for bar in historyATR.itertuples():
# self.algo.Debug(f"bar.date:{bar}")
# 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)
# 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 wille 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 *
class IndicatorClass:
def __init__(self, Algorithm, continuous_contract):
pass#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.
# 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.
# region imports
from AlgorithmImports import *
from AlphaModel import *
from DataClass import *
import math
class NasdaqStrategy(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2022,12,26)
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 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.Data[key] = DataClass(self, future) # Initiating DataClass for each Future & Passing our instance of QCAlgorithm Class
# 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)
# 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 Yesterdays' data 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
self.Debug(f"In Main: OnSecuritiesChanged: Date:{self.Time}, security:{changes.AddedSecurities[0].Mapped}")
def OnEndOfDay(self, symbol):
security=self.Securities[symbol]
self.Debug(f"OnEndOfDay self.Time:{self.Time}, Symbol:{security.Symbol},High:{security.High},Low:{security.Low},Close{security.Close},Open:{security.Open},Price:{security.Price}")
mappedSecurity=self.Securities[security.Mapped]
self.Debug(f" Mapped OnEndOfDay self.Time:{self.Time}, Symbol:{mappedSecurity.Symbol},High:{mappedSecurity.High},Low:{mappedSecurity.Low},Close{mappedSecurity.Close},Open:{mappedSecurity.Open},Price:{mappedSecurity.Price}")
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