| 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 0 Tracking Error 0 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.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 *
class Config:
def __init__(self):
self.set_symbols()
self.set_plot_metrics()
self.set_metrics()
def set_symbols(self):
self.symbols = ["AAPL"]
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),("OnePercentValue",OnePercentValue),("PercentageOfMarketPriceVsVWAP",PercentageOfMarketPriceVsVWAP)],
"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]),
("VWAP",VWAP),
("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])]
}
@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, 1, 20) # Set Start Date
self.SetCash(100000) # Set Strategy Cash
self.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()
# 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[float](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():
self.SecondHistory[symbol.Value].Add(bar.Volume)
if self.SecondHistory[symbol.Value].IsReady:
self.symbols[symbol.Value].update("real_time")
def WeekHistoryHandler(self, bar):
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):
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):
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
# __init__.py
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
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
# __init__.py
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
# __init__.py
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 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
# __init__.py
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(self.algorithm.SecondHistory[self.name])
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
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
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
self.vwap = VolumeWeightedAveragePriceIndicator(name, 1)
self.algorithm.RegisterIndicator(name, self.vwap, Resolution.Daily)
self.IsReady = False
def calculate(self):
# self.Value = (self.algorithm.DayHistory[self.name][0].Low + self.algorithm.DayHistory[self.name][0].High + self.algorithm.DayHistory[self.name][0].Close) / 3
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 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
# __init__.py
# __init__.py
from AlgorithmImports import *
from config import *
class Symbol:
def __init__(self, algo, name):
self.algorithm = algo
self.name = name
self.metrics = {}
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 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)