Constructing an AI Agent to Detect and Handle Anomalies in Time-Series Data

As a knowledge scientist working on time-series forecasting, I even have run into anomalies and outliers greater than I can count. Across demand forecasting, finance, traffic, and sales data, I keep running into spikes and dips which can be hard to interpret.

Anomaly handling is often a gray area, rarely black or white, but indicators of deeper issues. Some anomalies are real signals like holidays, weather events, promotions, or viral moments; others are only data glitches, but each look the identical at first glance. The faster we detect anomalies in data, the faster motion will be taken to forestall poor performance and damage.

We’re coping with critical time-series data, and detecting anomalies is crucial. Should you remove a real event, a worthwhile signal data point is removed, and for those who keep a false alarm signal, the training data accommodates noise.

Most ML-based detectors flag spikes based on Z-scores, IQR thresholds, or other static methods with none context. With recent advancements in AI, now we have a greater choice to design an anomaly-handling agent that reasons about each case. An agent that detects unusual behavior, checks context, and decides whether to repair the information, keep it as an actual signal, or flag it for review.

In this text, we construct such an agent step-by-step that mixes easy statistical detection with an AI agent that acts as a primary line of defense for time-series data, reducing manual intervention while preserving the signals that matter most. We’ll detect and handle anomalies in COVID-19 data by autonomous decision-making based on the severity of the anomaly, using:

1. Live epidemiological data from the disease.sh API.
2. Statistical anomaly detection.
3. Severity classification.
4. A GroqCloud-powered AI agent that takes autonomous decisions whether to:
   • Fix the anomaly
   • Keep the anomaly
   • Flag anomaly for human review

That is agentic decision intelligence, not merely anomaly detection.

Why is traditional anomaly detection alone not enough?

There are traditional ML methods like isolation forests designed for anomaly detection, but they lack end-to-end decision orchestration. They’re unable to act on them quickly enough in production environments. We’re implementing an AI agent to fill this gap by turning raw anomaly scores into autonomous, end-to-end decisions dynamically on live data.

Traditional Anomaly Detection

The standard anomaly detection follows the pipeline approach as drawn below:

Limitations of Traditional Anomaly Detection

• Works on static rules and manually sets thresholds.
• It’s single-dimensional and handles easy data.
• No contextual reasoning.
• Human-driven decision making.
• Manual-driven motion.

Anomaly Detection and Handling with an AI Agent 

The AI Agent anomaly detection follows the pipeline approach as drawn below:

Why does this work higher in practice?

• Works on real-time data.
• It’s multidimensional and may handle complex data.
• Works on contextual reasoning.
• Adaptive & self-learning decision making.
• Take autonomous motion.

Selecting a sensible dataset for our example

We’re using real-world COVID-19 data to detect anomalies, because it is noisy, shows spikes, and the outcomes assist in the advance of public health.

What do we wish the AI Agent to make your mind up?

The goal is to repeatedly monitor COVID-19 data, find anomalies, define their severity, and take autonomous decisions and judge motion to be taken:

• Flag anomaly for human review
• Fix the anomaly
• Keep the anomaly

Data Source

For the information, we’re using free, live disease.sh data via API. This API provides data on day by day confirmed cases, deaths and recoveries. For the AI Agent implementation, we’re specializing in day by day case counts, which are perfect for anomaly detection.

How do the pieces fit together?

High-Level system architecture of the anomaly detection on COVID-19 data using an AI Agent is as follows:

Constructing the AI Agent Step-by-Step 

Let’s go step-by-step to grasp methods to load data using disease.sh, detect anomalies, classify them, and implement an AI agent that reasons and takes appropriate motion as per the severity of the anomalies.

Step 1: Install Required Libraries

Step one is to put in required libraries like phidata, groq, python-dotenv, tabulate, and streamlit.

pip install phidata
pip install groq
pip install python-dotenv #library to load .env file
pip install tabulate
pip install streamlit

Step 2: Environment File Set-up

Open your IDE and create a project folder, and under that folder, create an environmental file “.env” to store GROQ_API_KEY.

GROQ_API_KEY="your_groq_api_key_here"

Step 3: Data Ingestion

