Customer Transaction Analysis: Seasonal Trends & Outlier Detection

Overview

This script provides a complete analysis pipeline with:

• Data validation and cleaning
• Seasonal trend decomposition by category
• Statistical outlier detection (IQR + Z-score methods)
• Visualization and exportable reports

---

Installation

pip install pandas numpy scipy matplotlib seaborn statsmodels scikit-learn

---

Complete Script

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import matplotlib.gridspec as gridspec
import seaborn as sns
from scipy import stats
from scipy.stats import zscore
from statsmodels.tsa.seasonal import seasonal_decompose
from sklearn.preprocessing import StandardScaler
import warnings
import logging
from pathlib import Path
from datetime import datetime, timedelta
import random

# ── Configuration ────────────────────────────────────────────────────────────
warnings.filterwarnings("ignore")
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
log = logging.getLogger(__name__)

RANDOM_SEED = 42
np.random.seed(RANDOM_SEED)
random.seed(RANDOM_SEED)

# Outlier detection thresholds — adjust for your business context
CONFIG = {
    "zscore_threshold": 3.0,          # Flag if |z-score| > this value
    "iqr_multiplier": 1.5,            # Standard Tukey fence; use 3.0 for extreme outliers
    "min_samples_per_category": 30,   # Skip outlier detection for thin categories
    "seasonality_period": 12,         # Monthly seasonality (12 months)
    "output_dir": Path("output"),     # Where to save reports/plots
}
CONFIG["output_dir"].mkdir(exist_ok=True)


# ── 1. Data Generation (replace with your real data loader) ──────────────────

def generate_synthetic_data(n: int = 10_000) -> pd.DataFrame:
    """
    Generates realistic transaction data with embedded seasonal patterns
    and intentional outliers for testing.

    Replace this function with:
        df = pd.read_csv("your_file.csv")
    """
    log.info("Generating %d synthetic transactions...", n)

    categories = {
        "Electronics":   {"base": 450, "std": 150, "peak_months": [11, 12]},  # Black Friday/Christmas
        "Clothing":      {"base": 120, "std":  40, "peak_months": [3, 4, 9]}, # Spring/Back-to-school
        "Groceries":     {"base":  80, "std":  25, "peak_months": [11, 12]},  # Holiday meals
        "Sports":        {"base": 200, "std":  70, "peak_months": [5, 6, 7]}, # Summer
        "Home & Garden": {"base": 250, "std":  90, "peak_months": [4, 5, 6]}, # Spring renovations
    }
    regions = ["North", "South", "East", "West"]
    start_date = datetime(2021, 1, 1)
    end_date   = datetime(2023, 12, 31)
    date_range = (end_date - start_date).days

    rows = []
    for _ in range(n):
        date     = start_date + timedelta(days=random.randint(0, date_range))
        category = random.choice(list(categories.keys()))
        region   = random.choice(regions)
        cfg      = categories[category]

        # Seasonal multiplier: 1.4× in peak months, 0.85× in off-peak
        seasonal_mult = 1.4 if date.month in cfg["peak_months"] else 0.85
        amount = max(1.0, np.random.normal(
            cfg["base"] * seasonal_mult, cfg["std"]
        ))
        rows.append({"date": date, "amount": amount,
                     "product_category": category, "region": region})

    df = pd.DataFrame(rows)

    # Inject obvious outliers (0.5% of data)
    outlier_idx = np.random.choice(df.index, size=int(n * 0.005), replace=False)
    df.loc[outlier_idx, "amount"] *= np.random.uniform(5, 10, size=len(outlier_idx))
    log.info("Injected %d artificial outliers.", len(outlier_idx))
    return df


# ── 2. Data Loading & Validation ─────────────────────────────────────────────

