Interactive time series charts (with Bokeh)

import os, sys
import datetime as dt
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import bokeh

from IPython.core.interactiveshell import InteractiveShell
InteractiveShell.ast_node_interactivity = 'all'  # display all output cells

from bokeh.plotting import figure, show, reset_output, save
from bokeh.models import Band
from bokeh.palettes import Category10

import random
from IPython.core.display import HTML
from bokeh.plotting.figure import Figure
from bokeh.plotting import figure, show, reset_output, save
from bokeh.io import output_notebook, output_file

output_notebook(hide_banner=True)

def hugo_safe_render(fig: Figure) -> HTML:
    """ Save bokeh plot as HTML and re-embed into notebook, so that it also renders in Hugo markdown. """
    reset_output()
    name = f'{random.getrandbits(32)}.html'
    output_file(name)
    _ = save(fig)
    
    return HTML(f'<embed type="text/html" src="{name}" width="100%" height="{int(fig.height+100)}"></embed>')

fig = figure(plot_height=200, sizing_mode='scale_width')

x = [1, 2, 3, 4, 5]
y = [6, 7, 2, 4, 5]

_ = fig.line(x, y, line_width=2)

hugo_safe_render(fig)

from bokeh.models import ColumnDataSource

try:
    from bokeh.sampledata.stocks import AAPL
except RuntimeError:
    from bokeh import sampledata
    sampledata.download()
    from bokeh.sampledata.stocks import AAPL

aapl_df = (
    pd.DataFrame(
        {'price': AAPL['adj_close']},
        index=pd.DatetimeIndex(AAPL['date']).rename('date')
    ).reset_index()
    .assign(lower=lambda x: x['price'] - 50, upper=lambda x: x['price'] + 50)   
)

cds_aapl = ColumnDataSource(aapl_df)

fig = figure(
    plot_height=200,
    sizing_mode='scale_width',
    x_axis_type='datetime'
)

_ = fig.line('date', 'price', source=cds_aapl)

hugo_safe_render(fig)

# reset_output()
# output_notebook()

from bokeh.models import HoverTool, DatetimeTickFormatter

fig = figure(
    plot_height=200,
    sizing_mode='scale_width',
    x_axis_type='datetime'
)

cds_aapl = ColumnDataSource(aapl_df)
_ = fig.line('date', 'price', source=cds_aapl)


def datetime(x):
    return np.array(x, dtype=np.datetime64)

fig.add_tools(HoverTool(
    tooltips=[
        ('Date', '@date{%F}'),
        ('Price', '@price')
    ],
    formatters={'@date': 'datetime'},
    mode='vline'
))

fig.xaxis.formatter = DatetimeTickFormatter(
    months=['%b %Y'],
    days=['%b %d']
)

hugo_safe_render(fig)

from bokeh.sampledata.stocks import AAPL, GOOG

goog = pd.Series(GOOG['adj_close'], index=pd.DatetimeIndex(GOOG['date']).rename('date')).rename('GOOG')
aapl = pd.Series(AAPL['adj_close'], index=pd.DatetimeIndex(AAPL['date']).rename('date')).rename('AAPL')

df = (
    pd.concat([goog, aapl], axis=1)
    .dropna()
    .rename_axis('stock', axis=1)
    .stack()
    .rename('price')
    .to_frame()
    .assign(lower=lambda x: x['price'] - 50, upper=lambda x: x['price'] + 50)
    .set_axis(['val', 'lower', 'upper'], axis=1, inplace=False)
)

df.head()

from typing import Union, Dict


def preproc_dispatcher(data: Union[pd.Series, pd.DataFrame]) -> Dict[str, pd.DataFrame]:
    """ Dispatch input-specific pre-processing functions for various potential data formats.
    
    Bokeh expects a DataFrame (without index) as input to ColumnDataSource.
    
    """
    
    n_idx_levels = len(data.index.names)
    is_srs = isinstance(data, pd.Series)
    is_df = isinstance(data, pd.DataFrame)
    
    if is_srs and n_idx_levels == 1:
        #print('Single index series')
        return preproc_single_index_series(data)

    elif is_srs and n_idx_levels == 2:
        #print('Multi-index series')
        return preproc_multi_index_series(data)

    elif is_df and n_idx_levels == 2:
        assert set(data.columns) == set(['val', 'lower', 'upper']), 'Expecting three column names: (val, upper, lower)'
        #print('Multi-index dataframe')
        return preproc_multi_index_df(data)
    
    else:
        raise ValueError

        
def preproc_single_index_series(srs: pd.Series) -> Dict[str, pd.DataFrame]:
    """ A single-index series just needs to be reset and given it's name. """

    return {srs.name: srs.reset_index()}


def preproc_multi_index_series(srs: pd.Series) -> Dict[str, pd.DataFrame]:
    """ A multi-index series needs to be split into groups. """

    return {
        x: sub_df.reset_index(level=1, drop=True).reset_index().sort_index()
        for x, sub_df in df.groupby(level=[1])
    }

