# AI Sentiment Features > "Provides Extract market sentiment from news, social media, and analyst reports" - Skill: `paulpas/ai-sentiment-features` (Agent Skill) - Install (CLI): `npx skillmds@latest add paulpas/ai-sentiment-features` - Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/ai-sentiment-features/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Coding & Dev Tools - License: MIT - Author: paulpas (https://skillmd.com/u/paulpas) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/paulpas/ai-sentiment-features --- **Role:** Convert sentiment signals into actionable trading features with proper temporal alignment **Philosophy:** Sentiment reflects market psychology but can lag prices. Prioritize real-time processing, sentiment drift detection, and integration with price-based signals. ## Key Principles 1. **Temporal Alignment**: Ensure sentiment matches the correct future period 2. **Sentiment Drift**: Track sentiment changes, not just absolute values 3. **Source Weighting**: Different sources have different predictive power 4. **Contrarian Signals**: Use sentiment extremes as counter-trend signals 5. **Volume Context**: Normalize sentiment by news volume to avoid outliers ## Implementation Guidelines ### Structure - Core logic: `sentiment/extractors.py` - Sentiment extraction pipeline - Integrator: `sentiment/integrator.py` - Feature integration with price data - Analyzers: `sentiment/analyzers.py` - Sentiment pattern detection - Config: `config/sentiment_config.yaml` - Sentiment parameters ### Patterns to Follow - Use sliding window for sentiment aggregation - Normalize by news volume to avoid burst artifacts - Track sentiment momentum (change rate) - Align sentiment with lagged price responses ## Adherence Checklist Before completing your task, verify: - [ ] Sentiment features temporally aligned with price movements - [ ] Sentiment normalized by news volume - [ ] Sentiment momentum/changes tracked - [ ] Contrarian signals detected at extremes - [ ] Source-specific sentiment weighted appropriately ## Code Examples ### Sentiment Feature Extractor ```python import numpy as np import pandas as pd from typing import Dict, List, Tuple from dataclasses import dataclass from collections import defaultdict @dataclass class SentimentRecord: """Single sentiment record for a time period.""" timestamp: int sentiment: float # -1 to 1 source: str volume: int # News volume for this period entities: List[str] class SentimentFeatureExtractor: """Extract sentiment features from news data.""" def __init__(self, window_sizes: List[int] = [1, 3, 5, 10]): self.window_sizes = window_sizes self.sources = defaultdict(list) def extract_from_news(self, news_data: pd.DataFrame) -> List[SentimentRecord]: """Extract sentiment from news articles.""" records = [] for _, row in news_data.iterrows(): timestamp = row['timestamp'] sentiment = self._calculate_sentiment(row['text']) source = row.get('source', 'unknown') volume = row.get('volume', 1) entities = row.get('entities', []) records.append(SentimentRecord( timestamp=timestamp, sentiment=sentiment, source=source, volume=volume, entities=entities )) return records def _calculate_sentiment(self, text: str) -> float: """Calculate sentiment score for text.""" # Simple keyword-based sentiment positive_words = ['good', 'strong', 'surge', 'gain', 'upgrade', 'profit', 'outperform'] negative_words = ['bad', 'weak', 'plunge', 'loss', 'downgrade', 'concern', 'underperform'] text_lower = text.lower() positive_count = sum(1 for word in positive_words if word in text_lower) negative_count = sum(1 for word in negative_words if word in text_lower) total = positive_count + negative_count if total == 0: return 0.0 score = (positive_count - negative_count) / total return np.clip(score, -1.0, 1.0) def aggregate_sentiment(self, records: List[SentimentRecord]) -> Dict[str, np.ndarray]: """Aggregate sentiment across time windows.""" if not records: return {} # Sort by timestamp records = sorted(records, key=lambda x: x.timestamp) timestamps = [r.timestamp for r in records] sentiments = [r.sentiment for r in records] volumes = [r.volume for r in records] # Calculate various aggregations features = { 'sentiment': np.array(sentiments), 'sentiment_weighted': np.array([r.sentiment * r.volume for r in records]) / (np.array(volumes) + 1e-6) } # Window-based aggregations for window in self.window_sizes: rolling_mean = pd.Series(sentiments).rolling(window).mean().fillna(0).values features[f'sentiment_ma_{window}'] = rolling_mean # Momentum if len(rolling_mean) > window: features[f'sentiment_momentum_{window}'] = ( rolling_mean - pd.Series(rolling_mean).shift(window).fillna(0).values ) # Sentiment extremes features['sentiment_extreme'] = np.abs(sentiments) > 0.7 return features def extract_all(self, news_data: pd.DataFrame) -> Dict[str, np.ndarray]: """Extract all sentiment features from news data.""" records = self.extract_from_news(news_data) return self.aggregate_sentiment(records) ``` ### Sentiment-Price Integration ```python import numpy as np import pandas as pd from typing import Dict, List, Tuple class SentimentPriceIntegrator: """Integrate sentiment features with price data.""" def __init__(self, price_lag: int = 1, sentiment_lag: int = 0): self.price_lag = price_lag self.sentiment_lag = sentiment_lag def align_sentiment_prices(self, sentiment_features: Dict[str, np.ndarray], prices: np.ndarray, sentiment_times: np.ndarray, price_times: np.ndarray) -> Dict[str, np.ndarray]: """Align sentiment features with price data.""" # Interpolate sentiment to price timestamps aligned_features = {} for feature_name, sentiment_values in sentiment_features.items(): # Simple nearest neighbor interpolation aligned = np.interp(price_times, sentiment_times, sentiment_values, left=sentiment_values[0], right=sentiment_values[-1]) aligned_features[feature_name] = aligned # Add price-based features aligned_features['price'] = prices aligned_features['returns'] = np.diff(np.log(np.concatenate([[prices[0]], prices]))) return aligned_features def create_lagged_features(self, features: Dict[str, np.ndarray], max_lag: int = 5) -> Dict[str, np.ndarray]: """Create lagged versions of features.""" lagged = {} for name, values in features.items(): values = np.asarray(values) lagged[name] = values for lag in range(1, max_lag + 1): if lag < len(values): lagged[f'{name}_lag{lag}'] = np.concatenate([ [values[0]] * lag, values[:-lag] ]) return lagged def create_derivative_features(self, features: Dict[str, np.ndarray]) -> Dict[str, np.ndarray]: """Create derivative (change) features.""" derivatives = {} for name, values in features.items(): values = np.asarray(values) if name.endswith('_lag'): continue # Skip lagged features derivatives[f'{name}_diff'] = np.diff(values, prepend=values[0]) derivatives[f'{name}_ pct_change'] = ( np.diff(values, prepend=values[0]) / (np.abs(values) + 1e-6) ) return derivatives ``` ### Sentiment Contrarian Signal Detector ```python import numpy as np from typing import Dict, List, Tuple class SentimentContrarianDetector: """Detect contrarian sentiment signals for counter-trend opportunities.""" def __init__(self, extremethreshold: float = 0.8, lookback_period: int = 20, divergence_threshold: float = 0.5): self.threshold = extremethreshold self.lookback = lookback_period self.divergence = divergence_threshold def detect_contrarian_signals(self, sentiment: np.ndarray, returns: np.ndarray) -> List[Dict]: """Detect contrarian signals where sentiment diverges from price.""" signals = [] for i in range(self.lookback, len(sentiment)): # Calculate sentiment extremes recent_sentiment = sentiment[i-self.lookback:i] sent_mean = np.mean(recent_sentiment) sent_std = np.std(recent_sentiment) if len(recent_sentiment) > 1 else 0 # Calculate price returns recent_returns = returns[i-self.lookback:i] return_sum = np.sum(recent_returns) # Normalize sentiment to z-score sent_z = (sentiment[i] - sent_mean) / (sent_std + 1e-6) # Detect contrarian signal is_extreme = np.abs(sent_z) > self.threshold has_divergence = (sent_z > 0 and return_sum < -self.divergence) or \ (sent_z < 0 and return_sum > self.divergence) if is_extreme and has_divergence: direction = 'short' if sent_z > 0 else 'long' signals.append({ 'timestamp': i, 'type': 'contrarian', 'direction': direction, 'strength': float(np.abs(sent_z)), 'sentiment': float(sentiment[i]), 'recent_return': float(return_sum), 'explanation': f'Extreme sentiment ({sentiment[i]:.2f}) against trend' }) return signals def calculate_sentiment_reversal(self, sentiment: np.ndarray, threshold: float = 0.3) -> np.ndarray: """Calculate sentiment reversal potential.""" if len(sentiment) < 2: return np.zeros_like(sentiment) # Calculate sentiment momentum momentum = np.diff(sentiment, prepend=sentiment[0]) # Calculate how far sentiment is from neutral from_neutral = np.abs(sentiment) # Reversal potential: high sentiment + high momentum change reversal_potential = momentum * from_neutral return reversal_potential def sentiment_volume_ratio(self, sentiment: np.ndarray, volume: np.ndarray) -> np.ndarray: """Calculate sentiment relative to news volume.""" # Normalize sentiment by volume volume_normalized = sentiment / (np.log(volume + 1) + 1e-6) return volume_normalized ``` ### Multi-Source Sentiment Aggregator ```python import numpy as np import pandas as pd from typing import Dict, List class MultiSourceSentimentAggregator: """Aggregate sentiment from multiple sources with source weighting.""" def __init__(self, source_weights: Dict[str, float] = None): self.source_weights = source_weights or { 'professional': 1.5, 'news_agency': 1.2, 'social_media': 0.8, 'analyst': 1.3, 'blog': 0.5 } def weighted_sentiment(self, records: List[Dict]) -> np.ndarray: """Calculate weighted average sentiment by source.""" if not records: return np.array([]) timestamps = [r['timestamp'] for r in records] sentiments = [r['sentiment'] for r in records] sources = [r['source'] for r in records] # Get weights for each record weights = np.array([ self.source_weights.get(source, 1.0) for source in sources ]) # Normalize weights weights = weights / np.sum(weights) # Weighted average weighted_sentiment = np.sum(weights * np.array(sentiments)) return weighted_sentiment def aggregate_by_time_window(self, records: List[Dict], window_size: int = 60) -> Dict[str, np.ndarray]: """Aggregate sentiment in time windows.""" if not records: return {} # Sort records records = sorted(records, key=lambda x: x['timestamp']) timestamps = [] aggregated = [] current_window_start = records[0]['timestamp'] window_records = [] for record in records: if record['timestamp'] - current_window_start > window_size: # End current window if window_records: timestamps.append(current_window_start) sent = self.weighted_sentiment(window_records) aggregated.append(sent) # Start new window current_window_start = record['timestamp'] window_records = [record] else: window_records.append(record) # Final window if window_records: timestamps.append(current_window_start) sent = self.weighted_sentiment(window_records) aggregated.append(sent) return { 'timestamp': np.array(timestamps), 'sentiment': np.array(aggregated), 'volume': np.array([len(w) for w in [records[i*window_size:(i+1)*window_size] for i in range(len(records)//window_size)]]) } ``` --- --- ## Constraints ### MUST DO - Validate input feature distributions against training data baselines; flag drift exceeding 2 standard deviations - Implement model versioning with reproducibility tags — every prediction must be traceable to the exact model artifact and config - Include confidence intervals or probability estimates alongside all point predictions, never return raw scores without context - Log all model inputs, outputs, and metadata to enable post-hoc analysis of prediction failures - Implement feature computation consistently between training and inference — use the same transformation pipeline for both ### MUST NOT DO - Do not train models on look-ahead biased features (e.g., using future prices or events in training data) - Avoid deploying a new model version without shadow-testing against the current production model first - Never retrain a model on a data window that includes regime changes without explicit regime-aware validation - Do not use accuracy as the primary metric for imbalanced datasets — use precision/recall, F1, or AUC-ROC - Avoid hardcoding feature names; load them from a schema or config file to prevent mismatches between training and inference ## 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. - [NLTK Book Chapter 1 - Processing Text](https://www.nltk.org/book/ch01.html) - [VADER Sentiment Analysis](https://www.nltk.org/howto/sentiment.html) - [Financial Sentiment Analysis with BERT](https://arxiv.org/abs/1908.10063) - [Sentiment Feature Engineering Guide](https://machinelearningmastery.com/natural-language-processing-for-finance/) - [Text Classification for Market Sentiment](https://scikit-learn.org/stable/modules/feature_extraction.html#text-feature-extraction)