​
SMA algorithm that accepts ‘series’ length

Indie’s standard indie.algorithms.Sma algorithm has a length parameter which cannot be dynamically changed. Its common usage looks like this: Sma.new(self.close, length=12), where the length parameter is typically a literal constant. While it is possible for the length to be a variable calculated at the time of indicator initialization or even during calculation over bars/candles, the general rule that should not be broken is as follows: once set, the value of the length parameter should not change. Otherwise, the algorithm breaks internally and starts giving inaccurate results*.

Sometimes, greater flexibility is needed. Fortunately, there is a way to achieve this, and this indicator with MySma algorithm demonstrates it. The algorithm is based on properties derived from prefix sums. Let’s look at an example to see how it works. Suppose we have a source series of values, src, like this:

src = …, 2, 7, 5, 4, 3, 9, 6

Think of this series as a series of close prices of some instrument on a chart. The last (i.e., current) value of this series is 6, which is accessed in Indie with the expression src[0].

To calculate the SMA with a length period, we simply divide the sum of length elements by the length value. The sum calculation seems straightforward, but it can be inefficient. For example, sum(src, length=3) is calculated as 3 + 9 + 6 = 18. This calculation is CPU-intensive, and the larger the length value, the more computationally expensive it becomes (the time complexity of this calculation is linear). This isn’t a big issue if the length is constant. However, consider a use case where length may vary from 2 to 300, and the size of the src series is around 20,000 bars, requiring SMA (and thus sum) calculations at every bar of this dataset. That would involve a lot of sum operations.

If we use prefix sums, which in Indie can be calculated with a cs = CumSum.new(src, length) statement, we can calculate sums in constant time (which is faster):

cs = …, 2, 9, 14, 18, 21, 30, 36

Calculation of any sum with any length using the cs series is as simple as subtracting two numbers. For example, sum(src, length=3) = cs[0] - cs[3] = 36 - 18 = 18. Let’s calculate sums with a few different lengths:

  • sum(src, length=4) = cs[0] - cs[4] = 36 - 14 = 22, which is the same as 4 + 3 + 9 + 6 = 22
  • sum(src, length=5) = cs[0] - cs[5] = 36 - 9 = 27, which is the same as 5 + 4 + 3 + 9 + 6 = 27

and so on. The general formula here is:

sum(src, length) = cs[0] - cs[length]

In this indicator, there are three SMA plot lines:

  • The thin green and thin red lines both use the indie.algorithm.Sma from the standard library. They use different but constant length parameters: short_len and long_len. These two plots serve as references with which we compare the third plot.
  • The thick light blue line uses the MySma algorithm described above, which is based on the prefix sums approach. MySma can accept a ‘series’ length, and depending on runtime conditions, either short_len or long_len is passed to it.

We see that the blue line is perfectly aligned with one of the two fixed-length SMA plots, proving its correctness.

This approach can be effectively used not only in the SMA algorithm but in any similar algorithm that can be calculated with the help of prefix sums or derivatives of SMA (like double-SMA and so on).

P. S. This is true as of Indie v4.

SMA algorithm that accepts 'series' length
# indie:lang_version = 4
from indie import (
    indicator, MutSeriesF, algorithm, SeriesF, plot, 
    color, param, line_style,
)
from indie.algorithms import CumSum, Sma


@algorithm
def MySum(self, src: SeriesF, length: int) -> SeriesF:
    cs = CumSum.new(src)
    return MutSeriesF.new(cs[0] - cs[length])


@algorithm
def MySma(self, src: SeriesF, length: int) -> SeriesF:
    s = MySum.new(src, length)
    return MutSeriesF.new(s[0] / length)


