API Reference¶

def load_data(filepath: Path, name: str) -> pd.DataFrame:
    """Load one CSV file into a pandas DataFrame.

    WHY: Reading from CSV into a DataFrame is the first step
    in almost every BI pipeline.


    Args:
        filepath: Path to the CSV file.
        name: A short name for logging.

    Returns:
        A pandas DataFrame with the file contents.
    """
    LOG.info(f"Loading {name} from {filepath}")
    df: pd.DataFrame = pd.read_csv(filepath)
    LOG.info(f"Loaded: {df.shape[0]} rows, {df.shape[1]} columns")
    return df

def main() -> None:
    """Main function to run the BI logic.
    This is where the main logic starts
    when this script is run.
    """

    # First, log the header for the BI module to indicate the start of the workflow.
    log_header(LOG, "BI")

    # Clearly indicate the start of the main function in the logs for easy tracking.
    LOG.info("========================")
    LOG.info("START main()")
    LOG.info("========================")

    # Use the imported log_path function to
    # log the paths of all critical paths and files for reference.
    log_path(LOG, "Raw data: ", DATA_RAW)
    log_path(LOG, "Customers:", CUSTOMERS_FILE)
    log_path(LOG, "Products: ", PRODUCTS_FILE)
    log_path(LOG, "Sales:    ", SALES_FILE)

    LOG.info("CALL a function to load each dataset.............")
    df_customers = load_data(CUSTOMERS_FILE, "customers")
    df_products = load_data(PRODUCTS_FILE, "products")
    df_sales = load_data(SALES_FILE, "sales")

    LOG.info("CALL a function to get sales by region........")
    df_region = sales_by_region(df_customers, df_sales)

    LOG.info("CALL a function to plot sales by region........")
    plot_bar(
        df=df_region,
        x="Region",
        y="SaleAmount",
        title="Total Sales by Region",
        xlabel="Region",
        ylabel="Total Sales Amount ($)",
        palette="Blues_d",
    )

    LOG.info("CALL a function to get sales by product category........")
    df_category = sales_by_category(df_products, df_sales)

    LOG.info("CALL a function to plot sales by product category........")
    plot_bar(
        df=df_category,
        x="Category",
        y="SaleAmount",
        title="Total Sales by Product Category",
        xlabel="Category",
        ylabel="Total Sales Amount ($)",
        palette="Greens_d",
    )

    LOG.info("CALL a function to summarize the datasets........")
    summarize(df_customers, df_products, df_sales)

    LOG.info("CALL a function to show charts........")
    plt.show()

    LOG.info("Workflow complete")
    LOG.info("CLOSE chart windows to continue.")
    LOG.info("Terminate this process with CTRL+c as needed.")
    LOG.info("========================")
    LOG.info("Executed successfully!")
    LOG.info("========================")

def plot_bar(
    df: pd.DataFrame,
    x: str,
    y: str,
    title: str,
    xlabel: str,
    ylabel: str,
    palette: str = "Blues_d",
) -> None:
    """Plot a vertical bar chart.

    WHY: Bar charts are the most common chart in BI reporting.
    A reusable function ensures consistent style and labeling
    across all modules.

    Args:
        df: DataFrame containing the data to plot.
        x: Column name for the x-axis (categories).
        y: Column name for the y-axis (values).
        title: Chart title.
        xlabel: X-axis label.
        ylabel: Y-axis label.
        palette: Seaborn color palette name.

    Returns:
        None
    """
    LOG.info(f"Creating chart: {title}")

    _, ax = plt.subplots(figsize=(9, 5))

    bar: Axes = sns.barplot(
        data=df,
        x=x,
        y=y,
        hue=x,
        legend=False,
        palette=palette,
        ax=ax,
    )

    bar.set_title(f"{title} (CLOSE chart to continue)")
    bar.set_xlabel(xlabel)
    bar.set_ylabel(ylabel)

    plt.tight_layout()

def sales_by_category(
    df_products: pd.DataFrame,
    df_sales: pd.DataFrame,
) -> pd.DataFrame:
    """Aggregate total sales amount by product category.

    WHY: Product category is another key business dimension.
    Understanding which categories drive revenue helps prioritize
    inventory, marketing, and purchasing decisions.

    Args:
        df_products: Products DataFrame with ProductID and Category columns.
        df_sales: Sales DataFrame with ProductID and SaleAmount columns.

    Returns:
        DataFrame with Category and SaleAmount columns, sorted by SaleAmount.
    """
    LOG.info("Aggregating sales by product category")

    # Make a copy of the sales DataFrame to avoid modifying the original
    df_sales = df_sales.copy()

    # Convert the SaleAmount column to numeric, coercing errors to NaN ("not a number")
    df_sales["SaleAmount"] = pd.to_numeric(df_sales["SaleAmount"], errors="coerce")

    # Merge the sales DataFrame with the products DataFrame on ProductID
    # to get the Category for each sale (like a "left join" in SQL)
    # A left join means we keep all rows from df_sales and add matching Category values from df_products.
    df_merged: pd.DataFrame = df_sales.merge(
        df_products[["ProductID", "Category"]],
        on="ProductID",
        how="left",
    )

    # Group the merged DataFrame by Category and sum the SaleAmount for each category.
    # This returns a Series (a single column of values, one per category).
    # We cast to Series because we are grouping a single column.
    grouped: pd.Series = pd.Series(df_merged.groupby("Category")["SaleAmount"].sum())

    # Reset the index to turn the Series back into a DataFrame with two columns:
    # Category and SaleAmount.
    # Then sort by SaleAmount descending so the highest-revenue category appears first.
    df_category: pd.DataFrame = grouped.reset_index().sort_values(
        "SaleAmount", ascending=False
    )

    # Use the built-in dataframe iloc (index location) method
    # to get the first row of the sorted DataFrame (the category with the highest sales)
    # as a string.
    # In Python, we start counting with 0 (no offset from the beginning of the list),
    # so the first row is at index 0.
    top_category: str = str(df_category.iloc[0]["Category"])

    # Use the built-in dataframe iloc (index location) method
    # to get the first row of the sorted DataFrame
    # (the category with the highest sales)
    # as a float.
    top_sales: float = float(df_category.iloc[0]["SaleAmount"])

    # Log the top category and its total sales amount for quick reference
    # Using handy dandy f-strings (formatted string literals).
    # Format the sales amount as currency with commas and two floating decimal places.
    LOG.info(f"  Top category: {top_category} (${top_sales:,.2f})")

    return df_category

