Accelerated Computing Cudf

[
  {
    "id": "cudf-apply-udf__generic",
    "question": "Task: Row-wise apply, applymap, and column-wise UDFs that should move to vectorized operations or Numba where appropriate\nTask folder: evals/files/cudf-apply-udf/\nPrompt variant: generic\n\nUser prompt: Help me run this on the GPU\n\nUse the provided starter workspace for this task. Modify the starter file(s) under the provided `code/` directory. Run the relevant smoke or validation command from that workspace when practical, and report the changed files and validation result.",
    "expected_skill": "accelerated-computing-cudf",
    "expected_script": null,
    "files": [
      "evals/files/cudf-apply-udf/code/generate_data.py",
      "evals/files/cudf-apply-udf/code/udf_pipeline.py"
    ],
    "ground_truth": "A successful answer uses the provided cudf-apply-udf starter files, especially code/udf_pipeline.py, to migrate the pandas DataFrame workload to cuDF where supported. It replaces row-wise apply/applymap or column UDF logic with vectorized cuDF expressions, Numba-compatible GPU logic, or a narrow compatibility boundary, preserves representative pandas results, and reports validation performed or the runtime blocker.",
    "expected_behavior": []
  },
  {
    "id": "cudf-csv-etl__generic",
    "question": "Task: Basic CSV ETL pipeline \u2014 read, filter, compute columns, groupby aggregate, write to parquet\nTask folder: evals/files/cudf-csv-etl/\nPrompt variant: generic\n\nUser prompt: Help me run this on the GPU\n\nUse the provided starter workspace for this task. Modify the starter file(s) under the provided `code/` directory. Run the relevant smoke or validation command from that workspace when practical, and report the changed files and validation result.",
    "expected_skill": "accelerated-computing-cudf",
    "expected_script": null,
    "files": [
      "evals/files/cudf-csv-etl/code/etl_pipeline.py",
      "evals/files/cudf-csv-etl/code/generate_data.py"
    ],
    "ground_truth": "A successful answer uses the provided cudf-csv-etl starter files, especially code/etl_pipeline.py, to move CSV read, filtering, computed columns, groupby aggregation, and parquet output to cuDF. It preserves filter predicates, computed-column formulas, grouping keys, aggregate columns, generated data paths, output paths, and reports validation performed or the runtime blocker.",
    "expected_behavior": []
  },
  {
    "id": "cudf-groupby-agg__generic",
    "question": "Task: Complex groupby with multiple agg functions, named aggregation, and transform\nTask folder: evals/files/cudf-groupby-agg/\nPrompt variant: generic\n\nUser prompt: Help me run this on the GPU\n\nUse the provided starter workspace for this task. Modify the starter file(s) under the provided `code/` directory. Run the relevant smoke or validation command from that workspace when practical, and report the changed files and validation result.",
    "expected_skill": "accelerated-computing-cudf",
    "expected_script": null,
    "files": [
      "evals/files/cudf-groupby-agg/code/generate_data.py",
      "evals/files/cudf-groupby-agg/code/groupby_analysis.py"
    ],
    "ground_truth": "A successful answer uses the provided cudf-groupby-agg starter files, especially code/groupby_analysis.py, to run the DataFrame loading and groupby work with cuDF. It preserves grouping keys, sum, mean, std, count, nunique, named aggregation, transform semantics or a documented compatibility boundary, output column names, and reports validation performed or the runtime blocker.",
    "expected_behavior": []
  },
  {
    "id": "cudf-multi-join__generic",
    "question": "Task: Three-table join (orders, customers, products) with left/inner joins followed by aggregation\nTask folder: evals/files/cudf-multi-join/\nPrompt variant: generic\n\nUser prompt: Help me run this on the GPU\n\nUse the provided starter workspace for this task. Modify the starter file(s) under the provided `code/` directory. Run the relevant smoke or validation command from that workspace when practical, and report the changed files and validation result.",
    "expected_skill": "accelerated-computing-cudf",
    "expected_script": null,
    "files": [
      "evals/files/cudf-multi-join/code/generate_data.py",
      "evals/files/cudf-multi-join/code/multi_join.py"
    ],
    "ground_truth": "A successful answer uses the provided cudf-multi-join starter files, especially code/multi_join.py, to migrate the orders, customers, and products joins plus downstream filtering and aggregation to cuDF. It preserves left and inner join types, join keys, suffix behavior, row-count expectations, post-join filters, output schema, and reports validation performed or the runtime blocker.",
    "expected_behavior": []
  },
  {
    "id": "cudf-null-handling__generic",
    "question": "Task: DataFrame with many nulls \u2014 fillna strategies, dropna, interpolate, isna masks, conditional fills\nTask folder: evals/files/cudf-null-handling/\nPrompt variant: generic\n\nUser prompt: Help me run this on the GPU\n\nUse the provided starter workspace for this task. Modify the starter file(s) under the provided `code/` directory. Run the relevant smoke or validation command from that workspace when practical, and report the changed files and validation result.",
    "expected_skill": "accelerated-computing-cudf",
    "expected_script": null,
    "files": [
      "evals/files/cudf-null-handling/code/generate_data.py",
      "evals/files/cudf-null-handling/code/null_pipeline.py"
    ],
    "ground_truth": "A successful answer uses the provided cudf-null-handling starter files, especially code/null_pipeline.py, to move null detection, fill, drop, mask, and conditional fill logic to cuDF where supported. It preserves scalar and dictionary fill rules, subset and threshold drop rules, NA-aware boolean masks, interpolation or other compatibility boundaries, and reports validation performed or the runtime blocker.",
    "expected_behavior": []
  },
  {
    "id": "cudf-parquet-io__generic",
    "question": "Task: Read multiple parquet files, concatenate, filter, write partitioned output\nTask folder: evals/files/cudf-parquet-io/\nPrompt variant: generic\n\nUser prompt: Help me run this on the GPU\n\nUse the provided starter workspace for this task. Modify the starter file(s) under the provided `code/` directory. Run the relevant smoke or validation command from that workspace when practical, and report the changed files and validation result.",
    "expected_skill": "accelerated-computing-cudf",
    "expected_script": null,
    "files": [
      "evals/files/cudf-parquet-io/code/generate_data.py",
      "evals/files/cudf-parquet-io/code/parquet_pipeline.py"
    ],
    "ground_truth": "A successful answer uses the provided cudf-parquet-io starter files, especially code/parquet_pipeline.py, to migrate parquet reads, concatenation, filtering, column selection, dtype handling, and parquet writes to cuDF. It preserves multi-file input handling, partitioned output behavior, generated data paths, output paths, and reports validation performed or the runtime blocker.",
    "expected_behavior": []
  },
  {
    "id": "cudf-pivot-melt__generic",
    "question": "Task: Pivot table creation, melt/unpivot, stack/unstack, and cross-tabulation\nTask folder: evals/files/cudf-pivot-melt/\nPrompt variant: generic\n\nUser prompt: Help me run this on the GPU\n\nUse the provided starter workspace for this task. Modify the starter file(s) under the provided `code/` directory. Run the relevant smoke or validation command from that workspace when practical, and report the changed files and validation result.",
    "expected_skill": "accelerated-computing-cudf",
    "expected_script": null,
    "files": [
      "evals/files/cudf-pivot-melt/code/generate_data.py",
      "evals/files/cudf-pivot-melt/code/reshape_analysis.py"
    ],
    "ground_truth": "A successful answer uses the provided cudf-pivot-melt starter files, especially code/reshape_analysis.py, to move supported reshape operations such as pivot, melt, stack/unstack, or crosstab-style logic to cuDF where practical. It preserves index labels, column labels, fill values, aggregation choices, output schema, compatibility boundaries, and reports validation performed or the runtime blocker.",
    "expected_behavior": []
  },
  {
    "id": "cudf-string-ops__generic",
    "question": "Task: Text cleaning pipeline using pandas string accessor \u2014 lowercase, strip, regex extract, contains, replace\nTask folder: evals/files/cudf-string-ops/\nPrompt variant: generic\n\nUser prompt: Help me run this on the GPU\n\nUse the provided starter workspace for this task. Modify the starter file(s) under the provided `code/` directory. Run the relevant smoke or validation command from that workspace when practical, and report the changed files and validation result.",
    "expected_skill": "accelerated-computing-cudf",
    "expected_script": null,
    "files": [
      "evals/files/cudf-string-ops/code/clean_contacts.py",
      "evals/files/cudf-string-ops/code/generate_data.py"
    ],
    "ground_truth": "A successful answer uses the provided cudf-string-ops starter files, especially code/clean_contacts.py, to migrate string cleaning to cuDF string accessors for lowercase, strip, contains, replace, and extract-style operations. It preserves regex patterns, extracted columns, null handling, string dtype behavior, representative cleaned values, and reports validation performed or the runtime blocker.",
    "expected_behavior": []
  },
  {
    "id": "cudf-timeseries-resample__generic",
    "question": "Task: Timestamped sensor data with resample to hourly/daily and rolling statistics\nTask folder: evals/files/cudf-timeseries-resample/\nPrompt variant: generic\n\nUser prompt: Help me run this on the GPU\n\nUse the provided starter workspace for this task. Modify the starter file(s) under the provided `code/` directory. Run the relevant smoke or validation command from that workspace when practical, and report the changed files and validation result.",
    "expected_skill": "accelerated-computing-cudf",
    "expected_script": null,
    "files": [
      "evals/files/cudf-timeseries-resample/code/generate_data.py",
      "evals/files/cudf-timeseries-resample/code/timeseries_analysis.py"
    ],
    "ground_truth": "A successful answer uses the provided cudf-timeseries-resample starter files, especially code/timeseries_analysis.py, to run datetime parsing, timestamp ordering, bucket creation, aggregation, and rolling computations with cuDF where supported. It preserves hourly and daily grouping semantics, missing buckets, rolling window sizes, output ordering, compatibility boundaries, and reports validation performed or the runtime blocker.",
    "expected_behavior": []
  },
  {
    "id": "cudf-window-functions__generic",
    "question": "Task: Ranking, cumulative sums, rolling averages, expanding stats, and shift/lag operations\nTask folder: evals/files/cudf-window-functions/\nPrompt variant: generic\n\nUser prompt: Help me run this on the GPU\n\nUse the provided starter workspace for this task. Modify the starter file(s) under the provided `code/` directory. Run the relevant smoke or validation command from that workspace when practical, and report the changed files and validation result.",
    "expected_skill": "accelerated-computing-cudf",
    "expected_script": null,
    "files": [
      "evals/files/cudf-window-functions/code/generate_data.py",
      "evals/files/cudf-window-functions/code/window_analysis.py"
    ],
    "ground_truth": "A successful answer uses the provided cudf-window-functions starter files, especially code/window_analysis.py, to migrate ranking, cumulative operations, rolling calculations, expanding calculations, and shift/lag work to cuDF where supported. It preserves group keys, ordering columns, rank methods, window sizes, edge and null behavior, output names, and reports validation performed or the runtime blocker.",
    "expected_behavior": []
  },
  {
    "id": "source-cudf-null-fillna-semantics__generic",
    "question": "Task: Preserve pandas nullable dtype and fillna semantics while migrating to cuDF.\nTask folder: evals/files/source-cudf-null-fillna-semantics/\nPrompt variant: generic\n\nUser prompt: Help me move this DataFrame cleanup to the GPU without messing up missing values.\n\nUse the provided starter workspace for this task. Modify the starter file(s) under the provided `code/` directory. Run the relevant smoke or validation command from that workspace when practical, and report the changed files and validation result.",
    "expected_skill": "accelerated-computing-cudf",
    "expected_script": null,
    "files": [
      "evals/files/source-cudf-null-fillna-semantics/NOTICE.md",
      "evals/files/source-cudf-null-fillna-semantics/code/null_cleanup.py"
    ],
    "ground_truth": "A successful answer uses the provided source-cudf-null-fillna-semantics starter files, especially code/null_cleanup.py, to migrate the cleanup workflow to cuDF without changing missing-value meaning. It preserves nullable integer, string, category-like, mask/where, fillna, and groupby semantics without lossy sentinel conversions, includes or describes pandas-versus-cuDF parity validation, and reports validation performed or the runtime blocker.",
    "expected_behavior": []
  },
  {
    "id": "cudf-native-stream-handoff-boundary__generic",
    "question": "Task: Fix a threaded native GPU wrapper so cross-stream handoff and close/free ordering are correct.\nTask folder: evals/files/cudf-native-stream-handoff-boundary/\nPrompt variant: generic\n\nUser prompt: This threaded GPU wrapper sometimes returns stale checksums after one\nworker hands a device buffer to another. Can you make the handoff correct\nwithout blocking the whole device on every transfer, and keep cleanup safe\nfor queued GPU work?\n\nUse the provided starter workspace for this task. Modify the starter file(s) under the provided `code/` directory. Run the relevant smoke or validation command from that workspace when practical, and report the changed files and validation result.",
    "expected_skill": "accelerated-computing-cudf",
    "expected_script": null,
    "files": [
      "evals/files/cudf-native-stream-handoff-boundary/NOTICE.md",
      "evals/files/cudf-native-stream-handoff-boundary/code/run_smoke.sh",
      "evals/files/cudf-native-stream-handoff-boundary/code/threaded_handoff.cu"
    ],
    "ground_truth": "A successful answer uses the provided cudf-native-stream-handoff-boundary starter files, especially code/threaded_handoff.cu, to fix cross-thread or cross-stream GPU handoff by tying CUDA event readiness to the object dependency. It orders consumer work after producer writes, orders destruction or free after last stream use, preserves asynchronous overlap where practical, and reports compile or smoke validation performed or the runtime blocker.",
    "expected_behavior": []
  },
  {
    "id": "negative-deep-learning-training__generic",
    "question": "Task: Assess whether a PyTorch training performance issue belongs in NVIDIA GPU data science migration guidance.\nTask folder: evals/files/negative-deep-learning-training/\nPrompt variant: generic\n\nUser prompt: This PyTorch training script underutilizes my H100. Help me speed up model\ntraining on the GPU.\n\nUse the provided starter workspace for this task. Modify the starter file(s) under the provided `code/` directory. Run the relevant smoke or validation command from that workspace when practical, and report the changed files and validation result.",
    "expected_skill": null,
    "expected_script": null,
    "files": [
      "evals/files/negative-deep-learning-training/code/train.py"
    ],
    "ground_truth": "A successful answer treats the provided train.py context as a PyTorch/deep-learning training performance task rather than a cuDF migration. It keeps guidance focused on model training, data loading, batching, mixed precision, profiling, or other training-specific tactics, and only mentions cuDF as optional upstream tabular ETL when that is directly relevant.",
    "expected_behavior": []
  }
]

# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

"""Generate synthetic insurance claims data for UDF processing."""

import os
import numpy as np
import pandas as pd

SEED = 42
N_ROWS = 40_000


def generate():
    if os.path.exists("claims.csv"):
        return

    rng = np.random.default_rng(SEED)

    policy_types = ["auto", "home", "health", "life", "travel"]
    risk_levels = ["low", "medium", "high"]
    regions = ["northeast", "southeast", "midwest", "west", "pacific"]

    df = pd.DataFrame({
        "claim_id": range(N_ROWS),
        "policy_type": rng.choice(policy_types, N_ROWS),
        "risk_level": rng.choice(risk_levels, N_ROWS, p=[0.5, 0.35, 0.15]),
        "region": rng.choice(regions, N_ROWS),
        "age": rng.integers(18, 85, N_ROWS),
        "claim_amount": np.round(rng.exponential(5000, N_ROWS), 2),
        "deductible": np.round(rng.choice([250, 500, 1000, 2000, 5000], N_ROWS).astype(float), 2),
        "premium_monthly": np.round(rng.uniform(50, 800, N_ROWS), 2),
        "years_as_customer": rng.integers(0, 30, N_ROWS),
        "num_prior_claims": rng.integers(0, 10, N_ROWS),
        "credit_score": rng.integers(300, 850, N_ROWS),
        "property_value": np.round(rng.uniform(50_000, 1_000_000, N_ROWS), 2),
    })

    df.to_csv("claims.csv", index=False)
    print(f"Generated {len(df)} insurance claims -> claims.csv")


if __name__ == "__main__":
    generate()

# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

"""UDF-heavy processing pipeline on insurance claims data.

Uses apply(), applymap(), and custom functions for row-wise and
element-wise transformations on a pandas DataFrame.
"""

import numpy as np
import pandas as pd

from generate_data import generate


def load_data():
    generate()
    df = pd.read_csv("claims.csv")
    print(f"Loaded {len(df)} claims")
    return df


# --- Row-wise UDFs used with apply(axis=1) ---

def calculate_risk_score(row):
    """Complex row-wise risk scoring function."""
    base_score = 50

    # Age factor
    if row["age"] < 25:
        base_score += 15
    elif row["age"] > 65:
        base_score += 10
    else:
        base_score -= 5

    # Claims history
    base_score += row["num_prior_claims"] * 8

    # Credit score factor
    if row["credit_score"] >= 750:
        base_score -= 20
    elif row["credit_score"] >= 650:
        base_score -= 10
    elif row["credit_score"] < 550:
        base_score += 15

    # Risk level multiplier
    if row["risk_level"] == "high":
        base_score *= 1.5
    elif row["risk_level"] == "medium":
        base_score *= 1.2

    # Loyalty discount
    if row["years_as_customer"] > 10:
        base_score *= 0.85
    elif row["years_as_customer"] > 5:
        base_score *= 0.92

    return round(base_score, 2)


def calculate_payout(row):
    """Calculate adjusted payout amount based on multiple conditions."""
    amount = row["claim_amount"]
    deductible = row["deductible"]

    net = max(0, amount - deductible)

    # Cap by policy type
    caps = {"auto": 50_000, "home": 200_000, "health": 100_000,
            "life": 500_000, "travel": 10_000}
    cap = caps.get(row["policy_type"], 50_000)
    net = min(net, cap)

    # Loyalty bonus: extra 5% for long-term customers
    if row["years_as_customer"] > 15:
        net *= 1.05

    # High-risk penalty: reduce by 10%
    if row["risk_level"] == "high" and row["num_prior_claims"] > 5:
        net *= 0.90

    return round(net, 2)


def classify_claim_tier(row):
    """Classify claim into processing tier based on multiple factors."""
    amount = row["claim_amount"]
    risk = row["risk_level"]
    priors = row["num_prior_claims"]

    if amount > 20_000 or (risk == "high" and priors > 3):
        return "tier_3_manual"
    elif amount > 5_000 or (risk == "medium" and priors > 2):
        return "tier_2_review"
    else:
        return "tier_1_auto"


# --- Column-wise UDFs ---

def normalize_score(series):
    """Min-max normalize a numeric series."""
    return (series - series.min()) / (series.max() - series.min())


def winsorize(series, lower=0.05, upper=0.95):
    """Clip values at the given percentiles."""
    lo = series.quantile(lower)
    hi = series.quantile(upper)
    return series.clip(lo, hi)


# --- Element-wise UDF ---

def format_currency(val):
    """Format a numeric value as currency string."""
    if pd.isna(val):
        return "$0.00"
    return f"${val:,.2f}"


def credit_bucket(val):
    """Bucket a credit score into a category."""
    if val >= 750:
        return "excellent"
    elif val >= 700:
        return "good"
    elif val >= 650:
        return "fair"
    elif val >= 550:
        return "poor"
    else:
        return "very_poor"


def process_claims(df):
    """Apply all UDFs to the claims DataFrame."""

    # Row-wise apply (the expensive operations)
    print("Computing risk scores (row-wise apply)...")
    df["risk_score"] = df.apply(calculate_risk_score, axis=1)

    print("Computing payouts (row-wise apply)...")
    df["payout"] = df.apply(calculate_payout, axis=1)

    print("Classifying claims (row-wise apply)...")
    df["claim_tier"] = df.apply(classify_claim_tier, axis=1)

    # Column-wise UDFs
    print("Normalizing and winsorizing...")
    df["risk_score_norm"] = normalize_score(df["risk_score"])
    df["claim_amount_winsorized"] = winsorize(df["claim_amount"])
    df["premium_norm"] = normalize_score(df["premium_monthly"])

    # Element-wise apply (applymap-style via apply on columns)
    print("Formatting and bucketing...")
    df["credit_bucket"] = df["credit_score"].apply(credit_bucket)
    df["payout_formatted"] = df["payout"].apply(format_currency)

    # Element-wise on multiple numeric columns
    numeric_cols = ["claim_amount", "deductible", "premium_monthly", "property_value"]
    formatted = df[numeric_cols].applymap(format_currency)
    for col in numeric_cols:
        df[f"{col}_fmt"] = formatted[col]

    return df


def summarize(df):
    """Summarize processed claims."""
    print(f"\nProcessed {len(df)} claims")
    print(f"Risk score stats: mean={df['risk_score'].mean():.1f}, "
          f"std={df['risk_score'].std():.1f}")
    print(f"Total payouts: ${df['payout'].sum():,.2f}")

    tier_counts = df["claim_tier"].value_counts()
    print(f"\nClaim tiers:\n{tier_counts}")

    credit_dist = df["credit_bucket"].value_counts()
    print(f"\nCredit distribution:\n{credit_dist}")

    by_type = df.groupby("policy_type").agg(
        avg_risk=("risk_score", "mean"),
        total_payout=("payout", "sum"),
        claim_count=("claim_id", "count"),
    ).round(2)
    print(f"\nBy policy type:\n{by_type}")


def main():
    df = load_data()
    df = process_claims(df)
    summarize(df)


if __name__ == "__main__":
    main()

# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

"""CSV ETL pipeline: read, filter, compute, groupby, write parquet.

Reads sales.csv, filters to completed orders, adds computed columns
(revenue, discounted_revenue, age_group), runs a groupby aggregation
by region and product, and writes the summary to parquet.
"""

import numpy as np
import pandas as pd

from generate_data import generate


def load_data():
    generate()
    df = pd.read_csv("sales.csv")
    print(f"Loaded {len(df)} rows from sales.csv")
    return df


def filter_completed(df):
    """Keep only completed orders with quantity >= 2."""
    mask = (df["status"] == "completed") & (df["quantity"] >= 2)
    filtered = df[mask].copy()
    print(f"Filtered to {len(filtered)} completed orders")
    return filtered


def add_computed_columns(df):
    """Add revenue, discounted revenue, and age group columns."""
    df["revenue"] = df["quantity"] * df["unit_price"]
    df["discounted_revenue"] = df["revenue"] * (1 - df["discount_pct"])

    bins = [0, 25, 35, 50, 65, 100]
    labels = ["18-25", "26-35", "36-50", "51-65", "65+"]
    df["age_group"] = pd.cut(df["customer_age"], bins=bins, labels=labels)

    df["high_value"] = (df["discounted_revenue"] > 500).astype(int)
    print(f"Added computed columns; {df['high_value'].sum()} high-value orders")
    return df


def aggregate_by_region_product(df):
    """Groupby region + product, compute summary statistics."""
    summary = (
        df.groupby(["region", "product"])
        .agg(
            total_revenue=("revenue", "sum"),
            total_discounted=("discounted_revenue", "sum"),
            order_count=("order_id", "count"),
            avg_quantity=("quantity", "mean"),
            avg_unit_price=("unit_price", "mean"),
            high_value_count=("high_value", "sum"),
        )
        .reset_index()
    )
    summary["avg_discount_impact"] = (
        1 - summary["total_discounted"] / summary["total_revenue"]
    )
    summary = summary.sort_values("total_revenue", ascending=False)
    print(f"Aggregated into {len(summary)} region-product groups")
    return summary


def write_output(summary):
    """Write the summary to a parquet file."""
    summary.to_parquet("sales_summary.parquet", index=False)
    print("Wrote sales_summary.parquet")


def main():
    df = load_data()
    df = filter_completed(df)
    df = add_computed_columns(df)
    summary = aggregate_by_region_product(df)
    write_output(summary)

    print("\nTop 5 region-product combos by revenue:")
    print(summary.head(5).to_string(index=False))


if __name__ == "__main__":
    main()

# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

"""Generate a synthetic sales CSV for the ETL pipeline."""

import os
import numpy as np
import pandas as pd

SEED = 42
N_ROWS = 50_000


