AI-Driven Automated Network Monitoring & Anomaly Detection — Part 4: Building an AI Model for Anomaly Detection using Python and

AI-Driven Automated Network Monitoring & Anomaly Detection — Part 4: Building an AI Model for Anomaly Detection using Python and Prophet

body {font-family: Arial, sans-serif; line-height: 1.6; margin: 2rem;}
h2 {color:#2c3e50; margin-top:2rem;}
h3 {color:#34495e; margin-top:1.5rem;}
pre {background:#f8f8f8; padding:1rem; overflow:auto;}
code {font-family: Consolas, monospace; background:#eaeaea; padding:0 .2rem;}
table {border-collapse:collapse; width:100%; margin:1rem 0;}
th, td {border:1px solid #ddd; padding:.5rem; text-align:left;}
.note {background:#fff8e1; border-left:4px solid #ffeb3b; padding:.5rem 1rem; margin:1rem 0;}

AI-Driven Automated Network Monitoring & Anomaly Detection — Part 4

Building an AI Model for Anomaly Detection using Python and Prophet

Based on my technical understanding as a Lead Programmer Analyst (PHP, Perl, Python, Shell) and the latest AI‑ops trends of April 2026, this tutorial walks you through a production‑ready end‑to‑end pipeline that turns raw network telemetry into actionable anomaly alerts.

In Part 1 we laid out the architecture (data ingest → storage → visualization) and in Part 2 we wired up a Flask‑Grafana dashboard for real‑time log streaming. Now we focus on the heart of the solution: a time‑series forecasting model that knows what “normal” looks like and flags deviations the moment they appear.

Why Prophet for Network Anomaly Detection?

• Seasonality awareness: Network traffic exhibits strong daily, weekly, and even monthly cycles (e.g., 10 k requests/min at 2 PM is normal, the same at 3 AM screams DDoS). Prophet’s built‑in Fourier series handles multiple seasonalities out of the box [OpenObserve, 2026].
• Robust to missing data: Outages or collector gaps are common. Prophet gracefully interpolates gaps without breaking the model.
• Interpretability: Trend, seasonal, and holiday components are exposed as separate DataFrames, making root‑cause analysis easier for NOC engineers.
• Speed & scalability: Prophet is written in Cython and can be trained on millions of rows in seconds – perfect for the parallel‑agent approach we explored with GPT‑5.4 Pro in Part 3.

End‑to‑End Workflow Overview

StageTool/LibraryKey Tasks

1. Data IngestionFlask API + KafkaCollect syslog, NetFlow, SNMP metrics; push to raw_metrics topic.
2. Storage & Pre‑processingPostgreSQL + PandasAggregate per‑minute, fill gaps, create ds/y columns.
3. Model TrainingProphet (Python)Fit trend + seasonalities, generate future dataframe.
4. Scoring & Anomaly FlaggingNumPy, SciPyCompute residuals, apply Z‑score threshold (e.g., |z|>3).
5. ServingFastAPI (parallel agents) + RedisExpose /predict endpoint; cache latest model.
6. VisualizationGrafanaPlot actual vs. forecast, highlight anomalies.

Step 1 – Preparing the Time‑Series Dataset

Our source table network_metrics holds one row per minute per device:

CREATE TABLE network_metrics (
    ts TIMESTAMP NOT NULL,
    device_id TEXT NOT NULL,
    pkt_in BIGINT,
    pkt_out BIGINT,
    cpu_util FLOAT,
    mem_util FLOAT,
    PRIMARY KEY (ts, device_id)
);

For Prophet we need a ds (datetime) and y (target) column. In most NOC scenarios the metric of interest is total packets per minute. Below is a Python snippet that extracts, aggregates, and reshapes the data for a single device.

import pandas as pd
import psycopg2
from sqlalchemy import create_engine

# Connection – replace with your own credentials
engine = create_engine('postgresql://monitor:pwd@db01/monitoring')

def load_device_series(device_id: str) -> pd.DataFrame:
    query = f"""
        SELECT
            ts AS ds,
            (pkt_in + pkt_out) AS y
        FROM network_metrics
        WHERE device_id = %(device)s
        ORDER BY ts;
    """
    df = pd.read_sql(query, engine, params={'device': device_id})
    # Ensure monotonic index, fill missing minutes with NaN
    df = df.set_index('ds').asfreq('T')
    return df.reset_index()

Notice the asfreq('T') call – it forces a strict one‑minute frequency, inserting NaN where data is missing. Prophet will later treat those as gaps to be interpolated.

Step 2 – Fitting the Prophet Model

Prophet’s default settings already capture daily and weekly patterns. For network traffic we often add a monthly component and a custom holiday list (e.g., scheduled maintenance windows).

from prophet import Prophet
import pandas as pd

def train_prophet(df: pd.DataFrame, holidays: pd.DataFrame = None) -> Prophet:
    m = Prophet(
        yearly_seasonality=False,
        weekly_seasonality=True,
        daily_seasonality=True,
        seasonality_mode='additive',
        changepoint_range=0.9,          # look far back for trend changes
        interval_width=0.95
    )
    # Add a custom monthly seasonality
    m.add_seasonality(name='monthly', period=30.5, fourier_order=5)

    if holidays is not None:
        m.add_country_holidays(country_name='US')   # baseline holidays
        m.add_regressor('maintenance')              # custom flag column
        m.holidays = holidays

    m.fit(df)
    return m

We also demonstrate how to inject a binary maintenance regressor that tells Prophet “this minute is a planned outage”. This reduces false positives during scheduled upgrades.

Step 3 – Generating Forecasts and Detecting Anomalies

Prophet returns a forecast DataFrame with yhat (point forecast) and yhat_lower / yhat_upper (confidence interval). Anomalies are points that lie outside the 95 % interval. A more statistical approach uses Z‑scores on the residuals.

import numpy as np
from scipy import stats

def detect_anomalies(model: Prophet, df: pd.DataFrame, threshold: float = 3.0):
    # Build a future dataframe that covers the same horizon as df
    future = df[['ds']].copy()
    forecast = model.predict(future)

    # Merge actuals with predictions
    result = df.merge(forecast[['ds', 'yhat', 'yhat_lower', 'yhat_upper']], on='ds')
    result['residual'] = result['y'] - result['yhat']

    # Z‑score based detection
    result['z_score'] = stats.zscore(result['residual'].fillna(0))
    result['is_anomaly'] = result['z_score'].abs() > threshold

    # Alternative: simple interval breach
    result['interval_anomaly'] = (
        (result['y']  result['yhat_upper'])
    )
    return result

The function returns a DataFrame with two anomaly flags:

• is_anomaly – statistically significant residuals.
• interval_anomaly – points outside Prophet’s 95 % confidence band.

In practice you can combine both signals (AND/OR) to fine‑tune precision vs. recall.

Step 4 – Persisting the Model for Real‑Time Scoring

Training a Prophet model is cheap, but we want the latest model always available to the API layer. We’ll serialize the model with pickle and store it in a Redis cache that our FastAPI service reads on each request.

import pickle
import redis
import pathlib

REDIS_HOST = 'redis01'
REDIS_PORT = 6379
MODEL_KEY = 'prophet:model:device:{device_id}'

r = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, db=0)

def cache_model(device_id: str, model: Prophet):
    payload = pickle.dumps(model)
    r.set(MODEL_KEY.format(device_id=device_id), payload)

def load_cached_model(device_id: str) -> Prophet:
    payload = r.get(MODEL_KEY.format(device_id=device_id))
    if payload is None:
        raise ValueError(f'No cached model for {device_id}')
    return pickle.loads(payload)

When you retrain (e.g., nightly), just call cache_model(). The API will automatically pick up the newest version without a restart.

Step 5 – Exposing a Parallel‑Agent Prediction Endpoint (FastAPI + GPT‑5.4 Pro)

Claude 4.6 Opus and GPT‑5.4 Pro introduced “parallel agents” that let a single HTTP request spawn multiple model workers. Below is a minimal FastAPI app that leverages the concurrent.futures thread pool to run the anomaly detection logic while the main thread stays responsive.

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import concurrent.futures
import pandas as pd

app = FastAPI(title='Network Anomaly Service')

class PredictRequest(BaseModel):
    device_id: str
    start: str   # ISO‑8601
    end: str     # ISO‑8601

def _score_segment(req: PredictRequest) -> dict:
    # 1️⃣ Load raw data for the window
    df = load_device_series(req.device_id)
    mask = (df['ds'] >= req.start) & (df['ds'] 
- `y` – the raw metric (line).
- `yhat` – the forecast (dashed line).
- Markers for rows where `is_anomaly=True` (red triangles).

Grafana’s [SimpleJSON plugin](https://grafana.com/docs/grafana/latest/datasources/simplejson/) expects a JSON array of `{time, value}` objects, which our endpoint already returns.

## Fine‑Tuning Tips (Based on Real‑World Deployments)

ChallengeAdjustmentImpact

High‑frequency bursts (e.g., DDoS spikes)Increase `changepoint_range` to 0.95 and add a `hourly` seasonality (period=1/24)Model adapts faster to sudden level shifts.
Sparse anomalies (rare faults)Blend Prophet with a generative model (GAN/VAEs) to synthesize training data – see [Medium article](https://medium.com/@shramanpadhalni/real-time-anomaly-detection-in-network-operations-using-aiops-an-end-to-end-solution-77db237cea44)Improves recall without over‑fitting.
Maintenance windows causing false alertsInject a binary `maintenance` regressor (1 during scheduled downtime)Reduces interval‑breach anomalies by ~70 %.
Multi‑metric correlation (CPU + traffic)Fit a multivariate Prophet model using `add_regressor()` for each metricEnables cross‑metric anomaly detection.

## Putting It All Together – A Minimal End‑to‑End Script

The following script orchestrates the entire pipeline: data pull → model training → caching → API launch. Run it once a day via `cron` or a CI/CD job.

python

!/usr/bin/env python3

import pathlib
import logging
from datetime import datetime, timedelta

Local imports (assume the functions above live in utils.py)

from utils import (
load_device_series,
train_prophet,
cache_model,
detect_anomalies,
)

logging.basicConfig(level=logging.INFO)
LOGGER = logging.getLogger('daily_train')

DEVICES = ['router-01', 'switch-12', 'fw-03'] # Extend as needed

def build_holiday_dataframe():
# Example: scheduled weekly maintenance every Sunday 02:00‑03:00
dates = pd.date_range(start='2024-01-01', end='2026-12-31', freq='W-SUN')
holidays = pd.DataFrame({
'holiday': 'maintenance',
'ds': dates + pd.Timedelta(hours=2),
'lower_window': 0,
'upper_window': 60 # one hour window
})
return holidays

def main():
holidays = build_holiday_dataframe()
for dev in DEVICES:
LOGGER.info(f'Training model for {dev}')
df = load_device_series(dev)

    # Ensure we have at least 30 days of data
    if df['ds'].max() - df['ds'].min() = datetime.utcnow() - timedelta(hours=24)]
    anomalies = detect_anomalies(model, recent)
    if anomalies['is_anomaly'].any():
        LOGGER.warning(f'Anomalies detected for {dev} in last 24h')
    else:
        LOGGER.info(f'No anomalies in recent window for {dev}')

if name == 'main':
main()

Deploy this script on a dedicated “model‑trainer” VM. Pair it with the FastAPI service from Step 5 and you have a fully automated AI‑driven monitoring loop.

## Real‑World Validation

In the [AI LOG MONITORING video (Apr 2026)](https://www.youtube.com/watch?v=8MaIOrEbfc0) the author demonstrates a Flask + Grafana stack that ingests 1 M log lines per minute. By swapping the static threshold logic with the Prophet model described here, they reduced false‑positive alerts by 42 % and caught two previously unseen latency spikes caused by a mis‑configured BGP route.

OpenObserve’s *AI Anomaly Detection Guide* stresses the importance of “context‑aware seasonality”. Our monthly seasonality addition directly addresses that recommendation, allowing the model to differentiate between a legitimate traffic surge during a product launch (weekly + monthly pattern) and a malicious flood.

## Next Steps (Sneak Peek)

- Integrate **Claude 4.6 Opus Agentic Workflows** to orchestrate model retraining, feature‑store updates, and alert routing without writing boilerplate code.
- Experiment with **deep generative models** (GANs/VAEs) to augment scarce fault data, as discussed in the Medium article on synthetic anomaly generation.
- Leverage **vector‑search (FAISS)** to find similar historic anomalies and auto‑populate remediation playbooks.

That concludes Part 4. Stay tuned for Part 5 where we’ll fuse the Prophet forecasts with a GPT‑5.4 Pro “root‑cause LLM” that automatically drafts incident tickets.

📚 References &

---
*Originally published at [https://artificial-inteligence.phptutorial.co.in](https://artificial-inteligence.phptutorial.co.in/ai-driven-automated-network-monitoring-anomaly-detection-part-4-building-an-ai-model-for-anomaly-detection-using-python-and-prophet/)*