form4api

Python client for Form4API — real-time SEC Form 4 insider trading, Form 144 intent-to-sell, institutional 13F-HR, and congressional STOCK Act trading data, including the insider/Congress convergence signal.

Supports Python 3.11+. Uses httpx for both sync and async HTTP.

Installation

pip install form4api

Sync quickstart

from form4api import Form4ApiClient

client = Form4ApiClient("YOUR_API_KEY")

# Recent open-market purchases at Apple (excluding 10b5-1 plan trades)
txns = client.transactions.list(ticker="AAPL", code="P", exclude_10b5=True, per_page=5)
for t in txns:
    print(t.insider_name, t.insider_title, t.shares_amount, "@", t.price_per_share)
    print(f"  open market: {t.is_open_market}, 10b5 plan: {t.is10b5_plan}, value: ${t.total_value:,.0f}")

# Company overview (includes SIC, state, website)
company = client.companies.get("MSFT")
print(company.name, company.active_insiders, "active insiders")
print(company.sic_description, company.state_of_incorporation)

# Similar companies in the same industry, ranked by Form 4 filings in the last 12 months
peers = client.companies.peers("MSFT", limit=5)
for p in peers.peers:
    print(p.ticker, p.name, p.filings12m, p.match_level)  # peers.basis: sic4 | sic2 | mixed | none

# Insider detail
insider = client.insiders.get("0001234567")
print(insider.name, insider.officer_title)

# Cluster-buy signals (Business plan)
signals = client.signals.list(cluster_buy=True)
for sig in signals:
    print(sig.company_name, sig.insider_count, "buyers on", sig.signal_date)

Async quickstart

import asyncio
from form4api import AsyncForm4ApiClient

async def main():
    async with AsyncForm4ApiClient("YOUR_API_KEY") as client:
        txns = await client.transactions.list(ticker="AAPL", per_page=5)
        for t in txns:
            print(t.insider_name, t.shares_amount, "@", t.price_per_share)

asyncio.run(main())

Resources

Congress date quality. Each congress trade (and convergence leg) has disclosure_lag_days: int | None and date_quality: str | None. When the filing's own dates are impossible or implausible, the lag is None and date_quality is one of "transaction_after_disclosure", "future_transaction_date" or "implausible_lag"; the raw dates are still returned. A lag over 45 days is not a legal finding.

for t in client.congress.trades(ticker="NVDA"):
    lag = f"dates unreliable ({t.date_quality})" if t.disclosure_lag_days is None else f"{t.disclosure_lag_days}d"
    print(t.politician.full_name, t.transaction_date, lag)

Every endpoint that returns typed JSON has a method here, including every plan-gated one. The exceptions are the CSV exports, the account/billing bootstrap calls and the usage endpoints, which have no typed response and are not wrapped. Calling one your key isn't entitled to raises PlanError (HTTP 402) carrying required_plan, current_plan, and upgrade_url rather than failing opaquely.

An AsyncForm4ApiClient mirrors the whole surface with the same resources and method names — await client.insiders.leaderboard().

For the full parameter reference see the REST docs. For LLM workflows, form4api-mcp exposes the same endpoints as tools.

Transaction filters

client.transactions.list(
    ticker="AAPL",        # filter by ticker — accepts up to 25 comma-separated symbols
    cik="0000320193",     # or by company CIK
    insider_cik="...",    # filter by insider CIK
    code="P",             # transaction code: P=purchase, S=sale, A=grant, etc.
    from_date="2026-01-01",
    to_date="2026-12-31",
    filed_from="2026-01-01",  # filters on Filing.FiledAt (when it hit EDGAR), not the trade date
    filed_to="2026-12-31",
    exclude_10b5=True,    # omit trades filed under a Rule 10b5-1 plan
    per_page=100,
    page=1,
)

# Multi-ticker lookup in one call
client.transactions.list(ticker="AAPL,MSFT,NVDA")

Granular filtering (v0.4.0+)

# The "just show me real buys & sells" preset: open-market only,
# no 10b5-1 plan trades, no derivatives.
client.transactions.list(ticker="AAPL", significant=True)

# Multi-code include / exclude (comma-separated SEC codes)
client.transactions.list(codes="P,S")
client.transactions.list(exclude_codes="A,M,F,G")

# Whole-category filters: open_market | grants | derivatives | gifts | other
client.transactions.list(category="open_market")
client.transactions.list(exclude_category="derivatives")
client.transactions.list(exclude_derivative=True)

# Trade-size screening
client.transactions.list(min_value=1_000_000)          # USD, shares x price — every plan
client.transactions.list(max_value=5_000_000)          # Pro+
client.transactions.list(min_shares=10_000, max_shares=100_000)  # Pro+

# Post-trade return screening (Pro+). Returns are FRACTIONS: 0.2 = +20%.
# Horizons: 1d, 1w, 1m, 3m, 6m. Rows with no computed return are excluded.
client.transactions.list(min_return_3m=0.10, max_return_3m=0.50)
client.transactions.list(has_returns=True)  # only rows with all five horizons computed

# Other filters and ordering
client.transactions.list(inst_ownership_trend="increasing")  # increasing | decreasing | stable (13F)
client.transactions.list(role="officer")                     # officer | director | ten_percent_owner
client.transactions.list(sort="value")                       # "date" (default) or "value" (largest first)