def generate():
    if os.path.exists("sales.csv"):
        return

    rng = np.random.default_rng(SEED)

    regions = ["North", "South", "East", "West"]
    products = ["Widget", "Gadget", "Doohickey", "Thingamajig", "Whatchamacallit"]
    statuses = ["completed", "pending", "returned", "cancelled"]

    df = pd.DataFrame({
        "order_id": range(N_ROWS),
        "region": rng.choice(regions, N_ROWS),
        "product": rng.choice(products, N_ROWS),
        "quantity": rng.integers(1, 50, N_ROWS),
        "unit_price": np.round(rng.uniform(5.0, 500.0, N_ROWS), 2),
        "discount_pct": np.round(rng.uniform(0.0, 0.3, N_ROWS), 3),
        "status": rng.choice(statuses, N_ROWS, p=[0.7, 0.1, 0.1, 0.1]),
        "customer_age": rng.integers(18, 80, N_ROWS),
    })

    df.to_csv("sales.csv", index=False)
    print(f"Generated {len(df)} rows -> sales.csv")


if __name__ == "__main__":
    generate()

# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

"""Generate synthetic employee performance data."""

import os
import numpy as np
import pandas as pd

SEED = 42
N_EMPLOYEES = 50_000


def generate():
    if os.path.exists("employees.csv"):
        return

    rng = np.random.default_rng(SEED)

    departments = ["Engineering", "Sales", "Marketing", "Finance", "HR", "Operations"]
    levels = ["Junior", "Mid", "Senior", "Lead", "Principal"]
    offices = ["NYC", "SF", "London", "Berlin", "Tokyo", "Sydney"]

    df = pd.DataFrame({
        "employee_id": range(N_EMPLOYEES),
        "department": rng.choice(departments, N_EMPLOYEES),
        "level": rng.choice(levels, N_EMPLOYEES, p=[0.3, 0.3, 0.2, 0.12, 0.08]),
        "office": rng.choice(offices, N_EMPLOYEES),
        "salary": np.round(rng.normal(85_000, 25_000, N_EMPLOYEES).clip(30_000, 300_000), 2),
        "bonus": np.round(rng.exponential(5_000, N_EMPLOYEES), 2),
        "performance_score": np.round(rng.normal(3.5, 0.8, N_EMPLOYEES).clip(1.0, 5.0), 2),
        "years_tenure": rng.integers(0, 25, N_EMPLOYEES),
        "projects_completed": rng.integers(0, 50, N_EMPLOYEES),
        "training_hours": np.round(rng.exponential(20, N_EMPLOYEES), 1),
    })

    df.to_csv("employees.csv", index=False)
    print(f"Generated {len(df)} employee records -> employees.csv")


if __name__ == "__main__":
    generate()

# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

"""Complex groupby aggregation and transform pipeline.

Performs department-level, multi-key groupby, named aggregation,
and transform-based feature engineering on employee data.
"""

import numpy as np
import pandas as pd

from generate_data import generate


def load_data():
    generate()
    df = pd.read_csv("employees.csv")
    print(f"Loaded {len(df)} employees")
    return df


def department_summary(df):
    """Basic department-level aggregation with multiple functions."""
    dept = df.groupby("department").agg(
        headcount=("employee_id", "count"),
        avg_salary=("salary", "mean"),
        median_salary=("salary", "median"),
        std_salary=("salary", "std"),
        total_bonus=("bonus", "sum"),
        avg_perf=("performance_score", "mean"),
        unique_levels=("level", "nunique"),
        unique_offices=("office", "nunique"),
        avg_tenure=("years_tenure", "mean"),
        total_projects=("projects_completed", "sum"),
    ).reset_index()
    dept = dept.sort_values("avg_salary", ascending=False)
    print(f"Department summary: {len(dept)} departments")
    return dept


def multi_key_aggregation(df):
    """Groupby on department + level with named aggregation."""
    result = df.groupby(["department", "level"]).agg(
        count=("employee_id", "count"),
        salary_mean=("salary", "mean"),
        salary_min=("salary", "min"),
        salary_max=("salary", "max"),
        salary_sum=("salary", "sum"),
        bonus_mean=("bonus", "mean"),
        perf_mean=("performance_score", "mean"),
        perf_std=("performance_score", "std"),
        tenure_mean=("years_tenure", "mean"),
        projects_sum=("projects_completed", "sum"),
    ).reset_index()
    result["salary_range"] = result["salary_max"] - result["salary_min"]
    print(f"Multi-key aggregation: {len(result)} groups")
    return result


def office_department_crosstab(df):
    """Three-key groupby: department + office + level."""
    cross = df.groupby(["department", "office", "level"]).agg(
        headcount=("employee_id", "count"),
        avg_salary=("salary", "mean"),
        total_training=("training_hours", "sum"),
    ).reset_index()
    print(f"Cross-tab: {len(cross)} groups")
    return cross


def add_transform_features(df):
    """Use groupby transform to add group-relative features."""
    # Department-level transforms
    df["dept_avg_salary"] = df.groupby("department")["salary"].transform("mean")
    df["dept_std_salary"] = df.groupby("department")["salary"].transform("std")
    df["salary_zscore"] = (df["salary"] - df["dept_avg_salary"]) / df["dept_std_salary"]

    # Level-level transforms
    df["level_avg_perf"] = df.groupby("level")["performance_score"].transform("mean")
    df["perf_vs_level"] = df["performance_score"] - df["level_avg_perf"]

    # Department rank by salary
    df["dept_salary_rank"] = df.groupby("department")["salary"].rank(
        method="dense", ascending=False
    )

    # Department + level cumulative count
    df["dept_level_count"] = df.groupby(["department", "level"]).cumcount() + 1

    # Percent of department total
    df["dept_salary_total"] = df.groupby("department")["salary"].transform("sum")
    df["salary_pct_of_dept"] = df["salary"] / df["dept_salary_total"]

    outlier_count = (df["salary_zscore"].abs() > 2).sum()
    print(f"Transform features added; {outlier_count} salary outliers (|z| > 2)")
    return df


def top_performers_per_dept(df):
    """Get top 5 performers per department using groupby + nlargest."""
    top = (
        df.groupby("department")
        .apply(lambda g: g.nlargest(5, "performance_score"))
        .reset_index(drop=True)
    )
    print(f"Top performers: {len(top)} rows")
    return top


def main():
    df = load_data()

    dept_summary = department_summary(df)
    multi_key = multi_key_aggregation(df)
    cross = office_department_crosstab(df)
    df_with_transforms = add_transform_features(df)
    top_perf = top_performers_per_dept(df)

    print(f"\nDepartment summary:\n{dept_summary.to_string(index=False)}")
    print(f"\nSample transformed rows:\n"
          f"{df_with_transforms[['department', 'level', 'salary', 'salary_zscore', 'perf_vs_level', 'dept_salary_rank']].head(10).to_string(index=False)}")


if __name__ == "__main__":
    main()

# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

"""Generate three related CSVs: orders, customers, products."""

import os
import numpy as np
import pandas as pd

SEED = 42
N_CUSTOMERS = 3_000
N_PRODUCTS = 200
N_ORDERS = 80_000


def generate():
    if os.path.exists("orders.csv"):
        return

    rng = np.random.default_rng(SEED)

    # --- customers ---
    tiers = ["bronze", "silver", "gold", "platinum"]
    customers = pd.DataFrame({
        "customer_id": range(N_CUSTOMERS),
        "customer_name": [f"Cust_{i:05d}" for i in range(N_CUSTOMERS)],
        "tier": rng.choice(tiers, N_CUSTOMERS, p=[0.4, 0.3, 0.2, 0.1]),
        "country": rng.choice(["US", "UK", "DE", "JP", "BR", "IN"], N_CUSTOMERS),
        "credit_limit": np.round(rng.uniform(500, 50_000, N_CUSTOMERS), 2),
    })

    # --- products ---
    categories = ["electronics", "clothing", "food", "tools", "toys"]
    products = pd.DataFrame({
        "product_id": range(N_PRODUCTS),
        "product_name": [f"Prod_{i:04d}" for i in range(N_PRODUCTS)],
        "category": rng.choice(categories, N_PRODUCTS),
        "base_price": np.round(rng.uniform(2.0, 800.0, N_PRODUCTS), 2),
        "weight_kg": np.round(rng.uniform(0.1, 30.0, N_PRODUCTS), 2),
    })

    # --- orders (some customer_ids intentionally out of range to test left join) ---
    orders = pd.DataFrame({
        "order_id": range(N_ORDERS),
        "customer_id": rng.integers(0, N_CUSTOMERS + 200, N_ORDERS),
        "product_id": rng.integers(0, N_PRODUCTS, N_ORDERS),
        "quantity": rng.integers(1, 20, N_ORDERS),
        "order_total": np.round(rng.uniform(5.0, 2000.0, N_ORDERS), 2),
        "channel": rng.choice(["web", "mobile", "store", "phone"], N_ORDERS),
    })

    customers.to_csv("customers.csv", index=False)
    products.to_csv("products.csv", index=False)
    orders.to_csv("orders.csv", index=False)
    print(f"Generated {N_CUSTOMERS} customers, {N_PRODUCTS} products, {N_ORDERS} orders")


