Data-Analysis-Jupyter/medium/visuals.py

import pandas as pd
import numpy as np
import statsmodels.api as sm
from sklearn.linear_model import LinearRegression
from sklearn.metrics import mean_squared_error


from scipy import stats

import plotly.graph_objs as go
import cufflinks
cufflinks.go_offline()


def make_hist(df, x, category=None):
    """
    Make an interactive histogram, optionally segmented by `category`

    :param df: dataframe of data
    :param x: string of column to use for plotting
    :param category: string representing column to segment by

    :return figure: a plotly histogram to show with iplot or plot
    """
    if category is not None:
        data = []
        for name, group in df.groupby(category):
            data.append(go.Histogram(dict(x=group[x], name=name)))
    else:
        data = [go.Histogram(dict(x=df[x]))]

    layout = go.Layout(
        yaxis=dict(title="Count"),
        xaxis=dict(title=x.replace('_', ' ').title()),
        title=f"{x.replace('_', ' ').title()} Distribution by {category.replace('_', ' ').title()}"
        if category
        else f"{x.replace('_', ' ').title()} Distribution",
    )

    figure = go.Figure(data=data, layout=layout)
    return figure


def make_cum_plot(df, y, category=None, ranges=False):
    """
    Make an interactive cumulative plot, optionally segmented by `category`

    :param df: dataframe of data, must have a `published_date` column
    :param y: string of column to use for plotting or list of two strings for double y axis
    :param category: string representing column to segment by
    :param ranges: boolean for whether to add range slider and range selector

    :return figure: a plotly plot to show with iplot or plot
    """
    if category is not None:
        data = []
        for i, (name, group) in enumerate(df.groupby(category)):
            group.sort_values("published_date", inplace=True)
            data.append(
                go.Scatter(
                    x=group["published_date"],
                    y=group[y].cumsum(),
                    mode="lines+markers",
                    text=group["title"],
                    name=name,
                    marker=dict(size=10, opacity=0.8,
                                symbol=i + 2),
                )
            )
    else:
        df.sort_values("published_date", inplace=True)
        if len(y) == 2:
            data = [
                go.Scatter(
                    x=df["published_date"],
                    y=df[y[0]].cumsum(),
                    name=y[0].title(),
                    mode="lines+markers",
                    text=df["title"],
                    marker=dict(size=10, color='blue', opacity=0.6, line=dict(color='black'),
                                )),
                go.Scatter(
                    x=df["published_date"],
                    y=df[y[1]].cumsum(),
                    yaxis='y2',
                    name=y[1].title(),
                    mode="lines+markers",
                    text=df["title"],
                    marker=dict(size=10, color='red', opacity=0.6, line=dict(color='black'),
                                )),
            ]
        else:
            data = [
                go.Scatter(
                    x=df["published_date"],
                    y=df[y].cumsum(),
                    mode="lines+markers",
                    text=df["title"],
                    marker=dict(size=12, color='blue', opacity=0.6, line=dict(color='black'),
                                ),
                )
            ]
    if len(y) == 2:
        layout = go.Layout(
            xaxis=dict(title="Published Date", type="date"),
            yaxis=dict(title=y[0].replace('_', ' ').title(), color='blue'),
            yaxis2=dict(title=y[1].replace('_', ' ').title(), color='red',
                        overlaying='y', side='right'),
            font=dict(size=14),
            title=f"Cumulative {y[0].title()} and {y[1].title()}",
        )
    else:
        layout = go.Layout(
            xaxis=dict(title="Published Date", type="date"),
            yaxis=dict(title=y.replace('_', ' ').title()),
            font=dict(size=14),
            title=f"Cumulative {y.replace('_', ' ').title()} by {category.replace('_', ' ').title()}"
            if category is not None
            else f"Cumulative {y.replace('_', ' ').title()}",
        )

    # Add a rangeselector and rangeslider for a data xaxis
    if ranges:
        rangeselector = dict(
            buttons=list(
                [
                    dict(count=1, label="1m", step="month", stepmode="backward"),
                    dict(count=6, label="6m", step="month", stepmode="backward"),
                    dict(count=1, label="1y", step="year", stepmode="backward"),
                    dict(step="all"),
                ]
            )
        )
        rangeslider = dict(visible=True)
        layout["xaxis"]["rangeselector"] = rangeselector
        layout["xaxis"]["rangeslider"] = rangeslider
        layout['width'] = 1000
        layout['height'] = 600

    figure = go.Figure(data=data, layout=layout)
    return figure


