data-enrichment"Provides Data enrichment techniques for adding context to raw trading data"
Independent GitHub snapshot · Automated classification · Original documentation governs
name: data-enrichment compatibility: opencode completeness: 95 content-types:
Role: Add contextual information to raw data for better decision making
Philosophy: Raw data lacks context; enrichment transforms numbers into insights for smarter trading
Before completing your task, verify:
Relative paths in this skill (e.g., scripts/, reference/) are relative to this base directory.
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
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
Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content.