from datetime import datetime, timedelta, date from typing import Tuple, List import altair as alt import pandas as pd import streamlit as st def overall(df: pd.DataFrame, artist: str, territory: List[str], period: Tuple[date, date]) -> None: """ Function to display overall merch metrics :param period: :param territory: :param df: Pandas Dataframe :param artist: Artist name to use for filtering :return: None """ st.write(f"Showing data from {period[0].date()} to {period[1].date()}") # type: ignore[attr-defined] labels = [ "Total Revenue", "Total Units", "Total Transactions", "Average Order Amount", "Unique Customers", ] df_columns = [ "SUM_ORDER_TOTAL", "TOTAL_UNITS", "TOTAL_ORDERS", "AVG_ORDER_TOTAL", "UNIQUE_CUSTOMERS", ] help_txt = [ "Money from products", "Total units sold", "Nr of completed transactions", "Includes product price + postage etc", "Nr of unique customers", ] for i, (label, df_column, help_, column) in enumerate( zip(labels, df_columns, help_txt, st.columns(5)) ): column.metric( label=label, value=( df[(df["ARTIST"] == artist) & (df["TERRITORY"].isin(territory))][df_column].sum() ).round(1), help=help_, ) def overall_raw( df: pd.DataFrame, artist: str, territory: List[str], period: Tuple[date, date] ) -> None: """ Function to calculate aggregate metrics and show graphs :param period: :param territory: :param df: Pandas Dataframe :param artist: Artist name to use for filtering :return: None """ st.write(f"Showing data from {period[0].date()} to {period[1].date()}") # type: ignore[attr-defined] df = df[(df["ARTIST"] == artist) & (df["TERRITORY"].isin(territory))] min_dt, max_dt = df["ORDER_DATE"].pipe(pd.to_datetime).dt.date.agg(["min", "max"]) df = df[(df["ORDER_DATE"] >= period[0].date()) & (df["ORDER_DATE"] <= period[1].date())] # type: ignore[attr-defined, assignment] if df.shape[0] > 0: df = ( df.groupby(["ARTIST", "TERRITORY"]) .agg( SUM_ORDER_TOTAL=pd.NamedAgg(column="ORDER_ITEMS_TOTAL", aggfunc="sum"), TOTAL_UNITS=pd.NamedAgg(column="ORDER_ITEMS_QUANTITY", aggfunc="sum"), AVG_ORDER_TOTAL=pd.NamedAgg(column="ORDER_TOTAL", aggfunc="mean"), TOTAL_ORDERS=pd.NamedAgg(column="ORDER_ID", aggfunc="count"), UNIQUE_CUSTOMERS=pd.NamedAgg(column="STORE_CUSTOMER_ID", aggfunc="nunique"), ) .reset_index() ) labels = [ "Total Revenue", "Total Units Sold", "Total Transactions", "Average Order Amount", "Unique Customers", ] df_columns = [ "SUM_ORDER_TOTAL", "TOTAL_UNITS", "TOTAL_ORDERS", "AVG_ORDER_TOTAL", "UNIQUE_CUSTOMERS", ] help_txt = [ "Money from products", "Total units sold", "Nr of completed transactions", "(avg for territory A + avg for territory B)/ nr of territories (Includes product price + postage etc)", "Nr of unique customers from territory A + Nr of unique customers from territory B etc", ] for i, (label, df_column, help_, column) in enumerate( zip(labels, df_columns, help_txt, st.columns(5)) ): column.metric( label=label, value=(df[df_column].sum()).round(1), help=help_, ) else: st.write( f"⚠️ Available date range for selected artist and territories is between {min_dt} to {max_dt}." ) def by_day(df: pd.DataFrame, period: Tuple[date, date], territories: List[str]) -> None: """ Function to show merch data by days :param period: :param df: Pandas Dataframe containing data grouped by artist and day :param territories: List of territories for filtering :return: None """ df = df[(df["ARTIST"] == st.session_state["artist"]) & (df["TERRITORY"].isin(territories))] df = df[(df["ORDER_DATE"] >= period[0].date()) & (df["ORDER_DATE"] <= period[1].date())] # type: ignore[attr-defined, assignment] all_territories = df["TERRITORY"].unique() try: all_dates = pd.date_range(df["ORDER_DATE"].min(), df["ORDER_DATE"].max()) full_index = pd.MultiIndex.from_product( [all_territories, all_dates], names=["TERRITORY", "ORDER_DATE"] ) df_full = ( df.set_index(["TERRITORY", "ORDER_DATE"]) .reindex(full_index) .groupby(level=0) .ffill() .reset_index() ) df_full["RUNNING_TOTAL"] = df_full["RUNNING_TOTAL"].fillna(0) base = alt.Chart( df_full.groupby(["ARTIST", "ORDER_DATE"])[["SUM_ORDER_TOTAL", "RUNNING_TOTAL"]] .sum() .reset_index() ).encode(alt.X("ORDER_DATE").title("Sales by date:")) line1 = base.mark_bar(opacity=0.8, color="#57A44C", interpolate="monotone").encode( alt.Y("SUM_ORDER_TOTAL").title("Total Sales($)") ) line2 = base.mark_line(stroke="#5276A7", interpolate="monotone").encode( alt.Y("RUNNING_TOTAL").title("Total New Fans") ) st.altair_chart( alt.layer(line1, line2).resolve_scale(y="independent"), use_container_width=True ) explain_main_metrics = st.toggle("Show explanation") if explain_main_metrics: st.write( "Green bars represent sales by day. Blue line represent how total unique fan count has increased from the first ever purchase." ) st.write( "So you can see what days were most profitable and what days contributed to the increase of first purchasers." ) st.divider() except ValueError: st.write("") # TODO: make this function below more universal def top_products( df: pd.DataFrame, df2: pd.DataFrame, period: Tuple[date, date], territories: List[str] ) -> None: """ Function to display product by artist and type :param period: dates for filtering :param df: Pandas dataframe containing product type data :param df2: Pandas dataframe containing product name data :param territories: List of territories for filtering :return: None """ df = df[(df["ORDER_DATE"] >= period[0].date()) & (df["ORDER_DATE"] <= period[1].date())] # type: ignore[attr-defined, assignment] df["ORDER_DATE"] = pd.to_datetime(df["ORDER_DATE"]) df["MONTH_YEAR"] = df["ORDER_DATE"].dt.to_period("M") df2 = df2[(df2["ORDER_DATE"] >= period[0].date()) & (df2["ORDER_DATE"] <= period[1].date())] # type: ignore[attr-defined, assignment] df2["ORDER_DATE"] = pd.to_datetime(df["ORDER_DATE"]) df2["MONTH_YEAR"] = df2["ORDER_DATE"].dt.to_period("M") st.write( '
Top selling products:
', unsafe_allow_html=True, ) top3 = ( df[(df["ARTIST"] == st.session_state["artist"]) & (df["TERRITORY"].isin(territories))] .groupby(["PRODUCT_TYPE"])["TOTAL_PRODUCTS"] .sum() .reset_index() .sort_values("TOTAL_PRODUCTS", ascending=False)[0:25] .rename(columns={"PRODUCT_TYPE": "Product Type", "TOTAL_PRODUCTS": "Total Products"}) ) top3_product_names = ( df2[(df2["ARTIST"] == st.session_state["artist"]) & (df2["TERRITORY"].isin(territories))] .groupby(["PRODUCT_NAME"])["TOTAL_PRODUCTS"] .sum() .reset_index() .sort_values("TOTAL_PRODUCTS", ascending=False)[0:25] .rename(columns={"PRODUCT_NAME": "Product Name", "TOTAL_PRODUCTS": "Total Products"}) ) product_by_day_df = ( df[(df["ARTIST"] == st.session_state["artist"]) & (df["TERRITORY"].isin(territories))] .groupby(["MONTH_YEAR", "PRODUCT_TYPE"])["TOTAL_PRODUCTS"] .sum() .reset_index() .pivot(index="MONTH_YEAR", columns="PRODUCT_TYPE", values="TOTAL_PRODUCTS") .reset_index() .fillna(value=0) ) product_by_day_df["MONTH_YEAR_ST"] = product_by_day_df["MONTH_YEAR"].dt.strftime("%Y-%m") product_by_day_df = product_by_day_df.drop("MONTH_YEAR", axis=1) st.bar_chart( product_by_day_df, x="MONTH_YEAR_ST", x_label="Year/Month", y_label="Nr of products" ) st.dataframe(top3, hide_index=True, use_container_width=False) st.divider() st.dataframe(top3_product_names, hide_index=True, use_container_width=False) def vs_last_week(df: pd.DataFrame, territories: List[str]) -> None: """ Function to display products sold this week vs last :param df: Pandas dataframe containing purchase data :param territories: List of territories for filtering data :return: None """ df["ORDER_DATE"] = pd.to_datetime(df["ORDER_DATE"]) # Get today's date today = datetime.today() # Get the start and end of the current week (Monday to Sunday) start_of_week = today - timedelta(days=today.weekday()) # Monday of current week # Get the start and end of last week start_of_last_week = start_of_week - timedelta(days=7) # Monday of last week end_of_last_week = start_of_week - timedelta(days=1) # Sunday of last week # Filter for current week current_week_df = df[(df["ORDER_DATE"] >= start_of_week)] current_week_df["WEEK"] = "CURRENT_WEEK" # Filter for last week last_week_df = df[ (df["ORDER_DATE"] >= start_of_last_week) & (df["ORDER_DATE"] <= end_of_last_week) ] last_week_df["WEEK"] = "LAST_WEEK" combined_df = pd.concat([current_week_df, last_week_df]) st.write( 'Sales this vs last week:
', unsafe_allow_html=True, ) st.dataframe( combined_df[ (combined_df["ARTIST"] == st.session_state["artist"]) & (combined_df["TERRITORY"].isin(territories)) & (combined_df["WEEK"].isin(["CURRENT_WEEK", "LAST_WEEK"])) ] .groupby(["WEEK"])["ORDER_ITEMS_TOTAL"] .count() .reset_index() .set_axis(["PERIOD", "TOTAL ITEMS"], axis=1), hide_index=True, )