analysis

name	analysis
description	Use when full-stack data analysis pipeline — clean, detect anomalies, generate reports, and create visualizations with production pandas. Turn raw data into paid deliverables.
domain	data
author	oyi77
license	Apache-2.0
subdomain	data-engineering
tags	["analytics","data-cleaning","anomaly-detection","reporting","visualization","pandas","money-making"]
version	1.0.0

Data Analysis Pipeline — Clean, Analyze, Report, Visualize

When Not to Use

• Simple or one-off tasks — if the task is straightforward, direct execution is faster than structured methodology.
• Already established workflows — follow existing team conventions rather than introducing new frameworks.
• When automation overhead exceeds benefit — for very small scopes, the setup cost may not be justified.

Dependencies

• Python 3.8+ or Node.js 18+
• Access to relevant APIs/services for your specific use case
• Basic understanding of the domain concepts

Commands

# Refer to the skill's usage section for specific commands
# Adapt these to your workflow

Money-Making Overview

Service	Client Price	Your Time	ROI
Data cleaning & preparation	$200–$800/project	1–3 hrs	$150–$400/hr
Anomaly detection & audit	$500–$2,000/report	2–4 hrs	$250–$500/hr
Custom report generation	$300–$1,500/report	1–3 hrs	$200–$500/hr
Data visualization dashboard	$500–$3,000/dashboard	3–8 hrs	$150–$375/hr
Full pipeline (clean→analyze→report→viz)	$1,500–$5,000/engagement	4–12 hrs	$300–$500/hr

Target clients: SaaS companies, e-commerce stores, real estate agencies, marketing agencies, startup founders who need data clarity but can't afford a full-time data analyst.

Delivery model: Fixed-price per pipeline or hourly consulting at $150–$400/hr.

Combined Capabilities

Capability	Tools	Output
Data Cleaning	pandas, numpy, pyjanitor	Clean, validated dataset + cleaning report
Anomaly Detection	pandas, scipy, sklearn, PyOD	Anomaly scores + flagged records + summary
Report Generation	Jinja2, pandas, HTML/PDF templates	Branded report (PDF, HTML, Markdown)
Data Visualization	matplotlib, seaborn, plotly, altair	Static charts + interactive dashboards

Concrete Action Flow

Phase 1: Ingest & Profile (30 min)

import pandas as pd
import numpy as np
from pathlib import Path

def profile_dataset(df: pd.DataFrame) -> dict:
    """Generate a comprehensive data profile in one pass."""
    profile = {
        "rows": len(df),
        "columns": len(df.columns),
        "dtypes": df.dtypes.astype(str).to_dict(),
        "missing": df.isnull().sum()[df.isnull().sum() > 0].to_dict(),
        "missing_pct": (df.isnull().mean() * 100).round(2).to_dict(),
        "duplicates": df.duplicated().sum(),
        "numeric_stats": df.describe().to_dict(),
        "unique_counts": {c: df[c].nunique() for c in df.select_dtypes("object").columns},
    }
    return profile

# Usage
df = pd.read_csv("client_data.csv")
profile = profile_dataset(df)
print(f"{profile['rows']} rows, {profile['duplicates']} duplicates, {len(profile['missing'])} columns with nulls")

Phase 2: Clean (1–2 hrs)

def clean_dataset(df: pd.DataFrame, config: dict = None) -> pd.DataFrame:
    """
    Production cleaning pipeline.
    config keys: drop_duplicates, fill_strategy, drop_high_missing, type_coercions
    """
    result = df.copy()
    config = config or {}

    # 1. Drop duplicates
    if config.get("drop_duplicates", True):
        before = len(result)
        result = result.drop_duplicates()
        dupes = before - len(result)
        if dupes:
            print(f"Dropped {dupes} duplicate rows")

    # 2. Handle missing values
    fill_strategy = config.get("fill_strategy", {})
    for col, strategy in fill_strategy.items():
        if col not in result.columns:
            continue
        if strategy == "median" and result[col].dtype.kind in "fc":
            result[col] = result[col].fillna(result[col].median())
        elif strategy == "mode":
            result[col] = result[col].fillna(result[col].mode().iloc[0] if not result[col].mode().empty else "")
        elif strategy == "zero":
            result[col] = result[col].fillna(0)
        elif strategy == "forward":
            result[col] = result[col].ffill()
         (strategy, (, , )):
            result[col] = result[col].fillna(strategy)

    
    high_missing_pct = config.get(, )
    missing_pct = result.isnull().mean() * 
    cols_to_drop = missing_pct[missing_pct > high_missing_pct].index.tolist()
     cols_to_drop:
        result = result.drop(columns=cols_to_drop)
        ()

    
     col, dtype  config.get(, {}).items():
         col  result.columns:
            :
                result[col] = result[col].astype(dtype)
             (ValueError, TypeError):
                ()

     result


 () -> :
    
    lines = [, ]
    lines.append()
    lines.append()
    lines.append()
    lines.append()
     c  before.columns:
        m = before[c].isnull().()
         m:
            lines.append()
    lines.append()
    lines.append()
     c  after.columns:
        m = after[c].isnull().()
         m:
            lines.append()
     .join(lines)

