What is 3kk0?

3kk0 (formerly Bond) is a real-time market data streaming platform designed for indie quants and AI agents. It aggregates data from multiple sources: crypto exchanges (via CCXT), social media (Twitter/X, Nitter), on-chain data (Solana), and custom WebSocket sources into unified, time-aligned WebSocket feeds. Instead of managing multiple APIs and normalizing different schemas, you describe what you want and 3kk0 delivers clean, aggregated JSON events.

"BTC price + tweets every 5s"

Key Features

• Natural Language Interface: Create streams using plain English powered by DeepSeek AI
• Unified Schema: All data sources normalized to consistent JSON format with timestamp alignment
• Multiple Sources: CCXT (600+ exchanges), Twitter/X, Solana on-chain, Google Trends, custom WebSockets
• Real-time Streaming: WebSocket delivery with optional replay buffer (last 60 seconds)
• Automatic Failover: Exchange failover and source-level retry with exponential backoff
• Python SDK: Async-first client library with convenience functions
• HMAC Authentication: Secure token-based access with time-limited expiration

Architecture

3kk0 uses a three-tier architecture: a Control Plane (FastAPI REST API for stream management), a Data Plane (multi-source pipeline with Redis Streams for event buffering), and a WebSocket Broker (fan-out to connected clients). Each stream runs independently with its own aggregation pipeline, ensuring isolation and reliability.

Common Use Cases

Installation

Install the 3kk0 Python SDK from the repository:

# Clone the repository
git clone https://github.com/yourusername/3kk0.git
cd 3kk0

# Install SDK in development mode
pip install -e sdk/python

Set Up Your Environment

Point the SDK to your 3kk0 API instance:

from ekko import EkkoClient

# Local development
client = EkkoClient(
    api_url="http://localhost:8000",
    ws_url="ws://localhost:8080"
)

# Or production
client = EkkoClient(
    api_url="https://api.3kk0.com",
    ws_url="wss://streams.3kk0.com"
)

Create Your First Stream

Use natural language to create a data stream:

import asyncio
from ekko import EkkoClient

async def main():
    client = EkkoClient()

    # Create stream with natural language
    result = await client.create_stream("BTC price + tweets every 5s")

    stream_id = result["stream_id"]
    token = result["access_token"]

    print(f"Stream created: {stream_id}")

    # Listen to events
    async for event in client.subscribe(stream_id, token):
        print(f"Price: ${event['price_avg']:.2f}, Tweets: {event['tweets']}")

asyncio.run(main())

Or Use the Convenience Function

For quick prototyping, use the listen() function:

from ekko import listen

async for event in listen("BTC price every 5s"):
    print(f"BTC: ${event['price_avg']:,.2f}")

Event Schema

Every event returns a unified, aggregated structure:

{
  "ts": "2025-10-21T13:30:05Z",
  "window_start": "2025-10-21T13:30:00Z",
  "window_end": "2025-10-21T13:30:05Z",

  "price_avg": 68123.5,
  "price_high": 68456.2,
  "price_low": 67890.1,
  "bid_avg": 68122.1,
  "ask_avg": 68125.0,
  "volume_sum": 134.2,

  "tweets": 8,
  "onchain_count": 3,
  "custom_count": 0,

  "raw_data": { /* detailed source data */ }
}

System Requirements

Python SDK Installation

Install the SDK from the source repository:

# Clone the repository
git clone https://github.com/yourusername/3kk0.git
cd 3kk0

# Install SDK in editable mode
pip install -e sdk/python

# Or install all development dependencies
pip install -r requirements.txt

Docker Setup (Recommended)

The easiest way to run 3kk0 locally is with Docker Compose:

# Navigate to infra directory
cd infra

# Start all services (API, Broker, Redis)
docker compose up -d

# Services will be available at:
# - API: http://localhost:8000
# - Broker: ws://localhost:8080
# - Redis: localhost:6379

Environment Configuration

Create a .env file in the project root:

# Redis connection
REDIS_URL=redis://localhost:6379

# WebSocket host for client URLs
WS_HOST=localhost:8080
PUBLIC_WS_BASE=ws://localhost:8080/ws

# HMAC secret (CHANGE IN PRODUCTION!)
BOND_SECRET=bond-dev-secret-change-in-production

# DeepSeek API key (for natural language parsing)
DEEPSEEK_API_KEY=sk-your-key-here

# Optional: Token TTL (default 3600 seconds)
BOND_TOKEN_TTL=3600

# Optional: Stream limits (default 5)
BOND_MAX_STREAMS=5

Verify Installation

Test that everything is working:

import asyncio
from ekko import EkkoClient

async def test():
    client = EkkoClient(api_url="http://localhost:8000")

    # Test API health
    health = await client.health()
    print(f"API Status: {health}")

    # List streams
    streams = await client.list_streams()
    print(f"Active streams: {len(streams)}")

asyncio.run(test())

Production Deployment

For production, ensure you:

• Change BOND_SECRET to a strong random value
• Use a managed Redis instance with authentication
• Set DEEPSEEK_API_KEY for NL parsing
• Configure proper CORS and rate limiting
• Use HTTPS/WSS endpoints

How Authentication Works

When you create a stream, 3kk0 generates two types of tokens:

• Short-lived token: Valid for 1 hour (default), used for WebSocket connections
• Persistent token: Valid for ~10 years, stored in Redis for long-term access

Token Format

Tokens use HMAC-SHA256 with time-based expiration:

# Token format: "{expiry_timestamp}.{signature}"
# Example:
"1698765432.f1189e3ca7ce6df8"
               ↑           ↑
          Unix timestamp  HMAC signature (16 chars)

Authentication Flow

Using Tokens

Include the token in your WebSocket connection:

# WebSocket URL includes token
ws_url = f"ws://localhost:8080/ws/{stream_id}?token={access_token}"

# Connect with SDK (handles automatically)
async for event in client.subscribe(stream_id, token):
    print(event)

Token Refresh

Before a token expires, request a new one:

# Refresh token via API
response = await client.refresh_token(stream_id)

new_token = response["token"]
expires_in = response["expires_in_sec"]  # 3600

# Reconnect with new token
async for event in client.subscribe(stream_id, new_token):
    print(event)

Token Verification

The broker verifies tokens on every WebSocket connection:

# 1. Parse token: "expiry.signature"
expiry, signature = token.split(".")

# 2. Check expiration
if time.time() > int(expiry):
    return "Token expired"

# 3. Verify HMAC signature (constant-time comparison)
expected = hmac.new(secret, f"{stream_id}.{expiry}", sha256).hexdigest()[:16]
if not hmac.compare_digest(signature, expected):
    return "Invalid signature"

Error Handling

WebSocket closes with specific codes for auth errors:

# Generate a strong secret
openssl rand -hex 32

# Set in environment
export BOND_SECRET=your-strong-random-secret-here

Token Lifespan

Configure token expiration via environment variables:

# Default: 1 hour (3600 seconds)
BOND_TOKEN_TTL=3600

# For long-running bots: 24 hours
BOND_TOKEN_TTL=86400

# For testing: 5 minutes
BOND_TOKEN_TTL=300

Paste your streams documentation here...

Paste your data sources documentation here...

Paste your filters documentation here...

Paste your webhooks documentation here...

Paste your Python SDK reference here...

Paste your WebSocket API documentation here...

Paste your REST API documentation here...

Paste your streaming API documentation here...

Paste your trading bot guide here...

Paste your dashboard guide here...

Paste your AI agent integration guide here...

Paste your error handling guide here...

Paste your examples here...

Paste your rate limits documentation here...

Paste your changelog here...

Paste your support information here...