How to Build a Real-Time Ad Fraud Dashboard with Python and WebSocket

Monitor your ad traffic quality in real-time. Here's a complete implementation using Python, WebSocket, and a simple frontend.

Architecture

Ad Traffic → Collector → Analysis Engine → WebSocket Server → Dashboard
                                ↓
                         Alert System

Backend (Python + FastAPI)

from fastapi import FastAPI, WebSocket
import asyncio
import json

app = FastAPI()
connected_clients = set()

class TrafficAnalyzer:
    def __init__(self):
        self.stats = {
            'total_visits': 0,
            'bot_detected': 0,
            'human_verified': 0,
            'suspicious': 0
        }

    def analyze(self, visit):
        self.stats['total_visits'] += 1

        # Three-layer check
        ip_score = self.check_ip(visit['ip'])
        fp_score = self.check_fingerprint(visit['fingerprint'])
        behavior_score = self.check_behavior(visit['mouse_data'])

        combined = (ip_score + fp_score + behavior_score) / 3

        if combined > 70:
            self.stats['human_verified'] += 1
            verdict = 'human'
        elif combined > 40:
            self.stats['suspicious'] += 1
            verdict = 'suspicious'
        else:
            self.stats['bot_detected'] += 1
            verdict = 'bot'

        return {
            'verdict': verdict,
            'score': combined,
            'stats': self.stats.copy()
        }

analyzer = TrafficAnalyzer()

@app.websocket('/ws')
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    connected_clients.add(websocket)
    try:
        while True:
            data = await websocket.receive_json()
            result = analyzer.analyze(data)
            # Broadcast to all dashboards
            for client in connected_clients:
                await client.send_json(result)
    finally:
        connected_clients.discard(websocket)

Frontend Dashboard

Use Chart.js for real-time graphs showing bot/human ratio, geographic distribution, and threat level over time.

Alerts

async def check_alerts(stats):
    bot_ratio = stats['bot_detected'] / max(stats['total_visits'], 1)
    if bot_ratio > 0.3:
        await send_alert(f'WARNING: Bot ratio at {bot_ratio:.1%}')