paulpas Role: Provide scalable, cost-effective storage for trading data with efficient query capabilities
Philosophy: Data is an asset; storage must balance cost, accessibility, and retention policies
Key Principles
- Partitioned Storage: Organize data by symbol, date, and time for efficient access
- Columnar Format: Use Parquet for compressed, columnar storage
- Hot/Warm/Cold Tiers: Automatically move data based on access patterns
- Data Lifecycle Policies: Define retention and archival rules
- Metadata Catalog: Track schema, size, and access patterns
Implementation Guidelines
Structure
- Core logic: data_lake/lake_manager.py
- Storage adapter: data_lake/storage.py
- Tests: tests/test_data_lake.py
Patterns to Follow
- Use partitioned directory structure
- Implement batch writes for efficiency
- Support time-travel queries
- Track storage metrics
Adherence Checklist
Before completing your task, verify:
- Data is partitioned for efficient queries
- Columnar compression reduces storage costs
- Lifecycle policies are enforced
- Metadata catalog is updated on writes
- Access patterns are monitored
Relative paths in this skill (e.g., scripts/, reference/) are relative to this base directory.
Python Implementation
import os
import time
from typing import Dict, List, Optional, Any, Iterable
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
import json
import logging
class StorageTier(Enum):
HOT = "hot" # Fast access, SSD
WARM = "warm" # Moderate access, HDD
COLD = "cold" # Infrequent access, archive
@dataclass
class DataFile:
"""Represents a data file in the lake."""
path: str
size_bytes: int
row_count: int
partition: Dict[str, str]
created_at: float
last_accessed: float
tier: StorageTier
@dataclass
class DataLakeConfig:
"""Configuration for data lake."""
base_path: str
partition_columns: List[str] = field(default_factory=lambda: ["symbol", "date", "hour"])
retention_days: int = 365
compression: str = "snappy"
file_format: str = "parquet"
class DataLake:
"""Manages data storage in the data lake."""
def __init__(self, config: DataLakeConfig):
self.config = config
self.files: Dict[str, DataFile] = {}
self._metadata_path = os.path.join(config.base_path, "_metadata")
self._initialize_metadata()
def _initialize_metadata(self):
"""Initialize metadata storage."""
os.makedirs(self._metadata_path, exist_ok=True)
self.metadata_file = os.path.join(self._metadata_path, "files.json")
if os.path.exists(self.metadata_file):
try:
with open(self.metadata_file, 'r') as f:
data = json.load(f)
for path, file_data in data.items():
self.files[path] = DataFile(**file_data)
except Exception as e:
logging.warning(f"Failed to load metadata: {e}")
def _save_metadata(self):
"""Save metadata to disk."""
data = {
path: {
"path": f.path,
"size_bytes": f.size_bytes,
"row_count": f.row_count,
"partition": f.partition,
"created_at": f.created_at,
"last_accessed": f.last_accessed,
"tier": f.tier.value
}
for path, f in self.files.items()
}
os.makedirs(self._metadata_path, exist_ok=True)
with open(self.metadata_file, 'w') as f:
json.dump(data, f, indent=2)
def write_data(
self,
data: List[Dict[str, Any]],
partition_values: Dict[str, str]
) -> str:
"""Write data to lake with partitioning."""
partition_path = self._build_partition_path(partition_values)
filepath = os.path.join(
self.config.base_path,
partition_path,
f"data_{int(time.time() * 1000)}.parquet"
)
# In a real implementation, this would write to Parquet
# For demonstration, just create the directory structure
os.makedirs(os.path.dirname(filepath), exist_ok=True)
file_info = DataFile(
path=filepath,
size_bytes=len(str(data)), # Approximate
row_count=len(data),
partition=partition_values,
created_at=time.time(),
last_accessed=time.time(),
tier=StorageTier.HOT
)
self.files[filepath] = file_info
self._save_metadata()
return filepath
def _build_partition_path(self, partition_values: Dict[str, str]) -> str:
"""Build partition directory path."""
parts = []
for col in self.config.partition_columns:
if col in partition_values:
parts.append(f"{col}={partition_values[col]}")
return "/".join(parts)
def read_data(
self,
symbol: str,
start_time: float,
end_time: float
) -> List[Dict[str, Any]]:
"""Read data for a symbol and time range."""
# Find relevant files
matching_files = []
for file in self.files.values():
if symbol == file.partition.get("symbol", symbol):
# Check time range (simplified)
if start_time <= file.last_accessed <= end_time:
matching_files.append(file)
# In a real implementation, this would read from storage
return []
def query(
self,
symbol: Optional[str] = None,
date_range: Optional[tuple] = None,
limit: int = 1000
) -> List[Dict[str, Any]]:
"""Query data lake with filters."""
results = []
for file in self.files.values():
# Apply filters
if symbol and file.partition.get("symbol") != symbol:
continue
results.append({
"path": file.path,
"partition": file.partition,
"size_bytes": file.size_bytes,
"row_count": file.row_count
})
if len(results) >= limit:
break
return results
def get_file(self, path: str) -> Optional[DataFile]:
"""Get file metadata."""
return self.files.get(path)
def list_files(
self,
symbol: Optional[str] = None,
tier: Optional[StorageTier] = None
) -> List[DataFile]:
"""List files with optional filters."""
results = []
for file in self.files.values():
if symbol and file.partition.get("symbol") != symbol:
continue
if tier and file.tier != tier:
continue
results.append(file)
return results
def delete_file(self, path: str) -> bool:
"""Delete a file."""
if path in self.files:
del self.files[path]
self._save_metadata()
return True
return False
def get_stats(self) -> Dict[str, Any]:
"""Get data lake statistics."""
total_size = sum(f.size_bytes for f in self.files.values())
total_rows = sum(f.row_count for f in self.files.values())
tier_counts = {}
for f in self.files.values():
tier = f.tier.value
tier_counts[tier] = tier_counts.get(tier, 0) + 1
return {
"total_files": len(self.files),
"total_size_bytes": total_size,
"total_rows": total_rows,
"files_by_tier": tier_counts,
"avg_file_size": total_size / len(self.files) if self.files else 0
}
class DataLifecycleManager:
"""Manages data lifecycle policies."""
def __init__(self, lake: DataLake):
self.lake = lake
self.policies: Dict[str, Dict] = {}
self._initialize_policies()
def _initialize_policies(self):
"""Initialize lifecycle policies."""
self.policies = {
"hot": {"max_age_days": 7, "tier": StorageTier.HOT},
"warm": {"max_age_days": 30, "tier": StorageTier.WARM},
"cold": {"max_age_days": 365, "tier": StorageTier.COLD}
}
def set_policy(self, symbol: str, hot_days: int, warm_days: int):
"""Set lifecycle policy for a symbol."""
self.policies[symbol] = {
"hot_days": hot_days,
"warm_days": warm_days
}
def get_tier_for_file(self, file: DataFile, symbol: str) -> StorageTier:
"""Determine tier for a file based on age."""
if symbol in self.policies:
policy = self.policies[symbol]
age_days = (time.time() - file.created_at) / 86400
if age_days < policy["hot_days"]:
return StorageTier.HOT
elif age_days < policy["warm_days"]:
return StorageTier.WARM
else:
return StorageTier.COLD
# Default policy
age_days = (time.time() - file.created_at) / 86400
if age_days < 7:
return StorageTier.HOT
elif age_days < 30:
return StorageTier.WARM
return StorageTier.COLD
def enforce_policies(self) -> List[Dict]:
"""Enforce lifecycle policies and return tier changes."""
changes = []
for file in self.lake.files.values():
symbol = file.partition.get("symbol")
if not symbol:
continue
new_tier = self.get_tier_for_file(file, symbol)
if new_tier != file.tier:
old_tier = file.tier
file.tier = new_tier
changes.append({
"path": file.path,
"from_tier": old_tier.value,
"to_tier": new_tier.value,
"age_days": (time.time() - file.created_at) / 86400
})
self.lake._save_metadata()
return changes
class DataCatalog:
"""Catalog of data lake contents."""
def __init__(self, lake: DataLake):
self.lake = lake
self.schema_registry: Dict[str, Dict] = {}
def register_schema(self, symbol: str, schema: Dict[str, str]):
"""Register schema for a symbol."""
self.schema_registry[symbol] = {
"schema": schema,
"updated_at": time.time()
}
def get_schema(self, symbol: str) -> Optional[Dict[str, str]]:
"""Get schema for a symbol."""
if symbol in self.schema_registry:
return self.schema_registry[symbol]["schema"]
return None
def search(self, keyword: str) -> List[Dict]:
"""Search catalog by keyword."""
results = []
for file in self.lake.files.values():
symbol = file.partition.get("symbol", "")
if keyword.lower() in symbol.lower() or keyword in file.path.lower():
results.append({
"symbol": symbol,
"path": file.path,
"partition": file.partition,
"size_bytes": file.size_bytes,
"created_at": file.created_at
})
return results
Pattern 2: Parquet Data Lake with Schema Evolution and Partitioning
from __future__ import annotations
import logging
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional
import pandas as pd
logger = logging.getLogger(__name__)
class DataLakeWriter:
"""Writes trading data to a partitioned Parquet data lake with schema management."""
def __init__(self, base_path: str = "/data/lake/trading"):
self.base_path = Path(base_path)
self._schema_version: dict[str, int] = {}
def write_candles(
self,
df: pd.DataFrame,
symbol: str,
timeframe: str = "1h",
schema_version: int = 1,
) -> dict:
"""Write OHLCV data to the data lake in partitioned Parquet format.
Partitions by date (year/month/day) for efficient time-range queries.
Args:
df: DataFrame with columns: timestamp, open, high, low, close, volume.
symbol: Trading pair identifier.
timeframe: Data granularity used for partition naming.
schema_version: Current schema version number.
Returns:
Summary dict with file paths written and row counts.
"""
df = df.copy()
df["timestamp"] = pd.to_datetime(df["timestamp"], utc=True)
df["symbol"] = symbol
df["timeframe"] = timeframe
# Create date partitions
df["date"] = df["timestamp"].dt.date
result = {"files_written": [], "total_rows": 0}
for date_val, group in df.groupby("date"):
partition_path = self.base_path / symbol / str(timeframe) / f"year={date_val.year}" / f"month={date_val.month:02d}" / f"day={date_val.day:02d}"
partition_path.mkdir(parents=True, exist_ok=True)
file_name = partition_path / f"candles_{symbol}_{timeframe}_{date_val.isoformat()}.parquet"
group[[
"timestamp", "open", "high", "low", "close", "volume",
"symbol", "timeframe", "date",
]].to_parquet(file_name, engine="pyarrow", index=False)
result["files_written"].append(str(file_name))
result["total_rows"] += len(group)
self._schema_version[symbol] = schema_version
logger.info("Wrote %d candle rows for %s to data lake (v%d)",
result["total_rows"], symbol, schema_version)
return result
def read_range(
self,
symbol: str,
start: datetime,
end: datetime,
timeframe: str = "1h",
) -> pd.DataFrame:
"""Read candle data from the data lake for a specific time range.
Uses partition pruning to scan only relevant date directories.
"""
files = []
current = start.date()
end_date = end.date()
while current <= end_date:
search_path = (
self.base_path / symbol / timeframe /
f"year={current.year}" / f"month={current.month:02d}" / f"day={current.day:02d}"
)
if search_path.exists():
for pf in search_path.glob(f"candles_{symbol}_{timeframe}_*.parquet"):
files.append(str(pf))
current += __import__("datetime").timedelta(days=1)
if not files:
logger.warning("No data found for %s in range %s → %s", symbol, start, end)
return pd.DataFrame()
df = pd.concat([pd.read_parquet(f) for f in files], ignore_index=True)
mask = (df["timestamp"] >= start) & (df["timestamp"] < end)
return df.loc[mask].sort_values("timestamp").reset_index(drop=True)
Constraints
MUST DO
- Validate all incoming data against schema constraints (type, range, nullability) before processing or storage
- Implement idempotent operations: re-processing the same data must produce identical results
- Track data lineage and provenance with timestamps, source identifiers, and transformation history for every record
- Handle out-of-order data by implementing a watermark-based ordering mechanism with configurable tolerance window
- Log data quality metrics (completeness, freshness, accuracy) per source with automatic alerting on degradation
MUST NOT DO
- Do not silently drop records that fail validation — log them to a quarantine table for review
- Avoid concatenating strings for timestamp comparison; use proper datetime/timedelta objects
- Never assume data arrives in chronological order from any external feed without explicit ordering guarantees
- Do not store raw and processed data in the same table without clear partitioning or separation strategy
- Avoid blocking on slow data sources — implement async prefetch with timeout-based fallback to cached data
Live References
Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content.