Skip to main content

​
Streaming

VecTrade uses Server-Sent Events (SSE) for real-time streaming of AI analysis responses.
SDK AI Streaming is coming soon. Direct client.ai.stream() is not yet available in the public SDK. AI streaming is currently available through the VTrade Copilot web interface for Standard and Professional plan users.

​
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