src/data.py

import os
from collections import OrderedDict

import pandas as pd

from src.core.config import get_settings

settings = get_settings()

CONFIG = {
    "min_net_sales_percentage": 0.005,
    "max_net_sales_percentage": 0.1,
    "min_extra_qty_percentage": 0.1,
    "percentage_threshold": 0.5,
}


def check_net_sales_percentage(data):
    """
    Check the extra net sales percentage of total sales is between the min and max thresholds.
    """
    min_threshold = CONFIG.get("min_net_sales_percentage", 0.005)
    max_threshold = CONFIG.get("max_net_sales_percentage", 0.10)
    pct_threshold = CONFIG.get("percentage_threshold", 0.5)

    data["CHECK_EXTRA_NS_PCT"] = (min_threshold <= data.EXTRA_NET_SALE_PCT) & (
        data.EXTRA_NET_SALE_PCT <= max_threshold
    )
    data.drop(columns=["EXTRA_NET_SALE_PCT"], inplace=True)
    pct = data["CHECK_EXTRA_NS_PCT"].mean()
    return pct > pct_threshold


def check_in_history(reco, history, type):
    """
    Check if the product/basket is already in the history of the store
    """
    pct_threshold = CONFIG.get("percentage_threshold", 0.5)
    # check if the product was sold last week or in the whole history
    reco["CHECK_IN_HISTORY"] = reco.QTY_CURRENT > 0 | reco[
        "DISPLAY_NAME"
    ].isin(history[type])
    pct = reco.CHECK_IN_HISTORY.mean()
    return pct > pct_threshold


def check_extra_qty(reco):
    """
    Check the extra quantity percentage of total sales is above the threshold.
    """
    min_threshold_pct = CONFIG.get("min_extra_qty_percentage", 0.1)
    pct_threshold = CONFIG.get("percentage_threshold", 0.5)
    min_threshold = reco["QTY_CURRENT"] * min_threshold_pct
    reco["CHECK_EXTRA_QTY_PCT"] = reco["QTY_CURRENT"] > min_threshold
    pct = reco["CHECK_EXTRA_QTY_PCT"].mean()
    return pct > pct_threshold


def checks(reco, history, type):
    """
    Run all checks
    """
    if type == "PROD_NAME":
        check_net_sales_percentage(reco)
    check_extra_qty(reco)
    check_in_history(reco, history, type)
    # Filter out the recommendations that don't pass the checks
    check_cols = [c for c in reco.columns if c.startswith("CHECK_")]
    reco = reco[reco[check_cols].all(axis=1)].drop(columns=check_cols)
    return reco