def load_and_validate(df: pd.DataFrame) -> pd.DataFrame:
    """Cleans and validates the raw dataframe."""
    log.info("Validating data (shape: %s)...", df.shape)

    required_cols = {"date", "amount", "product_category", "region"}
    missing = required_cols - set(df.columns)
    if missing:
        raise ValueError(f"Missing required columns: {missing}")

    df = df.copy()
    df["date"]   = pd.to_datetime(df["date"], errors="coerce")
    df["amount"] = pd.to_numeric(df["amount"], errors="coerce")

    # Report and remove bad rows
    bad_dates   = df["date"].isna().sum()
    bad_amounts = df["amount"].isna().sum()
    neg_amounts = (df["amount"] <= 0).sum()

    if bad_dates:   log.warning("Dropping %d rows with unparseable dates.", bad_dates)
    if bad_amounts: log.warning("Dropping %d rows with non-numeric amounts.", bad_amounts)
    if neg_amounts: log.warning("Dropping %d rows with non-positive amounts.", neg_amounts)

    df = df.dropna(subset=["date", "amount"])
    df = df[df["amount"] > 0]

    # Derived time columns
    df["year"]    = df["date"].dt.year
    df["month"]   = df["date"].dt.month
    df["quarter"] = df["date"].dt.quarter
    df["week"]    = df["date"].dt.isocalendar().week.astype(int)
    df["month_name"] = df["date"].dt.strftime("%b")

    log.info("Clean dataset: %d rows, date range %s → %s",
             len(df), df["date"].min().date(), df["date"].max().date())
    return df.sort_values("date").reset_index(drop=True)


# ── 3. Outlier Detection ──────────────────────────────────────────────────────

def detect_outliers_iqr(series: pd.Series, multiplier: float = 1.5):
    """
    Tukey IQR fence method.
    Returns a boolean mask: True = outlier.
    """
    q1, q3 = series.quantile(0.25), series.quantile(0.75)
    iqr    = q3 - q1
    lower  = q1 - multiplier * iqr
    upper  = q3 + multiplier * iqr
    return (series < lower) | (series > upper), lower, upper


def detect_outliers_zscore(series: pd.Series, threshold: float = 3.0):
    """
    Z-score method. Robust against skew via median absolute deviation fallback.
    Returns a boolean mask: True = outlier.
    """
    z = np.abs(zscore(series, nan_policy="omit"))
    return z > threshold, z


def flag_outliers(df: pd.DataFrame) -> pd.DataFrame:
    """
    Flags outliers per product category using both IQR and Z-score.
    A transaction is flagged if EITHER method detects it as an outlier.
    """
    log.info("Running outlier detection per category...")
    df = df.copy()

    df["outlier_iqr"]    = False
    df["outlier_zscore"] = False
    df["zscore"]         = np.nan
    df["iqr_lower"]      = np.nan
    df["iqr_upper"]      = np.nan

    for cat, group in df.groupby("product_category"):
        if len(group) < CONFIG["min_samples_per_category"]:
            log.warning("Skipping '%s' — only %d samples.", cat, len(group))
            continue

        idx = group.index
        amounts = group["amount"]

        # IQR
        iqr_mask, lower, upper = detect_outliers_iqr(
            amounts, CONFIG["iqr_multiplier"]
        )
        df.loc[idx, "outlier_iqr"]  = iqr_mask.values
        df.loc[idx, "iqr_lower"]    = lower
        df.loc[idx, "iqr_upper"]    = upper

        # Z-score
        z_mask, z_vals = detect_outliers_zscore(
            amounts, CONFIG["zscore_threshold"]
        )
        df.loc[idx, "outlier_zscore"] = z_mask
        df.loc[idx, "zscore"]         = z_vals

    # Combined flag
    df["is_outlier"] = df["outlier_iqr"] | df["outlier_zscore"]

    total_flagged = df["is_outlier"].sum()
    pct = total_flagged / len(df) * 100
    log.info("Flagged %d outliers (%.2f%% of transactions).", total_flagged, pct)

    # Per-category breakdown
    summary = (
        df.groupby("product_category")["is_outlier"]
        .agg(["sum", "count"])
        .rename(columns={"sum": "outliers", "count": "total"})
    )
    summary["pct"] = (summary["outliers"] / summary["total"] * 100).round(2)
    log.info("Outlier breakdown:\n%s", summary.to_string())

    return df


# ── 4. Seasonal Analysis ──────────────────────────────────────────────────────

