Part 4 — Per-Entity Baselines + Model Persistence (So Your Mini IDS Actually Learns)

Part 3 got you to a deployable pipeline: flow logs in, anomalies out, JSON events shipped.

image from miro

But there’s still a problem that shows up immediately in real networks:

A single global baseline is unfair.

A proxy/NAT gateway is “noisy normal.” A database server is “quiet normal.” A vulnerability scanner is “scan-shaped normal.”

If you baseline them together, you’ll either:

• drown in false positives, or
• raise thresholds so high you miss real issues.

Part 4 fixes that by adding:

• per-entity baselines (per host / per subnet / per role)
• model persistence (baseline survives restarts)
• stable anomaly scores across days/weeks
• lightweight drift handling (traffic changes over time)

This is the difference between a cool script and something operators can rely on.

1) Choose your “entity”: host, subnet, role, or service

Per-entity baselining can mean different things. Pick what matches your visibility and your ops workflow.

Common choices:

A) Per source IP (src_ip)

Great for catching:

• compromised endpoints
• unusual egress from a workstation
• one host suddenly scanning

B) Per destination IP (dst_ip)

Great for catching:

• a server suddenly receiving new ports
• inbound spikes to a specific service

C) Per subnet / segment

Great when hosts churn (DHCP, ephemeral nodes).

D) Per “role tag”

Best in real ops. If you can tag:

• proxy, db, app, scanner, vpn, nat …then baselines become way more stable.

In this Part 4 build, we’ll implement per-entity baselines where entity_id is computed from the flow record:

• default: src_ip
• optional: role tag if you have tags

2) The key design: aggregate “per entity per window”

In Parts 1–3, you aggregated global metrics per time window.

Now you do:

For each window, compute metrics per entity.

So a single window produces multiple rows:

• window_start=12:00:00, entity=10.10.1.10, bytes=…, uniq_dst_ips=…, …
• window_start=12:00:00, entity=10.10.2.15, bytes=…, uniq_dst_ips=…, …
• window_start=12:00:00, entity=proxy-1, bytes=…, uniq_dst_ips=…, …

That lets you compare each entity against its own historical normal.

3) Per-entity window aggregator (flows)

We’ll build an aggregator that produces features for each entity in each window.

Features per entity:

• total bytes, packets, flows
• unique dst IPs
• unique dst ports
• max dst ports in window (fan-out)
• entropy of dst ports (diversity change)
• external dst ratio (if you want egress monitoring)

Implementation

from collections import Counter, defaultdict
import ipaddress
import math
WINDOW_SECONDS = 60
PRIVATE_NETS = [
    ipaddress.ip_network("10.0.0.0/8"),
    ipaddress.ip_network("172.16.0.0/12"),
    ipaddress.ip_network("192.168.0.0/16"),
]
def is_private_ip(ip: str) -> bool:
    try:
        addr = ipaddress.ip_address(ip)
        return any(addr in n for n in PRIVATE_NETS)
    except Exception:
        return False
