utils.py

"""Description of the Ames Housing dataset.

This module uses the information available on the publication homepage and
defines a nested dictionary `ALL_COLUMNS` that can be used to decode the data
in the accompanying Excel file. For convenience, `ALL_VARIABLES` provides a
list of only the column names.

Furthermore, six helper dictionaries `CONTINUOUS_COLUMNS`, `DISCRETE_COLUMNS`,
`NUMERIC_COLUMNS`, `NOMINAL_COLUMNS`, `ORDINAL_COLUMNS`, and `LABEL_COLUMNS`
are defined that provide just the subset of the columns with the corresponding
data types. Note that the numeric dictionary unifies the continuous and
discrete data columns while the label dictionary unifies the nominal and
ordinal columns. For each of the six dictionaries, a list of the actual column
names is created with the same name and the suffix "_VARIABLES" instead of
"_COLUMNS", e.g., "CONTINUOUS_VARIABLES" instead of "CONTINUOUS_COLUMNS".

Lastly, the INDEX_COLUMNS and LABEL_TYPES list can be used to refer to the
actual values in a more readable way.

Source:
    https://www.amstat.org/publications/jse/v19n3/decock/DataDocumentation.txt

Implementation Note:
    This file defines the "constants" it exports dynamically. This is a bit
    advanced but intentional!
"""

import re

import numpy as np
import pandas as pd
import requests
import tabulate


FACTOR_VARIABLES = []
INDEX_COLUMNS = ["Order", "PID"]
LABEL_TYPES = ["nominal", "ordinal"]
TARGET_VARIABLES = ["SalePrice"]
# Note that these dictionaries and lists are not actually constants but
# filled in during import time which makes them "near"-constant.
ALL_COLUMNS = {}
ALL_VARIABLES = []
CONTINUOUS_COLUMNS = {}
CONTINUOUS_VARIABLES = []
DISCRETE_COLUMNS = {}
DISCRETE_VARIABLES = []
NUMERIC_COLUMNS = {}
NUMERIC_VARIABLES = []
NOMINAL_COLUMNS = {}
NOMINAL_VARIABLES = []
ORDINAL_COLUMNS = {}
ORDINAL_VARIABLES = []
LABEL_COLUMNS = {}
LABEL_VARIABLES = []


def _get_lines():
    """Obtain the non-empty lines of the data description file."""
    # Read cached data file.
    try:
        with open("data/data_documentation.txt", "r") as file:
            lines = file.readlines()
    # If there is no cached file, obtain in from the original source.
    except FileNotFoundError:
        response = requests.get(
            "https://www.amstat.org/publications"
            "/jse/v19n3/decock/DataDocumentation.txt"
        )
        # Cache the retrieved file.
        with open("data/data_documentation.txt", "w") as file:
            file.write(response.text)
        lines = response.text.split("\r\n")
    # Remove header, footer, and empty lines.
    lines = [x.replace("  ", " ").strip() for x in lines[13:545]]
    lines = [x for x in lines if x != ""]

    return lines


def _extract_meta_data(lines):
    """Extract variables and realizations for a line.

    This function parses the lines from the data documentation file and
    writes the results into the global dictionary ALL_COLUMNS that is exported
    by this module.

    A line can be a variable consisting of:
        - the name of the variable / column,
        - the variable's type (continuous, discrete, nominal, or ordinal), and
        - a text description of the variable.

    A line can also be a realization of a label column consisting of:
        - the encoding,
        - and the description.

    Implementation note:
        As the lines come in order, the "elif" condition below correctly refers
        to the last line representing a variable.
    """
    variable = re.compile(r"^(.*)(?:[\s]+)\(([\w]*)\)(?:\t)?: (.*)$")
    realization = re.compile(r"^(.*)\t(.*)$")
    # The two ID columns and the target variable "SalePrice"
    # are not put into the helper dicts / lists as they are
    # treated seperately in the modelling anyways.
    non_feature_columns = INDEX_COLUMNS + TARGET_VARIABLES

    for line in lines:
        # Process the next variable in the list.
        match = variable.match(line)
        if match:
            name, type_, description = match.groups()
            # Skip the non-feature columns (that are always non-label columns).
            if name in non_feature_columns:
                continue
            type_ = type_.lower()
            # Create an entry for the next variable in the list.
            ALL_COLUMNS[name] = {"type": type_, "description": description}
            # Only if the variable is a label type, a lookup table is needed.
            if type_ in LABEL_TYPES:
                ALL_COLUMNS[name].update({"lookups": {}})
            # Ordinal variables also store the order of their realizations
            # exactly as defined in the data description file.
            if type_ == "ordinal":
                ALL_COLUMNS[name].update({"order": []})
        # Add label realizations to a previously found label variable.
        elif type_ in LABEL_TYPES:
            match = realization.match(line)
            code, description = match.groups()
            code = code.strip()
            ALL_COLUMNS[name]["lookups"][code] = description
            if type_ == "ordinal":
                ALL_COLUMNS[name]["order"].append(code)