if __name__ == "__main__":
    generate()

# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

"""Three-table join pipeline with aggregation.

Joins orders with customers (left join) and products (inner join),
then computes per-customer and per-category summaries.
"""

import numpy as np
import pandas as pd

from generate_data import generate


def load_tables():
    generate()
    orders = pd.read_csv("orders.csv")
    customers = pd.read_csv("customers.csv")
    products = pd.read_csv("products.csv")
    print(f"Loaded orders={len(orders)}, customers={len(customers)}, products={len(products)}")
    return orders, customers, products


def join_tables(orders, customers, products):
    """Left-join orders->customers, then inner-join with products."""
    # Left join: keep all orders even if customer_id is missing
    merged = orders.merge(customers, on="customer_id", how="left")
    print(f"After left join with customers: {len(merged)} rows, "
          f"{merged['customer_name'].isna().sum()} unmatched customers")

    # Inner join: drop orders whose product_id doesn't match
    merged = merged.merge(products, on="product_id", how="inner")
    print(f"After inner join with products: {len(merged)} rows")

    # Computed columns
    merged["line_total"] = merged["quantity"] * merged["base_price"]
    merged["total_weight"] = merged["quantity"] * merged["weight_kg"]
    merged["over_credit"] = (merged["order_total"] > merged["credit_limit"]).fillna(False)

    return merged


def customer_summary(merged):
    """Per-customer aggregation."""
    cust = (
        merged.groupby("customer_id")
        .agg(
            num_orders=("order_id", "count"),
            total_spent=("order_total", "sum"),
            avg_order=("order_total", "mean"),
            unique_products=("product_id", "nunique"),
            total_weight=("total_weight", "sum"),
            times_over_credit=("over_credit", "sum"),
            tier=("tier", "first"),
            country=("country", "first"),
        )
        .reset_index()
        .sort_values("total_spent", ascending=False)
    )
    print(f"Customer summary: {len(cust)} customers")
    return cust


def category_summary(merged):
    """Per-category aggregation."""
    cat = (
        merged.groupby("category")
        .agg(
            num_orders=("order_id", "count"),
            total_revenue=("line_total", "sum"),
            avg_quantity=("quantity", "mean"),
            unique_customers=("customer_id", "nunique"),
            avg_weight=("total_weight", "mean"),
        )
        .reset_index()
        .sort_values("total_revenue", ascending=False)
    )
    print(f"Category summary: {len(cat)} categories")
    return cat


def tier_channel_summary(merged):
    """Cross-tabulation of tier x channel."""
    cross = (
        merged.groupby(["tier", "channel"])
        .agg(
            order_count=("order_id", "count"),
            revenue=("line_total", "sum"),
        )
        .reset_index()
    )
    # Pivot to wide format
    pivot = cross.pivot_table(
        index="tier", columns="channel", values="revenue",
        aggfunc="sum", fill_value=0,
    )
    print(f"Tier-channel pivot:\n{pivot}")
    return cross


def main():
    orders, customers, products = load_tables()
    merged = join_tables(orders, customers, products)

    cust_summary = customer_summary(merged)
    cat_summary = category_summary(merged)
    tier_ch = tier_channel_summary(merged)

    print(f"\nTop 5 customers by spend:\n{cust_summary.head(5).to_string(index=False)}")
    print(f"\nCategory breakdown:\n{cat_summary.to_string(index=False)}")


if __name__ == "__main__":
    main()

#!/usr/bin/env bash
# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

set -euo pipefail

script_dir="$(CDPATH= cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)"
tmp_dir="$(mktemp -d "${TMPDIR:-/var/tmp}/threaded-handoff.XXXXXX")"
trap 'rm -rf "$tmp_dir"' EXIT

nvcc -std=c++17 -O2 "$script_dir/threaded_handoff.cu" -o "$tmp_dir/threaded_handoff"
"$tmp_dir/threaded_handoff"

/*
 * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 * SPDX-License-Identifier: Apache-2.0
 */

#include <cuda_runtime.h>

#include <cstdint>
#include <cstdio>
#include <cstdlib>
#include <memory>

namespace {

constexpr int kRows = 1 << 20;
constexpr int kValue = 7;
constexpr int kTrials = 32;

void check_cuda(cudaError_t status, const char* what)
{
  if (status != cudaSuccess) {
    std::fprintf(stderr, "%s failed: %s\n", what, cudaGetErrorString(status));
    std::abort();
  }
}

__global__ void fill_kernel(int* data, int rows, int value)
{
  int idx = blockIdx.x * blockDim.x + threadIdx.x;
  if (idx >= rows) return;

  int adjusted = value;
  for (int spin = 0; spin < 4096; ++spin) {
    adjusted += (spin & 1);
    adjusted -= (spin & 1);
  }
  data[idx] = adjusted;
}

__global__ void checksum_kernel(const int* data, std::uint64_t* out, int rows)
{
  int idx = blockIdx.x * blockDim.x + threadIdx.x;
  if (idx < rows) {
    atomicAdd(reinterpret_cast<unsigned long long*>(out),
              static_cast<unsigned long long>(data[idx]));
  }
}

struct NativeGpuTable {
  int* data{};
  int rows{};
  cudaStream_t producer_stream{};

  explicit NativeGpuTable(int row_count) : rows(row_count)
  {
    check_cuda(cudaStreamCreateWithFlags(&producer_stream, cudaStreamNonBlocking),
               "create producer stream");
    check_cuda(cudaMalloc(&data, sizeof(int) * rows), "allocate table data");
  }

  ~NativeGpuTable()
  {
    if (data != nullptr) {
      cudaFree(data);
    }
    if (producer_stream != nullptr) {
      cudaStreamDestroy(producer_stream);
    }
  }
};

std::shared_ptr<NativeGpuTable> build_table_async(int value)
{
  auto table = std::make_shared<NativeGpuTable>(kRows);
  int block = 256;
  int grid = (table->rows + block - 1) / block;
  fill_kernel<<<grid, block, 0, table->producer_stream>>>(table->data, table->rows, value);
  check_cuda(cudaGetLastError(), "launch fill kernel");
  return table;
}

std::uint64_t consume_on_stream(const std::shared_ptr<NativeGpuTable>& table,
                                cudaStream_t consumer_stream)
{
  std::uint64_t* d_sum{};
  std::uint64_t h_sum{};
  int block = 256;
  int grid = (table->rows + block - 1) / block;

  check_cuda(cudaMalloc(&d_sum, sizeof(std::uint64_t)), "allocate checksum");
  check_cuda(cudaMemsetAsync(d_sum, 0, sizeof(std::uint64_t), consumer_stream),
             "clear checksum");

  checksum_kernel<<<grid, block, 0, consumer_stream>>>(table->data, d_sum, table->rows);
  check_cuda(cudaGetLastError(), "launch checksum kernel");
  check_cuda(cudaMemcpyAsync(&h_sum,
                             d_sum,
                             sizeof(std::uint64_t),
                             cudaMemcpyDeviceToHost,
                             consumer_stream),
             "copy checksum");
  check_cuda(cudaStreamSynchronize(consumer_stream), "sync consumer stream");
  check_cuda(cudaFree(d_sum), "free checksum");
  return h_sum;
}

}  // namespace