def sales_by_region(
    df_customers: pd.DataFrame,
    df_sales: pd.DataFrame,
) -> pd.DataFrame:
    """Aggregate total sales amount by customer region.

    Args:
        df_customers: Customers DataFrame with CustomerID and Region columns.
        df_sales: Sales DataFrame with CustomerID and SaleAmount columns.

    Returns:
        DataFrame with Region and SaleAmount columns, sorted by SaleAmount.
    """
    LOG.info("Aggregating sales by region")

    # Make a copy of the sales DataFrame to avoid modifying the original
    df_sales = df_sales.copy()

    # Convert the SaleAmount column to numeric, coercing errors to NaN ("not a number")
    df_sales["SaleAmount"] = pd.to_numeric(df_sales["SaleAmount"], errors="coerce")

    # Merge the sales DataFrame with the customers DataFrame on CustomerID
    # to get the Region for each sale (like a "left join" in SQL)
    # A left join means we keep all rows from df_sales and add matching Region values from df_customers.
    df_merged: pd.DataFrame = df_sales.merge(
        df_customers[["CustomerID", "Region"]],
        on="CustomerID",
        how="left",
    )

    # Clean up the Region column by stripping whitespace and capitalizing each word
    df_merged["Region"] = df_merged["Region"].str.strip().str.title()

    # Group the merged DataFrame by Region and sum the SaleAmount for each region.
    # This returns a Series (a single column of values, one per region).
    # We cast to Series because we are grouping a single column.
    grouped: pd.Series = pd.Series(df_merged.groupby("Region")["SaleAmount"].sum())

    # Reset the index to turn the Series back into a DataFrame with two columns:
    # Region and SaleAmount.
    # Then sort by SaleAmount descending so the highest-revenue region appears first.
    df_region: pd.DataFrame = grouped.reset_index().sort_values(
        "SaleAmount", ascending=False
    )

    # Use the built-in dataframe iloc (index location) method
    # to get the first row of the sorted DataFrame (the region with the highest sales)
    # as a string.
    # In Python, we start counting with 0
    # (no offset from the beginning of the list),
    # so the first row is at index 0.
    top_region: str = str(df_region.iloc[0]["Region"])

    # Use the built-in dataframe iloc (index location) method
    # to get the first row of the sorted DataFrame
    # (the region with the highest sales)
    # as a float.
    top_sales: float = float(df_region.iloc[0]["SaleAmount"])

    # Log the top region and its total sales amount for quick reference
    # Using handy dandy f-strings (formatted string literals).
    # Format the sales amount as currency with commas and two floating decimal places.
    LOG.info(f"  Top region: {top_region} (${top_sales:,.2f})")

    LOG.info("Returning DataFrame with total sales by region")
    return df_region

def summarize(
    df_customers: pd.DataFrame,
    df_products: pd.DataFrame,
    df_sales: pd.DataFrame,
) -> None:
    """Log a brief summary of all three datasets.

    Args:
        df_customers: Customers DataFrame.
        df_products: Products DataFrame.
        df_sales: Sales DataFrame.

    Returns:
        None
    """
    LOG.info("========================")
    LOG.info("SUMMARY")
    LOG.info("========================")

    # Get the number of rows and columns in each using the shape attribute (0=rows, 1=columns)
    cust_rows: int = df_customers.shape[0]
    cust_cols: int = df_customers.shape[1]

    # Get the number of rows and columns in the products DataFrame using the shape attribute (0=rows, 1=columns)
    prod_rows: int = df_products.shape[0]
    prod_cols: int = df_products.shape[1]

    # Get the number of rows and columns in the sales DataFrame using the shape attribute (0=rows, 1=columns)
    sale_rows: int = df_sales.shape[0]
    sale_cols: int = df_sales.shape[1]

    # Log the summary of each dataset using f-strings (formatted string literals)
    # Start with an f outside the string, then use curly braces {} to insert variables into the string.
    LOG.info(f"Customers:  {cust_rows} rows, {cust_cols} columns")
    LOG.info(f"Products:   {prod_rows} rows, {prod_cols} columns")
    LOG.info(f"Sales:      {sale_rows} rows, {sale_cols} columns")

    LOG.info("========================")
    LOG.info("ANALYST NOTES:")
    LOG.info("Note any data quality issues.")
    LOG.info("We will clean data later.")
    LOG.info("========================")