from typing import Dict
from AlgorithmImports import *
class Centurion(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2021, 3, 11) # Set Start Date
self.SetCash(100000) # Set Strategy Cash
# self.AddEquity("SPY", Resolution.Minute)
'''
Would it be easier to have a symboldata class?
What would it contain?
- initialization of Data
- dictionaries for stoploss, option types, rebalance weights
'''
# Assets=====================================
self.dow = self.AddEquity("DJI", Resolution.Minute).Symbol
self.bnd = self.AddEquity("BND", Resolution.Minute).Symbol
spy = self.AddEquity("SPY", Resolution.Minute)
spy.SetDataNormalizationMode(DataNormalizationMode.Raw)
self.spy = spy.Symbol
self.spycontract: Symbol = None
tqqq = self.AddEquity("TQQQ", Resolution.Minute)
tqqq.SetDataNormalizationMode(DataNormalizationMode.Raw)
self.tqqq = tqqq.Symbol
self.tqqqcontract: Symbol = None
soxl = self.AddEquity("SOXL", Resolution.Minute)
soxl.SetDataNormalizationMode(DataNormalizationMode.Raw)
self.soxl = soxl.Symbol
self.soxlcontract: Symbol = None
tecl = self.AddEquity("TECL", Resolution.Minute)
tecl.SetDataNormalizationMode(DataNormalizationMode.Raw)
self.tecl = tecl.Symbol
self.teclcontract: Symbol = None
#Scheduled Events============================================
self.Schedule.On(self.DateRules.MonthStart("SPY"), self.TimeRules.AfterMarketOpen("SPY", 1), self.monthlyRebalance)
self.Schedule.On(self.DateRules.EveryDay("SPY"), self.TimeRules.AfterMarketOpen("SPY", 1), self.captureOpening)
#Variables needed for stoploss
self.stoplosshold = 0
self.dailyOpen: Dict[Symbol,float] = {
self.spy: 0,
self.dow:0,
self.bnd:0
}
# Rebalancing Weights
self.weights = {
self.spy: .25,
self.dow: .2,
self.bnd: .4
}
# Underlying & Options (UNOs)
self.uNos = {
self.spy : self.spycontract,
self.tqqq : self.tqqqcontract,
self.tecl : self.teclcontract,
self.soxl : self.soxlcontract
}
# I dont really know what these do
def CheckOptions(self):
for underlying,contract in self.uNos.items():
if contract and self.Securities[contract].IsTradable:
if self.Portfolio[contract].UnrealizedProfitPercent > .1:
self.MarketOrder(underlying, -self.Portfolio[contract].Quantity * 100)
self.ExerciseOption(contract, 1)
self.RemoveSecurity(contract)
contract = None
def ComputeInstrinsicValue(self, symbol:Symbol, underlying:Symbol):
sign = 1 if symbol.ID.OptionRight == OptionRight.Call else -1
instrinsic = max(0,
sign * (self.Securities[symbol].Price - symbol.ID.StrikePrice)
)
return instrinsic
def OnData(self, data):
'''OnData event is the primary entry point for your algorithm. Each new data point will be pumped in here.
Arguments:
data: Slice object keyed by symbol containing the stock data
'''
if self.IsWarmingUp:
return
if self.Portfolio.Invested:
self.stoploss(data)
if not self.Portfolio.Invested:
self.monthlyRebalance()
self.setPuts()
self.timeDecayCheck()
#A Method to purchase options
self.buyPuts()
# **MISSING** method to exercise options
self.exerciseLoki()
def setPuts(self):
'''
I want to purchase a long puts for every key in dictionary uNos ((u)underlying (N)and (os)Options)
'''
for underlying in self.uNos:
targetStrike = (self.Securities[underlying].Price * 0.60) - (self.Securities[underlying].Price * 0.60)%5
contracts = self.OptionChainProvider.GetOptionContractList(underlying, self.Time)
puts = [x for x in contracts if x.ID.OptionRight == OptionRight.Put]
puts = sorted( sorted(puts, key = lambda x: x.ID.Date, reverse = True),
key = lambda x: x.ID.StrikePrice)
puts = [x for x in puts if x.ID.StrikePrice == targetStrike]
puts = [x for x in puts if 270 < (x.ID.Date - self.Time).days <= 420]
if len(puts) == 0:
#self.Log("No Puts")
continue
self.AddOptionContract(puts[0], Resolution.Minute)
self.uNos[underlying] = puts[0]
def timeDecayCheck(self):
#Don't let an options time to expiration be less than 6 months
for option,contract in self.uNos.items():
if (contract.ID.Date - self.Time).days < 180:
self.Liquidate(contract)
self.RemoveSecurity(contract)
contract = None
def buyPuts(self):
for _,contract in self.uNos.items():
if not self.Portfolio[contract].Invested:
self.SetHoldings(contract, 0.05)
def exerciseLoki(self):
for underlying,contract in self.uNos.items():
if contract and self.Securities[contract].IsTradable:
if self.Portfolio[contract].UnrealizedProfitPercent > .1:
# I really dont know what this line does
# self.MarketOrder(underlying, -self.Portfolio[contract].Quantity * 100)
self.ExerciseOption(contract, 1)
self.RemoveSecurity(contract)
contract = None
def captureOpening(self):
#Grabs the daily opening price of spy for our stoploss method
if self.IsWarmingUp:
return
for key, value in self.dailyOpen.items():
if self.CurrentSlice.Bars.ContainsKey(key):
self.dailyOpen[key] = self.CurrentSlice[key].Open
self.stoplosshold = 0
def monthlyRebalance(self):
# Rebalance portfolio monthly
if self.IsWarmingUp:
return
for key,value in self.weights.items():
self.SetHoldings(key,value)
def stoploss(self, data):
'''
Stoploss logic:
- If spy drops more than 5% liquidate entire equity portfolio
- Change stoplosshold value to 1, this indicates that the portfolios SL has been hit
and were going to hold until the next trading day
'''
if self.IsWarmingUp:
return
for symbol, weight in self.weights.items():
open = self.dailyOpen[symbol]
#See if any symbol has decreased more than 5% in a given day, liquidate if true and check next one...
if ((self.Securities[symbol].Price-open)/self.Securities[symbol].Price) < -.05:
self.SetHoldings(symbol, 0)
self.stoplosshold = 1
#self.Log('HIT')