MAPE in Machine Learning

import numpy as np
from sklearn.metrics import mean_absolute_percentage_error

# Ground truth and predictions (demand forecasting example)
y_true = np.array([100, 150, 200, 50, 300, 80])
y_pred = np.array([110, 130, 210, 45, 280, 95])

# 1. Standard MAPE via scikit-learn (returns as fraction, not %)
mape_sklearn = mean_absolute_percentage_error(y_true, y_pred)
print(f"MAPE (sklearn, fraction): {mape_sklearn:.4f}")
print(f"MAPE (percentage):        {mape_sklearn * 100:.2f}%")
# Output: MAPE (percentage): 11.08%

# 2. MAPE with sample weights (prioritize high-value predictions)
weights = np.array([2, 1, 1, 3, 1, 2])  # Weight perishables higher
mape_weighted = mean_absolute_percentage_error(
    y_true, y_pred, sample_weight=weights
)
print(f"Weighted MAPE: {mape_weighted * 100:.2f}%")

# 3. Multi-output MAPE (multiple forecast horizons)
y_true_multi = np.array([[100, 110], [200, 190], [150, 160]])
y_pred_multi = np.array([[105, 115], [180, 195], [155, 150]])

mape_multi = mean_absolute_percentage_error(
    y_true_multi, y_pred_multi, multioutput='raw_values'
)
print(f"Per-horizon MAPE: {mape_multi * 100}")  # Per column

# WARNING: sklearn produces inf when y_true contains zeros!
y_with_zero = np.array([100, 0, 200])
y_pred_zero = np.array([110, 5, 190])
mape_inf = mean_absolute_percentage_error(y_with_zero, y_pred_zero)
print(f"MAPE with zero: {mape_inf}")  # inf!

import numpy as np
from typing import Optional, Dict

def mape(
    y_true: np.ndarray,
    y_pred: np.ndarray,
    zero_strategy: str = "skip",
    epsilon: float = 1e-8
) -> float:
    """
    Standard MAPE with configurable zero handling.
    
    Args:
        y_true: Actual values
        y_pred: Predicted values
        zero_strategy: 'skip' (exclude zeros), 'clip' (clip denom to epsilon),
                       'error' (raise ValueError)
        epsilon: Minimum denominator for 'clip' strategy
    Returns:
        MAPE as a percentage (e.g., 12.5 means 12.5%)
    """
    y_true = np.asarray(y_true, dtype=np.float64)
    y_pred = np.asarray(y_pred, dtype=np.float64)
    
    if y_true.shape != y_pred.shape:
        raise ValueError(f"Shape mismatch: {y_true.shape} vs {y_pred.shape}")
    
    abs_errors = np.abs(y_true - y_pred)
    
    if zero_strategy == "skip":
        mask = np.abs(y_true) > epsilon
        if not mask.any():
            raise ValueError("All actual values are zero")
        ape = abs_errors[mask] / np.abs(y_true[mask])
    elif zero_strategy == "clip":
        denom = np.maximum(np.abs(y_true), epsilon)
        ape = abs_errors / denom
    elif zero_strategy == "error":
        if np.any(np.abs(y_true) < epsilon):
            raise ValueError("Zero actual values found; MAPE is undefined")
        ape = abs_errors / np.abs(y_true)
    else:
        raise ValueError(f"Unknown zero_strategy: {zero_strategy}")
    
    return float(np.mean(ape) * 100)


def smape(y_true: np.ndarray, y_pred: np.ndarray) -> float:
    """
    Symmetric MAPE. Range: [0%, 200%].
    Avoids division by zero when both actual and predicted are nonzero.
    """
    y_true = np.asarray(y_true, dtype=np.float64)
    y_pred = np.asarray(y_pred, dtype=np.float64)
    
    numerator = np.abs(y_true - y_pred)
    denominator = (np.abs(y_true) + np.abs(y_pred)) / 2
    
    # Handle case where both actual and predicted are zero
    mask = denominator > 0
    result = np.zeros_like(numerator)
    result[mask] = numerator[mask] / denominator[mask]
    
    return float(np.mean(result) * 100)


def wmape(y_true: np.ndarray, y_pred: np.ndarray) -> float:
    """
    Weighted MAPE. Equivalent to MAE / mean(|y_true|).
    Handles individual zeros gracefully; only fails if ALL actuals are zero.
    """
    y_true = np.asarray(y_true, dtype=np.float64)
    y_pred = np.asarray(y_pred, dtype=np.float64)
    
    total_abs_error = np.sum(np.abs(y_true - y_pred))
    total_actual = np.sum(np.abs(y_true))
    
    if total_actual == 0:
        raise ValueError("WMAPE undefined: sum of actuals is zero")
    
    return float(total_abs_error / total_actual * 100)