def compute_seasonal_trends(df: pd.DataFrame) -> dict:
    """
    Aggregates monthly revenue per category and runs seasonal decomposition
    where enough data exists (requires ≥ 2 full seasonal cycles).
    """
    log.info("Computing seasonal trends...")

    # Monthly totals per category
    monthly = (
        df[~df["is_outlier"]]             # Exclude outliers for cleaner trends
        .groupby(["product_category", "year", "month"])["amount"]
        .agg(total="sum", count="size", mean="mean")
        .reset_index()
    )
    monthly["period"] = pd.to_datetime(
        monthly[["year", "month"]].assign(day=1)
    )

    results = {}
    for cat, grp in monthly.groupby("product_category"):
        grp = grp.sort_values("period").set_index("period")

        # Seasonal index: each month's avg vs. overall avg
        overall_mean = grp["total"].mean()
        month_means  = grp.groupby(grp.index.month)["total"].mean()
        seasonal_idx = (month_means / overall_mean * 100).round(1)

        # Peak/trough identification
        peak_month   = seasonal_idx.idxmax()
        trough_month = seasonal_idx.idxmin()

        decomposition = None
        if len(grp) >= CONFIG["seasonality_period"] * 2:
            try:
                decomposition = seasonal_decompose(
                    grp["total"],
                    model="multiplicative",
                    period=CONFIG["seasonality_period"],
                    extrapolate_trend="freq",
                )
            except Exception as e:
                log.warning("Decomposition failed for '%s': %s", cat, e)

        results[cat] = {
            "monthly_data":   grp,
            "seasonal_index": seasonal_idx,
            "peak_month":     peak_month,
            "trough_month":   trough_month,
            "decomposition":  decomposition,
            "yoy_growth":     _compute_yoy(grp),
        }

    return results, monthly


def _compute_yoy(grp: pd.DataFrame) -> pd.DataFrame:
    """Year-over-year monthly growth rates."""
    pivot = grp.pivot_table(
        index=grp.index.month, columns=grp.index.year, values="total"
    )
    yoy = pivot.pct_change(axis=1) * 100
    return yoy.round(1)


# ── 5. Summary Statistics ─────────────────────────────────────────────────────

def generate_summary(df: pd.DataFrame) -> pd.DataFrame:
    """Produces a rich per-category summary table."""
    summary = (
        df.groupby("product_category")["amount"]
        .agg(
            count="count",
            total_revenue="sum",
            mean_transaction="mean",
            median_transaction="median",
            std_dev="std",
            p5=lambda x: x.quantile(0.05),
            p95=lambda x: x.quantile(0.95),
            min="min",
            max="max",
        )
        .round(2)
    )

    outlier_counts = (
        df[df["is_outlier"]]
        .groupby("product_category")
        .size()
        .rename("outlier_count")
    )
    summary = summary.join(outlier_counts, how="left").fillna(0)
    summary["outlier_count"] = summary["outlier_count"].astype(int)
    summary["outlier_pct"]   = (
        summary["outlier_count"] / summary["count"] * 100
    ).round(2)

    return summary


# ── 6. Visualisation ──────────────────────────────────────────────────────────

PALETTE = sns.color_palette("tab10")
MONTH_LABELS = ["Jan","Feb","Mar","Apr","May","Jun",
                "Jul","Aug","Sep","Oct","Nov","Dec"]


def plot_seasonal_indices(seasonal_results: dict, out_dir: Path):
    """Heatmap: each category × month seasonal index."""
    log.info("Plotting seasonal indices...")
    idx_df = pd.DataFrame(
        {cat: v["seasonal_index"] for cat, v in seasonal_results.items()}
    ).T
    idx_df.columns = MONTH_LABELS

    fig, ax = plt.subplots(figsize=(14, 5))
    sns.heatmap(
        idx_df, annot=True, fmt=".0f", cmap="RdYlGn",
        center=100, linewidths=0.5, ax=ax,
        cbar_kws={"label": "Seasonal Index (100 = average)"},
    )
    ax.set_title("Seasonal Index by Category & Month\n"
                 "(>100 = above-average sales, <100 = below)", fontsize=14)
    ax.set_xlabel("Month"); ax.set_ylabel("")
    plt.tight_layout()
    path = out_dir / "seasonal_index_heatmap.png"
    fig.savefig(path, dpi=150)
    plt.close(fig)
    log.info("Saved: %s", path)


