""" Engagement summary feature creation for user clustering. """ import pandas as pd import numpy as np from typing import Dict, List, Optional, Tuple, Any from functools import reduce class EngagementFeatureCreator: """Create engagement summary features from various user activity data.""" def create_engagement_features( self, features_df: pd.DataFrame, verbose: bool = True ) -> Tuple[pd.DataFrame, Dict[str, Any]]: """ Create engagement summary features. Args: features_df: DataFrame with all base features verbose: Whether to print progress messages Returns: Tuple of (updated features_df, summary dict) """ if verbose: print(" Creating engagement summary features...") # Check available columns available_columns = set(features_df.columns) if verbose: print(f" Key columns check:") print(f" • play_frequency: {'play_frequency' in available_columns}") print(f" • share_count: {'share_count' in available_columns}") print(f" • total_downloads: {'total_downloads' in available_columns}") print(f" • model_diversity: {'model_diversity' in available_columns}") # Create engagement score features_df = self._create_engagement_score(features_df, available_columns) # Identify user segments features_df["user_segment"] = features_df.apply( self._identify_user_segment, axis=1 ) # Create activity diversity features_df = self._create_activity_diversity(features_df, available_columns) # Create content diversity score features_df = self._create_content_diversity_score(features_df) # Create creator-consumer ratio - use np.where to avoid NaN features_df["creator_consumer_ratio"] = np.where( (features_df["reaction_frequency"] * 30) > 0, features_df["total_clips_created"] / (features_df["reaction_frequency"] * 30), features_df["total_clips_created"], # If no reactions, ratio is just clips ).round(3) # Create sharing propensity features_df = self._create_sharing_propensity(features_df, available_columns) # Create download intensity features_df = self._create_download_intensity(features_df, available_columns) # Encode user segments features_df = self._encode_user_segments(features_df) # Create engagement level features_df["engagement_level"] = features_df["engagement_score"].apply( self._categorize_engagement_level ) # Create additional features features_df = self._create_additional_features(features_df, available_columns) # Create summary summary = self._create_summary(features_df) return features_df, summary def _create_engagement_score( self, features_df: pd.DataFrame, available_columns: set ) -> pd.DataFrame: """Create overall engagement score.""" engagement_components = [] # Creator activity is primary engagement_components.append(features_df["total_clips_created"] * 2.0) engagement_components.append(features_df["clip_creation_rate"] * 5.0) # API/Bot usage is fundamental engagement_components.append(features_df["total_bot_actions"] * 2.5) if "api_usage_tier" in features_df.columns: engagement_components.append(features_df["api_usage_tier"] * 10.0) # Community engagement engagement_components.append(features_df["reaction_frequency"] * 2.0) if "community_interaction_score" in features_df.columns: engagement_components.append( features_df["community_interaction_score"] * 3.0 ) # Platform commitment engagement_components.append(features_df["subscription_tier"] * 5.0) engagement_components.append(features_df["is_recent_creator"] * 20.0) # Optional engagement metrics if "play_frequency" in available_columns: engagement_components.append(features_df["play_frequency"] * 2.5) if "share_count" in available_columns: engagement_components.append(features_df["share_count"] * 2.0) if "total_downloads" in available_columns: engagement_components.append(features_df["total_downloads"] * 0.5) if "model_diversity" in available_columns: engagement_components.append(features_df["model_diversity"] * 2.0) # Subtract controversy score if available if "controversy_score" in available_columns: engagement_components.append(-features_df["controversy_score"] * 5.0) # Sum all components features_df["engagement_score"] = reduce( lambda x, y: x + y, engagement_components ).round(2) return features_df def _identify_user_segment(self, row: pd.Series) -> str: """Identify user segment based on behavior patterns.""" clips = row["total_clips_created"] reactions = row["reaction_frequency"] shares = row.get("share_count", 0) public_ratio = row.get("public_clip_ratio", 0) public_count = row.get("public_clip_count", 0) advanced_ratio = row.get("advanced_model_ratio", 0) v4p5_ratio = row.get("v4p5_ratio", 0) daily_rate = row.get("daily_generation_rate", 0) bot_actions = row["total_bot_actions"] # Pro Power Users: Use advanced models significantly if v4p5_ratio > 0.3 or advanced_ratio > 0.5: return "pro_power_user" # Music Influencers: High sharing and public content elif shares > 10 and (public_ratio > 0.5 or public_count > 50): return "music_influencer" # Super Creators: Very high creation volume elif clips >= 500 or daily_rate > 50: return "super_creator" # Automation Users: High generation, low interaction elif bot_actions > clips * 3 and reactions < 0.1 and clips > 50: return "automation_user" # Casual Creators: Free tier patterns or low volume elif daily_rate <= 20 or (clips < 100 and daily_rate < 30): if clips >= 20: return "casual_creator" else: return "casual_experimenter" # Regular Active Users: Moderate activity elif clips >= 50: return "regular_active_user" # Low activity else: return "dormant_user" def _create_activity_diversity( self, features_df: pd.DataFrame, available_columns: set ) -> pd.DataFrame: """Create activity diversity metric.""" activity_components = [ (features_df["total_clips_created"] > 0).astype(int), (features_df["reaction_frequency"] > 0).astype(int), (features_df["total_bot_actions"] > 0).astype(int), (features_df["subscription_tier"] > 0).astype(int), ] if "total_downloads" in available_columns: activity_components.append((features_df["total_downloads"] > 0).astype(int)) if "share_count" in available_columns: activity_components.append((features_df["share_count"] > 0).astype(int)) if "play_frequency" in available_columns: activity_components.append((features_df["play_frequency"] > 0).astype(int)) features_df["activity_diversity"] = reduce( lambda x, y: x + y, activity_components ) return features_df def _create_content_diversity_score( self, features_df: pd.DataFrame ) -> pd.DataFrame: """Create content diversity score.""" content_diversity_cols = [ "model_diversity", "task_diversity", "source_diversity", ] content_diversity = 0 for col in content_diversity_cols: if col in features_df.columns: content_diversity = content_diversity + features_df[col] features_df["content_diversity_score"] = content_diversity / 3 return features_df def _create_sharing_propensity( self, features_df: pd.DataFrame, available_columns: set ) -> pd.DataFrame: """Create sharing propensity metric.""" downloads_for_interactions = ( features_df["total_downloads"] if "total_downloads" in features_df.columns else 0 ) share_count_for_propensity = ( features_df["share_count"] if "share_count" in features_df.columns else 0 ) total_interactions = ( features_df["total_clips_created"] + features_df["reaction_frequency"] * 30 + downloads_for_interactions + 1 # Add 1 to prevent division by zero ) features_df["sharing_propensity"] = np.where( total_interactions > 1, # Check > 1 since we added 1 share_count_for_propensity / total_interactions, 0, ).round(4) return features_df def _create_download_intensity( self, features_df: pd.DataFrame, available_columns: set ) -> pd.DataFrame: """Create download intensity metric.""" downloads_for_intensity = ( features_df["total_downloads"] if "total_downloads" in features_df.columns else 0 ) total_activity = ( features_df["total_clips_created"] + features_df["reaction_frequency"] * 30 + 1 # Add 1 to prevent division by zero ) features_df["download_intensity"] = np.where( total_activity > 1, # Check > 1 since we added 1 downloads_for_intensity / total_activity, 0, ).round(3) return features_df def _encode_user_segments(self, features_df: pd.DataFrame) -> pd.DataFrame: """Encode user segments as numeric values.""" user_segment_mapping = { "dormant_user": 0, "casual_experimenter": 1, "casual_creator": 2, "regular_active_user": 3, "automation_user": 4, "super_creator": 5, "music_influencer": 6, "pro_power_user": 7, } features_df["user_segment_encoded"] = features_df["user_segment"].map( user_segment_mapping ) return features_df def _categorize_engagement_level(self, score: float) -> str: """Categorize engagement score into levels.""" if score >= 100: return "elite" elif score >= 50: return "high" elif score >= 20: return "medium" elif score >= 5: return "low" else: return "minimal" def _create_additional_features( self, features_df: pd.DataFrame, available_columns: set ) -> pd.DataFrame: """Create additional enrichment features.""" # Content velocity - use np.where to avoid NaN if "days_creating" in features_df.columns: features_df["content_velocity"] = np.where( features_df["days_creating"] > 0, features_df["total_clips_created"] / features_df["days_creating"], 0, ).round(3) else: features_df["content_velocity"] = features_df["clip_creation_rate"] # Engagement efficiency - use np.where to avoid NaN features_df["engagement_efficiency"] = np.where( features_df["total_clips_created"] > 0, features_df["engagement_score"] / features_df["total_clips_created"], 0, ).round(2) # Viral potential score - use np.where to avoid NaN viral_numerator = 0 if "share_count" in available_columns: viral_numerator = viral_numerator + features_df["share_count"] * 2 if "total_downloads" in available_columns: viral_numerator = viral_numerator + features_df["total_downloads"] features_df["viral_potential"] = np.where( features_df["total_clips_created"] > 0, viral_numerator / features_df["total_clips_created"], 0, ).round(2) # Platform loyalty score features_df["platform_loyalty_score"] = ( features_df["subscription_tier"] * 3 + features_df["activity_diversity"] + features_df["is_recent_creator"] * 2 + (features_df["engagement_score"] > 50).astype(int) * 2 ) # Content specialization score - use np.where to avoid NaN if ( "model_diversity" in features_df.columns and "task_diversity" in features_df.columns ): max_model_div = features_df["model_diversity"].max() max_task_div = features_df["task_diversity"].max() normalized_model_div = np.where( max_model_div > 0, features_df["model_diversity"] / max_model_div, 0 ) normalized_task_div = np.where( max_task_div > 0, features_df["task_diversity"] / max_task_div, 0 ) features_df["content_specialization"] = ( 2 - normalized_model_div - normalized_task_div ).round(3) else: features_df["content_specialization"] = 1.0 # Interaction balance - use np.where to avoid NaN consumption_base = features_df["reaction_frequency"] * 30 if "play_frequency" in available_columns: consumption_base = consumption_base + features_df["play_frequency"] * 30 features_df["interaction_balance"] = np.where( consumption_base > 0, features_df["total_clips_created"] / consumption_base, features_df[ "total_clips_created" ], # If no consumption, balance is just creation ).round(3) # Community influence score influence_components = [ features_df["reaction_frequency"] * 30 * 0.2, features_df["total_clips_created"] * 0.2, ] if "community_interaction_score" in features_df.columns: influence_components.append( features_df["community_interaction_score"] * 0.3 ) if "share_count" in available_columns: influence_components.append(features_df["share_count"] * 0.3) features_df["community_influence"] = reduce( lambda x, y: x + y, influence_components ).round(2) return features_df def _create_summary(self, features_df: pd.DataFrame) -> Dict[str, Any]: """Create summary statistics.""" segment_dist = features_df["user_segment"].value_counts() segment_summary = {} for segment, count in segment_dist.items(): segment_summary[segment] = { "count": count, "percentage": count / len(features_df) * 100, } engagement_dist = features_df["engagement_level"].value_counts() engagement_summary = {} for level, count in engagement_dist.items(): engagement_summary[level] = { "count": count, "percentage": count / len(features_df) * 100, } summary = { "user_segments": segment_summary, "engagement_levels": engagement_summary, "avg_activity_diversity": features_df["activity_diversity"].mean(), "users_with_5plus_activities": ( features_df["activity_diversity"] >= 5 ).sum(), "total_features": len(features_df.columns) - 1, # Exclude user_id } return summary