Before constructing any agent, we want a knowledge source that’s noisy enough to surface real anomalies, but structured enough to reason about. COVID-19 day by day case counts are an excellent fit as they contain reporting delays, sudden spikes, and regime changes. For simplicity, we deliberately restrict ourselves to a single univariate time series.

Load data from the disease.sh using request URL and extract the date and day by day case count based on the chosen country and the variety of days for which you would like to extract data. The info is converted right into a structured dataframe by parsing json, formatting date and sorting chronologically.

# ---------------------------------------
# DATA INGESTION (disease.sh)
# ---------------------------------------

def load_live_covid_data(country: str , days:int):
    url = f"https://disease.sh/v3/covid-19/historical/{country}?lastdays={days}"
    response = requests.get(url)
    data = response.json()["timeline"]["cases"]

    df = (
        pd.DataFrame(list(data.items()), columns=["Date", "Cases"])
        .assign(Date=lambda d: pd.to_datetime(d["Date"], format="%m/%d/%y"))
        .sort_values("Date")
        .reset_index(drop=True)
    )
    return df

Step 4: Anomalies Detection

We’ll now detect abnormal behavior in COVID-19 time-series data by detecting sudden spikes and rapid growth trends. Case counts are generally stable, and enormous deviations or sharp increases indicate meaningful anomalies. We’ll now detect anomalies using statistical methods and binary labeling for deterministic and reproducible anomaly detection. Two parameters are calculated to detect anomalies.

1. Spike Detection
   • A sudden spike in data is detected using the Z-score; if any data point falls outside the Z-score range, it have to be an anomaly.
2. Growth Rate Detection
   • The day-over-day growth rate is calculated; if it exceeds 40%, it’s flagged.

# ---------------------------------------
# ANOMALY DETECTION
# ---------------------------------------
def detect_anomalies(df):
   values = df["Cases"].values
   mean, std = values.mean(), values.std()

   spike_idx = [
       i for i, v in enumerate(values)
       if abs(v - mean) > 3 * std
   ]

   growth = np.diff(values) / np.maximum(values[:-1], 1)
   growth_idx = [i + 1 for i, g in enumerate(growth) if g > 0.4]

   anomalies = set(spike_idx + growth_idx)
   df["Anomaly"] = ["YES" if i in anomalies else "NO" for i in range(len(df))]

   return df

If there’s an anomaly in response to either spike or growth or with each parameters, the “Anomaly” is ready to “YES”; otherwise set to “NO”.

Step 5: Severity Classification

All anomalies will not be equal; we are going to classify them as ‘CRITICAL’, ‘WARNING’, or ‘MINOR’ to guide AI Agent decisions. Fixed rolling windows and rule-based thresholds are used to categorise severity. Severity is classed only when an anomaly exists; otherwise, Severity, Agent Decision, and Motion parameters within the dataframe are set to ‘blank’.

# ---------------------------------------
# CONFIG
# ---------------------------------------
ROLLING_WINDOW = 7
MIN_ABS_INCREASE = 500

# ---------------------------------------
# SEVERITY CLASSIFICATION
# ---------------------------------------
def compute_severity(df):
    df = df.sort_values("Date").reset_index(drop=True)
    df["Severity"] = ""
    df["Agent Decision"] = ""
    df["Action"] = ""
    for i in range(len(df)):
        if df.loc[i, "Anomaly"] == "YES":
            if i < ROLLING_WINDOW:
                df.loc[i, "Severity"] = ""

            curr = df.loc[i, "Cases"]
            baseline = df.loc[i - ROLLING_WINDOW:i- 1, "Cases"].mean()

            abs_inc = curr - baseline
            growth = abs_inc / max(baseline, 1)

            if abs_inc < MIN_ABS_INCREASE:
                df.loc[i, "Severity"] = ""
            if growth >= 1.0:
                df.loc[i, "Severity"] = "CRITICAL"
            elif growth >= 0.4:
                df.loc[i, "Severity"] = "WARNING"
            else:
                df.loc[i, "Severity"] = "MINOR"
    return df

Within the above code, to categorise the anomaly severity, each anomaly is compared with 7-day historical data (ROLLING_WINDOW = 7), and absolute and relative growth are calculated.