paginate() accepts the same filters.

Transaction fields

Company fields

Webhooks

sub = client.webhooks.create(
    "https://example.com/hook",
    ["TransactionFiled", "ClusterBuy"],
    auth_header={"name": "Authorization", "value": "Bearer ..."},  # optional, sent with every delivery
)
print(sub.secret)  # shown once; store it

client.webhooks.update(sub.subscription_id, "https://example.com/new-hook")  # change the URL; returns WebhookSubscription incl. url_warning
client.webhooks.test(sub.subscription_id)      # WebhookTestResult(delivered, status_code, duration_ms, error)
client.webhooks.enable(sub.subscription_id)    # undo an auto-disable (disabled_reason == "delivery_failures")
client.webhooks.set_auth_header(sub.subscription_id, "Authorization", "Bearer new")
client.webhooks.delete_auth_header(sub.subscription_id)  # returns None (204, no body)

Event types (exact names):

Payloads are compact PascalCase references with no envelope (the body is the bare object), for example {"FilingId": ..., "AccessionNumber": ...} for TransactionFiled. The event name is in the X-Event-Type header, not the body. CongressTradeFiled and ConvergenceSignal carry the full event body.

webhooks.list() returns delivery health on each WebhookSubscription: is_active, disabled_at, disabled_reason, last_attempt_at, last_success_at, last_status_code, last_error_class, is_read_only and auth_header_name, all None when not applicable. The auth header value is write-only and never returned. webhooks.test() reports a failed send as delivered=False with an error, it does not raise.

Moving a webhook to a new URL. update(id, url) changes the delivery URL and keeps the id, secret, event types and auth header. It does not re-enable a subscription the system disabled, so call enable(id) afterwards if needed. The last_* health fields come back None because they described the old URL.

# Started on a tunnel, now moving to a stable endpoint
moved = client.webhooks.update(sub.subscription_id, "https://example.com/hooks/form4api")
if not moved.is_active:
    client.webhooks.enable(moved.subscription_id)

url_warning. create(), update() and each list() item carry url_warning: a string when the URL's host is a temporary tunnel or test host (trycloudflare, ngrok, webhook.site, loca.lt, serveo), otherwise None. It is a warning only; the subscription is saved either way, and deliveries stop when the tunnel closes. URLs over 500 characters are rejected with 400 INVALID_URL.

Verifying deliveries. Each delivery carries X-Insider-Signature: sha256=<hex HMAC-SHA256(secret, raw body)>. Verify it against the raw request body, before parsing JSON:

from form4api import verify_webhook

body = request.get_data()                     # raw bytes, e.g. Flask
sig = request.headers["X-Insider-Signature"]
if not verify_webhook(body, sig, WEBHOOK_SECRET):   # str or bytes; constant-time compare
    abort(401)

Search

client.search(q, limit=None) looks up companies and insiders by name or ticker in a single call — useful for a combined search box where you don't know in advance whether the user typed a ticker or a person's name. q must be 2-64 characters; outside that range the API returns HTTP 400 and the SDK raises Form4ApiError with error_code "QUERY_TOO_SHORT" or "QUERY_TOO_LONG". Insider matching is whitespace-tokenised, so "tim cook" matches an insider named "Cook Timothy D". limit caps the number of results per array (1-20, API default 8).

result = client.search("tim cook")
for c in result.companies:
    print(c.ticker, c.name, c.cik)
for i in result.insiders:
    # i.ticker is a ticker associated with the insider and may be None —
    # its exact meaning is still settling on the backend.
    print(i.cik, i.name, i.title, i.ticker)

Pagination

paginate() pages through the data until it runs out (a short or empty page) or, since the backend's 2026-08-01 plan-gated pagination depth (Free: 20 pages, Starter: 100, Pro+: unlimited), the next page is rejected with a 402. That 402 is not swallowed — it raises PaginationLimitError mid-iteration, after every page already yielded has been delivered to your loop. Pages you already received are real and complete; the error only means iteration stopped early. Pass max_pages to stop deliberately before that ever happens, or catch PaginationLimitError to know when a Free/Starter key ran out of depth on a bulk pull:

from form4api import PaginationLimitError

# transactions.paginate() — yields one list per page automatically
all_txns = []
try:
    for batch in client.transactions.paginate(ticker="NVDA", exclude_10b5=True, per_page=100):
        all_txns.extend(batch)
except PaginationLimitError as e:
    print(f"Stopped after {e.pages_yielded} pages — {e}")
    # all_txns still holds every page yielded before the limit hit

# signals.paginate()
all_signals = []
for batch in client.signals.paginate(cluster_buy=True, per_page=100, max_pages=10):
    all_signals.extend(batch)

Error handling

from form4api import (
    Form4ApiClient,
    AuthError,
    PlanError,
    PaginationLimitError,
    RateLimitError,
    NotFoundError,
)

client = Form4ApiClient("YOUR_API_KEY")

try:
    signals = client.signals.list()
except PaginationLimitError as e:
    print(f"Paginate stopped after {e.pages_yielded} pages — upgrade to go deeper")
except PlanError as e:
    print(f"Upgrade required")
except RateLimitError as e:
    print(f"Retry after {e.retry_after}s")
except AuthError:
    print("Invalid API key")
except NotFoundError:
    print("Resource not found")

License

MIT