| 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 4.457 Tracking Error 0.124 Treynor Ratio 0 Total Fees $0.00 Estimated Strategy Capacity $0 Lowest Capacity Asset |
from AlgorithmImports import *
from metrics.simple_price_metrics.Price import *
from metrics.simple_price_metrics.DMA import *
from metrics.simple_price_metrics.OnePercentValue import *
from metrics.simple_price_metrics.DayHigh import *
from metrics.simple_price_metrics.DayLow import *
from metrics.price_change_metrics.PercentageUpDownFromOpen import *
from metrics.price_change_metrics.Last52WeekHighPercentage import *
from metrics.price_change_metrics.MarketOpenChangeDMAPercentage import *
from metrics.price_change_metrics.DayChangePercentage import *
from metrics.price_change_metrics.DMAGapUpDown import *
from metrics.volume_metrics.VWAP import *
from metrics.volume_metrics.PercentageOfMarketPriceVsVWAP import *
from metrics.volume_metrics.VolumeDMA import *
from metrics.volume_metrics.PercentageAboveDailyVolume import *
from metrics.volume_metrics.VolumePerTimeMultipleVs30DMA import *
from metrics.volume_metrics.VolumeChangePercentage import *
from metrics.volume_metrics.NetVolumePercentage import *
from metrics.volume_metrics.FiftenSeconds30DMAVolume import *
from metrics.day_range_metrics.DayFactorPrice import *
from metrics.day_range_metrics.DayPriceRangePercentage import *
from metrics.day_range_metrics.MarketOpenRangeDMAPercentage import *
from metrics.day_range_metrics.PricePercentageOf30DayRange import *
from metrics.day_range_metrics.PricePercentageOf52WeekRange import *
from metrics.day_range_metrics.PricePercentageOfDayRange import *
from metrics.day_range_metrics.RangePercentageDMA import *
from metrics.day_range_metrics.RangeMultiple import *
from metrics.day_range_metrics.RangeMultiple7DMA import *
from metrics.day_range_metrics.DayPercentageChangeOfDMARangePercentage import *
from trades.tradecollections.TradeOne import TradeOne
from trades.tradecollections.TradeThree import TradeThree
from trades.tradecollections.TradeTwo import TradeTwo
class Config:
def __init__(self):
self.set_symbols()
self.set_plot_metrics()
self.set_metrics()
self.set_trades()
def set_symbols(self):
self.symbols = ['AAP', 'AAPL', 'ABBV', 'ABCL', 'ABNB', 'ABSI', 'ACAH', 'ACM', 'ACN', 'ADBE', 'ADI', 'ADMP', 'ADSK', 'AER', 'AFRM', 'AGCB', 'AGI', 'AI', 'ALCC', 'ALGN', 'ALIT', 'ALNY', 'AMAT', 'AMD', 'AMZN', 'ANAC', 'AON', 'APA', 'APG', 'APO', 'APSG', 'AR', 'ARKK', 'ARMK', 'ARNC', 'ASML', 'ASPU', 'ATIP', 'ATUS', 'ATVI', 'AUR', 'AUS', 'AVPT', 'AXP', 'AXTA', 'BABA', 'BAC', 'BBDC', 'BBWI', 'BEC']
def set_metrics(self):
# please note that the order matters so if metrics x depends on
# metrics y you need to add metrics x before metrics y
self.metrics={
"real_time" : [ ("Price",Price),("VWAP",VWAP),("OnePercentValue",OnePercentValue),("PercentageOfMarketPriceVsVWAP",PercentageOfMarketPriceVsVWAP),
("DayHigh", DayHigh), ("DayLow", DayLow)],
"weekly" : [ ("PricePercentageOf52WeekRange",PricePercentageOf52WeekRange)],
"daily" : [ ("1DayChangePercentage",DayChangePercentage,[1]),("7DayChangePercentage",DayChangePercentage,[7]),("30DayChangePercentage",DayChangePercentage,[30]),
("7DMAGapUpDown",DMAGapUpDown,[7]),("30DMAGapUpDown",DMAGapUpDown,[30]),
("MarketOpenChange7DMAPercentage",MarketOpenChangeDMAPercentage,[7]),("MarketOpenChange30DMAPercentage",MarketOpenChangeDMAPercentage,[30]),
("PercentageUpDownFromOpen",PercentageUpDownFromOpen),
("DayNetVolumePercentage",NetVolumePercentage,[1]),("WeekNetVolumePercentage",NetVolumePercentage,[7]),("MonthNetVolumePercentage",NetVolumePercentage,[30]),
("Volume7DMA",VolumeDMA,[7]),("Volume30DMA",VolumeDMA,[30]),
("DayLowerQuadrantPrice",DayFactorPrice,[0.25]),("DayMiddlePrice",DayFactorPrice,[0.5]),("DayUpperQuadrantPrice",DayFactorPrice,[0.75]),
("DayPriceRangePercentage",DayPriceRangePercentage),
("PricePercentageOfDayRange",PricePercentageOfDayRange),
("RangePercentage7DMA",RangePercentageDMA,[7]),("RangePercentage30DMA",RangePercentageDMA,[30]),
("HistoricalRangePercentage7DMA",RangePercentageDMA,[7,7]),("HistoricalRangePercentage30DMA",RangePercentageDMA,[30,30]),
("DayPercentageChangeOf7DMARangePercentage",DayPercentageChangeOfDMARangePercentage,[7]),("DayPercentageChangeOf30DMARangePercentage",DayPercentageChangeOfDMARangePercentage,[30]),
("PricePercentageOf30DayRange",PricePercentageOf30DayRange),
("RangeMultiple",RangeMultiple),("RangeMultiple7DMA",RangeMultiple7DMA),
("DMA50",DMA,[50]),("DMA100",DMA,[100])
],
"minute" : [ ("DayVolumeChangePercentage",VolumeChangePercentage,[1]),("WeekVolumeChangePercentage",VolumeChangePercentage,[7]),("MonthVolumeChangePercentage",VolumeChangePercentage,[30]),
("VolumePerTimeMultipleVs30DMA",VolumePerTimeMultipleVs30DMA),
("PercentageAboveDailyVolume",PercentageAboveDailyVolume)
],
"15_seconds" : [("Last52WeekHighPercentage",Last52WeekHighPercentage),("FiftenSeconds30DMAVolume",FiftenSeconds30DMAVolume)],
"15_minute_after_market_open":[("MarketOpenRange7DMAPercentage",MarketOpenRangeDMAPercentage,[7]),("MarketOpenRange30DMAPercentage",MarketOpenRangeDMAPercentage,[30])]
}
def set_trades(self):
self.trades = [TradeOne,TradeTwo]
@classmethod
def set_parameters(cls):
cls.parameters={
"PauseAlgo" : True,
"C1" : True,
"C2" : True,
"C3" : True,
"C4" : True,
"C5" : True,
"Trading Algorithm" : True,
"SameConditionTimeC1" : 0,
"SameConditionTimeC2" : 1,
"SameConditionTimeC3" : 30,
"SameConditionTimeC4" : 10,
"SameConditionTimeC5" : 10,
"Parameter 1" : 135,
"Parameter 2" : 145,
"Parameter 3" : 300,
"Parameter 4" : 7,
"Parameter 5" : 0.85,
"VWAP_PCT" : 40,
"Rally_X_Min_PCT" : 8,
"Rally_X_Max_PCT" : 20,
"Rally_Y_PCT" : 4,
"OffsetPCT" : 0.1,
"SharesToSell" : 100,
"StopLoss" : 200,
"ProfitTakeC1" : 15,
"ProfitTakeC2" : 15,
"ProfitTakeC3" : 10,
"ProfitTakeC4" : 15,
"ProfitTakeC5" : 15,
}
return cls.parameters
def set_plot_metrics(self):
self.plot_metrics=["VWAP"]from AlgorithmImports import *
from symbols.symbol import *
from datetime import timedelta
from config import Config
class App(QCAlgorithm):
symbols = {}
def Initialize(self):
self.SetStartDate(2023, 2, 8) # Set Start Date
self.SetCash(100000) # Set Strategy Cash
self.config: Config = Config()
# map for the disired history data of the application
self.WeekHistory = {}
self.DayHistory = {}
self.MinuteHistory = {}
self.SecondHistory = {}
# create symbols
for name in self.config.symbols:
sp = Symbol(self, name)
self.symbols[name] = sp
self.AddEquity(name, Resolution.Second)
self.symbols[name].addMetric()
self.symbols[name].addTrades()
# please add the desired rolling window with the max bars you need
self.WeekHistory[name] = RollingWindow[TradeBar](52)
self.Consolidate(name, CalendarType.Weekly, self.WeekHistoryHandler)
self.DayHistory[name] = RollingWindow[TradeBar](60)
self.Consolidate(name, Resolution.Daily, self.DayHistoryHandler)
self.MinuteHistory[name] = RollingWindow[TradeBar](24000) # 60 days
self.Consolidate(name, Resolution.Minute, self.MinuteHistoryHandler)
self.SecondHistory[name] = RollingWindow[TradeBar](15)
# this schedule assumes every symbol trade on the same exchange
self.Schedule.On(self.DateRules.EveryDay(self.config.symbols[0]),
self.TimeRules.AfterMarketOpen(self.config.symbols[0], 16),
self.MarketOpenUpdate)
self.Schedule.On(self.DateRules.EveryDay(self.config.symbols[0]),
self.TimeRules.Every(timedelta(seconds=15)),
self.FiftenSecondUpdate)
self.SetWarmUp(TimeSpan.FromDays(380), Resolution.Minute)
def OnData(self, data: Slice):
if self.IsWarmingUp:
return
for symbol, bar in data.Bars.items():
if symbol.Value not in self.SecondHistory:
continue
self.SecondHistory[symbol.Value].Add(bar)
if self.SecondHistory[symbol.Value].IsReady:
self.symbols[symbol.Value].update("real_time")
self.symbols[symbol.Value].update_trade()
self.symbols[symbol.Value].trade()
def OnOrderEvent(self, orderEvent: OrderEvent) -> None:
self.symbols[orderEvent.Symbol.Value].handle_order_event(orderEvent)
def WeekHistoryHandler(self, bar):
if bar is None or bar.Symbol is None or bar.Symbol.Value not in self.WeekHistory:
return
self.WeekHistory[bar.Symbol.Value].Add(bar)
if not self.WeekHistory[bar.Symbol.Value].IsReady:
return
self.symbols[bar.Symbol.Value].update("weekly")
def DayHistoryHandler(self, bar):
if bar is None or bar.Symbol is None or bar.Symbol.Value not in self.DayHistory:
return
self.DayHistory[bar.Symbol.Value].Add(bar)
if not self.DayHistory[bar.Symbol.Value].IsReady:
return
self.symbols[bar.Symbol.Value].update("daily")
def MinuteHistoryHandler(self, bar):
if bar is None or bar.Symbol is None or bar.Symbol.Value not in self.MinuteHistory:
return
self.MinuteHistory[bar.Symbol.Value].Add(bar)
if self.IsWarmingUp:
return
self.symbols[bar.Symbol.Value].update("minute")
def FiftenSecondUpdate(self):
if self.IsWarmingUp:
return
for symbol in self.symbols.keys():
if self.SecondHistory[symbol].IsReady:
self.symbols[symbol].update("15_seconds")
def MarketOpenUpdate(self):
for symbol in self.symbols.keys():
if self.MinuteHistory[symbol].IsReady:
self.symbols[symbol].update("15_minute_after_market_open")
# region imports
from AlgorithmImports import *
# endregion
from abc import ABC, abstractmethod
class Metric(ABC):
@abstractmethod
def calculate(self):
pass
from AlgorithmImports import *
from metrics.Metric import Metric
class DayFactorPrice(Metric):
"""
18) Day’s Lower ¼ Price
The price of the lower quadrant of the day range.
19) Day’s Middle Price
The MidDay Range Price
20) Day’s Upper ¼ Price
The upper quadrant price of the day range.
"""
def __init__(self, algo, name, factor):
self.name = name
self.algorithm = algo
self.factor = factor
self.IsReady = False
def calculate(self):
high = self.algorithm.DayHistory[self.name][0].High
low = self.algorithm.DayHistory[self.name][0].Low
self.Value = self.factor * (high - low) + low
self.IsReady = True
from metrics.Metric import Metric
from AlgorithmImports import *
from .RangePercentageDMA import *
class DayPercentageChangeOfDMARangePercentage(Metric):
"""
16) 7 Day Percentage Change of 7 DMA Range Percentage
The percentage change of the last day’s 7 DMA Range Vs the 7th historical day 7 DMA Range
17) 30 Day Percentage Change of 30 DMA Range Percentage
The percentage change of the last day’s 30 DMA Range Vs the 30th historical day 30 DMA Range
"""
def __init__(self, algo, name, length):
self.name = name
self.algorithm = algo
self.length = length
self.IsReady = False
def calculate(self):
if not self.algorithm.symbols[self.name].metrics["RangePercentage"+str(self.length)+"DMA"].IsReady:
return
if not self.algorithm.symbols[self.name].metrics["HistoricalRangePercentage"+str(self.length)+"DMA"].IsReady:
return
dma = self.algorithm.symbols[self.name].metrics["RangePercentage"+str(self.length)+"DMA"].Value
historical_dma = self.algorithm.symbols[self.name].metrics["HistoricalRangePercentage"+str(self.length)+"DMA"].Value
self.Value = (dma - historical_dma) * 100 / dma
self.IsReady = True
from AlgorithmImports import *
from metrics.Metric import Metric
class DayPriceRangePercentage(Metric):
"""
3) Day’s Price Range Percentage
(Day High - Day Low) * 100 / Day Open
"""
def __init__(self, algo, name):
self.name = name
self.algorithm = algo
self.IsReady = False
def calculate(self):
high = self.algorithm.DayHistory[self.name][0].High
low = self.algorithm.DayHistory[self.name][0].Low
open = self.algorithm.DayHistory[self.name][0].Open
self.Value = (high - low) * 100 / open
self.IsReady = True
from AlgorithmImports import *
from metrics.Metric import Metric
from statistics import mean
from datetime import time
class MarketOpenRangeDMAPercentage(Metric):
"""
21) Market Open Range 7 DMA Percentage
The 7 DMA of Daily Range.
22) Market Open Range 30 DMA Percentage
The 30 DMA of Daily Range.
"""
def __init__(self, algo, name, length):
self.name = name
self.algorithm = algo
self.length = length
self.IsReady = False
def calculate(self):
day_range = []
for i in range(390 * self.length):
if self.algorithm.MinuteHistory[self.name][i].EndTime.time() == time(hour=9, minute=45):
self.high = self.algorithm.MinuteHistory[self.name][i].High
self.low = self.algorithm.MinuteHistory[self.name][i].Low
if self.algorithm.MinuteHistory[self.name][i].EndTime.time() < time(hour=9, minute=45):
self.high = max(self.high, self.algorithm.MinuteHistory[self.name][i].High)
self.low = min(self.low, self.algorithm.MinuteHistory[self.name][i].Low)
if self.algorithm.MinuteHistory[self.name][i].EndTime.time() == time(hour=9, minute=31):
open = self.algorithm.MinuteHistory[self.name][i].Open
day_range.append((self.high - self.low) * 100 / open)
self.Value = mean(day_range)
self.IsReady = True
from metrics.Metric import Metric
from AlgorithmImports import *
class PricePercentageOf30DayRange(Metric):
"""
12) Price Percentage of 30 Day Range
The last price percentage of the the last 30 Days Price Range
"""
def __init__(self, algo, name):
self.name = name
self.algorithm = algo
self.IsReady = False
def calculate(self):
high = max([bar.High for bar in list(self.algorithm.DayHistory[self.name])[:30]])
low = min([bar.Low for bar in list(self.algorithm.DayHistory[self.name])[:30]])
close = self.algorithm.DayHistory[self.name][0].Close
self.Value = (close - low) * 100 / (high - low)
self.IsReady = True
from AlgorithmImports import *
from metrics.Metric import Metric
class PricePercentageOf52WeekRange(Metric):
"""
13) Price Percentage of 52 Week Range
The last price percentage of the the last 52 Weeks Price Range
"""
def __init__(self, algo, name):
self.name = name
self.algorithm = algo
self.IsReady = False
def calculate(self):
high = max([bar.High for bar in list(self.algorithm.WeekHistory[self.name])[:52]])
low = min([bar.Low for bar in list(self.algorithm.WeekHistory[self.name])[:52]])
close = self.algorithm.WeekHistory[self.name][0].Close
self.Value = (close - low) * 100 / (high - low)
self.IsReady = True
from AlgorithmImports import *
from metrics.Metric import Metric
class PricePercentageOfDayRange(Metric):
"""
2) Price Percentage of Day Range
The last price percentage of the day range.
(Day Close - Day Low) * 100 / (Day High - Day Low)
"""
def __init__(self, algo, name):
self.name = name
self.algorithm = algo
self.IsReady = False
def calculate(self):
high = self.algorithm.DayHistory[self.name][0].High
low = self.algorithm.DayHistory[self.name][0].Low
close = self.algorithm.DayHistory[self.name][0].Close
if high - low == 0:
return
self.Value = (close - low) * 100 / (high - low)
self.IsReady = True
from AlgorithmImports import *
from metrics.Metric import Metric
class RangeMultiple(Metric):
"""
6) Range Multiple
Day’s Price Range Percentage / 30 DMA Range Percentage
"""
def __init__(self, algo, name):
self.name = name
self.algorithm = algo
self.IsReady = False
def calculate(self):
if not self.algorithm.symbols[self.name].metrics['DayPriceRangePercentage'].IsReady:
return
if not self.algorithm.symbols[self.name].metrics['RangePercentage30DMA'].IsReady:
return
self.Value = self.algorithm.symbols[self.name].metrics['DayPriceRangePercentage'].Value /\
self.algorithm.symbols[self.name].metrics['RangePercentage30DMA'].Value
self.IsReady = True
from AlgorithmImports import *
from metrics.Metric import Metric
from collections import deque
from statistics import mean
class RangeMultiple7DMA(Metric):
"""
7) 7 DMA Range Multiple
The 7 Day Simple Moving Average of the Range Multiple
"""
def __init__(self, algo, name):
self.name = name
self.algorithm = algo
self.Values = RollingWindow[float](7)
self.IsReady = False
def calculate(self):
if not self.algorithm.symbols[self.name].metrics['RangeMultiple'].IsReady:
return
self.Values.Add(self.algorithm.symbols[self.name].metrics['RangeMultiple'].Value)
if not self.Values.IsReady:
return
self.Value = mean(list(self.Values))
self.IsReady = True
from AlgorithmImports import *
from metrics.Metric import Metric
from statistics import mean
class RangePercentageDMA(Metric):
"""
4) 7 DMA Range Percentage
The 7 Days Simple Moving Average of the Day’s Price Range Percentage
5) 30 DMA Range Percentage
The 30 Days Simple Moving Average of the Day’s Price Range Percentage
"""
def __init__(self, algo, name, length, ignore=0):
self.name = name
self.algorithm = algo
self.length = length
self.ignore = ignore
self.Values = RollingWindow[float](60)
self.IsReady = False
def calculate(self):
if not self.algorithm.symbols[self.name].metrics['DayPriceRangePercentage'].IsReady:
return
self.Values.Add(self.algorithm.symbols[self.name].metrics['DayPriceRangePercentage'].Value)
if not self.Values.IsReady:
return
self.Value = mean(list(self.Values)[self.ignore:self.ignore+self.length])
self.IsReady = True
from AlgorithmImports import *
from metrics.Metric import Metric
from statistics import mean
class DMAGapUpDown(Metric):
"""
40) 7 DMA Gap Up/Down
The 7 DMA of the Gap Percentage.
The Gap Percentage is the percentage of change of the day open relative to the last day’s close.
41) 30 DMA Gap Up/Down
The 30 DMA of the Gap Percentage.
The Gap Percentage is the percentage of change of the day open relative to the last day’s close.
"""
def __init__(self, algo, name, length):
self.name = name
self.algorithm = algo
self.length = length
self.IsReady = False
def calculate(self):
gaps = []
for i in range(self.length):
last_close = self.algorithm.DayHistory[self.name][i+1].Close
open_value = self.algorithm.DayHistory[self.name][i].Open
Value = (open_value - last_close) * 100 / last_close
gaps.append(Value)
self.Value = mean(gaps)
self.IsReady = True
from AlgorithmImports import *
from metrics.Metric import Metric
class DayChangePercentage(Metric):
"""
37) 1 Day Change Percentage
The change percentage of the day close relative to the close of day 1 daily bars ago.
38) 7 Day Change Percentage
The change percentage of the day close relative to the close of day 7 daily bars ago.
39) 30 Day Change Percentage
The change percentage of the day close relative to the close of the day 30 daily bars ago
"""
def __init__(self, algo, name, length):
self.name = name
self.algorithm = algo
self.length = length
self.IsReady = False
def calculate(self):
last_close = self.algorithm.DayHistory[self.name][self.length].Close
close_value = self.algorithm.DayHistory[self.name][0].Close
self.Value = (close_value - last_close) * 100 / last_close
self.IsReady = True
from AlgorithmImports import *
from metrics.Metric import Metric
class Last52WeekHighPercentage(Metric):
"""
14) 52 Week High Percentage
The percentage of the last 52 Weeks high relative to the last price.
(52 Week High - last price) * 100 / last price
"""
def __init__(self, algo, name):
self.name = name
self.algorithm = algo
self.IsReady = False
def calculate(self):
high_value = max([x.High for x in self.algorithm.WeekHistory[self.name]])
last_value = self.algorithm.Securities[self.name].Price
self.Value = (high_value - last_value) * 100 / last_value
self.IsReady = True
from AlgorithmImports import *
from metrics.Metric import Metric
from statistics import mean
class MarketOpenChangeDMAPercentage(Metric):
"""
23) Market Open Change 7 DMA Percentage
The 7 DMA of the Market Open Change.
The Market Open change is the percentage change between day open and day close relative to the day open.
24) Market Open Change 30 DMA Percentage
The 30 DMA of the Market Open Change.
The Market Open change is the percentage change between day open and day close relative to the day open.
"""
def __init__(self, algo, name, length):
self.name = name
self.algorithm = algo
self.length = length
self.IsReady = False
def calculate(self):
change = []
for bar in list(self.algorithm.DayHistory[self.name])[:self.length]:
open_value = bar.Open
close_value = bar.Close
Value = (close_value - open_value) * 100 / open_value
change.append(Value)
self.Value = mean(change)
self.IsReady = True
from AlgorithmImports import *
from metrics.Metric import Metric
class PercentageUpDownFromOpen(Metric):
"""
10) Percentage UP/Down From Open
The percentage change of price relative to the day open price.
(Day Close - Day Open) * 100 / Day Open
"""
def __init__(self, algo, name):
self.name = name
self.algorithm = algo
self.IsReady = False
def calculate(self):
open_value = self.algorithm.DayHistory[self.name][0].Open
close_value = self.algorithm.DayHistory[self.name][0].Close
self.Value = (close_value - open_value) * 100 / open_value
self.IsReady = True
from AlgorithmImports import *
from metrics.Metric import Metric
class DMA(Metric):
'''
The Day Simple Moving Average of Price
'''
def __init__(self, algo, name, length):
self.name = name
self.algorithm = algo
self.sma = self.algorithm.SMA(name, length, Resolution.Daily)
self.IsReady=False
def calculate(self):
if self.sma.IsReady:
self.Value=self.sma.Current.Value
self.IsReady=True
from datetime import time
from AlgorithmImports import *
from metrics.Metric import Metric
class DayHigh(Metric):
"""
Calculate The Day High of the symbol.
"""
def __init__(self, algo, name):
self.name = name
self.algorithm = algo
self.Value = None
self.IsReady = False
def calculate(self):
if (
self.Value is None
or self.algorithm.SecondHistory[self.name][0].Time.date()
!= self.algorithm.SecondHistory[self.name][1].Time.date()
):
self.Value = self.algorithm.SecondHistory[self.name][0].High
self.Value = max(self.Value, self.algorithm.SecondHistory[self.name][0].High)
self.IsReady = True
from datetime import time
from AlgorithmImports import *
from metrics.Metric import Metric
class DayLow(Metric):
"""
Calculate The Day Low of the symbol.
"""
def __init__(self, algo, name):
self.name = name
self.algorithm = algo
self.Value = None
self.IsReady = False
def calculate(self):
if (
self.Value is None
or self.algorithm.SecondHistory[self.name][0].Time.date()
!= self.algorithm.SecondHistory[self.name][1].Time.date()
):
self.Value = self.algorithm.SecondHistory[self.name][0].Low
self.Value = min(self.Value, self.algorithm.SecondHistory[self.name][0].Low)
self.IsReady = True
from AlgorithmImports import *
from metrics.Metric import Metric
class OnePercentValue(Metric):
"""
Calculate One Percent Value of last symbol price.
"""
def __init__(self, algo, name):
self.name = name
self.algorithm = algo
self.IsReady=False
def calculate(self):
self.Value = self.algorithm.Securities[self.name].Price * 0.1
self.IsReady=True
from AlgorithmImports import *
from metrics.Metric import Metric
class Price(Metric):
"""
Calculate The last update of the symbol price.
"""
def __init__(self, algo, name):
self.name = name
self.algorithm = algo
self.IsReady=False
def calculate(self):
self.Value = self.algorithm.Securities[self.name].Price
self.IsReady=True
from metrics.Metric import Metric
from AlgorithmImports import *
class FiftenSeconds30DMAVolume(Metric):
"""
28) 15 Seconds / 30 DMA Volume
The ratio between the last 15 seconds volume and the 30 DMA volume scaled per 15 seconds.
"""
def __init__(self, algo, name):
self.name = name
self.algorithm = algo
self.IsReady = False
def calculate(self):
total_vol = sum([bar.Volume for bar in list(self.algorithm.DayHistory[self.name])[:30]])
dma_15_second = total_vol / 46800
vol_15_seconds = sum(bar.Volume for bar in self.algorithm.SecondHistory[self.name])
if dma_15_second == 0:
return
self.Value = vol_15_seconds / dma_15_second
self.IsReady = True
from metrics.Metric import Metric
from AlgorithmImports import *
class NetVolumePercentage(Metric):
"""
29) Day Net Volume Percentage
The Day Net Volume Percentage is the Percentage of the positive volume relative to the total volume of the day.
The positive volume is the volume corresponding to the price bar where the close is equal or greater than the close price of the previous bar.
We can use 1 minute bars in this calculation.
30) Week Net Volume Percentage
The Day Net Volume Percentage is the Percentage of the positive volume relative to the total volume of the week.
31) Month Net Volume Percentage
The Day Net Volume Percentage is the Percentage of the positive volume relative to the total volume of the month.
"""
def __init__(self, algo, name, length):
self.name = name
self.algorithm = algo
self.length = length
self.IsReady = False
def calculate(self):
total_vol = 0
positive_vol = 0
last_day = self.algorithm.MinuteHistory[self.name][0].Time.date()
day_count = 0
for i in range(self.algorithm.MinuteHistory[self.name].Count):
if last_day != self.algorithm.MinuteHistory[self.name][i].Time.date():
day_count += 1
last_day = self.algorithm.MinuteHistory[self.name][i].Time.date()
if day_count >= self.length:
break
total_vol += self.algorithm.MinuteHistory[self.name][i].Volume
if self.algorithm.MinuteHistory[self.name][i].Close >= self.algorithm.MinuteHistory[self.name][i+1].Close:
positive_vol += self.algorithm.MinuteHistory[self.name][i].Volume
if total_vol == 0:
return
negative_vol = total_vol - positive_vol
self.Value = (positive_vol - negative_vol) * 100 / total_vol
self.IsReady = True
from AlgorithmImports import *
from metrics.Metric import Metric
class PercentageAboveDailyVolume(Metric):
"""
27) Percentage Above Daily Volume
The Percentage of the end of day accumulated volume estimated using
the up to current time accumulated volume relative to the 30 DMA Volume.
"""
def __init__(self, algo, name):
self.name = name
self.algorithm = algo
self.IsReady = False
def calculate(self):
if not self.algorithm.symbols[self.name].metrics['Volume30DMA'].IsReady:
return
vol_today = 0
minute_bar_count = 0
last_day = self.algorithm.MinuteHistory[self.name][0].Time.date()
for bar in self.algorithm.MinuteHistory[self.name]:
if bar.Time.date() != last_day:
break
vol_today += bar.Volume
minute_bar_count += 1
volume_per_minute = vol_today / minute_bar_count
volume = volume_per_minute * 390
if volume == 0:
return
self.Value = (volume - self.algorithm.symbols[self.name].metrics['Volume30DMA'].Value) * 100 / volume
self.IsReady = True
from AlgorithmImports import *
from metrics.Metric import Metric
class PercentageOfMarketPriceVsVWAP(Metric):
"""
9) Percentage of Market Price Vs VWAP
The percentage difference between last price and Vwap.
(Price - VWAP) * 100 / Price
"""
def __init__(self, algo, name):
self.name = name
self.algorithm = algo
self.IsReady = False
def calculate(self):
if not self.algorithm.symbols[self.name].metrics['VWAP'].IsReady:
return
self.Value = (self.algorithm.Securities[self.name].Price - self.algorithm.symbols[self.name].metrics['VWAP'].Value) * 100 /\
self.algorithm.Securities[self.name].Price
self.IsReady = True
from AlgorithmImports import *
from metrics.Metric import Metric
class VWAP(Metric):
"""
8) VWAP
The volume weighted moving average price.
"""
def __init__(self, algo, name):
self.name = name
self.algorithm = algo
symbol = self.algorithm.Securities[name].Symbol
self.vwap = self.algorithm.VWAP(symbol)
self.IsReady = False
def calculate(self):
if self.vwap.IsReady:
self.Value = self.vwap.Current.Value
self.IsReady = True
from AlgorithmImports import *
from metrics.Metric import Metric
class VolumeChangePercentage(Metric):
"""
34) 1 Day Volume Change Percentage
The percentage change of the current day volume up to the current time relative to the previous day volume up to the same time.
35) 7 Day Volume Change Percentage
The percentage change of the current day volume up to the current time relative to the day volume up to the same time 7 days ago.
36) 30 Day Volume Change Percentage
The percentage change of the current day volume up to the current time relative to the day volume up to the same time 30 days ago.
"""
def __init__(self, algo, name, length):
self.name = name
self.algorithm = algo
self.length = length
self.IsReady = False
def calculate(self):
vol_today = 0
minute_bar_count = 0
last_day = self.algorithm.MinuteHistory[self.name][0].Time.date()
for bar in self.algorithm.MinuteHistory[self.name]:
if bar.Time.date() != last_day:
break
vol_today += bar.Volume
minute_bar_count += 1
if vol_today == 0:
return
# if the day is not closed yet we compare current volume with the volume of length - 1 bars ago
length = self.length
if self.algorithm.MinuteHistory[self.name][0].Time.date() != self.algorithm.DayHistory[self.name][0].Time.date():
length -= 1
last_day = self.algorithm.DayHistory[self.name][length].Time.date()
start_bar = self.get_start_bar(last_day)
start_bar -= minute_bar_count - 1
last_vol = 0
for i in range(minute_bar_count):
last_vol += self.algorithm.MinuteHistory[self.name][i+start_bar].Volume
self.Value = (vol_today - last_vol) * 100 / vol_today
self.IsReady = True
def get_start_bar(self, date):
'''
binary search to get the start bar of minute history of x day ago
'''
low = 0
high = self.algorithm.MinuteHistory[self.name].Count - 1
while low < high:
mid = low + (high - low + 1) // 2
if self.algorithm.MinuteHistory[self.name][mid].Time.date() < date:
high = mid - 1
else:
low = mid
return low
from AlgorithmImports import *
from metrics.Metric import Metric
from statistics import mean
class VolumeDMA(Metric):
"""
32) 7 DMA Volume
The 7 Day Simple Moving Average of Volume
33) 30 DMA Volume
The 30 Day Simple Moving Average of Volume
"""
def __init__(self, algo, name, length):
self.name = name
self.algorithm = algo
self.length = length
self.IsReady = False
def calculate(self):
self.Value = mean([bar.Volume for bar in list(self.algorithm.DayHistory[self.name])[:self.length]])
self.IsReady = True
from AlgorithmImports import *
from metrics.Metric import Metric
class VolumePerTimeMultipleVs30DMA(Metric):
"""
11) Volume Per Time Multiple Vs 30DMA Volume Per Time Multiple
The day volume per minute calculated by the total daily volume and dividing it by the number of minutes passed in the day.
It is then divided by its 30 Day Simple Moving Average to get the volume multiple.
"""
def __init__(self, algo, name):
self.name = name
self.algorithm = algo
self.IsReady = False
def calculate(self):
vol_today = sum([x.Volume for x in self.algorithm.MinuteHistory[self.name]
if x.Time.date() == self.algorithm.MinuteHistory[self.name][0].Time.date()]) /\
len([x for x in self.algorithm.MinuteHistory[self.name]
if x.Time.date() == self.algorithm.MinuteHistory[self.name][0].Time.date()])
bar_count = 0
vol_30_day = 0
days_count = 0
last_day = self.algorithm.MinuteHistory[self.name][0].Time.date()
for bar in self.algorithm.MinuteHistory[self.name]:
if last_day != bar.Time.date():
days_count += 1
last_day = bar.Time.date()
if days_count > 30:
break
if not days_count:
continue
if bar.Time.time() <= self.algorithm.MinuteHistory[self.name][0].Time.time():
bar_count += 1
vol_30_day += bar.Volume
avg_vol_30_day = vol_30_day / bar_count
self.Value = vol_today / avg_vol_30_day
self.IsReady = True
from asyncio import FastChildWatcher
from AlgorithmImports import *
from config import *
from datetime import datetime
from trades.ITrade import Trade
from typing import List
class Symbol:
def __init__(self, algo, name):
self.algorithm = algo
self.last_trade= [datetime.min,datetime.min,datetime.min,datetime.min,datetime.min]
self.name = name
self.metrics = {}
self.trades: List[Trade] = []
def addMetric(self):
config_metrics=self.algorithm.config.metrics
for key in config_metrics:
for item in config_metrics[key]:
chart_name=item[0]
if len(item)==3:
self.metrics[chart_name]=item[1](self.algorithm,self.name,*item[2])
else:
self.metrics[chart_name]=item[1](self.algorithm,self.name)
def addTrades(self):
config_trades = self.algorithm.config.trades
for trade in config_trades:
self.trades.append(trade(self))
def update_trade(self):
for trade in self.trades:
if trade.in_order:
trade.update()
def trade(self):
if self.in_trade:
return
for trade in self.trades:
if self.in_trade is False:
trade.entry_order()
def handle_order_event(self, orderEvent: OrderEvent):
for trade in self.trades:
if trade.in_order:
trade.handle_order_event(orderEvent)
def update(self,type):
for metric in [item[0] for item in self.algorithm.config.metrics[type]]:
self.metrics[metric].calculate()
self.plot(type)
def plot(self,type):
for metric in [item[0] for item in self.algorithm.config.metrics[type]]:
if metric in self.algorithm.config.plot_metrics:
chart=Chart(f"{metric}")
self.algorithm.AddChart(chart)
if self.metrics[metric].IsReady :
self.algorithm.Plot(f"{metric}", f"{self.name}", self.metrics[metric].Value)
@property
def in_trade(self):
result = False
for trade in self.trades:
result |= trade.in_order
return resultfrom AlgorithmImports import *
from config import Config
class Condition:
def __init__(self):
self.parameters = Config.set_parameters()
def base_condition(self, symbol, C, SameConditionTimeC):
"Time Since the last Entry for the Symbol for Trade 1 is at least SameConditionTimeC1 minutes"
id = int(C[1]) - 1
if (
self.parameters["PauseAlgo"] == False
and self.open_position(symbol) == False
and self.parameters["Trading Algorithm"]
and self.parameters[C]
and (symbol.algorithm.Time - symbol.last_trade[id]).seconds / 60
>= self.parameters[SameConditionTimeC]
):
return True
return False
def open_position(self, symbol):
"There is open position for the Symbol"
symbol_open_orders = symbol.algorithm.Transactions.GetOpenOrders(symbol.name)
if len(symbol_open_orders) == 0:
return False
return True
def percentage_of_market_price_vs_vwap_condition(self, symbol):
"Percentage of Market Price Vs Vwap is less than 0 and has absolute value greater than 7 DMA Range Price * VWAP_PCT"
market_price_vwmp = symbol.metrics["PercentageOfMarketPriceVsVWAP"]
price_change_7dma = symbol.metrics["DayPercentageChangeOf7DMARangePercentage"]
if market_price_vwmp.IsReady and price_change_7dma.IsReady:
return (market_price_vwmp.Value, price_change_7dma.Value)
return (-1, -1)
def days_price_range_percentage_condition(self, symbol):
"Day’s Price Range Percentage is equal to or less than( -1 * (Parameter 1 / 100) * 30 DMA Range Price)"
price_day_range = symbol.metrics["PricePercentageOfDayRange"]
price_30day_range = symbol.metrics["PricePercentageOf30DayRange"]
if price_day_range.IsReady and price_30day_range.IsReady:
return (price_day_range.Value, price_30day_range.Value)
return (-1, -1)
def ratio_of_the_Maximum_Down_Percentage(self, symbol):
"""
The ratio of the Maximum Down Percentage in the last 30 minutes of the day to the 30 DMA Range Price Percentage
is less than or equal (-1 * Parameter 3) / 100.0
"""
price_30day_range = symbol.metrics["PricePercentageOf30DayRange"]
if price_30day_range.IsReady == False:
return False
last_30_minutes = list(symbol.algorithm.MinuteHistory[symbol.name])[:30]
max_high = 0
max_down_percentage = 0
for v in reversed(last_30_minutes):
if v.High > max_high:
max_high = v.High
continue
if max_high > v.Low:
max_down_percentage = max(max_down_percentage, (max_high - v.Low) * 100 / max_high)
ratio = max_down_percentage / price_30day_range.Value
return ratio <= (-1 * self.parameters["Parameter 3"] / 100)
def last_bar_makes_a_new_day_low(self, symbol):
low = symbol.algorithm.SecondHistory[symbol.name][0].Low
return low == symbol.metrics["DayLow"]
def last_bar_makes_a_new_day_high(self, symbol):
high = symbol.algorithm.SecondHistory[symbol.name][0].High
return high == symbol.metrics["DayHigh"]
def rally_condition(self, symbol):
price = symbol.metrics["Price"]
price_day_range = symbol.metrics["PricePercentageOfDayRange"]
if price_day_range.IsReady:
percentage_increase = 0
low = 90000000
high = 0
last_day = symbol.algorithm.MinuteHistory[symbol.name][0].Time.date()
lst_of_bars = []
for bar in symbol.algorithm.MinuteHistory[symbol.name]:
if bar.Time.date() != last_day:
break
lst_of_bars.append(bar)
lst_of_bars.reverse()
# part one
for bar in lst_of_bars:
if bar.Low < low:
low = bar.Low
high = 0
percentage_increase = 0
continue
if bar.High > high:
high = bar.High
percentage_increase = (high - low) * 100 / low
percentage_a = float(percentage_increase / price_day_range.Value) * 100
# part 2
if high == 0:
precentage_b = abs(
float(lst_of_bars[-1].High - lst_of_bars[-1].Low) * 100 / lst_of_bars[-1].High
)
else:
precentage_b = abs(float(high - lst_of_bars[-1].Low) * 100 / high)
precentage_b = float(precentage_b / price_day_range.Value) * 100
# conditions
reset = False
condition_1 = False
condition_2 = False
if (
self.parameters["Rally_X_Min_PCT"]
<= percentage_a
<= self.parameters["Rally_X_Max_PCT"]
):
condition_1 = True
if precentage_b >= self.parameters["Rally_Y_PCT"]:
condition_2 = True
if percentage_a > self.parameters["Rally_X_Max_PCT"]:
reset = True
return {"reset": reset, "condition": condition_1 and condition_2}
else:
return {"reset": False, "condition": False}
def short_rally_condition(self, symbol):
price = symbol.metrics["Price"]
price_day_range = symbol.metrics["PricePercentageOfDayRange"]
if price_day_range.IsReady:
percentage_increase = 0
low = 90000000
high = 0
last_day = symbol.algorithm.MinuteHistory[symbol.name][0].Time.date()
lst_of_bars = []
for bar in symbol.algorithm.MinuteHistory[symbol.name]:
if bar.Time.date() != last_day:
break
lst_of_bars.append(bar)
lst_of_bars.reverse()
# part one
for bar in lst_of_bars:
if bar.High > high:
high = bar.High
low = 90000000
percentage_increase = 0
continue
if bar.Low < low:
low = bar.Low
percentage_increase = abs(high - low) * 100 / high
percentage_a = float(percentage_increase / price_day_range.Value) * 100
# part 2
if low == 90000000:
precentage_b = abs(
float(lst_of_bars[-1].High - lst_of_bars[-1].Low) * 100 / lst_of_bars[-1].Low
)
else:
precentage_b = abs(float(lst_of_bars[-1].High - low) * 100 / low)
precentage_b = float(precentage_b / price_day_range.Value) * 100
# conditions
reset = False
condition_1 = False
condition_2 = False
if (
self.parameters["Rally_X_Min_PCT"]
<= percentage_a
<= self.parameters["Rally_X_Max_PCT"]
):
condition_1 = True
if precentage_b >= self.parameters["Rally_Y_PCT"]:
condition_2 = True
if percentage_a > self.parameters["Rally_X_Min_PCT"]:
reset = True
return {"reset": reset, "condition": condition_1 and condition_2}
else:
return {"reset": False, "condition": False}
from abc import ABC, abstractmethod
from AlgorithmImports import *
from symbols.symbol import Symbol
class Trade(ABC):
def __init__(self, symbol: Symbol) -> None:
self.symbol: Symbol = symbol
self.algorithm: QCAlgorithm = symbol.algorithm
self.ticket: OrderTicket = None
self.parameters = self.algorithm.config.set_parameters()
def base_condition(self, C, SameConditionTimeC):
id = int(C[1]) - 1
if (
self.parameters["PauseAlgo"] == False
and self.symbol.in_trade == False
and self.parameters["Trading Algorithm"]
and self.parameters[C]
and (self.algorithm.Time - self.symbol.last_trade[id]).seconds / 60
>= self.parameters[SameConditionTimeC]
):
return True
return False
@abstractmethod
def condition(self):
pass
@abstractmethod
def entry_order(self):
pass
@abstractmethod
def exit_order(self):
pass
@abstractmethod
def update(self):
pass
def handle_order_event(self, orderEvent: OrderEvent):
if orderEvent.Status == OrderStatus.Filled:
if orderEvent.OrderId == self.ticket.OrderId:
self.exit_order()
else:
self.Transactions.CancelOpenOrders(self.symbol.name)
self.ticket = None
@property
def in_order(self):
return self.ticket is not None
class LongTrade(Trade):
def __init__(self, symbol: Symbol) -> None:
super().__init__(symbol)
def rally_condition(self):
price = self.symbol.metrics["Price"]
price_day_range = self.symbol.metrics["PricePercentageOfDayRange"]
if price_day_range.IsReady:
percentage_increase = 0
low = 90000000
high = 0
last_day = self.symbol.algorithm.MinuteHistory[self.symbol.name][0].Time.date()
lst_of_bars = []
for bar in self.symbol.algorithm.MinuteHistory[self.symbol.name]:
if bar.Time.date() != last_day:
break
lst_of_bars.append(bar)
lst_of_bars.reverse()
# part one
for bar in lst_of_bars:
if bar.Low < low:
low = bar.Low
high = 0
percentage_increase = 0
continue
if bar.High > high:
high = bar.High
percentage_increase = (high - low) * 100 / low
percentage_a = float(percentage_increase / price_day_range.Value) * 100
# part 2
if high == 0:
precentage_b = abs(
float(lst_of_bars[-1].High - lst_of_bars[-1].Low) * 100 / lst_of_bars[-1].High
)
else:
precentage_b = abs(float(high - lst_of_bars[-1].Low) * 100 / high)
precentage_b = float(precentage_b / price_day_range.Value) * 100
# conditions
reset = False
condition_1 = False
condition_2 = False
if (
self.parameters["Rally_X_Min_PCT"]
<= percentage_a
<= self.parameters["Rally_X_Max_PCT"]
):
condition_1 = True
if precentage_b >= self.parameters["Rally_Y_PCT"]:
condition_2 = True
if percentage_a > self.parameters["Rally_X_Max_PCT"]:
reset = True
return {"reset": reset, "condition": condition_1 and condition_2}
else:
return {"reset": False, "condition": False}
def condition(self):
return super().condition()
def entry_order(self):
return super().entry_order()
def exit_order(self):
return super().exit_order()
def update(self):
return super().update()
class ShortTrade(Trade):
def __init__(self, symbol: Symbol) -> None:
super().__init__(symbol)
def rally_condition(self):
price = self.symbol.metrics["Price"]
price_day_range = self.symbol.metrics["PricePercentageOfDayRange"]
if price_day_range.IsReady:
percentage_increase = 0
low = 90000000
high = 0
last_day = self.symbol.algorithm.MinuteHistory[self.symbol.name][0].Time.date()
lst_of_bars = []
for bar in self.symbol.algorithm.MinuteHistory[self.symbol.name]:
if bar.Time.date() != last_day:
break
lst_of_bars.append(bar)
lst_of_bars.reverse()
# part one
for bar in lst_of_bars:
if bar.High > high:
high = bar.High
low = 90000000
percentage_increase = 0
continue
if bar.Low < low:
low = bar.Low
percentage_increase = abs(high - low) * 100 / high
percentage_a = float(percentage_increase / price_day_range.Value) * 100
# part 2
if low == 90000000:
precentage_b = abs(
float(lst_of_bars[-1].High - lst_of_bars[-1].Low) * 100 / lst_of_bars[-1].Low
)
else:
precentage_b = abs(float(lst_of_bars[-1].High - low) * 100 / low)
precentage_b = float(precentage_b / price_day_range.Value) * 100
# conditions
reset = False
condition_1 = False
condition_2 = False
if (
self.parameters["Rally_X_Min_PCT"]
<= percentage_a
<= self.parameters["Rally_X_Max_PCT"]
):
condition_1 = True
if precentage_b >= self.parameters["Rally_Y_PCT"]:
condition_2 = True
if percentage_a > self.parameters["Rally_X_Max_PCT"]:
reset = True
return {"reset": reset, "condition": condition_1 and condition_2}
else:
return {"reset": False, "condition": False}
def condition(self):
return super().condition()
def entry_order(self):
return super().entry_order()
def exit_order(self):
return super().exit_order()
def update(self):
return super().update()
from AlgorithmImports import *
from trades.ITrade import *
class TradeFive(LongTrade):
def __init__(self, symbol):
super().__init__(symbol)
self.condition1_5 = False
self.condition6 = False
self.condition7 = False
self.condition8 = False
def percentage_of_market_price_vs_vwap_condition(self):
market_price_vwmp = self.symbol.metrics["PercentageOfMarketPriceVsVWAP"]
price_change_7dma = self.symbol.metrics["DayPercentageChangeOf7DMARangePercentage"]
if market_price_vwmp.IsReady and price_change_7dma.IsReady:
return market_price_vwmp.Value > (
price_change_7dma.Value * self.parameters["VWAP_PCT"]
)
return False
def days_price_range_percentage_condition(self):
price_day_range = self.symbol.metrics["PricePercentageOfDayRange"]
price_30day_range = self.symbol.metrics["PricePercentageOf30DayRange"]
if price_day_range.IsReady and price_30day_range.IsReady:
return (
price_day_range.Value
>= (self.parameters["Parameter 2"] / 100) * price_30day_range.Value
)
return False
def last_bar_makes_a_new_day_high(self):
high = self.symbol.algorithm.SecondHistory[self.symbol.name][0].High
return high == self.symbol.metrics["DayHigh"]
def condition(self):
if self.condition1_5 and self.condition6 and self.condition7:
data = self.rally_condition()
if data["condition"]:
return True
if data["reset"] == False:
self.condition1_5 = False
self.condition6 = False
self.condition7 = False
self.condition8 = False
return False
else:
self.condition1_5 = self.base_condition("C5", "SameConditionTimeC5")
self.condition6 = self.percentage_of_market_price_vs_vwap_condition
self.condition7 = self.last_bar_makes_a_new_day_high()
self.condition8 = self.days_price_range_percentage_condition()
return False
def entry_order(self):
if self.in_order or self.condition() is False:
return
if not self.symbol.metrics["DayLow"].IsReady:
return
stop_price = self.Securities[self.symbol].Price - (
self.algorithm.config["Parameter 4"] / 100
) * (self.symbol.metrics["DayHigh"] - self.symbol.metrics["DayLow"])
limit_price_offset = (self.parameters["OffsetPCT"] / 100) * stop_price
limit_price = stop_price + limit_price_offset
quantity = self.algorithm.config["CashRisk-1"] // limit_price
self.ticket = self.StopLimitOrder(self.symbol, quantity, stop_price, limit_price)
def update(self):
if not self.in_order:
return
if self.ticket.Status == OrderStatus.Filled:
return
stop_price = self.Securities[self.symbol].Price - (
self.algorithm.config["Parameter 4"] / 100
) * (self.symbol.metrics["DayHigh"] - self.symbol.metrics["DayLow"])
if stop_price > self.ticket.Get(OrderField.StopPrice):
return
limit_price_offset = (self.parameters["OffsetPCT"] / 100) * stop_price
limit_price = stop_price - limit_price_offset
quantity = self.algorithm.config["CashRisk-1"] // limit_price
update_settings = UpdateOrderFields()
update_settings.StopPrice = stop_price
update_settings.LimitPrice = limit_price
update_settings.Quantity = quantity
response = self.ticket.Update(update_settings)
if not response.IsSuccess:
self.Debug("Order failed to update")
def exit_order(self):
size: float = -self.ticket.QuantityFilled * (self.algorithm.config["SharesToSell"] / 100)
fill_price: float = self.ticket.AverageFillPrice
tp_limit_price = (
fill_price
- fill_price
* (self.algorithm.config["ProfitTakeC5"] / 100)
* self.symbol.metrics["RangePercentage30DMA"]
)
self.algorithm.LimitOrder(self.symbol.name, size, tp_limit_price)
sl_price = (
fill_price
+ fill_price
* (self.algorithm.config["StopLoss"] / 100)
* self.symbol.metrics["RangePercentage30DMA"]
)
self.algorithm.StopMarketOrder(self.symbol.name, size, sl_price)
from AlgorithmImports import *
from trades.ITrade import *
class TradeFour(ShortTrade):
def __init__(self, symbol):
super().__init__(symbol)
self.condition1_5 = False
self.condition6 = False
self.condition7 = False
def percentage_of_market_price_vs_vwap_condition(self):
market_price_vwmp = self.symbol.metrics["PercentageOfMarketPriceVsVWAP"]
price_change_7dma = self.symbol.metrics["DayPercentageChangeOf7DMARangePercentage"]
if market_price_vwmp.IsReady and price_change_7dma.IsReady:
return market_price_vwmp.Value >= (
price_change_7dma.Value * self.parameters["VWAP_PCT"]
)
return False
def days_price_range_percentage_condition(self):
price_day_range = self.symbol.metrics["PricePercentageOfDayRange"]
price_30day_range = self.symbol.metrics["PricePercentageOf30DayRange"]
if price_day_range.IsReady and price_30day_range.IsReady:
return (
price_day_range.Value
>= (self.parameters["Parameter 1"] / 100) * price_30day_range.Value
)
return False
def condition(self):
if self.condition1_5 and self.condition6 and self.condition7:
data = self.rally_condition()
if data["condition"]:
return True
if data["reset"] == False:
self.condition1_5 = False
self.condition6 = False
self.condition7 = False
return False
else:
self.condition1_5 = self.base_condition("C4", "SameConditionTimeC4")
self.condition6 = self.percentage_of_market_price_vs_vwap_condition()
self.condition7 = self.days_price_range_percentage_condition()
return False
def entry_order(self):
if self.in_order or self.condition() is False:
return
if not self.symbol.metrics["DayLow"].IsReady:
return
stop_price = self.Securities[self.symbol].Price - (
self.algorithm.config["Parameter 4"] / 100
) * (self.symbol.metrics["DayHigh"] - self.symbol.metrics["DayLow"])
limit_price_offset = (self.parameters["OffsetPCT"] / 100) * stop_price
limit_price = stop_price - limit_price_offset
quantity = self.algorithm.config["CashRisk-1"] // limit_price
self.ticket = self.StopLimitOrder(self.symbol, quantity, stop_price, limit_price)
def update(self):
if not self.in_order:
return
if self.ticket.Status == OrderStatus.Filled:
return
stop_price = self.Securities[self.symbol].Price - (
self.algorithm.config["Parameter 4"] / 100
) * (self.symbol.metrics["DayHigh"] - self.symbol.metrics["DayLow"])
if stop_price > self.ticket.Get(OrderField.StopPrice):
return
limit_price_offset = (self.parameters["OffsetPCT"] / 100) * stop_price
limit_price = stop_price + limit_price_offset
quantity = self.algorithm.config["CashRisk-1"] // limit_price
update_settings = UpdateOrderFields()
update_settings.StopPrice = stop_price
update_settings.LimitPrice = limit_price
update_settings.Quantity = quantity
response = self.ticket.Update(update_settings)
if not response.IsSuccess:
self.Debug("Order failed to update")
def exit_order(self):
size: float = -self.ticket.QuantityFilled * (self.algorithm.config["SharesToSell"] / 100)
fill_price: float = self.ticket.AverageFillPrice
tp_limit_price = (
fill_price
- fill_price
* (self.algorithm.config["ProfitTakeC4"] / 100)
* self.symbol.metrics["RangePercentage30DMA"]
)
self.algorithm.LimitOrder(self.symbol.name, size, tp_limit_price)
sl_price = (
fill_price
+ fill_price
* (self.algorithm.config["StopLoss"] / 100)
* self.symbol.metrics["RangePercentage30DMA"]
)
self.algorithm.StopMarketOrder(self.symbol.name, size, sl_price)
from AlgorithmImports import *
from trades.ITrade import *
class TradeOne(LongTrade):
def __init__(self, symbol):
super().__init__(symbol)
self.condition1_5 = False
self.condition6 = False
self.condition7 = False
def percentage_of_market_price_vs_vwap_condition(self):
market_price_vwmp = self.symbol.metrics["PercentageOfMarketPriceVsVWAP"]
price_change_7dma = self.symbol.metrics["DayPercentageChangeOf7DMARangePercentage"]
if market_price_vwmp.IsReady and price_change_7dma.IsReady:
return market_price_vwmp.Value < 0 and abs(market_price_vwmp.Value) > (
price_change_7dma.Value * self.parameters["VWAP_PCT"]
)
return False
def days_price_range_percentage_condition(self):
price_day_range = self.symbol.metrics["PricePercentageOfDayRange"]
price_30day_range = self.symbol.metrics["PricePercentageOf30DayRange"]
if price_day_range.IsReady and price_30day_range.IsReady:
return (
price_day_range.Value
<= -1 * (self.parameters["Parameter 1"] / 100) * price_30day_range.Value
)
return False
def condition(self):
if self.condition1_5 and self.condition6 and self.condition7:
data = self.rally_condition()
if data["condition"]:
return True
if data["reset"] == False:
self.condition1_5 = False
self.condition6 = False
self.condition7 = False
return False
else:
self.condition1_5 = self.base_condition("C1", "SameConditionTimeC1")
self.condition6 = self.percentage_of_market_price_vs_vwap_condition()
self.condition7 = self.days_price_range_percentage_condition()
return False
def entry_order(self):
if self.in_order or self.condition() is False:
return
if not self.symbol.metrics["DayLow"].IsReady:
return
stop_price = self.Securities[self.symbol].Price + (
self.algorithm.config["Parameter 4"] / 100
) * (self.symbol.metrics["DayHigh"] - self.symbol.metrics["DayLow"])
limit_price_offset = (self.parameters["OffsetPCT"] / 100) * stop_price
limit_price = stop_price + limit_price_offset
quantity = self.algorithm.config["CashRisk-1"] // limit_price
self.ticket = self.StopLimitOrder(self.symbol, quantity, stop_price, limit_price)
def update(self):
if not self.in_order:
return
if self.ticket.Status == OrderStatus.Filled:
return
stop_price = self.Securities[self.symbol].Price + (
self.algorithm.config["Parameter 4"] / 100
) * (self.symbol.metrics["DayHigh"] - self.symbol.metrics["DayLow"])
if stop_price < self.ticket.Get(OrderField.StopPrice):
return
limit_price_offset = (self.parameters["OffsetPCT"] / 100) * stop_price
limit_price = stop_price + limit_price_offset
quantity = self.algorithm.config["CashRisk-1"] // limit_price
update_settings = UpdateOrderFields()
update_settings.StopPrice = stop_price
update_settings.LimitPrice = limit_price
update_settings.Quantity = quantity
response = self.ticket.Update(update_settings)
if not response.IsSuccess:
self.Debug("Order failed to update")
def exit_order(self):
size: float = -self.ticket.QuantityFilled * (self.algorithm.config["SharesToSell"] / 100)
fill_price: float = self.ticket.AverageFillPrice
tp_limit_price = (
fill_price
+ fill_price
* (self.algorithm.config["ProfitTakeC1"] / 100)
* self.symbol.metrics["RangePercentage30DMA"]
)
self.algorithm.LimitOrder(self.symbol.name, size, tp_limit_price)
sl_price = (
fill_price
- fill_price
* (self.algorithm.config["StopLoss"] / 100)
* self.symbol.metrics["RangePercentage30DMA"]
)
self.algorithm.StopMarketOrder(self.symbol.name, size, sl_price)
from AlgorithmImports import *
from trades.ITrade import *
class TradeThree(LongTrade):
def __init__(self, symbol):
super().__init__(symbol)
self.condition1_5 = False
self.condition6 = False
def ratio_of_the_Maximum_Down_Percentage(self):
price_30day_range = self.symbol.metrics["PricePercentageOf30DayRange"]
if price_30day_range.IsReady == False:
return False
last_30_minutes = list(self.symbol.algorithm.MinuteHistory[self.symbol.name])[:30]
max_high = 0
max_down_percentage = 0
for v in reversed(last_30_minutes):
if v.High > max_high:
max_high = v.High
continue
if max_high > v.Low:
max_down_percentage = max(max_down_percentage, (max_high - v.Low) * 100 / max_high)
ratio = max_down_percentage / price_30day_range.Value
return ratio <= (-1 * self.parameters["Parameter 3"] / 100)
def condition(self):
self.condition1_5 = self.condition.base_condition("C3", "SameConditionTimeC3")
self.condition6 = self.condition.ratio_of_the_Maximum_Down_Percentage()
return self.condition1_5 and self.condition6
def entry_order(self):
if self.in_order or self.condition() is False:
return
if not self.symbol.metrics["DayLow"].IsReady:
return
stop_price = self.Securities[self.symbol].Price + (
self.algorithm.config["Parameter 5"] / 100
) * (self.symbol.metrics["DayHigh"] - self.symbol.metrics["DayLow"])
limit_price_offset = (self.parameters["OffsetPCT"] / 100) * stop_price
limit_price = stop_price + limit_price_offset
quantity = self.algorithm.config["CashRisk-1"] // limit_price
self.ticket = self.StopLimitOrder(self.symbol, quantity, stop_price, limit_price)
def update(self):
if not self.in_order:
return
if self.ticket.Status == OrderStatus.Filled:
return
stop_price = self.Securities[self.symbol].Price + (
self.algorithm.config["Parameter 5"] / 100
) * (self.symbol.metrics["DayHigh"] - self.symbol.metrics["DayLow"])
if stop_price < self.ticket.Get(OrderField.StopPrice):
return
limit_price_offset = (self.parameters["OffsetPCT"] / 100) * stop_price
limit_price = stop_price + limit_price_offset
quantity = self.algorithm.config["CashRisk-1"] // limit_price
update_settings = UpdateOrderFields()
update_settings.StopPrice = stop_price
update_settings.LimitPrice = limit_price
update_settings.Quantity = quantity
response = self.ticket.Update(update_settings)
if not response.IsSuccess:
self.Debug("Order failed to update")
def exit_order(self):
size: float = -self.ticket.QuantityFilled * (self.algorithm.config["SharesToSell"] / 100)
fill_price: float = self.ticket.AverageFillPrice
tp_limit_price = (
fill_price
+ fill_price
* (self.algorithm.config["ProfitTakeC3"] / 100)
* self.symbol.metrics["RangePercentage30DMA"]
)
self.algorithm.LimitOrder(self.symbol.name, size, tp_limit_price)
sl_price = (
fill_price
- fill_price
* (self.algorithm.config["StopLoss"] / 100)
* self.symbol.metrics["RangePercentage30DMA"]
)
self.algorithm.StopMarketOrder(self.symbol.name, size, sl_price)
from AlgorithmImports import *
from trades.ITrade import *
class TradeTwo(LongTrade):
def __init__(self, symbol):
super().__init__(symbol)
self.condition1_5 = False
self.condition6 = False
self.condition7 = False
self.condition8 = False
def percentage_of_market_price_vs_vwap_condition(self):
market_price_vwmp = self.symbol.metrics["PercentageOfMarketPriceVsVWAP"]
price_change_7dma = self.symbol.metrics["DayPercentageChangeOf7DMARangePercentage"]
if market_price_vwmp.IsReady and price_change_7dma.IsReady:
return market_price_vwmp.Value < 0 and abs(market_price_vwmp.Value) > (
price_change_7dma.Value * self.parameters["VWAP_PCT"]
)
return False
def days_price_range_percentage_condition(self):
price_day_range = self.symbol.metrics["PricePercentageOfDayRange"]
price_30day_range = self.symbol.metrics["PricePercentageOf30DayRange"]
if price_day_range.IsReady and price_30day_range.IsReady:
return (
price_day_range.Value
>= (self.parameters["Parameter 2"] / 100) * price_30day_range.Value
)
return False
def last_bar_makes_a_new_day_low(self):
low = self.symbol.algorithm.SecondHistory[self.symbol.name][0].Low
return low == self.symbol.metrics["DayLow"]
def condition(self):
if self.condition1_5 and self.condition6 and self.condition7 and self.condition8:
data = self.rally_condition()
if data["condition"]:
return True
if data["reset"] == False:
self.condition1_5 = False
self.condition6 = False
self.condition7 = False
self.condition8 = False
return False
else:
self.condition1_5 = self.base_condition("C2", "SameConditionTimeC2")
self.condition6 = self.percentage_of_market_price_vs_vwap_condition()
self.condition7 = self.last_bar_makes_a_new_day_low()
self.condition8 = self.days_price_range_percentage_condition()
return False
def entry_order(self):
if self.in_order or self.condition() is False:
return
if not self.symbol.metrics["DayLow"].IsReady:
return
stop_price = self.Securities[self.symbol].Price + (
self.algorithm.config["Parameter 4"] / 100
) * (self.symbol.metrics["DayHigh"] - self.symbol.metrics["DayLow"])
limit_price_offset = (self.parameters["OffsetPCT"] / 100) * stop_price
limit_price = stop_price + limit_price_offset
quantity = self.algorithm.config["CashRisk-1"] // limit_price
self.ticket = self.StopLimitOrder(self.symbol, quantity, stop_price, limit_price)
def update(self):
if not self.in_order:
return
if self.ticket.Status == OrderStatus.Filled:
return
stop_price = self.Securities[self.symbol].Price + (
self.algorithm.config["Parameter 4"] / 100
) * (self.symbol.metrics["DayHigh"] - self.symbol.metrics["DayLow"])
if stop_price < self.ticket.Get(OrderField.StopPrice):
return
limit_price_offset = (self.parameters["OffsetPCT"] / 100) * stop_price
limit_price = stop_price + limit_price_offset
quantity = self.algorithm.config["CashRisk-1"] // limit_price
update_settings = UpdateOrderFields()
update_settings.StopPrice = stop_price
update_settings.LimitPrice = limit_price
update_settings.Quantity = quantity
response = self.ticket.Update(update_settings)
if not response.IsSuccess:
self.Debug("Order failed to update")
def exit_order(self):
size: float = -self.ticket.QuantityFilled * (self.algorithm.config["SharesToSell"] / 100)
fill_price: float = self.ticket.AverageFillPrice
tp_limit_price = (
fill_price
+ fill_price
* (self.algorithm.config["ProfitTakeC2"] / 100)
* self.symbol.metrics["RangePercentage30DMA"]
)
self.algorithm.LimitOrder(self.symbol.name, size, tp_limit_price)
sl_price = (
fill_price
- fill_price
* (self.algorithm.config["StopLoss"] / 100)
* self.symbol.metrics["RangePercentage30DMA"]
)
self.algorithm.StopMarketOrder(self.symbol.name, size, sl_price)