@indicator("Sma with 'series' length", overlay_main_pane=True)
@param.int('short_len', default=12)
@param.int('long_len', default=24)
@plot(color=color.BLUE(alpha=0.65), line_width=7)
@plot(color=color.RED)
@plot(color=color.GREEN)
def Main(self, short_len, long_len):
    # The long and short `Sma`s are calculated here with the algorithm 
    # from the `indie.algorithms` standard library, they accept only 
    # non-series lengths and they are plotted to be a reference that 
    # proves that MySma gives correct results.
    short_sma = Sma.new(self.high, short_len)
    long_sma = Sma.new(self.high, long_len)
    
    length = short_len
    if long_sma[0] >= short_sma[0]:
        length = long_len
    # So, strictly speaking, `length` is not a series, but just a number.
    # But it is kinda series, because it may have different values on different bars.
    # NOTE: If you need a truly series length, wrap it with `MutSeriesF.new(length)`.
    
    return (
        MySma.new(self.high, length)[0], # Do not use indie.algorithms.Sma here, 
                                         # because `length` is not constant over 
                                         # different bars. You may try and see how 
                                         # indicator starts giving bad results.
        short_sma[0],
        long_sma[0], 
    )

​
Green/Red Bar Counts - usage example of indie.MutSeries[T] generic class

Indie includes generic series types: indie.Series[T] and indie.MutSeries[T]. The float-types indie.SeriesF and indie.MutSeriesF are still available as aliases for indie.Series[float] and indie.MutSeries[float], respectively.

The “Green/Red Bar Counts” indicator demonstrates the usage of the new generic class indie.MutSeries[T], which is instantiated with T as the int type. The generic type parameter T can be any (well, almost any) Indie type, such as bool, str, etc.

Green/Red Bar Counts - usage example of indie.MutSeries[T] generic class
# indie:lang_version = 4
import math
from indie import indicator, MutSeries, Context, plot, color, plot_style

def is_green_bar(ctx: Context) -> bool:
    return ctx.close[0] >= ctx.open[0]

@indicator('Green/Red Bar Counts')
@plot(color=color.GREEN, style=plot_style.STEPS)
@plot(color=color.RED, style=plot_style.STEPS)
def Main(self):
    # `indie.MutSeries[T]` is a generic class, 
    # `T` can be `float`, `int`, `bool`, `str`, etc.
    green_count = MutSeries[int].new(0)
    red_count = MutSeries[int].new(0)
    if is_green_bar(self):
        green_count[0] = green_count[1] + 1
        red_count[0] = 0
    else:
        green_count[0] = 0
        red_count[0] = red_count[1] - 1

    return green_count[0], red_count[0]

​
Sma-Ema Crossover - usage example of indie.Var[T]

The indie.Var[T] container functions similarly to indie.MutSeries[T], but it has a single-value history depth, making it ideal for storing and updating the current state without accumulating historical values. Although indie.Var[T] does not store historical values, it differs significantly from a primitive field of type T of an Algorithm/Context or a global variable. The key difference lies in its behavior during real-time calculations. indie.Var[T] retains only the values set on closing bar updates, reverting to the previous value before each intrabar update.

The “SMA-EMA Crossover” indicator demonstrates the use of indie.Var[T] through the algorithm CrossoverWithVar, which detects and retains values on specific crossover events. It also includes the algorithm CrossoverWithSeries to illustrate how to achieve the same result without using indie.Var[T].

Sma-Ema Crossover - usage example of indie.Var[T]
# Copyright (c) 2024 @TakeProfit. All rights reserved.

# This work is licensed under the MIT License.
# For a copy, see <https://opensource.org/licenses/MIT>.

# indie:lang_version = 4
from indie import indicator, Var, plot, MutSeriesF, SeriesF, color, algorithm
from indie.algorithms import Sma, Ema

@algorithm
def CrossoverWithVar(self, s1: SeriesF, s2: SeriesF) -> float:
    my_var = Var[float].new(0)
    if s1[1] > s2[1] and s1[0] < s2[0]:
        my_var.set(s1[0])
    return my_var.get()

