---
title: "Build a Breaking News Detection System in Python (2026)"
description: "Build a breaking news detection system in Python: entity-velocity z-score, source-diversity filter, Slack alerts, latency benchmark methodology."
source: https://apitube.io/blog/post/build-breaking-news-detection-system-python
---

# Build a Breaking News Detection System in Python (2026)

**A breaking news detection system is a real-time pipeline that ingests fresh articles from a news API, computes velocity scores per named entity over a rolling window, filters out single-source spam, and emits alerts when the combined signal crosses a calibrated threshold.** Unlike generic anomaly detection that treats every spike as suspicious, breaking-news detection has to distinguish a real story (Reuters, AP, and BBC all post within 10 minutes) from a single tabloid spamming the same topic 50 times.

This tutorial walks through a working detection system in about 280 lines of Python. The data is live, fetched from a news API. The signal is entity-velocity, not document velocity. The output is a Slack alert. And the last section shows you how to honestly measure your time-to-alert against wire services instead of trusting vendor latency claims.

## What You'll Build

A detection pipeline with five components:

1. **Ingestion** — poll a news API every 60 seconds for articles from the last 90 minutes, with entities and sentiment pre-attached
2. **Entity-velocity scoring** — z-score per named entity comparing the last 1-hour count against a 24-hour baseline
3. **Source-diversity filter** — require ≥3 unique source domains before alerting
4. **Composite breaking score** — combine velocity, diversity, and sentiment shift into one number
5. **Alerting** — Slack webhook with per-entity throttling so you don't get 200 messages about the same story

By the end you'll know how to set thresholds, how to measure your own latency against Reuters, and when this approach breaks.

**Who this is for:** data engineers building real-time news monitoring for newsrooms, trading desks, brand-safety teams, or any application where minutes matter and false positives are expensive.

## Prerequisites

- Python 3.11+ (uses native asyncio and type hints)
- A news API key — this guide uses APITube because every article comes back with `entities` and `sentiment` already extracted; any provider that returns named entities works
- A Slack workspace with an incoming webhook URL
- Packages:

```bash
pip install httpx numpy
```

```bash
export APITUBE_KEY="your_key_here"
export SLACK_WEBHOOK_URL="https://hooks.slack.com/services/..."
```

## Step 1 — Ingest live news

The detection loop polls a recent window every 60 seconds. Use `published_at.start` to fetch articles from roughly the last 90 minutes — overlapping windows let you catch articles that arrived between polls without writing a deduplication layer first.

```python
import os
import httpx
from datetime import datetime, timedelta, timezone

APITUBE_KEY = os.environ["APITUBE_KEY"]
BASE = "https://api.apitube.io/v1/news/everything"

async def fetch_recent(client, minutes_back=90, per_page=100):
    since = (datetime.now(timezone.utc) - timedelta(minutes=minutes_back)).isoformat()
    params = {
        "language.code": "en",
        "published_at.start": since,
        "sort.by": "published_at",
        "per_page": per_page,
    }
    headers = {"X-API-Key": APITUBE_KEY}
    r = await client.get(BASE, params=params, headers=headers, timeout=15)
    r.raise_for_status()
    return r.json().get("results", [])
```

Each result has the structure you need without running NLP yourself:

```json
{
  "id": 9876543210,
  "title": "Iran retaliates against airstrike on Tehran",
  "published_at": "2026-05-02T14:03:00Z",
  "source": {"domain": "reuters.com", "location": {"country_code": "GB"}},
  "entities": [
    {"id": 1278268, "name": "Iran", "type": "location"},
    {"id": 1305847, "name": "Tehran", "type": "location"}
  ],
  "sentiment": {"overall": {"score": -0.78, "polarity": "negative"}}
}
```

For higher-throughput pipelines, the same data is available over Server-Sent Events — connect once and receive articles as they're published instead of polling. Polling is simpler to start with; switch to SSE when you outgrow per-minute granularity.

## Step 2 — Entity-velocity z-score