Phase 3: Anomaly Detection (1–2 hrs)

from scipy import stats
from sklearn.ensemble import IsolationForest

def detect_outliers_iqr(df: pd.DataFrame, columns: list[str], multiplier: float = 1.5) -> pd.DataFrame:
    """
    IQR-based outlier detection. Returns a flagged copy.
    """
    result = df.copy()
    for col in columns:
        if df[col].dtype.kind not in "fc":
            continue
        Q1 = df[col].quantile(0.25)
        Q3 = df[col].quantile(0.75)
        IQR = Q3 - Q1
        lower = Q1 - multiplier * IQR
        upper = Q3 + multiplier * IQR
        result[f"{col}_outlier"] = (df[col] < lower) | (df[col] > upper)
    return result


def detect_anomalies_isolation_forest(
    df: pd.DataFrame,
    feature_cols: list[str],
    contamination: float = 0.05,
) -> pd.DataFrame:
    """
    Isolation Forest anomaly detection for multivariate patterns.
    Returns scores and flags.
    """
    model = IsolationForest(contamination=contamination, random_state=42)
    df_clean = df[feature_cols].select_dtypes(include=[np.number]).dropna()
    scores = model.fit_predict(df_clean)
    result = df.copy()
    result.loc[df_clean.index, "anomaly_score"] = model.score_samples(df_clean)
    result.loc[df_clean.index, "is_anomaly"] = scores == -1
    return result


def anomaly_summary(df: pd.DataFrame, anomaly_col:  = ) -> :
    
    total = df[anomaly_col].()  df[anomaly_col].dtype.kind    df[anomaly_col].()
    lines = [, ]
    lines.append()
     df.get()   :
        lines.append()
     .join(lines)

Phase 4: Report Generation (1 hr)

from datetime import datetime
import json

def generate_report(
    profile: dict,
    cleaning: str,
    anomaly: str,
    charts: list[str],
    title: str = "Data Analysis Report",
) -> str:
    """
    Generate a standalone HTML report from all analysis phases.
    charts: list of <img> tags or Plotly HTML divs.
    """
    sections = f"""
<!DOCTYPE html>
<html>
<head><meta charset="utf-8"><title>{title}</title>
<style>
  body {{ font-family: -apple-system, sans-serif; max-width: 960px; margin: auto; padding: 2rem; }}
  h1 {{ color: #1a1a2e; border-bottom: 3px solid #e94560; }}
  h2 {{ color: #16213e; margin-top: 2rem; }}
  .stat {{ background: #f8f9fa; padding: 1rem; border-radius: 8px; margin: 0.5rem 0; }}
  .anomaly {{ background: #fff3cd; padding: 1rem; border-left: 4px solid #ffc107; }}
  pre {{ background: #f4f4f4; padding: 1rem; overflow-x: auto; }}
  .chart {{ margin: 1.5rem 0; }}
</style></head>
<body>
<h1>{title}</h1>
<p>Generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}</p>

<h2>Dataset Overview</h2>
<div class="stat">
  <strong>Rows:</strong> {profile['rows']} |
  <strong>Columns:</strong> {profile['columns']} |
  <strong>Duplicates:</strong> {profile['duplicates']}
</div>

<h2>Data Cleaning</div>
<pre>{cleaning}</pre>

<h2>Anomaly Detection</div>
<div class="anomaly">{anomaly}</div>

<h2>Visualizations</h2>
{"".join(f'<div class="chart">{c}</div>' for c in charts)}
</body></html>"""
    return sections


def report_to_markdown() -> :
    
    lines = [
        ,
        ,
        ,
        ,
        ,
        ,
        ,
        ,
        ,
        cleaning,
        ,
        ,
        anomaly,
    ]
     .join(lines)

Phase 5: Visualization (1–2 hrs)

import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
import seaborn as sns

def plot_numeric_distributions(
    df: pd.DataFrame,
    columns: list[str],
    output_dir: str = "charts",
) -> list[str]:
    """Generate distribution plots for numeric columns, return file paths."""
    Path(output_dir).mkdir(exist_ok=True)
    paths = []
    for col in columns:
        if col not in df or df[col].dtype.kind not in "fc":
            continue
        fig, ax = plt.subplots(figsize=(10, 4))
        sns.histplot(df[col].dropna(), kde=True, ax=ax)
        ax.set_title(f"Distribution of {col}")
        plt.tight_layout()
        path = f"{output_dir}/{col}_dist.png"
        fig.savefig(path, dpi=150)
        plt.close(fig)
        paths.append(path)
    return paths