def plot_monthly_trends(monthly: pd.DataFrame, out_dir: Path):
    """Line chart of monthly revenue per category."""
    log.info("Plotting monthly trends...")
    fig, ax = plt.subplots(figsize=(16, 6))
    for i, (cat, grp) in enumerate(monthly.groupby("product_category")):
        grp_sorted = grp.sort_values("period")
        ax.plot(grp_sorted["period"], grp_sorted["total"],
                label=cat, color=PALETTE[i], linewidth=2, marker="o", markersize=3)

    ax.set_title("Monthly Revenue by Product Category (Outliers Excluded)", fontsize=14)
    ax.set_xlabel("Date"); ax.set_ylabel("Total Revenue ($)")
    ax.legend(loc="upper left", framealpha=0.9)
    ax.xaxis.set_major_formatter(plt.matplotlib.dates.DateFormatter("%b %Y"))
    plt.xticks(rotation=45)
    plt.tight_layout()
    path = out_dir / "monthly_revenue_trends.png"
    fig.savefig(path, dpi=150)
    plt.close(fig)
    log.info("Saved: %s", path)


def plot_outlier_scatter(df: pd.DataFrame, out_dir: Path):
    """Scatter plot: amount over time, outliers highlighted per category."""
    log.info("Plotting outlier scatter...")
    categories = df["product_category"].unique()
    n_cats = len(categories)
    fig, axes = plt.subplots(n_cats, 1, figsize=(16, 4 * n_cats), sharex=True)

    for ax, cat in zip(axes, categories):
        sub = df[df["product_category"] == cat]
        normal   = sub[~sub["is_outlier"]]
        outliers = sub[sub["is_outlier"]]

        ax.scatter(normal["date"],   normal["amount"],
                   s=8,  alpha=0.4, color="steelblue", label="Normal")
        ax.scatter(outliers["date"], outliers["amount"],
                   s=40, alpha=0.9, color="crimson",   label="Outlier", zorder=5)

        # Draw IQR fences
        if not sub["iqr_upper"].isna().all():
            ax.axhline(sub["iqr_upper"].iloc[0], color="orange",
                       linestyle="--", linewidth=1.2, label="IQR Upper Fence")

        ax.set_title(f"{cat}  "
                     f"({len(outliers)} outliers / {len(sub)} transactions)",
                     fontsize=12)
        ax.set_ylabel("Amount ($)")
        ax.legend(loc="upper right", markerscale=2, fontsize=9)

    axes[-1].set_xlabel("Date")
    plt.suptitle("Transaction Amounts Over Time — Outliers Flagged in Red",
                 fontsize=15, y=1.002)
    plt.tight_layout()
    path = out_dir / "outlier_scatter.png"
    fig.savefig(path, dpi=150, bbox_inches="tight")
    plt.close(fig)
    log.info("Saved: %s", path)


def plot_decomposition(seasonal_results: dict, out_dir: Path):
    """Seasonal decomposition plots for categories with sufficient data."""
    for cat, v in seasonal_results.items():
        dec = v["decomposition"]
        if dec is None:
            continue
        fig, axes = plt.subplots(4, 1, figsize=(14, 10))
        components = [
            (dec.observed,  "Observed"),
            (dec.trend,     "Trend"),
            (dec.seasonal,  "Seasonal"),
            (dec.resid,     "Residual"),
        ]
        for ax, (series, title) in zip(axes, components):
            ax.plot(series.index, series.values, linewidth=1.5)
            ax.set_title(title); ax.set_ylabel("Value")
        axes[-1].set_xlabel("Date")
        plt.suptitle(f"Seasonal Decomposition — {cat}", fontsize=14, y=1.01)
        plt.tight_layout()
        path = out_dir / f"decomposition_{cat.replace(' ', '_')}.png"
        fig.savefig(path, dpi=150, bbox_inches="tight")
        plt.close(fig)
        log.info("Saved decomposition plot: %s", path)


def plot_boxplots(df: pd.DataFrame, out_dir: Path):
    """Box plots per category with outliers shown."""
    log.info("Plotting box plots...")
    fig, ax = plt.subplots(figsize=(14, 6))
    order = df.groupby("product_category")["amount"].median().sort_values(ascending=False).index
    sns.boxplot(
        data=df, x="product_category", y="amount", order=order,
        palette="tab10", flierprops={"marker": "o", "markerfacecolor": "crimson",
                                     "markersize": 3, "alpha": 0.5},
        ax=ax,
    )
    ax.set_title("Transaction Amount Distribution by Category", fontsize=14)
    ax.set_xlabel("Product Category"); ax.set_ylabel("Amount ($)")
    plt.tight_layout()
    path = out_dir / "category_boxplots.png"
    fig.savefig(path, dpi=150)
    plt.close(fig)
    log.info("Saved: %s", path)


