import logging import numpy as np import pandas as pd from typing import Dict from utils.base_attributes_generation import get_fan_attribute_values from .algo import Algo component_logger = logging.getLogger().getChild("enrichment.enrich_superfan") class EnrichSuperfan(Algo): """ Implement FanSifter's proprietary segmenting method described here: - https://fansifter.atlassian.net/wiki/spaces/FM/pages/198017029/Superfans """ source_attribute_ids = [ 75, # enrPurchaseQuantity 87, # userYoutubeFollower # to calculate media_follows 88, # userTwitterFollower # to calculate media_follows 89, # userSpotifyFollower # to calculate media_follows 90, # userFanClubMember # to calculate fan_club 91, # userTracksPlayed 45, # marketingOptIn # to calculate unique_newsletter & marketing_optin 86, # purchaseType 100, # presave 101, # subscription_type 102, # profile_followers ] result_attribute_ids = [ 68, # enrSuperfanScore 69, # enrSuperfan 81, # enrSuperfanSegment 103, # 'superfan_fanclub_member', 104, # 'superfan_media_follows_count', 105, # 'superfan_unique_newsletter_signups', 106, # 'superfan_vip_status', 107, # 'superfan_marketing_optin', 108, # 'superfan_presave_status', 109, # 'superfan_paid_status', 112, # 'superfan_tracks_played', 113, # 'superfan_purchase_quantity' ] segment_attribute_name = "enrSuperfan" segment_name = "{}" data_gathering_rule = "strict" max_superfan_score = 25 scores: Dict[str, int] = {} def __init__( self, collection_id: int, schema: str, user_id: str, default_value: str = "unknown", **kwargs, ): super().__init__(collection_id, schema, user_id, default_value) # @timeit def _run(self): """ Main enrichment method """ source_df = self.available_data["source"] # TODO! Consider adding those as ML input (in which case do this in feature_engineering instead) source_df = self._get_media_follows_count(source_df) # media_follows_count source_df = self._get_unique_newsletter_signups_count(source_df) # unique_newsletter_signups source_df = self._get_marketing_optin(source_df) # marketing_optin source_df = self._get_fan_club(source_df) # fanclub_member source_df = self._get_vip_indicator(source_df) # vip_status source_df = self._get_presave_indicator(source_df) # presave indicator source_df = self._get_paid_subscription_indicator(source_df) # paid subscription self.scores = self.get_attribute_scores() # Columns that get assigned a score. They exclude source attrs that are only used for feature engineering # NB! It is of paramount importance those features are integers, so they could be multiplied superfan_calc_cols = [ "fanclub_member", "media_follows_count", "unique_newsletter_signups", "75", # enrPurchaseQuantity "vip_status", "marketing_optin", "91", # tracksplayed "presave_status", "paid_status", "102", # social_media_followers ] """ We aggregate rows per fan. This is especially important with custom segments. Since we have integers, we can simply sum. We convert those cols to numeric first. We add 91 separately, because we have special handling for that, where we'""" if len(source_df) > 1000 and hasattr(source_df, "parallel_apply"): source_df[superfan_calc_cols] = source_df[superfan_calc_cols].parallel_apply( pd.to_numeric, axis=0, args=["coerce"] ) else: source_df[superfan_calc_cols] = source_df[superfan_calc_cols].apply( lambda x: pd.to_numeric(x, errors="coerce") ) source_df = source_df.groupby(["fan_id"]).agg({x: "sum" for x in superfan_calc_cols}) """ Assign scores from "attribute scoring map" to our selected columns (by multiplication). """ superfan_calc_cols.remove("91") # don't use these in "regular" calculation (we have special handling for this) superfan_calc_cols.remove("102") source_df["enrSuperfanScore"] = ( source_df[superfan_calc_cols] .assign(**self.scores) .mul(source_df[superfan_calc_cols].fillna(0)) # We need to fill nans with 0, else multiplication fails .clip(0, 25) # Maximum allowed score is 25 (as per agreed methodology) .sum(axis=1) ) # create columns to store each attributes SF score for score_item in superfan_calc_cols: source_df[score_item + "_score"] = source_df[score_item] * self.scores[score_item] """ Special case handling for "tracks played", where we calculate score based on quantiles. """ tracks_played_col = "91" # Need to convert to float for some basic opereations source_df = source_df.astype({tracks_played_col: np.float32}) # We can't operate with NaNs here, replace with 0. Also change type to float. source_df.fillna(value=0, inplace=True) # We won't go through this block when we don't have anything in tracks_played col if int(source_df[tracks_played_col].sum()) != 0: # Create 4 percentiles and give them labels 1, 2, 3, 4 q_labels = range(1, 5) # It can fail when not enough unique values for binning. try: # We want to calculate quantiles only where we have values for tracks played tracks_df = source_df.loc[source_df[tracks_played_col] > 0] q_groups = pd.qcut(tracks_df[tracks_played_col].rank(method="first"), q=4, labels=q_labels) # Create new temporary df with quantile values tracks_df = tracks_df.assign(tracks_quantile=q_groups.values) tracks_df = tracks_df.astype({"tracks_quantile": int}) # Add to total score in our original df source_df["tracks_quantile"] = tracks_df["tracks_quantile"] source_df.fillna(0, inplace=True) source_df["enrSuperfanScore"] = source_df["enrSuperfanScore"] + source_df["tracks_quantile"] # add score column source_df["tracks_played_score"] = source_df["tracks_quantile"] except Exception as e: component_logger.error( f"Could not calculate tracks_played score because not enough unique values: {e}" ) else: source_df["tracks_played_score"] = 0 """ Special case handling for "social media followers", where we calculate score based on 75th quantile. """ social_media_follow_col = "102" # Need to convert to float for some basic opereations source_df = source_df.astype({social_media_follow_col: np.float32}) # We can't operate with NaNs here, replace with 0. Also change type to float. source_df.fillna(value=0, inplace=True) # We won't go through this block when we don't have anything in social_media_followers col if int(source_df[social_media_follow_col].sum()) != 0: # Find 75th quantile q_75 = source_df[social_media_follow_col].quantile(0.75) # It can fail when not enough unique values for binning. try: # find if value is over/under q_75 source_df["social_media_followers_scores"] = np.where(source_df[social_media_follow_col] > q_75, 1, 0) source_df.fillna(0, inplace=True) source_df["enrSuperfanScore"] = ( source_df["enrSuperfanScore"] + source_df["social_media_followers_scores"] ) # add score column source_df["media_follows_count_score"] = source_df["social_media_followers_scores"] except Exception as e: component_logger.error( f"Could not calculate social_media_followers score because not enough unique values: {e}" ) else: source_df["media_follows_count_score"] = 0 self.max_superfan_score = int(source_df["enrSuperfanScore"].max()) # Divide Superfans into cohorts and assign a segment number (this nbr is mapped to name) cohorts = self._get_segment_cohorts() if len(source_df) > 1000 and hasattr(source_df, "parallel_apply"): source_df["enrSuperfanSegment"] = source_df[["enrSuperfanScore"]].parallel_apply( self._assign_segment, axis=1, args=[cohorts] ) else: source_df["enrSuperfanSegment"] = source_df[["enrSuperfanScore"]].apply( self._assign_segment, axis=1, args=[cohorts] ) source_df = self._assign_superfan_segment_names(source_df) source_df.reset_index(inplace=True) output_renaming_map = { "fanclub_member_score": "superfan_fanclub_member", "media_follows_count_score": "superfan_media_follows_count", "unique_newsletter_signups_score": "superfan_unique_newsletter_signups", "vip_status_score": "superfan_vip_status", "marketing_optin_score": "superfan_marketing_optin", "presave_status_score": "superfan_presave_status", "paid_status_score": "superfan_paid_status", "tracks_played_score": "superfan_tracks_played", "75_score": "superfan_purchase_quantity", } source_df.rename( columns=output_renaming_map, inplace=True, ) output_cols = ["fan_id", "enrSuperfanScore", "enrSuperfan", "enrSuperfanSegment"] + list( output_renaming_map.values() ) # If only one unique score is available, then we don't want to return any segments if source_df["enrSuperfanScore"].nunique() == 1: component_logger.info(f"Will not create Superfan segments because all fans have exactly the same score") # Provide empty result so we wouldn't get analytics either from fan_attribute self.result_df = pd.DataFrame({col: [] for col in output_cols}) # Don't generate segments self.run_generate_segments = False else: self.result_df = source_df[output_cols] @staticmethod def get_attribute_scores(): """ Creates a table of 7 verticals and 3 levels of intensity. The levels sub-dict values map to system labels, which come either from source data, from "collaborative data points", or from API integrations. The ids to name mapping are in commons.system_label """ # TODO! need a bit of feature engineering to get the "vip/special" merch/ticket # Instead of None we have empty string to avoid issues with df.assign (which requires keywords) action_attitude_map = { "social_media": { "L1": {"visits": ""}, "L2": {"follows": "media_follows_count", "likes": ""}, "L3": {"comments": "", "shares": ""}, }, "non_social_media": { "L1": {"visits_website": "", "opens_email": ""}, "L2": {"sign_ups": "unique_newsletter_signups", "optin": "marketing_optin"}, "L3": {"reacts_to_call_to_action": ""}, }, "pre_save": { "L1": {"visits_pre_save_landing_page": "presave_status", "subscription_type": "paid_status"}, "L2": {"song_preview": "", "click_to_service": ""}, "L3": {"intent_to_buy": ""}, }, "streams": {"L1": {"listens": ""}, "L2": {"likes": ""}, "L3": {"adds_to_playlist": "", "adds_album": ""},}, "merch": { "L1": {"visits_online_store": "", "intent_to_buy": ""}, "L2": {"buys_merch": "75"}, # We use enrPurchaseQuantity because we don't have merch purchase quantity "L3": {"buys_special": "vip_status"}, }, "concert": { "L1": {"visit_ticketing_page": ""}, "L2": {"buys_ticket": "75"}, # We use enrPurchaseQuantity because we don't have event purchase quantity "L3": {"buys_vip_ticket": ""}, }, "high_engagement": {"L1": {}, "L2": {}, "L3": {"fanclub_member": "fanclub_member"},}, } """ Create a dictionary of attributes and their scores. This implements method A as per docs """ scoring_method_a = {} for actions in action_attitude_map.values(): for item, attribute in actions["L1"].items(): scoring_method_a[attribute] = 2 for item, attribute in actions["L2"].items(): scoring_method_a[attribute] = 3 for item, attribute in actions["L3"].items(): scoring_method_a[attribute] = 4 # We haven't defined attributes for each vertical, so drop that "empty" key. # We use pop so it wouldn't fail when all attributes are present at some point. scoring_method_a.pop("", None) return scoring_method_a def _assign_superfan_segment_names(self, df): """ Getting the list of all segments and """ level_to_name_dict = { "1": "Most Intense Superfans", "2": "Superfans: Strong Engagement", "3": "Superfans: Moderate Engagement", "4": "Superfans: Low Engagement", "5": "Superfans: Weak Engagement", } df["enrSuperfan"] = df["enrSuperfanSegment"].astype(str) df.replace({"enrSuperfan": level_to_name_dict}, inplace=True) return df def _get_segment_cohorts(self): lst = range(1, self.max_superfan_score + 1, 1) cohorts = np.array_split(lst, 5) cohorts_list = [] for index, cohort in enumerate(cohorts): cohorts_list.append({"name": len(cohorts) - index, "range": cohort}) return cohorts_list @staticmethod def _assign_segment(row, cohorts_list): value = int(row[0]) if value == 0: # In case score is 0, we return the shittiest segment return 5 for cohort in cohorts_list: if value in list(cohort["range"]): return cohort["name"] return 5 @staticmethod def _get_fan_club(df): """ There are some special fields (of type text in our case) where we treat fan as part of fan club by merely having data in this column. For example, Avatar "CitizenShip". NB! Since we sum rows per fan when assigning scores, we need to make sure we count fanclub once. So for that, we first check if fanclub data exists, and then assign only one row per fan as 1. """ try: # If fanclub data is not null df.loc[~df["90"].isna(), "fanclub_exists"] = 1 # Get max row_id df["max_row_id"] = df["row_id"].groupby(df["fan_id"]).transform("max") # Create new column that conditionally equals 1, where max_row_id = row and fanclub data exists; else 0; df["fanclub_member"] = np.where((df["max_row_id"] == df["row_id"]) & (df["fanclub_exists"] == 1), 1, 0) except ValueError: # ValueError: cannot mask with array containing NA / NaN values (when only null values) df["fanclub_member"] = 0 # We don't need original attributes anymore. What's more, they pollute the scores (side-effect of df assign) df = df.drop(columns=["90", "max_row_id", "fanclub_exists"]) return df @staticmethod def _get_marketing_optin(df): """ Convert not null values for optin data to 1. In case it has string "no", assign it 0. NB! Since we sum rows per fan when assigning scores, we need to make sure we count optin once. So for that, we first check if optin data exists, and then assign only one row per fan as 1. """ try: df.loc[~df["45"].isna(), "optin_exists"] = 1 df.loc[df["45"].isna(), "optin_exists"] = 0 # Create a field which is 1 for all rows of fan in case optin data contains "no" df["no_marketing"] = df["45"].str.contains("no|FALSE|F|N|0", case=False).replace([True, False], [1, 0]) df["no_marketing"] = df["no_marketing"].groupby(df["fan_id"]).transform("sum") # Get max row_id df["max_row_id"] = df["row_id"].groupby(df["fan_id"]).transform("max") # Create new column that conditionally equals 1 where max_row_id=row, optin_exists=1, and no_marketing=0; df["marketing_optin"] = np.where( (df["max_row_id"] == df["row_id"]) & (df["optin_exists"] == 1) & (df["no_marketing"] == 0), 1, 0, ) except ValueError: # ValueError: cannot mask with array containing NA / NaN values (when only null values) df["marketing_optin"] = 0 # We don't need original attributes anymore. df = df.drop(columns=["max_row_id", "optin_exists", "no_marketing"]) return df @staticmethod def _get_media_follows_count(df): """Convert not null values for each of the attribute to 1. Do per row, we aggregate later. """ media_attribute_ids = ["87", "88", "89"] try: for attribute in media_attribute_ids: df.loc[~df[attribute].isna(), attribute] = "1" df.loc[df[attribute].isna(), attribute] = "0" if len(df) > 1000 and hasattr(df, "parallel_apply"): df[media_attribute_ids] = df[media_attribute_ids].parallel_apply(pd.to_numeric, axis=0, args=["coerce"]) else: df[media_attribute_ids] = df[media_attribute_ids].apply(lambda x: pd.to_numeric(x, errors="coerce")) df["media_follows_count"] = df["87"] + df["88"] + df["89"] except ValueError: # ValueError: cannot mask with array containing NA / NaN values (when only null values) df["media_follows_count"] = 0 # We don't need original attributes anymore. df = df.drop(columns=media_attribute_ids) return df @staticmethod def _get_unique_newsletter_signups_count(df): """Checks if string contains 'newsletter' and assigns 1 if yes. Do per row, we aggregate later. There's some duplicates in data, but we need uniques only. """ try: if df["45"].str.contains("newsletter", case=False).any(): # Count unique input column where it contains "newsletter", and join it back with original df unique_newsletters = ( df.groupby("fan_id") .apply(lambda x: pd.Series(x["45"].unique()).str.contains("newsletter", case=False).sum()) .reset_index() ) unique_newsletters.rename(columns={0: "unique_newsletters_total"}, inplace=True) df = df.merge(unique_newsletters, how="left", on="fan_id") else: df["unique_newsletters_total"] = 0 # Get max row_id df["max_row_id"] = df["row_id"].groupby(df["fan_id"]).transform("max") # Create new column that conditionally equals 1, where max_row_id = row and unique_newsletters_total; else 0; df["unique_newsletter_signups"] = np.where( (df["max_row_id"] == df["row_id"]) & (df["unique_newsletters_total"] >= 1), df["unique_newsletters_total"], 0, ) except ValueError: df["unique_newsletter_signups"] = 0 # We don't need original attributes anymore. df = df.drop(columns=["max_row_id", "unique_newsletters_total"]) return df @staticmethod def _get_vip_indicator(df): """ Checks if string contains any keywords in given regex. This is a binary feature, we need to apply score only once (so use same tricks as with fanclub. """ vip_string = "bundle|exclusive|signed|pre-order|anniversary|vip|vinyl|upgrade|limited" try: # Fill nans with blanks or the bottom loc[condition will fail df[["86"]] = df[["86"]].fillna("") # Count unique input column where it matches regex pattern, and join it back with original df if df["86"].str.contains(vip_string, case=False, regex=True).any(): vip_counts = ( df.groupby("fan_id") .apply(lambda x: pd.Series(x["86"].unique()).str.contains(vip_string, case=False, regex=True).sum()) .reset_index() ) vip_counts.rename(columns={0: "vip_total"}, inplace=True) df = df.merge(vip_counts, how="left", on="fan_id") else: df["vip_total"] = 0 # Get max row_id df["max_row_id"] = df["row_id"].groupby(df["fan_id"]).transform("max") # Create new column that conditionally equals 1, where max_row_id = row and vip total is greater than 0; df["vip_status"] = np.where((df["max_row_id"] == df["row_id"]) & (df["vip_total"] >= 1), 1, 0) except ValueError: df["vip_status"] = 0 # We don't need original attributes anymore. What's more, they pollute the scores (side-effect of df assign) df = df.drop(columns=["86", "max_row_id", "vip_total"]) return df @staticmethod def _get_presave_indicator(df): """ Checks if string contains any keywords in given regex. This is a binary feature, we need to apply score only once (so use same tricks as with fanclub. """ presave_string = "presave" try: # Fill nans with blanks or the bottom loc[condition will fail df[["100"]] = df[["100"]].fillna("") # Count unique input column where it matches regex pattern, and join it back with original df if df["100"].str.contains(presave_string, case=False, regex=True).any(): presave_counts = ( df.groupby("fan_id") .apply( lambda x: pd.Series(x["100"].unique()) .str.contains(presave_string, case=False, regex=True) .sum() ) .reset_index() ) presave_counts.rename(columns={0: "presave_total"}, inplace=True) df = df.merge(presave_counts, how="left", on="fan_id") else: df["presave_total"] = 0 # Get max row_id df["max_row_id"] = df["row_id"].groupby(df["fan_id"]).transform("max") # Create new column that conditionally equals 1, where max_row_id = row and vip total is greater than 0; df["presave_status"] = np.where((df["max_row_id"] == df["row_id"]) & (df["presave_total"] >= 1), 1, 0) except ValueError: df["presave_status"] = 0 # We don't need original attributes anymore. What's more, they pollute the scores (side-effect of df assign) df = df.drop(columns=["100", "max_row_id", "presave_total"]) return df @staticmethod def _get_paid_subscription_indicator(df): """ Checks if string contains any keywords in given regex. This is a binary feature, we need to apply score only once (so use same tricks as with fanclub. """ paid_string = "paid" try: # Fill nans with blanks or the bottom loc[condition will fail df[["101"]] = df[["101"]].fillna("") # Count unique input column where it matches regex pattern, and join it back with original df if df["101"].str.contains(paid_string, case=False, regex=True).any(): paid_counts = ( df.groupby("fan_id") .apply( lambda x: pd.Series(x["101"].unique()).str.contains(paid_string, case=False, regex=True).sum() ) .reset_index() ) paid_counts.rename(columns={0: "paid_total"}, inplace=True) df = df.merge(paid_counts, how="left", on="fan_id") else: df["paid_total"] = 0 # Get max row_id df["max_row_id"] = df["row_id"].groupby(df["fan_id"]).transform("max") # Create new column that conditionally equals 1, where max_row_id = row and vip total is greater than 0; df["paid_status"] = np.where((df["max_row_id"] == df["row_id"]) & (df["paid_total"] >= 1), 1, 0) except ValueError: df["paid_status"] = 0 # We don't need original attributes anymore. What's more, they pollute the scores (side-effect of df assign) df = df.drop(columns=["101", "max_row_id", "paid_total"]) return df # We have this here because we don't need to merge with original row (and cannot in case of multiple sources) def _generate_fan_attributes(self): """ Getting a list of attributes that are present in this enrichment""" condition = self.system_fields["a_id"].isin(self.result_attribute_ids) attributes_df = self.system_fields[condition][["a_system_name", "a_id"]] attributes_df.rename(columns={"a_system_name": "name", "a_id": "attribute_id"}, inplace=True) fan_attributes_df = get_fan_attribute_values( source_df=self.result_df, attributes_df=attributes_df, collection_id=self.enrichment_collection_id, ) self.fan_attribute = fan_attributes_df