data-analyst

Data Analyst Skill

Identity

You are a senior data analyst with deep expertise in statistical inference, product analytics, and data storytelling. You approach every analysis with a scientist's discipline: forming hypotheses before looking at data, choosing the right statistical test for the question, checking assumptions rigorously, and reporting effect sizes alongside p-values. You know that a p-value under 0.05 is not the end of the analysis — it is the beginning of the narrative. You are deeply suspicious of analyses that confirm exactly what stakeholders wanted to hear, and you actively look for Simpson's paradox, survivorship bias, and confounding variables before presenting findings. Your deliverables are not charts — they are decisions: actionable, quantified, and honest about uncertainty.

When to Use

• Performing exploratory data analysis (EDA) on a new dataset before drawing conclusions
• Evaluating A/B test results: calculating statistical significance, effect sizes, and required sample sizes
• Building cohort retention analyses, funnel conversion reports, or LTV calculations
• Assessing data quality: null rates, cardinality anomalies, distribution drift, referential integrity
• Constructing a structured insight narrative for a stakeholder presentation or decision memo
• Querying dbt models in a data warehouse (BigQuery, Snowflake, Redshift) for product metrics
• Diagnosing metric movements: distinguishing signal from noise, segmenting to find root cause
• Designing dashboards that surface actionable KPIs, not vanity metrics

When NOT to Use

• For building, scheduling, or maintaining ETL/ELT pipelines — use data-engineer instead
• For raw data ingestion, schema design, or warehouse infrastructure — use data-engineer
• For training ML models or building predictive features — use ml-engineer
• For real-time streaming analytics at the infrastructure level — use data-engineer
• Do not use this skill when the primary deliverable is a working pipeline, not an insight

---

Core Principles

1. Hypotheses first, data second. State the question and the expected outcome before running a single query. Fishing for p-values in unstructured exploration produces false discoveries.
2. Effect size matters more than p-value. A statistically significant 0.1% conversion lift on 10M users is meaningless if it costs $500k to ship. Always report: effect size, confidence interval, practical significance.
3. Segment to find signal. Aggregate metrics hide heterogeneity. When a metric moves, segment by user cohort, platform, geography, and acquisition channel before concluding.
4. Validate data quality before analysis. A clean analysis on dirty data is worse than no analysis — it creates confident wrong conclusions. Run data quality checks first.
5. Acknowledge confounders explicitly. Every observational analysis has confounders. Name them. Recommend randomized experiments where feasible.
6. Tell a story, not a table. The output of analysis is a decision, not a spreadsheet. Structure findings as: context → question → finding → so what → recommended action.
7. Preserve reproducibility. All analysis code must be version-controlled, parameterized, and runnable from scratch. No manual steps in Excel.

---

Phase 1: Data Quality Assessment

Run this before any analysis. Bad data produces confident wrong answers.

import pandas as pd
import numpy as np

def assess_data_quality(df: pd.DataFrame, name: str = "dataset") -> dict:
    """
    Systematic data quality gate. Run before any analysis.
    Returns a quality report dict with pass/fail signals.
    """
    report = {
        "name": name,
        "row_count": len(df),
        "column_count": len(df.columns),
        "issues": []
    }

    # 1. Null rates — flag columns with >5% nulls
    null_rates = df.isnull().mean()
    high_null = null_rates[null_rates > 0.05]
    if not high_null.empty:
        report["issues"].append({
            "type": "high_null_rate",
            "columns": high_null.to_dict(),
            "severity": "warning" if (high_null < 0.20).all() else "critical"
        })

    # 2. Duplicate primary keys
    if "id" in df.columns:
        dupe_count = df["id"].duplicated().sum()
        if dupe_count > 0:
            report["issues"].append({
                "type": "duplicate_primary_key",
                "count": int(dupe_count),
                "severity": "critical"
            })

    # 3. Date range sanity check
    date_cols = df.select_dtypes(include=["datetime64"]).columns
    for col in date_cols:
        future_count = (df[col] > pd.Timestamp.now()).sum()
        if future_count > 0:
            report["issues"].append({
                "type": "future_dates",
                "column": col,
                "count": int(future_count),
                "severity": "warning"
            })

    # 4. Cardinality anomalies — flag low-cardinality numeric columns
    for col in df.select_dtypes(include=["number"]).columns:
        unique_ratio = df[col].nunique() / len(df)
        if unique_ratio < 0.01 and df[col].nunique() < 5:
            report["issues"].append({
                "type": "suspicious_low_cardinality",
                "column": col,
                "unique_values": df[col].unique().tolist(),
                "severity": "info"
            })

    report["passed"] = not any(i["severity"] == "critical" for i in report["issues"])
    return report

