""" Functional Transformers """ from datetime import datetime, timedelta import numpy as np import pandas as pd def rollfoward_snapshot_dates(df, from_date= datetime.now(), rollforward_days=7, column='SNAPSHOT_DATE'): """ Updates snapshot year params: df (pd.DataFrame) - pandas dataframe from_date (date) - date to roll forward from """ _df = df.copy() end_date = from_date + timedelta(days=rollforward_days - 1) rollout_dates = pd.date_range(start=from_date, end=end_date) _df[column] = rollout_dates return _df def rollforward_snapshot_year(df, column='SNAPSHOT_YEAR'): """ Updates snapshot year """ _df = df.copy() _df[column] = np.array([datetime.now().year for _ in range(len(df))]) return _df def rollforward_snapshot_isoweek(df, column='SNAPSHOT_ISOWEEK'): """ Updates snapshot isoweek """ # roll forward snapshot date _df = df.copy() _df[column] = _df['SNAPSHOT_DATE'].apply(lambda date_x: date_x.isocalendar()[1]) return _df def rollforward_snapshot_month(df, column='SNAPSHOT_MONTH'): """ Updates snapshot month """ # roll forward snapshot date _df = df.copy() _df[column] = _df['SNAPSHOT_DATE'].apply(lambda date_x: date_x.month) return _df def rollforward_snapshot_day_of_year(df, column='SNAPSHOT_DAY_OF_YEAR'): """ Updates snapshot day of year """ # roll forward snapshot date _df = df.copy() _df[column] = _df['SNAPSHOT_DATE'].apply(lambda date_x: date_x.timetuple().tm_yday) return _df def rollforward_snapshot_day_of_week(df, column='SNAPSHOT_DAY_OF_WEEK'): """ Updates snapshot day of week """ # roll forward snapshot date _df = df.copy() _df[column] = _df['SNAPSHOT_DATE'].apply(lambda date_x: date_x.weekday()) return _df def rollforward_release_date(df, release_date:datetime, column='RELEASE_DATE'): """ Updates update release date """ _df = df.copy() if isinstance(df, str): release_date = pd.to_datetime(release_date) _df[column] = release_date return _df def rollforward_release_year(df, column='RELEASE_YEAR'): """ Updates update release date """ _df = df.copy() _df[column] = _df['RELEASE_DATE'].apply(lambda date_x: date_x.year) return _df def rollforward_release_month(df, column='RELEASE_MONTH'): """ Updates update release date """ _df = df.copy() _df[column] = _df['RELEASE_DATE'].apply(lambda date_x: date_x.month) return _df def rollforward_release_day_of_week(df, column='RELEASE_DOW'): """ Updates update release date """ _df = df.copy() _df[column] = _df['RELEASE_DATE'].apply(lambda date_x: date_x.weekday()) return _df def rollforward_release_weekiso(df, column='RELEASE_WEEKISO'): """ Updates update release date """ _df = df.copy() _df[column] = _df['RELEASE_DATE'].apply(lambda date_x: date_x.isocalendar()[1]) return _df def rollforward_release_age_days(df, column='RELEASE_AGE_DAYS'): """ Updates update release date """ _df = df.copy() _df[column] = _df.apply(lambda row: (pd.to_datetime(row['SNAPSHOT_DATE']) - pd.to_datetime(row['RELEASE_DATE'])).days, axis=1) return _df def rollforward_release_age_weeks(df, column='RELEASE_AGE_WEEKS'): """ Updates update release date """ _df = df.copy() _df[column] = _df.apply(lambda row: (pd.to_datetime(row['SNAPSHOT_DATE']) - pd.to_datetime(row['RELEASE_DATE'])).days //7 , axis=1) return _df