pareto-analyzer

pareto-analyzer

You are pareto-analyzer - a specialized skill for Pareto analysis to identify the vital few causes and prioritize improvement efforts.

Overview

This skill enables AI-powered Pareto analysis including:

• Basic Pareto chart creation
• Multi-level Pareto analysis
• Weighted Pareto analysis
• Before/after comparison
• Pareto by multiple dimensions
• Statistical validation
• Vital few identification
• Improvement prioritization

Capabilities

1. Basic Pareto Analysis

import pandas as pd
import numpy as np

def pareto_analysis(data: pd.DataFrame, category_col: str, value_col: str):
    """
    Perform basic Pareto analysis

    data: DataFrame with categories and values
    category_col: column name for categories
    value_col: column name for values (counts, costs, etc.)
    """
    # Aggregate by category
    summary = data.groupby(category_col)[value_col].sum().reset_index()
    summary.columns = ['category', 'value']

    # Sort descending
    summary = summary.sort_values('value', ascending=False).reset_index(drop=True)

    # Calculate percentages
    total = summary['value'].sum()
    summary['percentage'] = summary['value'] / total * 100
    summary['cumulative_value'] = summary['value'].cumsum()
    summary['cumulative_percentage'] = summary['cumulative_value'] / total * 100

    # Identify vital few (categories up to 80%)
    vital_few = summary[summary['cumulative_percentage'] <= 80]
    if len(vital_few) == 0:
        vital_few = summary.head(1)
    elif summary[summary['cumulative_percentage'] <= 80].iloc[-1]['cumulative_percentage'] < 80:
        # Add one more to cross 80%
        vital_few = summary.head(len(vital_few) + 1)

    trivial_many = summary[~summary['category'].isin(vital_few['category'])]

    return {
        "analysis": summary.to_dict('records'),
        "total_value": total,
        "vital_few": {
            "categories": vital_few['category'].tolist(),
            "count": len(vital_few),
            "value": vital_few['value'].sum(),
            "percentage": round(vital_few['value'].sum() / total * 100, 1)
        },
        "trivial_many": {
            "categories": trivial_many['category'].tolist(),
            "count": len(trivial_many),
            "value": trivial_many['value'].sum(),
            "percentage": round(trivial_many['value'].sum() / total * 100, 1)
        },
        "pareto_ratio": f"{len(vital_few)}/{len(summary)} categories cause {round(vital_few['value'].sum() / total * 100)}% of impact"
    }

2. Multi-Level Pareto

def multi_level_pareto(data: pd.DataFrame, levels: list, value_col: str):
    """
    Multi-level Pareto analysis for drilling down

    levels: list of column names for hierarchical analysis
    Example: ['department', 'defect_type', 'root_cause']
    """
    results = {}

    # Level 1 - Top level Pareto
    level1_result = pareto_analysis(data, levels[0], value_col)
    results['level_1'] = {
        'dimension': levels[0],
        'analysis': level1_result
    }

    # Subsequent levels - Pareto within top categories
    if len(levels) > 1:
        vital_categories = level1_result['vital_few']['categories']

        for level_idx in range(1, len(levels)):
            level_results = []

            for cat in vital_categories:
                filtered = data[data[levels[level_idx - 1]] == cat]

                if len(filtered) > 0:
                    sub_pareto = pareto_analysis(filtered, levels[level_idx], value_col)
                    level_results.append({
                        'parent_category': cat,
                        'analysis': sub_pareto
                    })

            results[f'level_{level_idx + 1}'] = {
                'dimension': levels[level_idx],
                'sub_analyses': level_results
            }

            # Update vital categories for next level
            vital_categories = []
            for sub in level_results:
                vital_categories.extend(sub['analysis']['vital_few']['categories'])

    return results

3. Weighted Pareto Analysis

def weighted_pareto(data: pd.DataFrame, category_col: str,
                   frequency_col: str, severity_col: str = None,
                   cost_col: str = None):
    """
    Weighted Pareto considering multiple factors

    Can weight by frequency × severity, or by actual cost
    """
    summary = data.groupby(category_col).agg({
        frequency_col: 'sum'
    }).reset_index()
    summary.columns = ['category', 'frequency']

    # Add severity weighting if provided
    if severity_col:
        severity_avg = data.groupby(category_col)[severity_col].mean().reset_index()
        severity_avg.columns = ['category', 'avg_severity']
        summary = summary.merge(severity_avg, on='category')
        summary['weighted_score'] = summary['frequency'] * summary['avg_severity']
    elif cost_col:
        cost_total = data.groupby(category_col)[cost_col].sum().reset_index()
        cost_total.columns = ['category', 'total_cost']
        summary = summary.merge(cost_total, on='category')
        summary['weighted_score'] = summary['total_cost']
    else:
        summary['weighted_score'] = summary['frequency']

    # Sort by weighted score
    summary = summary.sort_values('weighted_score', ascending=False).reset_index(drop=True)

    # Calculate cumulative
    total = summary['weighted_score'].sum()
    summary['percentage'] = summary['weighted_score'] / total * 100
    summary['cumulative_pct'] = summary['percentage'].cumsum()

    # Compare rankings
    freq_rank = summary.sort_values('frequency', ascending=False)['category'].tolist()
    weighted_rank = summary['category'].tolist()

    rank_comparison = []
    for i, cat in enumerate(weighted_rank):
        freq_position = freq_rank.index(cat) + 1
        rank_comparison.append({
            'category': cat,
            'weighted_rank': i + 1,
            'frequency_rank': freq_position,
            'rank_change': freq_position - (i + 1)
        })

    return {
        "weighted_analysis": summary.to_dict('records'),
        "rank_comparison": rank_comparison,
        "weighting_method": "severity" if severity_col else "cost" if cost_col else "frequency",
        "insight": identify_rank_changes(rank_comparison)
    }

def identify_rank_changes(comparisons):
    """Identify categories with significant rank changes"""
    movers = [c for c in comparisons if abs(c['rank_change']) >= 2]
    if movers:
        return f"{len(movers)} categories have significant rank changes when weighted"
    return "Rankings are consistent between frequency and weighted analysis"

4. Before/After Pareto Comparison

def compare_pareto_periods(before_data: pd.DataFrame, after_data: pd.DataFrame,
                          category_col: str, value_col: str):
    """
    Compare Pareto analysis between two periods
    """
    before = pareto_analysis(before_data, category_col, value_col)
    after = pareto_analysis(after_data, category_col, value_col)

    # Build comparison
    before_df = pd.DataFrame(before['analysis'])
    after_df = pd.DataFrame(after['analysis'])

    comparison = before_df.merge(
        after_df,
        on='category',
        how='outer',
        suffixes=('_before', '_after')
    )

    comparison = comparison.fillna(0)
    comparison['change'] = comparison['value_after'] - comparison['value_before']
    comparison['change_pct'] = np.where(
        comparison['value_before'] > 0,
        (comparison['change'] / comparison['value_before']) * 100,
        100 if comparison['value_after'] > 0 else 0
    )

    # Summary metrics
    total_before = before['total_value']
    total_after = after['total_value']

    # Identify improvements and deteriorations
    improved = comparison[comparison['change'] < 0].sort_values('change')
    deteriorated = comparison[comparison['change'] > 0].sort_values('change', ascending=False)

    return {
        "befor