Types
Adx
type
Average directional index algorithm. Read here about how to use it.
# indie:lang_version = 5
from indie import indicator
from indie.algorithms import Adx
@indicator('Example')
def Main(self):
minus_di, adx, plus_di = Adx.new(adx_len=9, di_len=12)
return minus_di[0], adx[0], plus_di[0]
# indie:lang_version = 5
from indie import algorithm, SeriesF, MutSeriesF
from indie.algorithms import Change, Rma, Tr, FixNan
from indie.math import divide
@algorithm
def Adx(self, adx_len: int, di_len: int) -> tuple[SeriesF, SeriesF, SeriesF]:
'''Average Directional Index'''
up = Change.new(self.ctx.high)[0]
down = -Change.new(self.ctx.low)[0]
plus_dm = MutSeriesF.new(0 if up <= down or up <= 0 else up)
minus_dm = MutSeriesF.new(0 if down <= up or down <= 0 else down)
truerange = Rma.new(Tr.new(), di_len)
plus = MutSeriesF.new(100 * divide(Rma.new(plus_dm, di_len)[0], truerange[0]))
minus = MutSeriesF.new(100 * divide(Rma.new(minus_dm, di_len)[0], truerange[0]))
plus[0] = FixNan.new(plus)[0]
minus[0] = FixNan.new(minus)[0]
sum = plus[0] + minus[0]
res = 100 * Rma.new(MutSeriesF.new(abs(plus[0] - minus[0]) / (sum if sum != 0 else 1)), adx_len)[0]
return minus, MutSeriesF.new(res), plus
Atr
type
Average true range algorithm. Read here about how to use it.
# indie:lang_version = 5
from indie import indicator
from indie.algorithms import Atr
@indicator('Example')
def Main(self):
atr = Atr.new(length=12, ma_algorithm='SMA')
return atr[0]
# indie:lang_version = 5
from indie import algorithm, SeriesF
from indie.algorithms import Tr, Ma
@algorithm
def Atr(self, length: int, ma_algorithm: str = 'RMA') -> SeriesF:
'''Average True Range'''
return Ma.new(Tr.new(True), length, ma_algorithm)
Bb
type
Bollinger bands algorithm. Read here about how to use it.
# indie:lang_version = 5
from indie import indicator
from indie.algorithms import Bb
@indicator('Example', overlay_main_pane=True)
def Main(self):
lower, middle, upper = Bb.new(self.close, length=20, mult=2.0)
return lower[0], middle[0], upper[0]
# indie:lang_version = 5
from indie import algorithm, SeriesF, MutSeriesF
from indie.algorithms import Sma, StdDev
@algorithm
def Bb(self, src: SeriesF, length: int, mult: float) -> tuple[SeriesF, SeriesF, SeriesF]:
'''Bollinger Bands'''
middle = Sma.new(src, length)[0]
dev = mult * StdDev.new(src, length)[0]
lower = middle - dev
upper = middle + dev
return MutSeriesF.new(lower), MutSeriesF.new(middle), MutSeriesF.new(upper)
Cci
type
Commodity channel index algorithm. Read here about how to use it.
# indie:lang_version = 5
from indie import indicator
from indie.algorithms import Cci
@indicator('Example')
def Main(self):
cci = Cci.new(self.close, length=20)
return cci[0]
# indie:lang_version = 5
from indie import algorithm, SeriesF, MutSeriesF
from indie.algorithms import Sma, Dev
from indie.math import divide
@algorithm
def Cci(self, src: SeriesF, length: int) -> SeriesF:
'''Commodity Channel Index'''
ma = Sma.new(src, length)[0]
dv = Dev.new(src, length)[0]
res = divide(src[0] - ma, 0.015 * dv)
return MutSeriesF.new(res)
Change
type
Algorithm to calculate change of a series of values. Read here about how to use it.
# indie:lang_version = 5
from indie import indicator
from indie.algorithms import Change
@indicator('Example')
def Main(self):
ch = Change.new(self.close)
return ch[0]
# indie:lang_version = 5
from indie import algorithm, SeriesF, MutSeriesF
@algorithm
def Change(self, src: SeriesF, length: int = 1) -> SeriesF:
return MutSeriesF.new(src[0] - src[length])
Corr
type
Correlation coefficient algorithm. Read here about how to use it.
# indie:lang_version = 5
from indie import indicator
from indie.algorithms import Corr
@indicator('Example')
def Main(self):
corr = Corr.new(self.close, self.open, length=14)
return corr[0]
# indie:lang_version = 5
from indie import algorithm, SeriesF, MutSeriesF
from indie.algorithms import Sma, StdDev
from indie.math import divide
@algorithm
def Corr(self, x: SeriesF, y: SeriesF, length: int) -> SeriesF:
# NOTE: `cov` could be calculated as `E(x*y) - E(x)*E(y)`,
# but that may cause precision issues with floating-point arithmetic
e_x = Sma.new(x, length)[0]
e_y = Sma.new(y, length)[0]
cov = 0.0
for i in range(length):
cov += (x[i] - e_x) * (y[i] - e_y)
cov /= length
std_devs = StdDev.new(x, length)[0] * StdDev.new(y, length)[0]
corr = divide(cov, std_devs, 0.0)
return MutSeriesF.new(corr)
CumSum
type
Cumulative sum algorithm. Read here about how to use it.
# indie:lang_version = 5
from indie import indicator
from indie.algorithms import CumSum
@indicator('Example')
def Main(self):
cs = CumSum.new(self.close)
return cs[0]
# indie:lang_version = 5
from indie import algorithm, SeriesF, MutSeriesF
@algorithm
def CumSum(self, src: SeriesF) -> SeriesF:
res = MutSeriesF.new(init=0)
res[0] += src[0]
return res
Dev
type
Mean absolute deviation algorithm. Read here about how to use it.
# indie:lang_version = 5
from indie import indicator
from indie.algorithms import Dev
@indicator('Example')
def Main(self):
dev = Dev.new(self.close, length=12)
return dev[0]
# indie:lang_version = 5
from indie import algorithm, SeriesF, MutSeriesF
from indie.algorithms import Sma
@algorithm
def Dev(self, src: SeriesF, length: int) -> SeriesF:
'''Mean Absolute Deviation'''
mean = Sma.new(src, length)[0]
sum = 0.0
for i in range(length):
sum += abs(src[i] - mean)
return MutSeriesF.new(sum / length)
Donchian
type
Donchian channels algorithm. Read here about how to use it.
# indie:lang_version = 5
from indie import indicator
from indie.algorithms import Donchian
@indicator('DC', overlay_main_pane=True)
def Main(self):
d = Donchian.new(length=20)
return d[0]
# indie:lang_version = 5
from indie import algorithm, MutSeriesF, SeriesF
from indie.algorithms import Lowest, Highest
@algorithm
def Donchian(self, length: int) -> SeriesF:
lowest = Lowest.new(self.ctx.low, length)[0]
highest = Highest.new(self.ctx.high, length)[0]
return MutSeriesF.new((lowest + highest) / 2)
Ema
type
Exponential moving average algorithm. Read here about how to use it.
# indie:lang_version = 5
from indie import indicator
from indie.algorithms import Ema
@indicator('Example', overlay_main_pane=True)
def Main(self):
ema = Ema.new(self.close, length=9)
return ema[0]
# indie:lang_version = 5
from math import isnan, nan
from indie import algorithm, SeriesF, MutSeriesF
from indie.algorithms import Sma
@algorithm
def Ema(self, src: SeriesF, length: int) -> SeriesF:
'''Exponential Moving Average'''
alpha = 2 / (length + 1)
s = MutSeriesF.new(init=0)
res_sma = Sma.new(src, length)[0]
result = 0.0
if isnan(src[0]):
s[0] = nan
result = res_sma
elif isnan(s[1]):
s[0] = res_sma
result = s[0]
else:
s[0] = alpha * src[0] + (1 - alpha) * s[1]
result = s[0]
return MutSeriesF.new(result)
FixNan
type
Algorithm that replaces all math.nan values with the most recent corresponding non-nan values in a series. Read here about how to use it.
# indie:lang_version = 5
import math
from indie import indicator, MutSeriesF
from indie.algorithms import FixNan
@indicator('Example', overlay_main_pane=True)
def Main(self):
# Variable `s` is just an example of a series with `nan` values:
s = MutSeriesF.new(math.nan)
if self.close[0] > self.open[0]:
s[0] = self.close[0]
# `FixNan` replaces every `nan` value in `s` with the closest non-`nan` value on the left of it:
s2 = FixNan.new(s)
return s2[0]
# indie:lang_version = 5
from math import isnan
from indie import algorithm, SeriesF, MutSeriesF
@algorithm
def FixNan(self, src: SeriesF) -> SeriesF:
res = MutSeriesF.new(init=src[0])
if not isnan(src[0]):
res[0] = src[0]
return res
See also:
NanToZeroHighest
type
Algorithm that returns the maximum value over a given period. Read here about how to use it.
# indie:lang_version = 5
from indie import indicator
from indie.algorithms import Highest
@indicator('My Indie 1', overlay_main_pane=True)
def Main(self):
h = Highest.new(self.high, length=12)
return h[0]
# indie:lang_version = 5
from indie import algorithm, SeriesF, MutSeriesF
@algorithm
def Highest(self, src: SeriesF, length: int) -> SeriesF:
src.request_size(length)
result = src[0]
for i in range(1, length):
result = max(src[i], result)
return MutSeriesF.new(result)
See also:
SinceHighestLinReg
type
Linear regression algorithm. Read here about how to use it.
# indie:lang_version = 5
from indie import indicator
from indie.algorithms import LinReg
@indicator('Example')
def Main(self):
lr = LinReg.new(self.close, length=14, offset=0)
return lr[0]
# indie:lang_version = 5
from indie import algorithm, SeriesF, MutSeriesF
from indie.algorithms import Sma, StdDev, Corr
from indie.math import divide
@algorithm
def LinReg(self, y: SeriesF, length: int, offset: int = 0) -> SeriesF:
x = MutSeriesF.new(self.ctx.bar_index)
e_x = Sma.new(x, length)[0]
e_y = Sma.new(y, length)[0]
dev_x = StdDev.new(x, length)[0]
dev_y = StdDev.new(y, length)[0]
corr = Corr.new(x, y, length)[0]
slope = corr * divide(dev_y, dev_x)
inter = e_y - slope * e_x
reg = (x[0] - offset) * slope + inter
return MutSeriesF.new(reg)
Lowest
type
Algorithm that returns the minimum value over a given period. Read here about how to use it.
# indie:lang_version = 5
from indie import indicator
from indie.algorithms import Lowest
@indicator('My Indie 1', overlay_main_pane=True)
def Main(self):
l = Lowest.new(self.low, length=12)
return l[0]
# indie:lang_version = 5
from indie import algorithm, SeriesF, MutSeriesF
@algorithm
def Lowest(self, src: SeriesF, length: int) -> SeriesF:
src.request_size(length)
result = src[0]
for i in range(1, length):
result = min(src[i], result)
return MutSeriesF.new(result)
See also:
SinceLowestMa
type
Moving average algorithm. Read here about how to use it.
# indie:lang_version = 5
from indie.algorithms import Ma
@indicator('Example', overlay_main_pane=True)
def Main(self):
ma = Ma.new(self.close, length=12, algorithm='SMA')
return ma[0]
# indie:lang_version = 5
from indie import algorithm, SeriesF, Optional, IndieError
from indie.algorithms import Ema, Sma, Rma, Vwma, Wma
@algorithm
def Ma(self, src: SeriesF, length: int, algorithm: str) -> SeriesF:
'''Moving Average'''
result: Optional[SeriesF]
if algorithm == 'EMA':
result = Ema.new(src, length)
elif algorithm == 'SMA':
result = Sma.new(src, length)
elif algorithm == 'SMMA (RMA)' or algorithm == 'RMA':
result = Rma.new(src, length)
elif algorithm == 'VWMA':
result = Vwma.new(src, length)
elif algorithm == 'WMA':
result = Wma.new(src, length)
else:
raise IndieError("moving average algorithm should be one of 'EMA', 'SMA', "
"'SMMA (RMA)', 'RMA', 'VWMA' or 'WMA' but it is " + algorithm)
return result.value()
Macd
type
Moving average convergence/divergence algorithm. Read here about how to use it.
# indie:lang_version = 5
from indie.algorithms import Macd
@indicator('Example', overlay_main_pane=True)
def Main(self):
macd = Macd.new(self.close, fast_length=12, slow_length=26, sig_length=9, ma_source='EMA', ma_signal='EMA')
return macd[0]
# indie:lang_version = 5
from indie import algorithm, SeriesF, MutSeriesF
from indie.algorithms import Ma
@algorithm
def Macd(self, src: SeriesF, fast_len: int, slow_len: int, sig_len: int,
ma_source: str = 'EMA', ma_signal: str = 'EMA') -> tuple[SeriesF, SeriesF, SeriesF]:
'''Moving Average Convergence Divergence'''
fast_ma = Ma.new(src, fast_len, ma_source)
slow_ma = Ma.new(src, slow_len, ma_source)
macd = MutSeriesF.new(fast_ma[0] - slow_ma[0])
signal = Ma.new(macd, sig_len, ma_signal)
hist = MutSeriesF.new(macd[0] - signal[0])
return macd, signal, hist
Median
type
Moving median algorithm. Read here about how to use it.
# indie:lang_version = 5
from indie import indicator
from indie.algorithms import Median
@indicator('Example', overlay_main_pane=True)
def Main(self):
m = Median.new(self.close, length=10)
return m[0]
# indie:lang_version = 5
from math import nan
from sortedcontainers import SortedList
from indie import Algorithm, Context, IndieError, MutSeriesF, Optional, SeriesF
class Median(Algorithm):
def __init__(self, ctx: Context) -> None:
super().__init__(ctx)
self._src_sorted: Optional[SortedList] = None
self._last_added_value = nan
def _get_median(self) -> float:
src_sorted = self._src_sorted.value()
l = len(src_sorted)
if l % 2 == 0:
return (src_sorted[l // 2 - 1] + src_sorted[l // 2]) / 2
return src_sorted[(l - 1) // 2]
def calc(self, src: SeriesF, length: int) -> SeriesF:
if self._src_sorted is None:
if length == 0:
raise IndieError('cannot calculate median on list of zero size')
self._src_sorted = SortedList([nan] * length)
src_sorted = self._src_sorted.value()
val_to_remove = nan
if not self.ctx.is_new_bar: # intrabar update
val_to_remove = self._last_added_value
else:
val_to_remove = src[length]
src_sorted.remove(val_to_remove)
src_sorted.add(src[0])
self._last_added_value = src[0]
result = self._get_median()
return MutSeriesF.new(result)
Mfi
type
Money flow index algorithm. Read here about how to use it.
# indie:lang_version = 5
from indie.algorithms import Mfi
@indicator('Example', overlay_main_pane=True)
def Main(self):
mfi = Mfi.new(self.hlc3, length=14)
return mfi[0]
# indie:lang_version = 5
from indie import algorithm, MutSeriesF, SeriesF
from indie.math import divide
@algorithm
def Mfv(self) -> SeriesF:
'''Money Flow Volume'''
h = self.ctx.high[0]
l = self.ctx.low[0]
c = self.ctx.close[0]
v = self.ctx.volume[0]
res = v * divide((c - l) - (h - c), (h - l), 0.0)
return MutSeriesF.new(res)
Mfv
type
Money flow volume algorithm. Read here about how to use it.
# indie:lang_version = 5
from indie import indicator
from indie.algorithms import Mfv
@indicator('Example')
def Main(self):
mfv = Mfv.new()
return mfv[0]
# indie:lang_version = 5
from indie import algorithm, MutSeriesF, SeriesF
from indie.math import divide
@algorithm
def Mfv(self) -> SeriesF:
'''Money Flow Volume'''
h = self.ctx.high[0]
l = self.ctx.low[0]
c = self.ctx.close[0]
v = self.ctx.volume[0]
res = v * divide((c - l) - (h - c), (h - l), 0.0)
return MutSeriesF.new(res)
NanToZero
type
Algorithm that replaces all math.nan values with zeros in a series. Read here about how to use it.
# indie:lang_version = 5
import math
from indie import indicator, MutSeriesF
from indie.algorithms import NanToZero
@indicator('Example')
def Main(self):
# Variable `s` is just an example of a series with `nan` values:
s = MutSeriesF.new(math.nan)
if self.close[0] > self.open[0]:
s[0] = 0 if math.isnan(s[1]) else s[1] + 1
# `NanToZero` replaces every `nan` value in `s` with zero
s2 = NanToZero.new(s)
return s2[0]
# indie:lang_version = 5
from math import isnan
from indie import algorithm, SeriesF, MutSeriesF
@algorithm
def NanToZero(self, src: SeriesF) -> SeriesF:
res = 0 if isnan(src[0]) else src[0]
return MutSeriesF.new(res)
See also:
FixNanNetVolume
type
Net volume algorithm. Read here about how to use it.
# indie:lang_version = 5
from indie import indicator
from indie.algorithms import NetVolume
@indicator('Example')
def Main(self):
nv = NetVolume.new(self.close)
return nv[0]
# indie:lang_version = 5
from math import nan
from indie import algorithm, SeriesF, MutSeriesF
from indie.algorithms import Change
@algorithm
def NetVolume(self, src: SeriesF) -> SeriesF:
der = Change.new(src)[0]
nv = MutSeriesF.new(init=nan)
if der > 0:
nv[0] = self.ctx.volume[0]
elif der < 0:
nv[0] = -self.ctx.volume[0]
elif der == 0:
nv[0] = 0
return nv
PercentRank
type
Percent rank algorithm. Read here about how to use it.
# indie:lang_version = 5
from indie import indicator
from indie.algorithms import PercentRank
@indicator('Example')
def Main(self):
pr = PercentRank.new(self.close, length=12)
return pr[0]
# indie:lang_version = 5
from indie import algorithm, SeriesF, MutSeriesF
@algorithm
def PercentRank(self, src: SeriesF, length: int) -> SeriesF:
num_leq = 0
for idx in range(1, length + 1):
if src[idx] <= src[0]:
num_leq += 1
return MutSeriesF.new(100.0 * num_leq / length)
Percentile
type
Percentile algorithm. Read here about how to use it.
# indie:lang_version = 5
from indie import indicator
from indie.algorithms import Percentile
@indicator('Example')
def Main(self):
pcl = Percentile.new(self.close, length=12, pct=95, interpolate=True)
return pcl[0]
# indie:lang_version = 5
from math import ceil, nan
from sortedcontainers import SortedList
from indie import Algorithm, Context, IndieError, MutSeriesF, Optional, SeriesF
class Percentile(Algorithm):
def __init__(self, ctx: Context) -> None:
super().__init__(ctx)
self._src_sorted: Optional[SortedList] = None
self._last_added_value = nan
def _get_percentile(self, pct: float, interpolate: bool) -> float:
if pct < 0 or pct > 100:
raise IndieError('percentile must be within [0..100] range')
src_sorted = self._src_sorted.value()
l = len(src_sorted)
x = (l - 1) * pct / 100
i = ceil(x)
if x == i:
return src_sorted[i]
if interpolate:
return src_sorted[i] - (src_sorted[i] - src_sorted[i - 1]) * (i - x)
return src_sorted[i]
def calc(self, src: SeriesF, length: int, pct: float, interpolate: bool) -> SeriesF:
if self._src_sorted is None:
if length == 0:
raise IndieError('cannot calculate median on list of zero size')
self._src_sorted = SortedList([nan] * length)
src_sorted = self._src_sorted.value()
val_to_remove = nan
if not self.ctx.is_new_bar: # intrabar update
val_to_remove = self._last_added_value
else:
val_to_remove = src[length]
src_sorted.remove(val_to_remove)
src_sorted.add(src[0])
self._last_added_value = src[0]
result = self._get_percentile(pct, interpolate)
return MutSeriesF.new(result)
PivotHighLow
type
PivotHighLow algorithm. Read here about how to use it.
# indie:lang_version = 5
from indie import indicator
from indie.algorithms import PivotHighLow
@indicator('Example')
def Main(self):
ph, pl = PivotHighLow.new(self.close, left_bars=5, right_bars=5)
return ph[0], pl[0]
# indie:lang_version = 5
from math import inf, nan
from sortedcontainers import SortedList
from indie import Algorithm, Context, IndieError, MutSeriesF, Optional, SeriesF, algorithm
class HighestLowest(Algorithm):
def __init__(self, ctx: Context) -> None:
super().__init__(ctx)
self._src_sorted: Optional[SortedList] = None
self._last_added_value = nan
def calc(self, src: SeriesF, length: int) -> tuple[SeriesF, SeriesF]:
if length == 0:
return MutSeriesF.new(inf), MutSeriesF.new(-inf)
if self._src_sorted is None:
self._src_sorted = SortedList([nan] * length)
src_sorted = self._src_sorted.value()
val_to_remove = nan
if not self.ctx.is_new_bar: # intrabar update
val_to_remove = self._last_added_value
else:
val_to_remove = src[length]
src_sorted.remove(val_to_remove)
src_sorted.add(src[0])
self._last_added_value = src[0]
return MutSeriesF.new(src_sorted[-1]), MutSeriesF.new(src_sorted[0])
@algorithm
def PivotHighLow(self, src: SeriesF, left_bars: int, right_bars: int) -> tuple[SeriesF, SeriesF]:
if left_bars < 0:
raise IndieError('left_bars cannot be negative')
if right_bars < 0:
raise IndieError('right_bars cannot be negative')
right_max, right_min = HighestLowest.new(src, right_bars)
left_max, left_min = HighestLowest.new(src, left_bars)
src_val = src[right_bars]
is_high_pivot = src_val > left_max[right_bars + 1] and src_val >= right_max[0]
is_low_pivot = src_val < left_min[right_bars + 1] and src_val <= right_min[0]
return MutSeriesF.new(src_val if is_high_pivot else nan), MutSeriesF.new(src_val if is_low_pivot else nan)
Rma
type
RMA algorithm is a moving average algorithm that is used to calculate RSI. Read here about how to use it.
# indie:lang_version = 5
from indie import indicator
from indie.algorithms import Rma
@indicator('My Indie 1', overlay_main_pane=True)
def Main(self):
rma = Rma.new(self.close, length=12)
return rma[0]
# indie:lang_version = 5
from math import isnan
from indie import algorithm, SeriesF, MutSeriesF
from indie.algorithms import Sma
@algorithm
def Rma(self, src: SeriesF, length: int) -> SeriesF:
'''RSI Moving Average'''
result = MutSeriesF.new(init=0)
if isnan(result[1]):
result[0] = Sma.new(src, length)[0]
else:
result[0] = (src[0] + (length - 1) * result[1]) / length
return result
Roc
type
Rate of change algorithm. Read here about how to use it.
# indie:lang_version = 5
from indie import indicator
from indie.algorithms import Roc
@indicator('Example')
def Main(self):
roc = Roc.new(self.close, length=9)
return roc[0]
# indie:lang_version = 5
from indie import algorithm, SeriesF, MutSeriesF
from indie.algorithms import Change
from indie.math import divide
@algorithm
def Roc(self, src: SeriesF, length: int) -> SeriesF:
'''Rate Of Change'''
res = 100 * divide(Change.new(src, length)[0], src[length])
return MutSeriesF.new(res)
Rsi
type
Relative strength index algorithm. Read here about how to use it.
# indie:lang_version = 5
from indie import indicator
from indie.algorithms import Rsi
@indicator('Example')
def Main(self):
rsi = Rsi.new(self.close, length=12)
return rsi[0]
# indie:lang_version = 5
from indie import algorithm, SeriesF, MutSeriesF
from indie.algorithms import Change, Rma
from indie.math import divide
@algorithm
def Rsi(self, src: SeriesF, length: int) -> SeriesF:
'''Relative Strength Index'''
u = max(Change.new(src)[0], 0) # upward change
rma_u = Rma.new(MutSeriesF.new(u), length)[0]
d = max(-Change.new(src)[0], 0) # downward change
rma_d = Rma.new(MutSeriesF.new(d), length)[0]
res = 100 * divide(rma_u, rma_u + rma_d, 1.0)
return MutSeriesF.new(res)
Sar
type
Parabolic stop and reverse algorithm. Read here about how to use it.
# indie:lang_version = 5
from indie.algorithms import Sar
@indicator('Example', overlay_main_pane=True)
def Main(self):
sar = Sar.new(start=0.02, increment=0.02, maximum=0.2)
return sar[0]
# indie:lang_version = 5
from math import nan
from indie import algorithm, SeriesF, Var, MutSeriesF
@algorithm
def Sar(self, start: float, increment: float, maximum: float) -> SeriesF:
result = Var[float].new(init=nan)
max_min = Var[float].new(init=nan)
acceleration = Var[float].new(init=nan)
is_below = Var[bool].new(init=False)
is_first_trend_bar = False
if self.ctx.bar_index == 1:
if self.ctx.close[0] > self.ctx.close[1]:
is_below.set(True)
max_min.set(self.ctx.high[0])
result.set(self.ctx.low[1])
else:
is_below.set(False)
max_min.set(self.ctx.low[0])
result.set(self.ctx.high[1])
is_first_trend_bar = True
acceleration.set(start)
result.set(result.get() + acceleration.get() * (max_min.get() - result.get()))
if is_below.get():
if result.get() > self.ctx.low[0]:
is_first_trend_bar = True
is_below.set(False)
result.set(max(self.ctx.high[0], max_min.get()))
max_min.set(self.ctx.low[0])
acceleration.set(start)
else:
if result.get() < self.ctx.high[0]:
is_first_trend_bar = True
is_below.set(True)
result.set(min(self.ctx.low[0], max_min.get()))
max_min.set(self.ctx.high[0])
acceleration.set(start)
if not is_first_trend_bar:
if is_below.get():
if self.ctx.high[0] > max_min.get():
max_min.set(self.ctx.high[0])
acceleration.set(min(acceleration.get() + increment, maximum))
else:
if self.ctx.low[0] < max_min.get():
max_min.set(self.ctx.low[0])
acceleration.set(min(acceleration.get() + increment, maximum))
if is_below.get():
result.set(min(result.get(), self.ctx.low[1]))
if self.ctx.bar_index > 1:
result.set(min(result.get(), self.ctx.low[2]))
else:
result.set(max(result.get(), self.ctx.high[1]))
if self.ctx.bar_index > 1:
result.set(max(result.get(), self.ctx.high[2]))
return MutSeriesF.new(result.get())
SinceHighest
type
Algorithm that returns the number of bars after the maximum price for the given period. Read here about how to use it.
# indie:lang_version = 5
from indie import indicator
from indie.algorithms import SinceHighest
@indicator('Example')
def Main(self):
sh = SinceHighest.new(self.high, length=10)
return sh[0]
# indie:lang_version = 5
from indie import algorithm, SeriesF, Series, MutSeries
@algorithm
def SinceHighest(self, src: SeriesF, length: int) -> Series[int]:
offset = 0
for idx in range(1, length):
if src[idx] >= src[offset]:
offset = idx
return MutSeries[int].new(offset)
See also:
HighestSinceLowest
type
Algorithm that returns the number of bars after the minimum price for the given period. Read here about how to use it.
# indie:lang_version = 5
from indie import indicator
from indie.algorithms import SinceLowest
@indicator('Example')
def Main(self):
sl = SinceLowest.new(self.high, length=10)
return sl[0]
# indie:lang_version = 5
from indie import algorithm, SeriesF, Series, MutSeries
@algorithm
def SinceLowest(self, src: SeriesF, length: int) -> Series[int]:
offset = 0
for idx in range(1, length):
if src[idx] <= src[offset]:
offset = idx
return MutSeries[int].new(offset)
See also:
LowestSinceTrue
type
Read here about how to use it.
# indie:lang_version = 5
from indie import indicator, MutSeries
from indie.algorithms import SinceTrue
@indicator('Example')
def Main(self):
cond = MutSeries[bool].new(self.close[0] > self.open[0])
index = SinceTrue.new(cond)
return index
# indie:lang_version = 5
from indie import algorithm, Series, MutSeries
@algorithm
def SinceTrue(self, condition: Series[bool]) -> Series[int]:
result = MutSeries[int].new(init=-1)
if condition[0]:
result[0] = 0
elif result[0] != -1:
result[0] += 1
return result
Sma
type
Simple moving average algorithm. Read here about how to use it.
# indie:lang_version = 5
from indie import indicator
from indie.algorithms import Sma
@indicator('Example')
def Main(self):
sma = Sma.new(self.close, length=12)
return sma[0]
# indie:lang_version = 5
from indie import algorithm, SeriesF, MutSeriesF
from indie.algorithms import Sum
@algorithm
def Sma(self, src: SeriesF, length: int) -> SeriesF:
'''Simple Moving Average'''
s = Sum.new(src, length)
return MutSeriesF.new(s[0] / length)
StdDev
type
Standard deviation algorithm. Read here about how to use it.
# indie:lang_version = 5
from indie import indicator
from indie.algorithms import StdDev
@indicator('Example')
def Main(self):
sd = StdDev.new(self.close, length=12)
return sd[0]
# indie:lang_version = 5
from math import sqrt
from indie import algorithm, SeriesF, MutSeriesF
from indie.algorithms import Sma
@algorithm
def StdDev(self, src: SeriesF, length: int) -> SeriesF:
# NOTE: `var_` could be calculated as `E(src^2) - E(src)^2`,
# but that may cause precision issues with floating-point arithmetic
avg = Sma.new(src, length)[0]
var_ = 0.0
for i in range(length):
dev = src[i] - avg
var_ += dev * dev
var_ /= length
return MutSeriesF.new(sqrt(var_))
Stoch
type
Stochastic algorithm. Read here about how to use it.
# indie:lang_version = 5
from indie import indicator
from indie.algorithms import Stoch
@indicator('Example')
def Main(self):
s = Stoch.new(self.close, self.low, self.high, length=14)
return s[0]
# indie:lang_version = 5
from indie import algorithm, SeriesF, MutSeriesF
from indie.algorithms import Lowest, Highest
from indie.math import divide
@algorithm
def Stoch(self, src: SeriesF, low: SeriesF, high: SeriesF, length: int) -> SeriesF:
lowest_low = Lowest.new(low, length)[0]
highest_high = Highest.new(high, length)[0]
res = 100 * divide(src[0] - lowest_low, highest_high - lowest_low, 1.0)
return MutSeriesF.new(res)
Sum
type
Sliding sum algorithm. Read here about how to use it.
# indie:lang_version = 5
from indie import indicator
from indie.algorithms import Sum
@indicator('Example')
def Main(self):
s = Sum.new(self.close, length=10)
return s[0]
# indie:lang_version = 5
from math import isnan, nan
from indie import algorithm, SeriesF, MutSeriesF, Var
def nan_to_zero(val: float) -> float:
return 0 if isnan(val) else val
@algorithm
def Sum(self, src: SeriesF, length: int) -> SeriesF:
src.request_size(length + 1)
nan_count = Var[int].new(0)
sum = Var[float].new(0)
sum.set(sum.get() + nan_to_zero(src[0]))
result = 0.0
if isnan(src[0]):
nan_count.set(nan_count.get() + 1)
result = nan
else:
if len(src) - nan_count.get() > length:
sum.set(sum.get() - src[length])
if len(src) - nan_count.get() < length:
result = nan
else:
result = sum.get()
return MutSeriesF.new(result)
Supertrend
type
Supertrend algorithm. Read here about how to use it.
# indie:lang_version = 5
from indie import indicator, color, plot
from indie.algorithms import Supertrend
@indicator('Example', overlay_main_pane=True)
def Main(self):
st, dir = Supertrend.new(factor=3.0, atr_period=10, ma_algorithm='SMA')
c = color.RED if dir[0] > 0 else color.GREEN
return plot.Line(st[0], color=c)
# indie:lang_version = 5
from math import isnan
from indie import algorithm, MutSeriesF, SeriesF
from indie.algorithms import Atr
def nan_to_val(maybe_nan: float, val: float) -> float:
return val if isnan(maybe_nan) else maybe_nan
@algorithm
def Supertrend(self, factor: float, atr_period: int, ma_algorithm: str) -> tuple[SeriesF, SeriesF]:
src = self.ctx.hl2
atr = Atr.new(atr_period, ma_algorithm)
upper_band = MutSeriesF.new(src[0] + factor * atr[0])
lower_band = MutSeriesF.new(src[0] - factor * atr[0])
prev_lower_band = nan_to_val(lower_band[1], 0)
prev_upper_band = nan_to_val(upper_band[1], 0)
lower_band[0] = lower_band[0] \
if lower_band[0] > prev_lower_band or self.ctx.close[1] < prev_lower_band \
else prev_lower_band
upper_band[0] = upper_band[0] \
if upper_band[0] < prev_upper_band or self.ctx.close[1] > prev_upper_band \
else prev_upper_band
direction = 0.0 # default value
super_trend = MutSeriesF.new(init=0)
prev_super_trend = super_trend[1]
if isnan(atr[1]):
direction = 1.0
elif prev_super_trend == prev_upper_band:
direction = -1.0 if self.ctx.close[0] > upper_band[0] else 1.0
else:
direction = 1.0 if self.ctx.close[0] < lower_band[0] else -1.0
super_trend[0] = lower_band[0] if direction < 0 else upper_band[0]
return super_trend, MutSeriesF.new(direction)
Tr
type
True range algorithm. Read here about how to use it.
# indie:lang_version = 5
from indie import indicator
from indie.algorithms import Tr
@indicator('Example')
def Main(self):
tr = Tr.new()
return tr[0]
# indie:lang_version = 5
from math import nan, isnan
from indie import algorithm, SeriesF, MutSeriesF
@algorithm
def Tr(self, handle_na: bool = False) -> SeriesF:
'''True Range'''
res = nan if handle_na else self.ctx.high[0] - self.ctx.low[0]
if not isnan(self.ctx.close[1]):
res = max(
self.ctx.high[0] - self.ctx.low[0],
abs(self.ctx.high[0] - self.ctx.close[1]),
abs(self.ctx.low[0] - self.ctx.close[1]),
)
return MutSeriesF.new(res)
Tsi
type
True strength index algorithm. Read here about how to use it.
# indie:lang_version = 5
from indie import indicator
from indie.algorithms import Tsi
@indicator('Example')
def Main(self):
tsi = Tsi.new(self.close, long_len=21, short_len=12)
return tsi[0]
# indie:lang_version = 5
from indie import algorithm, SeriesF, MutSeriesF
from indie.algorithms import Ema, Change
from indie.math import divide
@algorithm
def DoubleSmooth(self, src: SeriesF, long_len: int, short_len: int) -> SeriesF:
return Ema.new(Ema.new(src, long_len), short_len)
@algorithm
def Tsi(self, src: SeriesF, long_len: int, short_len: int) -> SeriesF:
'''True Strength Index'''
pc = Change.new(src)
double_smoothed_pc = DoubleSmooth.new(pc, long_len, short_len)[0]
abs_pc = MutSeriesF.new(abs(pc[0]))
double_smoothed_abs_pc = DoubleSmooth.new(abs_pc, long_len, short_len)[0]
res = divide(double_smoothed_pc, double_smoothed_abs_pc)
return MutSeriesF.new(res)
Uo
type
Ultimate oscillator algorithm. Read here about how to use it.
# indie:lang_version = 5
from indie.algorithms import Uo
@indicator('Example', overlay_main_pane=True)
def Main(self):
uo = Uo.new(fast_len=7, middle_len=14, slow_len=28)
return uo[0]
# indie:lang_version = 5
from math import nan
from indie import algorithm, SeriesF, Var, MutSeriesF
@algorithm
def Sar(self, start: float, increment: float, maximum: float) -> SeriesF:
result = Var[float].new(init=nan)
max_min = Var[float].new(init=nan)
acceleration = Var[float].new(init=nan)
is_below = Var[bool].new(init=False)
is_first_trend_bar = False
if self.ctx.bar_index == 1:
if self.ctx.close[0] > self.ctx.close[1]:
is_below.set(True)
max_min.set(self.ctx.high[0])
result.set(self.ctx.low[1])
else:
is_below.set(False)
max_min.set(self.ctx.low[0])
result.set(self.ctx.high[1])
is_first_trend_bar = True
acceleration.set(start)
result.set(result.get() + acceleration.get() * (max_min.get() - result.get()))
if is_below.get():
if result.get() > self.ctx.low[0]:
is_first_trend_bar = True
is_below.set(False)
result.set(max(self.ctx.high[0], max_min.get()))
max_min.set(self.ctx.low[0])
acceleration.set(start)
else:
if result.get() < self.ctx.high[0]:
is_first_trend_bar = True
is_below.set(True)
result.set(min(self.ctx.low[0], max_min.get()))
max_min.set(self.ctx.high[0])
acceleration.set(start)
if not is_first_trend_bar:
if is_below.get():
if self.ctx.high[0] > max_min.get():
max_min.set(self.ctx.high[0])
acceleration.set(min(acceleration.get() + increment, maximum))
else:
if self.ctx.low[0] < max_min.get():
max_min.set(self.ctx.low[0])
acceleration.set(min(acceleration.get() + increment, maximum))
if is_below.get():
result.set(min(result.get(), self.ctx.low[1]))
if self.ctx.bar_index > 1:
result.set(min(result.get(), self.ctx.low[2]))
else:
result.set(max(result.get(), self.ctx.high[1]))
if self.ctx.bar_index > 1:
result.set(max(result.get(), self.ctx.high[2]))
return MutSeriesF.new(result.get())
Vwap
type
Volume weighted average price algorithm. Read here about how to use it.
# indie:lang_version = 5
from indie import indicator
from indie.algorithms import Vwap
@indicator('Example', overlay_main_pane=True)
def Main(self):
main_line, upper, lower = Vwap.new(self.close, anchor='Day', std_dev_mult=1.0)
return main_line[0], upper[0], lower[0]
# indie:lang_version = 5
from math import sqrt
from datetime import datetime
from indie import algorithm, SeriesF, MutSeriesF, Var
@algorithm
def Vwap(self, src: SeriesF, anchor: str, std_dev_mult: float) -> tuple[SeriesF, SeriesF, SeriesF]:
'''
Volume Weighted Average Price
anchor can be 'Session', 'Week', 'Month', 'Year'
'''
vwap_sum = Var[float].new(0)
volume_sum = Var[float].new(0)
vwap_squares_sum = Var[float].new(0)
current_datetime = datetime.utcfromtimestamp(self.ctx.time[0])
prev_datetime = datetime.utcfromtimestamp(self.ctx.time.get(1, 0))
need_reset = False
if anchor == 'Session' and \
not self.ctx.trading_session.is_same_period(self.ctx.time[0], self.ctx.time.get(1, 0)):
need_reset = True
elif anchor == 'Week' and ((current_datetime.weekday() == 0 and prev_datetime.weekday() != 0) or
(current_datetime-prev_datetime).days >= 7):
need_reset = True
elif anchor == 'Month' and (current_datetime.month != prev_datetime.month or
(current_datetime-prev_datetime).days >= 31):
need_reset = True
elif anchor == 'Year' and current_datetime.year != prev_datetime.year:
need_reset = True
if need_reset:
vwap_sum.set(0)
volume_sum.set(0)
vwap_squares_sum.set(0)
vwap_sum.set(vwap_sum.get() + src[0] * self.ctx.volume[0])
vwap_squares_sum.set(vwap_squares_sum.get() + src[0] * src[0] * self.ctx.volume[0])
volume_sum.set(volume_sum.get() + self.ctx.volume[0])
vwap_avg = vwap_sum.get() / volume_sum.get()
vwap_std_dev = sqrt(max(vwap_squares_sum.get() / volume_sum.get() - vwap_avg * vwap_avg, 0))
std_dev = std_dev_mult * vwap_std_dev
lower = MutSeriesF.new(vwap_avg - std_dev)
upper = MutSeriesF.new(vwap_avg + std_dev)
return MutSeriesF.new(vwap_avg), upper, lower
Vwma
type
Volume weighted moving average algorithm. Read here about how to use it.
# indie:lang_version = 5
from indie import indicator
from indie.algorithms import Vwma
@indicator('Example', overlay_main_pane=True)
def Main(self):
vwma = Vwma.new(self.close, length=12)
return vwma[0]
# indie:lang_version = 5
from indie import algorithm, SeriesF, MutSeriesF
from indie.algorithms import Sma
from indie.math import divide
@algorithm
def Vwma(self, src: SeriesF, length: int) -> SeriesF:
'''Volume Weighted Moving Average'''
src_vol = MutSeriesF.new(src[0] * self.ctx.volume[0])
sma_src_vol = Sma.new(src_vol, length)
sma_vol = Sma.new(self.ctx.volume, length)
res = divide(sma_src_vol[0], sma_vol[0])
return MutSeriesF.new(res)
Wma
type
Weighted moving average algorithm. Read here about how to use it.
# indie:lang_version = 5
from indie import indicator
from indie.algorithms import Wma
@indicator('Example', overlay_main_pane=True)
def Main(self):
wma = Wma.new(self.close, length=12)
return wma[0]
# indie:lang_version = 5
from indie import algorithm, SeriesF, MutSeriesF
@algorithm
def Wma(self, src: SeriesF, length: int) -> SeriesF:
'''Weighted Moving Average'''
norm = 0.0
s = 0.0
for i in range(length):
weight = length - i
norm += weight
s += src[i] * weight
return MutSeriesF.new(s / norm)
ZigZag
type
ZigZag algorithm that identifies pivot high and low points based on price movements. Read here about how to use it.
# indie:lang_version = 5
from indie import indicator
from indie.algorithms import ZigZag
@indicator('Example', overlay_main_pane=True)
def Main(self):
new_high, upd_high, new_low, upd_low = ZigZag.new(
left_bars=5,
right_bars=5,
dev_threshold=5.0,
allow_zig_zag_within_one_bar=True
)
return 1.0 if new_high or new_low else 0.0
# indie:lang_version = 5
from math import isnan
from indie import Optional, Context, Var, Algorithm
from indie.algorithms import PivotHighLow
class ZigZag(Algorithm):
def __init__(self, ctx: Context):
super().__init__(ctx)
def calc(self, left_bars: int, right_bars: int, dev_threshold: float,
allow_zig_zag_on_one_bar: bool = True) -> tuple[bool, bool, bool, bool]:
ph, _ = PivotHighLow.new(self.ctx.high, left_bars, right_bars)
_, pl = PivotHighLow.new(self.ctx.low, left_bars, right_bars)
enough_bars = self.ctx.bar_count >= left_bars + 1 + right_bars # TODO: make this check configurable
new_high, upd_high = False, False
new_low, upd_low = False, False
if enough_bars and not isnan(ph[0]):
new_high, upd_high = self._check_pivot_point(ph[0], True, dev_threshold)
zigzag_updated = new_high or upd_high
check_low_pivot = not zigzag_updated or allow_zig_zag_on_one_bar
if enough_bars and not isnan(pl[0]) and check_low_pivot:
new_low, upd_low = self._check_pivot_point(pl[0], False, dev_threshold)
# Note: pivot information appears with a delay of `right_bars` bars, meaning
# if a boolean value is `True`, the event actually occurred `right_bars` bars ago.
# If `allow_zig_zag_on_one_bar` is `True`, then `new_high` and `new_low` can be `True` at the same time.
return new_high, upd_high, new_low, upd_low
def _check_pivot_point(self, price: float, is_high: bool,
dev_threshold: float) -> tuple[bool, bool]:
last_pivot = Var[Optional[float]].new(None)
last_pivot_side = Var[bool].new(False)
if last_pivot.get() is None:
last_pivot.set(price)
last_pivot_side.set(is_high)
return True, False
last_pivot_price = last_pivot.get().value()
if last_pivot_side.get() == is_high:
if self._has_better_price(last_pivot_price, price, is_high):
last_pivot.set(price)
return False, True
else:
if self._is_not_near(last_pivot_price, price, is_high, dev_threshold):
last_pivot.set(price)
last_pivot_side.set(is_high)
return True, False
return False, False
def _has_better_price(self, old_pivot: float, new_pivot: float, is_high: bool) -> bool:
if is_high:
return new_pivot > old_pivot
return new_pivot < old_pivot
def _is_not_near(self, old_pivot: float, new_pivot: float, is_high: bool,
dev_threshold: float) -> bool:
if old_pivot == 0:
return True
dev = 100 * (new_pivot - old_pivot) / abs(old_pivot)
if is_high:
return dev >= dev_threshold
return dev <= -dev_threshold
See also:
PivotHighLow