Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.vectrade.io/llms.txt

Use this file to discover all available pages before exploring further.

​
Streaming

VecTrade uses Server-Sent Events (SSE) for real-time streaming of AI analysis responses.

​
AI Streaming

from vectrade import VecTrade

client = VecTrade()

# Stream AI analysis
for chunk in client.ai.stream("What's the bull case for NVDA?"):
    print(chunk.text, end="", flush=True)

print()  # newline after stream ends

​
SSE Event Format

The stream sends events in standard SSE format: 
data: {"content": "NVIDIA's", "type": "text"}

data: {"content": " bull case", "type": "text"}

data: {"content": " rests on", "type": "text"}

data: {"type": "citation", "source": "Q4 Earnings", "url": "..."}

data: {"type": "done", "usage": {"prompt_tokens": 150, "completion_tokens": 420}}

​
Stream Event Types

TypeDescription
textContent chunk to display
citationSource reference
metadataModel info, timing
doneStream complete with usage stats

​
Async Streaming (Python)

import asyncio
from vectrade import AsyncVecTrade

async def main():
    client = AsyncVecTrade()

    async for chunk in client.ai.stream("Analyze TSLA technicals"):
        print(chunk.text, end="", flush=True)

    await client.close()

asyncio.run(main())

​
Edge/Worker Streaming (TypeScript)

For Vercel Edge Functions or Cloudflare Workers, convert to a ReadableStream: 
import { VecTrade, toReadableStream } from '@vectrade/sdk';

export const runtime = 'edge';

export async function POST(req: Request) {
  const { prompt } = await req.json();
  const client = new VecTrade();

  const stream = client.ai.stream(prompt);
  const readable = toReadableStream(stream);

  return new Response(readable, {
    headers: { 'Content-Type': 'text/event-stream' },
  });
}

​
Error Handling in Streams

Errors during streaming are delivered as error events before the connection closes: 
from vectrade import VecTrade, RateLimitError

try:
    for chunk in client.ai.stream("..."):
        print(chunk.text, end="")
except RateLimitError as e:
    print(f"\nRate limited: retry after {e.retry_after}s")

​
Connection Management

  • Streams auto-close when the [DONE] event is received
  • Use context managers for guaranteed cleanup:
with VecTrade() as client:
    for chunk in client.ai.stream("..."):
        print(chunk.text, end="")
# Connection is closed automatically