Stats
Actions
Tags
'%20stop-opacity%3D'0.16'%2F%3E%3Cstop%20offset%3D'1'%20stop-color%3D'rgb(200%2C90%2C60)'%20stop-opacity%3D'0.03'%2F%3E%3C%2FlinearGradient%3E%3C%2Fdefs%3E%3Crect%20width%3D'320'%20height%3D'200'%20fill%3D'url(%23g)'%2F%3E%3Ccircle%20cx%3D'250'%20cy%3D'56'%20r%3D'92'%20fill%3D'rgb(200%2C90%2C60)'%20fill-opacity%3D'0.06'%2F%3E%3Ccircle%20cx%3D'64'%20cy%3D'172'%20r%3D'58'%20fill%3D'rgb(200%2C90%2C60)'%20fill-opacity%3D'0.05'%2F%3E%3C%2Fsvg%3E)
From market-sentiment-analyzer
Analyzes market sentiment from social media, news, and on-chain metrics to produce a weighted Fear & Greed Index and trading signals.
How this command is triggered — by the user, by Claude, or both
Slash command
/market-sentiment-analyzer:analyze-sentimentThe summary Claude sees in its command listing — used to decide when to auto-load this command
# Analyze Market Sentiment Comprehensive multi-source sentiment analysis system that aggregates data from social media platforms, news outlets, on-chain metrics, derivatives markets, and whale tracking to predict market movements and identify trading opportunities. ## Overview Market sentiment analysis combines quantitative on-chain data with qualitative social signals to measure the collective mood of market participants. This command provides a sophisticated sentiment aggregation framework that: - **Monitors 6+ data sources** including Twitter/X, Reddit, Telegram, Discord, news feeds,...
Comprehensive multi-source sentiment analysis system that aggregates data from social media platforms, news outlets, on-chain metrics, derivatives markets, and whale tracking to predict market movements and identify trading opportunities.
Market sentiment analysis combines quantitative on-chain data with qualitative social signals to measure the collective mood of market participants. This command provides a sophisticated sentiment aggregation framework that:
The analyzer uses natural language processing (NLP), time-series analysis, and machine learning to identify sentiment extremes that often precede major price reversals, making it invaluable for contrarian trading strategies.
Twitter/X Analysis (15%)
Reddit Sentiment (10%)
Telegram & Discord (10%)
Mainstream Media Coverage
Regulatory News Impact
Funding Rates Analysis (12%)
Options Market Sentiment (13%)
Large Holder Movements
Exchange Flow Analysis
Network Value Metrics
The Fear & Greed Index aggregates multiple data sources into a single 0-100 score:
Score Interpretation:
Calculation Methodology:
Social Sentiment Component (35 points)
Market Momentum Component (20 points)
Derivatives Component (25 points)
Whale Behavior Component (15 points)
On-Chain Component (5 points)
#!/usr/bin/env python3
"""
Comprehensive Multi-Source Crypto Market Sentiment Analyzer
Aggregates sentiment from social media, news, derivatives, and on-chain data
"""
import asyncio
import aiohttp
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import Dict, List, Tuple, Optional
from dataclasses import dataclass, field
from enum import Enum
import statistics
import json
import re
from collections import defaultdict, deque
import tweepy
import praw
from telethon import TelegramClient
from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer
from transformers import pipeline
import ccxt.async_support as ccxt
from web3 import Web3
import redis
import logging
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class SentimentLevel(Enum):
"""Sentiment level classification"""
EXTREME_FEAR = "Extreme Fear"
FEAR = "Fear"
NEUTRAL = "Neutral"
GREED = "Greed"
EXTREME_GREED = "Extreme Greed"
@dataclass
class SentimentScore:
"""Individual sentiment score with metadata"""
source: str
score: float # -1.0 to 1.0
confidence: float # 0.0 to 1.0
weight: float
timestamp: datetime
sample_size: int
metadata: Dict = field(default_factory=dict)
@dataclass
class FearGreedIndex:
"""Fear & Greed Index result"""
score: int # 0-100
level: SentimentLevel
components: Dict[str, float]
historical_percentile: float
trend: str # 'increasing', 'decreasing', 'stable'
timestamp: datetime
@dataclass
class SentimentAlert:
"""Sentiment-based trading alert"""
alert_type: str
severity: str # 'low', 'medium', 'high', 'critical'
message: str
sentiment_score: float
trigger_condition: str
timestamp: datetime
class SocialMediaAnalyzer:
"""Analyze sentiment from social media platforms"""
def __init__(self, config: Dict):
self.config = config
self.vader = SentimentIntensityAnalyzer()
self.finbert = pipeline(
"sentiment-analysis",
model="ProsusAI/finbert",
device=-1 # CPU
)
# Initialize API clients
self.twitter_client = self._init_twitter()
self.reddit_client = self._init_reddit()
self.telegram_client = self._init_telegram()
# Sentiment cache
self.cache = redis.Redis(
host='localhost',
port=6379,
db=0,
decode_responses=True
)
def _init_twitter(self) -> Optional[tweepy.Client]:
"""Initialize Twitter API client"""
try:
return tweepy.Client(
bearer_token=self.config.get('twitter_bearer_token'),
wait_on_rate_limit=True
)
except Exception as e:
logger.error(f"Twitter initialization failed: {e}")
return None
def _init_reddit(self) -> Optional[praw.Reddit]:
"""Initialize Reddit API client"""
try:
return praw.Reddit(
client_id=self.config.get('reddit_client_id'),
client_secret=self.config.get('reddit_client_secret'),
user_agent='CryptoSentimentAnalyzer/1.0'
)
except Exception as e:
logger.error(f"Reddit initialization failed: {e}")
return None
def _init_telegram(self) -> Optional[TelegramClient]:
"""Initialize Telegram client"""
try:
return TelegramClient(
'sentiment_session',
self.config.get('telegram_api_id'),
self.config.get('telegram_api_hash')
)
except Exception as e:
logger.error(f"Telegram initialization failed: {e}")
return None
async def analyze_twitter(self, symbol: str, lookback_hours: int = 24) -> SentimentScore:
"""Analyze Twitter sentiment for cryptocurrency"""
if not self.twitter_client:
return self._empty_sentiment_score('twitter')
try:
# Search for tweets
query = f"${symbol} OR #{symbol} -is:retweet lang:en"
tweets = self.twitter_client.search_recent_tweets(
query=query,
max_results=100,
tweet_fields=['created_at', 'public_metrics', 'author_id']
)
if not tweets.data:
return self._empty_sentiment_score('twitter')
sentiments = []
total_engagement = 0
influential_count = 0
for tweet in tweets.data:
# Get engagement metrics
metrics = tweet.public_metrics
engagement = (
metrics['like_count'] +
metrics['retweet_count'] * 2 +
metrics['reply_count']
)
# Calculate sentiment with VADER
vader_score = self.vader.polarity_scores(tweet.text)
# Calculate FinBERT sentiment for financial context
try:
finbert_result = self.finbert(tweet.text[:512])[0]
finbert_score = self._finbert_to_score(finbert_result)
except Exception:
finbert_score = vader_score['compound']
# Weighted average of VADER and FinBERT
combined_score = 0.6 * finbert_score + 0.4 * vader_score['compound']
# Weight by engagement
sentiments.append({
'score': combined_score,
'weight': max(1, engagement)
})
total_engagement += engagement
# Track influential tweets (high engagement)
if engagement > 100:
influential_count += 1
# Calculate weighted average sentiment
weighted_sum = sum(s['score'] * s['weight'] for s in sentiments)
total_weight = sum(s['weight'] for s in sentiments)
avg_sentiment = weighted_sum / total_weight if total_weight > 0 else 0.0
# Calculate confidence based on sample size and agreement
sentiment_values = [s['score'] for s in sentiments]
std_dev = statistics.stdev(sentiment_values) if len(sentiment_values) > 1 else 1.0
confidence = min(1.0, (len(sentiments) / 100) * (1.0 - std_dev))
return SentimentScore(
source='twitter',
score=avg_sentiment,
confidence=confidence,
weight=0.15,
timestamp=datetime.now(),
sample_size=len(sentiments),
metadata={
'total_engagement': total_engagement,
'influential_tweets': influential_count,
'avg_engagement': total_engagement / len(sentiments),
'sentiment_distribution': {
'positive': sum(1 for s in sentiments if s['score'] > 0.05),
'neutral': sum(1 for s in sentiments if -0.05 <= s['score'] <= 0.05),
'negative': sum(1 for s in sentiments if s['score'] < -0.05)
}
}
)
except Exception as e:
logger.error(f"Twitter analysis error: {e}")
return self._empty_sentiment_score('twitter')
async def analyze_reddit(self, symbol: str, lookback_hours: int = 24) -> SentimentScore:
"""Analyze Reddit sentiment for cryptocurrency"""
if not self.reddit_client:
return self._empty_sentiment_score('reddit')
try:
# Define relevant subreddits
subreddits = ['CryptoCurrency', 'Bitcoin', 'ethereum', 'CryptoMarkets']
sentiments = []
total_upvotes = 0
post_count = 0
comment_count = 0
for subreddit_name in subreddits:
subreddit = self.reddit_client.subreddit(subreddit_name)
# Search posts
search_query = f"{symbol}"
for post in subreddit.search(search_query, time_filter='day', limit=50):
post_age_hours = (datetime.now().timestamp() - post.created_utc) / 3600
if post_age_hours > lookback_hours:
continue
# Analyze post title and text
text = f"{post.title} {post.selftext}"
sentiment = self.vader.polarity_scores(text)['compound']
# Weight by upvote ratio and score
weight = max(1, post.score * post.upvote_ratio)
sentiments.append({
'score': sentiment,
'weight': weight
})
total_upvotes += post.score
post_count += 1
# Analyze top comments
post.comment_sort = 'top'
post.comment_limit = 10
for comment in post.comments[:10]:
if hasattr(comment, 'body'):
comment_sentiment = self.vader.polarity_scores(comment.body)['compound']
comment_weight = max(1, comment.score)
sentiments.append({
'score': comment_sentiment,
'weight': comment_weight
})
comment_count += 1
if not sentiments:
return self._empty_sentiment_score('reddit')
# Calculate weighted average
weighted_sum = sum(s['score'] * s['weight'] for s in sentiments)
total_weight = sum(s['weight'] for s in sentiments)
avg_sentiment = weighted_sum / total_weight if total_weight > 0 else 0.0
# Calculate activity level
total_items = post_count + comment_count
activity_level = (
'HIGH' if total_items > 100 else
'MEDIUM' if total_items > 30 else
'LOW'
)
# Confidence based on sample size
confidence = min(1.0, total_items / 100)
return SentimentScore(
source='reddit',
score=avg_sentiment,
confidence=confidence,
weight=0.10,
timestamp=datetime.now(),
sample_size=len(sentiments),
metadata={
'post_count': post_count,
'comment_count': comment_count,
'total_upvotes': total_upvotes,
'activity_level': activity_level,
'avg_upvotes_per_post': total_upvotes / post_count if post_count > 0 else 0
}
)
except Exception as e:
logger.error(f"Reddit analysis error: {e}")
return self._empty_sentiment_score('reddit')
async def analyze_telegram(self, symbol: str, lookback_hours: int = 24) -> SentimentScore:
"""Analyze Telegram group sentiment"""
if not self.telegram_client:
return self._empty_sentiment_score('telegram')
try:
# Define target channels/groups
channels = self.config.get('telegram_channels', [])
sentiments = []
total_messages = 0
async with self.telegram_client:
for channel in channels:
try:
# Get recent messages
messages = await self.telegram_client.get_messages(
channel,
limit=100
)
for msg in messages:
# Check message age
msg_age = datetime.now() - msg.date
if msg_age.total_seconds() > lookback_hours * 3600:
continue
if msg.text and symbol.lower() in msg.text.lower():
sentiment = self.vader.polarity_scores(msg.text)['compound']
# Weight by reactions/views
weight = 1
if msg.reactions:
weight += sum(r.count for r in msg.reactions.results)
sentiments.append({
'score': sentiment,
'weight': weight
})
total_messages += 1
except Exception as e:
logger.warning(f"Error accessing channel {channel}: {e}")
continue
if not sentiments:
return self._empty_sentiment_score('telegram')
# Calculate weighted average
weighted_sum = sum(s['score'] * s['weight'] for s in sentiments)
total_weight = sum(s['weight'] for s in sentiments)
avg_sentiment = weighted_sum / total_weight if total_weight > 0 else 0.0
confidence = min(1.0, total_messages / 50)
return SentimentScore(
source='telegram',
score=avg_sentiment,
confidence=confidence,
weight=0.10,
timestamp=datetime.now(),
sample_size=total_messages,
metadata={
'channels_analyzed': len(channels),
'total_messages': total_messages
}
)
except Exception as e:
logger.error(f"Telegram analysis error: {e}")
return self._empty_sentiment_score('telegram')
def _finbert_to_score(self, result: Dict) -> float:
"""Convert FinBERT result to -1 to 1 score"""
label = result['label'].lower()
score = result['score']
if label == 'positive':
return score
elif label == 'negative':
return -score
else: # neutral
return 0.0
def _empty_sentiment_score(self, source: str) -> SentimentScore:
"""Return empty sentiment score"""
return SentimentScore(
source=source,
score=0.0,
confidence=0.0,
weight=0.0,
timestamp=datetime.now(),
sample_size=0,
metadata={}
)
class DerivativesAnalyzer:
"""Analyze derivatives market sentiment"""
def __init__(self):
self.exchanges = {
'binance': ccxt.binance(),
'bybit': ccxt.bybit(),
'okx': ccxt.okx()
}
async def analyze_funding_rates(self, symbol: str) -> SentimentScore:
"""Analyze perpetual futures funding rates"""
try:
funding_rates = []
for exchange_name, exchange in self.exchanges.items():
try:
# Get funding rate
ticker = await exchange.fetch_ticker(f"{symbol}/USDT:USDT")
if 'fundingRate' in ticker['info']:
rate = float(ticker['info']['fundingRate'])
funding_rates.append({
'exchange': exchange_name,
'rate': rate
})
except Exception as e:
logger.warning(f"Error fetching funding rate from {exchange_name}: {e}")
continue
if not funding_rates:
return self._empty_sentiment_score('funding_rates')
# Calculate average funding rate
avg_rate = statistics.mean([fr['rate'] for fr in funding_rates])
# Convert funding rate to sentiment score
# Positive funding = longs pay shorts = bullish = positive sentiment
# Negative funding = shorts pay longs = bearish = negative sentiment
# Normalize to -1 to 1 scale (typical range: -0.1% to +0.1%)
sentiment_score = np.clip(avg_rate * 1000, -1.0, 1.0)
# High absolute funding rates indicate extreme sentiment
confidence = min(1.0, abs(avg_rate) * 5000)
return SentimentScore(
source='funding_rates',
score=sentiment_score,
confidence=confidence,
weight=0.12,
timestamp=datetime.now(),
sample_size=len(funding_rates),
metadata={
'avg_funding_rate': avg_rate,
'rate_percentage': avg_rate * 100,
'exchanges': funding_rates,
'interpretation': (
'Strong bullish bias' if avg_rate > 0.0005 else
'Mild bullish bias' if avg_rate > 0.0001 else
'Neutral' if abs(avg_rate) <= 0.0001 else
'Mild bearish bias' if avg_rate > -0.0005 else
'Strong bearish bias'
)
}
)
except Exception as e:
logger.error(f"Funding rate analysis error: {e}")
return self._empty_sentiment_score('funding_rates')
finally:
# Close exchange connections
for exchange in self.exchanges.values():
await exchange.close()
async def analyze_options_sentiment(self, symbol: str) -> SentimentScore:
"""Analyze options market sentiment (put/call ratio, max pain)"""
try:
# For demonstration - would integrate with Deribit API
# This is a simplified simulation
# Fetch options data (simulated)
put_volume = 15000
call_volume = 22000
put_call_ratio = put_volume / call_volume
# Calculate sentiment from put/call ratio
# Low ratio (< 0.7) = bullish (more calls) = positive sentiment
# High ratio (> 1.3) = bearish (more puts) = negative sentiment
if put_call_ratio < 0.7:
sentiment_score = (0.7 - put_call_ratio) / 0.7 # 0 to 1
elif put_call_ratio > 1.3:
sentiment_score = -(put_call_ratio - 1.3) / 1.3 # -1 to 0
else:
# Neutral range
sentiment_score = 0.0
sentiment_score = np.clip(sentiment_score, -1.0, 1.0)
# Confidence based on volume
total_volume = put_volume + call_volume
confidence = min(1.0, total_volume / 50000)
return SentimentScore(
source='options',
score=sentiment_score,
confidence=confidence,
weight=0.13,
timestamp=datetime.now(),
sample_size=1,
metadata={
'put_call_ratio': put_call_ratio,
'put_volume': put_volume,
'call_volume': call_volume,
'interpretation': (
'Bullish (heavy call buying)' if put_call_ratio < 0.7 else
'Neutral' if 0.7 <= put_call_ratio <= 1.3 else
'Bearish (heavy put buying)'
)
}
)
except Exception as e:
logger.error(f"Options sentiment analysis error: {e}")
return self._empty_sentiment_score('options')
def _empty_sentiment_score(self, source: str) -> SentimentScore:
"""Return empty sentiment score"""
return SentimentScore(
source=source,
score=0.0,
confidence=0.0,
weight=0.0,
timestamp=datetime.now(),
sample_size=0,
metadata={}
)
class OnChainAnalyzer:
"""Analyze on-chain sentiment indicators"""
def __init__(self, config: Dict):
self.config = config
# Initialize Web3 connections for different chains
self.w3_eth = Web3(Web3.HTTPProvider(config.get('ethereum_rpc')))
self.session = None
async def _get_session(self) -> aiohttp.ClientSession:
"""Get or create aiohttp session"""
if self.session is None or self.session.closed:
self.session = aiohttp.ClientSession()
return self.session
async def analyze_whale_movements(self, symbol: str) -> SentimentScore:
"""Analyze large holder movements"""
try:
session = await self._get_session()
# Use Whale Alert API or similar service
whale_api_url = "https://api.whale-alert.io/v1/transactions"
params = {
'api_key': self.config.get('whale_alert_api_key'),
'min_value': 1000000, # $1M minimum
'symbol': symbol.lower(),
'limit': 100
}
async with session.get(whale_api_url, params=params) as response:
if response.status != 200:
return self._empty_sentiment_score('whale_movements')
data = await response.json()
transactions = data.get('transactions', [])
if not transactions:
return self._empty_sentiment_score('whale_movements')
# Analyze transaction patterns
exchange_deposits = 0 # Bearish
exchange_withdrawals = 0 # Bullish
unknown_transfers = 0
total_value = 0
for tx in transactions:
value = tx.get('amount_usd', 0)
total_value += value
from_type = tx.get('from', {}).get('owner_type', 'unknown')
to_type = tx.get('to', {}).get('owner_type', 'unknown')
if to_type == 'exchange':
exchange_deposits += value
elif from_type == 'exchange':
exchange_withdrawals += value
else:
unknown_transfers += value
# Calculate net exchange flow sentiment
net_flow = exchange_withdrawals - exchange_deposits
# Normalize to -1 to 1 scale
max_value = max(exchange_deposits, exchange_withdrawals)
if max_value > 0:
sentiment_score = net_flow / max_value
else:
sentiment_score = 0.0
sentiment_score = np.clip(sentiment_score, -1.0, 1.0)
# Confidence based on transaction count and value
confidence = min(1.0, (len(transactions) / 100) * (total_value / 100000000))
return SentimentScore(
source='whale_movements',
score=sentiment_score,
confidence=confidence,
weight=0.15,
timestamp=datetime.now(),
sample_size=len(transactions),
metadata={
'exchange_deposits_usd': exchange_deposits,
'exchange_withdrawals_usd': exchange_withdrawals,
'net_flow_usd': net_flow,
'total_value_usd': total_value,
'transaction_count': len(transactions),
'interpretation': (
'Strong accumulation (whale withdrawals)' if sentiment_score > 0.5 else
'Mild accumulation' if sentiment_score > 0.2 else
'Neutral' if abs(sentiment_score) <= 0.2 else
'Mild distribution' if sentiment_score > -0.5 else
'Strong distribution (whale deposits)'
)
}
)
except Exception as e:
logger.error(f"Whale movement analysis error: {e}")
return self._empty_sentiment_score('whale_movements')
async def analyze_network_value_metrics(self, symbol: str) -> SentimentScore:
"""Analyze MVRV, NVT, and other network value metrics"""
try:
# For demonstration - would integrate with Glassnode, CoinMetrics, etc.
# This is a simplified simulation
# MVRV ratio (Market Value to Realized Value)
# > 3.0 = overvalued/euphoric (negative sentiment)
# < 1.0 = undervalued/despair (positive sentiment for contrarian)
mvrv_ratio = 2.1
# NVT ratio (Network Value to Transactions)
# High NVT = overvalued relative to usage
nvt_ratio = 85
# Calculate sentiment from MVRV
# Convert to -1 to 1 scale where extremes indicate reversal potential
if mvrv_ratio > 3.0:
# Overvalued - bearish
mvrv_sentiment = -(min(mvrv_ratio - 3.0, 2.0) / 2.0)
elif mvrv_ratio < 1.0:
# Undervalued - bullish (contrarian)
mvrv_sentiment = (1.0 - mvrv_ratio)
else:
# Neutral range (1.0 to 3.0)
mvrv_sentiment = (2.0 - mvrv_ratio) / 2.0
# Calculate sentiment from NVT (inverse relationship)
# Low NVT = healthy usage = bullish
# High NVT = overvalued = bearish
if nvt_ratio > 100:
nvt_sentiment = -0.5
elif nvt_ratio < 50:
nvt_sentiment = 0.5
else:
nvt_sentiment = 0.0
# Combine metrics
combined_sentiment = (mvrv_sentiment * 0.6 + nvt_sentiment * 0.4)
combined_sentiment = np.clip(combined_sentiment, -1.0, 1.0)
# Confidence is moderate for on-chain metrics
confidence = 0.7
return SentimentScore(
source='network_metrics',
score=combined_sentiment,
confidence=confidence,
weight=0.05,
timestamp=datetime.now(),
sample_size=1,
metadata={
'mvrv_ratio': mvrv_ratio,
'nvt_ratio': nvt_ratio,
'mvrv_interpretation': (
'Extremely overvalued' if mvrv_ratio > 3.5 else
'Overvalued' if mvrv_ratio > 2.5 else
'Fair value' if 1.0 <= mvrv_ratio <= 2.5 else
'Undervalued'
),
'nvt_interpretation': (
'Overvalued vs usage' if nvt_ratio > 100 else
'Fair valuation' if 50 <= nvt_ratio <= 100 else
'Undervalued vs usage'
)
}
)
except Exception as e:
logger.error(f"Network value metrics analysis error: {e}")
return self._empty_sentiment_score('network_metrics')
def _empty_sentiment_score(self, source: str) -> SentimentScore:
"""Return empty sentiment score"""
return SentimentScore(
source=source,
score=0.0,
confidence=0.0,
weight=0.0,
timestamp=datetime.now(),
sample_size=0,
metadata={}
)
async def close(self):
"""Close aiohttp session"""
if self.session and not self.session.closed:
await self.session.close()
class MarketSentimentAggregator:
"""Aggregate all sentiment sources into unified score"""
def __init__(self, config: Dict):
self.config = config
self.social_analyzer = SocialMediaAnalyzer(config)
self.derivatives_analyzer = DerivativesAnalyzer()
self.onchain_analyzer = OnChainAnalyzer(config)
# Historical sentiment tracking
self.sentiment_history = deque(maxlen=168) # 7 days of hourly data
async def analyze_complete_sentiment(self, symbol: str) -> Dict:
"""Run complete sentiment analysis"""
logger.info(f"Starting comprehensive sentiment analysis for {symbol}")
# Gather all sentiment sources concurrently
tasks = [
self.social_analyzer.analyze_twitter(symbol),
self.social_analyzer.analyze_reddit(symbol),
self.social_analyzer.analyze_telegram(symbol),
self.derivatives_analyzer.analyze_funding_rates(symbol),
self.derivatives_analyzer.analyze_options_sentiment(symbol),
self.onchain_analyzer.analyze_whale_movements(symbol),
self.onchain_analyzer.analyze_network_value_metrics(symbol)
]
sentiment_scores = await asyncio.gather(*tasks, return_exceptions=True)
# Filter out exceptions
valid_scores = [
score for score in sentiment_scores
if isinstance(score, SentimentScore) and score.confidence > 0
]
if not valid_scores:
logger.warning("No valid sentiment scores obtained")
return self._empty_analysis_result(symbol)
# Calculate weighted aggregate sentiment
aggregate_sentiment = self._calculate_weighted_sentiment(valid_scores)
# Calculate Fear & Greed Index
fear_greed_index = self._calculate_fear_greed_index(valid_scores, aggregate_sentiment)
# Generate sentiment alerts
alerts = self._generate_alerts(valid_scores, fear_greed_index)
# Store in history
self.sentiment_history.append({
'timestamp': datetime.now(),
'sentiment': aggregate_sentiment,
'fear_greed_score': fear_greed_index.score
})
# Calculate trend
trend = self._calculate_trend()
# Perform correlation analysis with historical price
correlation_analysis = await self._analyze_sentiment_price_correlation(symbol)
result = {
'symbol': symbol,
'timestamp': datetime.now().isoformat(),
'aggregate_sentiment': {
'score': aggregate_sentiment,
'normalized_score': (aggregate_sentiment + 1) / 2, # 0 to 1 scale
'interpretation': self._interpret_sentiment(aggregate_sentiment)
},
'fear_greed_index': {
'score': fear_greed_index.score,
'level': fear_greed_index.level.value,
'historical_percentile': fear_greed_index.historical_percentile,
'trend': trend,
'components': fear_greed_index.components
},
'sentiment_sources': [
{
'source': score.source,
'score': score.score,
'confidence': score.confidence,
'weight': score.weight,
'sample_size': score.sample_size,
'metadata': score.metadata
}
for score in valid_scores
],
'alerts': [
{
'type': alert.alert_type,
'severity': alert.severity,
'message': alert.message,
'sentiment_score': alert.sentiment_score,
'trigger': alert.trigger_condition
}
for alert in alerts
],
'correlation_analysis': correlation_analysis,
'data_quality': {
'sources_analyzed': len(valid_scores),
'total_samples': sum(score.sample_size for score in valid_scores),
'avg_confidence': statistics.mean([score.confidence for score in valid_scores]),
'coverage_percentage': (len(valid_scores) / 7) * 100 # 7 possible sources
}
}
logger.info(f"Sentiment analysis complete: {fear_greed_index.level.value} "
f"(score: {fear_greed_index.score})")
return result
def _calculate_weighted_sentiment(self, scores: List[SentimentScore]) -> float:
"""Calculate weighted average sentiment"""
weighted_sum = sum(
score.score * score.weight * score.confidence
for score in scores
)
total_weight = sum(
score.weight * score.confidence
for score in scores
)
if total_weight == 0:
return 0.0
return weighted_sum / total_weight
def _calculate_fear_greed_index(
self,
scores: List[SentimentScore],
aggregate_sentiment: float
) -> FearGreedIndex:
"""Calculate Fear & Greed Index (0-100)"""
# Start with aggregate sentiment scaled to 0-100
base_score = (aggregate_sentiment + 1) * 50
# Build component scores
components = {}
for score in scores:
# Scale each source to 0-100
component_score = (score.score + 1) * 50
components[score.source] = component_score
# Calculate historical percentile
if len(self.sentiment_history) > 10:
historical_scores = [h['fear_greed_score'] for h in self.sentiment_history]
percentile = (
sum(1 for s in historical_scores if s < base_score) /
len(historical_scores) * 100
)
else:
percentile = 50.0 # Default to median if insufficient history
# Determine sentiment level
if base_score >= 75:
level = SentimentLevel.EXTREME_GREED
elif base_score >= 55:
level = SentimentLevel.GREED
elif base_score >= 45:
level = SentimentLevel.NEUTRAL
elif base_score >= 25:
level = SentimentLevel.FEAR
else:
level = SentimentLevel.EXTREME_FEAR
return FearGreedIndex(
score=int(base_score),
level=level,
components=components,
historical_percentile=percentile,
trend='', # Calculated separately
timestamp=datetime.now()
)
def _calculate_trend(self) -> str:
"""Calculate sentiment trend from history"""
if len(self.sentiment_history) < 5:
return 'insufficient_data'
# Compare recent average to older average
recent_scores = [h['fear_greed_score'] for h in list(self.sentiment_history)[-5:]]
older_scores = [h['fear_greed_score'] for h in list(self.sentiment_history)[-10:-5]]
recent_avg = statistics.mean(recent_scores)
older_avg = statistics.mean(older_scores)
diff = recent_avg - older_avg
if diff > 5:
return 'increasing'
elif diff < -5:
return 'decreasing'
else:
return 'stable'
def _generate_alerts(
self,
scores: List[SentimentScore],
fear_greed: FearGreedIndex
) -> List[SentimentAlert]:
"""Generate trading alerts based on sentiment extremes"""
alerts = []
# Extreme Fear alert (potential buying opportunity)
if fear_greed.level == SentimentLevel.EXTREME_FEAR:
alerts.append(SentimentAlert(
alert_type='contrarian_buy_opportunity',
severity='high',
message=f'Extreme Fear detected (score: {fear_greed.score}). '
'Historically, this has preceded price rebounds.',
sentiment_score=fear_greed.score,
trigger_condition='fear_greed_score < 25',
timestamp=datetime.now()
))
# Extreme Greed alert (potential selling opportunity)
if fear_greed.level == SentimentLevel.EXTREME_GREED:
alerts.append(SentimentAlert(
alert_type='contrarian_sell_opportunity',
severity='high',
message=f'Extreme Greed detected (score: {fear_greed.score}). '
'Market may be overheated - consider profit taking.',
sentiment_score=fear_greed.score,
trigger_condition='fear_greed_score > 75',
timestamp=datetime.now()
))
# Funding rate extreme alert
funding_score = next(
(s for s in scores if s.source == 'funding_rates'),
None
)
if funding_score and abs(funding_score.score) > 0.7:
direction = 'long' if funding_score.score > 0 else 'short'
alerts.append(SentimentAlert(
alert_type='funding_rate_extreme',
severity='medium',
message=f'Extreme {direction} funding rate detected. '
f'Potential for {direction} squeeze or reversal.',
sentiment_score=funding_score.score,
trigger_condition=f'abs(funding_rate_score) > 0.7',
timestamp=datetime.now()
))
# Whale movement alert
whale_score = next(
(s for s in scores if s.source == 'whale_movements'),
None
)
if whale_score and abs(whale_score.score) > 0.5:
direction = 'accumulation' if whale_score.score > 0 else 'distribution'
alerts.append(SentimentAlert(
alert_type='whale_activity',
severity='medium',
message=f'Significant whale {direction} detected. '
f'Smart money may be positioning for trend.',
sentiment_score=whale_score.score,
trigger_condition='abs(whale_score) > 0.5',
timestamp=datetime.now()
))
return alerts
async def _analyze_sentiment_price_correlation(self, symbol: str) -> Dict:
"""Analyze correlation between sentiment and price movements"""
try:
if len(self.sentiment_history) < 20:
return {
'status': 'insufficient_data',
'message': 'Need at least 20 data points for correlation analysis'
}
# Extract sentiment scores
sentiments = [h['sentiment'] for h in self.sentiment_history]
# Would fetch actual price data here
# For demonstration, simulating price correlation
# Calculate correlation coefficient (simulated)
correlation = 0.65 # Simulated positive correlation
# Calculate lead/lag relationship
# Does sentiment lead price, or vice versa?
sentiment_leads_price = True # Simulated
lag_hours = 6 # Simulated
return {
'status': 'success',
'correlation_coefficient': correlation,
'correlation_strength': (
'strong' if abs(correlation) > 0.7 else
'moderate' if abs(correlation) > 0.4 else
'weak'
),
'relationship': (
'Sentiment leads price' if sentiment_leads_price
else 'Price leads sentiment'
),
'lag_hours': lag_hours,
'sample_size': len(sentiments),
'interpretation': (
'Sentiment changes typically precede price movements by ~6 hours. '
'Strong correlation suggests sentiment is a useful leading indicator.'
if sentiment_leads_price and abs(correlation) > 0.6
else 'Moderate correlation. Sentiment should be combined with other factors.'
)
}
except Exception as e:
logger.error(f"Correlation analysis error: {e}")
return {
'status': 'error',
'message': str(e)
}
def _interpret_sentiment(self, score: float) -> str:
"""Interpret aggregate sentiment score"""
if score > 0.6:
return 'Very Bullish'
elif score > 0.2:
return 'Bullish'
elif score > -0.2:
return 'Neutral'
elif score > -0.6:
return 'Bearish'
else:
return 'Very Bearish'
def _empty_analysis_result(self, symbol: str) -> Dict:
"""Return empty analysis result"""
return {
'symbol': symbol,
'timestamp': datetime.now().isoformat(),
'error': 'Unable to gather sufficient sentiment data',
'aggregate_sentiment': None,
'fear_greed_index': None,
'sentiment_sources': [],
'alerts': []
}
async def close(self):
"""Cleanup resources"""
await self.onchain_analyzer.close()
async def main():
"""Example usage"""
# Configuration
config = {
'twitter_bearer_token': 'YOUR_TWITTER_TOKEN',
'reddit_client_id': 'YOUR_REDDIT_CLIENT_ID',
'reddit_client_secret': 'YOUR_REDDIT_SECRET',
'telegram_api_id': 'YOUR_TELEGRAM_API_ID',
'telegram_api_hash': 'YOUR_TELEGRAM_HASH',
'telegram_channels': ['bitcoin', 'cryptocurrency'],
'whale_alert_api_key': 'YOUR_WHALE_ALERT_KEY',
'ethereum_rpc': 'https://mainnet.infura.io/v3/YOUR_KEY'
}
# Initialize aggregator
aggregator = MarketSentimentAggregator(config)
try:
# Analyze sentiment
result = await aggregator.analyze_complete_sentiment('BTC')
# Pretty print results
print("\n" + "="*80)
print(f"MARKET SENTIMENT ANALYSIS - {result['symbol']}")
print("="*80)
print(f"\nFear & Greed Index: {result['fear_greed_index']['score']} - "
f"{result['fear_greed_index']['level']}")
print(f"Trend: {result['fear_greed_index']['trend']}")
print(f"Historical Percentile: {result['fear_greed_index']['historical_percentile']:.1f}%")
print(f"\nAggregate Sentiment: {result['aggregate_sentiment']['interpretation']}")
print(f"Score: {result['aggregate_sentiment']['score']:.3f}")
print("\nSentiment by Source:")
print("-" * 80)
for source in result['sentiment_sources']:
print(f" {source['source']:20} | Score: {source['score']:6.3f} | "
f"Confidence: {source['confidence']:.2f} | Samples: {source['sample_size']}")
if result['alerts']:
print("\nAlerts:")
print("-" * 80)
for alert in result['alerts']:
print(f" [{alert['severity'].upper()}] {alert['message']}")
print("\nData Quality:")
print("-" * 80)
print(f" Sources Analyzed: {result['data_quality']['sources_analyzed']}")
print(f" Total Samples: {result['data_quality']['total_samples']}")
print(f" Avg Confidence: {result['data_quality']['avg_confidence']:.2%}")
print(f" Coverage: {result['data_quality']['coverage_percentage']:.1f}%")
if result['correlation_analysis']['status'] == 'success':
print("\nSentiment-Price Correlation:")
print("-" * 80)
corr = result['correlation_analysis']
print(f" Correlation: {corr['correlation_coefficient']:.3f} ({corr['correlation_strength']})")
print(f" Relationship: {corr['relationship']}")
print(f" {corr['interpretation']}")
print("\n" + "="*80)
# Export to JSON
output_file = f"sentiment_analysis_{result['symbol']}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
with open(output_file, 'w') as f:
json.dump(result, f, indent=2)
print(f"\nFull analysis exported to: {output_file}")
finally:
await aggregator.close()
if __name__ == '__main__':
asyncio.run(main())
# Basic sentiment analysis
/analyze-sentiment BTC
# Analysis with specific lookback period
/analyze-sentiment ETH --lookback 48h
# Export to JSON
/analyze-sentiment BTC --output sentiment_btc.json
# Monitor sentiment in real-time
/analyze-sentiment BTC --monitor --interval 1h
# Use sentiment for entry signals
result = await aggregator.analyze_complete_sentiment('BTC')
if result['fear_greed_index']['level'] == 'Extreme Fear':
# Potential buying opportunity
if result['aggregate_sentiment']['score'] < -0.6:
print("STRONG BUY SIGNAL: Extreme fear with very bearish sentiment")
# Check for reversal signals
if len(result['alerts']) > 0:
for alert in result['alerts']:
if alert['type'] == 'contrarian_buy_opportunity':
print(f"Contrarian signal: {alert['message']}")
29plugins reuse this command
First indexed Dec 31, 2025
Showing the 6 earliest of 29 plugins
npx claudepluginhub jamon8888/claude-code-plugins-plus --plugin market-sentiment-analyzer/analyze-sentimentAnalyzes market sentiment from social media, news, and on-chain metrics to produce a weighted Fear & Greed Index and trading signals.
/aggregate-newsAggregates crypto news from 50+ sources with AI sentiment analysis, deduplication, and market impact scoring to surface actionable intelligence.
/analyze-chainMonitors blockchain network health, holder distribution, whale movements, exchange flows, and transaction velocity across multiple chains. Produces on-chain analytics reports.
/sentimentChecks market sentiment using Fear & Greed indices, market movers, and optional Google Trends keywords, then synthesizes a risk assessment summary.
/monitor-whalesMonitors large cryptocurrency transactions (whales) across multiple blockchains in real-time, analyzes market impact, and sends alerts via Slack, Discord, Telegram, or email.
/generate-signalGenerates cryptocurrency trading signals by analyzing technical indicators, chart patterns, and market conditions across multiple timeframes.