# Note: this function does the same thing but using MutSeriesF
@algorithm
def CrossoverWithSeries(self, s1: SeriesF, s2: SeriesF) -> float:
    my_ser = MutSeriesF.new(init=0)
    if s1[1] > s2[1] and s1[0] < s2[0]:
        my_ser[0] = s1[0]
    return my_ser[0]

@indicator('Sma-Ema Crossover', overlay_main_pane=True)
@plot(color=color.MAROON)
@plot(color=color.LIME)
@plot(color=color.BLUE, line_width=2)
def Main(self):
    sma = Sma.new(self.close, 5)
    ema = Ema.new(self.close, 5)
    return sma[0], ema[0], CrossoverWithVar.new(sma, ema)

​
Green/Red Bar Count - usage example of indie.Var[T]

This update of “Green/Red Bar Counts” indicator showcases the indie.Var[T] container usage.

In this implementation:

  • State Tracking: The indie.Var[T] container counts consecutive green (up) or red (down) bars without retaining any history. indie.Var[T] is ideal here since it efficiently holds only the most recent value, unlike indie.MutSeries[T], which maintains a historical sequence of values.

Why Use indie.Var Instead of a Simple Float or Integer?

Unlike using a single float or int, indie.Var is fully compatible with Indie’s series-based environment, where values update bar by bar. Plain variables would reset on each new bar calculation, causing them to lose their previous state. Moreover, indie.Var ensures that values persist correctly in both historical and real-time data. A simple local number variable would reset or lack the ability to update accurately in real-time, making it unsuitable for cumulative calculations. In contrast, indie.Var provides stable, consistent updates across all bars.

​
Key Differences with the Previous Version: General Transformation from MutSeries to Var

In cases where only the latest value is required without retaining historical data, you can convert code from indie.MutSeries to indie.Var by following these general rules:

  1. Replace MutSeries.new() with Var.new():

my_var = Var[float].new(0) # instead of my_series = MutSeries[float].new(0)

  1. Use .get() and .set() instead of indexing:

var_value = my_var.get() # instead of series_value = my_series[0]

my_var.set(new_value) # instead of my_series[0] = new_value

Green/Red Bar Count - usage example of indie.Var[T]
# Copyright (c) 2024 @TakeProfit. All rights reserved.

# This work is licensed under the MIT License.
# For a copy, see <https://opensource.org/licenses/MIT>.

# indie:lang_version = 4
from indie import indicator, Var, Context, plot, color, plot_style

def is_green_bar(ctx: Context) -> bool:
    return ctx.close[0] >= ctx.open[0]

@indicator('Green/Red Bar Counts')
@plot(color=color.GREEN, style=plot_style.STEPS)
@plot(color=color.RED, style=plot_style.STEPS)
def Main(self):
    # `indie.Var[T]` is a generic class, 
    # `T` can be `float`, `int`, `bool`, `str`, etc.
    green_count = Var[int].new(0)
    red_count = Var[int].new(0)
    if is_green_bar(self):
        green_count.set(green_count.get() + 1)
        red_count.set(0)
    else:
        green_count.set(0)
        red_count.set(red_count.get() - 1)

    return green_count.get(), red_count.get()

​
Median Algorithm (written with Indie v4 classes)

This is an example of an Indie algorithm that uses class syntax, introduced in the language since version 4.

The Median class of this indicator inherits from the indie.Algorithm base class. It has two methods: the __init__ constructor and calc. It is essential that the __init__ constructor is executed only once during the creation of the indicator. This makes it a good location to create all objects that hold the algorithm’s state, ensuring these objects do not lose their data when a candle update arrives (triggering the calc method). The stateful objects (fields) are:

  • self._sorted_vals: list[float] - A list of the last length values of the input series src. The algorithm keeps those values ordered.
  • self._prev_src_val: float - The previous value of the last element in the src series.
  • self._bar_count: int - A counter for bars that we use to detect when a new bar arrives.

The main idea of this Median algorithm is to maintain a sliding window of the last length values of the src series and keep these values in sorted order. With such an array, the median is easily calculated as the value in the middle of it.

