Python SDK

The official Python SDK gives you a single typed surface across the Forecite scored feed, the Verdict Engine, reference data, webhooks, and the realtime stream. It targets Python 3.10+.

Beta

The Python SDK is in beta while we harden the surface ahead of a stable 1.0. Method names and shapes may still change.

The REST client is synchronous — ideal for scripts, notebooks, and services. Each resource is a namespace on the client: forecite.feeds, forecite.providers, forecite.score(...), forecite.stream(...). The realtime stream is callback-based and can run inline or on a background thread.

Quickstart

Install the package

pip install forecite

Create a client

Pass your fc_live_… key (get one from the API keys page). Keep it out of source control.

import os
from forecite import Forecite

forecite = Forecite(os.environ["FORECITE_API_KEY"])

Fetch scored feeds

page = forecite.feeds.list(symbol="NVDA", actionability=True, limit=10)

for item in page["data"]:
    print(item["title"], item["scoring"]["sentiment_score"])

SDK patterns

The client uses consistent patterns for configuration, pagination, and errors across every resource.

Response models

Reads return plain dictionaries with snake_case keys matching the API. A feed item carries the source artifact plus its Verdict Engine scoring block:

{
    "id": "0b3f…",
    "title": "Q3 EPS $4.93 vs $4.59 est; FY guide raised",
    "source": "globenewswire",
    "published_at": "2026-06-30T13:31:02Z",
    "scoring": {
        "actionability": True,
        "sentiment_score": 8,          # 0–10 (5 = neutral)
        "sentiment_comment": "…",
    },
    "symbols": [{"symbol": "NVDA", "exchange": "NASDAQ"}],
}

Environment configuration

Production is the default. Pass base_url / ws_url to target local dev or a custom deployment; timeout sets the per-request timeout in seconds.

forecite = Forecite(
    os.environ["FORECITE_API_KEY"],
    base_url="http://localhost:3000/api",
    ws_url="ws://localhost:8080",
    timeout=30.0,
)

Pagination

List endpoints return {"data": [...], "next_cursor": ...}. Pass next_cursor back as cursor to page; a None cursor means you've reached the end.

cursor = None
while True:
    page = forecite.feeds.list(limit=100, cursor=cursor)
    for item in page["data"]:
        handle(item)
    cursor = page["next_cursor"]
    if not cursor:
        break

Error handling

Every failed request raises a ForeciteError carrying the HTTP status and the API's code / message.

from forecite import Forecite, ForeciteError

try:
    page = forecite.feeds.list(limit=10)
except ForeciteError as err:
    if err.status == 429:
        ...  # Rate limited — retry with backoff.
    elif err.status == 401:
        ...  # Bad or missing API key.
    else:
        raise

Feed data

Read the scored feed — list with filters, or fetch a single item by id.

# Filtered list — actionable NVDA items, most bullish first
page = forecite.feeds.list(symbol="NVDA", actionability=True, sentiment_min=7)

# One item with full per-symbol detail
item = forecite.feeds.get("0b3f2c1e-…")

Available filters: since / until (ISO 8601), symbol, exchange, aggregator, actionability, sentiment_min / sentiment_max, limit, cursor.

Discovery

Browse the reference data that powers the feed's filters.

providers = forecite.providers.list()
one = forecite.providers.get(providers[0]["id"])

sources = forecite.sources.list()  # list[str]
activity = forecite.sources.get("globenewswire")

symbols = forecite.symbols.list(limit=500)
nvda = forecite.symbols.get("NVDA")

by_dimension = forecite.tags.list()
corp_activity = forecite.tags.get("corp_activity")

Scoring

Run the Verdict Engine on your own text. Pass an artifact dict and optional scoring options.

verdict = forecite.score(
    {
        "type": "filing",
        "title": "Acme cuts FY guide on softening demand",
        "body": "Acme Corp lowered its full-year revenue outlook…",
        "related_symbols": ["ACME"],
    },
    options={"model": "pro", "score_sentiment": "if_actionable"},
)

print(verdict["actionability"]["score"])

Realtime streams

Subscribe to the scored feed over WebSocket with callbacks. Run it inline with run() (blocking) or on a background thread with run_in_thread().

def on_feed(item):
    print(item["title"], item["scoring"]["sentiment_score"])

stream = forecite.stream(
    filters={"symbols": ["NVDA", "TSLA"], "actionable": True},
    snapshot=10,
    on_feed=on_feed,
    on_welcome=lambda info: print("connected:", info["tier"]),
    on_quota_exceeded=lambda info: print(info["message"]),
)

stream.run()  # blocks; or: thread = stream.run_in_thread()
# stream.close() to stop

Account

Inspect the authenticated key and its usage.

key = forecite.me()             # tier, limits
usage = forecite.usage(days=14) # daily feed vs. score usage

Webhooks

Manage outbound webhook endpoints.

hook = forecite.webhooks.create(
    url="https://example.com/forecite",
    description="prod",
)

forecite.webhooks.list()
forecite.webhooks.update(hook["id"], enabled=False)
forecite.webhooks.delete(hook["id"])

Changelog

`0.1.2`

First public beta of the forecite package.
Synchronous clients for feeds, scoring, discovery, account, webhooks.
Callback-based realtime feed stream with snapshot replay and threaded run.

On this page