paulpas Role: Add contextual information to raw data for better decision making
Philosophy: Raw data lacks context; enrichment transforms numbers into insights for smarter trading
Key Principles
- Reference Data Integration: Add symbol metadata, corporate actions, sector info
- Market Context: Incorporate index data, volatility indexes, macro indicators
- Event-Driven Enrichment: Add news, earnings, and corporate event data
- Derived Features: Calculate ratios, correlations, and relative strength
- Data Quality Scoring: Tag enriched data with confidence levels
Implementation Guidelines
Structure
- Core logic: enrichment/enricher.py
- Context providers: enrichment/context_providers.py
- Tests: tests/test_data_enrichment.py
Patterns to Follow
- Use dependency injection for context providers
- Implement enrichment pipelines for complex transformations
- Support asynchronous enrichment for external sources
- Track enrichment quality metrics
Adherence Checklist
Before completing your task, verify:
- Enrichment sources are configurable
- Data quality scores are calculated
- Enrichment latency is monitored
- Fallback values are provided when context unavailable
- Enrichment history is tracked
Relative paths in this skill (e.g., scripts/, reference/) are relative to this base directory.
Python Implementation
import time
from typing import Dict, List, Optional, Any, Callable
from dataclasses import dataclass, field
from enum import Enum
import logging
class EnrichmentType(Enum):
REFERENCE = "reference" # Symbol metadata, sector, etc.
MARKET = "market" # Index data, volatility
EVENT = "event" # Earnings, news, splits
DERIVED = "derived" # Ratios, correlations
@dataclass
class EnrichmentSource:
"""Configuration for an enrichment source."""
name: str
source_type: EnrichmentType
enabled: bool = True
timeout_seconds: float = 5.0
@dataclass
class EnrichmentResult:
"""Result of data enrichment."""
original_data: Dict[str, Any]
enriched_data: Dict[str, Any]
quality_score: float = 1.0 # 0-1
enrichment_log: List[Dict] = field(default_factory=list)
class EnrichmentPipeline:
"""Pipeline for applying multiple enrichments to data."""
def __init__(self):
self.sources: Dict[str, EnrichmentSource] = {}
self.providers: Dict[str, Callable] = {}
self._initialize_default_sources()
def _initialize_default_sources(self):
"""Initialize default enrichment sources."""
self.register_source(EnrichmentSource(
name="symbol_metadata",
source_type=EnrichmentType.REFERENCE
))
self.register_source(EnrichmentSource(
name="market_indices",
source_type=EnrichmentType.MARKET
))
self.register_source(EnrichmentSource(
name="news_events",
source_type=EnrichmentType.EVENT
))
self.register_source(EnrichmentSource(
name="technical_ratios",
source_type=EnrichmentType.DERIVED
))
def register_source(self, source: EnrichmentSource):
"""Register an enrichment source."""
self.sources[source.name] = source
def register_provider(self, source_name: str, provider_func: Callable):
"""Register enrichment provider function."""
self.providers[source_name] = provider_func
def enrich(
self,
data: Dict[str, Any],
sources: Optional[List[str]] = None
) -> EnrichmentResult:
"""Apply enrichments to data."""
result = EnrichmentResult(
original_data=data.copy(),
enriched_data=data.copy()
)
sources_to_apply = sources or list(self.sources.keys())
total_sources = len(sources_to_apply)
successful = 0
for source_name in sources_to_apply:
if source_name not in self.sources:
continue
source = self.sources[source_name]
if not source.enabled:
result.enrichment_log.append({
"source": source_name,
"status": "skipped",
"reason": "disabled"
})
continue
if source_name not in self.providers:
result.enrichment_log.append({
"source": source_name,
"status": "failed",
"reason": "no provider registered"
})
continue
try:
enrichment_start = time.time()
enrichment = self.providers[source_name](data, source_name)
enrichment_duration = time.time() - enrichment_start
if enrichment is not None:
result.enriched_data.update(enrichment)
result.enrichment_log.append({
"source": source_name,
"status": "success",
"duration_seconds": enrichment_duration,
"keys_added": len(enrichment)
})
successful += 1
else:
result.enrichment_log.append({
"source": source_name,
"status": "skipped",
"reason": "provider returned None"
})
except Exception as e:
result.enrichment_log.append({
"source": source_name,
"status": "failed",
"error": str(e)
})
# Calculate quality score
if total_sources > 0:
result.quality_score = successful / total_sources
return result
def enrich_batch(
self,
data_list: List[Dict[str, Any]],
sources: Optional[List[str]] = None
) -> List[EnrichmentResult]:
"""Enrich a batch of data items."""
return [self.enrich(data, sources) for data in data_list]
class SymbolEnricher:
"""Enriches data with symbol-specific information."""
def __init__(self):
self.symbols: Dict[str, Dict[str, Any]] = {}
self._initialize_symbols()
def _initialize_symbols(self):
"""Initialize default symbol metadata."""
self.symbols.update({
"AAPL": {
"name": "Apple Inc.",
"sector": "Technology",
"industry": "Consumer Electronics",
"currency": "USD",
"market_cap": 2500000000000,
"country": "USA",
"is_etf": False,
"leverage": 1.0
},
"SPY": {
"name": "SPDR S&P 500 ETF",
"sector": "ETF",
"industry": "Equity ETF",
"currency": "USD",
"market_cap": 400000000000,
"country": "USA",
"is_etf": True,
"leverage": 1.0
}
})
def register_symbol(self, symbol: str, metadata: Dict[str, Any]):
"""Register metadata for a symbol."""
self.symbols[symbol] = metadata
def enrich_symbol(
self,
data: Dict[str, Any],
source_name: str
) -> Optional[Dict[str, Any]]:
"""Enrich data with symbol metadata."""
symbol = data.get("symbol") or data.get("symbol_name")
if not symbol:
return None
if symbol not in self.symbols:
return None
metadata = self.symbols[symbol]
enrichment = {}
for key, value in metadata.items():
enrichment[f"{source_name}.{key}"] = value
# Add derived fields
if "is_etf" in metadata:
enrichment[f"{source_name}.is_equity"] = not metadata["is_etf"]
if "sector" in metadata:
enrichment[f"{source_name}.sector_group"] = self._get_sector_group(metadata["sector"])
return enrichment
def _get_sector_group(self, sector: str) -> str:
"""Get sector group from sector."""
sector_groups = {
"Technology": "Technology",
"Communication": "Communication",
"Consumer Discretionary": "Consumer",
"Consumer Staples": "Consumer",
"Healthcare": "Healthcare",
"Financials": "Financial",
"Industrials": "Industrials",
"Utilities": "Utilities",
"Energy": "Energy",
"Materials": "Materials",
"Real Estate": "Real Estate"
}
return sector_groups.get(sector, "Other")
class MarketContextEnricher:
"""Enriches data with market context."""
def __init__(self):
self.indices: Dict[str, Dict[str, float]] = {}
self._initialize_indices()
def _initialize_indices(self):
"""Initialize market indices."""
self.indices.update({
"SPY": {"price": 450.0, "volatility_20d": 0.015, "volume": 100000000},
"QQQ": {"price": 350.0, "volatility_20d": 0.020, "volume": 80000000},
"VIX": {"price": 15.0, "description": "Low volatility"}
})
def enrich_market_context(
self,
data: Dict[str, Any],
source_name: str
) -> Optional[Dict[str, Any]]:
"""Enrich data with market context."""
symbol = data.get("symbol")
if not symbol:
return None
enrichment = {}
# Calculate relative strength to market
if "close" in data and "SPY" in self.indices:
spy_close = self.indices["SPY"]["price"]
market_return = (spy_close - 440) / 440 if spy_close > 0 else 0
stock_return = (data["close"] - data.get("open", data["close"])) / data.get("open", data["close"])
enrichment[f"{source_name}.relative_strength"] = stock_return - market_return
# Add index correlation estimates
for index_name, index_data in self.indices.items():
enrichment[f"{source_name}.{index_name}_price"] = index_data["price"]
enrichment[f"{source_name}.{index_name}_volatility"] = index_data["volatility_20d"]
# Market state
if "VIX" in self.indices:
vix = self.indices["VIX"]["price"]
enrichment[f"{source_name}.market_state"] = self._get_market_state(vix)
return enrichment
def _get_market_state(self, vix: float) -> str:
"""Determine market state based on VIX."""
if vix > 30:
return "high_volatility"
elif vix > 20:
return "moderate_volatility"
else:
return "low_volatility"
class EventEnricher:
"""Enriches data with event information."""
def __init__(self):
self.events: Dict[str, List[Dict]] = {}
def register_event(self, symbol: str, event: Dict[str, Any]):
"""Register an event for a symbol."""
if symbol not in self.events:
self.events[symbol] = []
self.events[symbol].append(event)
def enrich_events(
self,
data: Dict[str, Any],
source_name: str
) -> Optional[Dict[str, Any]]:
"""Enrich data with event context."""
symbol = data.get("symbol")
if not symbol or symbol not in self.events:
return None
events = self.events[symbol]
# Find relevant events
enrichment = {
f"{source_name}.event_count": len(events),
f"{source_name}.has_upcoming_events": len(events) > 0
}
# Add event details
for i, event in enumerate(events):
enrichment[f"{source_name}.event_{i}_type"] = event.get("type", "unknown")
enrichment[f"{source_name}.event_{i}_date"] = event.get("date", "")
return enrichment
class EnrichmentQualityMonitor:
"""Monitors enrichment quality."""
def __init__(self):
self.metrics: Dict[str, Dict] = {}
self._lock = None
def record_enrichment(self, result: EnrichmentResult):
"""Record enrichment result."""
for log in result.enrichment_log:
source = log["source"]
status = log["status"]
if source not in self.metrics:
self.metrics[source] = {"success": 0, "failed": 0, "skipped": 0}
if status == "success":
self.metrics[source]["success"] += 1
elif status == "failed":
self.metrics[source]["failed"] += 1
else:
self.metrics[source]["skipped"] += 1
def get_quality_report(self) -> Dict[str, Dict]:
"""Get enrichment quality report."""
report = {}
for source, counts in self.metrics.items():
total = counts["success"] + counts["failed"] + counts["skipped"]
report[source] = {
"success_rate": counts["success"] / total if total > 0 else 0,
"success_count": counts["success"],
"failed_count": counts["failed"],
"skipped_count": counts["skipped"],
"total": total
}
return report
Pattern 2: Feature Engineering Pipeline with Validation
from __future__ import annotations
import logging
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Optional
logger = logging.getLogger(__name__)
@dataclass(frozen=True)
class EnrichedCandle:
"""A candle enriched with computed technical features."""
timestamp: datetime
open_price: float
high_price: float
low_price: float
close_price: float
volume: float
# Computed features
sma_20: Optional[float] = None
rsi_14: Optional[float] = None
atr_14: Optional[float] = None
macd_signal: Optional[str] = None
vwap: Optional[float] = None
@property
def has_all_features(self) -> bool:
return all([
self.sma_20 is not None,
self.rsi_14 is not None,
self.atr_14 is not None,
self.vwap is not None,
])
class FeatureEnricher:
"""Computes technical features and enriches raw candle data."""
def __init__(self, window_sizes: dict[str, int] | None = None):
self._windows = window_sizes or {"sma": 20, "rsi": 14, "atr": 14}
def enrich_batch(
self, candles: list[dict],
) -> list[EnrichedCandle]:
"""Apply feature engineering to a batch of raw candle dicts.
Args:
candles: List of raw OHLCV dicts with keys: timestamp, open, high, low, close, volume.
Returns:
List of EnrichedCandle objects with computed technical features.
"""
if not candles:
return []
prices = [c["close"] for c in candles]
highs = [c["high"] for c in candles]
lows = [c["low"] for c in candles]
volumes = [c["volume"] for c in candles]
sma_values = self._compute_sma(prices, self._windows["sma"])
rsi_values = self._compute_rsi(prices, self._windows["rsi"])
atr_values = self._compute_atr(highs, lows, prices, self._windows["atr"])
vwap_values = self._compute_vwap(prices, volumes)
enriched: list[EnrichedCandle] = []
for i, candle in enumerate(candles):
ec = EnrichedCandle(
timestamp=datetime.fromisoformat(candle["timestamp"]),
open_price=candle["open"],
high_price=candle["high"],
low_price=candle["low"],
close_price=candle["close"],
volume=candle["volume"],
sma_20=sma_values[i],
rsi_14=rsi_values[i],
atr_14=atr_values[i],
vwap=vwap_values[i],
)
enriched.append(ec)
logger.info("Enriched %d candles with %d features", len(enriched), 4)
return enriched
@staticmethod
def _compute_sma(closes: list[float], window: int) -> list[Optional[float]]:
result: list[Optional[float]] = [None] * window
for i in range(window, len(closes)):
avg = sum(closes[i - window:i]) / window
result.append(round(avg, 6))
return result
@staticmethod
def _compute_rsi(closes: list[float], period: int) -> list[Optional[float]]:
result: list[Optional[float]] = [None] * period
if len(closes) < period + 1:
return result
deltas = [closes[i] - closes[i - 1] for i in range(1, len(closes))]
gains = [d if d > 0 else 0 for d in deltas]
losses = [-d if d < 0 else 0 for d in deltas]
avg_gain = sum(gains[:period]) / period
avg_loss = sum(losses[:period]) / period
if avg_loss == 0:
result.append(100.0)
else:
rs = avg_gain / avg_loss
result.append(round(100 - (100 / (1 + rs)), 4))
for i in range(period, len(deltas)):
avg_gain = (avg_gain * (period - 1) + gains[i]) / period
avg_loss = (avg_loss * (period - 1) + losses[i]) / period
if avg_loss == 0:
result.append(100.0)
else:
rs = avg_gain / avg_loss
result.append(round(100 - (100 / (1 + rs)), 4))
return result
@staticmethod
def _compute_atr(highs, lows, closes, period):
result = [None] * period
if len(highs) < period + 1:
return result
trs = []
for i in range(1, len(highs)):
tr = max(highs[i] - lows[i], abs(highs[i] - closes[i - 1]), abs(lows[i] - closes[i - 1]))
trs.append(tr)
avg_tr = sum(trs[:period]) / period
result.append(round(avg_tr, 6))
for i in range(period, len(trs)):
avg_tr = (avg_tr * (period - 1) + trs[i]) / period
result.append(round(avg_tr, 6))
return result
@staticmethod
def _compute_vwap(prices, volumes):
result: list[Optional[float]] = []
cum_pv = 0.0
cum_vol = 0.0
for p, v in zip(prices, volumes):
cum_pv += p * v
cum_vol += v
result.append(round(cum_pv / cum_vol, 6) if cum_vol > 0 else None)
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.