Median Algorithm (written with Indie v4 classes)
# indie:lang_version = 4
import math
from indie import indicator, Algorithm, Context, SeriesF, MutSeriesF, plot, color, param


def swap(a: list[float], i: int, j: int) -> None:
    tmp = a[i]
    a[i] = a[j]
    a[j] = tmp


# The main idea of this algorithm is to have a sliding window 
# of `length` last values of `src` series and keep them in a sorted order.
# Having such an array, the median is easily calculated as the value 
# in the middle of such an array.
class Median(Algorithm):
    def __init__(self, ctx: Context):
        super().__init__(ctx)
        self._sorted_vals: list[float] = []
        self._prev_src_val = math.nan
        self._bar_count = 0
        
    def calc(self, src: SeriesF, length: int) -> SeriesF:
        # TODO: Need self.ctx.is_new_bar here
        is_new_bar = self._bar_count != len(src)
        self._bar_count = len(src)

        # First we search in our sorted array for an index where 
        # the new element will be inserted. We add new element at position 
        # of some existing element which is need to leave the array anyway
        insert_index = -1
        if len(self._sorted_vals) == length or not is_new_bar:
            val_to_remove = math.nan
            if not is_new_bar:
                val_to_remove = self._prev_src_val
            else:
                val_to_remove = src[length]
            for i in range(len(self._sorted_vals)):
                val = self._sorted_vals[i]
                if val == val_to_remove:
                    insert_index = i
                    break
        else:
            self._sorted_vals.append(math.nan)
            insert_index = len(self._sorted_vals) - 1

        # Insert the new element in our sorted array.
        # After this line the sorted order could be broken (most likely it is)
        self._sorted_vals[insert_index] = self._prev_src_val = src[0]

        # Restore sorted order in our array
        while insert_index > 0 and self._sorted_vals[insert_index - 1] > self._sorted_vals[insert_index]:
            swap(self._sorted_vals, insert_index, insert_index - 1)
            insert_index -= 1
        while insert_index < len(self._sorted_vals) - 1 and self._sorted_vals[insert_index] > self._sorted_vals[insert_index + 1]:
            swap(self._sorted_vals, insert_index, insert_index + 1)
            insert_index += 1

        # Find the median value
        res = math.nan
        if len(self._sorted_vals) == length:
            if length % 2 == 1:
                # Odd number of elements, e.g. [1, 3, 5] so the median is 3
                res = self._sorted_vals[(length - 1) // 2]
            else:
                # Even number of elements, e.g. [1, 3, 5, 7] 
                # so the median is (3 + 5) / 2 = 8
                mid_index = (length - 1) // 2
                left = self._sorted_vals[mid_index]
                right = self._sorted_vals[mid_index + 1]
                res = (left + right) / 2

        return MutSeriesF.new(res)


@indicator('Median', overlay_main_pane=True)
@param.int('length', default=10)
@plot(color=color.WHITE)
def Main(self, length):
    return Median.new(self.close, length)[0]

​
Custom VWAP - usage example of indie.Schedule class

The “Custom VWAP” indicator demonstrates the usage of the Schedule class. If we don’t want to reset VWAP at the start of each new default trading day (using the ‘Session’ option of the anchor parameter), we can create a custom schedule to define the reset time.

For example, by default, the VWAP reset time for ‘BTC/USDT’ is at 00:00 UTC. However, we may want to change this reset time to 00:00 in our chosen timezone (e.g., ‘America/New_York’ for this indicator).

Key Points:

  • The custom schedule is defined in the __init__ function because creating it on each bar in the Context.calc function is resource-inefficient. Alternatively, the Context.pre_calc function can be used for schedule definition if we need to access data from self.info or self.trading_session.
  • The is_same_period method (for both Schedule and Session options) is used to detect whether two timestamps belong to the same schedule period or fall on different days.
Custom VWAP - usage example of indie.Schedule class
# Copyright (c) 2024 @TakeProfit. All rights reserved.

# This work is licensed under the MIT License.
# For a copy, see <https://opensource.org/licenses/MIT>.

# indie:lang_version = 4
from math import sqrt
from datetime import datetime, time
from indie import algorithm, SeriesF, MutSeriesF, Var, Optional, IndieError, MainContext
from indie import indicator, param, source, plot, fill, color, Plot, Fill
from indie.schedule import Schedule, ScheduleRule


@algorithm
def Vwap(self, src: SeriesF, anchor: str, std_dev_mult: float, schedule: Optional[Schedule] = None) -> tuple[SeriesF, SeriesF, SeriesF]:
    '''
    Custom Volume Weighted Average Price
    anchor can be 'Session', 'Day', 'Week', 'Month', 'Year', 'Schedule'
    '''
    cum_weighted_price = Var[float].new(0)
    cum_volume = Var[float].new(0)
    vwap_sum = Var[float].new(0)
    vwap_count = Var[int].new(0)
    vwap_dev_squares = 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 == 'Schedule' and schedule is None:
        raise IndieError("Schedule parameter cannot be None when anchor is set to 'Schedule'")

    if anchor == 'Schedule' and \
            not schedule.value().is_same_period(self.ctx.time[0], self.ctx.time.get(1, 0)):
        need_reset = True
    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
    if anchor == 'Day' and (current_datetime.day != prev_datetime.day or
                            (current_datetime-prev_datetime).days >= 1):
        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:
        cum_weighted_price.set(0)
        cum_volume.set(0)
        vwap_sum.set(0)
        vwap_count.set(0)
        vwap_dev_squares.set(0)

    cum_weighted_price.set(cum_weighted_price.get() + src[0] * self.ctx.volume[0])
    cum_volume.set(cum_volume.get() + self.ctx.volume[0])
    vwap_value = cum_weighted_price.get() / cum_volume.get()

    vwap_sum.set(vwap_sum.get() + vwap_value)
    vwap_count.set(vwap_count.get() + 1)
    vwap_avg = vwap_sum.get() / vwap_count.get()
    vwap_dev_squares.set(vwap_dev_squares.get() + (vwap_value - vwap_avg) ** 2)
    vwap_std_dev = sqrt(vwap_dev_squares.get() / vwap_count.get())

    std_dev = std_dev_mult * vwap_std_dev
    lower = MutSeriesF.new(vwap_value - std_dev)
    upper = MutSeriesF.new(vwap_value + std_dev)

    return MutSeriesF.new(vwap_value), upper, lower

@indicator('Custom VWAP', overlay_main_pane=True)  # Volume Weighted Average Price
@param.source('src', default=source.HLC3, title='Source')
@param.string('anchor', default='Schedule', options=['Session', 'Day', 'Week', 'Month', 'Year', 'Schedule'], title='Anchor')
@param.int('offset', default=0, min=-500, max=500, title='Offset')
@plot('vwap', title='VWAP', color=color.BLUE)
@plot('upper', title='Upper band', color=color.GREEN)
@plot('lower', title='Lower band', color=color.GREEN)
@fill('upper', 'lower', color=color.GREEN(0.1), title='Background')
class Main(MainContext):
    def __init__(self, src, anchor, offset):
        rule = ScheduleRule(start=time(hour=0), end=time(hour=0)) # 24-hour rule
        self.schedule = Schedule(rules=[rule], timezone='America/New_York')
        self.src = src
        self.anchor = anchor
        self.offset = offset

    def calc(self):
        std_dev_mult = 1.0
        main_line, upper, lower = Vwap.new(self.src, self.anchor, std_dev_mult, self.schedule)
        return (
            Plot(main_line[0], offset=self.offset),
            Plot(upper[0], offset=self.offset),
            Plot(lower[0], offset=self.offset),
            Fill(),
        )

Was this page helpful?

Built-in indicatorsChangelog