Insider Trading Alerts in Python: Polling New Filings Without Missing Any

You want a message when a director buys shares in a company you follow. The obvious design is a loop: every few minutes, ask the API for anything published since the last check. That design has a bug that only shows up weeks later, when you notice a filing your alert never sent.

This post builds the alert properly, in about 40 lines of Python.

The naive version

last_seen = "2026-11-02T00:00:00Z"
while True:
    rows = get("/transactions", published_after=last_seen)
    for row in rows:
        alert(row)
    last_seen = max(row["published_at"] for row in rows)
    sleep(300)

It looks right. It will miss filings.

Why: publication time is not arrival time

published_at is the time the register published the filing, taken from the register itself. It is not the time the filing reached the API. Those differ, sometimes by days:

• Some registers only give a date. A filing published "on 2 November" carries a midnight timestamp, though it may only appear on the register that evening.
• Registers are polled on a schedule, and a busy register's queue takes time.
• Some registers publish after a check. BaFin, for example, publishes a notification after an administrative review, dated the day it went live.
• Recovered filings arrive late by design. When a filing that couldn't be parsed is fixed and re-read, it enters the API with its original publication time.

So a filing with published_at = 09:00 can show up at 15:00, after your cursor has moved past 09:00. A strict "after the last one I saw" query never returns it.

The fix: overlap, then deduplicate

Ask for a window that reaches back further than any plausible delay, and remember which rows you have already handled. Every transaction has a stable integer id, so deduplication is a set lookup.

import os
import time
from datetime import datetime, timedelta, timezone

import requests

API = "https://api.duceusapi.com"
HEADERS = {"Authorization": f"Bearer {os.environ['DUCEUS_API_KEY']}"}
OVERLAP = timedelta(days=7)   # longer than any delay we have seen
PAGE = 500

seen: set[int] = set()        # persist this (a file, SQLite) in real use


def fetch_since(since: datetime) -> list[dict]:
    rows, offset = [], 0
    while True:
        response = requests.get(
            f"{API}/transactions",
            headers=HEADERS,
            params={
                "transaction_type": "BUY",
                "published_after": since.isoformat(),
                "limit": PAGE,
                "offset": offset,
            },
            timeout=30,
        )
        if response.status_code == 429:  # rate limited: wait for the window to reset
            time.sleep(int(response.headers.get("Retry-After", "60")))
            continue
        response.raise_for_status()
        items = response.json()["items"]
        rows.extend(items)
        if len(items) < PAGE:
            return rows
        offset += PAGE


def alert(deal: dict) -> None:
    print(
        f"{deal['published_at'][:16]}  {deal['issuer_name']}: "
        f"{deal['person_role']} bought {deal['volume']:g} @ {deal['unit_price']} {deal['currency']}"
    )


while True:
    for deal in fetch_since(datetime.now(timezone.utc) - OVERLAP):
        if deal["id"] not in seen and not deal["is_closely_associated"]:
            alert(deal)
        seen.add(deal["id"])
    time.sleep(15 * 60)

Three details matter:

• The window is relative to now, not to the last row seen. A late row is caught on the next pass whatever its timestamp.
• Seen ids are kept for at least the overlap period, so nothing alerts twice.
• Withdrawn filings disappear on their own. When a register marks a filing cancelled or superseded, the API stops returning it. If you store rows locally, re-check the ones you hold during the overlap window.

Keep the request count down

Each pass re-reads the whole overlap window, one request per 500 rows. Filter on the server (country, transaction_type, issuer_id, role) rather than in Python so a pass stays at one or two pages, and pick the polling interval to fit your plan's hourly limit. Every response carries X-RateLimit-Remaining, and a 429 comes with Retry-After.

To watch only some companies, run the same loop per company with issuer_id=iss_…. Ids are stable: if two company records are later merged, the old id still resolves.

Watch both datasets at once

/feed returns insider deals and major shareholding notifications in one stream, each tagged transaction or holding, with the same published_after parameter. The overlap-and-deduplicate pattern works the same way. Key the seen-set on (kind, id), because the two datasets number their rows separately.

Only the interesting buys

Most alerts are noise: a director's spouse buying €3,000 of shares. The /signals endpoint returns only market purchases at a price by the insider themselves, each with a score and the reasons behind it (CEO or CFO buyer, cluster of buyers, unusually large deal). Point the same loop at it:

params={"preset": "insider_buy", "min_score": 50, "published_after": since.isoformat()}

Summary

• published_at is when the register published, not when the row arrived; rows can arrive days later.
• Poll a sliding window that reaches back further than any plausible delay, and deduplicate on id.
• Filter on the server, page with offset, and honour Retry-After.
• /feed covers both datasets; /signals cuts the noise.