def plot_correlation_heatmap(
    df: pd.DataFrame,
    columns: list[str] = None,
    output_dir: str = "charts",
) -> str:
    """Generate a correlation heatmap, return file path."""
    numeric = df.select_dtypes(include=[np.number])
     columns:
        numeric = numeric[[c  c  columns  c  numeric.columns]]
    Path(output_dir).mkdir(exist_ok=)
    fig, ax = plt.subplots(figsize=(, ))
    sns.heatmap(numeric.corr(), annot=, fmt=, cmap=, ax=ax)
    ax.set_title()
    plt.tight_layout()
    path = 
    fig.savefig(path, dpi=)
    plt.close(fig)
     path


 () -> :
    
    Path(output_dir).mkdir(exist_ok=)
    fig, ax = plt.subplots(figsize=(, ))
    plot_df = df[[date_col, value_col]].dropna().sort_values(date_col)
    ax.plot(plot_df[date_col], plot_df[value_col], linewidth=)
    ax.set_title()
    ax.set_xlabel(date_col)
    plt.xticks(rotation=)
    plt.tight_layout()
    path = 
    fig.savefig(path, dpi=)
    plt.close(fig)
     path

Phase 6: Deliver & Bill

#!/usr/bin/env python3
"""
Full pipeline orchestration script.
Usage: python run_pipeline.py --input client_data.csv --output report.html
"""
import argparse
import pandas as pd
from pathlib import Path

def run_pipeline(input_path: str, output_path: str):
    df = pd.read_csv(input_path)
    profile = profile_dataset(df)

    # Clean
    df_clean = clean_dataset(df, {
        "drop_duplicates": True,
        "fill_strategy": {"age": "median", "category": "mode", "revenue": "zero"},
    })
    cleaning = cleaning_report(df, df_clean)

    # Anomaly detection
    numeric_cols = df_clean.select_dtypes(include=[np.number]).columns.tolist()
    df_flagged = detect_outliers_iqr(df_clean, numeric_cols)
    anomaly = anomaly_summary(df_flagged, f"{numeric_cols[0]}_outlier" if numeric_cols else None)

    # Visualize
    charts = []
    chart_paths = plot_numeric_distributions(df_clean, numeric_cols[:5])
    for p in chart_paths:
        charts.append(f'<img src="{p}" alt="distribution" style="max-width:100%">')

    # Report
    report = generate_report(profile, cleaning, anomaly, charts)
    Path(output_path).write_text(report)
    print(f"Report written to {output_path}")

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument(, required=)
    parser.add_argument(, default=)
    args = parser.parse_args()
    run_pipeline(args., args.output)

First Action in 60 Minutes

1. Find a dataset — Download any CSV (e.g., Kaggle, your own exports, or mock data via pip install faker && python -c "from faker import Faker; ...")
2. Run the profile — profile_dataset(df) and note missing values, data types, outliers
3. Clean & flag — Apply clean_dataset() and detect_outliers_iqr() with sensible config
4. Generate one chart — plot_numeric_distributions(df, df.columns[:3])
5. Produce a report — generate_report(profile, cleaning, anomaly, charts)
6. Package as a deliverable — Bundle the report HTML + charts into a zip. That's your $500 minimum viable product.

Anti-Rationalization Table

Rationalization	Reality
"I need a BI tool for this"	Pandas + matplotlib covers 90% of client needs without $10k/month tooling.
"Cleaning is not billable"	Cleaning IS the value. Clients pay for confidence in their data, not for charts.
"I will build the report template later"	Every client wants branding. Template-first saves 2 hrs per engagement. Charge for it.
"Anomaly detection needs ML"	IQR and z-score catch 80% of real-world anomalies faster than any model.
"One pipeline fits every client"	Abstract per-phase but wire per-client. Your clean_dataset() config dict is the differentiator.

Client Deliverable Checklist

• Cleaned dataset (CSV/Parquet) with cleaning_log.txt
• Anomaly-flagged dataset with score column + summary
• 3–5 distribution charts (PNG or interactive Plotly)
• Correlation heatmap
• Time-series chart (if temporal data exists)
• Branded HTML report or Markdown summary
• One-page executive brief (top 3 findings)
• Raw pipeline script for reproducibility (optional upsell)

Output Format

For every client engagement, deliver a folder:

client-name-report/
  data/
    raw.csv              # original untouched
    cleaned.csv          # cleaned dataset
    flagged.csv          # with anomaly flags
  charts/
    age_dist.png
    revenue_dist.png
    correlation.png
    revenue_timeseries.png
  report.html            # full branded report
  executive-summary.md   # 3-bullet findings
  pipeline.py            # reproducible script

Bill for the whole folder, not per-file.

Verification Checklist

• Dataset profiled (missing, duplicates, dtypes documented)
• Cleaning applied with config-driven strategy
• Anomaly detection executed with IQR or Isolation Forest
• At least one visualization generated per key metric
• Report generated in deliverable format (HTML or Markdown)
• Pipeline is reproducible from CLI
• No hardcoded paths or client data in reusable code
• Output is client-ready (no debug prints, proper branding)

When to Use

Use this skill when working with analysis.

Workflow

See the parent skill for authoritative workflow documentation.