def maape(y_true: np.ndarray, y_pred: np.ndarray) -> float:
    """
    Mean Arctangent APE. Bounded in [0, pi/2). Handles zeros gracefully.
    Proposed by Kim & Kim (2016).
    """
    y_true = np.asarray(y_true, dtype=np.float64)
    y_pred = np.asarray(y_pred, dtype=np.float64)
    
    abs_errors = np.abs(y_true - y_pred)
    # Use safe division, defaulting to 0 where y_true is 0
    with np.errstate(divide='ignore', invalid='ignore'):
        ratios = np.where(y_true != 0, abs_errors / np.abs(y_true), abs_errors)
    
    return float(np.mean(np.arctan(ratios)))


# Demonstration with demand forecasting data
y_actual = np.array([120, 0, 85, 200, 15, 300, 0, 50])
y_forecast = np.array([130, 5, 80, 190, 20, 310, 3, 55])

print("=== MAPE Variants Comparison ===")
print(f"MAPE (skip zeros):   {mape(y_actual, y_forecast, 'skip'):.2f}%")
print(f"MAPE (clip zeros):   {mape(y_actual, y_forecast, 'clip'):.2f}%")
print(f"SMAPE:               {smape(y_actual, y_forecast):.2f}%")
print(f"WMAPE:               {wmape(y_actual, y_forecast):.2f}%")
print(f"MAAPE (radians):     {maape(y_actual, y_forecast):.4f}")
print(f"\nZero observations:   {np.sum(y_actual == 0)} / {len(y_actual)}")

import torch
import torch.nn as nn
from torch.utils.data import DataLoader, TensorDataset


class MAPEMetric:
    """
    Stateful MAPE metric for PyTorch evaluation loops.
    Accumulates predictions across batches, computes MAPE at the end.
    
    WARNING: Do NOT use MAPE as a training loss. Use MAE or Huber instead,
    then evaluate with MAPE after training.
    """
    
    def __init__(self, epsilon: float = 1e-8):
        self.epsilon = epsilon
        self.reset()
    
    def reset(self):
        self.total_ape = 0.0
        self.count = 0
        self.skipped_zeros = 0
    
    def update(self, y_pred: torch.Tensor, y_true: torch.Tensor):
        """Accumulate APE values from a batch."""
        abs_errors = torch.abs(y_true - y_pred)
        
        # Skip near-zero actuals
        mask = torch.abs(y_true) > self.epsilon
        valid_ape = abs_errors[mask] / torch.abs(y_true[mask])
        
        self.total_ape += valid_ape.sum().item()
        self.count += mask.sum().item()
        self.skipped_zeros += (~mask).sum().item()
    
    def compute(self) -> dict:
        """Compute MAPE from accumulated values."""
        if self.count == 0:
            return {"mape_pct": float('inf'), "valid_count": 0, "skipped": self.skipped_zeros}
        
        mape_val = (self.total_ape / self.count) * 100
        return {
            "mape_pct": mape_val,
            "valid_count": self.count,
            "skipped_zeros": self.skipped_zeros
        }


# Example: Evaluate a trained model
def evaluate_model_mape(model, dataloader, device='cpu'):
    """Evaluate regression model and report MAPE."""
    model.eval()
    mape_metric = MAPEMetric()
    mae_total = 0.0
    n_samples = 0
    
    with torch.no_grad():
        for X_batch, y_batch in dataloader:
            X_batch, y_batch = X_batch.to(device), y_batch.to(device)
            predictions = model(X_batch)
            
            # Update MAPE (for percentage-based reporting)
            mape_metric.update(predictions, y_batch)
            
            # Also track MAE (for unit-based reporting)
            mae_total += torch.abs(y_batch - predictions).sum().item()
            n_samples += len(y_batch)
    
    results = mape_metric.compute()
    results["mae"] = mae_total / n_samples
    return results


# Demo with synthetic data
torch.manual_seed(42)
model = nn.Sequential(
    nn.Linear(5, 32), nn.ReLU(),
    nn.Linear(32, 16), nn.ReLU(),
    nn.Linear(16, 1)
)

# Train with MAE loss (NOT MAPE)
optimizer = torch.optim.Adam(model.parameters(), lr=0.01)
criterion = nn.L1Loss()  # MAE for training

X = torch.randn(500, 5)
y = torch.randn(500, 1) * 10 + 50  # Positive target values

dataset = TensorDataset(X, y)
train_loader = DataLoader(dataset, batch_size=32, shuffle=True)

# Quick training
for epoch in range(20):
    for X_batch, y_batch in train_loader:
        optimizer.zero_grad()
        loss = criterion(model(X_batch), y_batch)
        loss.backward()
        optimizer.step()

