How to Build a Real-Time Threat Intelligence Feed Aggregator in Python

Security teams drown in alerts. One underappreciated root cause: nobody is systematically collecting and correlating threat intelligence before incidents happen. A threat intelligence feed aggregator pulls IOCs (indicators of compromise) from multiple sources — OSINT feeds, commercial APIs, government advisories — into a single normalized store you can query in real time.

Here's how to build a production-ready one in Python.

What We're Building

The goal is a daemon that:

• Polls multiple threat intel sources on a schedule (AbuseIPDB, AlienVault OTX, CISA KEV)
• Normalizes IOCs into a common schema
• Stores them in SQLite (easy to swap for Postgres later)
• Exposes a lookup function to check if a given IP, domain, or hash is known-bad

We're skipping STIX/TAXII for simplicity, but the data model is compatible if you need to go that route later.

Defining the Data Model

Every source has different field names, different formats, different confidence scores. The first thing you need is a normalized IOC schema:

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum

class IOCType(str, Enum):
    IP = "ip"
    DOMAIN = "domain"
    URL = "url"
    HASH_MD5 = "md5"
    HASH_SHA256 = "sha256"
    CVE = "cve"

@dataclass
class IOC:
    value: str
    ioc_type: IOCType
    source: str
    confidence: int  # 0-100
    tags: list[str] = field(default_factory=list)
    first_seen: datetime = field(default_factory=datetime.utcnow)
    last_seen: datetime = field(default_factory=datetime.utcnow)
    raw: dict = field(default_factory=dict)  # preserve original payload

Keep the raw payload. You'll want it when a field matters later that you didn't normalize today.

Writing the Feed Adapters

Each source gets its own adapter class that returns a list of IOC objects. Here's a minimal one for AbuseIPDB:

import httpx
from datetime import datetime

class AbuseIPDBAdapter:
    BASE_URL = "https://api.abuseipdb.com/api/v2"

    def __init__(self, api_key: str):
        self.api_key = api_key

    def fetch(self, limit: int = 1000) -> list[IOC]:
        headers = {"Key": self.api_key, "Accept": "application/json"}
        params = {"confidenceMinimum": 75, "limit": limit}
        resp = httpx.get(
            f"{self.BASE_URL}/blacklist",
            headers=headers,
            params=params,
            timeout=30,
        )
        resp.raise_for_status()
        data = resp.json()

        iocs = []
        for entry in data.get("data", []):
            iocs.append(IOC(
                value=entry["ipAddress"],
                ioc_type=IOCType.IP,
                source="abuseipdb",
                confidence=entry.get("abuseConfidenceScore", 75),
                tags=entry.get("usageType", "").split(","),
                last_seen=datetime.fromisoformat(
                    entry["lastReportedAt"].replace("Z", "+00:00")
                ),
                raw=entry,
            ))
        return iocs

The pattern is always the same: fetch() returns list[IOC]. This makes the aggregation loop trivial to extend with new sources.

The Aggregation Loop

We want this running continuously, each adapter on its own schedule (CVE feeds change daily, IP reputation can change hourly):

import sqlite3
import schedule
import time
import json
import logging
from pathlib import Path

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger(__name__)

DB_PATH = Path("threat_intel.db")

def init_db(conn: sqlite3.Connection):
    conn.execute("""
        CREATE TABLE IF NOT EXISTS iocs (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            value TEXT NOT NULL,
            ioc_type TEXT NOT NULL,
            source TEXT NOT NULL,
            confidence INTEGER DEFAULT 0,
            tags TEXT DEFAULT '[]',
            first_seen TEXT,
            last_seen TEXT,
            raw TEXT DEFAULT '{}',
            UNIQUE(value, source)
        )
    """)
    conn.execute("CREATE INDEX IF NOT EXISTS idx_iocs_value ON iocs(value)")
    conn.commit()

