| Overall Statistics |
|
Total Orders 12 Average Win 0.10% Average Loss -1.99% Compounding Annual Return -44.095% Drawdown 2.500% Expectancy -0.475 Start Equity 100000 End Equity 98106.32 Net Profit -1.894% Sharpe Ratio -4.242 Sortino Ratio 0 Probabilistic Sharpe Ratio 1.580% Loss Rate 50% Win Rate 50% Profit-Loss Ratio 0.05 Alpha -0.407 Beta -0.01 Annual Standard Deviation 0.096 Annual Variance 0.009 Information Ratio -2.323 Tracking Error 0.191 Treynor Ratio 42.003 Total Fees $193.68 Estimated Strategy Capacity $230000.00 Lowest Capacity Asset TCS VL8CJEL7BM91 Portfolio Turnover 3.45% |
# region imports
from AlgorithmImports import *
# endregion
class SymbolData:
def __init__(self, symbol, algo):
self.symbol = symbol
self.algo = algo
self.bar_window = RollingWindow[TradeBar](2) #COULD make this closes only....
self._state = 0
self.db = self.algo.debug_lvl
def Update(self, bar):
self.bar_window.Add(bar) # self.tradeBarWindow.Add(data["SPY"]) --- To Add (Then ref as if a BAR)
@property
def CheckStates(self):
#IF crossed above, go from 0 to 1
if self.state == 0:
if self.CrossedAboveLvl:
self.state = 1
if self.db > 0:
self.algo.Debug(f'S0 -> S1 --- {self.symbol}')
#IF crossed BELOW, go from 1 to 2
if self.state == 1:
if self.CrossedBelowLvl:
self.state = 2
if self.db > 0:
self.algo.Debug(f'S1 -> S2 --- {self.symbol}')
return self.state
@property
def IsReady(self):
return self.bar_window.IsReady
@property
def CrossedAboveLvl(self):
if not self.IsReady:
return False
prev = self.bar_window[1].Close
curr = self.bar_window[0].Close
if curr > self.algo.high_cross_lvl and prev <= self.algo.high_cross_lvl:
if self.db >= 2:
self.algo.Debug(f'{self.symbol} -- {curr} > {self.algo.high_cross_lvl}')
return True
return False
@property
def CrossedBelowLvl(self):
if not self.IsReady:
return False
prev = self.bar_window[1].Close
curr = self.bar_window[0].Close
if curr < self.algo.low_cross_lvl and prev >= self.algo.low_cross_lvl:
if self.db >= 2:
self.algo.Debug(f'{self.symbol} -- {curr} < {self.algo.low_cross_lvl}')
return True
return False
# -------------------- Added a Property for state (increased safety)
@property
def state(self):
return self._state
@state.setter
def state(self, new):
if new in [0,1,2]:
self._state = new
# region imports
from AlgorithmImports import *
# endregion
from SymbolData import SymbolData
# Logic + Params Explained:
# 1. Universe: Select ALL symbols with price < $1, with MAX price of last (lookback) days < $1 ( < max_beg_price)
# 2. Check State Updates on Universe:
# S0 - Below max_beg_price for past 30 (lookback) days -- (Inside Universe)
# S1 - Crossed above 1.40 (high cross_lvl)
# S2 - Crossed below 1.00 (again) (low cross lvl)
#
# 3. Submit Stop Order 4 ticks above beg price (entry_offset_ticks + max_beg_price)
# 4. IF Entry filled, Submit Stop + Target as well.
# (stop_offset_ticks & tgt_ticks)
# 5. If day ended, cancel all pending orders, all symbols go back to state 0
## 7.31.2024 -- added eodx
class MeasuredRedCoyote(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2024, 7, 20) #Changed to test entries + stops
# self.SetEndDate(2024, 8,31)
self.SetCash(100000)
self.AddEquity("SPY", Resolution.Minute)
self.AddUniverse(self.SelectCoarse, self.SelectFine)
# ---------------- Parameters -------------- #
self.n_shares = 10 * 1000 #10k
self.lookback = 30
self.max_beg_price = 1.00 # Price must start below, and be below this for lb days (and cross below again for state 2)
self.high_cross_lvl = 1.20 # Price must cross this to enter state 1
self.low_cross_lvl = self.max_beg_price
self.STOP_OFF = True # TURNS STOP LOSS OFF
self.entry_offset_ticks = 4 # Number of ticks to submit stop order (ABOVE max_beg_price)
self.stop_offset_ticks = 10 # Number of ticks for STOP LOSS (10 = .1 --> entry - .1)
self.tgt_1 = 4.50 #4.50 price
self.tgt_2 = 8.00 # 8 price
self.tgt_3 = 12.0
self.tgt_4 = 14.0
self.limit_entry_on = True # Turn Stop Limit Entries on (Vs Stop Market)
self.max_slip_ticks = 4 # IF using Entry Limit, MAX ticks off trigger price to fill.
self.extended_hours = False
self.debug_lvl = 2 # Prints details to log (State counts, Universe, etc)
# enable ETH -- via universe settings.
# https://www.quantconnect.com/docs/v2/writing-algorithms/algorithm-framework/universe-selection/universe-settings
if self.extended_hours:
self.universe_settings.extended_market_hours = True
# ---------------- Declarations ------------ #
self.symbol_list = []
self.stop_orders = {}
self.symbol_data = {}
# ---------------- Scheduled Events
self._test_entries = False
# TO TEST ENTRIES + EXITS (Sets all universe to state 2)
# schedule eod event.
self.Schedule.On(
self.DateRules.EveryDay("SPY"),
self.TimeRules.BeforeMarketClose("SPY", 5),
self.EODX
)
def EODX(self):
self.liquidate(tag="EOD Exit")
def OnEndOfDay(self):
data = self.CurrentSlice
# NOTE -- this was NOT running before.... but now I'm adding EODX to it
#Reset instances to state 0 (Begin)
# TO RM same day req, RM this block
for symbol, obj in self.symbol_data.items():
obj.state = 0
# Clear Out Old Pending orders (IF NOT INVESTED) -- IF Invested, KEEP the orders.
not_invested = [kvp.Key for kvp in self.Portfolio if not kvp.Value.Invested]
for i in not_invested:
self.Transactions.CancelOpenOrders(i)
def SelectCoarse(self, coarse):
filtered = [ x for x in coarse if x.Price < self.max_beg_price]
# keep ONLY symbols that have been BELOW 1 for last 30 days.
# (Check if MAX of 30 days is > 1.0)
_symbols = [x.Symbol for x in filtered]
history = self.History(_symbols, self.lookback, Resolution.Daily)
chosen = []
for sym in _symbols:
try:
h = history.loc[sym]
except:
self.Debug(f'Error with hist.loc[{sym}]')
max_in_lb = h.high.max()
if max_in_lb < self.max_beg_price:
chosen.append(sym)
if self.debug_lvl > 0:
self.Debug(f'Universe ( {len(chosen)} symbols ) ---- {[str(i) for i in chosen]}')
self.symbol_list = chosen
return self.symbol_list
def SelectFine(self, fine):
return [f.Symbol for f in fine]
def OnData(self, data):
data = data.Bars
if not data.ContainsKey("SPY"):
self.Debug(f'Returning out -- No Data for SPY. (No data period)')
return
self.ClearErrors() #Clear up any remainders...
has_data = [i for i in self.symbol_list if data.ContainsKey(i)]
# Begin looking for various states!! (rolling windows to check CROSS)
for symbol in has_data:
# Check if FREE margin ? for 1 mroe entry?
# approx_margin = self.n_shares * 1
# if self.Portfolio.MarginRemaining < approx_margin: continue
# CHECK if already invested ---
if self.Portfolio[symbol].Invested: continue
# CHECK if already PENDING ----
openOrders = self.Transactions.GetOpenOrders(symbol)
if len(openOrders) > 0: continue
# Check States of Symbols
if symbol not in self.symbol_data:
self.symbol_data[symbol] = SymbolData(symbol, self) #__init__(self, symbol, algo):
bar = data[symbol]
self.symbol_data[symbol].Update(bar)
_state = self.symbol_data[symbol].CheckStates
if _state == 2:
entry_price = self.low_cross_lvl + (self.entry_offset_ticks / 100)
remaining_usd = self.Portfolio.MarginRemaining
#close = self.Securities[symbol].Close
close = bar.Close
if close == 0:
continue
# shares = int(remaining_usd / close) #Set Position Size.... (COULD be fixed? )
shares = self.n_shares
## ------------- REAL Entries -------------- ##
if self.limit_entry_on:
entryTicket = self.StopLimitOrder(symbol, shares, entry_price, entry_price + self.max_slip_ticks * .01, "Entry")
self.Debug(f'Entry -- Stop Limit Order Submitted -- {symbol}')
else:
entryTicket = self.StopMarketOrder(symbol, shares, entry_price, "Entry")
self.Debug(f'Entry -- Stop Market Order Submitted -- {symbol}')
self.Debug(f'Submitted Entry Order -- {symbol} @ {entry_price}')
if self.debug_lvl >= 2:
state_1 = [k for k,v in self.symbol_data.items() if v.state == 1]
state_2 = [k for k,v in self.symbol_data.items() if v.state == 2] #SETTING TO STATE 0 UPON ENTRY ORDER, so this likely wont show much / any.
self.Debug(f'S1 : {len(state_1)} ---- S2 : {len(state_2)}')
def ClearErrors(self):
#Clear ALL barely invested... (Remainders)
barely_invested = [kvp.Key for kvp in self.Portfolio if kvp.Value.Invested and kvp.Value.Quantity < 10] #Tiny remainder shares.... way below 1/4 of 10k
for symbol in barely_invested:
self.Liquidate(symbol)
is_short = [kvp.Key for kvp in self.Portfolio if kvp.Value.Invested and not kvp.Value.IsLong]
for symbol in is_short:
self.Liquidate(symbol)
not_invested = [kvp.Key for kvp in self.Portfolio if not kvp.Value.Invested]
#Clear all NOT invested (IF theres a leftover non - entry order, cancel it.)
for symbol in not_invested:
openOrders = self.Transactions.GetOpenOrders(symbol)
#CHECK if leftover targets / stops (NOT entries)
for i in openOrders:
if i.Tag != "Entry":
self.Transactions.CancelOpenOrders(symbol)
def OnOrderEvent(self, orderEvent):
"""
Manages OCO like behavior of STP vs Trail orders.
Entry orders are ignored -- returns out.
IF stop FILLED -- CANCEL Trail
if Trail filled -- Cancel Stop
## CAN ALSO USE DIRECTION --> OrderDirection.Buy
"""
#ONLY concerned with FILLED orders. Wait on partials, etc.
if orderEvent.Status != OrderStatus.Filled:
return
order_symbol = orderEvent.Symbol
oid = orderEvent.OrderId
order = self.Transactions.GetOrderById(oid)
shares = orderEvent.AbsoluteFillQuantity
entry_price = orderEvent.FillPrice
#dir = orderEvent.Direction
fill_price = orderEvent.FillPrice
if order.Tag == "Entry":
#RESET to state 0 upon entry : )
self.symbol_data[order_symbol].state = 0
self.Debug(f'Entry Filled -- {order_symbol} @ {fill_price} -- Submitting OCO Stop & Tgt')
## ------------------ STOP LOSS ----------------- ##
if not self.STOP_OFF:
stopTicket = self.StopMarketOrder(order_symbol, -shares, entry_price - (self.stop_offset_ticks / 100), "Stop")
self.stop_orders[order_symbol] = stopTicket #ADD to dict...
if self.debug_lvl >= 1:
self.Debug(f'SL Submitted -- {order_symbol} @ {entry_price - (self.stop_offset_ticks / 100)}')
## ----------------- Profit Targets ------------------ ##
tgt1 = self.LimitOrder(order_symbol, -int(shares / 4), self.tgt_1, "TGT")
tgt2 = self.LimitOrder(order_symbol, -int(shares / 4), self.tgt_2, "TGT")
tgt3 = self.LimitOrder(order_symbol, -int(shares / 4), self.tgt_3, "TGT")
tgt4 = self.LimitOrder(order_symbol, -int(shares / 4), self.tgt_4, "TGT4")
if self.debug_lvl >= 1:
self.Debug(f'TP1 Submitted -- {order_symbol} @ {self.tgt_1}')
self.Debug(f'TP2 Submitted -- {order_symbol} @ {self.tgt_2}')
self.Debug(f'TP3 Submitted -- {order_symbol} @ {self.tgt_3}')
self.Debug(f'TP4 Submitted -- {order_symbol} @ {self.tgt_4}')
return
# (IF NOT ENTRY)
## -------- IF EXIT ORDER FILLED -------- ##
self.Debug(f'Exit Order Fill -- {order_symbol} @ {fill_price} ')
## IF a Target order filled ... (RESET stop quantity) --- IF stop off, ignore
if order.Tag == "TGT" and not self.STOP_OFF:
#self.Debug(f' -------------------------------- TGT HIT (Resetting Quantity for SL)')
new_shares_held = self.Portfolio[order_symbol].Quantity
updateSettings = UpdateOrderFields()
updateSettings.Quantity = -new_shares_held
# order.UpdateOrderQuantity() #Unsure of syntax for this...
#Subtract Manually... (NOT based on new_shares_held via pf object)
#old = self.stop_orders[order_symbol].Quantity
#new = old - shares
self.stop_orders[order_symbol].Update(updateSettings)
#updateSettings.StopPrice = curr_trail
#response = self.TrailOrder.Update(updateSettings)
self.Debug(f"Stop Quantity Updated to {new_shares_held}")
return
# IF a Stop or Final Target Filled (Cancel all remaining)
if order.Tag in ["Stop", "TGT4"]:
#self.Debug(f'{self.Portfolio[order_symbol].Quantity} remaining...')
self.Debug(f'Stop / Tgt4 hit (NOW flat -- cancel any pending)')
# Logging:
self.Debug(f'Running OCO -- {order_symbol} (Cancelling Alt Exits.)')
#COULD just CANCEL ALL orders here... ?
self.Transactions.CancelOpenOrders(order_symbol)
return