int main()
{
  cudaStream_t consumer_stream{};
  check_cuda(cudaStreamCreateWithFlags(&consumer_stream, cudaStreamNonBlocking),
             "create consumer stream");

  std::uint64_t expected = static_cast<std::uint64_t>(kRows) * kValue;
  int failures = 0;

  for (int trial = 0; trial < kTrials; ++trial) {
    auto table = build_table_async(kValue);
    std::uint64_t actual = consume_on_stream(table, consumer_stream);
    if (actual != expected) {
      std::fprintf(stderr,
                   "trial %d checksum mismatch: got %llu expected %llu\n",
                   trial,
                   static_cast<unsigned long long>(actual),
                   static_cast<unsigned long long>(expected));
      ++failures;
    }
  }

  check_cuda(cudaStreamDestroy(consumer_stream), "destroy consumer stream");
  if (failures != 0) {
    std::fprintf(stderr, "%d stale handoff checks observed\n", failures);
    return 1;
  }
  std::puts("all handoffs matched expected checksum");
  return 0;
}

# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

"""Generate synthetic data with intentional null patterns."""

import os
import numpy as np
import pandas as pd

SEED = 42
N_ROWS = 40_000


def generate():
    if os.path.exists("messy_data.csv"):
        return

    rng = np.random.default_rng(SEED)

    df = pd.DataFrame({
        "id": range(N_ROWS),
        "group": rng.choice(["A", "B", "C", "D"], N_ROWS),
        "temperature": rng.normal(22.0, 3.0, N_ROWS),
        "humidity": rng.uniform(20, 90, N_ROWS),
        "pressure": rng.normal(1013, 5, N_ROWS),
        "wind_speed": rng.exponential(10, N_ROWS),
        "visibility": rng.uniform(1, 30, N_ROWS),
        "uv_index": rng.integers(0, 12, N_ROWS).astype(float),
        "air_quality": rng.choice(["good", "moderate", "poor", "hazardous"], N_ROWS),
        "station_code": rng.choice(["ST01", "ST02", "ST03", "ST04", "ST05"], N_ROWS),
    })

    # Introduce nulls with different patterns
    # Random scattered nulls (~15% each)
    for col in ["temperature", "humidity", "pressure"]:
        mask = rng.random(N_ROWS) < 0.15
        df.loc[mask, col] = np.nan

    # Block nulls (sensor offline for stretches)
    for start in [5000, 15000, 28000]:
        df.loc[start:start + 500, "wind_speed"] = np.nan
        df.loc[start:start + 300, "visibility"] = np.nan

    # Correlated nulls (uv_index missing when visibility is low)
    low_vis = df["visibility"] < 5
    df.loc[low_vis & (rng.random(N_ROWS) < 0.7), "uv_index"] = np.nan

    # String column nulls
    str_mask = rng.random(N_ROWS) < 0.10
    df.loc[str_mask, "air_quality"] = np.nan

    df["temperature"] = df["temperature"].round(2)
    df["humidity"] = df["humidity"].round(1)
    df["pressure"] = df["pressure"].round(1)
    df["wind_speed"] = df["wind_speed"].round(2)
    df["visibility"] = df["visibility"].round(1)

    df.to_csv("messy_data.csv", index=False)
    null_pcts = df.isnull().mean() * 100
    print(f"Generated {len(df)} rows with null percentages:\n{null_pcts.to_string()}")


if __name__ == "__main__":
    generate()

# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

"""Null handling pipeline: detect, fill, drop, interpolate, and report.

Demonstrates various pandas null-handling strategies on messy weather data.
"""

import numpy as np
import pandas as pd

from generate_data import generate


def load_data():
    generate()
    df = pd.read_csv("messy_data.csv")
    print(f"Loaded {len(df)} rows")
    print(f"Null counts:\n{df.isnull().sum()}")
    return df


def analyze_nulls(df):
    """Build a null analysis report."""
    null_counts = df.isnull().sum()
    null_pcts = df.isnull().mean() * 100
    report = pd.DataFrame({
        "null_count": null_counts,
        "null_pct": null_pcts.round(2),
        "dtype": df.dtypes,
    })

    # Per-group null rates
    group_nulls = df.groupby("group").apply(
        lambda g: g.isnull().sum()
    ).reset_index()
    print(f"Null report:\n{report}")
    return report, group_nulls


def fill_with_strategies(df):
    """Apply different fill strategies to different columns."""
    filled = df.copy()

    # Scalar fill
    filled["uv_index"] = filled["uv_index"].fillna(0)

    # Dict fill (different values per column)
    filled = filled.fillna({
        "air_quality": "unknown",
        "visibility": filled["visibility"].median(),
    })

    # Forward fill for block-missing wind data
    filled["wind_speed"] = filled["wind_speed"].ffill()
    # Backward fill for any remaining at the start
    filled["wind_speed"] = filled["wind_speed"].bfill()

    # Group-specific mean fill for temperature
    group_means = df.groupby("group")["temperature"].transform("mean")
    filled["temperature"] = filled["temperature"].fillna(group_means)

    # Conditional fill: humidity depends on air_quality
    quality_median = df.groupby("air_quality")["humidity"].median()
    for quality, median_val in quality_median.items():
        mask = filled["humidity"].isna() & (filled["air_quality"] == quality)
        filled.loc[mask, "humidity"] = median_val
    # Fill remaining humidity nulls with global median
    filled["humidity"] = filled["humidity"].fillna(filled["humidity"].median())

    print(f"After fills, remaining nulls:\n{filled.isnull().sum()}")
    return filled


def interpolate_pressure(df):
    """Interpolate pressure readings within each station."""
    interp_frames = []
    for station, group in df.groupby("station_code"):
        g = group.copy()
        g["pressure"] = g["pressure"].interpolate(method="linear", limit=10)
        g["pressure"] = g["pressure"].bfill().ffill()
        interp_frames.append(g)
    result = pd.concat(interp_frames, ignore_index=True)
    remaining = result["pressure"].isna().sum()
    print(f"After interpolation, {remaining} pressure nulls remain")
    return result


def dropna_analysis(df):
    """Demonstrate dropna with various parameters."""
    # Drop rows where all numeric columns are null
    numeric_cols = df.select_dtypes(include=[np.number]).columns.tolist()
    dropped_all = df.dropna(subset=numeric_cols, how="all")
    print(f"dropna(how='all') on numeric: {len(df)} -> {len(dropped_all)}")

    # Drop rows where more than 3 columns are null
    dropped_thresh = df.dropna(thresh=len(df.columns) - 3)
    print(f"dropna(thresh={len(df.columns) - 3}): {len(df)} -> {len(dropped_thresh)}")

    # Drop rows with any null in key columns
    key_cols = ["temperature", "humidity", "pressure"]
    dropped_subset = df.dropna(subset=key_cols)
    print(f"dropna(subset={key_cols}): {len(df)} -> {len(dropped_subset)}")

    return dropped_subset


def create_null_indicators(df):
    """Create boolean indicator columns for null patterns."""
    indicator_cols = ["temperature", "humidity", "pressure", "wind_speed", "uv_index"]

    for col in indicator_cols:
        df[f"{col}_missing"] = df[col].isna().astype(int)

    df["total_missing"] = df[[f"{c}_missing" for c in indicator_cols]].sum(axis=1)
    df["has_any_missing"] = (df["total_missing"] > 0).astype(int)

    # Null pattern string
    df["null_pattern"] = ""
    for col in indicator_cols:
        df["null_pattern"] = df["null_pattern"] + df[f"{col}_missing"].astype(str)

    pattern_counts = df["null_pattern"].value_counts().head(10)
    print(f"\nTop null patterns:\n{pattern_counts}")

    return df


def main():
    df = load_data()
    report, group_nulls = analyze_nulls(df)
    df_with_indicators = create_null_indicators(df)
    dropped = dropna_analysis(df)
    filled = fill_with_strategies(df)
    result = interpolate_pressure(filled)

    print(f"\nFinal null check:\n{result.isnull().sum()}")
    print(f"\nSample rows:\n{result.head(5).to_string(index=False)}")


if __name__ == "__main__":
    main()

# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

"""Generate multiple parquet files simulating partitioned log data."""

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

SEED = 42
N_PER_FILE = 10_000
N_FILES = 6


