Metadata-Version: 2.4
Name: form4api
Version: 0.11.0
Summary: Python client for the 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
License-Expression: MIT
Project-URL: Homepage, https://www.form4api.com
Project-URL: Documentation, https://www.form4api.com/docs
Project-URL: Repository, https://github.com/theodor90/form4api-py.git
Keywords: insider-trading,sec,sec-edgar,edgar,form-4,form4,form-144,13f,13f-hr,congress,stock-act,financial-data,stock-market,stocks,fintech,api,sdk,webhooks,form4api
Requires-Python: >=3.11
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: httpx>=0.27
Provides-Extra: dev
Requires-Dist: pytest>=8; extra == "dev"
Requires-Dist: pytest-httpx>=0.30; extra == "dev"
Requires-Dist: respx>=0.21; extra == "dev"
Dynamic: license-file

# form4api

Python client for [Form4API](https://www.form4api.com) — 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.

[![PyPI version](https://img.shields.io/pypi/v/form4api.svg)](https://pypi.org/project/form4api/)
[![PyPI downloads](https://img.shields.io/pypi/dm/form4api.svg)](https://pypi.org/project/form4api/)
[![Python versions](https://img.shields.io/pypi/pyversions/form4api.svg)](https://pypi.org/project/form4api/)
[![license](https://img.shields.io/pypi/l/form4api.svg)](https://github.com/theodor90/form4api-py/blob/main/LICENSE)

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

## Installation

```bash
pip install form4api
```

## Sync quickstart

```python
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)

# 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

```python
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

| Resource | Methods |
|---|---|
| `client.search(q, limit=None)` | Combined company + insider lookup — not namespaced under a resource |
| `client.transactions` | `.list(**params)`, `.paginate(**params)` |
| `client.insiders` | `.search(name, **params)`, `.get(cik)`, `.list(**params)`, `.directory(**params)`, `.transactions(cik, **params)`, `.summary(cik)` *(Pro)*, `.scorecard(cik)` *(Pro)*, `.leaderboard(**params)` *(Business)* |
| `client.companies` | `.get(ticker)`, `.insiders(ticker)`, `.list(**params)` |
| `client.signals` | `.list(**params)`, `.paginate(**params)`, `.explain(ticker)`, `.sentiment(ticker, **params)` — Business; `.convergence(**params)` — Pro |
| `client.congress` | `.trades(**params)`, `.politicians(**params)` *(Pro)*, `.politician(id_or_slug)` *(Pro)*, `.ticker(ticker)` *(Pro)* |
| `client.form144` | `.list(**params)` — Business plan |
| `client.holdings` | `.list(**params)`, `.managers(**params)` — Business plan |
| `client.ownership` | `.schedule_13dg(**params)` — Business plan (Schedule 13D/13G >5% ownership crossings; filters `ticker`, `form_type`, `is_amendment`, `page`, `per_page`) |
| `client.filings` | `.list(**params)`, `.recent(**params)`, `.get(accession_number)` |
| `client.stats` | `.get()` — public, no key required |
| `client.data_quality` | `.get()` — public, no key required |
| `client.status` | `.history()` — takes no parameters |
| `client.webhooks` | `.create(url, event_types, auth_header=None)`, `.list()`, `.update(id, url)`, `.delete(id)`, `.events(**params)`, `.enable(id)`, `.test(id)`, `.set_auth_header(id, name, value)`, `.delete_auth_header(id)` |
| `verify_webhook(body, signature, secret)` | Top-level function (not on the client) that checks the `X-Insider-Signature` header on a delivery |

**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.

```python
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](https://form4api.com/docs).
For LLM workflows, `form4api-mcp` exposes the same endpoints as tools.

### Transaction filters

```python
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+)

```python
# 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

| Field | Type | Description |
|-------|------|-------------|
| `ticker` | `str` | Stock ticker |
| `company_name` | `str` | Company name |
| `insider_name` | `str` | Insider full name |
| `insider_cik` | `str` | Insider CIK |
| `insider_title` | `str \| None` | Officer title as reported on the Form 4 |
| `is_director` | `bool` | Director flag |
| `is_officer` | `bool` | Officer flag |
| `is10_pct_owner` | `bool` | 10% owner flag |
| `accession_number` | `str` | SEC accession number |
| `security_title` | `str` | Security type |
| `transaction_code` | `str` | Transaction code |
| `is_open_market` | `bool` | `True` when code is P or S (not grants/awards) |
| `is10b5_plan` | `bool` | Filed under a Rule 10b5-1 pre-scheduled trading plan |
| `shares_amount` | `float` | Shares transacted |
| `price_per_share` | `float \| None` | Price per share |
| `total_value` | `float \| None` | `shares_amount × price_per_share` in USD |
| `shares_owned_after` | `float \| None` | Holdings after transaction |
| `direct_indirect` | `str \| None` | "D" (direct) or "I" (indirect) |
| `is_derivative` | `bool` | Derivative security flag |
| `transaction_date` | `str` | ISO datetime |
| `period_of_report` | `str` | ISO datetime |
| `return1d`/`return1w`/`return1m`/`return3m`/`return6m` | `float \| None` | Forward return since the transaction date, as a fraction (0.109 = +10.9%); `None` until enough time/price data has elapsed |
| `value_quality` | `str \| None` | Confidence flag on `total_value`'s pricing source |
| `filed_at` | `str \| None` | When the filing became public on EDGAR |
| `accepted_at` | `str \| None` | Precise UTC SEC-acceptance instant; `None` if not captured |
| `document_url` | `str \| None` | Public SEC document URL |
| `institutional_ownership` | `InstitutionalOwnership \| None` | Latest 13F institutional ownership snapshot for the ticker, when available |

### Company fields

| Field | Type | Description |
|-------|------|-------------|
| `cik` | `str` | SEC CIK |
| `name` | `str` | Company name |
| `ticker` | `str \| None` | Stock ticker |
| `exchange` | `str \| None` | Exchange |
| `total_filings` | `int` | Total Form 4 filings |
| `active_insiders` | `int` | Distinct insiders who have filed |
| `sic_description` | `str \| None` | SEC SIC industry description |
| `state_of_incorporation` | `str \| None` | Two-letter state code |
| `website` | `str \| None` | Company website as filed with SEC |

### Webhooks

```python
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):

| Event | Plan |
|---|---|
| `TransactionFiled` | Free |
| `ClusterBuy` | Free |
| `ClusterSell` | Free |
| `CongressTradeFiled` | Starter+ |
| `ConvergenceSignal` | Pro+ |

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.

```python
# 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:

```python
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).

```python
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:

```python
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

```python
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