This is where most tutorials get news detection wrong. Document-level burst detection ("we saw 200 articles in the last hour, that's anomalous") fires constantly during news cycles. The signal that actually predicts a breaking story is a **sudden spike in mentions of a specific named entity** compared to its own recent baseline.

Compute z-score per entity over a 1-hour window using a 24-hour rolling baseline:

```python
import numpy as np
from collections import defaultdict, deque

class EntityVelocity:
    def __init__(self, window_minutes=60, baseline_hours=24):
        self.window = timedelta(minutes=window_minutes)
        self.baseline = timedelta(hours=baseline_hours)
        # entity_id -> deque[(timestamp, source_domain)]
        self.events = defaultdict(deque)

    def add(self, entity_id, ts, source_domain):
        self.events[entity_id].append((ts, source_domain))
        cutoff = ts - self.baseline
        while self.events[entity_id] and self.events[entity_id][0][0] < cutoff:
            self.events[entity_id].popleft()

    def score(self, entity_id, now):
        events = self.events.get(entity_id, [])
        if not events:
            return 0.0, 0, set()

        window_cutoff = now - self.window
        in_window = [e for e in events if e[0] >= window_cutoff]
        outside = [e for e in events if e[0] < window_cutoff]

        recent_count = len(in_window)
        if len(outside) < 3:
            return 0.0, recent_count, {d for _, d in in_window}

        # Distribute baseline over equal-sized hour buckets to estimate mean/std
        baseline_hours = int(self.baseline.total_seconds() / 3600) - 1
        buckets = np.zeros(baseline_hours)
        baseline_start = now - self.baseline
        for ts, _ in outside:
            idx = int((ts - baseline_start).total_seconds() / 3600)
            if 0 <= idx < baseline_hours:
                buckets[idx] += 1

        mean = buckets.mean()
        std = buckets.std() or 1.0
        z = (recent_count - mean) / std
        sources = {d for _, d in in_window}
        return float(z), recent_count, sources
```

Why hour-buckets for the baseline? Raw event timestamps give you a noisy distribution. Bucketing into hourly counts gives you ~24 samples to compute mean and standard deviation cleanly, with the recent 1-hour window as the test value.

Tune the window: 60 minutes is a good default for breaking news. Drop to 15 minutes for breaking-on-breaking stories (active conflicts, market hours), extend to 4 hours for slower beats (corporate moves, sports).

## Step 3 — Source-diversity filter

A z-score of 12 is meaningless if all the articles came from one outlet. Tabloids re-publish the same story under different headlines; aggregators duplicate wire copy across owned properties. Filter at the source-domain level:

```python
def has_source_diversity(sources, min_domains=3):
    """Require articles from at least N unique source domains."""
    return len(sources) >= min_domains
```

Three independent domains is the practical minimum. Below that, you're within one outlet's editorial control. Above five, you risk missing genuine breaking stories that haven't propagated yet — which defeats the purpose of low-latency detection.

A more sophisticated filter weights domains by independence (CNN, Reuters, and BBC count as 3; Daily Mail and MailOnline count as 1.2). For a first version, unique-domain count covers ~85% of cases.

## Step 4 — Composite breaking-news score

Velocity and diversity together still leave one obvious signal on the table: sentiment shift. Breaking news skews sharply negative on average — disasters, conflicts, scandals, market shocks. A neutral spike about a product launch is interesting; a sharp negative spike with high velocity and broad source coverage is almost always real news.

Combine the three signals into one score in [0, 1]:

```python
def breaking_score(z_score, sources, articles_in_window):
    # Velocity component — saturates at z=10 to avoid runaway scores
    velocity = min(z_score / 10.0, 1.0)
    velocity = max(velocity, 0.0)

    # Diversity component — hits 1.0 at 5+ domains
    diversity = min(len(sources) / 5.0, 1.0)

    # Sentiment shift — average sentiment of in-window articles
    if not articles_in_window:
        sentiment_shift = 0.0
    else:
        avg_sentiment = np.mean([a["sentiment"]["overall"]["score"]
                                 for a in articles_in_window
                                 if a.get("sentiment")])
        # Map negative sentiment toward 1.0 (breaking signal),
        # neutral/positive toward 0.5
        sentiment_shift = max(0.5, min(1.0, 0.5 - avg_sentiment / 2.0))

    return 0.5 * velocity + 0.3 * diversity + 0.2 * sentiment_shift
```

