Insider Threat Detection with Graphs
Detect malicious insiders using ML anomaly detection and graph analysis.
Introduction
30 Second Summary
Every organization's biggest security risk isn't external hackers. It's the trusted employees who already have access to sensitive data and systems.
In this project, you will build an insider threat detection system that combines machine learning anomaly detection with graph-based behavioral analysis to identify suspicious users from enterprise activity logs.
What You'll Build
A Python pipeline that generates synthetic enterprise logs, engineers behavioral features, trains anomaly detection models, and builds a behavior graph to surface suspicious user clusters with ranked risk scores.
By the end of this project, you'll have:
- A synthetic data generator producing realistic CERT-style activity logs with injected insider threat scenarios across five event categories.
- Trained Isolation Forest and XGBoost anomaly detection models with precision, recall, and ROC-AUC evaluation against ground truth labels.
- A NetworkX behavior graph that correlates users by shared behavioral patterns and produces combined risk scores using graph centrality metrics.
- Secret Mission: Build an automated investigation agent that traverses the behavior graph and generates structured threat narrative reports for flagged users.
Are there any prerequisites?
This project assumes basic Python proficiency and familiarity with scikit-learn concepts like training models and train/test splits. You do not need prior experience with graph theory or cybersecurity.
Before We Start
Insider threats are one of the hardest problems in cybersecurity. Unlike external attackers, insiders already have legitimate access to systems. Detecting them requires analyzing behavioral patterns over time, which is where machine learning and graph analysis come in.
In this step, you'll set up your Python environment and install the libraries needed for data generation, anomaly detection, and graph theory analysis. You'll also create all the project files you'll fill in over the next steps.
In this step, get ready to:
- Verify Python 3.11+ and create a virtual environment.
- Install scikit-learn, XGBoost, NetworkX, and supporting libraries.
- Create the project file structure.
Verify your Python version
This project requires Python 3.11 or higher because scikit-learn 1.9.0 and NetworkX 3.6.1 both require it.
- Open your terminal by pressing Cmd+Space (macOS) or the Windows key (Windows) and searching for Terminal (macOS) or PowerShell (Windows).
- Check your Python version by running this command:
macOS
python3 --versionWindows
python --versionYou should see a version number in the output. Check whether it is 3.11 or higher.
✔️ I see version 3.11 or higher
Your Python version is ready. Move on to the next instruction below.
ⓧ I see an older version
Your system has a Python version below 3.11. On macOS, the default python3 often points to an older version that shipped with the OS. You need to install Python 3.11+ alongside it.
- Visit python.org/downloads and download the latest Python 3.11+ installer for your platform.
- Run the installer. On Windows, check the Add python.exe to PATH box on the first screen.
- After installing, close and reopen your terminal, then re-run the version check.
- On macOS, if python3 --version still shows the old version, try python3.11 --version or python3.12 --version to confirm the new install is available. Use whichever versioned command works for all Python commands going forward.
Use pyenv for version management
If you frequently work with multiple Python versions, consider using pyenv (macOS/Linux) or pyenv-win (Windows) to install and switch between versions cleanly.
ⓧ Command not found
Python is not installed on your system. You need to install Python 3.11 or higher.
- Visit python.org/downloads and download the latest Python 3.11+ installer for your platform.
- Run the installer. On Windows, check the Add python.exe to PATH box on the first screen.
- On macOS, open the downloaded .pkg file and follow the prompts.
- After installing, close and reopen your terminal, then run the version check again.
Create your project directory and virtual environment
A virtual environment isolates your project's packages from your system Python. This prevents version conflicts with other projects on your machine.
- Move to your Desktop by running this command:
cd ~/Desktop- Create your project folder and move into it by running these commands:
mkdir insider-threat-detection
cd insider-threat-detection- Create a virtual environment by running this command:
macOS
python3 -m venv venvWindows
python -m venv venv- Activate the virtual environment:
macOS
source venv/bin/activateWindows
venv\Scripts\activateYour terminal prompt should now show (venv) at the beginning. This confirms the virtual environment is active.
Not seeing (venv) in your prompt?
On Windows, you may get a permissions error when activating. Open PowerShell as Administrator and run Set-ExecutionPolicy RemoteSigned, then try activating again in your regular terminal.
On macOS, make sure you used source venv/bin/activate (not just venv/bin/activate). Help me activate my virtual environment.
Install the required packages
This project uses several libraries. scikit-learn handles machine learning. XGBoost provides a gradient-boosted classifier. NetworkX builds and analyzes graphs.
- Install all required packages by running this command:
pip install scikit-learn==1.9.0 xgboost-cpu==3.2.0 networkx==3.6.1 pandas numpy matplotlibWhy xgboost-cpu instead of xgboost?
The main xgboost package includes GPU dependencies that fail to install on Windows without CUDA. The xgboost-cpu package provides identical CPU functionality without the GPU dependency, so it installs cleanly on all platforms.
Installation may take a minute or two. You should see a success message for each package.
- Verify everything installed correctly by opening a Python shell and importing each package:
macOS
python3 -c "import sklearn; import xgboost; import networkx; import pandas; import numpy; import matplotlib; print('All imports successful!')"Windows
python -c "import sklearn; import xgboost; import networkx; import pandas; import numpy; import matplotlib; print('All imports successful!')"You should see All imports successful! printed. If any import fails, the error message will name the missing package.
Seeing an import error?
Make sure your virtual environment is active (you should see (venv) in your prompt). If you installed the packages without the venv active, deactivate, reactivate, and run the pip install command again.
If xgboost fails to import, confirm you installed xgboost-cpu (not xgboost). Help me fix my package imports.
Create the project file structure
Your project will have five Python files, each handling a distinct part of the detection pipeline.
- Create all five project files by running this command:
macOS
touch data_generator.py feature_engineering.py detector.py behavior_graph.py main.pyWindows
New-Item data_generator.py, feature_engineering.py, detector.py, behavior_graph.py, main.py- Confirm the files exist by running:
lsYou should see behavior_graph.py, data_generator.py, detector.py, feature_engineering.py, and main.py listed in the output (along with your venv folder).
What is each file for?
- data_generator.py produces synthetic enterprise activity logs with injected threat scenarios.
- feature_engineering.py transforms raw events into numerical feature vectors for ML models.
- detector.py trains and evaluates Isolation Forest and XGBoost anomaly detectors.
- behavior_graph.py builds a user similarity graph and computes centrality-based risk metrics.
- main.py orchestrates the full pipeline from data generation to risk scoring.
Files not showing up?
Make sure you ran the touch or New-Item command from inside the insider-threat-detection folder. Run pwd (macOS) or Get-Location (Windows) to confirm your current directory. Help me find my project files.
Your environment is ready. Next up, you'll build a synthetic data generator that produces realistic enterprise activity logs with injected insider threat scenarios.
Generate Synthetic Insider Threat Data
Your environment is set up and ready to go. Now it's time to build the foundation of the entire detection system: the data.
Real insider threat datasets (like the CERT Insider Threat Test Dataset from CMU) contain gigabytes of activity logs and require lengthy downloads. Instead, you'll build a synthetic data generator that mimics the same structure: logon events, file access, emails, web browsing, and USB device connections. This gives you full control over the threat scenarios injected into the data.
In this step, get ready to:
- Build a data generator that produces 5 types of activity events for 200 users over 30 days.
- Inject 5 malicious users with distinct threat behaviors that escalate over time.
- Validate the generated data to confirm the injected patterns are detectable.
Build the synthetic data generator
The generator simulates a 30-day window of enterprise activity. Normal users follow predictable patterns (busy on weekdays, quiet on weekends). Malicious users start normal but escalate suspicious behavior over time, mimicking real insider threat progression.
- Open data_generator.py in your code editor and add the module docstring and imports by pasting this code:
"""Synthetic insider threat data generator.
Produces CERT-style activity logs for N users over D days,
with injected malicious behavioral patterns.
"""
import random
from datetime import datetime, timedelta
import numpy as np
import pandas as pdWhat do these imports do?
- random and numpy handle random number generation for simulating realistic event distributions.
- datetime and timedelta create timestamps spanning the 30-day simulation window.
- pandas structures all generated events into a DataFrame for downstream analysis.
You won't see a change yet. The next chunk adds the main generator function that uses these imports.
- Below the imports, add the main generate_insider_threat_data function by pasting this code:
def generate_insider_threat_data(
n_users: int = 200,
n_days: int = 30,
n_malicious: int = 5,
seed: int = 42,
) -> tuple[pd.DataFrame, list[str]]:
"""
Generate synthetic user activity logs with insider threats.
Returns:
events_df: DataFrame of all activity events
malicious_users: list of user IDs who are malicious
"""
rng = np.random.default_rng(seed)
random.seed(seed)
user_ids = [f"USER_{i:04d}" for i in range(n_users)]
malicious_users = rng.choice(user_ids, size=n_malicious, replace=False).tolist()
start_date = datetime(2024, 1, 1)
events = []
for user_id in user_ids:
is_malicious = user_id in malicious_users
if is_malicious:
threat_type = random.choice(["exfiltration", "flight_risk", "credential_abuse"])
else:
threat_type = None
for day_offset in range(n_days):
current_date = start_date + timedelta(days=day_offset)
is_weekday = current_date.weekday() < 5
base_logons = rng.poisson(3 if is_weekday else 1)
base_files = rng.poisson(8 if is_weekday else 1)
base_emails = rng.poisson(12 if is_weekday else 2)
base_http = rng.poisson(20 if is_weekday else 5)
base_usb = rng.poisson(0.3)
escalation = min(day_offset / n_days, 1.0)
if threat_type == "exfiltration" and day_offset > n_days * 0.5:
base_files += int(rng.poisson(5 * escalation))
base_usb += int(rng.poisson(2 * escalation))
elif threat_type == "flight_risk" and day_offset > n_days * 0.4:
base_emails += int(rng.poisson(8 * escalation))
elif threat_type == "credential_abuse" and day_offset > n_days * 0.3:
base_logons += int(rng.poisson(4 * escalation))
for _ in range(base_logons):
hour = _generate_hour(rng, is_weekday, is_malicious and threat_type == "credential_abuse")
events.append({
"timestamp": current_date + timedelta(hours=hour, minutes=rng.integers(0, 60)),
"user_id": user_id,
"event_type": "logon",
"after_hours": hour < 7 or hour > 19,
"failed": rng.random() < (0.15 if threat_type == "credential_abuse" else 0.02),
"machine_id": f"PC_{rng.integers(1, 5 if not is_malicious else 12):03d}",
})
for _ in range(base_files):
hour = _generate_hour(rng, is_weekday, is_malicious and threat_type == "exfiltration")
events.append({
"timestamp": current_date + timedelta(hours=hour, minutes=rng.integers(0, 60)),
"user_id": user_id,
"event_type": "file_access",
"after_hours": hour < 7 or hour > 19,
"sensitive": rng.random() < (0.4 if threat_type == "exfiltration" else 0.05),
"machine_id": f"PC_{rng.integers(1, 5):03d}",
})
for _ in range(base_emails):
hour = _generate_hour(rng, is_weekday, False)
events.append({
"timestamp": current_date + timedelta(hours=hour, minutes=rng.integers(0, 60)),
"user_id": user_id,
"event_type": "email",
"after_hours": hour < 7 or hour > 19,
"external": rng.random() < (0.6 if threat_type == "flight_risk" else 0.1),
"large_attachment": rng.random() < (0.3 if threat_type == "exfiltration" else 0.05),
"recipient_domain": _generate_domain(rng, is_malicious and threat_type == "flight_risk"),
})
for _ in range(base_http):
hour = _generate_hour(rng, is_weekday, False)
events.append({
"timestamp": current_date + timedelta(hours=hour, minutes=rng.integers(0, 60)),
"user_id": user_id,
"event_type": "http",
"after_hours": hour < 7 or hour > 19,
"domain": f"site_{rng.integers(1, 50 if not is_malicious else 200):04d}.com",
"upload_mb": float(rng.exponential(0.1 if not is_malicious else 0.5)),
})
for _ in range(base_usb):
hour = _generate_hour(rng, is_weekday, is_malicious and threat_type == "exfiltration")
events.append({
"timestamp": current_date + timedelta(hours=hour, minutes=rng.integers(0, 60)),
"user_id": user_id,
"event_type": "usb",
"after_hours": hour < 7 or hour > 19,
"device_id": f"USB_{rng.integers(1, 3 if not is_malicious else 8):03d}",
})
events_df = pd.DataFrame(events)
events_df = events_df.sort_values("timestamp").reset_index(drop=True)
return events_df, malicious_usersWhat does this function do?
This is the core generator. It simulates 200 users going about their daily work over 30 days, producing five event types: logon, file access, email, HTTP browsing, and USB connections.
- Normal users follow Poisson-distributed activity rates that spike on weekdays and drop on weekends.
- Five users are randomly marked malicious and assigned a threat type: exfiltration (after-hours file access + USB), flight_risk (mass external emails), or credential_abuse (unusual logon patterns).
- The escalation variable makes malicious behavior intensify over time, mimicking how real insiders gradually push boundaries before acting.
- Each event carries type-specific attributes (e.g., after_hours, sensitive, external) that become features in later steps.
You won't see a change yet. The function references two helper functions that you'll add next.
- Below the main function, add the two helper functions by pasting this code:
def _generate_hour(rng: np.random.Generator, is_weekday: bool, bias_after_hours: bool) -> int:
"""Generate a realistic hour for an event."""
if bias_after_hours and rng.random() < 0.4:
return int(rng.choice([*range(0, 7), *range(20, 24)]))
if is_weekday:
return int(rng.normal(12, 3).clip(6, 22))
return int(rng.normal(14, 4).clip(0, 23))
def _generate_domain(rng: np.random.Generator, personal_bias: bool) -> str:
"""Generate a recipient email domain."""
if personal_bias and rng.random() < 0.5:
return rng.choice(["gmail.com", "yahoo.com", "protonmail.com", "outlook.com"])
return "company.com"What do these helpers do?
- _generate_hour creates realistic timestamps. Normal users cluster around noon on weekdays. Malicious users with bias_after_hours have a 40% chance of acting between 8pm and 7am.
- _generate_domain simulates email recipients. Flight risk users send 50% of emails to personal domains (gmail, yahoo, protonmail) instead of the corporate domain.
- Save data_generator.py.
Seeing a syntax error on save?
- Check that the triple-quoted docstrings use three double-quotes ("""), not single quotes.
- Make sure the f-strings inside events.append() use matching quote types. The outer dict uses double quotes, so f-string expressions are fine inside.
- Confirm the helper functions are outside (below) the main function, not indented inside it.
help me fix a syntax error in my data_generator.py file
✔️ Awesome, I've got everything!
Great. Double check you've saved data_generator.py before moving on.
ⓧ I'd like to double check the full code
"""Synthetic insider threat data generator.
Produces CERT-style activity logs for N users over D days,
with injected malicious behavioral patterns.
"""
import random
from datetime import datetime, timedelta
import numpy as np
import pandas as pd
def generate_insider_threat_data(
n_users: int = 200,
n_days: int = 30,
n_malicious: int = 5,
seed: int = 42,
) -> tuple[pd.DataFrame, list[str]]:
"""
Generate synthetic user activity logs with insider threats.
Returns:
events_df: DataFrame of all activity events
malicious_users: list of user IDs who are malicious
"""
rng = np.random.default_rng(seed)
random.seed(seed)
user_ids = [f"USER_{i:04d}" for i in range(n_users)]
malicious_users = rng.choice(user_ids, size=n_malicious, replace=False).tolist()
start_date = datetime(2024, 1, 1)
events = []
for user_id in user_ids:
is_malicious = user_id in malicious_users
if is_malicious:
threat_type = random.choice(["exfiltration", "flight_risk", "credential_abuse"])
else:
threat_type = None
for day_offset in range(n_days):
current_date = start_date + timedelta(days=day_offset)
is_weekday = current_date.weekday() < 5
base_logons = rng.poisson(3 if is_weekday else 1)
base_files = rng.poisson(8 if is_weekday else 1)
base_emails = rng.poisson(12 if is_weekday else 2)
base_http = rng.poisson(20 if is_weekday else 5)
base_usb = rng.poisson(0.3)
escalation = min(day_offset / n_days, 1.0)
if threat_type == "exfiltration" and day_offset > n_days * 0.5:
base_files += int(rng.poisson(5 * escalation))
base_usb += int(rng.poisson(2 * escalation))
elif threat_type == "flight_risk" and day_offset > n_days * 0.4:
base_emails += int(rng.poisson(8 * escalation))
elif threat_type == "credential_abuse" and day_offset > n_days * 0.3:
base_logons += int(rng.poisson(4 * escalation))
for _ in range(base_logons):
hour = _generate_hour(rng, is_weekday, is_malicious and threat_type == "credential_abuse")
events.append({
"timestamp": current_date + timedelta(hours=hour, minutes=rng.integers(0, 60)),
"user_id": user_id,
"event_type": "logon",
"after_hours": hour < 7 or hour > 19,
"failed": rng.random() < (0.15 if threat_type == "credential_abuse" else 0.02),
"machine_id": f"PC_{rng.integers(1, 5 if not is_malicious else 12):03d}",
})
for _ in range(base_files):
hour = _generate_hour(rng, is_weekday, is_malicious and threat_type == "exfiltration")
events.append({
"timestamp": current_date + timedelta(hours=hour, minutes=rng.integers(0, 60)),
"user_id": user_id,
"event_type": "file_access",
"after_hours": hour < 7 or hour > 19,
"sensitive": rng.random() < (0.4 if threat_type == "exfiltration" else 0.05),
"machine_id": f"PC_{rng.integers(1, 5):03d}",
})
for _ in range(base_emails):
hour = _generate_hour(rng, is_weekday, False)
events.append({
"timestamp": current_date + timedelta(hours=hour, minutes=rng.integers(0, 60)),
"user_id": user_id,
"event_type": "email",
"after_hours": hour < 7 or hour > 19,
"external": rng.random() < (0.6 if threat_type == "flight_risk" else 0.1),
"large_attachment": rng.random() < (0.3 if threat_type == "exfiltration" else 0.05),
"recipient_domain": _generate_domain(rng, is_malicious and threat_type == "flight_risk"),
})
for _ in range(base_http):
hour = _generate_hour(rng, is_weekday, False)
events.append({
"timestamp": current_date + timedelta(hours=hour, minutes=rng.integers(0, 60)),
"user_id": user_id,
"event_type": "http",
"after_hours": hour < 7 or hour > 19,
"domain": f"site_{rng.integers(1, 50 if not is_malicious else 200):04d}.com",
"upload_mb": float(rng.exponential(0.1 if not is_malicious else 0.5)),
})
for _ in range(base_usb):
hour = _generate_hour(rng, is_weekday, is_malicious and threat_type == "exfiltration")
events.append({
"timestamp": current_date + timedelta(hours=hour, minutes=rng.integers(0, 60)),
"user_id": user_id,
"event_type": "usb",
"after_hours": hour < 7 or hour > 19,
"device_id": f"USB_{rng.integers(1, 3 if not is_malicious else 8):03d}",
})
events_df = pd.DataFrame(events)
events_df = events_df.sort_values("timestamp").reset_index(drop=True)
return events_df, malicious_users
def _generate_hour(rng: np.random.Generator, is_weekday: bool, bias_after_hours: bool) -> int:
"""Generate a realistic hour for an event."""
if bias_after_hours and rng.random() < 0.4:
return int(rng.choice([*range(0, 7), *range(20, 24)]))
if is_weekday:
return int(rng.normal(12, 3).clip(6, 22))
return int(rng.normal(14, 4).clip(0, 23))
def _generate_domain(rng: np.random.Generator, personal_bias: bool) -> str:
"""Generate a recipient email domain."""
if personal_bias and rng.random() < 0.5:
return rng.choice(["gmail.com", "yahoo.com", "protonmail.com", "outlook.com"])
return "company.com"Validate the generated data
Before moving on, you need to confirm the generator produces realistic logs with detectable malicious patterns. If the signal isn't visible in raw counts, no ML model will find it either.
- Run the generator and print summary statistics by running this in your terminal:
python -c "
from data_generator import generate_insider_threat_data
import pandas as pd
events_df, malicious_users = generate_insider_threat_data()
print(f'Total events: {len(events_df):,}')
print(f'Unique users: {events_df[\"user_id\"].nunique()}')
print(f'Malicious users: {malicious_users}')
print(f'\nEvent type distribution:')
print(events_df['event_type'].value_counts().to_string())
print(f'\nClass balance: {len(malicious_users)} malicious / {events_df[\"user_id\"].nunique() - len(malicious_users)} benign')
"What should I see?
You should see output showing approximately 250,000+ total events spread across 200 users. The event type distribution should be dominated by HTTP events (highest base rate of 20/day), followed by emails (12/day), file accesses (8/day), logons (3/day), and USB events (rare at 0.3/day).
The 5 malicious user IDs are printed so you can track them through later steps. The class balance confirms the extreme imbalance: only 2.5% of users are threats.
- Verify that malicious users show elevated activity by running this comparison:
python -c "
from data_generator import generate_insider_threat_data
import pandas as pd
events_df, malicious_users = generate_insider_threat_data()
for user_id in malicious_users:
user_events = events_df[events_df['user_id'] == user_id]
print(f'{user_id}: {len(user_events)} events')
print(f' Event types: {user_events[\"event_type\"].value_counts().to_dict()}')
after_hours = user_events[user_events.get('after_hours', False) == True]
print(f' After-hours events: {len(after_hours)}')
print()
# Compare with a benign user
benign = [u for u in events_df['user_id'].unique() if u not in malicious_users][0]
benign_events = events_df[events_df['user_id'] == benign]
print(f'Benign comparison ({benign}): {len(benign_events)} events')
print(f' Event types: {benign_events[\"event_type\"].value_counts().to_dict()}')
"What patterns should I look for?
Malicious users should show noticeably higher event counts in their threat category. An exfiltration user will have more file access and USB events than the benign baseline. A flight risk user will have elevated email counts. A credential abuse user will have more logon events.
The after-hours event count for malicious users should also be higher. These differences confirm the injected patterns are producing detectable signal in the raw data.
Seeing a ModuleNotFoundError?
- Make sure you're running the command from your project directory (the folder containing data_generator.py).
- Confirm your virtual environment is still activated. You should see the environment name in your terminal prompt.
- If you see an error about numpy or pandas, re-run the pip install command from Step 1.
help me fix a ModuleNotFoundError when running my data generator
Your synthetic data is generating realistic activity patterns with clearly detectable malicious signals. Next up, you'll transform these raw event logs into structured feature vectors that ML models can learn from.
Engineer Behavioral Features
Raw event logs are useful for auditing, but machine learning models need structured numerical inputs. Each log entry is a single event (a logon, a file access, an email). To feed these into an anomaly detector, you need to compress thousands of events per user into a fixed-size feature vector that captures behavioral patterns.
In this step, you will build a feature engineering pipeline that transforms raw events into 17 daily behavioral metrics per user, then aggregates those into 34 user-level features. You will also validate that the engineered features contain enough signal to distinguish malicious users from benign ones.
In this step, get ready to:
- Transform raw event logs into 17 daily behavioral features across 5 activity categories.
- Aggregate daily features into user-level vectors and normalize with StandardScaler.
- Validate that malicious users produce measurably different feature values than benign users.
Transform raw event logs into user-day feature vectors
The feature engineering pipeline needs to compute daily behavioral metrics for each user, then summarize those daily values into a stable per-user profile. Computing both the mean and the standard deviation captures not just typical behavior, but how consistently a user behaves.
- Open feature_engineering.py in your editor and add the module docstring and imports by pasting this code:
"""Feature engineering pipeline for insider threat detection.
Transforms raw event logs into user-level behavioral feature vectors.
"""
import numpy as np
import pandas as pd
from sklearn.preprocessing import StandardScalerWhat do these imports do?
- numpy provides numerical operations for array manipulation.
- pandas gives you DataFrames for tabular data processing and the groupby aggregation you will use to summarize daily features.
- StandardScaler from scikit-learn normalizes features to zero mean and unit variance, which prevents high-magnitude features from dominating the anomaly detector.
- Below the imports, add the engineer_features() function by pasting the following code:
def engineer_features(events_df: pd.DataFrame, n_days: int = 30) -> pd.DataFrame:
"""Transform raw events into user-level feature vectors.
Computes 17 daily features per user, then aggregates to
mean and std per user (34 features total).
Returns:
DataFrame with user_id as index, 34 feature columns.
"""
users = events_df["user_id"].unique()
dates = pd.to_datetime(events_df["timestamp"]).dt.date.unique()
daily_features = []
for user_id in users:
user_events = events_df[events_df["user_id"] == user_id].copy()
user_events["date"] = pd.to_datetime(user_events["timestamp"]).dt.date
for date in dates:
day_events = user_events[user_events["date"] == date]
logon_events = day_events[day_events["event_type"] == "logon"]
file_events = day_events[day_events["event_type"] == "file_access"]
email_events = day_events[day_events["event_type"] == "email"]
http_events = day_events[day_events["event_type"] == "http"]
usb_events = day_events[day_events["event_type"] == "usb"]
features = {
"user_id": user_id,
"date": date,
# Logon features
"total_logons": len(logon_events),
"after_hours_logons": logon_events["after_hours"].sum() if len(logon_events) > 0 else 0,
"failed_logons": logon_events["failed"].sum() if len(logon_events) > 0 else 0,
"unique_machines": logon_events["machine_id"].nunique() if len(logon_events) > 0 else 0,
# File features
"files_accessed": len(file_events),
"sensitive_files": file_events["sensitive"].sum() if len(file_events) > 0 else 0,
"after_hours_file_access": (file_events["after_hours"].sum() if len(file_events) > 0 else 0),
# Email features
"emails_sent": len(email_events),
"external_emails": email_events["external"].sum() if len(email_events) > 0 else 0,
"large_attachments": email_events["large_attachment"].sum() if len(email_events) > 0 else 0,
"unique_external_recipients": (
email_events[email_events["external"] == True]["recipient_domain"].nunique()
if len(email_events) > 0 else 0
),
# HTTP features
"http_requests": len(http_events),
"unique_domains": http_events["domain"].nunique() if len(http_events) > 0 else 0,
"upload_volume_mb": http_events["upload_mb"].sum() if len(http_events) > 0 else 0.0,
# Device features
"usb_connects": len(usb_events),
"after_hours_usb": usb_events["after_hours"].sum() if len(usb_events) > 0 else 0,
"unique_usb_devices": usb_events["device_id"].nunique() if len(usb_events) > 0 else 0,
}
daily_features.append(features)
daily_df = pd.DataFrame(daily_features)
# Aggregate to user level: mean and std of each daily feature
feature_cols = [
"total_logons", "after_hours_logons", "failed_logons", "unique_machines",
"files_accessed", "sensitive_files", "after_hours_file_access",
"emails_sent", "external_emails", "large_attachments", "unique_external_recipients",
"http_requests", "unique_domains", "upload_volume_mb",
"usb_connects", "after_hours_usb", "unique_usb_devices",
]
agg_dict = {col: ["mean", "std"] for col in feature_cols}
user_features = daily_df.groupby("user_id").agg(agg_dict)
# Flatten multi-level columns
user_features.columns = [f"{col}_{stat}" for col, stat in user_features.columns]
user_features = user_features.fillna(0)
return user_featuresWhat does this function do?
This function performs a two-stage transformation on your raw event data.
- Stage 1 (daily features): For every user on every day, it counts 17 behavioral metrics across 5 categories. Logon features track login volume, after-hours access, and failed attempts. File features capture sensitive document access. Email features measure external communication. HTTP features record browsing breadth and upload volume. Device features track USB usage.
- Stage 2 (user aggregation): It uses pandas groupby().agg() to compute the mean and standard deviation of each daily metric per user. This produces 34 features (17 means + 17 standard deviations) that describe both typical behavior and behavioral consistency.
- The fillna(0) call handles users who had identical values every day (producing NaN for standard deviation) by treating zero variance as the safest assumption.
- Save feature_engineering.py.
Seeing an IndentationError?
- Make sure the entire function body is indented consistently with 4 spaces. Mixing tabs and spaces causes Python to reject the file.
- Confirm the features = { dictionary and the daily_features.append(features) call are both inside the inner for date in dates: loop.
Still stuck? Help me fix an IndentationError in my engineer_features function
Handle the feature matrix
Now you need a second function that normalizes the features so that no single metric dominates the anomaly detection model. A user with 200 HTTP requests per day would overwhelm a user with 2 USB connections without normalization.
- In feature_engineering.py, add the normalize_features() function below engineer_features() by pasting this code:
def normalize_features(user_features: pd.DataFrame) -> tuple[pd.DataFrame, StandardScaler]:
"""Apply StandardScaler normalization to feature matrix.
Returns:
normalized DataFrame and the fitted scaler.
"""
scaler = StandardScaler()
normalized_array = scaler.fit_transform(user_features)
normalized_df = pd.DataFrame(
normalized_array,
index=user_features.index,
columns=user_features.columns,
)
return normalized_df, scalerWhat does StandardScaler do?
StandardScaler transforms each feature column to have a mean of 0 and a standard deviation of 1. After scaling, a value of 2.0 in any column means "2 standard deviations above the population average". This makes features directly comparable regardless of their original units or magnitudes.
The function also returns the fitted scaler object so you can apply the same transformation to new data later without re-fitting.
- Save feature_engineering.py.
- Test the pipeline by running this command in your terminal:
python -c "
from data_generator import generate_insider_threat_data
from feature_engineering import engineer_features, normalize_features
events_df, malicious_users = generate_insider_threat_data()
user_features = engineer_features(events_df, n_days=30)
normalized_features, scaler = normalize_features(user_features)
print(f'Feature matrix shape: {normalized_features.shape}')
print(f'NaN values: {normalized_features.isna().sum().sum()}')
print(f'Users: {normalized_features.shape[0]}, Features: {normalized_features.shape[1]}')
"You should see output showing a feature matrix shape of (200, 34) with 0 NaN values.
Seeing a ModuleNotFoundError?
- Make sure your virtual environment is activated. You should see the environment name in your terminal prompt.
- Confirm you are running the command from inside your project directory where data_generator.py and feature_engineering.py both live.
Need more help? Help me debug a ModuleNotFoundError when importing my feature engineering module
Explore the data to confirm signal exists
Before building anomaly detection models, you should confirm that malicious users actually produce different feature values than benign users. If the features look identical for both groups, the models will have nothing to learn from.
- Run this comparison script to check for signal separation:
python -c "
import pandas as pd
from data_generator import generate_insider_threat_data
from feature_engineering import engineer_features
events_df, malicious_users = generate_insider_threat_data()
user_features = engineer_features(events_df, n_days=30)
labels = pd.Series(
[1 if uid in malicious_users else 0 for uid in user_features.index],
index=user_features.index,
)
malicious_means = user_features[labels == 1].mean()
benign_means = user_features[labels == 0].mean()
comparison = pd.DataFrame({
'malicious_mean': malicious_means,
'benign_mean': benign_means,
'ratio': malicious_means / (benign_means + 1e-10),
}).sort_values('ratio', ascending=False)
print('Top features separating malicious from benign users:')
print(comparison.head(10).to_string())
"You should see a table where several features have ratios significantly above 1.0. Features like after_hours_usb_mean, sensitive_files_mean, and external_emails_mean should show malicious users producing notably higher values than benign users.
Why check for signal before modeling?
This step mirrors real-world data science practice. If the ratio between malicious and benign means is close to 1.0 across all features, your feature engineering has failed. The model would have no patterns to learn from. Seeing strong separation here confirms that the data generator injected detectable behavioral differences and that your features capture them.
✔️ Awesome, I've got everything!
Great. Double check you have saved feature_engineering.py and confirm both your feature matrix shape and signal comparison ran successfully.
ⓧ I'd like to double check the full code
Here is the complete feature_engineering.py file for comparison:
"""Feature engineering pipeline for insider threat detection.
Transforms raw event logs into user-level behavioral feature vectors.
"""
import numpy as np
import pandas as pd
from sklearn.preprocessing import StandardScaler
def engineer_features(events_df: pd.DataFrame, n_days: int = 30) -> pd.DataFrame:
"""Transform raw events into user-level feature vectors.
Computes 17 daily features per user, then aggregates to
mean and std per user (34 features total).
Returns:
DataFrame with user_id as index, 34 feature columns.
"""
users = events_df["user_id"].unique()
dates = pd.to_datetime(events_df["timestamp"]).dt.date.unique()
daily_features = []
for user_id in users:
user_events = events_df[events_df["user_id"] == user_id].copy()
user_events["date"] = pd.to_datetime(user_events["timestamp"]).dt.date
for date in dates:
day_events = user_events[user_events["date"] == date]
logon_events = day_events[day_events["event_type"] == "logon"]
file_events = day_events[day_events["event_type"] == "file_access"]
email_events = day_events[day_events["event_type"] == "email"]
http_events = day_events[day_events["event_type"] == "http"]
usb_events = day_events[day_events["event_type"] == "usb"]
features = {
"user_id": user_id,
"date": date,
# Logon features
"total_logons": len(logon_events),
"after_hours_logons": logon_events["after_hours"].sum() if len(logon_events) > 0 else 0,
"failed_logons": logon_events["failed"].sum() if len(logon_events) > 0 else 0,
"unique_machines": logon_events["machine_id"].nunique() if len(logon_events) > 0 else 0,
# File features
"files_accessed": len(file_events),
"sensitive_files": file_events["sensitive"].sum() if len(file_events) > 0 else 0,
"after_hours_file_access": (file_events["after_hours"].sum() if len(file_events) > 0 else 0),
# Email features
"emails_sent": len(email_events),
"external_emails": email_events["external"].sum() if len(email_events) > 0 else 0,
"large_attachments": email_events["large_attachment"].sum() if len(email_events) > 0 else 0,
"unique_external_recipients": (
email_events[email_events["external"] == True]["recipient_domain"].nunique()
if len(email_events) > 0 else 0
),
# HTTP features
"http_requests": len(http_events),
"unique_domains": http_events["domain"].nunique() if len(http_events) > 0 else 0,
"upload_volume_mb": http_events["upload_mb"].sum() if len(http_events) > 0 else 0.0,
# Device features
"usb_connects": len(usb_events),
"after_hours_usb": usb_events["after_hours"].sum() if len(usb_events) > 0 else 0,
"unique_usb_devices": usb_events["device_id"].nunique() if len(usb_events) > 0 else 0,
}
daily_features.append(features)
daily_df = pd.DataFrame(daily_features)
# Aggregate to user level: mean and std of each daily feature
feature_cols = [
"total_logons", "after_hours_logons", "failed_logons", "unique_machines",
"files_accessed", "sensitive_files", "after_hours_file_access",
"emails_sent", "external_emails", "large_attachments", "unique_external_recipients",
"http_requests", "unique_domains", "upload_volume_mb",
"usb_connects", "after_hours_usb", "unique_usb_devices",
]
agg_dict = {col: ["mean", "std"] for col in feature_cols}
user_features = daily_df.groupby("user_id").agg(agg_dict)
# Flatten multi-level columns
user_features.columns = [f"{col}_{stat}" for col, stat in user_features.columns]
user_features = user_features.fillna(0)
return user_features
def normalize_features(user_features: pd.DataFrame) -> tuple[pd.DataFrame, StandardScaler]:
"""Apply StandardScaler normalization to feature matrix.
Returns:
normalized DataFrame and the fitted scaler.
"""
scaler = StandardScaler()
normalized_array = scaler.fit_transform(user_features)
normalized_df = pd.DataFrame(
normalized_array,
index=user_features.index,
columns=user_features.columns,
)
return normalized_df, scalerYour feature pipeline is complete. The strong signal separation confirms that your data generator is working and the feature engineering captures meaningful behavioral differences. Next up, you will train anomaly detection models to automatically identify which users are suspicious.
Detect Anomalies with ML Models
Your feature vectors are ready. Each of the 200 users is now represented by 34 numbers that capture their behavioral patterns across logon, file, email, HTTP, and USB activity.
Now you need to answer the core question: which users are behaving suspiciously? You will train two different models to answer this. Isolation Forest works without labels (unsupervised), flagging users who look statistically unlike their peers. XGBoost uses the ground truth labels (supervised) to learn what malicious behavior looks like directly.
In this step, get ready to:
- Train an Isolation Forest to flag anomalous users without labels.
- Train an XGBoost classifier with stratified cross-validation to handle class imbalance.
- Build an evaluation function that compares both models' precision, recall, F1, and ROC-AUC.
Train an Isolation Forest as the unsupervised baseline
Isolation Forest isolates anomalies by randomly partitioning data. Points that can be isolated in fewer partitions are more anomalous. It needs no labels, making it ideal for real-world insider threat detection where you rarely know who the bad actors are.
- Open detector.py and add the module docstring and imports by pasting this code:
"""Anomaly detection models for insider threat detection.
Implements Isolation Forest (unsupervised) and XGBoost (supervised) approaches.
"""
import numpy as np
import pandas as pd
from sklearn.ensemble import IsolationForest
from sklearn.metrics import (
classification_report,
f1_score,
precision_score,
recall_score,
roc_auc_score,
)
from sklearn.model_selection import StratifiedKFold
from xgboost import XGBClassifierWhat are these imports for?
- IsolationForest is the unsupervised anomaly detector from scikit-learn.
- The metrics imports (precision_score, recall_score, f1_score, roc_auc_score) measure how well each model catches malicious users.
- StratifiedKFold ensures each cross-validation fold preserves the 2.5% malicious ratio.
- XGBClassifier is the supervised gradient boosting model from the xgboost-cpu package.
- Below the imports, add the train_isolation_forest() function:
def train_isolation_forest(
features_df: pd.DataFrame,
contamination: float = 0.025,
seed: int = 42,
) -> tuple[IsolationForest, pd.Series]:
"""Train Isolation Forest and return anomaly scores.
Returns:
Fitted model and Series of anomaly scores (lower = more anomalous).
"""
model = IsolationForest(
contamination=contamination,
random_state=seed,
n_estimators=200,
)
model.fit(features_df)
# score_samples returns negative scores; more negative = more anomalous
scores = pd.Series(
model.score_samples(features_df),
index=features_df.index,
name="anomaly_score",
)
return model, scoresWhat does this code do?
The contamination=0.025 parameter tells Isolation Forest to expect 2.5% of users to be anomalous. This matches the 5 malicious users out of 200 total.
The score_samples() method returns a score for each user where lower (more negative) values mean the user is more anomalous. The function wraps these scores in a pandas Series so you can easily look up any user's anomaly score by their ID.
- Save detector.py.
Seeing a red underline on the imports?
Make sure your virtual environment is activated. Your editor needs to use the Python interpreter from your virtual environment to find the installed packages.
help me fix import errors for scikit-learn or xgboost in my detector.py
Train an XGBoost classifier as the supervised comparator
Isolation Forest knows nothing about which users are actually malicious. XGBoost can learn directly from labeled examples, but with only 5 malicious users in 200, the class imbalance is extreme. Stratified cross-validation and the scale_pos_weight parameter address this.
- Below train_isolation_forest(), add the train_xgboost_cv() function:
def train_xgboost_cv(
features_df: pd.DataFrame,
labels: pd.Series,
n_splits: int = 5,
seed: int = 42,
) -> tuple[XGBClassifier, dict]:
"""Train XGBoost with stratified cross-validation.
Returns:
Model trained on full data, and dict of mean CV metrics.
"""
n_positive = labels.sum()
n_negative = len(labels) - n_positive
scale_weight = n_negative / max(n_positive, 1)
cv_metrics = {"precision": [], "recall": [], "f1": [], "roc_auc": []}
skf = StratifiedKFold(n_splits=n_splits, shuffle=True, random_state=seed)
for train_idx, val_idx in skf.split(features_df, labels):
X_train = features_df.iloc[train_idx]
y_train = labels.iloc[train_idx]
X_val = features_df.iloc[val_idx]
y_val = labels.iloc[val_idx]
model = XGBClassifier(
n_estimators=100,
max_depth=4,
learning_rate=0.1,
scale_pos_weight=scale_weight,
random_state=seed,
eval_metric="logloss",
verbosity=0,
)
model.fit(X_train, y_train)
y_pred = model.predict(X_val)
y_proba = model.predict_proba(X_val)[:, 1]
cv_metrics["precision"].append(precision_score(y_val, y_pred, zero_division=0))
cv_metrics["recall"].append(recall_score(y_val, y_pred, zero_division=0))
cv_metrics["f1"].append(f1_score(y_val, y_pred, zero_division=0))
cv_metrics["roc_auc"].append(roc_auc_score(y_val, y_proba) if y_val.sum() > 0 else 0.0)
# Train final model on all data
final_model = XGBClassifier(
n_estimators=100,
max_depth=4,
learning_rate=0.1,
scale_pos_weight=scale_weight,
random_state=seed,
eval_metric="logloss",
verbosity=0,
)
final_model.fit(features_df, labels)
mean_metrics = {k: np.mean(v) for k, v in cv_metrics.items()}
return final_model, mean_metricsWhat does this code do?
The function computes scale_pos_weight as the ratio of benign to malicious users (195/5 = 39). This tells XGBoost to penalize misclassifying the rare malicious class 39x more heavily than benign users.
Stratified 5-fold cross-validation splits the 200 users into 5 groups while preserving the 2.5% malicious ratio in each fold. Each fold trains on 160 users and validates on 40. The function collects precision, recall, F1, and ROC-AUC from each fold, then averages them.
After cross-validation, a final model trains on ALL 200 users. This is the model you will use for predictions going forward. The CV metrics tell you how well it generalizes.
- Save detector.py.
Why cross-validation instead of a single train/test split?
With only 5 malicious users, a single 80/20 split might put 0 or 1 malicious users in the test set. Stratified 5-fold CV guarantees every fold sees at least 1 malicious user, giving you reliable metric estimates despite the tiny dataset.
Seeing indentation errors?
The nested for loop and model fitting inside it must be indented consistently. Make sure the model = XGBClassifier(...) block inside the loop uses the same indentation level as X_train = features_df.iloc[train_idx].
help me fix indentation in my train_xgboost_cv function
Evaluate and compare results
You have two models that produce scores, but you need a consistent way to evaluate Isolation Forest against ground truth labels. The evaluation function converts continuous anomaly scores into binary predictions using the contamination rate as a threshold.
- Below train_xgboost_cv(), add the evaluate_isolation_forest() function:
def evaluate_isolation_forest(
scores: pd.Series,
labels: pd.Series,
contamination: float = 0.025,
) -> dict:
"""Evaluate Isolation Forest predictions against ground truth."""
threshold = scores.quantile(contamination)
predictions = (scores <= threshold).astype(int)
return {
"precision": precision_score(labels, predictions, zero_division=0),
"recall": recall_score(labels, predictions, zero_division=0),
"f1": f1_score(labels, predictions, zero_division=0),
"roc_auc": roc_auc_score(labels, -scores),
}What does this code do?
Isolation Forest gives each user a continuous score, not a binary "malicious or not" label. To evaluate it, you need a cutoff. The function takes the bottom 2.5% of scores (the most anomalous users) and labels them as predicted-malicious.
Notice -scores is passed to roc_auc_score. This flips the sign so that higher values correspond to more anomalous users, which is what ROC-AUC expects (higher score = more likely positive class).
- Save detector.py.
- Verify the module loads correctly by running this command in your terminal:
python -c "from detector import train_isolation_forest, train_xgboost_cv, evaluate_isolation_forest; print('All detector functions loaded successfully.')"You should see All detector functions loaded successfully. printed with no errors.
How do the two models compare?
Isolation Forest needs no labels. It flags users purely based on statistical isolation from their peers. This is powerful in the real world where you rarely have confirmed insider threat labels.
XGBoost learns directly from labeled examples, so it achieves higher precision when labels are available. The tradeoff is that it requires labeled training data, which is expensive to obtain in practice.
When you run these models on your data in the next step, you will see that XGBoost identifies the exact malicious users with fewer false positives, while Isolation Forest catches anomalous behavior patterns that might not have labels yet.
Module not found error?
Make sure you are running the command from your project directory (the folder containing detector.py). Also confirm your virtual environment is still activated.
help me fix a ModuleNotFoundError when importing from detector.py
✔️ Awesome, I've got everything!
Great. Double check you have saved detector.py with all three functions: train_isolation_forest(), train_xgboost_cv(), and evaluate_isolation_forest().
ⓧ I'd like to double check the full code
"""Anomaly detection models for insider threat detection.
Implements Isolation Forest (unsupervised) and XGBoost (supervised) approaches.
"""
import numpy as np
import pandas as pd
from sklearn.ensemble import IsolationForest
from sklearn.metrics import (
classification_report,
f1_score,
precision_score,
recall_score,
roc_auc_score,
)
from sklearn.model_selection import StratifiedKFold
from xgboost import XGBClassifier
def train_isolation_forest(
features_df: pd.DataFrame,
contamination: float = 0.025,
seed: int = 42,
) -> tuple[IsolationForest, pd.Series]:
"""Train Isolation Forest and return anomaly scores.
Returns:
Fitted model and Series of anomaly scores (lower = more anomalous).
"""
model = IsolationForest(
contamination=contamination,
random_state=seed,
n_estimators=200,
)
model.fit(features_df)
# score_samples returns negative scores; more negative = more anomalous
scores = pd.Series(
model.score_samples(features_df),
index=features_df.index,
name="anomaly_score",
)
return model, scores
def train_xgboost_cv(
features_df: pd.DataFrame,
labels: pd.Series,
n_splits: int = 5,
seed: int = 42,
) -> tuple[XGBClassifier, dict]:
"""Train XGBoost with stratified cross-validation.
Returns:
Model trained on full data, and dict of mean CV metrics.
"""
n_positive = labels.sum()
n_negative = len(labels) - n_positive
scale_weight = n_negative / max(n_positive, 1)
cv_metrics = {"precision": [], "recall": [], "f1": [], "roc_auc": []}
skf = StratifiedKFold(n_splits=n_splits, shuffle=True, random_state=seed)
for train_idx, val_idx in skf.split(features_df, labels):
X_train = features_df.iloc[train_idx]
y_train = labels.iloc[train_idx]
X_val = features_df.iloc[val_idx]
y_val = labels.iloc[val_idx]
model = XGBClassifier(
n_estimators=100,
max_depth=4,
learning_rate=0.1,
scale_pos_weight=scale_weight,
random_state=seed,
eval_metric="logloss",
verbosity=0,
)
model.fit(X_train, y_train)
y_pred = model.predict(X_val)
y_proba = model.predict_proba(X_val)[:, 1]
cv_metrics["precision"].append(precision_score(y_val, y_pred, zero_division=0))
cv_metrics["recall"].append(recall_score(y_val, y_pred, zero_division=0))
cv_metrics["f1"].append(f1_score(y_val, y_pred, zero_division=0))
cv_metrics["roc_auc"].append(roc_auc_score(y_val, y_proba) if y_val.sum() > 0 else 0.0)
# Train final model on all data
final_model = XGBClassifier(
n_estimators=100,
max_depth=4,
learning_rate=0.1,
scale_pos_weight=scale_weight,
random_state=seed,
eval_metric="logloss",
verbosity=0,
)
final_model.fit(features_df, labels)
mean_metrics = {k: np.mean(v) for k, v in cv_metrics.items()}
return final_model, mean_metrics
def evaluate_isolation_forest(
scores: pd.Series,
labels: pd.Series,
contamination: float = 0.025,
) -> dict:
"""Evaluate Isolation Forest predictions against ground truth."""
threshold = scores.quantile(contamination)
predictions = (scores <= threshold).astype(int)
return {
"precision": precision_score(labels, predictions, zero_division=0),
"recall": recall_score(labels, predictions, zero_division=0),
"f1": f1_score(labels, predictions, zero_division=0),
"roc_auc": roc_auc_score(labels, -scores),
}Your detection engine is ready. Next up, you will build a behavior graph that connects users by shared behavioral patterns and reveals suspicious clusters that neither model can surface alone.
Build the Behavior Graph
Your ML models can flag individual users as anomalous, but they treat each user in isolation. In a real enterprise, insider threats often involve correlated behavior patterns across multiple users, or malicious users whose behavior bridges otherwise distinct groups.
In this step, you will build a NetworkX behavior graph that links users by shared behavioral patterns. You will then combine ML anomaly scores with graph centrality metrics into a single risk score that outperforms either signal alone.
In this step, get ready to:
- Construct a user similarity graph using cosine similarity and NetworkX.
- Compute degree centrality and betweenness centrality to surface suspicious structure.
- Produce a combined risk score and run the full detection pipeline end-to-end.
Construct a user similarity graph with NetworkX
Two users who access sensitive files after hours, send emails externally, and connect USB devices in the same pattern are behaviorally similar. A graph captures these pairwise relationships and reveals cluster structure that per-user ML scores cannot.
- Open behavior_graph.py in your editor and add the module imports and the build_behavior_graph() function by pasting the following code:
"""Behavior graph construction and analysis.
Builds a user similarity graph and computes graph-based risk metrics.
"""
import numpy as np
import pandas as pd
import networkx as nx
from sklearn.metrics.pairwise import cosine_similarity
def build_behavior_graph(
features_df: pd.DataFrame,
similarity_threshold: float = 0.85,
) -> nx.Graph:
"""Build a user similarity graph from feature vectors.
Users are nodes. Edges connect users with cosine similarity
above the threshold.
Returns:
NetworkX Graph with similarity weights on edges.
"""
# Compute pairwise cosine similarity between all user feature vectors
sim_matrix = cosine_similarity(features_df)
users = features_df.index.tolist()
# Create graph with all users as nodes
G = nx.Graph()
G.add_nodes_from(users)
# Add edges only where similarity exceeds threshold
for i in range(len(users)):
for j in range(i + 1, len(users)):
if sim_matrix[i, j] >= similarity_threshold:
G.add_edge(users[i], users[j], weight=sim_matrix[i, j])
return GWhat does this code do?
This function computes cosine similarity between every pair of user feature vectors (a 200x200 matrix). It then creates a graph where each user is a node.
Edges only form between users whose similarity score is 0.85 or higher. This threshold keeps the graph sparse and meaningful. Users connected by an edge share very similar behavioral patterns across all 34 features.
Why graphs instead of just ML scores?
ML models score each user independently. A graph reveals relational patterns: do flagged users cluster together? Does a user bridge two otherwise separate groups (suggesting they are adapting behavior to blend in)?
These structural signals add a dimension that per-user anomaly scores cannot capture on their own.
Compute graph metrics to identify suspicious structure
With the graph constructed, centrality metrics quantify each user's structural position. A user with high betweenness centrality sits between distinct behavioral clusters, which may indicate they are adapting their behavior over time.
- Add the compute_graph_metrics() function below build_behavior_graph() in behavior_graph.py:
def compute_graph_metrics(G: nx.Graph) -> pd.DataFrame:
"""Compute centrality metrics for all nodes.
Returns:
DataFrame with degree_centrality, betweenness_centrality per user.
"""
# Degree centrality: fraction of nodes each user is connected to
degree_cent = nx.degree_centrality(G)
# Betweenness centrality: how often a user lies on shortest paths between others
betweenness_cent = nx.betweenness_centrality(G)
metrics_df = pd.DataFrame({
"degree_centrality": degree_cent,
"betweenness_centrality": betweenness_cent,
})
return metrics_dfWhat do these centrality metrics tell us?
- Degree centrality measures what fraction of all users a given user is connected to. High degree means the user shares common behavioral patterns with many peers.
- Betweenness centrality measures how often a user sits on the shortest path between two other users. A high score means the user bridges otherwise separate behavioral groups.
Produce a combined risk score and run the full pipeline
The final piece fuses ML anomaly scores with graph-derived signals into a single ranked risk score. This combined approach captures both individual deviations (from Isolation Forest) and structural suspicion (from graph centrality).
- Add the compute_risk_scores() function below compute_graph_metrics() in behavior_graph.py:
def compute_risk_scores(
anomaly_scores: pd.Series,
graph_metrics: pd.DataFrame,
xgb_predictions: pd.Series,
G: nx.Graph,
) -> pd.DataFrame:
"""Compute combined risk scores from ML and graph signals.
risk_score = 0.6 * norm_anomaly + 0.25 * norm_betweenness + 0.15 * flagged_neighbor_ratio
Returns:
DataFrame with risk_score and component scores, sorted descending.
"""
# Normalize anomaly scores (flip so higher = more suspicious)
norm_anomaly = 1 - (anomaly_scores - anomaly_scores.min()) / (anomaly_scores.max() - anomaly_scores.min())
# Normalize betweenness
betweenness = graph_metrics["betweenness_centrality"]
norm_betweenness = (betweenness - betweenness.min()) / (betweenness.max() - betweenness.min() + 1e-10)
# Compute flagged neighbor ratio
flagged_neighbor_ratio = pd.Series(0.0, index=anomaly_scores.index)
for user in G.nodes():
neighbors = list(G.neighbors(user))
if len(neighbors) > 0:
flagged_count = sum(1 for n in neighbors if xgb_predictions.get(n, 0) == 1)
flagged_neighbor_ratio[user] = flagged_count / len(neighbors)
# Combined score
risk_score = (
0.6 * norm_anomaly
+ 0.25 * norm_betweenness
+ 0.15 * flagged_neighbor_ratio
)
result = pd.DataFrame({
"risk_score": risk_score,
"anomaly_component": norm_anomaly,
"betweenness_component": norm_betweenness,
"neighbor_component": flagged_neighbor_ratio,
}).sort_values("risk_score", ascending=False)
return resultHow does the risk formula work?
- 60% weight on the normalized anomaly score from Isolation Forest. This is the strongest individual signal.
- 25% weight on normalized betweenness centrality. Users bridging behavioral clusters may be adapting their patterns.
- 15% weight on flagged neighbor ratio. If a user's graph neighbors are also flagged by XGBoost, guilt-by-association provides supporting evidence.
- Save behavior_graph.py.
✔️ Awesome, I've got everything!
Great. Make sure you have saved the file before moving on to the main pipeline.
ⓧ I'd like to double check the full code
"""Behavior graph construction and analysis.
Builds a user similarity graph and computes graph-based risk metrics.
"""
import numpy as np
import pandas as pd
import networkx as nx
from sklearn.metrics.pairwise import cosine_similarity
def build_behavior_graph(
features_df: pd.DataFrame,
similarity_threshold: float = 0.85,
) -> nx.Graph:
"""Build a user similarity graph from feature vectors.
Users are nodes. Edges connect users with cosine similarity
above the threshold.
Returns:
NetworkX Graph with similarity weights on edges.
"""
sim_matrix = cosine_similarity(features_df)
users = features_df.index.tolist()
G = nx.Graph()
G.add_nodes_from(users)
for i in range(len(users)):
for j in range(i + 1, len(users)):
if sim_matrix[i, j] >= similarity_threshold:
G.add_edge(users[i], users[j], weight=sim_matrix[i, j])
return G
def compute_graph_metrics(G: nx.Graph) -> pd.DataFrame:
"""Compute centrality metrics for all nodes.
Returns:
DataFrame with degree_centrality, betweenness_centrality per user.
"""
degree_cent = nx.degree_centrality(G)
betweenness_cent = nx.betweenness_centrality(G)
metrics_df = pd.DataFrame({
"degree_centrality": degree_cent,
"betweenness_centrality": betweenness_cent,
})
return metrics_df
def compute_risk_scores(
anomaly_scores: pd.Series,
graph_metrics: pd.DataFrame,
xgb_predictions: pd.Series,
G: nx.Graph,
) -> pd.DataFrame:
"""Compute combined risk scores from ML and graph signals.
risk_score = 0.6 * norm_anomaly + 0.25 * norm_betweenness + 0.15 * flagged_neighbor_ratio
Returns:
DataFrame with risk_score and component scores, sorted descending.
"""
norm_anomaly = 1 - (anomaly_scores - anomaly_scores.min()) / (anomaly_scores.max() - anomaly_scores.min())
betweenness = graph_metrics["betweenness_centrality"]
norm_betweenness = (betweenness - betweenness.min()) / (betweenness.max() - betweenness.min() + 1e-10)
flagged_neighbor_ratio = pd.Series(0.0, index=anomaly_scores.index)
for user in G.nodes():
neighbors = list(G.neighbors(user))
if len(neighbors) > 0:
flagged_count = sum(1 for n in neighbors if xgb_predictions.get(n, 0) == 1)
flagged_neighbor_ratio[user] = flagged_count / len(neighbors)
risk_score = (
0.6 * norm_anomaly
+ 0.25 * norm_betweenness
+ 0.15 * flagged_neighbor_ratio
)
result = pd.DataFrame({
"risk_score": risk_score,
"anomaly_component": norm_anomaly,
"betweenness_component": norm_betweenness,
"neighbor_component": flagged_neighbor_ratio,
}).sort_values("risk_score", ascending=False)
return resultNow you need an orchestrator that ties together every module you have built: data generation, feature engineering, anomaly detection, and the behavior graph.
- Open main.py and paste the full pipeline code:
"""Main pipeline for insider threat detection system."""
import pandas as pd
from data_generator import generate_insider_threat_data
from feature_engineering import engineer_features, normalize_features
from detector import (
train_isolation_forest,
train_xgboost_cv,
evaluate_isolation_forest,
)
from behavior_graph import (
build_behavior_graph,
compute_graph_metrics,
compute_risk_scores,
)
def main():
print("=" * 60)
print("INSIDER THREAT DETECTION SYSTEM")
print("=" * 60)
# Step 1: Generate synthetic data
print("\n[1/5] Generating synthetic insider threat data...")
events_df, malicious_users = generate_insider_threat_data(
n_users=200, n_days=30, n_malicious=5, seed=42
)
print(f" Total events: {len(events_df):,}")
print(f" Users: {events_df['user_id'].nunique()}")
print(f" Malicious users: {malicious_users}")
print(f" Event types: {events_df['event_type'].value_counts().to_dict()}")
# Step 2: Feature engineering
print("\n[2/5] Engineering behavioral features...")
user_features = engineer_features(events_df, n_days=30)
normalized_features, scaler = normalize_features(user_features)
print(f" Feature matrix shape: {normalized_features.shape}")
print(f" Features per user: {normalized_features.shape[1]}")
# Create labels
labels = pd.Series(
[1 if uid in malicious_users else 0 for uid in normalized_features.index],
index=normalized_features.index,
name="is_malicious",
)
print(f" Class distribution: {labels.value_counts().to_dict()}")
# Step 3: Anomaly detection
print("\n[3/5] Training anomaly detection models...")
# Isolation Forest
iso_model, anomaly_scores = train_isolation_forest(normalized_features)
iso_metrics = evaluate_isolation_forest(anomaly_scores, labels)
print(f"\n Isolation Forest Results:")
print(f" Precision: {iso_metrics['precision']:.3f}")
print(f" Recall: {iso_metrics['recall']:.3f}")
print(f" F1: {iso_metrics['f1']:.3f}")
print(f" ROC-AUC: {iso_metrics['roc_auc']:.3f}")
# XGBoost
xgb_model, xgb_metrics = train_xgboost_cv(normalized_features, labels)
print(f"\n XGBoost CV Results (mean):")
print(f" Precision: {xgb_metrics['precision']:.3f}")
print(f" Recall: {xgb_metrics['recall']:.3f}")
print(f" F1: {xgb_metrics['f1']:.3f}")
print(f" ROC-AUC: {xgb_metrics['roc_auc']:.3f}")
# XGBoost predictions on full data
xgb_predictions = pd.Series(
xgb_model.predict(normalized_features),
index=normalized_features.index,
)
# Feature importances
importances = pd.Series(
xgb_model.feature_importances_,
index=normalized_features.columns,
).sort_values(ascending=False)
print(f"\n Top 5 important features:")
for feat, imp in importances.head(5).items():
print(f" {feat}: {imp:.4f}")
# Step 4: Behavior graph
print("\n[4/5] Building behavior graph...")
G = build_behavior_graph(normalized_features, similarity_threshold=0.85)
graph_metrics = compute_graph_metrics(G)
print(f" Nodes: {G.number_of_nodes()}")
print(f" Edges: {G.number_of_edges()}")
print(f" Connected components: {nx.number_connected_components(G)}")
# Step 5: Combined risk scoring
print("\n[5/5] Computing combined risk scores...")
risk_df = compute_risk_scores(anomaly_scores, graph_metrics, xgb_predictions, G)
print(f"\n Top 10 Riskiest Users:")
print(f" {'User':<12} {'Risk Score':<12} {'Actual':<10}")
print(f" {'-'*34}")
for user_id, row in risk_df.head(10).iterrows():
actual = "MALICIOUS" if user_id in malicious_users else "benign"
print(f" {user_id:<12} {row['risk_score']:.4f} {actual}")
# Precision@K
top5_users = risk_df.head(5).index.tolist()
top10_users = risk_df.head(10).index.tolist()
p_at_5 = sum(1 for u in top5_users if u in malicious_users) / 5
p_at_10 = sum(1 for u in top10_users if u in malicious_users) / 10
print(f"\n Precision@5: {p_at_5:.2f}")
print(f" Precision@10: {p_at_10:.2f}")
print("\n" + "=" * 60)
print("DETECTION COMPLETE")
print("=" * 60)
if __name__ == "__main__":
main()What does the main pipeline do?
This orchestrator runs all five stages sequentially: data generation, feature engineering, both ML models, graph construction, and combined risk scoring.
At the end, it prints the top 10 riskiest users alongside their ground truth labels, plus Precision@5 and Precision@10 metrics. These tell you what fraction of the top-ranked users are actually malicious.
- Save main.py.
- Run the full pipeline by executing this command in your terminal:
python main.py- You should see output from all five stages, ending with a ranked table of the top 10 riskiest users and Precision@5 and Precision@10 scores.
What should I see?
The pipeline prints progress for each stage. At the end, look for the Top 10 Riskiest Users table. Malicious users should appear near the top of the ranking. Precision@5 tells you how many of the top 5 ranked users are actually malicious (out of the 5 total malicious users in the dataset).
Seeing an import error or NameError?
- Make sure you are running the command from the project directory that contains all five .py files.
- Confirm your virtual environment is activated. You should see the environment name in your terminal prompt.
- If you see ModuleNotFoundError, double-check that all packages were installed in the active virtual environment.
Help me debug an import error when running main.py
✔️ Awesome, I've got everything!
Great. Make sure you have saved both files before continuing.
ⓧ I'd like to double check the full code
"""Main pipeline for insider threat detection system."""
import pandas as pd
from data_generator import generate_insider_threat_data
from feature_engineering import engineer_features, normalize_features
from detector import (
train_isolation_forest,
train_xgboost_cv,
evaluate_isolation_forest,
)
from behavior_graph import (
build_behavior_graph,
compute_graph_metrics,
compute_risk_scores,
)
def main():
print("=" * 60)
print("INSIDER THREAT DETECTION SYSTEM")
print("=" * 60)
print("\n[1/5] Generating synthetic insider threat data...")
events_df, malicious_users = generate_insider_threat_data(
n_users=200, n_days=30, n_malicious=5, seed=42
)
print(f" Total events: {len(events_df):,}")
print(f" Users: {events_df['user_id'].nunique()}")
print(f" Malicious users: {malicious_users}")
print(f" Event types: {events_df['event_type'].value_counts().to_dict()}")
print("\n[2/5] Engineering behavioral features...")
user_features = engineer_features(events_df, n_days=30)
normalized_features, scaler = normalize_features(user_features)
print(f" Feature matrix shape: {normalized_features.shape}")
print(f" Features per user: {normalized_features.shape[1]}")
labels = pd.Series(
[1 if uid in malicious_users else 0 for uid in normalized_features.index],
index=normalized_features.index,
name="is_malicious",
)
print(f" Class distribution: {labels.value_counts().to_dict()}")
print("\n[3/5] Training anomaly detection models...")
iso_model, anomaly_scores = train_isolation_forest(normalized_features)
iso_metrics = evaluate_isolation_forest(anomaly_scores, labels)
print(f"\n Isolation Forest Results:")
print(f" Precision: {iso_metrics['precision']:.3f}")
print(f" Recall: {iso_metrics['recall']:.3f}")
print(f" F1: {iso_metrics['f1']:.3f}")
print(f" ROC-AUC: {iso_metrics['roc_auc']:.3f}")
xgb_model, xgb_metrics = train_xgboost_cv(normalized_features, labels)
print(f"\n XGBoost CV Results (mean):")
print(f" Precision: {xgb_metrics['precision']:.3f}")
print(f" Recall: {xgb_metrics['recall']:.3f}")
print(f" F1: {xgb_metrics['f1']:.3f}")
print(f" ROC-AUC: {xgb_metrics['roc_auc']:.3f}")
xgb_predictions = pd.Series(
xgb_model.predict(normalized_features),
index=normalized_features.index,
)
importances = pd.Series(
xgb_model.feature_importances_,
index=normalized_features.columns,
).sort_values(ascending=False)
print(f"\n Top 5 important features:")
for feat, imp in importances.head(5).items():
print(f" {feat}: {imp:.4f}")
print("\n[4/5] Building behavior graph...")
G = build_behavior_graph(normalized_features, similarity_threshold=0.85)
graph_metrics = compute_graph_metrics(G)
print(f" Nodes: {G.number_of_nodes()}")
print(f" Edges: {G.number_of_edges()}")
print(f" Connected components: {nx.number_connected_components(G)}")
print("\n[5/5] Computing combined risk scores...")
risk_df = compute_risk_scores(anomaly_scores, graph_metrics, xgb_predictions, G)
print(f"\n Top 10 Riskiest Users:")
print(f" {'User':<12} {'Risk Score':<12} {'Actual':<10}")
print(f" {'-'*34}")
for user_id, row in risk_df.head(10).iterrows():
actual = "MALICIOUS" if user_id in malicious_users else "benign"
print(f" {user_id:<12} {row['risk_score']:.4f} {actual}")
top5_users = risk_df.head(5).index.tolist()
top10_users = risk_df.head(10).index.tolist()
p_at_5 = sum(1 for u in top5_users if u in malicious_users) / 5
p_at_10 = sum(1 for u in top10_users if u in malicious_users) / 10
print(f"\n Precision@5: {p_at_5:.2f}")
print(f" Precision@10: {p_at_10:.2f}")
print("\n" + "=" * 60)
print("DETECTION COMPLETE")
print("=" * 60)
if __name__ == "__main__":
main()Secret mission
Build the Investigation Agent
Your detection system flags suspicious users, but a SOC analyst still needs to manually investigate each one. Can you build an automated investigation agent that traverses the behavior graph, correlates temporal patterns, and generates structured threat narratives for flagged users?
Clean Up Your Resources
Clean Up Your Resources
This project runs entirely locally, so there are no ongoing costs or cloud resources to worry about.
Decide whether to keep your project files for future reference, or delete them to free up disk space.
Resources you used:
- Project folder on your Desktop containing data_generator.py, feature_engineering.py, detector.py, behavior_graph.py, main.py, and investigator.py.
- Python virtual environment inside the project folder.
- Installed pip packages: scikit-learn, xgboost-cpu, networkx, pandas, numpy, matplotlib.
Keep everything running
No action needed. Choose this if you want to keep experimenting with the detection pipeline, tweaking thresholds, or extending the investigation agent.
- Your project files and virtual environment stay on your Desktop, ready to use whenever you come back.
- The installed packages only exist inside the virtual environment, so they do not affect your system Python.
Pause - I'll come back to this later
Since this project has no running processes or servers, there is nothing to pause. Your files sit idle on disk until you return.
- If your virtual environment is still activated, deactivate it by running:
deactivate- When you come back, navigate to your project folder and reactivate the virtual environment to pick up where you left off.
Delete - I don't want to use this again
Remove all project files and the virtual environment to start fresh. This frees up the disk space used by the installed packages.
- If your virtual environment is still activated, deactivate it first by running:
deactivate- Navigate to your Desktop by running:
cd ~/Desktop- Delete the entire project folder. On macOS or Linux, run:
rm -rf insider-threat-detection- On Windows, run:
rmdir /s /q insider-threat-detection- This removes all Python files, the virtual environment, and all installed packages in one step.
Nice Work!
Nice Work!
You just built a full insider threat detection system that combines machine learning anomaly detection with graph-based behavioral correlation.
You've learned how to:
- Engineer 34 behavioral features from raw enterprise activity logs across 5 event categories, applying temporal aggregation and domain knowledge to transform noisy data into structured user-level vectors.
- Train and compare Isolation Forest (unsupervised) and XGBoost (supervised) anomaly detectors, understanding the tradeoffs between labeled and unlabeled detection approaches for rare-event problems.
- Build a NetworkX behavior graph with centrality metrics and a combined risk scoring formula that outperforms either ML or graph signals in isolation.
- Secret Mission: Built an automated investigation agent that traverses the behavior graph and generates structured threat narrative reports for flagged users.
Ready to quiz yourself?