def make_scatter_plot(df, x, y, fits=None, xlog=False, ylog=False, category=None, scale=None, sizeref=2, annotations=None, ranges=False, title_override=None):
    """
    Make an interactive scatterplot, optionally segmented by `category`

    :param df: dataframe of data
    :param x: string of column to use for xaxis
    :param y: string of column to use for yaxis
    :param fits: list of strings of fits
    :param xlog: boolean for making a log xaxis
    :param ylog boolean for making a log yaxis
    :param category: string representing categorical column to segment by, this must be a categorical
    :param scale: string representing numerical column to size and color markers by, this must be numerical data
    :param sizeref: float or integer for setting the size of markers according to the scale, only used if scale is set
    :param annotations: text to display on the plot (dictionary)
    :param ranges: boolean for whether to add a range slider and selector
    :param title_override: String to override the title

    :return figure: a plotly plot to show with iplot or plot
    """
    if category is not None:
        title = f"{y.replace('_', ' ').title()} vs {x.replace('_', ' ').title()} by {category.replace('_', ' ').title()}"
        data = []
        for i, (name, group) in enumerate(df.groupby(category)):
            data.append(go.Scatter(x=group[x],
                                   y=group[y],
                                   mode='markers',
                                   text=group['title'],
                                   name=name,
                                   marker=dict(size=8, symbol=i + 2)))

    else:
        if scale is not None:
            title = f"{y.replace('_', ' ').title()} vs {x.replace('_', ' ').title()} Scaled by {scale.title()}"
            data = [go.Scatter(x=df[x],
                               y=df[y],
                               mode='markers',
                               text=df['title'], marker=dict(size=df[scale],
                                                             line=dict(color='black', width=0.5), sizemode='area', sizeref=sizeref, opacity=0.8,
                                                             colorscale='Viridis', color=df[scale], showscale=True, sizemin=2))]
        else:

            df.sort_values(x, inplace=True)
            title = f"{y.replace('_', ' ').title()} vs {x.replace('_', ' ').title()}"
            data = [go.Scatter(x=df[x],
                               y=df[y],
                               mode='markers',
                               text=df['title'], marker=dict(
                size=12, color='blue', opacity=0.8, line=dict(color='black')),
                name='observations')]
            if fits is not None:
                for fit in fits:
                    data.append(go.Scatter(x=df[x], y=df[fit], text=df['title'],
                                           mode='lines+markers', marker=dict
                                           (size=8, opacity=0.6),
                                           line=dict(dash='dash'), name=fit))

                title += ' with Fit'
    layout = go.Layout(annotations=annotations,
                       xaxis=dict(title=x.replace('_', ' ').title() + (' (log scale)' if xlog else ''),
                                  type='log' if xlog else None),
                       yaxis=dict(title=y.replace('_', ' ').title() + (' (log scale)' if ylog else ''),
                                  type='log' if ylog else None),
                       font=dict(size=14),
                       title=title if title_override is None else title_override,
                       )

    # Add a rangeselector and rangeslider for a data xaxis
    if ranges:
        rangeselector = dict(
            buttons=list(
                [
                    dict(count=1, label="1m", step="month", stepmode="backward"),
                    dict(count=6, label="6m", step="month", stepmode="backward"),
                    dict(count=1, label="1y", step="year", stepmode="backward"),
                    dict(step="all"),
                ]
            )
        )
        rangeslider = dict(visible=True)
        layout["xaxis"]["rangeselector"] = rangeselector
        layout["xaxis"]["rangeslider"] = rangeslider
        layout['width'] = 1000
        layout['height'] = 600

    figure = go.Figure(data=data, layout=layout)
    return figure