def window_key(ts: int) -> int:
    return (ts // WINDOW_SECONDS) * WINDOW_SECONDS
def entropy(counter: Counter) -> float:
    total = sum(counter.values())
    if total == 0:
        return 0.0
    ent = 0.0
    for c in counter.values():
        p = c / total
        ent -= p * math.log2(p)
    return ent
def get_entity_id(flow: dict, mode: str = "src_ip"):
    # mode can be: src_ip, dst_ip, role
    if mode == "dst_ip":
        return flow.get("dst_ip") or "unknown"
    if mode == "role":
        # if you have a tag map, resolve here; else fallback to src_ip
        src = flow.get("src_ip")
        role = (TAGS_BY_IP.get(src) or [""])[0] if src else ""
        return role if role else (src or "unknown")
    return flow.get("src_ip") or "unknown"

class PerEntityWindowAgg:
    def __init__(self, entity_mode="src_ip"):
        self.entity_mode = entity_mode
        self.reset()
    def reset(self):
        # window -> entity -> counters
        self.entities = defaultdict(lambda: defaultdict(dict))
        self.data = defaultdict(lambda: defaultdict(self._new_entity_bucket))
    def _new_entity_bucket(self):
        return {
            "flows": 0,
            "bytes": 0,
            "packets": 0,
            "dst_ips": set(),
            "dst_ports": Counter(),
            "external_flows": 0,
            "external_bytes": 0,
        }
    def add_flow(self, f: dict):
        ts = f.get("ts")
        if not ts:
            return
        wk = window_key(int(ts))
        ent_id = get_entity_id(f, self.entity_mode)
        b = self.data[wk][ent_id]
        b["flows"] += 1
        b["bytes"] += int(f.get("bytes", 0))
        b["packets"] += int(f.get("packets", 0))
        dst_ip = f.get("dst_ip")
        if dst_ip:
            b["dst_ips"].add(dst_ip)
        dport = int(f.get("dst_port", 0) or 0)
        if dport:
            b["dst_ports"][dport] += 1
        # egress hint
        if dst_ip and not is_private_ip(dst_ip):
            b["external_flows"] += 1
            b["external_bytes"] += int(f.get("bytes", 0))
    def flush_window(self, wk: int):
        # Return list of (features, context) for each entity in this window
        out = []
        for ent_id, b in self.data[wk].items():
            uniq_dst_ips = len(b["dst_ips"])
            uniq_dst_ports = len(b["dst_ports"])
            max_port_count = max(b["dst_ports"].values(), default=0)
            features = {
                "window_start": wk,
                "entity_id": ent_id,
                "flows": b["flows"],
                "bytes": b["bytes"],
                "packets": b["packets"],
                "unique_dst_ips": uniq_dst_ips,
                "unique_dst_ports": uniq_dst_ports,
                "dst_port_entropy": entropy(b["dst_ports"]),
                "max_dst_port_hits": max_port_count,
                "external_flow_ratio": (b["external_flows"] / b["flows"]) if b["flows"] else 0.0,
                "external_byte_ratio": (b["external_bytes"] / b["bytes"]) if b["bytes"] else 0.0,
            }
            context = {
                "top_dst_ports": b["dst_ports"].most_common(8),
                "external_flows": b["external_flows"],
                "external_bytes": b["external_bytes"],
                "unique_dst_ips": uniq_dst_ips,
            }
            out.append((features, context))
        # cleanup this window
        del self.data[wk]
        return out

This gives you “entity rows” per window.

4) Baselines that persist: store rolling stats per entity

You need two things:

1. history buffer per entity (last N windows)
2. persistence to disk so it survives restarts

We’ll implement:

• a small JSON file per entity, or
• a single SQLite DB (cleaner, scalable)

For Medium + simplicity, SQLite is a great “production-ish” choice:

• built-in, no server
• structured
• safe enough for small tools

We’ll store:

• per-entity recent history (compressed)
• a few computed baseline stats (median, MAD per feature)
• last_updated timestamp

Option A (recommended): SQLite baseline store

import sqlite3
import json
from datetime import datetime, timezone
class BaselineStoreSQLite:
    def __init__(self, path="baseline_store.sqlite"):
        self.conn = sqlite3.connect(path)
        self._init_db()
    def _init_db(self):
        cur = self.conn.cursor()
        cur.execute("""
        CREATE TABLE IF NOT EXISTS entity_baseline (
            entity_id TEXT PRIMARY KEY,
            history_json TEXT NOT NULL,
            updated_at TEXT NOT NULL
        )
        """)
        self.conn.commit()
    def load_history(self, entity_id: str):
        cur = self.conn.cursor()
        cur.execute("SELECT history_json FROM entity_baseline WHERE entity_id=?", (entity_id,))
        row = cur.fetchone()
        if not row:
            return []
        return json.loads(row[0])
    def save_history(self, entity_id: str, history: list):
        cur = self.conn.cursor()
        cur.execute("""
        INSERT INTO entity_baseline(entity_id, history_json, updated_at)
        VALUES(?, ?, ?)
        ON CONFLICT(entity_id) DO UPDATE SET
            history_json=excluded.history_json,
            updated_at=excluded.updated_at
        """, (entity_id, json.dumps(history), datetime.now(timezone.utc).isoformat()))
        self.conn.commit()