def _populate_dicts_and_lists():
    """Populate all "secondary" dictionaries and lists.

    The ALL_COLUMNS dictionary is the "main" dictionary and all other global
    dictionaries and lists are considered derived from it and thus considered
    "secondary".
    """
    # The global data structures are not re-assigned to so as to keep all
    # references in the Jupyter notebooks alive. Instead, they are emptied
    # and re-filled.
    ALL_VARIABLES[:] = sorted(ALL_COLUMNS)
    CONTINUOUS_COLUMNS.clear()
    CONTINUOUS_COLUMNS.update(
        {
            key: value
            for (key, value) in ALL_COLUMNS.items()
            if value["type"] == "continuous"
        }
    )
    CONTINUOUS_VARIABLES[:] = sorted(CONTINUOUS_COLUMNS)
    DISCRETE_COLUMNS.clear()
    DISCRETE_COLUMNS.update(
        {
            key: value
            for (key, value) in ALL_COLUMNS.items()
            if value["type"] == "discrete"
        }
    )
    DISCRETE_VARIABLES[:] = sorted(DISCRETE_COLUMNS)
    FACTOR_VARIABLES[:] = [
        key
        for (key, value) in ALL_COLUMNS.items()
        if value["type"] == "factor"
    ]
    NUMERIC_COLUMNS.clear()
    NUMERIC_COLUMNS.update({**CONTINUOUS_COLUMNS, **DISCRETE_COLUMNS})
    NUMERIC_VARIABLES[:] = sorted(NUMERIC_COLUMNS)
    NOMINAL_COLUMNS.clear()
    NOMINAL_COLUMNS.update(
        {
            key: value
            for (key, value) in ALL_COLUMNS.items()
            if value["type"] == "nominal"
        }
    )
    NOMINAL_VARIABLES[:] = sorted(NOMINAL_COLUMNS)
    ORDINAL_COLUMNS.clear()
    ORDINAL_COLUMNS.update(
        {
            key: value
            for (key, value) in ALL_COLUMNS.items()
            if value["type"] == "ordinal"
        }
    )
    ORDINAL_VARIABLES[:] = sorted(ORDINAL_COLUMNS)
    LABEL_COLUMNS.clear()
    LABEL_COLUMNS.update({**NOMINAL_COLUMNS, **ORDINAL_COLUMNS})
    LABEL_VARIABLES[:] = sorted(LABEL_COLUMNS)


def _rename_column(old_name, new_name):
    """Change the name of a column."""
    ALL_COLUMNS[new_name] = ALL_COLUMNS[old_name]
    del ALL_COLUMNS[old_name]


def correct_column_names(data_columns, *, repopulate=True):
    """Cross-check the column names between data and description file.

    In rare cases, the variable name in the data description file was slightly
    changed, i.e., a dash or a space needs to be removed.

    This function adjusts the keys in all the dictionaries and lists.
    """
    for desc_column in ALL_VARIABLES:
        if desc_column not in data_columns:
            for data_column in data_columns:
                # Column name was truncated in description file.
                if data_column.startswith(desc_column):
                    _rename_column(desc_column, data_column)
                    break
                # Spaces between words in Excel were removed.
                adj_data_column = data_column.replace(" ", "")
                if adj_data_column == desc_column:
                    _rename_column(desc_column, data_column)
                    break
                # Spaces between words in description file were removed.
                adj_desc_column = desc_column.replace(" ", "")
                if adj_data_column == adj_desc_column:
                    _rename_column(desc_column, data_column)
                    break
                # Dashes in description file were removed.
                adj_desc_column = desc_column.replace("-", "")
                if data_column == adj_desc_column:
                    _rename_column(desc_column, data_column)
                    break
    # Propagate the change to all "secondary" dictionaries and lists.
    if repopulate:
        _populate_dicts_and_lists()


def update_column_descriptions(columns_to_be_kept, *, correct_columns=False):
    """Remove discarded columns for all the module's exported data structures.

    After dropping some columns from the DataFrame, these removals must be
    propagated to the helper data structures defined in this module.
    """
    if correct_columns:
        correct_column_names(columns_to_be_kept, repopulate=False)
    columns_to_be_removed = list(set(ALL_COLUMNS) - set(columns_to_be_kept))
    for column in columns_to_be_removed:
        del ALL_COLUMNS[column]
    # Propagate the change to all "secondary" dictionaries and lists.
    _populate_dicts_and_lists()