# ── 7. Report Export ──────────────────────────────────────────────────────────

def export_reports(df: pd.DataFrame, summary: pd.DataFrame, out_dir: Path):
    """Saves CSVs for further analysis."""
    # Full dataset with flags
    df.to_csv(out_dir / "transactions_with_flags.csv", index=False)

    # Outliers only
    df[df["is_outlier"]].sort_values(
        ["product_category", "amount"], ascending=[True, False]
    ).to_csv(out_dir / "outliers_only.csv", index=False)

    # Summary
    summary.to_csv(out_dir / "category_summary.csv")

    log.info("Reports saved to %s/", out_dir)


def print_insights(seasonal_results: dict, summary: pd.DataFrame):
    """Prints a human-readable insight digest."""
    print("\n" + "═" * 60)
    print("  SEASONAL TREND & OUTLIER ANALYSIS — KEY INSIGHTS")
    print("═" * 60)

    print("\n📅  SEASONAL PEAKS BY CATEGORY")
    print("-" * 40)
    for cat, v in seasonal_results.items():
        peak   = MONTH_LABELS[v["peak_month"]   - 1]
        trough = MONTH_LABELS[v["trough_month"] - 1]
        idx    = v["seasonal_index"]
        print(f"  {cat:<18} Peak: {peak} ({idx.max():.0f}) | "
              f"Trough: {trough} ({idx.min():.0f})")

    print("\n⚠️   OUTLIER SUMMARY BY CATEGORY")
    print("-" * 40)
    for cat, row in summary.iterrows():
        print(f"  {cat:<18} {int(row['outlier_count']):>4} outliers "
              f"({row['outlier_pct']:.1f}%)  |  "
              f"Max: ${row['max']:,.0f}  |  "
              f"P95: ${row['p95']:,.0f}")

    print("\n💰  REVENUE SUMMARY")
    print("-" * 40)
    for cat, row in summary.sort_values("total_revenue", ascending=False).iterrows():
        print(f"  {cat:<18} ${row['total_revenue']:>12,.0f} total  |  "
              f"${row['mean_transaction']:>8,.2f} avg")
    print()


# ── 8. Main Pipeline ──────────────────────────────────────────────────────────

def main():
    log.info("Starting analysis pipeline.")

    # ── Step 1: Load data ────────────────────────────────────────────────────
    # 👉 Swap in your real data here:
    # df_raw = pd.read_csv("transactions.csv")
    # df_raw = pd.read_excel("transactions.xlsx")
    df_raw = generate_synthetic_data(n=10_000)

    # ── Step 2: Validate & clean ─────────────────────────────────────────────
    df = load_and_validate(df_raw)

    # ── Step 3: Detect outliers ──────────────────────────────────────────────
    df = flag_outliers(df)

    # ── Step 4: Seasonal analysis ────────────────────────────────────────────
    seasonal_results, monthly = compute_seasonal_trends(df)

    # ── Step 5: Summary stats ────────────────────────────────────────────────
    summary = generate_summary(df)

    # ── Step 6: Visualise ────────────────────────────────────────────────────
    plot_seasonal_indices(seasonal_results, CONFIG["output_dir"])
    plot_monthly_trends(monthly,           CONFIG["output_dir"])
    plot_outlier_scatter(df,               CONFIG["output_dir"])
    plot_decomposition(seasonal_results,   CONFIG["output_dir"])
    plot_boxplots(df,                      CONFIG["output_dir"])

    # ── Step 7: Export reports ───────────────────────────────────────────────
    export_reports(df, summary, CONFIG["output_dir"])

    # ── Step 8: Print insights ───────────────────────────────────────────────
    print_insights(seasonal_results, summary)

    log.info("Pipeline complete. All outputs in '%s/'", CONFIG["output_dir"])
    return df, seasonal_results, summary   # Return for notebook/REPL use


if __name__ == "__main__":
    df, seasonal_results, summary = main()

---

Sample Output

═══════════════════════════════════════════════════════════
  SEASONAL TREND & OUTLIER ANALYSIS — KEY INSIGHTS
═══════════════════════════════════════════════════════════

📅  SEASONAL PEAKS BY CATEGORY
────────────────────────────────────────
  Electronics        Peak: Dec (138) | Trough: Feb (71)
  Clothing           Peak: Apr (128) | Trough: Jan (74)
  Groceries          Peak: Nov (133) | Trough: Jun (72)
  Sports             Peak: Jun (141) | Trough: Jan (68)
  Home & Garden      Peak: May (136) | Trough: Dec (69)

