Implementing Anomaly Detection in WFM Forecasts vs Actual Call Volumes

How-To-Articles
1

Implementing Anomaly Detection in WFM Forecasts vs Actual Call Volumes

What This Guide Covers

You are building an automated anomaly detection system that continuously compares Genesys Cloud WFM (or NICE CXone WFM) forecast volumes against actual real-time call volumes. When complete, your system will detect significant deviations the moment they occur-identifying when actual volume is tracking 30%+ above forecast within the first 30 minutes of a day-send automated alerts to the WFM Intraday Manager, and trigger pre-configured contingency actions (emergency agent notifications, overtime authorization, or overflow queue activation) based on the severity of the deviation.

Prerequisites, Roles & Licensing

  • Genesys Cloud: Any CX tier with WFM.
  • Permissions required:
    • Analytics > Queue Observation > View
    • Analytics > Queue Aggregates > View
    • WFM > Forecast > View (for accessing published forecast data)
  • Infrastructure:
    • A scheduled polling service (AWS Lambda on EventBridge Scheduler, every 15 minutes).
    • A notification gateway (Slack, PagerDuty, or SMS gateway).
    • Time-series storage for tracking deviation trends (CloudWatch Metrics or DynamoDB).

The Implementation Deep-Dive

1. The Forecast Accuracy Problem

WFM forecasts are generated days or weeks in advance based on historical patterns. In practice, external events cause actual volumes to deviate significantly:

  • A system outage drives 3× normal inbound volume (customers calling to report the issue).
  • A viral social media post causes a complaint spike.
  • A seasonal promotion launch was not communicated to the WFM team.
  • An IVR configuration error is routing calls incorrectly.

Without automated detection, WFM teams typically discover the deviation 45-60 minutes after it starts, when SLA has already collapsed. Automated anomaly detection compresses this to under 15 minutes.

2. Extracting Forecast Data from Genesys WFM

import requests
from datetime import datetime, timedelta

GENESYS_API = "https://api.mypurecloud.com"

def get_published_forecast_intervals(
    management_unit_id: str,
    week_start: str,  # YYYY-MM-DD format (Monday of the WFM week)
    access_token: str
) -> dict[str, float]:
    """
    Fetches published WFM forecast data for the current week.
    Returns a dict: {interval_datetime_str: expected_offered_volume}
    """
    headers = {"Authorization": f"Bearer {access_token}"}
    
    # Get the list of forecasts for the management unit
    resp = requests.get(
        f"{GENESYS_API}/api/v2/workforcemanagement/managementunits/{management_unit_id}/weeks/{week_start}/schedules/generationresults",
        headers=headers
    )
    resp.raise_for_status()
    
    # Alternatively, if using the Forecast API directly:
    forecast_resp = requests.get(
        f"{GENESYS_API}/api/v2/workforcemanagement/managementunits/{management_unit_id}/weeks/{week_start}/forecasts",
        headers=headers
    )
    
    forecast_data = {}
    for interval in forecast_resp.json().get("offeredPerInterval", []):
        ts = interval.get("intervalStartTime")
        offered = interval.get("offered", 0)
        forecast_data[ts] = offered
    
    return forecast_data

3. Extracting Actual Volume from the Analytics API

def get_actual_volume_by_interval(
    queue_ids: list[str],
    start_time: datetime,
    end_time: datetime,
    access_token: str
) -> dict[str, int]:
    """
    Fetches actual offered volume per 15-minute interval from the Aggregates API.
    Returns: {interval_start_time: actual_offered_count}
    """
    headers = {"Authorization": f"Bearer {access_token}", "Content-Type": "application/json"}
    
    payload = {
        "interval": f"{start_time.isoformat()}Z/{end_time.isoformat()}Z",
        "granularity": "PT15M",  # 15-minute intervals
        "groupBy": ["queueId"],
        "filter": {
            "type": "orFilter",
            "filters": [
                {"type": "term", "dimension": "queueId", "value": qid}
                for qid in queue_ids
            ]
        },
        "metrics": ["nOffered"]
    }
    
    resp = requests.post(
        f"{GENESYS_API}/api/v2/analytics/queues/aggregates/query",
        headers=headers,
        json=payload
    )
    resp.raise_for_status()
    
    actual_volume = {}
    for result in resp.json().get("results", []):
        for interval_data in result.get("data", []):
            if interval_data.get("metric") == "nOffered":
                ts = result.get("interval")
                actual_volume[ts] = actual_volume.get(ts, 0) + interval_data.get("stats", {}).get("count", 0)
    
    return actual_volume

4. The Anomaly Detection Engine

from dataclasses import dataclass

@dataclass
class ForecastAnomaly:
    interval_start: str
    forecast_volume: float
    actual_volume: int
    deviation_pct: float
    severity: str  # "MINOR", "MODERATE", "CRITICAL"