# Evaluate with MAPE
results = evaluate_model_mape(model, train_loader)
print(f"MAPE: {results['mape_pct']:.2f}%")
print(f"MAE:  {results['mae']:.3f}")
print(f"Valid samples: {results['valid_count']}")
print(f"Skipped (zero actuals): {results['skipped_zeros']}")

import numpy as np
from collections import deque
from typing import List


class MAPEMonitor:
    """Production MAPE monitor with asymmetry and drift detection."""
    
    def __init__(self, mape_threshold: float = 15.0, history_size: int = 168):
        self.mape_threshold = mape_threshold
        self.history: deque = deque(maxlen=history_size)
        self._errors: List[float] = []
        self._actuals: List[float] = []
        self._over_count = 0
        self._total_count = 0
    
    def record(self, y_true: float, y_pred: float):
        """Record a single prediction."""
        self._errors.append(abs(y_true - y_pred))
        self._actuals.append(abs(y_true))
        self._total_count += 1
        if y_pred > y_true:
            self._over_count += 1
    
    def flush_window(self) -> dict:
        """Compute WMAPE and asymmetry for current window."""
        errors = np.array(self._errors)
        actuals = np.array(self._actuals)
        
        total_actual = np.sum(actuals)
        wmape = (np.sum(errors) / total_actual * 100) if total_actual > 0 else float('inf')
        over_ratio = self._over_count / max(self._total_count, 1)
        
        result = {
            'wmape_pct': wmape,
            'over_predict_ratio': over_ratio,
            'sample_count': self._total_count,
            'zero_count': int(np.sum(actuals < 1e-8))
        }
        self.history.append(result)
        
        # Reset
        self._errors, self._actuals = [], []
        self._over_count, self._total_count = 0, 0
        return result
    
    def check_alerts(self) -> List[str]:
        """Check threshold, asymmetry, and drift alerts."""
        alerts = []
        if not self.history:
            return alerts
        latest = self.history[-1]
        
        if latest['wmape_pct'] > self.mape_threshold:
            alerts.append(f"WMAPE {latest['wmape_pct']:.1f}% > {self.mape_threshold}%")
        if latest['over_predict_ratio'] > 0.65:
            alerts.append(f"Over-prediction bias: {latest['over_predict_ratio']:.0%}")
        if len(self.history) >= 24:
            drift = latest['wmape_pct'] - self.history[-24]['wmape_pct']
            if drift > 3.0:
                alerts.append(f"MAPE drift: +{drift:.1f}pp over 24h")
        return alerts


# Demo
monitor = MAPEMonitor(mape_threshold=15.0)
np.random.seed(42)
for hour in range(48):
    actuals = np.random.exponential(50, 100)
    preds = actuals + np.random.normal(3, 5 + hour * 0.2, 100)
    for a, p in zip(actuals, preds):
        monitor.record(float(a), float(p))
    result = monitor.flush_window()
    alerts = monitor.check_alerts()
    if hour % 12 == 0:
        print(f"Hour {hour}: WMAPE={result['wmape_pct']:.1f}%")
    for alert in alerts:
        print(f"  [ALERT] {alert}")

# MAPE monitoring configuration for demand forecasting pipeline
metrics:
  mape:
    variant: "wmape"          # Prefer WMAPE for zero-safety
    report_standard_mape: true  # Also compute standard MAPE for reference
    report_smape: false
    
    zero_handling:
      strategy: "skip"         # skip | clip | error
      clip_epsilon: 1e-6       # Min denominator for clip strategy
      max_zero_ratio: 0.10     # Alert if >10% zeros
    
    thresholds:
      excellent: 5.0           # MAPE < 5%
      good: 10.0               # 5% <= MAPE < 10%
      acceptable: 20.0         # 10% <= MAPE < 20%
      critical: 30.0           # MAPE >= 30% triggers alert
    
    # Asymmetry monitoring
    asymmetry:
      track_direction: true
      over_predict_alert: 0.65  # Alert if >65% over-predictions
      under_predict_alert: 0.65
    
    # Slicing for segment-level MAPE
    slices:
      - dimension: "product_category"
        values: ["electronics", "grocery", "fashion", "home"]
      - dimension: "demand_volume"
        values: ["low", "medium", "high"]
      - dimension: "region"
        values: ["north", "south", "east", "west"]
    
    # Companion metrics (always compute alongside MAPE)
    companion_metrics:
      - "mae"                   # Absolute error in original units
      - "median_absolute_percentage_error"  # Robust to outliers
      - "forecast_bias"         # Mean error (directional)
      - "p90_absolute_percentage_error"     # 90th percentile APE
    
    # Drift detection
    drift:
      baseline_window: "7d"    # Compare against last 7 days
      drift_threshold_pp: 3.0  # Alert on 3 percentage point increase
      check_interval: "1h"