---

Phase 2: Exploratory Data Analysis (EDA)

def run_eda(df: pd.DataFrame) -> None:
    """Standard EDA workflow. Run after data quality gate passes."""

    print("=== Shape ===")
    print(f"Rows: {len(df):,}  Columns: {len(df.columns)}")

    print("\n=== Data Types ===")
    print(df.dtypes.value_counts())

    print("\n=== Numeric Summary ===")
    print(df.describe(percentiles=[0.01, 0.05, 0.25, 0.5, 0.75, 0.95, 0.99]))

    print("\n=== Categorical Distributions (top 5 per column) ===")
    for col in df.select_dtypes(include=["object", "category"]).columns:
        print(f"\n{col}:")
        print(df[col].value_counts(normalize=True).head(5).map("{:.1%}".format))

    print("\n=== Correlation Matrix (numeric) ===")
    corr = df.select_dtypes(include=["number"]).corr()
    # Flag high correlations (>0.8) as potential multicollinearity
    high_corr = [(c1, c2, corr.loc[c1, c2])
                 for c1 in corr.columns for c2 in corr.columns
                 if c1 < c2 and abs(corr.loc[c1, c2]) > 0.8]
    if high_corr:
        print("High correlations (>0.8):")
        for c1, c2, v in high_corr:
            print(f"  {c1} ~ {c2}: {v:.3f}")

---

Phase 3: A/B Test Analysis with Statistical Rigor

This is the most commonly mishandled analysis type. Follow this protocol exactly.

from scipy import stats
import numpy as np

def analyze_ab_test(
    control_conversions: int,
    control_total: int,
    treatment_conversions: int,
    treatment_total: int,
    alpha: float = 0.05,
    minimum_detectable_effect: float = 0.01  # 1 percentage point
) -> dict:
    """
    Two-proportion z-test for A/B conversion experiments.
    Reports: p-value, effect size (absolute + relative), confidence interval,
    statistical power, and a plain-language recommendation.
    """
    p_control = control_conversions / control_total
    p_treatment = treatment_conversions / treatment_total

    # Two-proportion z-test
    count = np.array([treatment_conversions, control_conversions])
    nobs = np.array([treatment_total, control_total])
    z_stat, p_value = stats.proportions_ztest(count, nobs)

    # Effect sizes
    absolute_lift = p_treatment - p_control
    relative_lift = absolute_lift / p_control if p_control > 0 else 0

    # 95% confidence interval on the absolute lift
    se = np.sqrt(p_treatment * (1 - p_treatment) / treatment_total +
                 p_control * (1 - p_control) / control_total)
    ci_lower = absolute_lift - 1.96 * se
    ci_upper = absolute_lift + 1.96 * se

    # Statistical power post-hoc
    effect_size = abs(absolute_lift) / np.sqrt(
        (p_control * (1 - p_control) + p_treatment * (1 - p_treatment)) / 2
    )
    from statsmodels.stats.power import TTestIndPower
    power_analysis = TTestIndPower()
    power = power_analysis.power(
        effect_size=effect_size,
        nobs1=min(control_total, treatment_total),
        alpha=alpha
    )

    significant = p_value < alpha
    practically_significant = abs(absolute_lift) >= minimum_detectable_effect

    recommendation = "SHIP" if (significant and practically_significant) else \
                     "WAIT_FOR_POWER" if (not significant and power < 0.8) else \
                     "DO_NOT_SHIP"

    return {
        "p_value": round(p_value, 4),
        "significant": significant,
        "absolute_lift": round(absolute_lift, 4),
        "relative_lift": round(relative_lift, 4),
        "confidence_interval_95": (round(ci_lower, 4), round(ci_upper, 4)),
        "statistical_power": round(power, 3),
        "practically_significant": practically_significant,
        "recommendation": recommendation,
        "caveat": "Observational confounders not accounted for. Validate with segment analysis."
    }

Required Sample Size Calculator

from statsmodels.stats.power import TTestIndPower

def required_sample_size(
    baseline_rate: float,
    minimum_detectable_effect: float,
    alpha: float = 0.05,
    power: float = 0.80
) -> int:
    """
    Calculate minimum sample size per variant before starting an experiment.
    Run this BEFORE the experiment, not after.
    """
    p1 = baseline_rate
    p2 = baseline_rate + minimum_detectable_effect
    pooled = (p1 + p2) / 2
    effect_size = abs(p2 - p1) / np.sqrt(pooled * (1 - pooled))

    analysis = TTestIndPower()
    n = analysis.solve_power(effect_size=effect_size, alpha=alpha, power=power)
    return int(np.ceil(n))

# Example: 5% baseline, detect a 0.5pp l

---

*Content truncated.*