Imbalanced Data and Anomaly Detection: Average Precision, Rebalancing, Alert Budgets and Unseen Failures

"""Failure warnings for Brackwater Wind's turbines. You write the functions marked TODO, one step at a time."""
import json

import numpy as np
import pandas as pd
from sklearn.ensemble import HistGradientBoostingClassifier

FEATURES = ["vib_rms", "oil_particles", "bearing_temp", "ambient_temp", "power_ratio"]
LABEL = "fails_within_14d"
SPLIT_DAY = 300     # train on days before, test on days from here on
HORIZON = 14        # a warning counts if it comes within 14 days before the failure


def load():
    """(train, test): days before SPLIT_DAY, and the 150 days after."""
    return pd.read_csv("train_days.csv"), pd.read_csv("test_days.csv")


def load_events():
    with open("events.json") as f:
        return json.load(f)


def fit_model(train, sample_weight=None, rows=None):
    """The supervised model: gradient-boosted trees on FEATURES. rows: optional row indices to train on (repeats
    allowed); sample_weight: optional weight per training row."""
    data = train if rows is None else train.iloc[rows]
    return HistGradientBoostingClassifier(random_state=0).fit(data[FEATURES], data[LABEL], sample_weight=sample_weight)


# ---------- Step 1: the accuracy trap ----------

def confusion(y, pred):
    """{"tp", "fp", "fn", "tn"} counts for 0/1 arrays y (truth) and pred."""
    # TODO (Step 1): turn y and pred into boolean arrays and count the four combinations.
    raise NotImplementedError("Step 1: write confusion()")


def average_precision(y, scores):
    """Rank rows by score, highest first; the mean, over the positive rows, of the precision among all rows ranked
    at or above each one. (Assume no tied scores.)"""
    # TODO (Step 1): order y by score (np.argsort(-scores)), precision at every rank =
    # np.cumsum(y) / rank, then the mean of it at the positive rows.
    raise NotImplementedError("Step 1: write average_precision()")


# ---------- Step 2: rebalancing ----------

def oversample(y, seed=0):
    """Row indices for a balanced training set: every row once, plus positive rows drawn with replacement
    (np.random.default_rng(seed).choice) until there are as many positives as negatives."""
    raise NotImplementedError("oversample() arrives in Step 2")


def balanced_weights(y):
    """A weight per row so both classes carry equal total weight: n / (2 * count of the row's class)."""
    raise NotImplementedError("balanced_weights() arrives in Step 2")


# ---------- Step 3: honest probabilities ----------

def correct_prior(p, trained_rate, true_rate):
    """Undo the base-rate shift of training on a set with positive rate trained_rate: multiply the odds p / (1 - p)
    by (true_rate / (1 - true_rate)) / (trained_rate / (1 - trained_rate)) and turn them back into probabilities."""
    raise NotImplementedError("correct_prior() arrives in Step 3")


def reliability(y, p, bins=(0, 0.02, 0.05, 0.1, 0.2, 0.5, 1.0)):
    """A DataFrame with one row per probability bin that holds rows: "bin" (a label like "0.05-0.1"), "n",
    "predicted" (mean p) and "actual" (positive rate). A value equal to an upper edge belongs to that bin; p = 0
    to the first."""
    raise NotImplementedError("reliability() arrives in Step 3")


# ---------- Step 4: a crew that can check one turbine a day ----------

def daily_top_k(df, scores, k=1):
    """The alerts a crew that inspects k turbines a day would act on: for each day, the k rows with the highest
    score. A DataFrame with columns turbine and day."""
    raise NotImplementedError("daily_top_k() arrives in Step 4")


def events_caught(alerts, events, since=SPLIT_DAY, horizon=HORIZON):
    """For every event with day >= since: {"turbine", "day", "mode", "caught", "lead_days"}. caught: some alert on
    that turbine came 1 to horizon days before the failure; lead_days: days between the earliest such alert and the
    failure (0 when missed)."""
    raise NotImplementedError("events_caught() arrives in Step 4")


# ---------- Step 5: failures nobody labelled ----------

def healthy_signals(df, coef):
    """What a healthy turbine keeps steady: vib_rms, oil_particles, the bearing temperature minus what ambient
    explains (np.polyval(coef, ambient_temp)), and power_ratio. A (rows, 4) array."""
    raise NotImplementedError("healthy_signals() arrives in Step 5")


def turbine_baselines(df, coef):
    """Each turbine's own normal: the median of every column of healthy_signals() over its rows. A DataFrame indexed
    by turbine with columns 0-3."""
    raise NotImplementedError("turbine_baselines() arrives in Step 5")


def centred(df, coef, baselines):
    """healthy_signals(df, coef) minus the baseline of each row's own turbine."""
    raise NotImplementedError("centred() arrives in Step 5")


class RobustZ:
    """Distance from normal in robust units: median and 1.4826 x median absolute deviation per column, fitted on
    healthy rows. score() is the largest |z| over the columns of each row."""

    def fit(self, X):
        raise NotImplementedError("RobustZ.fit() arrives in Step 5")

    def score(self, X):
        raise NotImplementedError("RobustZ.score() arrives in Step 5")


def anomaly_detector(train):
    """(coef, baselines, detector), all from the training rows NOT followed by a failure: coef =
    np.polyfit(ambient_temp, bearing_temp, 1), baselines = turbine_baselines(), and a RobustZ fitted on centred()."""
    raise NotImplementedError("anomaly_detector() arrives in Step 5")


Z_ALERT = 5.0