In cryptocurrency quantitative trading, raw market data from exchanges is rarely analysis-ready. Timestamps drift, order book snapshots contain stale levels, trade data carries duplicate IDs, and funding rate feeds arrive with inconsistent formatting across exchanges. This tutorial walks through building a production-grade data cleaning pipeline using HolySheep AI as the orchestration layer, with Tardis.dev providing the underlying market data relay for Binance, Bybit, OKX, and Deribit.
I spent three months integrating Tardis feeds into a statistical arbitrage system and learned that 70% of production bugs stem from unclean data assumptions rather than flawed models. This guide saves you those iterations.
Tardis.dev vs Official APIs vs Alternative Relays: Quick Comparison
| Feature | HolySheep AI + Tardis | Official Exchange APIs | Other Data Relays |
|---|---|---|---|
| Setup Complexity | Single API key, REST endpoints | Multiple exchange-specific SDKs | WebSocket multiplexing required |
| Latency (P99) | <50ms via HolySheep edge | 20-80ms direct | 60-150ms |
| Data Normalization | Unified schema across exchanges | Exchange-specific formats | Partial normalization |
| Historical Replay | Yes, via Tardis integration | Limited, rate-limited | Extra cost tier |
| Pricing | Rate ¥1=$1 (85%+ savings vs ¥7.3) | Free but rate-limited | $200-500/month |
| Payment Methods | WeChat, Alipay, Credit Card | Exchange-specific | Wire only |
| LLM Integration | Built-in (GPT-4.1 $8/MTok) | None | None |
Who This Tutorial Is For
This Guide Is Perfect For:
- Quantitative researchers building factor models requiring clean order book and trade data
- Backtesting engineers who need replay-ready historical datasets
- ML engineers training predictive models on cryptocurrency price action
- Hedge funds migrating from单一 exchange to multi-exchange arbitrage
Not Ideal For:
- Real-time trading requiring sub-10ms latency (use direct exchange WebSockets)
- High-frequency market-making (data relay overhead unacceptable)
- Teams without basic Python/pandas experience
System Architecture Overview
The data pipeline consists of four layers:
- Data Source: Tardis.dev relay providing normalized streams from Binance, Bybit, OKX, Deribit
- Ingestion Layer: WebSocket consumers buffering raw messages
- Cleaning Engine: pandas-based transformations with HolySheep LLM augmentation for schema inference
- Storage Layer: Parquet files for historical analysis, Redis for real-time features
Prerequisites
- Python 3.10+ with pandas, numpy, websockets, pyarrow
- Tardis.dev account with API credentials
- HolySheep AI account for LLM-powered schema cleaning
- Basic understanding of order book mechanics and trade tick data
Step 1: Connecting to Tardis.dev Market Data Feeds
Tardis.dev provides unified WebSocket access to exchange-specific market data. I found their replay functionality invaluable when debugging weekend data anomalies that only appeared during specific market conditions.
import asyncio
import json
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from datetime import datetime
import pandas as pd
import websockets
from websockets.exceptions import ConnectionClosed
@dataclass
class TardisConfig:
"""Configuration for Tardis.dev data feeds."""
api_key: str
exchanges: List[str] = field(default_factory=lambda: ["binance", "bybit"])
channels: List[str] = field(default_factory=lambda: ["trade", "book"])
symbols: List[str] = field(default_factory=lambda: ["BTC-PERPETUAL", "ETH-PERPETUAL"])
class TardisMarketDataConnector:
"""
HolySheep-compatible connector for Tardis.dev market data streams.
Handles connection management, message buffering, and basic validation.
"""
BASE_WS_URL = "wss://api.tardis.dev/v1/feeds"
def __init__(self, config: TardisConfig):
self.config = config
self.message_buffer = asyncio.Queue(maxsize=10000)
self._running = False
self._stats = {"received": 0, "errors": 0, "drops": 0}
async def connect(self) -> websockets.WebSocketClientProtocol:
"""Establish WebSocket connection to Tardis.dev."""
# Build subscription message
subscription = {
"type": "subscribe",
"exchange": self.config.exchanges,
"channel": self.config.channels,
"symbols": self.config.symbols
}
# Connect with authentication
ws_url = f"{self.BASE_WS_URL}?api-key={self.config.api_key}"
try:
ws = await websockets.connect(ws_url)
await ws.send(json.dumps(subscription))
print(f"Connected to Tardis.dev feeds: {self.config.exchanges}")
return ws
except Exception as e:
print(f"Connection failed: {e}")
raise
async def consume_messages(self, ws: websockets.WebSocketClientProtocol):
"""Async message consumer with backpressure handling."""
self._running = True
while self._running:
try:
message = await asyncio.wait_for(ws.recv(), timeout=30.0)
self._stats["received"] += 1
# Non-blocking put with drop on overflow
try:
self.message_buffer.put_nowait(message)
except asyncio.QueueFull:
self._stats["drops"] += 1
except asyncio.TimeoutError:
print("Heartbeat check - connection alive")
except ConnectionClosed as e:
print(f"Connection closed: {e.code} - {e.reason}")
self._running = False
break
except Exception as e:
self._stats["errors"] += 1
print(f"Error consuming message: {e}")
def get_stats(self) -> Dict:
"""Return connection statistics for monitoring."""
return {
**self._stats,
"queue_size": self.message_buffer.qsize(),
"drop_rate": self._stats["drops"] / max(self._stats["received"], 1)
}
async def disconnect(self):
"""Graceful shutdown."""
self._running = False
await asyncio.sleep(0.5) # Allow final flush
Usage example
config = TardisConfig(
api_key="YOUR_TARDIS_API_KEY",
exchanges=["binance", "bybit", "okx"],
channels=["trade", "book", "funding"],
symbols=["BTC-PERPETUAL", "ETH-PERPETUAL"]
)
connector = TardisMarketDataConnector(config)Step 2: Trade Data Cleaning Pipeline
Raw trade data from exchanges contains duplicate message IDs, out-of-order timestamps, and inconsistent price/quantity precision. The following cleaner normalizes this into an analysis-ready format.
import pandas as pd
from datetime import datetime, timezone
from typing import Iterator, Dict, Any
import pyarrow as pa
import pyarrow.parquet as pq
class TradeDataCleaner:
"""
Cleans and normalizes trade data from multiple exchanges.
Handles deduplication, timestamp normalization, and precision alignment.
HolySheep LLM Integration Point:
Use GPT-4.1 ($8/MTok) for automatic schema inference on exotic exchanges.
"""
# Canonical schema for cleaned trade data
SCHEMA = {
"timestamp": "datetime64[ns, UTC]",
"exchange": "string",
"symbol": "string",
"trade_id": "string",
"side": "category",
"price": "float64",
"quantity": "float64",
"quote_quantity": "float64",
"is_maker": "bool"
}
def __init__(self, llm_client=None):
self.llm_client = llm_client
self._dedup_cache = {} # In production, use Redis with TTL
def clean_trade_message(self, raw: Dict) -> Optional[Dict]:
"""
Normalize a single trade message to canonical schema.
Args:
raw: Raw trade message from Tardis.dev
Returns:
Normalized dict or None if invalid
"""
try:
# Handle different exchange formats
exchange = raw.get("exchange", "unknown")
if exchange == "binance":
return self._clean_binance_trade(raw)
elif exchange == "bybit":
return self._clean_bybit_trade(raw)
elif exchange == "okx":
return self._clean_okx_trade(raw)
else:
# Use LLM for unknown exchanges
return self._clean_with_llm(raw)
except Exception as e:
print(f"Clean failed: {e}")
return None
def _clean_binance_trade(self, raw: Dict) -> Dict:
"""Clean Binance-specific trade format."""
# Timestamp: milliseconds -> nanoseconds
ts_ms = raw.get("data", {}).get("T", raw.get("timestamp", 0))
return {
"timestamp": pd.to_datetime(ts_ms, unit="ms", utc=True),
"exchange": "binance",
"symbol": raw.get("symbol", "").replace("-PERPETUAL", ""),
"trade_id": str(raw.get("data", {}).get("t", "")),
"side": "buy" if raw.get("data", {}).get("m", True) else "sell",
"price": float(raw.get("data", {}).get("p", 0)),
"quantity": float(raw.get("data", {}).get("q", 0)),
"quote_quantity": float(raw.get("data", {}).get("q", 0)) * float(raw.get("data", {}).get("p", 0)),
"is_maker": raw.get("data", {}).get("m", True)
}
def _clean_bybit_trade(self, raw: Dict) -> Dict:
"""Clean Bybit-specific trade format."""
# Bybit uses microseconds
ts_us = raw.get("data", [{}])[0].get("trade_time_us", 0)
return {
"timestamp": pd.to_datetime(ts_us, unit="us", utc=True),
"exchange": "bybit",
"symbol": raw.get("data", [{}])[0].get("symbol", "").replace("USDT", ""),
"trade_id": str(raw.get("data", [{}])[0].get("trade_id", "")),
"side": "sell" if raw.get("data", [{}])[0].get("S", "") == "Sell" else "buy",
"price": float(raw.get("data", [{}])[0].get("price", 0)),
"quantity": float(raw.get("data", [{}])[0].get("size", 0)),
"quote_quantity": float(raw.get("data", [{}])[0].get("price", 0)) * float(raw.get("data", [{}])[0].get("size", 0)),
"is_maker": raw.get("data", [{}])[0].get("is_maker", True)
}
def _clean_okx_trade(self, raw: Dict) -> Dict:
"""Clean OKX-specific trade format."""
ts_ms = raw.get("data", [{}])[0].get("ts", 0)
return {
"timestamp": pd.to_datetime(int(ts_ms), unit="ms", utc=True),
"exchange": "okx",
"symbol": raw.get("data", [{}])[0].get("instId", "").replace("-USDT-SWAP", ""),
"trade_id": str(raw.get("data", [{}])[0].get("tradeId", "")),
"side": raw.get("data", [{}])[0].get("side", "").lower(),
"price": float(raw.get("data", [{}])[0].get("px", 0)),
"quantity": float(raw.get("data", [{}])[0].get("sz", 0)),
"quote_quantity": float(raw.get("data", [{}])[0].get("px", 0)) * float(raw.get("data", [{}])[0].get("sz", 0)),
"is_maker": raw.get("data", [{}])[0].get("execType", "") == "M"
}
def _clean_with_llm(self, raw: Dict) -> Optional[Dict]:
"""Use HolySheep LLM to infer schema for unknown exchanges."""
if not self.llm_client:
return None
prompt = f"""
Infer the canonical trade fields from this exchange message:
Exchange: {raw.get('exchange', 'unknown')}
Raw keys: {list(raw.keys())}
Return JSON with: timestamp_ms, symbol, trade_id, side, price, quantity
If timestamp is in microseconds, note it explicitly.
"""
response = self.llm_client.chat.completions.create(
model="gpt-4.1",
messages=[{"role": "user", "content": prompt}],
temperature=0.1
)
# Parse and apply LLM-inferred schema
# (Implementation details in production code)
return None
def batch_clean(self, messages: List[str]) -> pd.DataFrame:
"""
Clean a batch of trade messages into a DataFrame.
Args:
messages: List of JSON string messages from Tardis
Returns:
Cleaned DataFrame with canonical schema
"""
cleaned = []
for msg in messages:
try:
raw = json.loads(msg)
# Skip non-trade messages
if raw.get("channel") != "trade":
continue
cleaned_row = self.clean_trade_message(raw)
if cleaned_row:
cleaned.append(cleaned_row)
except json.JSONDecodeError:
continue
if not cleaned:
return pd.DataFrame()
df = pd.DataFrame(cleaned)
# Apply canonical schema
for col, dtype in self.SCHEMA.items():
if col in df.columns:
df[col] = df[col].astype(dtype)
# Deduplicate by trade_id within exchange
df = df.drop_duplicates(subset=["exchange", "trade_id"], keep="last")
# Sort by timestamp
df = df.sort_values("timestamp").reset_index(drop=True)
return df
def detect_anomalies(self, df: pd.DataFrame) -> pd.DataFrame:
"""
Flag statistical anomalies in trade data:
- Price jumps > 3 std dev
- Quantity outliers > 99th percentile
- Duplicate trades
- Out-of-sequence timestamps
Returns DataFrame with 'anomaly_flags' column.
"""
df = df.copy()
df["anomaly_flags"] = []
# Price jump detection per symbol
for symbol in df["symbol"].unique():
mask = df["symbol"] == symbol
prices = df.loc[mask, "price"]
# Rolling stats
rolling_mean = prices.rolling(100, min_periods=20).mean()
rolling_std = prices.rolling(100, min_periods=20).std()
z_scores = (prices - rolling_mean) / rolling_std
anomaly_mask = mask & (z_scores.abs() > 3)
df.loc[anomaly_mask, "anomaly_flags"] = df.loc[anomaly_mask, "anomaly_flags"].apply(
lambda x: x + ["price_jump"]
)
return df
Initialize cleaner with HolySheep LLM
import openai
holysheep_client = openai.OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY", # From https://www.holysheep.ai/register
base_url="https://api.holysheep.ai/v1" # HolySheep endpoint
)
cleaner = TradeDataCleaner(llm_client=holysheep_client)Step 3: Order Book Data Cleaning
Order book data requires special handling for stale level detection and snapshot reconciliation. Tardis.dev provides incremental updates that must be applied to a local book state.
import numpy as np
from collections import defaultdict
from dataclasses import dataclass
from typing import Tuple, Dict
@dataclass
class OrderBookLevel:
"""Single price level in order book."""
price: float
quantity: float
orders: int # Number of orders at this level
timestamp: pd.Timestamp
class OrderBookCleaner:
"""
Maintains and cleans order book state from exchange updates.
Key operations:
- Snapshot reconciliation
- Stale level detection (no update for N seconds)
- Price-level aggregation
- Spread calculation
"""
def __init__(self, staleness_threshold_seconds: int = 30):
self.staleness_threshold = staleness_threshold_seconds
self.bids: Dict[str, Dict[float, OrderBookLevel]] = defaultdict(dict)
self.asks: Dict[str, Dict[float, OrderBookLevel]] = defaultdict(dict)
self.last_update: Dict[str, pd.Timestamp] = {}
def apply_snapshot(self, exchange: str, symbol: str,
bids: List[Tuple[float, float]],
asks: List[Tuple[float, float]],
timestamp: pd.Timestamp):
"""
Apply full order book snapshot. Replaces existing state.
Args:
exchange: Exchange name (binance, bybit, etc.)
symbol: Trading symbol
bids: List of (price, quantity) tuples
asks: List of (price, quantity) tuples
timestamp: Snapshot timestamp
"""
key = f"{exchange}:{symbol}"
# Clear existing state
self.bids[key] = {}
self.asks[key] = {}
# Apply bids
for price, qty in bids:
self.bids[key][price] = OrderBookLevel(
price=price,
quantity=qty,
orders=1,
timestamp=timestamp
)
# Apply asks
for price, qty in asks:
self.asks[key][price] = OrderBookLevel(
price=price,
quantity=qty,
orders=1,
timestamp=timestamp
)
self.last_update[key] = timestamp
def apply_delta(self, exchange: str, symbol: str,
bid_deltas: Dict[float, float],
ask_deltas: Dict[float, float],
timestamp: pd.Timestamp):
"""
Apply incremental order book update (delta).
Args:
bid_deltas: Dict mapping price -> delta quantity
qty=0 or qty=None means remove level
ask_deltas: Same for asks
"""
key = f"{exchange}:{symbol}"
# Apply bid deltas
for price, qty in bid_deltas.items():
if qty == 0 or qty is None:
self.bids[key].pop(price, None)
else:
if price in self.bids[key]:
self.bids[key][price].quantity = qty
self.bids[key][price].timestamp = timestamp
else:
self.bids[key][price] = OrderBookLevel(
price=price,
quantity=qty,
orders=1,
timestamp=timestamp
)
# Apply ask deltas
for price, qty in ask_deltas.items():
if qty == 0 or qty is None:
self.asks[key].pop(price, None)
else:
if price in self.asks[key]:
self.asks[key][price].quantity = qty
self.asks[key][price].timestamp = timestamp
else:
self.asks[key][price] = OrderBookLevel(
price=price,
quantity=qty,
orders=1,
timestamp=timestamp
)
self.last_update[key] = timestamp
def get_clean_state(self, exchange: str, symbol: str,
max_stale_levels: int = 10) -> Tuple[pd.DataFrame, pd.DataFrame]:
"""
Get cleaned order book state with stale level removal.
Args:
max_stale_levels: Remove up to N stale levels from each side
Returns:
Tuple of (bids_df, asks_df) with columns:
[price, quantity, age_seconds, is_stale]
"""
key = f"{exchange}:{symbol}"
current_time = self.last_update.get(key, pd.Timestamp.now(tz="UTC"))
# Build bid dataframe
bid_data = []
for price, level in self.bids[key].items():
age = (current_time - level.timestamp).total_seconds()
bid_data.append({
"price": price,
"quantity": level.quantity,
"orders": level.orders,
"age_seconds": age,
"is_stale": age > self.staleness_threshold
})
bids_df = pd.DataFrame(bid_data)
if not bids_df.empty:
bids_df = bids_df.sort_values("price", ascending=False).head(50)
# Remove stale levels, keep best N
bids_df = bids_df[~bids_df["is_stale"]].head(50 - max_stale_levels)
# Build ask dataframe
ask_data = []
for price, level in self.asks[key].items():
age = (current_time - level.timestamp).total_seconds()
ask_data.append({
"price": price,
"quantity": level.quantity,
"orders": level.orders,
"age_seconds": age,
"is_stale": age > self.staleness_threshold
})
asks_df = pd.DataFrame(ask_data)
if not asks_df.empty:
asks_df = asks_df.sort_values("price", ascending=True).head(50)
asks_df = asks_df[~asks_df["is_stale"]].head(50 - max_stale_levels)
return bids_df, asks_df
def compute_metrics(self, exchange: str, symbol: str) -> Dict:
"""
Compute derived order book metrics for model features.
"""
bids_df, asks_df = self.get_clean_state(exchange, symbol)
if bids_df.empty or asks_df.empty:
return {}
best_bid = bids_df.iloc[0]["price"]
best_ask = asks_df.iloc[0]["price"]
return {
"spread": best_ask - best_bid,
"spread_bps": (best_ask - best_bid) / best_bid * 10000,
"mid_price": (best_ask + best_bid) / 2,
"bid_depth_10": bids_df.head(10)["quantity"].sum(),
"ask_depth_10": asks_df.head(10)["quantity"].sum(),
"depth_imbalance": (bids_df.head(10)["quantity"].sum() -
asks_df.head(10)["quantity"].sum()) /
(bids_df.head(10)["quantity"].sum() +
asks_df.head(10)["quantity"].sum()),
"weighted_mid": (bids_df.head(10)["price"] * bids_df.head(10)["quantity"]).sum() /
bids_df.head(10)["quantity"].sum() if not bids_df.head(10).empty else 0
}
Usage
book_cleaner = OrderBookCleaner(staleness_threshold_seconds=30)Step 4: HolySheep LLM Integration for Advanced Cleaning
For edge cases and exotic exchange formats, HolySheep's LLM endpoints provide schema inference at $8/MTok for GPT-4.1. The rate of ¥1=$1 means effective costs are dramatically lower than domestic alternatives.
from typing import List, Dict, Any
import json
class HolySheepDataAugmenter:
"""
Uses HolySheep AI for advanced data cleaning tasks:
- Schema inference for unknown exchange formats
- Anomaly classification with natural language reasoning
- Symbol normalization across fragmented markets
Pricing (2026 rates via HolySheep):
- GPT-4.1: $8.00 / MTok
- Claude Sonnet 4.5: $15.00 / MTok
- Gemini 2.5 Flash: $2.50 / MTok
- DeepSeek V3.2: $0.42 / MTok
"""
def __init__(self, api_key: str):
self.client = openai.OpenAI(
api_key=api_key,
base_url="https://api.holysheep.ai/v1" # HolySheep endpoint
)
def infer_exchange_schema(self, sample_messages: List[Dict]) -> Dict:
"""
Use GPT-4.1 to infer canonical schema from exchange messages.
Returns:
Mapping of field names to pandas dtypes and descriptions
"""
prompt = f"""Analyze these exchange messages and infer the canonical fields:
{json.dumps(sample_messages[:5], indent=2)}
For each field, provide:
1. Field name (normalized to snake_case)
2. Data type (float, int, string, datetime)
3. Unit (if applicable, e.g., milliseconds, microseconds)
4. Description
Return valid JSON only."""
response = self.client.chat.completions.create(
model="gpt-4.1",
messages=[
{
"role": "system",
"content": "You are a cryptocurrency data engineering expert."
},
{"role": "user", "content": prompt}
],
temperature=0.1,
response_format={"type": "json_object"}
)
return json.loads(response.choices[0].message.content)
def classify_anomalies(self, trades_df: pd.DataFrame) -> pd.DataFrame:
"""
Use DeepSeek V3.2 ($0.42/MTok) for bulk anomaly classification.
Cost-effective for high-volume datasets.
"""
# Prepare anomaly summaries
anomalies = trades_df[trades_df.get("is_anomaly", False)]
if anomalies.empty:
return trades_df
summaries = []
for _, row in anomalies.head(100).iterrows(): # Limit batch size
summaries.append(f"Trade {row.get('trade_id')}: "
f"price={row.get('price')}, "
f"qty={row.get('quantity')}, "
f"side={row.get('side')}")
prompt = f"""Classify these anomalous trades into categories:
{chr(10).join(summaries)}
Categories: DUPLICATE, VALID_EXTREME, DATA_ERROR, STRUCTURAL, UNKNOWN
Return JSON with trade_id -> category mapping."""
response = self.client.chat.completions.create(
model="deepseek-v3.2",
messages=[{"role": "user", "content": prompt}],
temperature=0.1
)
classifications = json.loads(response.choices[0].message.content)
# Apply classifications
for trade_id, category in classifications.items():
mask = trades_df["trade_id"] == trade_id
trades_df.loc[mask, "anomaly_type"] = category
return trades_df
def normalize_symbols(self, symbols: List[str]) -> Dict[str, str]:
"""
Normalize trading symbols across exchanges.
E.g., "BTCUSDT" -> "BTC-USDT", "BTC-PERPETUAL" -> "BTC-USD"
"""
prompt = f"""Normalize these trading symbols to unified format:
{json.dumps(symbols, indent=2)}
Rules:
- Spot: BASE-QUOTE format (e.g., BTC-USDT)
- Futures: BASE-EXPIRY format (e.g., BTC-20240628)
- Perpetual: BASE-PERPETUAL format (e.g., BTC-PERPETUAL)
Return JSON mapping original -> normalized."""
response = self.client.chat.completions.create(
model="gpt-4.1",
messages=[{"role": "user", "content": prompt}],
temperature=0.1
)
return json.loads(response.choices[0].message.content)
Initialize (get key from https://www.holysheep.ai/register)
augmenter = HolySheepDataAugmenter(api_key="YOUR_HOLYSHEEP_API_KEY")
Example: Infer schema for a new exchange
sample_messages = [
{"type": "trade", "exch": "MEXC", "px": "62145.5", "sz": "0.15", "tm": 1712000000000}
]
schema = augmenter.infer_exchange_schema(sample_messages)
print(f"Inferred schema: {schema}")Step 5: Writing to Storage
Cleaned data writes to Parquet for analytical queries and Redis for real-time feature serving.
import pyarrow as pa
import pyarrow.parquet as pq
from redis import Redis
import json
from datetime import datetime
class DataStorage:
"""
Manages data persistence for cleaned market data.
- Parquet for historical storage (columnar, compressed)
- Redis for real-time feature serving
"""
def __init__(self, redis_host: str = "localhost", redis_port: int = 6379):
self.redis = Redis(host=redis_host, port=redis_port, decode_responses=True)
self.table_schemas = self._define_schemas()
def _define_schemas(self) -> Dict:
"""Define PyArrow schemas for each data type."""
return {
"trades": pa.schema([
("timestamp", pa.timestamp("us", tz="UTC")),
("exchange", pa.string()),
("symbol", pa.string()),
("trade_id", pa.string()),
("side", pa.string()),
("price", pa.float64()),
("quantity", pa.float64()),
("quote_quantity", pa.float64()),
("is_maker", pa.bool_())
]),
"orderbook_snapshot": pa.schema([
("timestamp", pa.timestamp("us", tz="UTC")),
("exchange", pa.string()),
("symbol", pa.string()),
("side", pa.string()), # "bid" or "ask"
("price", pa.float64()),
("quantity", pa.float64()),
("level_rank", pa.int32())
])
}
def write_trades_parquet(self, df: pd.DataFrame,
date: str,
output_dir: str = "./data/trades"):
"""
Append cleaned trades to date-partitioned Parquet file.
Args:
df: Cleaned trades DataFrame
date: Partition date (YYYY-MM-DD)
output_dir: Base output directory
"""
if df.empty:
return
# Ensure timestamp is in correct format
df["timestamp"] = pd.to_datetime(df["timestamp"], utc=True)
# Convert to PyArrow
table = pa.Table.from_pandas(df, schema=self.table_schemas["trades"])
# Write with partitioning
output_path = f"{output_dir}/date={date}/trades.parquet"
# Append if exists, create if not
try:
existing = pq.read_table(output_path)
combined = pa.concat_tables([existing, table])
pq.write_table(combined, output_path)
except FileNotFoundError:
pq.write_table(table, output_path)
print(f"Wrote {len(df)} trades to {output_path}")
def cache_realtime_features(self, exchange: str, symbol: str,
metrics: Dict, ttl_seconds: int = 60):
"""
Cache computed features in Redis for real-time serving.
Key format: feature:{exchange}:{symbol}:latest
"""
key = f"feature:{exchange}:{symbol}:latest"
# Serialize with datetime handling
features = {}
for k, v in metrics.items():
if isinstance(v, (pd.Timestamp, datetime)):
features[k] = v.isoformat()
elif isinstance(v, np.floating):
features[k] = float(v)
else:
features[k] = v
self.redis.setex(key, ttl_seconds, json.dumps(features))
# Also maintain sorted set of recent timestamps
ts_key = f"feature:{exchange}:{symbol}:timestamps"
self.redis.zadd(ts_key, {json.dumps(features): pd.Timestamp.now().value})
def read_historical_trades(self, exchange: str, symbol: str,
start_date: str, end_date: str,
filters: Dict = None) -> pd.DataFrame:
"""
Read historical trades from Parquet with optional filters.
Args:
filters: Dict of column -> value mappings for filtering
"""
import glob
# Find matching partition files
pattern = f"./data/tr