def print_column_list(subset=None):
    """Print (a subset of) the data's column headers.

    Note that this function is built to handle both *_COLUMNS dicts and
    *_VARIABLES lists.
    """
    if subset is None:
        subset = ALL_VARIABLES
    else:
        subset = set(subset)
        # Handle variables withoutdescription seperately.
        without_desc = subset - set(ALL_VARIABLES)
        subset -= without_desc
    columns = [(c, ALL_COLUMNS[c]["description"]) for c in subset]
    if without_desc:
        for column in sorted(without_desc):
            columns.append((column, ""))
    print(tabulate.tabulate(sorted(columns), tablefmt="plain"))


def load_clean_data(path=None):
    """Return the clean project data as a pandas DataFrame.

    This utility function ensures that each column is cast to its correct type.

    It takes an optional path argument to a clean CSV file (defaults to
    "data/data_clean.csv").

    The target variables are always included as the last columns.

    Implementation Notes:

    One caveat is that all columns need to be casted as generic object type
    first, then the column names in the global dicts and lists are updated to
    reflect the slightly different column names (between data and description
    files), after which only the numeric columns can be casted correctly.

    Another difficulty is that some values, e.g., "NA" strings are cast as
    np.NaN / None by pandas although they represent actual label values.
    """
    # pragma pylint:disable=invalid-name
    df = pd.read_csv(
        "data/data_clean.csv" if path is None else path,
        index_col=INDEX_COLUMNS,
        dtype=object,
        na_values="",  # There are no missing values in the clean data file.
        keep_default_na=False,  # "NA" strings are casted as actual values.
    )
    # Remove columns that are in the description but not in the data file.
    update_column_descriptions(df.columns, correct_columns=True)
    # Cast the numeric types correctly.
    for column in CONTINUOUS_VARIABLES + TARGET_VARIABLES:
        df[column] = df[column].astype(float)
    for column in DISCRETE_VARIABLES:
        df[column] = df[column].astype(int)
    # Cast the label types as Categoricals.
    for column, mapping in NOMINAL_COLUMNS.items():
        labels = pd.api.types.CategoricalDtype(
            mapping["lookups"].keys(), ordered=False
        )
        df[column] = df[column].astype(labels)
    for column, mapping in ORDINAL_COLUMNS.items():
        labels = pd.api.types.CategoricalDtype(
            reversed(mapping["order"]), ordered=True
        )
        df[column] = df[column].astype(labels)
    # After the raw data cleaning, several derived variables were created.
    derived_columns = set(df.columns) - set(ALL_VARIABLES + TARGET_VARIABLES)
    if derived_columns:
        for column in derived_columns:
            # Check if the derived variable is a target variable.
            for target in TARGET_VARIABLES[:]:
                if column.startswith(target):
                    df[column] = df[column].astype(float)
                    TARGET_VARIABLES.append(column)
                    break
            else:
                df[column] = df[column].astype(float)
                is_int = (df[column] == df[column].astype(int)).all()
                n_unique = len(df[column].unique())
                if is_int & (n_unique == 2):
                    df[column] = df[column].astype(int)
                    type_ = "factor"
                elif is_int & (n_unique < 150):
                    df[column] = df[column].astype(int)
                    type_ = "discrete"
                else:
                    df[column] = df[column].astype(float)
                    type_ = "continuous"
                ALL_COLUMNS[column] = {
                    "type": type_,
                    "description": "derived variable",
                }
        _populate_dicts_and_lists()
    return df


def encode_ordinals(df):
    """Replace ordinal columns' labels with integer codes."""
    # pragma pylint:disable=invalid-name
    df = df.copy()
    for column in df.columns:
        if column in ORDINAL_VARIABLES:
            df[column] = df[column].cat.codes.astype(int)
    return df


def bias_score(y_true, y_pred):
    """Determine the bias of a prediction."""
    assert y_true.shape == y_pred.shape
    assert y_true.ndim == 1
    return np.mean(y_pred - y_true)


def max_deviation(y_true, y_pred):
    """Determine the maximum deviation of a prediction."""
    assert y_true.shape == y_pred.shape
    assert y_true.ndim == 1
    return np.max(np.abs(y_pred - y_true))


# This code is executed once during import time and
# populates all the "constants" directly or indirectly.
_extract_meta_data(_get_lines())
_populate_dicts_and_lists()