kronical 0.1.2

Python SDK for the Kronical API — real-time causal intelligence

Kronical Python SDK

Python client library for the Kronical API.

Installation

pip install kronical

# With WebSocket support
pip install kronical[websocket]

Quick Start

from kronical import KronicalClient

client = KronicalClient(
    api_key="kron_hf_...",
    base_url="https://api.kronical.ai",
)

# Health check
print(client.health())

# Get latest headlines
feed = client.feed.get(limit=10)
for h in feed["headlines"]:
    print(f"{h['text']} ({h['context']['novelty']})")

# Natural language search
results = client.feed.get(query="semiconductor export controls", days_back=30)
print(f"Found {results['count']} headlines")

# Enrich a news item
enriched = client.enrichment.submit("Fed raises rates by 25bps to 5.75%")
print(enriched["enrichment"]["context_narrative"])

# Batch enrichment
batch = client.enrichment.batch([
    "Fed raises rates by 25bps",
    "ECB holds rates steady",
])
print(f"Batch {batch['batch_id']} submitted")

# Poll for results
results = client.enrichment.get_batch(batch["batch_id"])
print(f"Status: {results['batch']['status']}")

# Pulse
columns = client.pulse.list()
client.pulse.create(name="Fed Policy", query="Federal Reserve monetary policy")
client.pulse.update(columns["columns"][0]["column_id"], enabled=False)

# Smart agents
columns = client.smart_columns.list()
new_col = client.smart_columns.create(
    name="China Semi",
    instruction="Any news about semiconductor export restrictions involving China",
)
matches = client.smart_columns.matches(hours=24)

# Watchlist
states = client.watchlist.ticker_states(["AAPL", "NVDA", "TSLA"])
impacts = client.watchlist.impacts("NVDA", hours=48)
sparkline = client.watchlist.sparkline("AAPL")

# Usage stats
usage = client.usage.get(days=30)
print(f"Total requests: {usage['summary']['total_requests']}")

Async Client

import asyncio
from kronical import AsyncKronicalClient

async def main():
    async with AsyncKronicalClient(
        api_key="kron_hf_...",
        base_url="https://api.kronical.ai",
    ) as client:
        feed = await client.feed.get(limit=10)
        print(f"Got {feed['count']} headlines")

asyncio.run(main())

WebSocket Streams

Requires pip install kronical[websocket].

import asyncio
from kronical.websocket import KronicalWebSocket

async def main():
    ws = KronicalWebSocket(
        api_key="kron_hf_...",
        base_url="wss://api.kronical.ai",
    )

    # Live headline feed (runs forever)
    # The connected message includes pulse_columns list.
    # Each headline message includes matched_columns list.
    async def on_headline(msg):
        if msg.get("type") == "connected":
            print(f"Columns: {msg.get('pulse_columns')}")
            return
        print(f"[{msg.get('novelty')}] {msg.get('text')} (columns: {msg.get('matched_columns')})")

    await ws.pulse(on_message=on_headline)

asyncio.run(main())

Enrichment Feed (Bidirectional)

async def enrich_stream():
    ws = KronicalWebSocket(
        api_key="kron_hf_...",
        base_url="wss://api.kronical.ai",
    )
    await ws.connect_enrichment()

    result = await ws.enrich("Fed raises rates by 25bps to 5.75%")
    print(result)

    await ws.close()

Error Handling

from kronical import KronicalClient
from kronical.exceptions import RateLimitError, AuthenticationError, NotFoundError

client = KronicalClient(api_key="kron_hf_...", base_url="https://api.kronical.ai")

try:
    results = client.feed.get(query="oil prices", days_back=30)
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after}s")
except AuthenticationError:
    print("Invalid API key")
except NotFoundError:
    print("Resource not found")

The SDK automatically retries on 429 (rate limit) responses with exponential backoff.

Configuration

Parameter	Default	Description
api_key	required	Your Kronical API key (kron_hf_...)
base_url	required	API base URL (https://api.kronical.ai)
timeout	30.0	Request timeout in seconds
max_retries	3	Max retries on rate limit (429)