def generate():
    outdir = Path("raw_logs")
    if outdir.exists() and len(list(outdir.glob("*.parquet"))) == N_FILES:
        return

    outdir.mkdir(exist_ok=True)
    rng = np.random.default_rng(SEED)

    endpoints = ["/api/users", "/api/orders", "/api/products",
                 "/api/health", "/api/search", "/api/auth"]
    methods = ["GET", "POST", "PUT", "DELETE"]
    status_codes = [200, 201, 204, 301, 400, 401, 403, 404, 500, 502, 503]
    status_weights = [0.50, 0.10, 0.05, 0.02, 0.08, 0.05, 0.03, 0.07, 0.04, 0.03, 0.03]
    regions = ["us-east-1", "us-west-2", "eu-west-1", "ap-southeast-1"]

    for i in range(N_FILES):
        base_date = pd.Timestamp("2024-01-01") + pd.Timedelta(days=i * 5)
        timestamps = base_date + pd.to_timedelta(
            rng.integers(0, 5 * 86400, N_PER_FILE), unit="s"
        )

        df = pd.DataFrame({
            "timestamp": timestamps,
            "endpoint": rng.choice(endpoints, N_PER_FILE),
            "method": rng.choice(methods, N_PER_FILE, p=[0.6, 0.2, 0.1, 0.1]),
            "status_code": rng.choice(status_codes, N_PER_FILE, p=status_weights),
            "response_time_ms": np.round(rng.exponential(150, N_PER_FILE), 2),
            "bytes_sent": rng.integers(100, 50_000, N_PER_FILE),
            "user_id": rng.integers(1, 5_000, N_PER_FILE),
            "region": rng.choice(regions, N_PER_FILE),
            "is_cached": rng.choice([True, False], N_PER_FILE, p=[0.3, 0.7]),
        })

        fname = outdir / f"logs_batch_{i:03d}.parquet"
        df.to_parquet(fname, index=False)
        print(f"Wrote {fname} ({len(df)} rows)")

    print(f"Generated {N_FILES} parquet files in {outdir}/")


if __name__ == "__main__":
    generate()

# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

"""Parquet I/O pipeline: read multiple files, concatenate, filter, write partitioned.

Reads log data from multiple parquet files, concatenates them, applies
filters and transformations, then writes partitioned parquet output.
"""

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

from generate_data import generate


def load_all_parquet(input_dir):
    """Read all parquet files from a directory and concatenate."""
    generate()
    parquet_files = sorted(Path(input_dir).glob("*.parquet"))
    print(f"Found {len(parquet_files)} parquet files in {input_dir}")

    frames = []
    for f in parquet_files:
        df = pd.read_parquet(f)
        df["source_file"] = f.stem
        frames.append(df)

    combined = pd.concat(frames, ignore_index=True)
    print(f"Combined: {len(combined)} rows, {combined.columns.tolist()}")
    return combined


def filter_and_transform(df):
    """Apply filters and add computed columns."""
    # Filter out health check endpoints
    df = df[df["endpoint"] != "/api/health"].copy()
    print(f"After filtering health checks: {len(df)} rows")

    # Categorize status codes
    df["status_category"] = pd.cut(
        df["status_code"],
        bins=[0, 199, 299, 399, 499, 599],
        labels=["1xx", "2xx", "3xx", "4xx", "5xx"],
    )

    # Performance buckets
    df["is_slow"] = (df["response_time_ms"] > 500).astype(int)
    df["perf_bucket"] = pd.cut(
        df["response_time_ms"],
        bins=[0, 50, 200, 500, 1000, float("inf")],
        labels=["fast", "normal", "slow", "very_slow", "timeout"],
    )

    # Extract hour from timestamp
    df["hour"] = df["timestamp"].dt.hour
    df["day_of_week"] = df["timestamp"].dt.dayofweek

    return df


def compute_summaries(df):
    """Compute endpoint and region summaries."""
    endpoint_summary = df.groupby("endpoint").agg(
        request_count=("user_id", "count"),
        unique_users=("user_id", "nunique"),
        avg_response_ms=("response_time_ms", "mean"),
        p95_response_ms=("response_time_ms", lambda x: x.quantile(0.95)),
        error_count=("is_slow", "sum"),
        total_bytes=("bytes_sent", "sum"),
    ).reset_index()

    region_summary = df.groupby("region").agg(
        request_count=("user_id", "count"),
        avg_response_ms=("response_time_ms", "mean"),
        cache_hit_rate=("is_cached", "mean"),
    ).reset_index()

    print(f"Endpoint summary:\n{endpoint_summary.to_string(index=False)}")
    print(f"\nRegion summary:\n{region_summary.to_string(index=False)}")

    return endpoint_summary, region_summary


def write_partitioned(df, output_dir):
    """Write partitioned parquet output by region."""
    output_path = Path(output_dir)
    if output_path.exists():
        import shutil
        shutil.rmtree(output_path)
    output_path.mkdir(parents=True)

    # Convert categoricals to string for parquet compatibility
    for col in df.select_dtypes(include=["category"]).columns:
        df[col] = df[col].astype(str)

    for region, group in df.groupby("region"):
        region_dir = output_path / f"region={region}"
        region_dir.mkdir(exist_ok=True)
        out_file = region_dir / "data.parquet"
        group.to_parquet(out_file, index=False)
        print(f"Wrote {out_file} ({len(group)} rows)")


def write_summaries(endpoint_summary, region_summary, output_dir):
    """Write summary tables as parquet."""
    output_path = Path(output_dir)
    output_path.mkdir(parents=True, exist_ok=True)
    endpoint_summary.to_parquet(output_path / "endpoint_summary.parquet", index=False)
    region_summary.to_parquet(output_path / "region_summary.parquet", index=False)
    print(f"Wrote summary parquets to {output_path}")


def main():
    df = load_all_parquet("raw_logs")
    df = filter_and_transform(df)
    endpoint_summary, region_summary = compute_summaries(df)
    write_partitioned(df, "processed_logs")
    write_summaries(endpoint_summary, region_summary, "processed_logs/summaries")

    # Verify round-trip by reading back
    read_back = pd.read_parquet("processed_logs/summaries/endpoint_summary.parquet")
    print(f"\nRound-trip verification: {len(read_back)} endpoint summary rows read back")


if __name__ == "__main__":
    main()

# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

"""Generate synthetic retail sales data for pivot/melt operations."""

import os
import numpy as np
import pandas as pd

SEED = 42
N_ROWS = 60_000


def generate():
    if os.path.exists("retail_sales.csv"):
        return

    rng = np.random.default_rng(SEED)

    stores = [f"Store_{i:02d}" for i in range(1, 16)]
    products = ["Laptop", "Phone", "Tablet", "Headphones", "Charger",
                "Case", "Cable", "Monitor", "Keyboard", "Mouse"]
    quarters = ["Q1", "Q2", "Q3", "Q4"]
    years = [2022, 2023, 2024]
    channels = ["online", "in-store", "phone"]

    df = pd.DataFrame({
        "transaction_id": range(N_ROWS),
        "store": rng.choice(stores, N_ROWS),
        "product": rng.choice(products, N_ROWS),
        "year": rng.choice(years, N_ROWS),
        "quarter": rng.choice(quarters, N_ROWS),
        "channel": rng.choice(channels, N_ROWS, p=[0.5, 0.35, 0.15]),
        "units_sold": rng.integers(1, 20, N_ROWS),
        "revenue": np.round(rng.uniform(10, 2000, N_ROWS), 2),
        "cost": np.round(rng.uniform(5, 1500, N_ROWS), 2),
        "customer_satisfaction": rng.integers(1, 6, N_ROWS),
    })

    df["profit"] = df["revenue"] - df["cost"]

    df.to_csv("retail_sales.csv", index=False)
    print(f"Generated {len(df)} retail sales rows -> retail_sales.csv")


if __name__ == "__main__":
    generate()

# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

"""Pivot, melt, stack/unstack, and cross-tabulation on retail data.

Demonstrates various DataFrame reshape operations for sales analysis.
"""

import numpy as np
import pandas as pd

from generate_data import generate


def load_data():
    generate()
    df = pd.read_csv("retail_sales.csv")
    print(f"Loaded {len(df)} retail sales rows")
    return df


