experimentation_mlops/mlops/modules/transformations.py

"""
Stores all transformations
"""
def createLag(data, amt=10):
    """
    Create a lag inside dataframe, in business days

    :param pandas.DataFrame data:
    :param int amt: Unit of lag period

    :return: Copy of pandas Dataframe
    """
    import pandas as pd
    if 'ds' in data:
        copy = data.copy()
        copy['ds'] = copy['ds'] + pd.tseries.offsets.BusinessDay(amt)
        return copy
    else:
        print(f"No 'ds' column found inside dataframe")
        return data
    
def scaleStandard(df_col):
    """
    Fits and returns a standard scaled version of a dataframe column
    """
    import pandas as pd
    from sklearn.preprocessing import StandardScaler
    scaler = StandardScaler()
    df_col = scaler.fit_transform(df_col)
    df_col = pd.DataFrame(df_col)
    return df_col, scaler

def logReturn(data, df_col):
    """
    Perform log return for a dataframe column
    """
    import numpy as np
    new_col = np.log1p(data[df_col].pct_change())
    return new_col