def preproc_multi_index_df(df: pd.DataFrame) -> Dict[str, pd.DataFrame]:
    """ A multi-index dataframe needs to be split into groups. """

    return {
        x: sub_df.reset_index(level=1, drop=True).reset_index().sort_index()
        for x, sub_df in df.groupby(level=[1])
    }

print('Multi-line time series\n'.upper())
multi_index_srs = df['val']

print('Input:')
multi_index_srs.head()

print('Output:')
preproc_dispatcher(multi_index_srs)

print('Single-line time series\n'.upper())
single_index_srs = df['val'].xs('GOOG', level=1)

print('Input:')
single_index_srs.head()

print('Output:')
preproc_dispatcher(single_index_srs)

print('Multi-line time series with bands\n'.upper())

print('Input:')
df.head()

print('Output:')
preproc_dispatcher(df)

from typing import Dict
from abc import ABC, abstractmethod

class TimeSeries(ABC):
    def __init__(self, data: Union[pd.Series, pd.DataFrame], title: str = ''):
        """ """
        
        self.data = self._process_data(data)
        
        indices = [df.iloc[:, 0] for df in self.data.values()]
        assert all((x == indices[0]).all() for x in indices), 'Indices are not the same'
        x_range = (indices[0].min(), indices[0].max())
        
        self.fig = self._setup_plot(title, x_range)
        fig = self._plot(self.fig, self.data)
        
    
    @staticmethod
    def _process_data(data) -> Dict[str, pd.DataFrame]:
        
        return preproc_dispatcher(data)
            
    
    @staticmethod
    def _setup_plot(title, x_range) -> Figure:
        """ Wrapper that sets up fig and applys common formatting to a bokeh plot. """

        from bokeh.models import DataRange1d

        fig = figure(
            plot_height=300,
            sizing_mode='scale_width',
            title=title,
            x_axis_type='datetime',
            # get rid of gap on x-axis
            x_range=DataRange1d(start=x_range[0], end=x_range[1])  
        )

        # FORMATTING
        fig.add_tools(HoverTool(
        tooltips=[
            ('Date', '@date{%F}'),
            ('Name', '$name'),
            
            ('Value', '@val')
        ],
        formatters={'@date': 'datetime'},
        ))

        return fig
    
    @abstractmethod
    def _plot(fig, df_dict: Dict[str, pd.DataFrame]) -> Figure:
        """ Core plotting logic for a multiple time series chart. """
    
        return NotImplementedError
    
    
    def show(self):
        #show(self.fig)
        return hugo_safe_render(self.fig)
        

class MultipleTimeSeries(TimeSeries):

    @staticmethod
    def _plot(fig, df_dict: Dict[str, pd.DataFrame]) -> Figure:
        """ Core plotting logic for a multiple time series chart. """
        
        for i, (stock, sub_df) in enumerate(df_dict.items()):
            color = Category10[3][i]
            cds = ColumnDataSource(sub_df)

            fig.line(            
                'date',
                'val',
                source=cds,
                name=stock,
                color=color
            )

        return fig

MultipleTimeSeries(multi_index_srs, title='Demo of multiple time series').show()

single_index_srs = df['val'].xs('GOOG', level=1)

MultipleTimeSeries(single_index_srs, title='Demo of single time series').show()

class MultipleTimeSeriesWithBands(TimeSeries):

    @staticmethod
    def _plot(fig, df_dict: Dict[str, pd.DataFrame]) -> Figure:
        """ Core plotting logic for a multiple time series chart. """
    
        for i, (dim, sub_df) in enumerate(df_dict.items()):
            color = Category10[3][i]
            cds = ColumnDataSource(sub_df)

            fig.line(
                'date',
                'val',
                source=cds,
                name=dim,
                color=color
            )

            band = Band(
                base='date',
                lower='lower',
                upper='upper',
                source=cds,
                level='underlay',
                fill_alpha=0.1,
                fill_color=color,

                line_width=0
            )

            fig.add_layout(band)
        return fig

MultipleTimeSeriesWithBands(df, title='Demo of upper/lower bands').show()

class MultipleTimeSeriesWithMovingAverage(TimeSeries):

    @staticmethod
    def _plot(fig, df_dict: Dict[str, pd.DataFrame]) -> Figure:
        """ Core plotting logic for a multiple time series chart. """
    
        for i, (dim, sub_df) in enumerate(df_dict.items()):
            color = Category10[3][i]
            
            sub_df.loc[:, 'val_ma7d'] = sub_df['val'].rolling(window=7, min_periods=1).mean()
            cds = ColumnDataSource(sub_df)

            # plot 1d
            fig.line(
                'date',
                'val',
                source=cds,
                name=f'{dim} (1d)',

                color=color,
                alpha=0.1
            )

            # plot 7d
            fig.line(
                'date',
                'val_ma7d',
                source=cds,
                name=f'{dim} (7d)',
                color=color
            )

        return fig

    
recent_df = df.loc['2012-09-28':, :]

MultipleTimeSeriesWithMovingAverage(recent_df, title='Demo of moving averages').show()