Typed Python client for the AlphAI financial-news REST API — relevance-scored, ticker-linked news and SEC Form 4 insider data, built for AI agents and trading bots.
- Sync and async clients (
Client/AsyncClient) overhttpx - Pydantic v2 response models — autocomplete, validation,
Decimalmoney - Cursor auto-pagination, automatic retry on 429/5xx, rate-limit inspection
- Typed errors; covers news search, feeds, symbols, Brief and Radar.
Calendar, macro and insider-trades are one
Client.request()call away.
API reference: https://api.alphai.io/api/schema/ · Developer guide: https://alphai.io/developers
pip install alphai-sdkRequires Python 3.10+. The import name is alphai.
Create an API key at https://alphai.io/account/api-keys, then pass it
explicitly or via the ALPHAI_API_KEY environment variable.
from alphai import Client
# reads $ALPHAI_API_KEY when api_key is omitted
with Client(api_key="ak_live_…") as client:
page = client.news.list(symbol="NVDA")
for article in page.results:
print(article.title, "→", article.relevance_score)Rate limits are per account and two-layer — a per-minute burst plus a per-day
volume cap: Free 20/min · 100/day · Basic 60/min · 10,000/day · Pro 150/min ·
100,000/day. News-archive depth is tiered too: Free keys page the feeds back
30 days, Basic 90, Pro 180 (paging past your horizon returns a 403 with an
upgrade hint).
from alphai import Client, NewsCategory
with Client() as client:
page = client.news.list(
symbol="NVDA",
category=[NewsCategory.EARNINGS, "insider"], # enum or str; OR-matched
min_relevance=7,
collapse_stories=True, # dedupe syndicated reprints
page_size=20, # 10 default; 1-20 on any key, 21-50 needs Pro
)
print(page.next_cursor) # opaque cursor for the next (older) page
print(page.has_more)from_date / to_date bound the feed to a publication window (inclusive), the
same names the MCP tools use. A datetime.date or a bare YYYY-MM-DD string
means the whole day, so equal bounds return that day, not an empty page; a
datetime is an exact instant (naive is read as UTC):
from datetime import date
with Client() as client:
july = client.news.list(
symbol="NVDA",
from_date=date(2026, 7, 1),
to_date=date(2026, 7, 31), # through July 31 23:59:59.999999 UTC
)The window respects your plan's archive depth (past the horizon is a 403 on
the first page) and applies to the default sort="published" mode only — delta
polling never walks back into history, so combining a window with
sort="ingested" is a 400. On news.insider the window bounds when the
filing reached the feed, not the trade date inside the insider block.
Articles reach the feed after their publish time, so a poller that tracks
time_published silently skips late arrivals. sort="ingested" orders the feed
by arrival instead, and its cursor is a polling position rather than an
end-of-feed marker:
cursor = load_cursor() # None on the first run
with Client() as client:
page = client.news.list(
sort="ingested", cursor=cursor, symbol="NVDA", page_size=20, min_relevance=7
)
for article in page.results:
handle(article) # article.original.created_at = when we received it
save_cursor(page.next_cursor) # always set; empty results = caught up
# Ask the page, never the cursor: in this mode next_cursor is never null,
# so `caught_up` (and its inverse `has_more`) is the only honest signal.
if page.caught_up:
sleep_until_next_poll()Pass the same sort on every call of a run. Each mode mints its own cursor
family, so replaying an ingested cursor into the default mode is a 400, not a
silent restart. Cursors are opaque: hand one back unchanged, never build one.
Keep up with the feed. A delta poll returns one page, so a poller that
drains slower than the feed publishes drifts backwards and its articles read as
hours old — the data is current, the position is not. Raise page_size and
narrow the stream (min_relevance, symbol, category) until one poll covers
one interval, and remember the per-day call cap bounds how much of the feed a
plan can drain at all.
On Free and Basic the archive horizon applies to where a poll resumes, so a
cursor left unused for longer than your window comes back 403
(extra.reason = "archive_horizon"). Poll on your plan's cadence and you will
not see it; Pro has no window.
iter() follows the cursor for you and flattens articles across pages:
with Client() as client:
for article in client.news.iter(category="earnings", max_items=100):
print(article.uid, article.title)with Client() as client:
client.news.trending() # top ≤10 from the last 48h
art = client.news.get("788e477c66f3849b")
client.news.related(art.uid) # up to 6 related articles
client.news.insider(symbol="NVDA") # SEC Form 4 feed (or .insider_iter())from decimal import Decimal
with Client() as client:
client.symbols.list(limit=100) # active tickers (bare list)
client.symbols.list(search="bitcoin") # name / brand / prefix lookup → BTC-USD
nvda = client.symbols.get("NVDA") # detail (404 if unknown)
btc = client.symbols.get("BTC-USD") # crypto + foreign listings too
# Multi-market: .asset_type ("Stock"/"ETF"/"Crypto"), .country, .currency,
# .supports_insider (US SEC names only). Crypto is "<SYM>-USD"; foreign uses
# the Yahoo suffix (e.g. "VOD.L").
sent = client.symbols.sentiment_summary("NVDA") # 7-day AI sentiment
ins = client.symbols.insider_summary("NVDA") # 30-day Form 4 rollup
assert isinstance(ins.buy_value_usd, Decimal | None) # money is DecimalAlphAI's own structured read of a company's earnings filings, with every figure checked against the filing text (8-K item 2.02 for US filers, a 6-K earnings release for foreign private issuers):
with Client() as client:
hist = client.symbols.earnings("NVDA")
print(hist.next_report_date) # company-confirmed (date | None; never an estimate)
for read in hist.reports: # newest first, capped at 20; empty = normal
a = read.analysis # EarningsReport | None
if a is None or not a.key_metrics:
continue # a read can publish without metrics; don't index blindly
print(read.fiscal_period, a.verdict, a.key_metrics[0].name, a.key_metrics[0].value)
latest = client.symbols.earnings_latest("NVDA") # None when no read exists yet (HTTP 204)
if latest is not None:
article = client.news.get(latest.uid) # full enrichmentEarningsRead.source_type distinguishes the filing kind (sec_form8k /
sec_form6k), and next_report_date is None whenever AlphAI holds no
confirmed date — the SDK deliberately does not substitute an estimate.
Each KeyMetric keeps value exactly as the filing printed it and adds
numeric, unit and scale next to it ("$19,345" → 19345.0, "USD",
"millions" when the filing's table header says so). scale is None when
nothing in the filing said it; don't assume millions.
from alphai import Client
with Client() as client:
brief = client.news.brief(tickers=["NVDA", "AMD", "BTC-USD"], hours=24, limit=10)
for event in [*brief.events, *brief.filings]:
print(event.matched_tickers, event.title)
for earnings in brief.upcoming_earnings:
print(earnings.ticker, earnings.report_date)
print("Unknown:", brief.unknown_tickers)
print("More coverage:", brief.events_truncated, brief.filings_truncated)One call covers 1–100 explicit tickers on every tier. This REST method does
not read your saved account watchlist. hours is a publication window (1–168,
default 24); limit caps each section independently (1–20, default 20). Events
shared by requested symbols appear once with matched_tickers. A missing
confirmed earnings date is left missing.
Brief is a ranked snapshot, with no cursor. Check both truncation flags;
narrow the window or use the paginated feed for more coverage. For a complete
incremental alert stream, use news.list(sort="ingested", ...) and persist its
cursor. Repeated Brief calls are not a lossless replacement.
from alphai import Client, ConflictError
with Client() as client:
snapshot = client.radar.snapshot(window="24h", market="us_equity", limit=10)
print(snapshot.snapshot_id, snapshot.as_of, snapshot.freshness)
for reading in snapshot.results:
print(reading.ticker, reading.news_z, reading.sent.value)
# scope="watchlist" reads the API key owner's existing saved symbols.
saved = client.radar.snapshot(scope="watchlist")
print(saved.watchlist_coverage)
try:
for reading in client.radar.iter(window="4h", max_items=100):
print(reading.ticker, reading.stories)
except ConflictError:
print("Snapshot expired or context changed. Start a new scan without a cursor.")Filters: window (4h / 24h), scope, market, exact ticker or tickers,
search, sort, order, sentiment, min_z, limit, offset, cursor.
Ticker filters do not expand aliases. limit is page size (1–100, default 50),
not a tier ticker cap; an explicitly empty ticker list is rejected.
The server delays the whole snapshot: Free 60 minutes, Basic 15, Pro no added
delay. Check as_of, access and freshness; processing adds latency. Scores and
sentiment may be None; keep them missing. Radar describes news activity and is
not a confirmed trading signal or a point-in-time backtest.
For manual pagination, pass next_cursor as cursor with the same filters and
limit. Cursors pin immutable snapshots, retained for three hours. HTTP 409 raises
ConflictError when the snapshot/context expires or changes; start a new scan
without a cursor. Iteration never silently restarts. HTTP 503 raises ServerError
after bounded retries when no eligible snapshot is available. An empty saved
watchlist returns an empty result.
Runnable examples: watchlist brief and
Radar. Both methods also work with AsyncClient; the Radar
iterator is an async generator.
Every method mirrors the sync client with await; iter() is an async generator:
import asyncio
from alphai import AsyncClient
async def main() -> None:
async with AsyncClient() as client:
async for article in client.news.iter(symbol="NVDA", max_items=20):
print(article.title)
asyncio.run(main())- alphai-news-to-email — a small, deployable app that emails you a deduplicated digest of high-relevance news for your watchlist. Built entirely on this SDK.
- alphai-earnings-week — one markdown card per week for a watchlist: confirmed next report dates, the latest filing-verified read per name, and the week's macro calendar, in 26 calls on the Free tier. Write-up with every response of the run: Earnings week from the filings, not the headlines.
All errors derive from AlphaAIError:
from alphai import Client, RateLimitError, NotFoundError, AuthenticationError
with Client() as client:
try:
client.symbols.get("ZZZZ")
except NotFoundError:
...
except RateLimitError as e:
print("retry after", e.retry_after, "seconds; limit", e.limit)
except AuthenticationError:
...| Status | Exception |
|---|---|
| 400 | BadRequestError (.fields for validation errors; .allowed_params lists the endpoint's real parameter names when you sent an unknown one) |
| 401 | AuthenticationError |
| 403 | PermissionDeniedError |
| 404 | NotFoundError |
| 409 | ConflictError (restart a new Radar scan without a cursor) |
| 429 | RateLimitError (.retry_after, .limit, .remaining, .reset) |
| 5xx | ServerError |
| network/timeout | APIConnectionError |
| 2xx, unparseable body | InvalidResponseError |
GET requests are automatically retried on 429 / 5xx / connection errors
(max_retries, default 2) with jittered backoff that honors Retry-After (capped
at max_retry_after, default 60s, so a bad value can't freeze your process). A
2xx with a non-JSON / empty body raises InvalidResponseError.
Every keyed response carries the X-RateLimit-* trio. The last one seen is on
the client:
with Client() as client:
client.news.list()
rl = client.last_rate_limit
if rl:
print(f"{rl.remaining}/{rl.limit} left, resets at {rl.reset}")Client(
api_key=None, # else $ALPHAI_API_KEY
base_url="https://api.alphai.io", # API host
timeout=30.0,
max_retries=2, # clamped to >= 0
backoff_factor=0.5,
max_retry_after=60.0, # cap on honored Retry-After (seconds)
user_agent="alphai-sdk-python/<version>",
http_client=None, # bring your own httpx.Client (advanced)
)The same keyword arguments apply to AsyncClient. When you pass a custom
http_client, the SDK still applies its Authorization header and base URL on
every request — your client just supplies the transport (proxies, custom
timeout, mounts). You own its lifecycle (the SDK won't close a client you passed in).
from datetime import datetime, timedelta, timezone
from alphai import Client
until = datetime.now(timezone.utc)
since = until - timedelta(days=29) # inside the Free archive window
with Client() as client:
page = client.news.search(query="Jane Street", from_date=since, to_date=until, page_size=20)
print(page.query.mode, page.query.note, page.matched)
for article in page.results:
print(article.title, article.enrichment.tickers)
if article.search_match:
print(article.search_match.context)
if page.next_cursor:
next_page = client.news.search(
query="Jane Street",
from_date=since,
to_date=until,
page_size=20,
cursor=page.next_cursor,
)news.search() is also available on AsyncClient with await. It returns a
NewsSearchPage, including typed query interpretation and per-article
search_match details. Quoted phrases such as query='"going concern"',
-word exclusions and OR are passed through unchanged. Add symbol,
category, source_type, item, min_relevance or collapse_stories to narrow
results. item="5.02" limits search to 8-Ks carrying that item. Dates accept ISO
strings, date or datetime, with the same semantics as the news feed.
Inspect query.mode and query.note: broadened can match only some words;
no_match and no_terms are empty answers within the searched coverage, not
proof an event did not happen. matched counts visible candidates among the best
200, not all articles in the archive. count is the current page size. Context
can be None for matches in titles/entity names; scores compare only within a
response. To continue, reuse the query, filters and window with next_cursor.
Invalid cursors raise BadRequestError; search downtime raises ServerError
(after the configured retries), never a successful empty page.
Runnable example: news search. Try the same queries in AlphAI Search.
uv venv && uv pip install -e ".[dev]"
ruff check . && ruff format --check .
mypy src/alphai
pytest # offline suite
pytest -m integration # live tests (needs ALPHAI_API_KEY)MIT — see LICENSE. API access still requires a valid key.