| Overall Statistics |
|
Total Orders 159 Average Win 2.97% Average Loss -1.99% Compounding Annual Return 323.065% Drawdown 16.300% Expectancy 0.262 Start Equity 100000 End Equity 146134.8 Net Profit 46.135% Sharpe Ratio 4.235 Sortino Ratio 6.689 Probabilistic Sharpe Ratio 81.049% Loss Rate 49% Win Rate 51% Profit-Loss Ratio 1.49 Alpha 1.678 Beta 0.884 Annual Standard Deviation 0.47 Annual Variance 0.221 Information Ratio 3.588 Tracking Error 0.457 Treynor Ratio 2.25 Total Fees $392.73 Estimated Strategy Capacity $600000000.00 Lowest Capacity Asset CL Z4A01OQ8Q37L Portfolio Turnover 123.79% Drawdown Recovery 61 |
# region imports
from AlgorithmImports import *
# endregion
# Import datetime
import datetime
# Import json
import json
class SwimmingGreenBull(QCAlgorithm):
def initialize(self):
# Set start date
self.set_start_date(2026, 3, 20)
# Set cash
self.set_cash(100000)
# Add future
future_object = self.add_future(ticker = Futures.Energy.CRUDE_OIL_WTI, resolution = Resolution.MINUTE, fill_forward = False, extended_market_hours = True)
# Set filter
future_object.set_filter(min_expiry_days = 0, max_expiry_days = 60)
# Register SPY
SPY = self.add_equity(ticker = "SPY")
# Time to store volume
self.time_to_store_volume = False
# Symbols list
self.symbols_list = []
# Total volume today
self.total_volume_today = {}
# Minute by minute volume
self.minute_by_minute_volume = {}
# Highest average minute-by-minute volume
self.highest_average_minute_by_minute_volume = None
# Last time alert logged
self.last_time_alert_logged = None
# Run alerts
self.run_alerts = False
# Minute by minute bars
self.minute_by_minute_bars = {}
# Minute by minute bars of most traded contract
self.highest_minute_by_minute_bars = None
# Webhook URL
self.discord_webhook_URL = "https://discord.com/api/webhooks/1495790766931312742/KlQz8y-few3ZMeg22BnPQTqhmIPflYhHEOQHQDLZdvk876xxUjJnQ7JgSohBj7nu6BON"
# Invested symbol store
self.invested_symbol = None
# Scheduled event for start of daily break
self.schedule.on(
self.date_rules.every_day(future_object.symbol),
self.time_rules.at(hour = 17, minute = 0, second = 0),
self.reset_total_volume
)
# Scheduled event for start of regular trading hours for SPY
self.schedule.on(
self.date_rules.every_day(SPY.symbol),
self.time_rules.after_market_open(symbol = SPY.symbol, minutes_after_open = 0),
self.regular_trading_hours_open_for_SPY
)
# Scheduled event for end of regular trading hours for SPY
self.schedule.on(
self.date_rules.every_day(SPY.symbol),
self.time_rules.before_market_close(symbol = SPY.symbol, minutes_before_close = 0),
self.regular_trading_hours_close_for_SPY
)
# Scheduled event for 5 PM on Fridays
self.schedule.on(
self.date_rules.week_end(symbol = future_object.symbol, days_offset = 0),
self.time_rules.at(hour = 17, minute = 0, second = 0),
self.reset_minute_by_minute_volume
)
# Scheduled event for 7 PM on Sundays
self.schedule.on(
self.date_rules.week_start(symbol = future_object.symbol, days_offset = 0),
self.time_rules.at(hour = 19, minute = 0, second = 0),
self.start_running_alerts
)
# Set warm up
self.set_warm_up(time_span = datetime.timedelta(days = 14))
def reset_total_volume(self):
# Loop total volume today dictionary
for symbol in self.total_volume_today:
# Reset
self.total_volume_today[symbol] = 0
def regular_trading_hours_open_for_SPY(self):
# Set time to store volume to True
self.time_to_store_volume = True
def regular_trading_hours_close_for_SPY(self):
# Set time to store volume to False
self.time_to_store_volume = False
def reset_minute_by_minute_volume(self):
# Average minute-by-minute volume of contracts
average_minute_by_minute_volume_of_contracts = []
# Loop
for symbol in self.minute_by_minute_volume:
# Calculate average minute-by-minute volume
average_minute_by_minute_volume = sum(self.minute_by_minute_volume[symbol]) / len(self.minute_by_minute_volume[symbol])
# Store in local list
average_minute_by_minute_volume_of_contracts.append(average_minute_by_minute_volume)
# Reset
self.minute_by_minute_volume[symbol] = []
# Get highest average minute-by-minute volume from local list
self.highest_average_minute_by_minute_volume = max(average_minute_by_minute_volume_of_contracts)
# Set run alerts variable to False
self.run_alerts = False
# Local dictionary to store minute by minute bars of contracts
local_minute_by_minute_bars = {}
# Local dictionary to store total volume of contracts
local_total_volume = {}
# Loop
for symbol in self.minute_by_minute_bars:
# Local variable to store total volume
local_total_volume_for_symbol = 0
# Loop bars of symbol
for trade_bar in self.minute_by_minute_bars[symbol]:
# Add volume to local total volume for symbol variable
local_total_volume_for_symbol += trade_bar.volume
# Add variable to local total volume dictionary
local_total_volume[symbol] = local_total_volume_for_symbol
# Add minute by minute bars list to local minute by minute bars dictionary
local_minute_by_minute_bars[symbol] = self.minute_by_minute_bars[symbol]
# Reset global minute by minute bars list of symbol
self.minute_by_minute_bars[symbol] = []
# Get symbol with highest total volume
symbol_with_highest_total_volume = max(local_total_volume, key=local_total_volume.get)
# Store its minute by minute bars
self.highest_minute_by_minute_bars = local_minute_by_minute_bars[symbol_with_highest_total_volume]
def start_running_alerts(self):
# Set run alerts variable to True
self.run_alerts = True
def on_securities_changed(self, changes):
# Loop added securities
for contract in changes.added_securities:
# Check if it's not a continuous futures contract
if "/" not in contract.symbol.value:
# Check if it's not SPY
if contract.symbol.value != "SPY":
# Store symbol
self.symbols_list.append(contract.symbol)
# Store in total volume today dictionary
self.total_volume_today[contract.symbol] = 0
# Store in minute by minute volume dictionary
self.minute_by_minute_volume[contract.symbol] = []
# Store in minute by minute bars dictionary
self.minute_by_minute_bars[contract.symbol] = []
# Loop removed securities
for contract in changes.removed_securities:
# Remove from symbols list
self.symbols_list.remove(contract.symbol)
# Remove from total volume today dictionary
self.total_volume_today.pop(contract.symbol)
# Remove from minute by minute volume dictionary
self.minute_by_minute_volume.pop(contract.symbol)
# Remove from minute by minute bars dictionary
self.minute_by_minute_bars.pop(contract.symbol)
# If invested symbol store is not None
if self.invested_symbol is not None:
# If removed contract is invested contract
if contract.symbol == self.invested_symbol:
# Reset
self.invested_symbol = None
def on_data(self, data: Slice):
# If invested symbol not None
if self.invested_symbol is not None:
# Check if long position active
if self.portfolio[self.invested_symbol].is_long == True:
# Run 10% take profit
if self.securities[self.invested_symbol].close > self.portfolio[self.invested_symbol].average_price * 1.1:
# Close position
self.market_order(symbol = self.invested_symbol, quantity = -self.portfolio[self.invested_symbol].quantity, tag = "LONG TAKE PROFIT")
# Reset
self.invested_symbol = None
# Check if short position active
elif self.portfolio[self.invested_symbol].is_short == True:
# Run 10% take profit
if self.securities[self.invested_symbol].close < self.portfolio[self.invested_symbol].average_price * 0.9:
# Close position
self.market_order(symbol = self.invested_symbol, quantity = -self.portfolio[self.invested_symbol].quantity, tag = "SHORT TAKE PROFIT")
# Reset
self.invested_symbol = None
# Loop symbols list
for symbol in self.symbols_list:
# If contract is traded
if data.bars.contains_key(symbol) == True:
# Get tradebar
trade_bar = data.bars[symbol]
# Volume
symbol_volume = trade_bar.volume
# Store volume
self.total_volume_today[symbol] += symbol_volume
# Check if regular trading hours
if self.time_to_store_volume == True:
# Store volume in minute by minute volume list
self.minute_by_minute_volume[symbol].append(symbol_volume)
# Store bar in minute by minute bar dictionary
self.minute_by_minute_bars[symbol].append(trade_bar)
# Check if not warming up
if self.is_warming_up == False:
# If run alerts is True
if self.run_alerts == True:
# Get symbol of contract with highest total volume
symbol_with_highest_total_volume = max(self.total_volume_today, key=self.total_volume_today.get)
# If current iteration symbol is symbol with highest total volume
if symbol == symbol_with_highest_total_volume:
# If it is not regular trading hours
if self.time_to_store_volume == False:
# If symbol volume is 3x greater than highest average minute by minute volume
if symbol_volume > self.highest_average_minute_by_minute_volume * 3:
# Local list
local_list = []
# Loop
for old_minute_bar in self.highest_minute_by_minute_bars:
# Check if hour and minute is the same as current time
if old_minute_bar.end_time.hour == self.time.hour and old_minute_bar.end_time.minute == self.time.minute:
# Add volume of minute bar in local list
local_list.append(old_minute_bar.volume)
# If local list not empty
if len(local_list) > 0:
# If current minute volume is greater than 2 times highest volume from local list
if symbol_volume > max(local_list) * 2:
# If last alert time is None
if self.last_time_alert_logged is None:
# Message
message = "Volume spike for :" + symbol.value + " volume amount: " + str(symbol_volume)
# Convert to dictionary
message_converted = {"content": message}
# Convert into json formatted string
json_formatted = json.dumps(message_converted)
# Send alert
#self.notify.web(address = self.discord_webhook_URL, data = json_formatted)
# Store current time
self.last_time_alert_logged = self.time
# Check if trade bar that triggered alert is green
if trade_bar.close > trade_bar.open:
# If not invested in anything
if self.invested_symbol is None:
# Market order
self.market_order(symbol = symbol, quantity = 1)
# Store symbol
self.invested_symbol = symbol
# If short position active
elif self.portfolio[self.invested_symbol].is_short == True:
# Close position
self.market_order(symbol = self.invested_symbol, quantity = -self.portfolio[self.invested_symbol].quantity)
# Market order
self.market_order(symbol = symbol, quantity = 1)
# Store symbol
self.invested_symbol = symbol
# Check if trade bar that triggered alert is red
else:
# If not invested in anything
if self.invested_symbol is None:
# Market order
self.market_order(symbol = symbol, quantity = -1)
# Store symbol
self.invested_symbol = symbol
# If long position active
elif self.portfolio[self.invested_symbol].is_long == True:
# Close position
self.market_order(symbol = self.invested_symbol, quantity = -self.portfolio[self.invested_symbol].quantity)
# Market order
self.market_order(symbol = symbol, quantity = -1)
# Store symbol
self.invested_symbol = symbol
# Else
else:
# If current time more than 2 hours from last alert time
if self.time > self.last_time_alert_logged + datetime.timedelta(hours = 2):
# Message
message = "Volume spike for :" + symbol.value + " volume amount: " + str(symbol_volume)
# Convert to dictionary
message_converted = {"content": message}
# Convert into json formatted string
json_formatted = json.dumps(message_converted)
# Send alert
#self.notify.web(address = self.discord_webhook_URL, data = json_formatted)
# Store current time
self.last_time_alert_logged = self.time
# Check if trade bar that triggered alert is green
if trade_bar.close > trade_bar.open:
# If not invested in anything
if self.invested_symbol is None:
# Market order
self.market_order(symbol = symbol, quantity = 1)
# Store symbol
self.invested_symbol = symbol
# If short position active
elif self.portfolio[self.invested_symbol].is_short == True:
# Close position
self.market_order(symbol = self.invested_symbol, quantity = -self.portfolio[self.invested_symbol].quantity)
# Market order
self.market_order(symbol = symbol, quantity = 1)
# Store symbol
self.invested_symbol = symbol
# Check if trade bar that triggered alert is red
else:
# If not invested in anything
if self.invested_symbol is None:
# Market order
self.market_order(symbol = symbol, quantity = -1)
# Store symbol
self.invested_symbol = symbol
# If long position active
elif self.portfolio[self.invested_symbol].is_long == True:
# Close position
self.market_order(symbol = self.invested_symbol, quantity = -self.portfolio[self.invested_symbol].quantity)
# Market order
self.market_order(symbol = symbol, quantity = -1)
# Store symbol
self.invested_symbol = symbol
# Else
else:
# If last alert time is None
if self.last_time_alert_logged is None:
# Message
message = "Volume spike for :" + symbol.value + " volume amount: " + str(symbol_volume)
# Convert to dictionary
message_converted = {"content": message}
# Convert into json formatted string
json_formatted = json.dumps(message_converted)
# Send alert
#self.notify.web(address = self.discord_webhook_URL, data = json_formatted)
# Store current time
self.last_time_alert_logged = self.time
# Check if trade bar that triggered alert is green
if trade_bar.close > trade_bar.open:
# If not invested in anything
if self.invested_symbol is None:
# Market order
self.market_order(symbol = symbol, quantity = 1)
# Store symbol
self.invested_symbol = symbol
# If short position active
elif self.portfolio[self.invested_symbol].is_short == True:
# Close position
self.market_order(symbol = self.invested_symbol, quantity = -self.portfolio[self.invested_symbol].quantity)
# Market order
self.market_order(symbol = symbol, quantity = 1)
# Store symbol
self.invested_symbol = symbol
# Check if trade bar that triggered alert is red
else:
# If not invested in anything
if self.invested_symbol is None:
# Market order
self.market_order(symbol = symbol, quantity = -1)
# Store symbol
self.invested_symbol = symbol
# If long position active
elif self.portfolio[self.invested_symbol].is_long == True:
# Close position
self.market_order(symbol = self.invested_symbol, quantity = -self.portfolio[self.invested_symbol].quantity)
# Market order
self.market_order(symbol = symbol, quantity = -1)
# Store symbol
self.invested_symbol = symbol
# Else
else:
# If current time more than 2 hours from last alert time
if self.time > self.last_time_alert_logged + datetime.timedelta(hours = 2):
# Message
message = "Volume spike for :" + symbol.value + " volume amount: " + str(symbol_volume)
# Convert to dictionary
message_converted = {"content": message}
# Convert into json formatted string
json_formatted = json.dumps(message_converted)
# Send alert
#self.notify.web(address = self.discord_webhook_URL, data = json_formatted)
# Store current time
self.last_time_alert_logged = self.time
# Check if trade bar that triggered alert is green
if trade_bar.close > trade_bar.open:
# If not invested in anything
if self.invested_symbol is None:
# Market order
self.market_order(symbol = symbol, quantity = 1)
# Store symbol
self.invested_symbol = symbol
# If short position active
elif self.portfolio[self.invested_symbol].is_short == True:
# Close position
self.market_order(symbol = self.invested_symbol, quantity = -self.portfolio[self.invested_symbol].quantity)
# Market order
self.market_order(symbol = symbol, quantity = 1)
# Store symbol
self.invested_symbol = symbol
# Check if trade bar that triggered alert is red
else:
# If not invested in anything
if self.invested_symbol is None:
# Market order
self.market_order(symbol = symbol, quantity = -1)
# Store symbol
self.invested_symbol = symbol
# If long position active
elif self.portfolio[self.invested_symbol].is_long == True:
# Close position
self.market_order(symbol = self.invested_symbol, quantity = -self.portfolio[self.invested_symbol].quantity)
# Market order
self.market_order(symbol = symbol, quantity = -1)
# Store symbol
self.invested_symbol = symbol
# If regular trading hours
else:
# If symbol volume is 6x greater than highest average minute by minute volume
if symbol_volume > self.highest_average_minute_by_minute_volume * 6:
# Local list
local_list = []
# Loop
for old_minute_bar in self.highest_minute_by_minute_bars:
# Check if hour and minute is the same as current time
if old_minute_bar.end_time.hour == self.time.hour and old_minute_bar.end_time.minute == self.time.minute:
# Add volume of minute bar in local list
local_list.append(old_minute_bar.volume)
# If local list not empty
if len(local_list) > 0:
# If current minute volume is greater than 2 times highest volume from local list
if symbol_volume > max(local_list) * 2:
# If last alert time is None
if self.last_time_alert_logged is None:
# Message
message = "Volume spike for :" + symbol.value + " volume amount: " + str(symbol_volume)
# Convert to dictionary
message_converted = {"content": message}
# Convert into json formatted string
json_formatted = json.dumps(message_converted)
# Send alert
#self.notify.web(address = self.discord_webhook_URL, data = json_formatted)
# Store current time
self.last_time_alert_logged = self.time
# Check if trade bar that triggered alert is green
if trade_bar.close > trade_bar.open:
# If not invested in anything
if self.invested_symbol is None:
# Market order
self.market_order(symbol = symbol, quantity = 1)
# Store symbol
self.invested_symbol = symbol
# If short position active
elif self.portfolio[self.invested_symbol].is_short == True:
# Close position
self.market_order(symbol = self.invested_symbol, quantity = -self.portfolio[self.invested_symbol].quantity)
# Market order
self.market_order(symbol = symbol, quantity = 1)
# Store symbol
self.invested_symbol = symbol
# Check if trade bar that triggered alert is red
else:
# If not invested in anything
if self.invested_symbol is None:
# Market order
self.market_order(symbol = symbol, quantity = -1)
# Store symbol
self.invested_symbol = symbol
# If long position active
elif self.portfolio[self.invested_symbol].is_long == True:
# Close position
self.market_order(symbol = self.invested_symbol, quantity = -self.portfolio[self.invested_symbol].quantity)
# Market order
self.market_order(symbol = symbol, quantity = -1)
# Store symbol
self.invested_symbol = symbol
# Else
else:
# If current time more than 2 hours from last alert time
if self.time > self.last_time_alert_logged + datetime.timedelta(hours = 2):
# Message
message = "Volume spike for :" + symbol.value + " volume amount: " + str(symbol_volume)
# Convert to dictionary
message_converted = {"content": message}
# Convert into json formatted string
json_formatted = json.dumps(message_converted)
# Send alert
#self.notify.web(address = self.discord_webhook_URL, data = json_formatted)
# Store current time
self.last_time_alert_logged = self.time
# Check if trade bar that triggered alert is green
if trade_bar.close > trade_bar.open:
# If not invested in anything
if self.invested_symbol is None:
# Market order
self.market_order(symbol = symbol, quantity = 1)
# Store symbol
self.invested_symbol = symbol
# If short position active
elif self.portfolio[self.invested_symbol].is_short == True:
# Close position
self.market_order(symbol = self.invested_symbol, quantity = -self.portfolio[self.invested_symbol].quantity)
# Market order
self.market_order(symbol = symbol, quantity = 1)
# Store symbol
self.invested_symbol = symbol
# Check if trade bar that triggered alert is red
else:
# If not invested in anything
if self.invested_symbol is None:
# Market order
self.market_order(symbol = symbol, quantity = -1)
# Store symbol
self.invested_symbol = symbol
# If long position active
elif self.portfolio[self.invested_symbol].is_long == True:
# Close position
self.market_order(symbol = self.invested_symbol, quantity = -self.portfolio[self.invested_symbol].quantity)
# Market order
self.market_order(symbol = symbol, quantity = -1)
# Store symbol
self.invested_symbol = symbol
# Else
else:
# If last alert time is None
if self.last_time_alert_logged is None:
# Message
message = "Volume spike for :" + symbol.value + " volume amount: " + str(symbol_volume)
# Convert to dictionary
message_converted = {"content": message}
# Convert into json formatted string
json_formatted = json.dumps(message_converted)
# Send alert
#self.notify.web(address = self.discord_webhook_URL, data = json_formatted)
# Store current time
self.last_time_alert_logged = self.time
# Check if trade bar that triggered alert is green
if trade_bar.close > trade_bar.open:
# If not invested in anything
if self.invested_symbol is None:
# Market order
self.market_order(symbol = symbol, quantity = 1)
# Store symbol
self.invested_symbol = symbol
# If short position active
elif self.portfolio[self.invested_symbol].is_short == True:
# Close position
self.market_order(symbol = self.invested_symbol, quantity = -self.portfolio[self.invested_symbol].quantity)
# Market order
self.market_order(symbol = symbol, quantity = 1)
# Store symbol
self.invested_symbol = symbol
# Check if trade bar that triggered alert is red
else:
# If not invested in anything
if self.invested_symbol is None:
# Market order
self.market_order(symbol = symbol, quantity = -1)
# Store symbol
self.invested_symbol = symbol
# If long position active
elif self.portfolio[self.invested_symbol].is_long == True:
# Close position
self.market_order(symbol = self.invested_symbol, quantity = -self.portfolio[self.invested_symbol].quantity)
# Market order
self.market_order(symbol = symbol, quantity = -1)
# Store symbol
self.invested_symbol = symbol
# Else
else:
# If current time more than 2 hours from last alert time
if self.time > self.last_time_alert_logged + datetime.timedelta(hours = 2):
# Message
message = "Volume spike for :" + symbol.value + " volume amount: " + str(symbol_volume)
# Convert to dictionary
message_converted = {"content": message}
# Convert into json formatted string
json_formatted = json.dumps(message_converted)
# Send alert
#self.notify.web(address = self.discord_webhook_URL, data = json_formatted)
# Store current time
self.last_time_alert_logged = self.time
# Check if trade bar that triggered alert is green
if trade_bar.close > trade_bar.open:
# If not invested in anything
if self.invested_symbol is None:
# Market order
self.market_order(symbol = symbol, quantity = 1)
# Store symbol
self.invested_symbol = symbol
# If short position active
elif self.portfolio[self.invested_symbol].is_short == True:
# Close position
self.market_order(symbol = self.invested_symbol, quantity = -self.portfolio[self.invested_symbol].quantity)
# Market order
self.market_order(symbol = symbol, quantity = 1)
# Store symbol
self.invested_symbol = symbol
# Check if trade bar that triggered alert is red
else:
# If not invested in anything
if self.invested_symbol is None:
# Market order
self.market_order(symbol = symbol, quantity = -1)
# Store symbol
self.invested_symbol = symbol
# If long position active
elif self.portfolio[self.invested_symbol].is_long == True:
# Close position
self.market_order(symbol = self.invested_symbol, quantity = -self.portfolio[self.invested_symbol].quantity)
# Market order
self.market_order(symbol = symbol, quantity = -1)
# Store symbol
self.invested_symbol = symbol