History is stored as a JSON list of dicts like: [{bytes:..., flows:..., unique_dst_ips:...}, ...]

Yes, it’s not the most compact. But it’s simple and works well for moderate entity counts.

5) Per-entity rolling robust z-score (with warm-up)

We’ll compute robust z-scores per entity using that entity’s stored history.

Key points:

• You need a warm-up period (e.g., 30 windows) before alerting.
• Your baseline window size depends on environment (often 1–24 hours of history).
• You should protect against MAD=0.

import pandas as pd
def robust_abs_z(val: float, series: pd.Series) -> float:
    median = series.median()
    mad = (series - median).abs().median()
    if mad == 0:
        return 0.0
    rz = 0.6745 * (val - median) / mad
    return abs(float(rz))

class PerEntityBaseline:
    def __init__(self, store: BaselineStoreSQLite, feature_cols, history_size=240, warmup=30):
        self.store = store
        self.feature_cols = feature_cols
        self.history_size = history_size
        self.warmup = warmup
    def score_and_update(self, entity_id: str, current: dict):
        history = self.store.load_history(entity_id)
        # history is a list of dicts
        hist_df = pd.DataFrame(history) if history else pd.DataFrame()
        scores = {}
        ready = len(history) >= self.warmup
        for col in self.feature_cols:
            val = float(current.get(col, 0))
            if not ready or hist_df.empty or col not in hist_df:
                scores[f"abs_rz_{col}"] = 0.0
            else:
                scores[f"abs_rz_{col}"] = robust_abs_z(val, hist_df[col].astype(float))
        # update history after scoring
        row = {col: float(current.get(col, 0)) for col in self.feature_cols}
        history.append(row)
        if len(history) > self.history_size:
            history = history[-self.history_size:]
        self.store.save_history(entity_id, history)
        return scores, ready

6) Per-entity alert logic: less noisy, more relevant

Per-entity alerting allows lower thresholds (more sensitivity) without global noise.

Example reasons:

• bytes spike for entity
• flow spike for entity
• unique dst IPs spike (host sweep)
• unique dst ports spike (port scan)
• external ratio jump (egress anomaly)

def entity_reasons(features: dict, rz: dict):
    reasons = []
    if rz["abs_rz_bytes"] > 6:
        reasons.append("bytes spike for entity")
    if rz["abs_rz_flows"] > 6:
        reasons.append("flow spike for entity")
    if rz["abs_rz_unique_dst_ips"] > 6:
        reasons.append("dst IP fan-out spike for entity")
    if rz["abs_rz_unique_dst_ports"] > 6:
        reasons.append("dst port diversity spike for entity")
    if rz["abs_rz_dst_port_entropy"] > 6:
        reasons.append("dst port entropy jump")
    # direct scan-ish thresholds (entity-specific)
    if features["unique_dst_ports"] >= 120:
        reasons.append("possible port scan (entity hit many dst ports)")
    if features["unique_dst_ips"] >= 300:
        reasons.append("possible host sweep (entity hit many dst IPs)")
    # egress anomaly
    if features["external_flow_ratio"] > 0.8 and features["flows"] > 100:
        reasons.append("high external egress ratio")
    return reasons

Tune those numbers to your environment; per-entity baselines let you tune separately later by role.

7) Putting it together: flow file → per-entity windows → persistent scoring → JSON alerts

This runner:

• reads flows
• groups into windows
• for each entity in window, scores it against its baseline
• writes JSONL alerts

def write_jsonl(path: str, obj: dict):
    import json
    with open(path, "a", encoding="utf-8") as f:
        f.write(json.dumps(obj) + "\n")
