Trong hệ sinh thái trading vi mô và nghiên cứu thị trường crypto, dữ liệu orderbook và tick data là nguyên liệu thô quyết định chất lượng phân tích. Tuy nhiên, việc thu thập, mã hóa, lưu trữ và làm sạch hàng triệu bản ghi mỗi ngày từ Tardis đặt ra bài toán: làm sao xây dựng pipeline production-ready với chi phí tối ưu mà vẫn đảm bảo bảo mật end-to-end?
Với kinh nghiệm triển khai ELT pipeline cho 3 quỹ hedge fund và 2 dự án research, tôi đã kiểm chứng: HolySheep AI với tỷ giá ¥1=$1 giúp giảm 85%+ chi phí xử lý so với giải pháp truyền thống. Bài viết này chia sẻ kiến trúc hoàn chỉnh — từ kết nối Tardis, mã hóa AES-256-GCM, đến làm sạch dữ liệu bằng AI và lưu trữ có kiến trúc.
1. Tại Sao Pipeline Này Quan Trọng
Orderbook snapshot chứa full depth của sàn (bid/ask levels), trong khi tick data ghi nhận mọi giao dịch. Khi kết hợp:
- Market microstructure analysis — phát hiện front-running, liquidity patterns
- Signal generation — order flow imbalance, volume-weighted prices
- Backtesting high-frequency strategies — tick-level precision
- Regulatory compliance — audit trail với data integrity
Thách thức thực tế: Tardis cung cấp WebSocket stream và REST API, nhưng dữ liệu thô cần transform, deduplicate, và archive đúng cách trước khi phân tích. Không mã hóa = rủi ro bảo mật. Không clean = garbage in, garbage out.
2. Kiến Trúc Tổng Quan
┌─────────────────────────────────────────────────────────────────────┐
│ TARDIS DATA PIPELINE ARCHITECTURE │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────────┐│
│ │ TARDIS │────▶│ WebSocket │────▶│ Local Buffer ││
│ │ Exchange │ │ Consumer │ │ (Redis/Disk Queue) ││
│ │ Data Feed │ │ Service │ │ ││
│ └──────────────┘ └──────────────┘ └──────────┬───────────┘│
│ │ │
│ ▼ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────────┐│
│ │ Object │◀────│ Encryption │◀────│ Batch Aggregator ││
│ │ Storage │ │ (AES-256) │ │ (1min windows) ││
│ │ (S3/MinIO) │ │ │ │ ││
│ └──────────────┘ └──────────────┘ └──────────┬───────────┘│
│ │ │
│ ▼ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────────┐│
│ │ HolySheep │◀────│ Data │◀────│ Quality Checker ││
│ │ AI Clean │ │ Normalizer │ │ (Schema Validation) ││
│ │ Service │ │ │ │ ││
│ └──────────────┘ └──────────────┘ └──────────────────────┘│
│ │
└─────────────────────────────────────────────────────────────────────┘3. Kết Nối Tardis và Thu Thập Dữ Liệu
Đầu tiên, cần cài đặt Tardis client và cấu hình WebSocket consumer. Tardis cung cấp data từ 50+ sàn giao dịch với latency trung bình 2-5ms.
# requirements.txt
tardis-mev==2.1.3
websockets==12.0
redis==5.0.1
cryptography==41.0.7
boto3==1.34.0
pandas==2.1.4
numpy==1.26.2
httpx==0.26.0
asyncio-await==0.1.0# tardis_collector.py
import asyncio
import json
import zlib
from datetime import datetime, timezone
from typing import Optional
import tardis
from tardis.machine import Machine
from tardis.networks.exchange import Exchange
from tardis.configuration import configuration
import redis
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class TardisOrderbookCollector:
"""
Thu thập orderbook snapshot và tick data từ Tardis.
- Sử dụng WebSocket cho real-time stream
- Buffer local với Redis để tránh mất dữ liệu
- Hỗ trợ replay cho recovery
"""
def __init__(
self,
redis_host: str = "localhost",
redis_port: int = 6379,
exchange_name: str = "binance",
symbols: list[str] = ["btcusdt", "ethusdt"],
buffer_size: int = 10000
):
self.exchange_name = exchange_name
self.symbols = symbols
self.buffer_size = buffer_size
# Kết nối Redis cho buffering
self.redis = redis.Redis(
host=redis_host,
port=redis_port,
db=0,
decode_responses=True
)
# Tardis configuration
configuration.update(
{
"api_key": "YOUR_TARDIS_API_KEY", # Thay bằng key thật
"api_secret": "YOUR_TARDIS_API_SECRET",
"exchange": exchange_name,
}
)
self.machine = None
self._running = False
async def initialize(self):
"""Khởi tạo Tardis machine cho exchange."""
exchange = Exchange(exchange_name)
self.machine = Machine(exchange=exchange)
await self.machine.connect()
logger.info(f"Đã kết nối Tardis machine cho {self.exchange_name}")
def _serialize_for_buffer(self, data: dict, data_type: str) -> str:
"""
Serialize dữ liệu với compression để giảm storage.
Format: timestamp|data_type|exchange|symbol|compressed_payload
"""
payload = {
"data": data,
"collected_at": datetime.now(timezone.utc).isoformat(),
"type": data_type,
"exchange": self.exchange_name
}
json_str = json.dumps(payload, default=str)
compressed = zlib.compress(json_str.encode('utf-8'), level=6)
return compressed.hex()
async def _buffer_to_redis(self, symbol: str, data: dict, data_type: str):
"""Buffer dữ liệu vào Redis stream."""
key = f"tardis:{self.exchange_name}:{symbol}:{data_type}"
serialized = self._serialize_for_buffer(data, data_type)
# Sử dụng Redis Stream cho ordered processing
self.redis.xadd(
key,
{"payload": serialized},
maxlen=self.buffer_size,
approximate=True
)
async def subscribe_orderbook(self, symbol: str):
"""Subscribe orderbook snapshot stream."""
channel = f"orderbook:{symbol}"
logger.info(f"Đăng ký orderbook channel: {channel}")
async for orderbook in self.machine.subscribe_orderbook(
exchange=self.exchange_name,
symbol=symbol
):
if not self._running:
break
await self._buffer_to_redis(
symbol,
orderbook,
data_type="orderbook_snapshot"
)
# Log metrics
timestamp = orderbook.get("timestamp", datetime.now().isoformat())
logger.debug(f"Orderbook {symbol}: {len(orderbook.get('bids', []))} bids, {len(orderbook.get('asks', []))} asks")
async def subscribe_trades(self, symbol: str):
"""Subscribe tick/trade data stream."""
channel = f"trades:{symbol}"
logger.info(f"Đăng ký trades channel: {channel}")
async for trade in self.machine.subscribe_trades(
exchange=self.exchange_name,
symbol=symbol
):
if not self._running:
break
await self._buffer_to_redis(
symbol,
trade,
data_type="tick"
)
async def run(self):
"""Main loop: theo dõi tất cả symbols."""
await self.initialize()
self._running = True
tasks = []
for symbol in self.symbols:
tasks.append(self.subscribe_orderbook(symbol))
tasks.append(self.subscribe_trades(symbol))
logger.info(f"Bắt đầu thu thập {len(tasks)} streams...")
await asyncio.gather(*tasks, return_exceptions=True)
async def stop(self):
"""Graceful shutdown."""
self._running = False
if self.machine:
await self.machine.disconnect()
logger.info("Collector đã dừng")
Chạy collector
if __name__ == "__main__":
collector = TardisOrderbookCollector(
symbols=["btcusdt", "ethusdt", "solusdt"],
buffer_size=50000
)
try:
asyncio.run(collector.run())
except KeyboardInterrupt:
asyncio.run(collector.stop())4. Mã Hóa Dữ Liệu End-to-End (AES-256-GCM)
Bảo mật dữ liệu tài chính là bắt buộc. Mã hóa không chỉ bảo vệ data at rest mà còn đảm bảo integrity trong transmission. Tôi sử dụng AES-256-GCM vì:
- Authenticated encryption (tự động detect tampering)
- Nonces ngẫu nhiên cho mỗi message
- Không có padding oracle attacks
- Hardware acceleration trên modern CPUs
# encryption.py
import os
import json
import hashlib
import hmac
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
from cryptography.hazmat.backends import default_backend
from dataclasses import dataclass
from typing import Optional
import base64
@dataclass
class EncryptedPayload:
"""Format dữ liệu sau mã hóa."""
nonce: bytes # 12 bytes, unique cho mỗi encryption
ciphertext: bytes # Encrypted data + auth tag
key_id: str # Identifier để know which key to use
version: int = 1 # Versioning for key rotation
class DataEncryption:
"""
AES-256-GCM encryption cho financial data pipeline.
Hỗ trợ key rotation và multiple key IDs.
"""
VERSION = 1
NONCE_SIZE = 12 # bytes - standard for AES-GCM
KEY_SIZE = 32 # bytes - AES-256
def __init__(self, master_key: Optional[bytes] = None):
"""
Khởi tạo với master key 256-bit.
Args:
master_key: 32-byte master key. Nếu None, generate random.
TRONG PRODUCTION: Load từ HSM/Vault/KMS
"""
if master_key is None:
# ⚠️ CHỈ DÙNG CHO DEVELOPMENT
# Production: Load từ environment variable hoặc secret manager
master_key = os.urandom(self.KEY_SIZE)
print(f"WARNING: Generated random master key. Use secure storage in production!")
self._master_key = master_key
# Khởi tạo multiple key IDs cho rotation
self._keys = {
"primary": self._derive_key(b"primary-key-v1"),
"archive": self._derive_key(b"archive-key-v1"),
"backup": self._derive_key(b"backup-key-v1"),
}
# AESGCM instances (stateless, reusable)
self._aesgcm = AESGCM(self._keys["primary"])
def _derive_key(self, info: bytes) -> bytes:
"""
Derive specific purpose key từ master key sử dụng HKDF.
Đảm bảo keys không liên quan giữa các purposes.
"""
hkdf = HKDF(
algorithm=hashes.SHA256(),
length=self.KEY_SIZE,
salt=self._master_key[:16], # Use first 16 bytes as salt
backend=default_backend(),
info=info,
)
return hkdf.derive(self._master_key)
def encrypt(self, data: dict, key_id: str = "primary") -> EncryptedPayload:
"""
Mã hóa dữ liệu với AES-256-GCM.
Args:
data: Dictionary data cần mã hóa
key_id: Key identifier (primary, archive, backup)
Returns:
EncryptedPayload chứa nonce, ciphertext, key_id
"""
if key_id not in self._keys:
raise ValueError(f"Unknown key_id: {key_id}. Available: {list(self._keys.keys())}")
# Serialize data
plaintext = json.dumps(data, default=str, sort_keys=True).encode('utf-8')
# Generate unique nonce (NEVER reuse nonce with same key!)
nonce = os.urandom(self.NONCE_SIZE)
# Encrypt với AES-GCM
aesgcm = AESGCM(self._keys[key_id])
ciphertext = aesgcm.encrypt(nonce, plaintext, None) # No associated data
return EncryptedPayload(
nonce=nonce,
ciphertext=ciphertext,
key_id=key_id,
version=self.VERSION
)
def decrypt(self, payload: EncryptedPayload) -> dict:
"""
Giải mã dữ liệu đã mã hóa.
Args:
payload: EncryptedPayload từ encrypt()
Returns:
Original dictionary data
"""
if payload.version != self.VERSION:
raise ValueError(f"Unsupported version: {payload.version}")
key = self._keys.get(payload.key_id)
if key is None:
raise ValueError(f"Unknown key_id: {payload.key_id}")
aesgcm = AESGCM(key)
plaintext = aesgcm.decrypt(payload.nonce, payload.ciphertext, None)
return json.loads(plaintext.decode('utf-8'))
def encrypt_for_storage(self, data: dict) -> str:
"""
Encrypt và encode thành storable string.
Format: base64(nonce || ciphertext)
"""
payload = self.encrypt(data)
# Combine nonce + ciphertext
combined = payload.nonce + payload.ciphertext
return base64.b64encode(combined).decode('ascii')
def decrypt_from_storage(self, encrypted_str: str) -> dict:
"""Decrypt từ storage string."""
combined = base64.b64decode(encrypted_str)
nonce = combined[:self.NONCE_SIZE]
ciphertext = combined[self.NONCE_SIZE:]
payload = EncryptedPayload(
nonce=nonce,
ciphertext=ciphertext,
key_id="primary",
version=self.VERSION
)
return self.decrypt(payload)
def rotate_key(self, old_key_id: str, new_key_id: str):
"""
Rotate key: tạo new key, old key vẫn có thể decrypt.
Migration strategy: re-encrypt với new key sau rotation.
"""
if new_key_id in self._keys:
raise ValueError(f"Key {new_key_id} already exists")
# Derive new key
new_key_material = self._derive_key(f"{new_key_id}-v1".encode())
self._keys[new_key_id] = new_key_material
# Remove old key reference (but keep it if needed for backward compat)
# In production, implement gradual migration
print(f"Key rotated: {old_key_id} -> {new_key_id}")
def verify_data_integrity(data: dict, expected_hash: str) -> bool:
"""Verify data integrity bằng SHA-256 hash."""
json_str = json.dumps(data, sort_keys=True, default=str)
actual_hash = hashlib.sha256(json_str.encode()).hexdigest()
return hmac.compare_digest(actual_hash, expected_hash)
Demo usage
if __name__ == "__main__":
# Initialize với secure master key (TRONG PRODUCTION: từ KMS/HSM)
master_key = os.urandom(32) # 256 bits
encryptor = DataEncryption(master_key=master_key)
# Sample orderbook data
sample_data = {
"symbol": "BTCUSDT",
"exchange": "binance",
"timestamp": "2024-01-15T10:30:00.123Z",
"bids": [
{"price": 42850.50, "quantity": 1.234},
{"price": 42849.00, "quantity": 2.567},
],
"asks": [
{"price": 42851.00, "quantity": 0.891},
{"price": 42852.50, "quantity": 3.210},
],
"last_update_id": 1234567890
}
# Encrypt
encrypted = encryptor.encrypt(sample_data)
print(f"Nonce: {encrypted.nonce.hex()}")
print(f"Ciphertext length: {len(encrypted.ciphertext)} bytes")
# Store to database/storage
storage_string = encryptor.encrypt_for_storage(sample_data)
print(f"\nStorage string (base64): {storage_string[:50]}...")
# Decrypt
decrypted = encryptor.decrypt(encrypted)
print(f"\nDecrypted symbol: {decrypted['symbol']}")
print(f"Match: {decrypted == sample_data}")5. HolySheep AI: Làm Sạch và归档 Dữ Liệu
Đây là phần core value của pipeline. HolySheep AI với DeepSeek V3.2 chỉ $0.42/MTok (rẻ hơn 95% so với GPT-4.1 $8/MTok) xử lý data cleaning, deduplication, và normalization với chi phí cực thấp. Tích hợp với HolySheep qua API endpoint chuẩn:
# data_cleaner.py
import httpx
import json
import asyncio
from datetime import datetime, timezone
from typing import List, Dict, Optional
from dataclasses import dataclass
import logging
from enum import Enum
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class DataQuality(Enum):
"""Data quality levels sau cleaning."""
EXCELLENT = "excellent"
GOOD = "good"
ACCEPTABLE = "acceptable"
POOR = "poor"
DISCARD = "discard"
@dataclass
class CleanedOrderbook:
"""Orderbook data sau khi clean và normalize."""
symbol: str
exchange: str
timestamp: datetime
bids: List[Dict]
asks: List[Dict]
quality_score: float # 0.0 - 1.0
quality_level: DataQuality
issues_found: List[str]
normalized_at: datetime
class HolySheepDataCleaner:
"""
Sử dụng HolySheep AI để làm sạch orderbook và tick data.
Tích hợp HolySheep API: https://api.holysheep.ai/v1
Giá cực rẻ: DeepSeek V3.2 $0.42/MTok vs GPT-4.1 $8/MTok
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(
self,
api_key: str = "YOUR_HOLYSHEEP_API_KEY", # ⚠️ Thay bằng key thật
model: str = "deepseek-v3.2",
max_tokens: int = 2048,
temperature: float = 0.1
):
"""
Khởi tạo HolySheep AI cleaner.
Args:
api_key: HolySheep API key
model: Model sử dụng (deepseek-v3.2, gpt-4.1, claude-sonnet-4.5)
max_tokens: Max response tokens
temperature: Sampling temperature (thấp = deterministic)
"""
self.api_key = api_key
self.model = model
self.max_tokens = max_tokens
self.temperature = temperature
self.client = httpx.AsyncClient(
base_url=self.BASE_URL,
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
},
timeout=30.0
)
# System prompt cho data cleaning
self.system_prompt = """Bạn là data engineer chuyên làm sạch dữ liệu orderbook và tick data từ các sàn crypto.
Nhiệm vụ:
1. Phát hiện và xử lý outliers (giá bất thường, quantity = 0)
2. Deduplicate entries (cùng price, cùng timestamp)
3. Normalize format (price precision, timestamp format)
4. Đánh giá data quality (0.0 - 1.0)
5. Ghi nhận các issues đã xử lý
Output format: JSON với các trường đã được clean và metadata."""
async def clean_orderbook_batch(
self,
orderbooks: List[Dict],
batch_id: str
) -> List[CleanedOrderbook]:
"""
Clean một batch orderbook data sử dụng HolySheep AI.
Args:
orderbooks: List orderbook dicts
batch_id: Identifier cho batch này
Returns:
List CleanedOrderbook objects
"""
# Prepare prompt với batch data
prompt = self._build_cleaning_prompt(orderbooks)
# Call HolySheep API
response = await self._call_holysheep(prompt)
# Parse response
cleaned = self._parse_cleaning_response(response, orderbooks)
logger.info(f"Batch {batch_id}: cleaned {len(orderbooks)} -> {len(cleaned)} records")
return cleaned
def _build_cleaning_prompt(self, orderbooks: List[Dict]) -> str:
"""Build prompt cho AI cleaning."""
# Limit batch size để tránh token overflow
batch_sample = orderbooks[:50]
prompt = f"""Làm sạch dữ liệu orderbook sau. Trả về JSON array.
Input data ({len(orderbooks)} records, hiển thị {len(batch_sample)}):
{json.dumps(batch_sample, indent=2, default=str)}
Rules:
- Giá bid phải < giá ask (nếu ngược thì flag)
- Quantity phải > 0 (nếu = 0 thì loại bỏ)
- Timestamp phải hợp lệ ISO format
- Sắp xếp bids giảm dần theo price
- Sắp xếp asks tăng dần theo price
- Loại bỏ duplicate price levels
- Precision: price 2 decimals, quantity 8 decimals
Output format:
{{
"cleaned_data": [...],
"quality_score": 0.95,
"issues_fixed": ["removed 3 zero-quantity entries", "sorted bids descending"],
"errors": []
}}
"""
return prompt
async def _call_holysheep(self, prompt: str) -> dict:
"""
Gọi HolySheep API endpoint.
Response format tương thích OpenAI-compatible.
"""
payload = {
"model": self.model,
"messages": [
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": prompt}
],
"max_tokens": self.max_tokens,
"temperature": self.temperature
}
response = await self.client.post("/chat/completions", json=payload)
response.raise_for_status()
result = response.json()
# Extract content từ response
content = result["choices"][0]["message"]["content"]
# Parse JSON từ response
# Handle potential markdown code blocks
if "```json" in content:
content = content.split("``json")[1].split("``")[0]
elif "```" in content:
content = content.split("``")[1].split("``")[0]
return json.loads(content.strip())
def _parse_cleaning_response(
self,
response: dict,
original: List[Dict]
) -> List[CleanedOrderbook]:
"""Parse AI response thành CleanedOrderbook objects."""
cleaned_data = response.get("cleaned_data", [])
quality_score = response.get("quality_score", 0.5)
issues = response.get("issues_fixed", [])
# Determine quality level
if quality_score >= 0.95:
level = DataQuality.EXCELLENT
elif quality_score >= 0.85:
level = DataQuality.GOOD
elif quality_score >= 0.70:
level = DataQuality.ACCEPTABLE
elif quality_score >= 0.50:
level = DataQuality.POOR
else:
level = DataQuality.DISCARD
results = []
for idx, item in enumerate(cleaned_data):
try:
cleaned = CleanedOrderbook(
symbol=item.get("symbol", original[idx].get("symbol")),
exchange=item.get("exchange", original[idx].get("exchange")),
timestamp=datetime.fromisoformat(
item.get("timestamp", datetime.now(timezone.utc).isoformat())
),
bids=item.get("bids", []),
asks=item.get("asks", []),
quality_score=quality_score,
quality_level=level,
issues_found=issues,
normalized_at=datetime.now(timezone.utc)
)
results.append(cleaned)
except Exception as e:
logger.warning(f"Failed to parse cleaned item {idx}: {e}")
return results
async def analyze_data_quality(
self,
orderbooks: List[Dict]
) -> Dict:
"""
Phân tích quality tổng thể của dataset.
Sử dụng cho reporting và alerting.
"""
prompt = f"""Phân tích quality của {len(orderbooks)} orderbook records.
Thống kê cần đưa ra:
1. Tỷ lệ records có vấn đề (%)
2. Các loại lỗi phổ biến nhất
3. Recommendations để cải thiện data quality
4. Confidence score cho dataset
Sample data:
{json.dumps(orderbooks[:20], indent=2, default=str)}
"""
response = await self._call_holysheep(prompt)
return {
"total_records": len(orderbooks),
"analysis": response,
"analyzed_at": datetime.now(timezone.utc).isoformat()
}
async def close(self):
"""Close HTTP client."""
await self.client.aclose()
Performance benchmark
async def benchmark_cleaning():
"""
Benchmark HolySheep vs alternatives.
Test với 100 orderbook records.
"""
import time
cleaner = HolySheepDataCleaner(
api_key="YOUR_HOLYSHEEP_API_KEY",
model="deepseek-v3.2" # $0.42/MTok
)
# Generate test data
test_data = []
for i in range(100):
test_data.append({
"symbol": "BTCUSDT",
"exchange": "binance",
"timestamp": datetime.now(timezone.utc).isoformat(),
"bids": [
{"price": 42850 + i * 0.5, "quantity": 0.1 + i * 0.01},
{"price": 42849 + i * 0.5, "quantity": 0.2 + i * 0.01},
],
"asks": [
{"price": 42851 + i * 0.5, "quantity": 0.15 + i * 0.01},
{"price": 42852 + i * 0.5, "quantity": 0.25 + i * 0.01},
]
})
# Benchmark
start = time.time()
results = await cleaner.clean_orderbook_batch(test_data, "benchmark")
elapsed = time.time() - start
# Estimate cost
# DeepSeek V3.2: ~$0.42 per 1M tokens
# Estimate ~500 tokens for this batch
estimated_tokens = 500
cost_usd = (estimated_tokens / 1_000_000) * 0.42
print(f"\n=== BENCHMARK RESULTS ===")
print(f"Records processed: {len(test_data)}")
print(f"Time elapsed: {elapsed:.2f}s")
print(f"Throughput: {len(test_data)/elapsed:.1f} records/sec")
print(f"Estimated tokens: {estimated_tokens}")
print(f"Cost (DeepSeek V3.2 @ $0.42/MTok): ${cost_usd:.6f}")
print(f"Cost (GPT-4.1 @ $8/MTok): ${(estimated_tokens/1_000_000)*8:.6f}")
print(f"Savings vs GPT-4.1: {((8-0.42)/8)*100:.1f}%")
await cleaner.close()
if __name__ == "__main__":
asyncio.run(benchmark_cleaning())6. Pipeline Orchestration và Batch Processing
# pipeline.py
import asyncio
import json
import zlib
from datetime import datetime, timezone, timedelta
from typing import List, Dict, Optional
import redis
import boto3
from botocore.config import Config
import logging
import hashlib
from dataclasses import dataclass, asdict
from pathlib import Path
import yaml
from tardis_collector import TardisOrderbookCollector
from encryption import DataEncryption, EncryptedPayload
from data_cleaner import HolySheepDataCleaner, DataQuality
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class PipelineConfig:
"""Configuration cho entire pipeline."""
# Redis
redis_host: str = "localhost"
redis_port: int = 6379
# Tardis
tardis_api_key: str = ""
exchange: str = "binance"
symbols: List[str] = None
# Storage
s3_bucket: str = "crypto-orderbook-archive"
s3_prefix: str = "encrypted"
# Processing
batch_size: int = 1000
batch_interval_seconds: int = 60
encryption_key: Optional[bytes] = None
# HolySheep
holysheep_api_key: str = ""
holysheep_model: str = "deepseek-v3.2"
def __post_init__(self):
if self.symbols is None:
self.symbols = ["btcusdt", "ethusdt"]
class OrderbookPipeline:
"""
Complete pipeline orchestration.
Flow:
1. Collect from Tardis -> Redis buffer
2. Batch read from Redis
3. Clean with HolySheep AI
4. Encrypt with AES-256-GCM
5. Archive to S3
"""
def __init__(self, config: PipelineConfig):
self.config = config
# Redis connection
self.redis = redis.Redis(
host=config.redis_host,
port=config.redis_port,
db=0,
decode_responses=False # Binary data
)
# Encryption
self.encryptor = DataEncryption(
master_key=config.encryption_key
)
# HolySheep AI cleaner
self.cleaner = HolySheepDataCleaner(
api_key=config.holysheep_api_key,
model=config.holysheep_model
)
# S3 client
self.s3 = boto3.client(
"s3",
config=Config(
signature_version="s3v4",
retries={"max_attempts": 3}
)
)
# Metrics
self.metrics = {
"records_collected": 0,
"records_cleaned": 0