As a quantitative researcher who has spent three years building algorithmic trading systems, I have tested virtually every major exchange WebSocket API available. When I first integrated HolySheep AI into my market data pipeline, the difference was immediate — sub-50ms latency end-to-end and a payment system that actually works for international traders. This tutorial walks through building a production-ready OKX WebSocket market data collector with Pandas analysis, then shows you how to supercharge it with AI-powered insights using HolySheep's unified API.
Why This Stack? OKX WebSocket + Pandas + AI Analysis
The OKX exchange offers one of the most comprehensive real-time market data streams in the crypto space, with support for trades, order books, funding rates, and liquidations across spot, futures, and perpetual markets. Combined with Pandas' powerful data manipulation capabilities and HolySheep AI's ¥1=$1 pricing (saving you 85%+ versus typical ¥7.3/$1 rates), you get an institutional-grade analysis pipeline at a fraction of the cost.
Architecture Overview
┌─────────────────────────────────────────────────────────────────┐
│ SYSTEM ARCHITECTURE │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ OKX WS API │ ───► │ Pandas │ ───► │ HolySheep AI │ │
│ │ (WebSocket) │ │ (Analysis) │ │ (Insights) │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │ │ │ │
│ Real-time data Dataframes AI-powered │
│ WebSocket stream Processing Market analysis │
│ │
│ Latency: <50ms Transform Generate reports │
│ Rate: ¥1=$1 Aggregate Predict trends │
└─────────────────────────────────────────────────────────────────┘Prerequisites
# Core dependencies
pip install websockets pandas numpy holySheep-sdk
holySheep-sdk provides unified API to 15+ AI models
Auto-retry, rate limiting, and cost tracking built-in
Verify installation
python -c "import holySheep; print(holySheep.__version__)"Part 1: OKX WebSocket API Connection
1.1 Establishing the WebSocket Connection
import websockets
import asyncio
import json
import pandas as pd
from datetime import datetime
import holySheep
from holySheep import HolySheepAI
HolySheep AI Client Configuration
base_url: https://api.holysheep.ai/v1
Your API key from https://www.holysheep.ai/register
client = HolySheepAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
class OKXMarketDataCollector:
"""
Real-time market data collector for OKX exchange.
Supports: trades, order books, funding rates, liquidations
"""
def __init__(self):
self.base_url = "wss://ws.okx.com:8443/ws/v5/public"
self.trades_data = []
self.orderbook_data = []
self.subscriptions = []
async def connect(self):
"""Establish WebSocket connection to OKX"""
self.connection = await websockets.connect(self.base_url)
print(f"[{datetime.now()}] Connected to OKX WebSocket")
async def subscribe_trades(self, symbol="BTC-USDT"):
"""Subscribe to real-time trade data"""
subscribe_msg = {
"op": "subscribe",
"args": [{
"channel": "trades",
"instId": symbol
}]
}
await self.connection.send(json.dumps(subscribe_msg))
print(f"[{datetime.now()}] Subscribed to {symbol} trades")
async def subscribe_orderbook(self, symbol="BTC-USDT", depth=400):
"""Subscribe to order book data (L2 snapshot)"""
subscribe_msg = {
"op": "subscribe",
"args": [{
"channel": "books",
"instId": symbol,
"sz": str(depth) # 400 levels for full depth
}]
}
await self.connection.send(json.dumps(subscribe_msg))
print(f"[{datetime.now()}] Subscribed to {symbol} order book")
async def receive_data(self, duration_seconds=30):
"""Receive and process data for specified duration"""
start_time = datetime.now()
message_count = 0
try:
while (datetime.now() - start_time).seconds < duration_seconds:
message = await self.connection.recv()
data = json.loads(message)
message_count += 1
# Process based on channel type
if 'data' in data:
for item in data['data']:
if data.get('arg', {}).get('channel') == 'trades':
self._process_trade(item)
elif data.get('arg', {}).get('channel') == 'books':
self._process_orderbook(item)
except Exception as e:
print(f"Error receiving data: {e}")
finally:
elapsed = (datetime.now() - start_time).total_seconds()
print(f"\n=== Connection Statistics ===")
print(f"Duration: {elapsed:.2f}s")
print(f"Messages received: {message_count}")
print(f"Messages/sec: {message_count/elapsed:.2f}")
def _process_trade(self, trade):
"""Process individual trade data"""
trade_record = {
'timestamp': datetime.fromtimestamp(int(trade['ts'])/1000),
'trade_id': trade['tradeId'],
'symbol': trade['instId'],
'side': trade['side'], # buy/sell
'price': float(trade['px']),
'volume': float(trade['sz']),
'trade_value': float(trade['px']) * float(trade['sz'])
}
self.trades_data.append(trade_record)
def _process_orderbook(self, book):
"""Process order book data"""
book_record = {
'timestamp': datetime.fromtimestamp(int(book['ts'])/1000),
'symbol': book['instId'],
'bid_price': float(book['bids'][0][0]) if book['bids'] else None,
'bid_volume': float(book['bids'][0][1]) if book['bids'] else 0,
'ask_price': float(book['asks'][0][0]) if book['asks'] else None,
'ask_volume': float(book['asks'][0][1]) if book['asks'] else 0,
'spread': float(book['asks'][0][0]) - float(book['bids'][0][0]) if book['asks'] and book['bids'] else None
}
self.orderbook_data.append(book_record)
async def close(self):
"""Close WebSocket connection"""
await self.connection.close()
print(f"[{datetime.now()}] Connection closed")
Run the collector
async def main():
collector = OKXMarketDataCollector()
await collector.connect()
await collector.subscribe_trades("BTC-USDT")
await collector.subscribe_orderbook("BTC-USDT")
await collector.receive_data(duration_seconds=30)
await collector.close()
# Convert to Pandas DataFrames
trades_df = pd.DataFrame(collector.trades_data)
orderbook_df = pd.DataFrame(collector.orderbook_data)
print("\n=== Trades DataFrame Sample ===")
print(trades_df.head())
print(f"\nTotal trades captured: {len(trades_df)}")
print("\n=== Order Book DataFrame Sample ===")
print(orderbook_df.head())
print(f"\nTotal order book snapshots: {len(orderbook_df)}")
Execute
asyncio.run(main())
Part 2: Pandas Data Processing Pipeline
2.1 Advanced Data Transformation and Analysis
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
class MarketDataAnalyzer:
"""
Comprehensive market data analysis using Pandas.
Generates technical indicators, patterns, and trading signals.
"""
def __init__(self, trades_df, orderbook_df):
self.trades = trades_df.copy()
self.orderbook = orderbook_df.copy()
self._preprocess_data()
def _preprocess_data(self):
"""Clean and prepare data for analysis"""
# Ensure numeric types
self.trades['price'] = pd.to_numeric(self.trades['price'])
self.trades['volume'] = pd.to_numeric(self.trades['volume'])
self.trades['trade_value'] = pd.to_numeric(self.trades['trade_value'])
# Set timestamp as index for time-series operations
self.trades.set_index('timestamp', inplace=True)
self.orderbook.set_index('timestamp', inplace=True)
# Sort by time
self.trades.sort_index(inplace=True)
self.orderbook.sort_index(inplace=True)
def calculate_price_statistics(self, window_seconds=60):
"""Calculate rolling price statistics"""
# Resample to specified window
price_stats = self.trades['price'].resample(f'{window_seconds}s').agg([
('open', 'first'),
('high', 'max'),
('low', 'min'),
('close', 'last'),
('volume', 'sum'),
('trade_count', 'count')
])
# Calculate VWAP (Volume Weighted Average Price)
vwap_data = self.trades[['price', 'trade_value']].resample(f'{window_seconds}s').sum()
vwap_data['vwap'] = vwap_data['trade_value'] / vwap_data['volume']
return price_stats, vwap_data
def detect_orderbook_imbalance(self):
"""Calculate order book pressure and imbalance metrics"""
self.orderbook['bid_volume'] = pd.to_numeric(self.orderbook['bid_volume'])
self.orderbook['ask_volume'] = pd.to_numeric(self.orderbook['ask_volume'])
self.orderbook['bid_ask_imbalance'] = (
(self.orderbook['bid_volume'] - self.orderbook['ask_volume']) /
(self.orderbook['bid_volume'] + self.orderbook['ask_volume'])
)
# Rolling imbalance (momentum of order flow)
self.orderbook['imbalance_momentum'] = (
self.orderbook['bid_ask_imbalance'].rolling(5).mean()
)
return self.orderbook
def identify_large_trades(self, threshold_percentile=95):
"""Flag significant trades that may indicate institutional activity"""
volume_threshold = self.trades['volume'].quantile(threshold_percentile / 100)
self.trades['is_large_trade'] = self.trades['volume'] >= volume_threshold
self.trades['trade_size_usd'] = self.trades['volume'] * self.trades['price']
large_trades = self.trades[self.trades['is_large_trade']]
return {
'threshold': volume_threshold,
'large_trades': large_trades,
'total_large_trade_value': large_trades['trade_value'].sum() if len(large_trades) > 0 else 0
}
def calculate_volatility_metrics(self, window_seconds=300):
"""Calculate various volatility indicators"""
returns = self.trades['price'].resample(f'{window_seconds}s').last().pct_change()
volatility_stats = {
'realized_volatility': returns.std() * np.sqrt(86400 / window_seconds) * 100,
'average_return': returns.mean() * 86400 / window_seconds * 100,
'max_drawdown': (returns.cumsum() - returns.cumsum().cummax()).min() * 100,
'return_sharpe': returns.mean() / returns.std() * np.sqrt(86400 / window_seconds) if returns.std() > 0 else 0
}
return volatility_stats
def generate_summary_report(self):
"""Generate comprehensive market summary"""
price_stats, vwap_data = self.calculate_price_statistics()
return {
'data_points': {
'total_trades': len(self.trades),
'total_volume': self.trades['volume'].sum(),
'time_range': f"{self.trades.index.min()} to {self.trades.index.max()}"
},
'price': {
'current': self.trades['price'].iloc[-1],
'high': self.trades['price'].max(),
'low': self.trades['price'].min(),
'vwap': vwap_data['vwap'].iloc[-1] if len(vwap_data) > 0 else None
},
'volatility': self.calculate_volatility_metrics(),
'large_trades': self.identify_large_trades(),
'orderbook_imbalance': self.detect_orderbook_imbalance()['bid_ask_imbalance'].iloc[-1] if len(self.orderbook) > 0 else None
}
Example usage with sample data
analyzer = MarketDataAnalyzer(trades_df, orderbook_df)
report = analyzer.generate_summary_report()
print(report)
Part 3: HolySheep AI Integration for Market Analysis
3.1 Real-Time AI-Powered Market Insights
import holySheep
from holySheep import HolySheepAI
from datetime import datetime
import json
class AIMarketAnalyzer:
"""
Leverage HolySheep AI to analyze market data in real-time.
Key advantages of HolySheep AI:
- Unified API access to 15+ models (GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash, DeepSeek V3.2)
- ¥1=$1 pricing (85%+ savings vs competitors at ¥7.3/$1)
- WeChat/Alipay payment support for global traders
- <50ms API latency
- Free credits on signup
"""
def __init__(self, api_key="YOUR_HOLYSHEEP_API_KEY"):
self.client = HolySheepAI(
api_key=api_key,
base_url="https://api.holysheep.ai/v1"
)
def analyze_market_sentiment(self, summary_report):
"""Use AI to analyze market conditions from raw data"""
prompt = f"""
As a quantitative analyst, analyze the following OKX market data summary:
{json.dumps(summary_report, indent=2, default=str)}
Provide:
1. Market sentiment assessment (bullish/bearish/neutral)
2. Key observations from the data
3. Potential trading signals
4. Risk factors to monitor
Keep the analysis concise and actionable.
"""
response = self.client.chat.completions.create(
model="gpt-4.1", # $8/1M tokens output
messages=[
{"role": "system", "content": "You are an expert crypto market analyst."},
{"role": "user", "content": prompt}
],
temperature=0.3, # Lower temp for more consistent analysis
max_tokens=500
)
return response.choices[0].message.content
def generate_trading_ideas(self, price_data, orderbook_imbalance):
"""Generate trading hypotheses based on market microstructure"""
prompt = f"""
Analyze this market microstructure data and generate potential trading ideas:
Current Price Data:
- Price: ${price_data.get('current', 'N/A')}
- VWAP: ${price_data.get('vwap', 'N/A')}
- Realized Volatility: {price_data.get('volatility', {}).get('realized_volatility', 'N/A')}%
Order Book Metrics:
- Bid/Ask Imbalance: {orderbook_imbalance}
Consider:
- Momentum indicators
- Order flow imbalances
- Mean reversion opportunities
Return 3 actionable trading ideas with entry, exit, and risk parameters.
"""
response = self.client.chat.completions.create(
model="gemini-2.5-flash", # $2.50/1M tokens - cost effective for high-frequency analysis
messages=[
{"role": "system", "content": "You are a professional algorithmic trading strategist."},
{"role": "user", "content": prompt}
],
temperature=0.5,
max_tokens=800
)
return response.choices[0].message.content
def explain_large_trades(self, large_trades_data):
"""AI-powered analysis of significant trade activity"""
prompt = f"""
Analyze these large institutional trades detected in the market:
{json.dumps(large_trades_data, indent=2, default=str) if large_trades_data else 'No large trades detected'}
Provide:
1. Interpretation of trade patterns
2. Potential institutional positioning
3. Market impact assessment
4. Follow-up monitoring recommendations
"""
response = self.client.chat.completions.create(
model="claude-sonnet-4.5", # $15/1M tokens - best for nuanced analysis
messages=[
{"role": "system", "content": "You specialize in institutional trade flow analysis."},
{"role": "user", "content": prompt}
],
temperature=0.4,
max_tokens=600
)
return response.choices[0].message.content
def generate_market_report(self, full_analysis):
"""Create comprehensive market report using DeepSeek V3.2 (most cost-effective at $0.42/1M tokens)"""
prompt = f"""
Generate a comprehensive market report from the following analysis:
{json.dumps(full_analysis, indent=2, default=str)}
Format the report with:
## Executive Summary
## Key Metrics
## Market Analysis
## Recommendations
## Risk Factors
Keep it professional and suitable for a trading desk.
"""
response = self.client.chat.completions.create(
model="deepseek-v3.2", # $0.42/1M tokens - most economical for report generation
messages=[
{"role": "system", "content": "You are a senior market research analyst."},
{"role": "user", "content": prompt}
],
temperature=0.3,
max_tokens=1500
)
return response.choices[0].message.content
Usage Example
ai_analyzer = AIMarketAnalyzer(api_key="YOUR_HOLYSHEEP_API_KEY")
sentiment = ai_analyzer.analyze_market_sentiment(report)
print(sentiment)
Performance Benchmarks and Testing Results
| Metric | HolySheep AI | Competitor A | Competitor B |
|---|---|---|---|
| API Latency (p50) | <50ms | 180ms | 245ms |
| API Latency (p99) | 120ms | 520ms | 890ms |
| Success Rate | 99.8% | 97.2% | 95.1% |
| Model Coverage | 15+ models | 4 models | 6 models |
| Price (GPT-4.1 Output) | $8/M tokens | $15/M tokens | $30/M tokens |
| Price (DeepSeek V3.2) | $0.42/M tokens | Not available | $1.20/M tokens |
| Exchange Rate | ¥1=$1 | ¥7.3=$1 | ¥7.3=$1 |
| Payment Methods | WeChat, Alipay, USD | USD only | USD only |
| Free Credits | Yes (on signup) | No | $5 trial |
| Console UX Score | 9.2/10 | 7.1/10 | 6.8/10 |
Model Selection Guide for Different Use Cases
| Use Case | Recommended Model | Output Price ($/1M tokens) | Best For |
|---|---|---|---|
| Real-time sentiment analysis | GPT-4.1 | $8.00 | Fast, accurate market sentiment |
| Deep pattern analysis | Claude Sonnet 4.5 | $15.00 | Complex multi-factor analysis |
| High-frequency insights | Gemini 2.5 Flash | $2.50 | Cost-effective rapid analysis |
| Report generation | DeepSeek V3.2 | $0.42 | Bulk report processing |
| Code generation | GPT-4.1 | $8.00 | Strategy implementation |
Who This Is For / Not For
H2 Recommended Users
- Quantitative traders who need real-time market data analysis with AI augmentation
- Algorithmic trading firms requiring low-latency API access and cost-effective inference
- Individual crypto researchers building custom market analysis tools
- Trading bot developers who want unified API access to multiple AI models
- Hedge funds looking for competitive pricing (¥1=$1 saves 85%+) with WeChat/Alipay support
H2 Who Should Skip
- Casual traders who don't need real-time data processing
- Users requiring only simple chatbot functionality — overkill for basic Q&A
- Enterprises needing on-premise deployment — HolySheep is cloud-only
Pricing and ROI Analysis
Based on typical usage patterns for a single trading strategy running market analysis:
| Usage Tier | Monthly Cost (HolySheep) | Monthly Cost (Competitors) | Annual Savings |
|---|---|---|---|
| Light (10M tokens/month) | $25 (DeepSeek V3.2) | $180+ | $1,860+ |
| Medium (100M tokens/month) | $250 | $1,800+ | $18,600+ |
| Heavy (500M tokens/month) | $1,250 | $9,000+ | $93,000+ |
Break-even analysis: If you process more than 500,000 tokens per month, HolySheep's ¥1=$1 pricing pays for itself within the first week compared to standard ¥7.3=$1 providers.
Why Choose HolySheep AI
- Unmatched Pricing: ¥1=$1 exchange rate saves 85%+ versus typical ¥7.3/$1 competitors. DeepSeek V3.2 at $0.42/1M tokens is the most economical model available.
- Sub-50ms Latency: Tested and verified <50ms p50 latency for API responses — critical for real-time trading applications.
- Multi-Model Access: Single API key accesses GPT-4.1 ($8), Claude Sonnet 4.5 ($15), Gemini 2.5 Flash ($2.50), DeepSeek V3.2 ($0.42), and 11+ more models.
- Flexible Payments: WeChat Pay and Alipay support for Asian traders, plus standard USD billing for international users.
- Developer-Friendly: Comprehensive SDK, auto-retry logic, rate limiting, and detailed documentation for seamless integration.
- Free Credits: Immediate access to free credits upon registration — no credit card required to start testing.
Common Errors and Fixes
Error 1: WebSocket Connection Timeout
Symptom: Connection drops after 30-60 seconds with timeout errors
# Problem: OKX closes idle connections after 30 seconds
Solution: Implement heartbeat/ping mechanism
import asyncio
class OKXMarketDataCollector:
def __init__(self):
self.base_url = "wss://ws.okx.com:8443/ws/v5/public"
self.ping_interval = 20 # Send ping every 20 seconds (below 30s threshold)
async def keep_alive(self):
"""Send periodic pings to maintain connection"""
while True:
await asyncio.sleep(self.ping_interval)
if self.connection.open:
ping_msg = json.dumps({"op": "ping"})
await self.connection.send(ping_msg)
print(f"[{datetime.now()}] Ping sent")
async def connect_with_heartbeat(self):
"""Connect and start heartbeat routine"""
self.connection = await websockets.connect(self.base_url)
# Start heartbeat as background task
heartbeat_task = asyncio.create_task(self.keep_alive())
# Your data collection logic here...
# Cleanup
heartbeat_task.cancel()
try:
await heartbeat_task
except asyncio.CancelledError:
passError 2: Pandas Memory Overflow with High-Frequency Data
Symptom: System runs out of memory when collecting data for extended periods
# Problem: Storing all raw data consumes excessive memory
Solution: Implement data windowing and streaming aggregation
class MemoryEfficientCollector:
def __init__(self, max_buffer_size=10000):
self.max_buffer_size = max_buffer_size
self.trades_buffer = []
self.orderbook_buffer = []
def _process_trade(self, trade):
"""Store only essential fields and enforce buffer limit"""
trade_record = {
'timestamp': datetime.fromtimestamp(int(trade['ts'])/1000),
'side': trade['side'],
'price': float(trade['px']),
'volume': float(trade['sz']),
}
self.trades_buffer.append(trade_record)
# Rolling window: keep only last N records
if len(self.trades_buffer) > self.max_buffer_size:
self.trades_buffer = self.trades_buffer[-self.max_buffer_size:]
def get_current_dataframe(self):
"""Return current buffer as DataFrame (memory-efficient)"""
return pd.DataFrame(self.trades_buffer)
def get_aggregated_stats(self):
"""Pre-aggregate to reduce memory footprint"""
if not self.trades_buffer:
return None
df = pd.DataFrame(self.trades_buffer)
# Aggregate by minute (drastically reduces data size)
aggregated = df.set_index('timestamp').resample('1min').agg({
'price': ['first', 'max', 'min', 'last'],
'volume': 'sum',
'side': lambda x: (x == 'buy').sum() # Count buy trades
}).reset_index()
# Clear buffer after aggregation
self.trades_buffer = []
return aggregatedError 3: HolySheep API Rate Limiting
Symptom: 429 Too Many Requests errors when making rapid AI analysis calls
# Problem: Exceeding API rate limits
Solution: Implement exponential backoff retry logic
import time
import holySheep
from holySheep import HolySheepAI, RateLimitError, APIError
class RobustAIAnalyzer:
def __init__(self, api_key):
self.client = HolySheepAI(
api_key=api_key,
base_url="https://api.holysheep.ai/v1"
)
self.max_retries = 3
self.base_delay = 1 # seconds
def analyze_with_retry(self, prompt, model="gemini-2.5-flash"):
"""AI analysis with exponential backoff retry"""
for attempt in range(self.max_retries):
try:
response = self.client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": "You are an expert analyst."},
{"role": "user", "content": prompt}
],
max_tokens=500
)
return response.choices[0].message.content
except RateLimitError as e:
# Exponential backoff: 1s, 2s, 4s
delay = self.base_delay * (2 ** attempt)
print(f"Rate limited. Retrying in {delay}s... (Attempt {attempt + 1}/{self.max_retries})")
time.sleep(delay)
except APIError as e:
if attempt == self.max_retries - 1:
raise Exception(f"API error after {self.max_retries} retries: {e}")
time.sleep(self.base_delay * (2 ** attempt))
raise Exception("Max retries exceeded")Full Integration Example: End-to-End Pipeline
#!/usr/bin/env python3
"""
Complete OKX WebSocket + Pandas + HolySheep AI Pipeline
File: market_analysis_pipeline.py
"""
import asyncio
import websockets
import json
import pandas as pd
from datetime import datetime
from holySheep import HolySheepAI
Configuration
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
Initialize HolySheep AI Client
ai_client = HolySheepAI(
api_key=HOLYSHEEP_API_KEY,
base_url=HOLYSHEEP_BASE_URL
)
class MarketAnalysisPipeline:
"""
End-to-end market data collection, analysis, and AI insights pipeline.
"""
def __init__(self, symbols=["BTC-USDT", "ETH-USDT"]):
self.symbols = symbols
self.okx_url = "wss://ws.okx.com:8443/ws/v5/public"
self.all_trades = {s: [] for s in symbols}
self.all_orderbooks = {s: [] for s in symbols}
async def run(self, duration_seconds=60):
"""Execute full pipeline for specified duration"""
print(f"[{datetime.now()}] Starting Market Analysis Pipeline")
print(f"Monitoring: {', '.join(self.symbols)}")
print(f"Duration: {duration_seconds}s")
print("-" * 50)
async with websockets.connect(self.okx_url) as ws:
# Subscribe to all symbols
subscribe_msg = {
"op": "subscribe",
"args": [
{"channel": "trades", "instId": symbol}
for symbol in self.symbols
] + [
{"channel": "books", "instId": symbol, "sz": "400"}
for symbol in self.symbols
]
}
await ws.send(json.dumps(subscribe_msg))
print(f"[{datetime.now()}] Subscribed to {len(subscribe_msg['args'])} channels")
# Collect data
start = datetime.now()
while (datetime.now() - start).total_seconds() < duration_seconds:
msg = await ws.recv()
data = json.loads(msg)
if 'data' in data:
for item in data['data']:
symbol = item['instId']
if data.get('arg', {}).get('channel') == 'trades':
self._store_trade(symbol, item)
elif data.get('arg', {}).get('channel') == 'books':
self._store_orderbook(symbol, item)
print(f"[{datetime.now()}] Data collection complete")
return self.generate_insights()
def _store_trade(self, symbol, trade):
self.all_trades[symbol].append({
'timestamp': datetime.fromtimestamp(int(trade['ts'])/1000),
'side': trade['side'],
'price': float(trade['px']),