undefined

undefined

class RecruitmentFunnelAnalyzer:
    def __init__(self, recruitment_data):
        self.data = recruitment_data

    def analyze_funnel(self, position_id=None, department=None, period=None):
        """
        Analyze conversion rates at each stage of the recruitment funnel
        """
        filtered_data = self.filter_data(position_id, department, period)

        funnel = {
            'job_impressions': filtered_data['impressions'].sum(),
            'applications': filtered_data['applications'].sum(),
            'resumes_passed': filtered_data['resume_passed'].sum(),
            'first_interviews': filtered_data['first_interview'].sum(),
            'second_interviews': filtered_data['second_interview'].sum(),
            'final_interviews': filtered_data['final_interview'].sum(),
            'offers_sent': filtered_data['offers_sent'].sum(),
            'offers_accepted': filtered_data['offers_accepted'].sum(),
            'onboarded': filtered_data['onboarded'].sum(),
            'probation_passed': filtered_data['probation_passed'].sum(),
        }

        # Calculate conversion rates between stages
        stages = list(funnel.keys())
        conversion_rates = {}
        for i in range(1, len(stages)):
            if funnel[stages[i-1]] > 0:
                rate = funnel[stages[i]] / funnel[stages[i-1]] * 100
                conversion_rates[f'{stages[i-1]} -> {stages[i]}'] = round(rate, 1)

        # Calculate key metrics
        key_metrics = {
            'application_rate': self.safe_divide(funnel['applications'], funnel['job_impressions']),
            'resume_pass_rate': self.safe_divide(funnel['resumes_passed'], funnel['applications']),
            'interview_show_rate': self.safe_divide(funnel['first_interviews'], funnel['resumes_passed']),
            'offer_acceptance_rate': self.safe_divide(funnel['offers_accepted'], funnel['offers_sent']),
            'onboarding_rate': self.safe_divide(funnel['onboarded'], funnel['offers_accepted']),
            'probation_retention_rate': self.safe_divide(funnel['probation_passed'], funnel['onboarded']),
            'overall_conversion_rate': self.safe_divide(funnel['probation_passed'], funnel['applications']),
        }

        return {
            'funnel': funnel,
            'conversion_rates': conversion_rates,
            'key_metrics': key_metrics,
        }

    def calculate_recruitment_cycle(self, department=None):
        """
        Calculate average time-to-hire (in days), from job posting to candidate onboarding
        """
        filtered = self.filter_data(department=department)

        cycle_metrics = {
            'avg_time_to_hire_days': filtered['days_to_hire'].mean(),
            'median_time_to_hire_days': filtered['days_to_hire'].median(),
            'resume_screening_time': filtered['days_resume_screening'].mean(),
            'interview_process_time': filtered['days_interview_process'].mean(),
            'offer_approval_time': filtered['days_offer_approval'].mean(),
            'candidate_decision_time': filtered['days_candidate_decision'].mean(),
        }

        # Analysis by position type
        by_position_type = filtered.groupby('position_type').agg({
            'days_to_hire': ['mean', 'median', 'min', 'max']
        }).round(1)

        return {
            'overall': cycle_metrics,
            'by_position_type': by_position_type,
        }

    def channel_roi_analysis(self):
        """
        ROI analysis for each recruitment channel
        """
        channel_data = self.data.groupby('channel').agg({
            'cost': 'sum',                   # Channel cost
            'applications': 'sum',           # Number of resumes
            'offers_accepted': 'sum',        # Number of hires
            'probation_passed': 'sum',       # Passed probation
            'quality_score': 'mean',         # Candidate quality score
        }).reset_index()

        channel_data['cost_per_resume'] = (
            channel_data['cost'] / channel_data['applications']
        ).round(2)
        channel_data['cost_per_hire'] = (
            channel_data['cost'] / channel_data['offers_accepted']
        ).round(2)
        channel_data['cost_per_effective_hire'] = (
            channel_data['cost'] / channel_data['probation_passed']
        ).round(2)

        # Channel efficiency ranking
        channel_data['composite_efficiency_score'] = (
            channel_data['quality_score'] * 0.4 +
            (1 / channel_data['cost_per_hire']) * 10000 * 0.3 +
            channel_data['probation_passed'] / channel_data['offers_accepted'] * 100 * 0.3
        ).round(2)

        return channel_data.sort_values('composite_efficiency_score', ascending=False)

    def safe_divide(self, numerator, denominator):
        if denominator == 0:
            return 0
        return round(numerator / denominator * 100, 1)

    def filter_data(self, position_id=None, department=None, period=None):
        filtered = self.data.copy()
        if position_id:
            filtered = filtered[filtered['position_id'] == position_id]
        if department:
            filtered = filtered[filtered['department'] == department]
        if period:
            filtered = filtered[filtered['period'] == period]
        return filtered

undefined

undefined

undefined

undefined