def upsert_iocs(conn: sqlite3.Connection, iocs: list[IOC]):
    for ioc in iocs:
        conn.execute("""
            INSERT INTO iocs
              (value, ioc_type, source, confidence, tags, first_seen, last_seen, raw)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
            ON CONFLICT(value, source) DO UPDATE SET
                confidence = excluded.confidence,
                last_seen  = excluded.last_seen,
                tags       = excluded.tags,
                raw        = excluded.raw
        """, (
            ioc.value, ioc.ioc_type.value, ioc.source, ioc.confidence,
            json.dumps(ioc.tags), ioc.first_seen.isoformat(),
            ioc.last_seen.isoformat(), json.dumps(ioc.raw),
        ))
    conn.commit()
    logger.info("Upserted %d IOCs from %s", len(iocs), iocs[0].source if iocs else "?")

def run_adapter(adapter, conn: sqlite3.Connection):
    try:
        iocs = adapter.fetch()
        upsert_iocs(conn, iocs)
    except Exception as e:
        logger.error("Adapter %s failed: %s", type(adapter).__name__, e)

def main():
    conn = sqlite3.connect(DB_PATH)
    init_db(conn)

    abuse_adapter = AbuseIPDBAdapter(api_key="YOUR_ABUSEIPDB_KEY")

    schedule.every(1).hours.do(run_adapter, abuse_adapter, conn)
    run_adapter(abuse_adapter, conn)  # run once on startup

    while True:
        schedule.run_pending()
        time.sleep(30)

if __name__ == "__main__":
    main()

A few things worth noting: error handling per adapter means one failing source doesn't block the others. The UNIQUE(value, source) constraint plus upsert means you can safely re-run without duplicates. The 30-second sleep in the loop is intentional — schedule is not async, and you don't need async here.

Querying the Store

The point of all this is fast lookups at runtime — when a log event contains an IP or domain, you want to know immediately if it's in your intel store:

def lookup_ioc(
    conn: sqlite3.Connection,
    value: str,
    min_confidence: int = 70,
) -> list[dict]:
    cursor = conn.execute("""
        SELECT value, ioc_type, source, confidence, tags, last_seen
        FROM iocs
        WHERE value = ? AND confidence >= ?
        ORDER BY confidence DESC
    """, (value, min_confidence))
    return [
        {
            "value": r[0], "type": r[1], "source": r[2],
            "confidence": r[3], "tags": json.loads(r[4]), "last_seen": r[5],
        }
        for r in cursor.fetchall()
    ]

# Integration example: check every extracted IP from a log line
matches = lookup_ioc(conn, "185.220.101.45")
if matches:
    logger.warning("Known threat actor: %s", matches)

The min_confidence parameter matters. Set it to 50 and you'll get noisy results. Set it to 90 and you'll miss things. 70 is a reasonable starting point — tune it per source based on observed false positive rates in your environment.

What to Add Next

This scaffold handles the core use case in under 200 lines. Here's what a production deployment adds:

More feed sources: AlienVault OTX has a Python SDK (OTXv2). Feodo Tracker is free with no auth — just a CSV download. The CISA KEV catalog is a JSON file updated daily at a stable URL.

Operational hardening:

• Rate limiting per adapter — most free-tier APIs have hourly caps. Add time.sleep() between requests or a token bucket.
• IOC expiration — IPs rotate. Add a scheduled cleanup: DELETE FROM iocs WHERE last_seen < datetime('now', '-30 days').
• Parallel fetching — once you have five adapters, switch to concurrent.futures.ThreadPoolExecutor for the fetch phase. Keep DB writes single-threaded to avoid contention.
• Alerting — call a Telegram or Slack webhook inside run_adapter when high-confidence IOCs are found.

For teams who want to go deeper on hardening the infrastructure around this kind of tooling — API exposure, secret management, network segmentation — the security checklists we publish cover those patterns step by step.

The Takeaway

A threat intel aggregator is not a 10,000-line project. The core is under 200 lines: a normalized data model, one adapter class per source, a scheduling loop, and an indexed SQLite store. Start there, add sources one by one, and measure your false positive rate before adding complexity.

The mistake most teams make is waiting for a commercial TIP budget that never gets approved. You can have something useful running this afternoon.

---

I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.