paulpas Role: Store and retrieve engineered features for consistent model training and inference
Philosophy: Features are the foundation of ML models; feature store ensures reproducibility and consistency across training and production
Key Principles
- Feature Versioning: Version features to track changes and enable rollback
- Feature Lineage: Track feature origins and transformations
- Offline vs Online Store: Separate storage for training and real-time inference
- Feature Discovery: Searchable catalog of available features
- Consistency Checks: Validate feature consistency across store types
Implementation Guidelines
Structure
- Core logic: features/feature_store.py
- Offline store: features/offline_store.py
- Online store: features/online_store.py
- Tests: tests/test_feature_store.py
Patterns to Follow
- Use DuckDB or Parquet for offline storage
- Use Redis or Memcached for online storage
- Implement feature groups for batch operations
- Support time-travel queries
Adherence Checklist
Before completing your task, verify:
- Feature versioning is enforced
- Lineage tracking is maintained
- Offline and online stores are synchronized
- Feature queries include metadata
- Consistency checks run on sync
Relative paths in this skill (e.g., scripts/, reference/) are relative to this base directory.
Python Implementation
import time
import uuid
from typing import Dict, List, Optional, Any, Tuple
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime
import hashlib
import logging
class FeatureType(Enum):
NUMERIC = "numeric"
CATEGORICAL = "categorical"
BOOLEAN = "boolean"
TEXT = "text"
EMBEDDING = "embedding"
@dataclass
class FeatureMetadata:
"""Metadata for a feature."""
name: str
feature_type: FeatureType
description: str = ""
default_value: Any = None
valid_range: Tuple[Any, Any] = None
categories: List[str] = None
version: int = 1
@dataclass
class FeatureVersion:
"""Version of a feature."""
version: int
created_at: float
definition: Dict[str, Any]
metadata: FeatureMetadata
checksum: str
@dataclass
class FeatureGroup:
"""Group of related features."""
name: str
description: str
features: List[str]
version: int = 1
created_at: float = field(default_factory=time.time)
class FeatureStore:
"""Manages features for ML models."""
def __init__(self):
self.features: Dict[str, List[FeatureVersion]] = {}
self.feature_groups: Dict[str, FeatureGroup] = {}
self.metadata: Dict[str, FeatureMetadata] = {}
self._initialize_default_features()
def _initialize_default_features(self):
"""Initialize default feature definitions."""
# Price-based features
self.register_feature(
"price_return_1h",
FeatureMetadata(
name="price_return_1h",
feature_type=FeatureType.NUMERIC,
description="1-hour price return",
valid_range=(-1.0, 1.0)
)
)
self.register_feature(
"price_volatility_24h",
FeatureMetadata(
name="price_volatility_24h",
feature_type=FeatureType.NUMERIC,
description="24-hour price volatility",
valid_range=(0.0, 0.5)
)
)
# Volume-based features
self.register_feature(
"volume_zscore_1h",
FeatureMetadata(
name="volume_zscore_1h",
feature_type=FeatureType.NUMERIC,
description="1-hour volume z-score",
valid_range=(-10.0, 10.0)
)
)
def register_feature(self, name: str, metadata: FeatureMetadata):
"""Register a new feature."""
if name not in self.metadata:
self.metadata[name] = metadata
self.features[name] = []
# Create initial version
version_data = {
"metadata": metadata.__dict__,
"created_at": time.time()
}
checksum = hashlib.md5(
str(version_data).encode()
).hexdigest()
version = FeatureVersion(
version=1,
created_at=time.time(),
definition=version_data,
metadata=metadata,
checksum=checksum
)
self.features[name].append(version)
def register_feature_group(self, group: FeatureGroup):
"""Register a feature group."""
self.feature_groups[group.name] = group
def get_feature(self, name: str, version: int = None) -> Optional[FeatureVersion]:
"""Get feature by name and optional version."""
if name not in self.features:
return None
versions = self.features[name]
if version is None:
return versions[-1] # Latest version
for v in reversed(versions):
if v.version == version:
return v
return None
def get_feature_metadata(self, name: str) -> Optional[FeatureMetadata]:
"""Get feature metadata."""
return self.metadata.get(name)
def get_features_by_group(self, group_name: str) -> List[str]:
"""Get all features in a group."""
group = self.feature_groups.get(group_name)
if group:
return group.features
return []
def search_features(self, keyword: str) -> List[FeatureMetadata]:
"""Search features by keyword."""
results = []
for name, metadata in self.metadata.items():
if keyword.lower() in name.lower() or keyword.lower() in metadata.description.lower():
results.append(metadata)
return results
def log_feature_values(
self,
symbol: str,
timestamp: float,
features: Dict[str, Any]
):
"""Log feature values for a timestamp."""
# This would persist to storage in a real implementation
logging.debug(
f"Logged features for {symbol} at {datetime.fromtimestamp(timestamp)}"
)
def get_feature_values(
self,
symbol: str,
feature_names: List[str],
start_time: float,
end_time: float
) -> Optional[Dict[str, List[Any]]]:
"""Retrieve feature values for a time range."""
# This would query storage in a real implementation
return None # Placeholder
def get_current_features(
self,
symbol: str,
feature_names: List[str]
) -> Dict[str, Any]:
"""Get most recent feature values."""
# This would query online store in a real implementation
return {name: 0.0 for name in feature_names}
class FeatureValidator:
"""Validates feature values."""
def __init__(self, store: FeatureStore):
self.store = store
def validate_feature_value(
self,
name: str,
value: Any
) -> Tuple[bool, Optional[str]]:
"""Validate a feature value."""
metadata = self.store.get_feature_metadata(name)
if not metadata:
return False, f"Unknown feature: {name}"
# Type check
if not self._check_type(value, metadata.feature_type):
return False, f"Invalid type for {name}"
# Range check
if metadata.valid_range:
min_val, max_val = metadata.valid_range
if value < min_val or value > max_val:
return False, f"Value {value} out of range [{min_val}, {max_val}]"
# Categorical check
if metadata.categories and value not in metadata.categories:
return False, f"Value {value} not in categories {metadata.categories}"
return True, None
def _check_type(self, value: Any, feature_type: FeatureType) -> bool:
"""Check if value matches feature type."""
type_map = {
FeatureType.NUMERIC: (int, float),
FeatureType.CATEGORICAL: str,
FeatureType.BOOLEAN: bool,
FeatureType.TEXT: str,
FeatureType.EMBEDDING: list
}
valid_types = type_map.get(feature_type, (int, float, str))
return isinstance(value, valid_types)
def validate_batch(
self,
feature_values: Dict[str, Any]
) -> List[Tuple[str, Optional[str]]]:
"""Validate multiple feature values."""
results = []
for name, value in feature_values.items():
valid, error = self.validate_feature_value(name, value)
results.append((name, error))
return results
class FeatureCache:
"""Cache for frequently accessed features."""
def __init__(self, max_size: int = 10000, ttl: float = 300.0):
self.max_size = max_size
self.ttl = ttl
self._cache: Dict[str, Tuple[Any, float]] = {}
self._lock = None # Would use threading.Lock in practice
def get(self, key: str) -> Optional[Any]:
"""Get cached feature value."""
if key in self._cache:
value, timestamp = self._cache[key]
if time.time() - timestamp < self.ttl:
return value
del self._cache[key]
return None
def set(self, key: str, value: Any):
"""Set cached feature value."""
if len(self._cache) >= self.max_size:
# Remove oldest entry
oldest_key = next(iter(self._cache))
del self._cache[oldest_key]
self._cache[key] = (value, time.time())
def delete(self, key: str):
"""Delete feature from cache."""
if key in self._cache:
del self._cache[key]
Pattern 2: Feature Store with Online/Offline Consistency
from __future__ import annotations
import hashlib
import json
import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Optional
logger = logging.getLogger(__name__)
@dataclass(frozen=True)
class FeatureVector:
"""Immutable feature vector with lineage tracking."""
entity_id: str
entity_type: str
features: dict[str, float]
version: int
generated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
@property
def fingerprint(self) -> str:
"""Deterministic hash for cache invalidation and consistency checks."""
content = f"{self.entity_id}:{self.version}:{json.dumps(self.features, sort_keys=True)}"
return hashlib.sha256(content.encode()).hexdigest()[:16]
class FeatureStore:
"""Feature store with both offline (batch) and online (low-latency) serving paths."""
def __init__(self, offline_backend, online_backend):
self._offline = offline_backend # e.g., Parquet files, S3, data warehouse
self._online = online_backend # e.g., Redis, in-memory cache
def compute_and_write(
self,
entity_id: str,
entity_type: str,
features: dict[str, float],
version: int = 1,
) -> FeatureVector:
"""Compute a new feature vector and write to both offline and online stores.
Guarantees that the same feature computation produces identical results
regardless of when or where it's called (deterministic + idempotent).
Args:
entity_id: Unique identifier for the entity (e.g., symbol, user).
entity_type: Entity category (e.g., "symbol", "user_profile").
features: Dict of feature name → float value.
version: Feature vector version number.
Returns:
The created FeatureVector with computed fingerprint.
"""
fv = FeatureVector(
entity_id=entity_id,
entity_type=entity_type,
features={k: round(v, 6) for k, v in features.items()},
version=version,
)
# Write to offline store (batch path — data warehouse / parquet)
self._offline.write(fv.entity_type, fv.entity_id, {
"features": fv.features,
"version": fv.version,
"generated_at": fv.generated_at.isoformat(),
"fingerprint": fv.fingerprint,
})
# Write to online store (low-latency path — Redis / DynamoDB)
self._online.set(
key=f"feature:{fv.entity_type}:{fv.entity_id}",
value=json.dumps({
"features": fv.features,
"version": fv.version,
"fingerprint": fv.fingerprint,
}),
ttl=3600, # 1 hour TTL for online cache
)
logger.info("Feature vector written: %s/%s v%d (fp=%s)",
entity_type, entity_id, version, fv.fingerprint)
return fv
def get_latest(self, entity_id: str, entity_type: str) -> Optional[FeatureVector]:
"""Retrieve the latest feature vector from online store.
Falls back to offline store if online cache misses or is stale.
"""
# Try online store first (fast path)
if online_data:
data = json.loads(online_data)
return FeatureVector(
entity_id=entity_id,
entity_type=entity_type,
features=data["features"],
version=data["version"],
)
# Fall back to offline store (slow path)
offline_data = self._offline.read(entity_type, entity_id)
if offline_data:
return FeatureVector(
entity_id=entity_id,
entity_type=entity_type,
features=offline_data["features"],
version=offline_data["version"],
)
logger.debug("No feature vector found for %s/%s", entity_type, entity_id)
return None
def get_batch(self, entity_ids: list[str], entity_type: str) -> dict[str, FeatureVector]:
"""Fetch feature vectors for multiple entities in a single batch call."""
result = {}
for eid in entity_ids:
fv = self.get_latest(eid, entity_type)
if fv is not None:
result[eid] = fv
return result
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.