data_preprocessing_function.py

import streamlit as st
import numpy as np
import pandas as pd
from sklearn.preprocessing import LabelEncoder
from sklearn.preprocessing import StandardScaler, MinMaxScaler
from scipy import stats


@st.cache_data
def remove_selected_columns(df, columns_remove):
    if not columns_remove:
        return df
    return df.drop(columns=list(columns_remove), errors="ignore")

# Create a function to remove rows with missing values in specific columns
@st.cache_data
def remove_rows_with_missing_data(df, columns):
    if columns:
        return df.dropna(subset=columns)
    return df

# Create a function to fill missing data with mean, median, or mode (for numerical columns)
@st.cache_data
def fill_missing_data(df, columns, method):
    result_df = df.copy()
    for column in columns:
        if method == 'mean':
            result_df[column] = result_df[column].fillna(result_df[column].mean())
        elif method == 'median':
            result_df[column] = result_df[column].fillna(result_df[column].median())
        elif method == 'mode':
            mode_val = result_df[column].mode().iloc[0]
            result_df[column] = result_df[column].fillna(mode_val)
    return result_df

@st.cache_data
def one_hot_encode(df, columns):
    return pd.get_dummies(df, columns=columns, prefix=columns, drop_first=False)

@st.cache_data
def label_encode(df, columns):
    result_df = df.copy()
    label_encoder = LabelEncoder()
    for col in columns:
        result_df[col] = label_encoder.fit_transform(result_df[col])
    return result_df


@st.cache_data
def standard_scale(df, columns):
    result_df = df.copy()
    scaler = StandardScaler()
    result_df[columns] = scaler.fit_transform(result_df[columns])
    return result_df

@st.cache_data
def min_max_scale(df, columns, feature_range=(0, 1)):
    result_df = df.copy()
    scaler = MinMaxScaler(feature_range=feature_range)
    result_df[columns] = scaler.fit_transform(result_df[columns])
    return result_df

@st.cache_data
def detect_outliers_iqr(df, column_name):
    data = pd.to_numeric(df[column_name], errors='coerce')
    valid_data = data.dropna()
    if valid_data.empty:
        return []

    q25, _, q75 = np.percentile(valid_data, [25, 50, 75])
    iqr = q75 - q25
    lower_bound = q25 - 1.5 * iqr
    upper_bound = q75 + 1.5 * iqr
    outlier_mask = (data < lower_bound) | (data > upper_bound)
    return np.where(outlier_mask.fillna(False))[0].tolist()



# Function to detect outliers using z-score
@st.cache_data
def detect_outliers_zscore(df, column_name):
    data = pd.to_numeric(df[column_name], errors='coerce')
    if data.isna().all():
        return []

    mean = np.nanmean(data)
    std = np.nanstd(data)
    if std == 0 or np.isnan(std):
        return []

    z_scores = np.abs((data - mean) / std)
    threshold = 3
    outlier_mask = (z_scores > threshold) & ~z_scores.isna()
    return np.where(outlier_mask)[0].tolist()

@st.cache_data
def remove_outliers(df, column_name, outlier_indices):
    if not outlier_indices:
        return df

    valid_indices = [idx for idx in sorted(set(outlier_indices)) if 0 <= idx < len(df)]
    if not valid_indices:
        return df

    return df.drop(index=df.index[valid_indices]).reset_index(drop=True)

@st.cache_data
def transform_outliers(df, column_name, outlier_indices):
    result_df = df.copy()
    if not outlier_indices:
        return result_df

    valid_positions = [idx for idx in sorted(set(outlier_indices)) if 0 <= idx < len(result_df)]
    if not valid_positions:
        return result_df

    column_idx = result_df.columns.get_loc(column_name)
    non_outlier_positions = [idx for idx in range(len(result_df)) if idx not in valid_positions]
    if not non_outlier_positions:
        return result_df

    median_value = result_df.iloc[non_outlier_positions, column_idx].median()
    result_df.iloc[valid_positions, column_idx] = median_value
    return result_df