1. Absolute Growth

A MIN_ABS_INCREASE = 500 is defined as a config parameter where changes below this value are considered very small, a negligible change. If absolutely the growth is lower than MIN_ABS_INCREASE, then ignore it and keep the severity blank. Absolute growth detects meaningful real-world impact, doesn’t react to noise or minor fluctuations, and prevents false alarms when growth percentage is high.

2. Relative Growth:

Relative growth helps in detecting explosive trends. If growth is larger than or equal to 100% increase over baseline, it means a sudden outbreak, and it’s assigned as ‘CRITICAL’; if growth is larger than 40%, it means sustained acceleration and desires monitoring, and it’s assigned as ‘WARNING’; otherwise assigned as ‘MINOR’. 

After severity classification, it is prepared for the AI Agent to make an autonomous decision and motion.

Step 6: Construct Prompt for AI Agent

Below is the prompt that defines how the AI agent reasons and makes decisions based on structured context and predefined severity when an anomaly is detected.  The agent is restricted to 3 explicit actions and must return a single, deterministic response for protected automation.

def build_agent_prompt(obs):
    return f"""
You might be an AI monitoring agent for COVID-19 data.

Observed anomaly:
Date: {obs['date']}
Cases: {obs['cases']}
Severity: {obs['severity']}

Decision rules:
- FIX_ANOMALY: noise, reporting fluctuation
- KEEP_ANOMALY: real outbreak signal
- FLAG_FOR_REVIEW: severe or ambiguous anomaly

Respond with ONLY certainly one of:
FIX_ANOMALY
KEEP_ANOMALY
FLAG_FOR_REVIEW
"""

Three data points, i.e., date, variety of cases reported, and severity, are provided to the prompt explicitly, which helps the AI Agent to make a call autonomously.

Step 7: Create your Agent with GroqCloud

We at the moment are creating an autonomous AI agent using GroqCloud that makes intelligent contextual decisions on detected anomalies and their severities and takes appropriate actions. Three predefined actions for the AI Agent implement validated outputs only.

# ---------------------------------------
# BUILDING AI AGENT
# ---------------------------------------
agent = Agent(
    name="CovidAnomalyAgent",
    model=Groq(id="openai/gpt-oss-120b"),
    instructions="""
You might be an AI agent monitoring live COVID-19 time-series data.
Detect anomalies, determine in response to the anomaly:
"FIX_ANOMALY", "KEEP_ANOMALY", "FLAG_FOR_REVIEW"."""
)
for i in range(len(df)):
    if df.loc[i, "Anomaly"] == "YES":
        obs = build_observation(df, i)
        prompt = build_agent_prompt(obs)
        response = agent.run(prompt)

        decision = response.messages[-1].content.strip()
        decision = decision if decision in VALID_ACTIONS else "FLAG_FOR_REVIEW"
        df = agent_action(df, i, decision)

An AI agent named “CovidAnomalyAgent” is created, which uses an LLM model hosted by GroqCloud for fast and low-latency reasoning. AI Agent runs a well-defined prompt, observes data, contextual reasoning, makes an autonomous decision, and takes actions inside protected constraints.

An AI Agent will not be handling anomalies but making intelligent decisions for every detected anomaly. The agent’s decision accurately reflects anomaly severity and required motion.

# ---------------------------------------
# Agent ACTION DECIDER
# ---------------------------------------
def agent_action(df, idx,motion):
    df.loc[idx, "Agent Decision"] = motion

    if motion == "FIX_ANOMALY":
        fix_anomaly(df, idx)

    elif motion == "KEEP_ANOMALY":
        df.loc[idx, "Action"] = "Accepted as an actual outbreak signal"

    elif motion == "FLAG_FOR_REVIEW":
        df.loc[idx, "Action"] = "Flagged for human review"
    return df

AI Agent ignores normal data points with no anomaly and considers only data points with “ANOMALY= YES”. The AI agent is constrained to return only three valid decisions: “FIX_ANOMALY“, “KEEP_ANOMALY“, and “FLAG_FOR_REVIEW“, and accordingly, motion is taken as defined within the table below:

