Six small, complete programs against the FilingPulse API — four in Python, two in
Node. Every script on this page is executed against a live API instance by our
regression suite before it ships, and the download links serve the exact tested files.
Each one reads FILINGPULSE_API_KEY from the environment —
free keys take under a minute. Most use an official SDK
(pip install filingpulse / npm install filingpulse); one is
plain stdlib HTTP on purpose, to show the API needs no client at all.
Paginates /v1/insider-trades, flattens each filing’s transaction
legs into one row per transaction, and coerces the schema’s strings-as-filed
numerics to Decimal at the edge — the API never coerces for you, and some
filings genuinely state no number.
python export_insider_trades_csv.py KMI --since 2026-01-01
def to_decimal(value):
"""Values arrive exactly as filed; not every filing states every number."""
if value is None:
return None
try:
return Decimal(value)
except InvalidOperation:
return None # as-filed junk ("N/A", ...) — keep the row, drop the number
def main() -> None:
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument("ticker", help="Issuer trading symbol, e.g. KMI")
ap.add_argument("--since", help="Only filings filed on/after YYYY-MM-DD")
ap.add_argument("--out", help="Output path (default <TICKER>_insider_trades.csv)")
args = ap.parse_args()
api_key = os.environ.get("FILINGPULSE_API_KEY")
if not api_key:
sys.exit("Set FILINGPULSE_API_KEY (free key: https://filingpulse.io/signup.html)")
fp = FilingPulse(api_key=api_key,
base_url=os.environ.get("FILINGPULSE_BASE_URL",
"https://api.filingpulse.io"))
out_path = args.out or f"{args.ticker.upper()}_insider_trades.csv"
fields = ["period", "form_type", "owner", "officer_title", "security_title",
"transaction_date", "transaction_code", "acquired_disposed",
"shares", "price_per_share", "shares_owned_after", "is_derivative"]
rows, offset = [], 0
while True:
page = fp.insider_trades(ticker=args.ticker, since=args.since,
limit=PAGE_SIZE, offset=offset)
for filing in page["data"]:
owner = filing["reporting_owners"][0]
for tx in filing["transactions"]:
rows.append({
"period": filing["period"],
"form_type": filing["form_type"],
"owner": owner["name"],
"officer_title": owner["officer_title"] or "",
"security_title": tx["security_title"],
"transaction_date": tx["transaction_date"],
"transaction_code": tx["transaction_code"],
"acquired_disposed": tx["acquired_disposed"],
"shares": to_decimal(tx["shares"]),
"price_per_share": to_decimal(tx["price_per_share"]),
"shares_owned_after": to_decimal(tx["shares_owned_after"]),
"is_derivative": tx["is_derivative"],
})
offset += PAGE_SIZE
if offset >= page["total"]:
breakFull script: export_insider_trades_csv.py
Item 2.02 is “Results of Operations and Financial Condition” —
the 8-K companies file when they report results. The same script digests any item code
(--item 5.02 for officer changes, 1.01 for material
agreements). Output is grouped by filing day; items are presented exactly as
classified, nothing more.
python earnings_8k_digest.py --since 2026-08-01
since = args.since or (dt.date.today() - dt.timedelta(days=7)).isoformat()
events = fp.events(item=args.item, since=since, limit=args.limit)["data"]
if not events:
print(f"No item {args.item} events filed since {since}.")
return
caption = next(i["caption"] for e in events for i in e["items"]
if i["code"] == args.item)
print(f"8-K item {args.item} ({caption}) since {since} — {len(events)} filings\n")
def day(e): # filed_date is YYYYMMDD as filed
d = e["filed_date"] or ""
return f"{d[:4]}-{d[4:6]}-{d[6:8]}" if len(d) == 8 else d
for filed_day, group in groupby(events, key=day):
print(filed_day)
for e in group:
other = [i["code"] for i in e["items"] if i["code"] != args.item]
extra = f" (also items {', '.join(other)})" if other else ""
print(f" {e['issuer']['name']} [{e['form_type']}]{extra}")
print()Full script: earnings_8k_digest.py
FilingPulse signs every delivery with your subscription’s secret
(X-FilingPulse-Signature: sha256=…). Always verify before
trusting a payload — anyone can POST JSON at your endpoint; only FilingPulse
knows the secret. This receiver is a complete stdlib HTTP server: it rejects bad
signatures with 400, prints one factual line per delivery, and forwards it
to a Slack incoming webhook if SLACK_WEBHOOK_URL is set.
FILINGPULSE_WEBHOOK_SECRET=whsec_… python webhook_receiver.py --port 8080
class Handler(BaseHTTPRequestHandler):
def do_POST(self):
body = self.rfile.read(int(self.headers.get("Content-Length", 0)))
signature = self.headers.get("X-FilingPulse-Signature", "")
if not verify_webhook_signature(SECRET, body, signature):
self.send_response(400)
self.end_headers()
self.wfile.write(b"bad signature")
return
notify(summarize(json.loads(body)))
self.send_response(200)
self.end_headers()
self.wfile.write(b"ok")
def log_message(self, *args): # signature check already logs what matters
passFull script: webhook_receiver.py
The SEC file number (333-XXXXXX) is the join key EDGAR itself uses
across one offering: the initial S-1/F-1, each amendment, the EFFECT notice, and the
final priced prospectus all carry it. Filter /v1/registrations on
file_number and the whole thread comes back in one call. Plain
urllib — the endpoint works with any HTTP client.
python ipo_lifecycle.py 333-296288
thread = get("/v1/registrations", file_number=file_number, limit=200)["data"]
if not thread:
sys.exit(f"No events on file {file_number}.")
thread.sort(key=lambda e: e["filed_date"] or "")
issuer = thread[-1]["issuer"]
print(f"{issuer['name']} (CIK {issuer['cik']}) — file {file_number}")
if issuer.get("sic"):
print(f" industry: {issuer['sic']} (SIC {issuer['sic_code']})")
for former in thread[-1].get("former_names", []):
print(f" formerly: {former['name']} (until {iso(former['date_changed'])})")
print()
for e in thread:
label = STAGE_LABEL.get(e["stage"], e["stage"])
print(f" {iso(e['filed_date'])} {e['form_type']:<7} {label}")
print(f"\n{len(thread)} tracked event(s). EFFECT notices for untracked form types "
"(S-3, S-8, ...) won't appear unless they share this file number.")Full script: ipo_lifecycle.py
JavaScript. The naive poller re-fetches the whole list every
minute and re-handles filings it already processed. This one does the two things that
make polling cheap and correct: it narrows the window with since=, and it
remembers what it already emitted. The id it dedupes on is accession —
the SEC accession number, the unique, stable identity of every filing, present on
every object the API returns. It also shows the shape of a well-behaved client: a
RateLimitError carries retryAfter, so backing off correctly
is four lines.
node watch_filings.mjs --tickers KMI,PRTH --item 2.02
* One pass over every feed we watch. Returns [{id, text}] newest-first, or null
* if the tier's rate limit was hit — the caller just tries again next cycle.
*
* `accession` is the id: the SEC accession number is the unique, stable identity
* of every filing, and every object type carries it (Form 4 since 2026-08-08,
* 8-K and registration objects since 2026-08-17).
*/
async function poll() {
const found = [];
try {
for (const query of tickers.length ? tickers.map((t) => ({ ticker: t })) : [{}]) {
const page = await fp.insiderTrades({ ...query, since, limit: PAGE });
for (const filing of page.data) {
found.push({ id: filing.accession, text: form4Line(filing) });
}
}
if (item) {
const page = await fp.events({ item, since, limit: PAGE });
for (const event of page.data) {
found.push({ id: event.accession, text: eventLine(event) });
}
}
} catch (err) {
if (err instanceof RateLimitError) {
console.error(`rate limited — backing off ${err.retryAfter}s`);
await sleep(err.retryAfter * 1000);
return null;
}
throw err;
}
return found;
}Each cycle prints only what it has not seen before, oldest first:
// The API returns each list newest-first; print oldest-first so the newest
// filing is always the last line on your screen.
const fresh = found.filter((f) => !seen.has(f.id)).reverse();
for (const f of fresh) {
remember(f.id);
console.log(` ${f.text}`);
}
if (first) {
console.log(` — ${fresh.length} filing(s) already on record`);
first = false;
}
}
if (once) break;
for (let waited = 0; waited < intervalMs && !stopping; waited += 500) {
await sleep(500);
}
}Full script: watch_filings.mjs
JavaScript. Same contract as the Python receiver, with one
language-specific trap worth stating plainly: verification runs over the raw
request bytes. express.json(), Fastify’s default JSON parser
and Next.js route handlers all hand you a parsed object and discard the original bytes —
and re-serializing that object produces different bytes, so every signature fails. Read
the body first, verify, then parse. (In Express: express.raw({ type:
"application/json" }) on this route.) verifyWebhookSignature is
async because it uses WebCrypto’s constant-time HMAC verify.
FILINGPULSE_WEBHOOK_SECRET=whsec_… node webhook_receiver.mjs --port 8080
async function readRawBody(req) {
const chunks = [];
for await (const chunk of req) chunks.push(chunk);
return Buffer.concat(chunks); // a Buffer IS a Uint8Array — pass it straight through
}
const server = http.createServer(async (req, res) => {
if (req.method !== "POST") {
res.writeHead(405).end("post only");
return;
}
const raw = await readRawBody(req);
const signature = req.headers["x-filingpulse-signature"] ?? "";
if (!(await verifyWebhookSignature(SECRET, raw, signature))) {
res.writeHead(400).end("bad signature");
return;
}
// Signature checked against the raw bytes — only now is it safe to parse.
try {
await notify(summarize(JSON.parse(raw.toString("utf8"))));
} catch (err) {
console.error(`delivery accepted but not handled: ${err.message}`);
res.writeHead(500).end("handler error"); // non-2xx makes FilingPulse retry
return;
}
res.writeHead(200).end("ok");
});Full script: webhook_receiver.mjs
All six scripts also accept FILINGPULSE_BASE_URL to point at a
different API base. The Node scripts need Node 18 or newer (they use global
fetch and WebCrypto). The examples present filings exactly as filed and
interpret nothing — FilingPulse is data infrastructure, not investment advice.