def make_linear_regression(df, x, y, intercept_0):
    """
    Create a linear regression, either with the intercept set to 0 or
    the intercept allowed to be fitted

    :param df: dataframe with data
    :param x: string or list of stringsfor the name of the column with x data
    :param y: string for the name of the column with y data
    :param intercept_0: boolean indicating whether to set the intercept to 0
    """
    if isinstance(x, list):
        lin_model = LinearRegression()
        lin_model.fit(df[x], df[y])

        slopes, intercept, = lin_model.coef_, lin_model.intercept_
        df['predicted'] = lin_model.predict(df[x])
        r2 = lin_model.score(df[x], df[y])
        rmse = np.sqrt(mean_squared_error(
            y_true=df[y], y_pred=df['predicted']))
        equation = f'{y.replace("_", " ")} ='

        names = ['r2', 'rmse', 'intercept']
        values = [r2, rmse, intercept]
        for i, (p, s) in enumerate(zip(x, slopes)):
            if (i + 1) % 3 == 0:
                equation += f'<br>{s:.2f} * {p.replace("_", " ")} +'
            else:
                equation += f' {s:.2f} * {p.replace("_", " ")} +'
            names.append(p)
            values.append(s)

        equation += f' {intercept:.2f}'
        annotations = [dict(x=0.4 * df.index.max(), y=0.9 * df[y].max(), showarrow=False,
                            text=equation,
                            font=dict(size=10))]

        df['index'] = list(df.index)
        figure = make_scatter_plot(df, x='index', y=y, fits=[
                                   'predicted'], annotations=annotations)
        summary = pd.DataFrame({'name': names, 'value': values})
    else:
        if intercept_0:
            lin_reg = sm.OLS(df[y], df[x]).fit()
            df['fit_values'] = lin_reg.fittedvalues
            summary = lin_reg.summary()
            slope = float(lin_reg.params)
            equation = f"${y.replace('_', ' ')} = {slope:.2f} * {x.replace('_', ' ')}$"

        else:
            lin_reg = stats.linregress(df[x], df[y])
            intercept, slope = lin_reg.intercept, lin_reg.slope
            params = ['pvalue', 'rvalue', 'slope', 'intercept']
            values = []
            for p in params:
                values.append(getattr(lin_reg, p))
            summary = pd.DataFrame({'param': params, 'value': values})
            df['fit_values'] = df[x] * slope + intercept
            equation = f"${y.replace('_', ' ')} = {slope:.2f} * {x.replace('_', ' ')} + {intercept:.2f}$"

        annotations = [dict(x=0.75 * df[x].max(), y=0.9 * df[y].max(), showarrow=False,
                            text=equation,
                            font=dict(size=32))]
        figure = make_scatter_plot(
            df, x=x, y=y, fits=['fit_values'], annotations=annotations)
    return figure, summary


def make_poly_fits(df, x, y, degree=6):
    """
    Generate fits and make interactive plot with fits

    :param df: dataframe with data
    :param x: string representing x data column
    :param y: string representing y data column
    :param degree: integer degree of fits to go up to

    :return fit_stats: dataframe with information about fits
    :return figure: interactive plotly figure that can be shown with iplot or plot
    """

    # Don't want to alter original data frame
    df = df.copy()
    fit_list = []
    rmse = []
    fit_params = []

    # Make each fit
    for i in range(1, degree + 1):
        fit_name = f'fit degree = {i}'
        fit_list.append(fit_name)
        z, res, *rest = np.polyfit(df[x], df[y], i, full=True)
        fit_params.append(z)
        df.loc[:, fit_name] = np.poly1d(z)(df[x])
        rmse.append(np.sqrt(res[0]))

    fit_stats = pd.DataFrame(
        {'fit': fit_list, 'rmse': rmse, 'params': fit_params})
    figure = make_scatter_plot(df, x=x, y=y, fits=fit_list)
    return figure, fit_stats


def make_extrapolation(df, y, years, degree=4):
    """
    Extrapolate `y` into the future `years` with `degree`  polynomial fit

    :param df: dataframe of data
    :param y: string of column to extrapolate
    :param years: number of years to extrapolate into the future
    :param degree: integer degree of polynomial fit

    :return figure: plotly figure for display using iplot or plot
    :return future_df: extrapolated numbers into the future
    """

    df = df.copy()
    x = 'days_since_start'
    df['days_since_start'] = (
        (df['published_date'] - df['published_date'].min()).
        dt.total_seconds() / (3600 * 24)).astype(int)

    cumy = f'cum_{y}'
    df[cumy] = df.sort_values(x)[y].cumsum()

    figure, summary = make_poly_fits(df, x, cumy, degree=degree)

    min_date = df['published_date'].min()
    max_date = df['published_date'].max()

    date_range = pd.date_range(start=min_date,
                               end=max_date + pd.Timedelta(days=int(years * 365)))

    future_df = pd.DataFrame({'date': date_range})

    future_df[x] = (
        (future_df['date'] - future_df['date'].min()).
        dt.total_seconds() / (3600 * 24)).astype(int)

    newcumy = f'cumulative_{y}'

    future_df = future_df.merge(df[[x, cumy]], on=x, how='left').\
        rename(columns={cumy: newcumy})

    z = np.poly1d(summary.iloc[-1]['params'])
    pred_name = f'predicted_{y}'
    future_df[pred_name] = z(future_df[x])
    future_df['title'] = ''

    last_date = future_df.loc[future_df['date'].idxmax()]
    prediction_text = (
        f"On {last_date['date'].date()} the {y} will be {float(last_date[pred_name]):,.0f}.")
    annotations = [dict(x=future_df['date'].quantile(0.4),
                        y=0.8 * future_df[pred_name].max(), text=prediction_text, showarrow=False,
                        font=dict(size=16))]

    title_override = f'{y.replace("_", " ").title()} with Extrapolation {years} Years into the Future'

    figure = make_scatter_plot(future_df, 'date', newcumy, fits=[
                               pred_name], annotations=annotations, ranges=True, title_override=title_override)
    return figure, future_df