def run_flow_file_per_entity(flow_path: str, fmt="jsonl", out_alerts="entity_alerts.jsonl"):
    reader = read_flows_jsonl if fmt == "jsonl" else read_flows_csv
    store = BaselineStoreSQLite("baseline_store.sqlite")
    feature_cols = ["flows", "bytes", "packets", "unique_dst_ips", "unique_dst_ports", "dst_port_entropy", "external_flow_ratio"]
    baseline = PerEntityBaseline(store, feature_cols=feature_cols, history_size=720, warmup=60)  # 12h history if 60s windows
    agg = PerEntityWindowAgg(entity_mode="src_ip")
    current_wk = None
    def flush(wk: int):
        entity_rows = agg.flush_window(wk)
        for features, context in entity_rows:
            entity_id = features["entity_id"]
            rz, ready = baseline.score_and_update(entity_id, features)
            if not ready:
                continue
            reasons = entity_reasons(features, rz)
            if reasons:
                event = {
                    "event_type": "mini_ids_entity_anomaly",
                    "window_start": wk,
                    "window_seconds": WINDOW_SECONDS,
                    "entity_id": entity_id,
                    "reasons": reasons,
                    "features": features,
                    "scores": rz,
                    "context": context,
                }
                write_jsonl(out_alerts, event)
                print(f"[ENTITY ALERT] wk={wk} entity={entity_id} reasons={','.join(reasons)} bytes={features['bytes']} flows={features['flows']}")
    for f in reader(flow_path):
        ts = f.get("ts")
        if not ts:
            continue
        wk = window_key(int(ts))
        if current_wk is None:
            current_wk = wk
        if wk != current_wk:
            flush(current_wk)
            current_wk = wk
        agg.add_flow(f)
    if current_wk is not None:
        flush(current_wk)

This is the core of a “learning” mini IDS.

8) Handling drift: traffic changes over weeks (don’t fight it)

Two drift problems happen in production:

A) “Monday looks different than Saturday”

Fix: separate baselines by time bucket:

• weekday/weekend
• hour-of-day bands

A lightweight approach:

• baseline key = entity_id + day_type + hour_band

Example:

• 10.10.1.15|weekday|09-17
• 10.10.1.15|offhours

You can implement this by changing entity_id in the baseline store to include the band.

B) “Normal slowly shifts”

Fix: use rolling history (already) and keep history size meaningful:

• too small → baseline jitter
• too big → baseline lags behind change

A common sweet spot:

• 6–24 hours for short-term anomalies
• optionally, add a long-term baseline (7–14 days) for “trend anomalies”

Keep it simple unless you really need long-term.

9) Prevent alert storms: cooldown + dedup

Real ops improvement: avoid spamming when one host is noisy for 30 minutes.

Implement:

• cooldown per entity (e.g., don’t alert more than once every 10 minutes unless severity increases)
• dedup by (entity_id, reason set)

Sketch:

import time
class Cooldown:
    def __init__(self, seconds=600):
        self.seconds = seconds
        self.last = {}
    def allow(self, key: str) -> bool:
        now = time.time()
        t = self.last.get(key, 0)
        if now - t >= self.seconds:
            self.last[key] = now
            return True
        return False

Use key like: f"{entity_id}|{','.join(sorted(reasons))}".

10) What Part 4 buys you operationally

After Part 4, your system:

• learns normal per entity
• survives restart
• produces stable anomaly scoring
• reduces global noise dramatically
• supports realistic triage (entity context + scoring)

This is where teams start trusting it.

Part 4 Wrap-up

Per-entity baselines are the real unlock:

• a proxy can be “loud normal” without breaking everything
• a database can be “quiet normal” and still trigger on subtle changes
• a scanner can be “scan-shaped normal” without constant noise

Persistence is the second unlock:

• if the tool forgets its baseline every restart, it never becomes reliable

With Part 4, your mini IDS becomes a continuously learning monitoring component — still lightweight, but finally aligned with how real networks behave.