For minor anomalies, the AI agent routinely fixes the information, preserves valid anomalies as-is, and flags critical cases for human review.

Step 8: Fix Anomaly

Minor anomalies are attributable to reporting noise and are corrected using local rolling mean smoothing over recent historical values.

# ---------------------------------------
# FIX ANOMALY
# ---------------------------------------

def fix_anomaly(df, idx):
    window = df.loc[max(0, idx - 3):idx - 1, "Cases"]
    if len(window) > 0:
        df.loc[idx, "Cases"] = int(window.mean())

    df.loc[idx, "Severity"] = ""
    df.loc[idx, "Action"] = "Auto-corrected by an AI agent"

It takes the immediate 3 days of past data, calculates its mean, and smooths the anomaly by replacing its value with this average. By the local rolling mean smoothing approach, temporary spikes and data glitches will be handled. 

Once an anomaly is fixed, the information point isn’t any longer considered dangerous, and severity is intentionally removed to avoid confusion. “Motion” is updated to “Auto-corrected by an AI agent”.

Complete Code

Kindly undergo the whole code for the statistical anomaly detection and AI Agent implementation for anomaly handling.

https://github.com/rautmadhura4/anomaly_detection_agent/tree/major

Results

Let’s compare the outcomes for the country, “India,” with several types of severity detected and the way the AI Agent handles them.

Scenario 1: A Native Implementation

The primary attempt is a native implementation where we detect minor anomalies and the AI Agent fixes them routinely. Below is the snapshot of the COVID data table of India with severity.

We now have also implemented a Streamlit dashboard to review the AI Agent’s decisions and actions. Within the below result snapshot, you possibly can see that various minor anomalies are fixed by the AI Agent.

This works best when anomalies are localized noise moderately than regime changes.

Scenario 2: A Boundary Condition

Here, critical anomalies are detected, and the AI Agent raises a flag for review as shown within the snapshot of the COVID data table of India with severity.

On the Streamlit dashboard AI Agent’s decisions and actions are shown within the result snapshot. You’ll be able to see that each one the critical anomalies were flagged for human review by the AI Agent.

Severity gating prevents destructive auto-corrections in high-impact anomalies.

Scenario 3: A Limitation 

For the limitation scenario, warning and important anomalies are detected as shown within the snapshot of the COVID data table of India with severity.

On the Streamlit dashboard AI Agent’s decisions and actions are shown below within the result snapshot. You’ll be able to see that the critical anomaly is flagged for human review by AI Agent, however the WARNING anomaly is routinely fixed. In lots of real settings, a WARNING-level anomaly needs to be preserved and monitored moderately than corrected.

This failure highlights why WARNING thresholds needs to be tuned and why human review stays essential.

Use the whole code and check out anomaly detection for the COVID-19 dataset, with different parameters.

Future Scope and Enhancements

We now have used a really limited dataset and implemented rule-based anomaly detection, but in the longer term, some enhancements will be done within the AI Agent implementation:

1. In our implementation, an anomaly is detected, and a call is made based on case count only. In the longer term, data will be more elaborate with features like hospitalization records, vaccination data, and others.

2. Anomaly detection is completed here using statistical methods, which may also be ML-driven in the longer term to discover more complex patterns.

3. Now, now we have implemented a single-agent architecture; in the longer term multi-agent architecture will be implemented to enhance scalability, clarity, and resilience.
4. In the longer term human feedback loop must also take care to make improved decisions.

Final Takeaways

Smarter AI agents enable operational AI that makes decisions using contextual reasoning, takes motion to repair anomalies, and escalates to humans when needed. There are some practical takeaways to take note while constructing an AI Agent for anomaly detection:

• To detect anomalies, use statistical methods and implement AI agents for contextual decision-making.

• Minor anomalies are protected to be autocorrected as they’re generally reported as noise. Critical should never be autocorrected and flagged for review by domain experts in order that real-world signals don’t get suppressed.

• This AI agent must not be utilized in situations where anomalies directly trigger irreversible actions.

When statistical methods and an AI agent approach are combined properly, they transform anomaly detection from just an alerting system right into a controlled, decision-driven system without compromising safety.