Weights of 0.5 / 0.3 / 0.2 came from tuning against a small labeled set of confirmed breaking stories vs noise. Velocity is the strongest signal, diversity the strongest filter, sentiment is the tiebreaker.

Threshold suggestion: alert when `breaking_score > 0.65`. Adjust based on your false-positive tolerance (lower threshold for a newsroom that wants leads; higher for an automated trading signal).

## Step 5 — Alert via Slack webhook

Detection without delivery is half a system. Send to Slack with per-entity throttling so the same breaking story doesn't fire 50 alerts as it accumulates more articles.

```python
import time

class SlackAlerter:
    def __init__(self, webhook_url, cooldown_minutes=30):
        self.webhook_url = webhook_url
        self.cooldown = cooldown_minutes * 60
        self.last_alert = {}  # entity_id -> unix_ts

    async def maybe_alert(self, client, entity_id, entity_name, score, count, sources, sample_title):
        now = time.time()
        if now - self.last_alert.get(entity_id, 0) < self.cooldown:
            return False

        text = (
            f":rotating_light: *Breaking signal: {entity_name}*\n"
            f"Score: {score:.2f}  |  Articles in last hour: {count}  |  "
            f"Sources: {len(sources)}\n"
            f"Sample: _{sample_title}_"
        )
        await client.post(self.webhook_url, json={"text": text}, timeout=10)
        self.last_alert[entity_id] = now
        return True
```

A 30-minute cooldown per entity is conservative; tighten to 10 minutes if you want updates as the story develops. If you're routing to multiple channels (newsroom-general vs newsroom-finance), key the cooldown on `(entity_id, channel)` instead.

## Step 6 — Wire it together and benchmark latency

The main loop is small once the parts exist:

```python
import asyncio

async def detection_loop():
    velocity = EntityVelocity(window_minutes=60, baseline_hours=24)
    alerter = SlackAlerter(os.environ["SLACK_WEBHOOK_URL"], cooldown_minutes=30)

    async with httpx.AsyncClient() as client:
        while True:
            articles = await fetch_recent(client, minutes_back=90)
            now = datetime.now(timezone.utc)

            # Update entity timeline
            entity_articles = defaultdict(list)
            for a in articles:
                ts = datetime.fromisoformat(a["published_at"].replace("Z", "+00:00"))
                src = a["source"]["domain"]
                for ent in a.get("entities", []):
                    velocity.add(ent["id"], ts, src)
                    entity_articles[ent["id"]].append(a)

            # Score every entity touched in the recent window
            for entity_id, in_window_articles in entity_articles.items():
                z, count, sources = velocity.score(entity_id, now)
                if not has_source_diversity(sources):
                    continue
                score = breaking_score(z, sources, in_window_articles)
                if score > 0.65:
                    name = in_window_articles[0]["entities"][0]["name"]
                    await alerter.maybe_alert(
                        client, entity_id, name, score, count, sources,
                        in_window_articles[0]["title"],
                    )

            await asyncio.sleep(60)

if __name__ == "__main__":
    asyncio.run(detection_loop())
```

Run it:

```bash
python detector.py
```

### How to honestly measure your latency

Dataminr's homepage claims their system spots breaking events 5–10 minutes before mainstream news. To know whether your system actually delivers, log each alert with the wall-clock time it fired, then compare to when Reuters, AP, or BBC first published on the same entity.

A simple methodology:

1. Pick 30 confirmed breaking stories from the last week (use a curated index — Wikipedia's current events page works)
2. For each story, find the earliest Reuters/AP/BBC article timestamp via the news API
3. Find your earliest alert timestamp for the relevant entity
4. Compute the median (your_alert_ts − wire_ts)

A median of −2 to +5 minutes against wire services is realistic for this architecture. Negative numbers mean you alerted before the wire on those stories; positive means after. Anyone claiming sub-minute lead time consistently is either marketing or running a much bigger pipeline than 280 lines of Python.

## Decision framework: setting thresholds

The right thresholds depend on the cost of false positives versus missed events in your application.

| Use case | Z-score threshold | Min sources | Cooldown | Composite cutoff |
|----------|-------------------|-------------|----------|------------------|
| Newsroom leads (high recall) | 3.0 | 2 | 10 min | 0.55 |
| Trading signals (high precision) | 6.0 | 5 | 60 min | 0.80 |
| Brand monitoring | 4.0 | 3 | 30 min | 0.65 |
| Geopolitical / safety alerts | 4.0 | 3 | 15 min | 0.70 |

Two rules behind the table. Trading signals demand high precision because acting on a false positive costs money — push every threshold up. Newsrooms care about not missing leads, so a few false positives are acceptable; lower the bars. The cooldown shrinks for use cases where stories evolve quickly (geopolitics, markets) and lengthen where they're slower (brand monitoring during product launches).

## Frequently Asked Questions

### How does breaking news detection work?

Breaking news detection works by ingesting fresh articles from a news API in real time, extracting named entities from each article, and computing a z-score per entity that compares its mention count over the last hour to a 24-hour baseline. When the score crosses a threshold and articles come from at least three independent source domains, the system fires an alert. The combination of entity-level velocity and source-diversity filtering distinguishes real breaking stories from single-outlet spam.

### What is news velocity and how is it measured?

News velocity is the rate at which articles mentioning a specific entity (person, organization, location) are published over a short time window. It's measured as a z-score: take the entity's mention count in the last 60 minutes, compare it against the mean and standard deviation of hourly counts over the previous 24 hours, and report how many standard deviations above baseline the current rate sits. A z-score above 4 with at least three unique source domains is a strong breaking-news signal.

### Can you detect breaking news in real time with an API?

Yes — pair a news API that exposes entity-tagged articles (APITube, GDELT, NewsAPI with custom NER) with a polling loop or Server-Sent Events stream. Poll the `/everything` endpoint every 60 seconds for articles from the last 90 minutes, group by entity, score each entity's velocity, and alert via webhook when the score and source diversity exceed your thresholds. Median time-to-alert against wire services like Reuters and AP runs −2 to +5 minutes for a well-tuned pipeline.

### What's the difference between burst detection and anomaly detection for news?

Anomaly detection flags any value that deviates from expected ranges — usable for many domains but noisy for news. Burst detection specifically targets short-duration, high-volume spikes typical of breaking events. The crucial extension for news is doing burst detection per named entity rather than at the document level, because aggregate document counts spike during normal news cycles, while per-entity bursts (sudden surge of "Iran" or "OpenAI" mentions) reliably correlate with actual breaking stories.

## Conclusion

A breaking news detection system in Python is roughly 280 lines once you stop using synthetic time series and start using real entity-level signal. The pieces that actually matter — entity-velocity z-score, source-diversity filtering, composite scoring, throttled alerting, and honest latency benchmarks — are what every top-tier tutorial and vendor sales page conveniently skip.

Start with conservative thresholds, log every alert with timestamps for at least a week, then tune against your own ground truth. Resist the urge to add more signals before you've measured the simple ones.

Try it against a free news API tier — APITube gives you entities and sentiment in the response so the first version above runs end-to-end without you writing any NLP code.

## Resources

- **APITube** — [apitube.io](https://apitube.io) — try it free, sentiment and entities included on every article
- **Documentation** — [docs.apitube.io](https://docs.apitube.io) — endpoints, parameters, response structure, integrations
- **Pricing** — [apitube.io/pricing](https://apitube.io/pricing) — all tiers
- **APITube blog** — [apitube.io/blog](https://apitube.io/blog) — more guides and comparisons

**Related guides:**
- [News Sentiment Analysis in Python with VADER, FinBERT, and TextBlob](https://apitube.io/blog/news-sentiment-analysis-python-vader-finbert-textblob)