def detect_anomalies(
    forecast: dict[str, float],
    actuals: dict[str, int],
    minor_threshold: float = 0.20,   # 20% deviation
    moderate_threshold: float = 0.40,  # 40% deviation
    critical_threshold: float = 0.60   # 60% deviation
) -> list[ForecastAnomaly]:
    """
    Compares forecast vs actual volumes and returns a list of anomalous intervals.
    Only evaluates intervals that have passed (actuals are available).
    """
    anomalies = []
    
    for interval_ts, forecast_vol in forecast.items():
        actual_vol = actuals.get(interval_ts)
        
        if actual_vol is None:
            continue  # Interval hasn't occurred yet
        
        if forecast_vol == 0:
            continue  # Skip zero-forecast intervals (overnight)
        
        deviation = (actual_vol - forecast_vol) / forecast_vol
        abs_deviation = abs(deviation)
        
        # Only flag intervals with significant deviation (above/below)
        if abs_deviation < minor_threshold:
            continue
        
        if abs_deviation >= critical_threshold:
            severity = "CRITICAL"
        elif abs_deviation >= moderate_threshold:
            severity = "MODERATE"
        else:
            severity = "MINOR"
        
        anomalies.append(ForecastAnomaly(
            interval_start=interval_ts,
            forecast_volume=forecast_vol,
            actual_volume=actual_vol,
            deviation_pct=deviation * 100,
            severity=severity
        ))
    
    return anomalies

def build_alert_message(anomalies: list[ForecastAnomaly]) -> str:
    """Builds a structured Slack alert message from detected anomalies."""
    critical = [a for a in anomalies if a.severity == "CRITICAL"]
    moderate = [a for a in anomalies if a.severity == "MODERATE"]
    
    if not critical and not moderate:
        return ""
    
    lines = ["🚨 *WFM FORECAST DEVIATION ALERT*"]
    
    if critical:
        lines.append(f"\n🔴 *CRITICAL DEVIATIONS ({len(critical)} intervals):*")
        for a in critical[:3]:  # Show top 3
            direction = "ABOVE" if a.deviation_pct > 0 else "BELOW"
            lines.append(f"  • `{a.interval_start}`: {abs(a.deviation_pct):.0f}% {direction} forecast "
                         f"(Forecast: {a.forecast_volume:.0f}, Actual: {a.actual_volume})")
    
    if moderate:
        lines.append(f"\n🟡 *MODERATE DEVIATIONS ({len(moderate)} intervals)*")
    
    lines.append("\n*Recommended Actions:*")
    if critical:
        lines.append("• Activate overflow queue or emergency overtime bank")
        lines.append("• Notify operations manager immediately")
    
    return "\n".join(lines)

5. Sustained Deviation Detection (Moving Average)

A single interval spike may be noise. A sustained 30%+ deviation over 3 consecutive intervals is a genuine pattern requiring action.

def detect_sustained_anomaly(
    anomalies: list[ForecastAnomaly],
    consecutive_threshold: int = 3
) -> bool:
    """Returns True if the last N intervals all show significant positive deviation."""
    recent = sorted(anomalies, key=lambda x: x.interval_start, reverse=True)[:consecutive_threshold]
    
    if len(recent) < consecutive_threshold:
        return False
    
    return all(a.deviation_pct > 20 and a.severity in ("MODERATE", "CRITICAL") for a in recent)

Validation, Edge Cases & Troubleshooting

Edge Case 1: Forecast Granularity Mismatch

Your WFM publishes forecasts at 30-minute intervals, but the Analytics Aggregates API returns data at 15-minute intervals. Comparing them directly produces a mismatch.
Solution: Aggregate the 15-minute actuals to 30-minute intervals before comparison. Sum the nOffered values of each pair of consecutive 15-minute intervals to produce a 30-minute actual total that aligns with the forecast granularity.

Edge Case 2: Systematic Forecast Bias

If your WFM model consistently underestimates Monday morning volume by 15%, your anomaly detector fires every Monday morning with false alerts. The operations team starts ignoring them.
Solution: Apply a historical bias correction to the forecast before comparison. Track the rolling 8-week average deviation for each day-of-week and 30-minute interval. If Monday 9:00 AM consistently runs 15% over forecast, apply a +15% correction factor to the forecast before computing deviations.

Edge Case 3: Analytics API Latency

Actual volume data for the last completed interval may not be available in the Aggregates API for 5-15 minutes after the interval ends. Querying at the boundary of an interval returns partial data, making the deviation appear larger than it actually is.
Solution: Offset your polling by 15 minutes. When evaluating at 10:30 AM, only analyze intervals up to 10:00 AM (not 10:15 AM or 10:30 AM). This ensures all data for analyzed intervals is fully committed to the analytics data mart.

Official References