def pivot_revenue_by_product_quarter(df):
    """Pivot table: average revenue by product and quarter."""
    pivot = pd.pivot_table(
        df,
        values="revenue",
        index="product",
        columns="quarter",
        aggfunc="mean",
        fill_value=0,
    )
    pivot = pivot.round(2)
    print(f"Revenue pivot (product x quarter):\n{pivot}")
    return pivot


def pivot_multi_agg(df):
    """Pivot table with multiple aggregation functions."""
    pivot = pd.pivot_table(
        df,
        values=["revenue", "units_sold"],
        index=["store"],
        columns=["year"],
        aggfunc={"revenue": ["sum", "mean"], "units_sold": "sum"},
        fill_value=0,
    )
    print(f"Multi-agg pivot shape: {pivot.shape}")
    print(f"Columns: {pivot.columns.tolist()[:8]}...")
    return pivot


def melt_pivot_back(pivot_df):
    """Melt a pivoted DataFrame back to long format."""
    # Reset index to make product a column
    flat = pivot_df.reset_index()
    melted = pd.melt(
        flat,
        id_vars=["product"],
        var_name="quarter",
        value_name="avg_revenue",
    )
    melted = melted.sort_values(["product", "quarter"])
    print(f"Melted back to long format: {len(melted)} rows")
    return melted


def stack_unstack_demo(df):
    """Demonstrate stack and unstack operations."""
    # Create a multi-index aggregation
    agg = df.groupby(["store", "product"]).agg(
        total_revenue=("revenue", "sum"),
        total_units=("units_sold", "sum"),
    )

    # Unstack product to columns
    unstacked = agg["total_revenue"].unstack(fill_value=0)
    print(f"Unstacked shape: {unstacked.shape}")

    # Stack it back
    stacked = unstacked.stack()
    stacked.name = "total_revenue"
    stacked = stacked.reset_index()
    print(f"Re-stacked: {len(stacked)} rows")

    return unstacked, stacked


def crosstab_analysis(df):
    """Cross-tabulation of channel vs product."""
    # Count cross-tab
    ct_count = pd.crosstab(
        df["channel"],
        df["product"],
        margins=True,
        margins_name="Total",
    )
    print(f"Count crosstab:\n{ct_count}")

    # Value cross-tab (average satisfaction)
    ct_sat = pd.crosstab(
        df["channel"],
        df["product"],
        values=df["customer_satisfaction"],
        aggfunc="mean",
    ).round(2)
    print(f"\nSatisfaction crosstab:\n{ct_sat}")

    # Normalized cross-tab
    ct_norm = pd.crosstab(
        df["channel"],
        df["product"],
        normalize="index",
    ).round(4)
    print(f"\nNormalized crosstab:\n{ct_norm}")

    return ct_count, ct_sat, ct_norm


def year_over_year_pivot(df):
    """Pivot to compare year-over-year performance by store."""
    yearly = df.groupby(["store", "year"]).agg(
        revenue=("revenue", "sum"),
        units=("units_sold", "sum"),
        avg_profit=("profit", "mean"),
    ).reset_index()

    # Pivot years to columns for side-by-side comparison
    yoy = yearly.pivot_table(
        index="store",
        columns="year",
        values="revenue",
        aggfunc="sum",
        fill_value=0,
    )
    yoy.columns = [f"revenue_{y}" for y in yoy.columns]
    yoy = yoy.reset_index()

    # Compute growth rates
    if "revenue_2023" in yoy.columns and "revenue_2022" in yoy.columns:
        yoy["growth_22_23"] = (
            (yoy["revenue_2023"] - yoy["revenue_2022"]) / yoy["revenue_2022"]
        ).round(4)
    if "revenue_2024" in yoy.columns and "revenue_2023" in yoy.columns:
        yoy["growth_23_24"] = (
            (yoy["revenue_2024"] - yoy["revenue_2023"]) / yoy["revenue_2023"]
        ).round(4)

    print(f"Year-over-year:\n{yoy.head().to_string(index=False)}")
    return yoy


def main():
    df = load_data()

    # Pivot operations
    revenue_pivot = pivot_revenue_by_product_quarter(df)
    multi_pivot = pivot_multi_agg(df)

    # Melt
    melted = melt_pivot_back(revenue_pivot)

    # Stack / Unstack
    unstacked, stacked = stack_unstack_demo(df)

    # Cross-tabulation
    ct_count, ct_sat, ct_norm = crosstab_analysis(df)

    # Year-over-year pivot
    yoy = year_over_year_pivot(df)

    print(f"\nAll reshape operations completed successfully.")


if __name__ == "__main__":
    main()

# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

"""Text cleaning pipeline using pandas string operations.

Reads messy contact data and applies a series of string transformations:
lowercase, strip whitespace, regex extraction, contains checks, and replacements.
"""

import pandas as pd

from generate_data import generate


def load_data():
    generate()
    df = pd.read_csv("raw_contacts.csv")
    df["notes"] = df["notes"].fillna("")
    print(f"Loaded {len(df)} raw contacts")
    return df


def clean_names(df):
    """Normalize first and last names."""
    df["first_name"] = df["first_name"].str.strip().str.lower().str.title()
    df["last_name"] = df["last_name"].str.strip().str.lower().str.title()
    df["full_name"] = df["first_name"] + " " + df["last_name"]
    return df


def clean_emails(df):
    """Strip and lowercase emails, extract domain."""
    df["email"] = df["email"].str.strip().str.lower()
    df["email_domain"] = df["email"].str.extract(r"@([a-z0-9\.\-]+)$", expand=False)
    df["is_company_email"] = df["email_domain"].str.contains(
        r"\.(org|net)$", regex=True
    ).astype(int)
    return df


def normalize_phones(df):
    """Extract digits from phone numbers into a standard 10-digit format."""
    digits = df["phone"].str.replace(r"[^\d]", "", regex=True)
    # Remove leading '1' for 11-digit US numbers
    digits = digits.str.replace(r"^1(\d{10})$", r"\1", regex=True)
    df["phone_clean"] = (
        "(" + digits.str[:3] + ") " + digits.str[3:6] + "-" + digits.str[6:10]
    )
    return df


def parse_addresses(df):
    """Extract state and zip from address strings."""
    df["address"] = df["address"].str.strip()
    df["state"] = df["address"].str.extract(r",\s*([A-Z]{2})\s+\d{5}", expand=False)
    df["zipcode"] = df["address"].str.extract(r"(\d{5})\s*$", expand=False)
    return df


def process_notes(df):
    """Extract reference numbers, detect flags, clean up notes."""
    df["notes"] = df["notes"].str.strip()

    # Extract reference numbers like Ref#12345 or REF#99887
    df["ref_number"] = df["notes"].str.extract(
        r"[Rr][Ee][Ff]#(\d+)", expand=False
    )

    # Flag rows
    df["is_vip"] = df["notes"].str.contains("VIP", case=False, na=False).astype(int)
    df["has_bounced"] = df["notes"].str.contains("BOUNCED", case=False, na=False).astype(int)
    df["needs_followup"] = df["notes"].str.contains(
        "follow-up|pending", case=False, regex=True, na=False
    ).astype(int)

    # Redact discount details
    df["notes_redacted"] = df["notes"].str.replace(
        r"Discount:\s*\d+%", "Discount: [REDACTED]", regex=True
    )

    return df


def summarize(df):
    """Print summary statistics about the cleaned data."""
    print(f"\nCleaned {len(df)} contacts")
    print(f"  Unique domains: {df['email_domain'].nunique()}")
    print(f"  Company emails: {df['is_company_email'].sum()}")
    print(f"  VIP customers: {df['is_vip'].sum()}")
    print(f"  Bounced emails: {df['has_bounced'].sum()}")
    print(f"  With ref numbers: {df['ref_number'].notna().sum()}")
    print(f"  States found: {df['state'].nunique()}")


def main():
    df = load_data()
    df = clean_names(df)
    df = clean_emails(df)
    df = normalize_phones(df)
    df = parse_addresses(df)
    df = process_notes(df)
    summarize(df)

    df.to_csv("cleaned_contacts.csv", index=False)
    print("\nWrote cleaned_contacts.csv")


if __name__ == "__main__":
    main()