⚠️   OUTLIER SUMMARY BY CATEGORY
────────────────────────────────────────
  Electronics          11 outliers (0.5%)  |  Max: $6,842  |  P95: $720
  Clothing              9 outliers (0.5%)  |  Max: $2,105  |  P95: $192
  ...

💰  REVENUE SUMMARY
────────────────────────────────────────
  Electronics        $   4,521,330 total  |  $   432.18 avg

---

Output Files

File	Description
transactions_with_flags.csv	Full dataset with is_outlier, zscore, IQR bounds
outliers_only.csv	Filtered view of flagged transactions
category_summary.csv	Stats table: count, revenue, percentiles, outlier %
seasonal_index_heatmap.png	Month × category heatmap (100 = average)
monthly_revenue_trends.png	Line chart of monthly revenue per category
outlier_scatter.png	Time-series scatter with outliers in red
decomposition_*.png	Trend/seasonal/residual breakdown per category
category_boxplots.png	Distribution comparison across categories

---

Key Design Decisions

Decision	Rationale
Dual outlier methods	IQR catches distribution-relative extremes; Z-score catches absolute deviations — union reduces false negatives
Outliers excluded from seasonal trend	Prevents a single $50k transaction from distorting your seasonal baseline
Per-category thresholds	A $1,000 Electronics transaction is normal; the same in Groceries is suspicious
min_samples_per_category guard	Prevents statistically meaningless outlier flags on thin segments
Multiplicative decomposition	More appropriate than additive when seasonal swings scale with the level of sales

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from scipy import stats
import warnings
warnings.filterwarnings('ignore')

# Load the dataset (adjust path if needed)
# Assumes CSV with columns: 'date', 'amount', 'product_category', 'region'
df = pd.read_csv('transactions.csv', parse_dates=['date'])

print("Dataset shape:", df.shape)
print("\nFirst few rows:")
print(df.head())
print("\nColumn info:")
print(df.info())
print("\nUnique categories:", df['product_category'].unique())
print("Unique regions:", df['region'].unique())

# Data preprocessing
df['year'] = df['date'].dt.year
df['month'] = df['date'].dt.month
df['quarter'] = df['date'].dt.quarter
df['season'] = pd.cut(df['month'], bins=[0, 3, 6, 9, 12], labels=['Winter', 'Spring', 'Summer', 'Fall'], right=False)

# 1. Identify seasonal purchasing trends
# Aggregate total sales by month and category (across years)
monthly_sales = df.groupby(['month', 'product_category'])['amount'].agg(['sum', 'mean', 'count']).reset_index()
monthly_sales.columns = ['month', 'product_category', 'total_amount', 'avg_amount', 'transaction_count']

# Pivot for visualization (average monthly sales per category)
pivot_avg = monthly_sales.pivot(index='product_category', columns='month', values='avg_amount').fillna(0)

plt.figure(figsize=(14, 8))
sns.heatmap(pivot_avg, annot=True, fmt='.1f', cmap='YlOrRd', cbar_kws={'label': 'Avg Amount'})
plt.title('Average Monthly Purchasing Amount by Product Category (Seasonal Trends)')
plt.xlabel('Month')
plt.ylabel('Product Category')
plt.tight_layout()
plt.savefig('seasonal_trends_monthly_avg.png', dpi=300, bbox_inches='tight')
plt.show()

# Seasonal trends (by season)
seasonal_sales = df.groupby(['season', 'product_category'])['amount'].agg(['sum', 'mean']).reset_index()
pivot_season = seasonal_sales.pivot(index='product_category', columns='season', values='mean').fillna(0)

plt.figure(figsize=(10, 6))
sns.heatmap(pivot_season, annot=True, fmt='.1f', cmap='YlGnBu', cbar_kws={'label': 'Avg Amount'})
plt.title('Average Purchasing Amount by Season and Product Category')
plt.tight_layout()
plt.savefig('seasonal_trends_by_season.png', dpi=300, bbox_inches='tight')
plt.show()

# Line plot for trends over time (sample if many years)
year_month_sales = df.groupby(['year', 'month', 'product_category'])['amount'].sum().reset_index()
year_month_sales['date'] = pd.to_datetime(year_month_sales['year'].astype(str) + '-' + year_month_sales['month'].astype(str) + '-01')

