Day 96 — Traffic Anomaly Detection (Z-Score + EMA)

Key Concepts

Exponential Moving Average (EMA):

• EMA_new = alpha * value + (1 - alpha) * EMA_old
• alpha = 0.1 means recent values matter but history is preserved
• Tracks mean and variance without storing all historical data

Z-score: (current_rate - mean) / stddev — IPs > 3 sigma above mean are anomalous.

Exercise — Statistical Traffic Analyzer

package ddos

import (
    "math"
    "net/http"
    "sync"
    "time"
)

type ipStats struct {
    mu       sync.Mutex
    requests []time.Time // sliding window entries (last 60s)
    rate     float64     // current request rate (req/s)
}

func (s *ipStats) record(now time.Time, window time.Duration) float64 {
    s.mu.Lock()
    defer s.mu.Unlock()
    cutoff := now.Add(-window)
    valid := s.requests[:0]
    for _, t := range s.requests {
        if t.After(cutoff) {
            valid = append(valid, t)
        }
    }
    s.requests = append(valid, now)
    s.rate = float64(len(s.requests)) / window.Seconds()
    return s.rate
}

type TrafficAnalyzer struct {
    mu      sync.RWMutex
    ipStats map[string]*ipStats
    window  time.Duration

    // EMA-based global baseline
    emaMean float64
    emaVar  float64
    alpha   float64

    Threshold float64 // z-score threshold (default 3.0)
    Alerts    chan Alert
    done      chan struct{}
}

type Alert struct {
    IP        string
    Rate      float64
    ZScore    float64
    Timestamp time.Time
}

func NewTrafficAnalyzer() *TrafficAnalyzer {
    ta := &TrafficAnalyzer{
        ipStats:   make(map[string]*ipStats),
        window:    60 * time.Second,
        alpha:     0.1,
        Threshold: 3.0,
        Alerts:    make(chan Alert, 100),
        done:      make(chan struct{}),
    }
    go ta.cleanup()
    return ta
}

func (ta *TrafficAnalyzer) Record(ip string) {
    ta.mu.Lock()
    s, ok := ta.ipStats[ip]
    if !ok {
        s = &ipStats{}
        ta.ipStats[ip] = s
    }
    ta.mu.Unlock()

    now := time.Now()
    rate := s.record(now, ta.window)

    // Update global EMA
    ta.mu.Lock()
    diff := rate - ta.emaMean
    ta.emaMean += ta.alpha * diff
    ta.emaVar = (1-ta.alpha) * (ta.emaVar + ta.alpha*diff*diff)
    mean := ta.emaMean
    variance := ta.emaVar
    ta.mu.Unlock()

    // Compute z-score
    stddev := math.Sqrt(variance)
    if stddev < 0.01 {
        return // not enough data
    }
    z := (rate - mean) / stddev
    if z > ta.Threshold {
        select {
        case ta.Alerts <- Alert{IP: ip, Rate: rate, ZScore: z, Timestamp: now}:
        default:
        }
    }
}

func (ta *TrafficAnalyzer) Middleware(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        ip, _, _ := net.SplitHostPort(r.RemoteAddr)
        ta.Record(ip)
        next.ServeHTTP(w, r)
    })
}

func (ta *TrafficAnalyzer) cleanup() {
    ticker := time.NewTicker(5 * time.Minute)
    defer ticker.Stop()
    for {
        select {
        case <-ta.done:
            return
        case <-ticker.C:
            ta.mu.Lock()
            // Remove IPs with no recent activity (empty window)
            for ip, s := range ta.ipStats {
                s.mu.Lock()
                cutoff := time.Now().Add(-ta.window)
                fresh := false
                for _, t := range s.requests {
                    if t.After(cutoff) {
                        fresh = true
                        break
                    }
                }
                s.mu.Unlock()
                if !fresh {
                    delete(ta.ipStats, ip)
                }
            }
            ta.mu.Unlock()
        }
    }
}