Intrinsic Time implementation, from the Alpha Engine whitepaper

intrinsic_time.py

class IntrinsicTimeBar:
    reversal: False
    direction: None
    time: None
    high: None
    low: None

    def __init__(self, time, high: float, low: float, direction=None, reversal=False):
        self.reversal = reversal
        self.direction = direction
        self.time = time
        self.high = high
        self.low = low


class IntrinsicTimeChart:
    maxPercentageChange = None
    maxLength = None
    lengthNow = 0
    OnStep = None
    upper = None
    lower = None

    def __init__(self, max_percentage_change: float, length=2):
        self.maxPercentageChange = max_percentage_change
        self.maxLength = max(2, length)
        self.window = []
        self.upper = None
        self.lower = None

    def __AddBar(self, new_bar: IntrinsicTimeBar):
        self.window.insert(0, new_bar)
        length = len(self.window)

        if length > self.maxLength:
            self.window.pop(self.maxLength)
        else:
            self.lengthNow = length

        if self.OnStep:
            self.OnStep(new_bar)

    def NewCandlestick(self, time, bar):
        if self.upper == None:
            self.upper = bar['Open']
            self.lower = bar['Open']

        self.NewPrice(time, bar['Open'])

        # process high and low in order of proximity from the open
        ohlc_order = [
            ('High',  abs(bar['Open'] - bar['High'])),
            ('Low',   abs(bar['Open'] - bar['Low']))
        ]
        for prop in [t[0] for t in sorted(ohlc_order, key=lambda t: t[1])]:
            self.NewPrice(time, bar[prop])

        self.NewPrice(time, bar['Close'])

    def NewPrice(self, time, price: float):
        self.upper = price if self.upper == None else max(price, self.upper)
        self.lower = price if self.lower == None else min(price, self.lower)

        allowable_change = (self.upper + self.lower) * \
            0.5 * self.maxPercentageChange

        possible_high = self.lower + allowable_change
        possible_low = self.upper - allowable_change
        surpassed_high = price > possible_high
        surpassed_low = price < possible_low

        if not surpassed_high and not surpassed_low:
            return

        new_level = None
        new_bar = IntrinsicTimeBar(
            time, self.upper, self.lower, direction=None, reversal=False)

        if surpassed_high:
            new_bar.high = new_level = possible_high
            new_bar.direction = 'u'
            if self.lengthNow > 0 and self.window[0].direction == 'd':
                new_bar.reversal = True

        elif surpassed_low:
            new_bar.low = new_level = possible_low
            new_bar.direction = 'd'
            if self.lengthNow > 0 and self.window[0].direction == 'u':
                new_bar.reversal = True

        else:
            raise Exception('the impossible has happened')

        self.upper = new_level
        self.lower = new_level
        self.__AddBar(new_bar)
        self.NewPrice(time, price)

research_book.py

qb = QuantBook()
spy = qb.AddEquity("SPY")
days = 360
history = qb.History(qb.Securities.Keys, days, Resolution.Daily)

#--------------------------------------------------------------#

import pandas as pd
import mplfinance as mpf
from IntrinsicTime import IntrinsicTimeChart

itc = IntrinsicTimeChart(0.03, days + 1)

# write every new bar into table struct at IT run-time
new_table = {"Date":[], "Open": [], "High": [], "Low": [], "Close": []}
def writeBarToTable(bar):
    new_table["Date"].append(bar.time)
    new_table["Low"].append(bar.low)
    new_table["High"].append(bar.high)
    if bar.direction == 'u':
        new_table["Open"].append(bar.low)
        new_table["Close"].append(bar.high)
    else:
        new_table["Open"].append(bar.high)
        new_table["Close"].append(bar.low)
itc.OnStep = writeBarToTable

# push price history into IT
for index, row in history.iterrows():
    bar = TradeBar()
    bar.Open  = row['open']
    bar.High  = row['high']
    bar.Low   = row['low']
    bar.Close = row['close']
    itc.NewCandlestick(index[1], bar)

# plot intrinsic time series
df = pd.DataFrame(new_table)
df.index = df["Date"]
mpf.plot(df, type='candlestick')

# plot historical price as OHLC
spy = history.loc['SPY']
spy.rename(columns={"open": "Open", "high": "High", "low": "Low", "close": "Close", "volume": "Volume"}, inplace=True)
mpf.plot(spy)

spy_intrinsic_time.png

spy_ohlc.png