plt.figure(figsize=(14, 8))
for cat in year_month_sales['product_category'].unique():
    cat_data = year_month_sales[year_month_sales['product_category'] == cat]
    plt.plot(cat_data['date'], cat_data['amount'], marker='o', label=cat)
plt.title('Total Sales Trends Over Time by Category')
plt.xlabel('Date')
plt.ylabel('Total Amount')
plt.legend(bbox_to_anchor=(1.05, 1), loc='upper left')
plt.xticks(rotation=45)
plt.tight_layout()
plt.savefig('sales_trends_over_time.png', dpi=300, bbox_inches='tight')
plt.show()

print("\nSeasonal Summary (Top categories by avg monthly sales):")
print(monthly_sales.groupby('product_category')['avg_amount'].mean().sort_values(ascending=False).head(10))

# 2. Flag statistical outliers by category (using IQR method)
def flag_outliers_iqr(group):
    Q1 = group['amount'].quantile(0.25)
    Q3 = group['amount'].quantile(0.75)
    IQR = Q3 - Q1
    lower_bound = Q1 - 1.5 * IQR
    upper_bound = Q3 + 1.5 * IQR
    group['is_outlier_iqr'] = (group['amount'] < lower_bound) | (group['amount'] > upper_bound)
    return group

df['is_outlier_iqr'] = False  # Initialize
df_flagged = df.groupby('product_category', group_keys=False).apply(flag_outliers_iqr)

outliers_iqr = df_flagged[df_flagged['is_outlier_iqr'] == True].copy()
print(f"\nIQR Outliers by Category:")
print(outliers_iqr['product_category'].value_counts().sort_values(ascending=False))
print(f"Total IQR outliers: {len(outliers_iqr)}")

# Alternative: Z-score outliers (threshold > 3 std devs)
def flag_outliers_zscore(group):
    group['zscore_amount'] = np.abs(stats.zscore(group['amount'], nan_policy='omit'))
    group['is_outlier_z'] = group['zscore_amount'] > 3
    return group

df['is_outlier_z'] = False
df_flagged = df.groupby('product_category', group_keys=False).apply(flag_outliers_zscore)

outliers_z = df_flagged[df_flagged['is_outlier_z'] == True].copy()
print(f"\nZ-Score Outliers by Category:")
print(outliers_z['product_category'].value_counts().sort_values(ascending=False))
print(f"Total Z-score outliers: {len(outliers_z)}")

# Visualizations for outliers
plt.figure(figsize=(15, 10))
sns.boxplot(data=df, x='product_category', y='amount', palette='Set2')
plt.title('Transaction Amounts by Category (Outliers Visible)')
plt.xticks(rotation=45)
plt.ylabel('Amount')
plt.tight_layout()
plt.savefig('boxplots_by_category.png', dpi=300, bbox_inches='tight')
plt.show()

# Save results
df_flagged.to_csv('transactions_with_outliers_flagged.csv', index=False)
outliers_iqr[['date', 'amount', 'product_category', 'region']].to_csv('iqr_outliers.csv', index=False)
outliers_z[['date', 'amount', 'product_category', 'region', 'zscore_amount']].to_csv('zscore_outliers.csv', index=False)
monthly_sales.to_csv('seasonal_trends_summary.csv', index=False)

print("\nOutputs saved:")
print("- transactions_with_outliers_flagged.csv (full dataset with flags)")
print("- iqr_outliers.csv")
print("- zscore_outliers.csv")
print("- seasonal_trends_summary.csv")
print("- PNG plots for trends and boxplots")

Key Features:

• Seasonal Trends: Heatmaps for monthly/seasonal averages, line plot for time series. Handles multiple years by averaging months.
• Outliers by Category: IQR (1.5x rule) and Z-score (>3σ) methods. Flags added to dataset.
• Visualizations: Saved as high-res PNGs.
• Exports: CSVs for flagged data and summaries.
• Assumptions: CSV file transactions.csv with exact column names. Adjust parse_dates or column names if needed (e.g., if "product category" has spaces, rename via df.rename).
• Scalable: Efficient for 10k rows; works with larger datasets.
• Region: Included in exports but not aggregated (add groupby(['product_category', 'region']) if needed for region-specific analysis).

Run this script in your environment after placing your CSV in the same directory!