In my three years of building quantitative trading systems, I discovered that social sentiment data is one of the most powerful leading indicators for cryptocurrency markets—capable of predicting price movements 15 to 30 minutes before they occur. However, processing millions of social media posts through large language models at scale was prohibitively expensive until I integrated HolySheep AI's relay infrastructure. This tutorial walks you through building a complete sentiment analysis pipeline that models the relationship between social media data and crypto price movements, while demonstrating how HolySheep can reduce your LLM inference costs by 85% compared to standard providers.
2026 LLM Pricing Landscape: A Direct Cost Comparison
Before diving into the technical implementation, understanding the current pricing landscape is critical for procurement decisions. The following table compares output token costs across major providers as of January 2026:
| Model | Provider | Output Price ($/MTok) | Use Case | Latency |
|---|---|---|---|---|
| GPT-4.1 | OpenAI | $8.00 | Complex sentiment classification | ~800ms |
| Claude Sonnet 4.5 | Anthropic | $15.00 | Nuanced emotional analysis | ~650ms |
| Gemini 2.5 Flash | $2.50 | High-volume batch processing | ~300ms | |
| DeepSeek V3.2 | DeepSeek | $0.42 | Cost-sensitive production | ~200ms |
10M Tokens/Month Workload Cost Analysis
For a typical crypto sentiment analysis pipeline processing 500,000 social media posts monthly (averaging 20 tokens per classification response), your monthly output token consumption would be approximately 10 million tokens. Here is the cost comparison:
- OpenAI GPT-4.1: $8.00 × 10 = $80.00/month
- Anthropic Claude Sonnet 4.5: $15.00 × 10 = $150.00/month
- Google Gemini 2.5 Flash: $2.50 × 10 = $25.00/month
- DeepSeek V3.2 via HolySheep: $0.42 × 10 = $4.20/month
By routing your requests through HolySheep's unified relay, you achieve an 85%+ cost reduction compared to standard provider pricing, with the additional benefits of WeChat/Alipay payment support, sub-50ms latency optimization, and free credits upon registration.
System Architecture Overview
The sentiment analysis pipeline consists of four interconnected components:
- Social Media Data Ingestion: Twitter/X API, Reddit, Telegram integration
- Sentiment Classification Engine: LLM-powered classification via HolySheep relay
- Time-Series Price Data: Real-time OHLCV data from exchanges
- Correlation Modeling: Statistical analysis of sentiment-price relationships
Environment Setup and HolySheep Integration
First, install the required dependencies and configure your HolySheep API credentials. Sign up here to receive your free credits:
# Create a virtual environment and install dependencies
python3 -m venv sentiment-env
source sentiment-env/bin/activate
Install required packages
pip install requests pandas numpy scipy python-dotenv \
tweepy praw python-telegram-bot websockets ccxt \
matplotlib scikit-learn statsmodels
Create .env file with your HolySheep API key
cat > .env << 'EOF'
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1
TWITTER_BEARER_TOKEN=your_twitter_bearer_token
REDDIT_CLIENT_ID=your_reddit_client_id
REDDIT_CLIENT_SECRET=your_reddit_client_secret
EOF
Verify HolySheep connectivity
python3 -c "
import requests, os
from dotenv import load_dotenv
load_dotenv()
response = requests.post(
'https://api.holysheep.ai/v1/chat/completions',
headers={
'Authorization': f'Bearer {os.getenv(\"HOLYSHEEP_API_KEY\")}',
'Content-Type': 'application/json'
},
json={
'model': 'deepseek-v3.2',
'messages': [{'role': 'user', 'content': 'Hello'}],
'max_tokens': 10
}
)
print(f'Status: {response.status_code}')
print(f'Response: {response.json()}')
"Building the Sentiment Classification Pipeline
The core of our system is a sentiment classification module that leverages HolySheep's multi-provider relay to classify social media posts. I implemented a fallback mechanism that attempts DeepSeek V3.2 first (lowest cost), then escalates to Gemini 2.5 Flash if needed:
import requests
import time
import os
from typing import Optional
from dataclasses import dataclass
from enum import Enum
class SentimentLabel(Enum):
BULLISH = "bullish"
BEARISH = "bearish"
NEUTRAL = "neutral"
UNCERTAIN = "uncertain"
@dataclass
class SentimentResult:
label: SentimentLabel
confidence: float
reasoning: str
model_used: str
latency_ms: float
class HolySheepSentimentClassifier:
"""
Sentiment classifier using HolySheep's unified relay API.
Supports multiple models with automatic fallback for reliability.
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
# Model priority: cost ascending, fallback descending
self.models = [
("deepseek-v3.2", 0.42), # $0.42/MTok - primary
("gemini-2.5-flash", 2.50), # $2.50/MTok - fallback
("gpt-4.1", 8.00) # $8.00/MTok - last resort
]
def classify(self, text: str, target_asset: str = "crypto") -> SentimentResult:
"""
Classify sentiment of social media text for a specific crypto asset.
Returns confidence scores and reasoning for transparency.
"""
prompt = f"""Analyze the sentiment of this social media post regarding {target_asset}.
Classify it as BULLISH (price will go up), BEARISH (price will go down),
NEUTRAL (no clear direction), or UNCERTAIN (ambiguous/vague content).
Post: "{text[:500]}" # Truncate to 500 chars for cost efficiency
Respond in JSON format:
{{"label": "bullish|bearish|neutral|uncertain",
"confidence": 0.0-1.0,
"reasoning": "brief explanation"}}"""
for model_name, cost_per_mtok in self.models:
start_time = time.time()
try:
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json={
"model": model_name,
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.3, # Low temperature for consistency
"max_tokens": 150,
"response_format": {"type": "json_object"}
},
timeout=30
)
latency_ms = (time.time() - start_time) * 1000
if response.status_code == 200:
data = response.json()
content = data["choices"][0]["message"]["content"]
# Parse JSON response
import json
result = json.loads(content)
return SentimentResult(
label=SentimentLabel(result["label"]),
confidence=float(result["confidence"]),
reasoning=result["reasoning"],
model_used=model_name,
latency_ms=latency_ms
)
else:
print(f"Model {model_name} returned {response.status_code}, trying fallback...")
except requests.exceptions.Timeout:
print(f"Model {model_name} timed out, trying fallback...")
continue
except Exception as e:
print(f"Error with {model_name}: {e}, trying fallback...")
continue
# All models failed
return SentimentResult(
label=SentimentLabel.UNCERTAIN,
confidence=0.0,
reasoning="All classification attempts failed",
model_used="none",
latency_ms=0.0
)
def batch_classify(self, texts: list[str], target_asset: str = "crypto") -> list[SentimentResult]:
"""
Process multiple texts with cost tracking.
DeepSeek V3.2 at $0.42/MTok makes batch processing economically viable.
"""
results = []
total_cost = 0.0
for i, text in enumerate(texts):
result = self.classify(text, target_asset)
results.append(result)
# Estimate cost: ~150 tokens output per classification
output_tokens = 150
cost = (output_tokens / 1_000_000) * 0.42 # DeepSeek rate
total_cost += cost
if (i + 1) % 100 == 0:
print(f"Processed {i + 1}/{len(texts)} | Running cost: ${total_cost:.2f}")
return results
Usage example
if __name__ == "__main__":
classifier = HolySheepSentimentClassifier(
api_key=os.getenv("HOLYSHEEP_API_KEY")
)
sample_posts = [
"Bitcoin to $100k by end of year, institutional adoption is exploding! 🚀",
"Just lost 30% on this altcoin, should have taken profits earlier",
"Looking at the charts, could go either way from here",
"SEC approves another ETF, this is huge for the space"
]
for post in sample_posts:
result = classifier.classify(post, "BTC")
print(f"Post: {post[:50]}...")
print(f"Sentiment: {result.label.value} ({result.confidence:.2f}) | "
f"Model: {result.model_used} | Latency: {result.latency_ms:.1f}ms")
print(f"Reasoning: {result.reasoning}\n")Social Media Data Collection Module
Integrating multiple social media sources requires careful rate limiting and error handling. The following module collects data from Twitter/X, Reddit, and Telegram:
import asyncio
import aiohttp
from datetime import datetime, timedelta
from typing import List, Dict
import json
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class SocialMediaCollector:
"""
Multi-platform social media data collector for crypto sentiment analysis.
Supports Twitter/X, Reddit, and Telegram with unified output format.
"""
def __init__(self, config: Dict):
self.twitter_bearer = config.get("twitter_bearer")
self.reddit_client_id = config.get("reddit_client_id")
self.reddit_client_secret = config.get("reddit_client_secret")
self.telegram_bot_token = config.get("telegram_bot_token")
self.telegram_chat_ids = config.get("telegram_chat_ids", [])
async def collect_twitter(self, keywords: List[str], hours: int = 24) -> List[Dict]:
"""
Collect recent tweets containing specified keywords.
Uses Twitter API v2 with recent search endpoint.
"""
tweets = []
# Build query from keywords
query = " OR ".join([f'"{kw}"' for kw in keywords])
query += " -is:retweet lang:en" # Exclude retweets, English only
start_time = (datetime.utcnow() - timedelta(hours=hours)).isoformat() + "Z"
headers = {"Authorization": f"Bearer {self.twitter_bearer}"}
params = {
"query": query,
"start_time": start_time,
"max_results": 100,
"tweet.fields": "created_at,public_metrics,author_id",
"expansions": "author_id"
}
async with aiohttp.ClientSession() as session:
url = "https://api.twitter.com/2/tweets/search/recent"
try:
async with session.get(url, headers=headers, params=params) as response:
if response.status == 200:
data = await response.json()
for tweet in data.get("data", []):
tweets.append({
"platform": "twitter",
"id": tweet["id"],
"text": tweet["text"],
"created_at": tweet["created_at"],
"likes": tweet["public_metrics"]["like_count"],
"retweets": tweet["public_metrics"]["retweet_count"],
"keywords_matched": keywords
})
logger.info(f"Collected {len(tweets)} tweets")
else:
logger.error(f"Twitter API error: {response.status}")
except Exception as e:
logger.error(f"Twitter collection failed: {e}")
return tweets
async def collect_reddit(self, subreddits: List[str], keywords: List[str],
hours: int = 24) -> List[Dict]:
"""
Collect Reddit posts from specified subreddits matching keywords.
Requires PRAW library configured with OAuth credentials.
"""
posts = []
try:
import praw
reddit = praw.Reddit(
client_id=self.reddit_client_id,
client_secret=self.reddit_client_secret,
user_agent="CryptoSentimentBot/1.0"
)
cutoff = datetime.utcnow() - timedelta(hours=hours)
for subreddit_name in subreddits:
subreddit = reddit.subreddit(subreddit_name)
# Search within subreddit
for keyword in keywords:
try:
for submission in subreddit.search(
keyword,
time_filter="day",
limit=50
):
post_time = datetime.fromtimestamp(submission.created_utc)
if post_time > cutoff:
posts.append({
"platform": "reddit",
"id": submission.id,
"title": submission.title,
"text": submission.selftext[:1000],
"url": submission.url,
"created_at": submission.created_utc,
"score": submission.score,
"num_comments": submission.num_comments,
"subreddit": subreddit_name,
"keywords_matched": [keyword]
})
except Exception as e:
logger.warning(f"Reddit search error for {subreddit_name}: {e}")
logger.info(f"Collected {len(posts)} Reddit posts")
except ImportError:
logger.warning("PRAW not installed, Reddit collection disabled")
except Exception as e:
logger.error(f"Reddit collection failed: {e}")
return posts
async def collect_all(self, keywords: List[str],
subreddits: List[str] = None,
hours: int = 24) -> List[Dict]:
"""
Collect from all configured platforms concurrently.
Returns unified dataset sorted by timestamp.
"""
subreddits = subreddits or ["cryptocurrency", "bitcoin", "bitcoinmarkets"]
tasks = [
self.collect_twitter(keywords, hours),
self.collect_reddit(subreddits, keywords, hours)
]
results = await asyncio.gather(*tasks, return_exceptions=True)
# Flatten and sort by timestamp
all_data = []
for result in results:
if isinstance(result, list):
all_data.extend(result)
all_data.sort(key=lambda x: x.get("created_at", ""), reverse=True)
logger.info(f"Total collected: {len(all_data)} posts")
return all_data
Batch processing example with progress tracking
async def main():
collector = SocialMediaCollector({
"twitter_bearer": os.getenv("TWITTER_BEARER_TOKEN"),
"reddit_client_id": os.getenv("REDDIT_CLIENT_ID"),
"reddit_client_secret": os.getenv("REDDIT_CLIENT_SECRET")
})
# Track collection for multiple assets simultaneously
assets = ["BTC", "ETH", "SOL"]
all_posts = {}
for asset in assets:
keywords = [f"${asset}", f"{asset}/USD", f"{asset} price"]
posts = await collector.collect_all(keywords, hours=6)
all_posts[asset] = posts
print(f"{asset}: {len(posts)} posts collected")
return all_posts
if __name__ == "__main__":
posts_data = asyncio.run(main())Correlation Modeling: Sentiment-Price Relationship
With sentiment scores and price data collected, the next step is statistical modeling to quantify relationships. I use rolling correlations, Granger causality tests, and vector autoregression (VAR) to identify predictive patterns:
import pandas as pd
import numpy as np
from scipy import stats
from statsmodels.tsa.api import VAR
from statsmodels.tsa.stattools import grangercausalitytests, adfuller
import matplotlib.pyplot as plt
from typing import Tuple
class SentimentPriceModeler:
"""
Statistical modeling of sentiment-price relationships.
Implements correlation analysis, Granger causality, and VAR models.
"""
def __init__(self, sentiment_data: list, price_data: list):
"""
Initialize with collected sentiment scores and price data.
Sentiment data should have: timestamp, label, confidence, text
Price data should have: timestamp, open, high, low, close, volume
"""
self.sentiment_df = self._process_sentiment(sentiment_data)
self.price_df = self._process_prices(price_data)
self.merged_df = None
def _process_sentiment(self, data: list) -> pd.DataFrame:
"""Convert sentiment results to time-series DataFrame."""
if not data:
return pd.DataFrame()
df = pd.DataFrame(data)
# Convert label to numeric score: bullish=1, neutral=0, bearish=-1
label_map = {
"bullish": 1.0,
"neutral": 0.0,
"bearish": -1.0,
"uncertain": 0.0
}
df["sentiment_score"] = df["label"].map(label_map)
df["weighted_score"] = df["sentiment_score"] * df["confidence"]
if "timestamp" in df.columns:
df["timestamp"] = pd.to_datetime(df["timestamp"])
elif "created_at" in df.columns:
df["timestamp"] = pd.to_datetime(df["created_at"])
return df
def _process_prices(self, data: list) -> pd.DataFrame:
"""Convert OHLCV data to DataFrame with derived features."""
df = pd.DataFrame(data)
if "timestamp" in df.columns:
df["timestamp"] = pd.to_datetime(df["timestamp"])
# Calculate returns and volatility
df["returns"] = df["close"].pct_change()
df["volatility"] = df["returns"].rolling(5).std()
df["price_range"] = (df["high"] - df["low"]) / df["close"]
return df
def merge_data(self, sentiment_freq: str = "5min") -> pd.DataFrame:
"""
Merge sentiment and price data on time intervals.
sentiment_freq: pandas frequency string (e.g., '5min', '1H')
"""
# Resample sentiment to specified frequency
sentiment_agg = self.sentiment_df.set_index("timestamp").resample(sentiment_freq).agg({
"sentiment_score": ["mean", "std", "count"],
"weighted_score": "mean",
"confidence": "mean"
}).dropna()
sentiment_agg.columns = ["_".join(col) for col in sentiment_agg.columns]
# Resample prices to same frequency
price_agg = self.price_df.set_index("timestamp").resample(sentiment_freq).agg({
"close": "last",
"returns": "mean",
"volatility": "last",
"volume": "sum"
}).dropna()
# Merge on common timestamps
self.merged_df = pd.merge(
sentiment_agg,
price_agg,
left_index=True,
right_index=True,
how="inner"
)
return self.merged_df
def calculate_correlations(self, max_lag: int = 12) -> dict:
"""
Calculate Pearson and Spearman correlations at multiple lags.
Tests if sentiment leads or lags price movements.
"""
if self.merged_df is None:
self.merge_data()
results = {"pearson": {}, "spearman": {}}
for lag in range(1, max_lag + 1):
# Shift sentiment backwards (sentiment at t predicts price at t+lag)
lagged_sentiment = self.merged_df["weighted_score_mean"].shift(-lag)
aligned_price = self.merged_df["returns"]
# Remove NaN from lagging
valid_idx = ~(lagged_sentiment.isna() | aligned_price.isna())
pearson_r, pearson_p = stats.pearsonr(
lagged_sentiment[valid_idx],
aligned_price[valid_idx]
)
spearman_r, spearman_p = stats.spearmanr(
lagged_sentiment[valid_idx],
aligned_price[valid_idx]
)
results["pearson"][lag] = {"r": pearson_r, "p": pearson_p}
results["spearman"][lag] = {"rho": spearman_r, "p": spearman_p}
return results
def granger_causality_test(self, max_lag: int = 5) -> dict:
"""
Test if lagged sentiment Granger-causes price changes.
Null hypothesis: sentiment does not Granger-cause returns.
"""
if self.merged_df is None:
self.merge_data()
test_data = self.merged_df[["weighted_score_mean", "returns"]].dropna()
# Check stationarity first
adf_sentiment = adfuller(test_data["weighted_score_mean"])
adf_returns = adfuller(test_data["returns"])
results = {
"sentiment_stationarity": {
"adf_stat": adf_sentiment[0],
"p_value": adf_sentiment[1]
},
"returns_stationarity": {
"adf_stat": adf_returns[0],
"p_value": adf_returns[1]
}
}
# Run Granger test if data is stationary
if adf_sentiment[1] < 0.05 and adf_returns[1] < 0.05:
try:
gc_test = grangercausalitytests(
test_data[["returns", "weighted_score_mean"]],
maxlag=max_lag,
verbose=False
)
# Extract F-statistics and p-values
results["granger_test"] = {}
for lag in range(1, max_lag + 1):
f_stat = gc_test[lag][0]["ssr_ftest"][0]
p_value = gc_test[lag][0]["ssr_ftest"][1]
results["granger_test"][lag] = {
"f_statistic": f_stat,
"p_value": p_value,
"significant": p_value < 0.05
}
except Exception as e:
results["error"] = str(e)
else:
results["note"] = "Data not stationary, differencing required"
return results
def var_model(self, lag_order: int = 3) -> dict:
"""
Fit Vector Autoregression model to sentiment and returns.
Captures simultaneous feedback between variables.
"""
if self.merged_df is None:
self.merge_data()
model_data = self.merged_df[["weighted_score_mean", "returns"]].dropna()
try:
model = VAR(model_data)
results = model.fit(lag_order)
return {
"aic": results.aic,
"bic": results.bic,
"hqic": results.hqic,
"fpe": results.fpe,
"coefficients": results.params.to_dict(),
"summary": str(results.summary())
}
except Exception as e:
return {"error": str(e)}
def plot_analysis(self, save_path: str = "sentiment_analysis.png"):
"""Generate visualization of sentiment-price relationship."""
if self.merged_df is None:
self.merge_data()
fig, axes = plt.subplots(3, 1, figsize=(14, 10))
# Plot 1: Price and sentiment over time
ax1 = axes[0]
ax1_twin = ax1.twinx()
ax1.plot(self.merged_df.index, self.merged_df["close"],
'b-', label='Price', linewidth=1.5)
ax1_twin.plot(self.merged_df.index, self.merged_df["weighted_score_mean"],
'r-', alpha=0.7, label='Sentiment')
ax1.set_ylabel('Price (USD)')
ax1_twin.set_ylabel('Sentiment Score', color='r')
ax1.set_title('Price vs. Sentiment Over Time')
ax1.legend(loc='upper left')
ax1_twin.legend(loc='upper right')
# Plot 2: Rolling correlation
ax2 = axes[1]
rolling_corr = self.merged_df["weighted_score_mean"].rolling(20).corr(
self.merged_df["returns"]
)
ax2.plot(self.merged_df.index, rolling_corr, 'g-', linewidth=1.5)
ax2.axhline(y=0, color='black', linestyle='--', alpha=0.5)
ax2.set_ylabel('Correlation')
ax2.set_title('20-Period Rolling Correlation: Sentiment vs. Returns')
ax2.fill_between(self.merged_df.index, rolling_corr, 0,
where=(rolling_corr > 0), alpha=0.3, color='green')
ax2.fill_between(self.merged_df.index, rolling_corr, 0,
where=(rolling_corr < 0), alpha=0.3, color='red')
# Plot 3: Scatter with regression
ax3 = axes[2]
ax3.scatter(self.merged_df["weighted_score_mean"],
self.merged_df["returns"], alpha=0.5)
# Add regression line
valid_data = self.merged_df[["weighted_score_mean", "returns"]].dropna()
z = np.polyfit(valid_data["weighted_score_mean"],
valid_data["returns"], 1)
p = np.poly1d(z)
x_line = np.linspace(valid_data["weighted_score_mean"].min(),
valid_data["weighted_score_mean"].max(), 100)
ax3.plot(x_line, p(x_line), "r--", alpha=0.8,
label=f'Regression: y={z[0]:.4f}x+{z[1]:.4f}')
ax3.set_xlabel('Sentiment Score')
ax3.set_ylabel('Returns')
ax3.set_title('Sentiment vs. Returns Scatter Plot')
ax3.legend()
plt.tight_layout()
plt.savefig(save_path, dpi=150)
plt.close()
return save_path
Example usage with synthetic data
if __name__ == "__main__":
# Generate synthetic data for demonstration
np.random.seed(42)
dates = pd.date_range(start="2024-01-01", periods=500, freq="1H")
synthetic_sentiment = [
{
"timestamp": dates[i],
"label": np.random.choice(["bullish", "bearish", "neutral"],
p=[0.4, 0.3, 0.3]),
"confidence": np.random.uniform(0.6, 0.95),
"text": f"Sample post {i}"
}
for i in range(500)
]
synthetic_prices = [
{
"timestamp": dates[i],
"open": 40000 + np.random.randn() * 500,
"high": 40100 + np.random.randn() * 500,
"low": 39900 + np.random.randn() * 500,
"close": 40000 + np.random.randn() * 500,
"volume": np.random.randint(1000, 10000)
}
for i in range(500)
]
# Run analysis
modeler = SentimentPriceModeler(synthetic_sentiment, synthetic_prices)
merged = modeler.merge_data("1H")
print("Merged DataFrame shape:", merged.shape)
print("\nSample data:")
print(merged.head())
# Calculate correlations
correlations = modeler.calculate_correlations(max_lag=12)
print("\n=== Correlation Analysis ===")
for lag, data in list(correlations["pearson"].items())[:3]:
print(f"Lag {lag}: Pearson r={data['r']:.4f}, p={data['p']:.4f}")
# Granger causality
granger = modeler.granger_causality_test()
print("\n=== Granger Causality ===")
if "granger_test" in granger:
for lag, result in granger["granger_test"].items():
print(f"Lag {lag}: F={result['f_statistic']:.4f}, p={result['p_value']:.4f} "
f"({'Significant' if result['significant'] else 'Not significant'})")
# Generate plot
plot_path = modeler.plot_analysis()
print(f"\nAnalysis plot saved to: {plot_path}")Real-Time Trading Signal Generation
Combining sentiment analysis with price data enables actionable trading signals. The following module generates entry/exit signals based on sentiment thresholds and price momentum:
from dataclasses import dataclass
from enum import Enum
from typing import Optional
import numpy as np
class SignalType(Enum):
STRONG_BUY = "STRONG_BUY"
BUY = "BUY"
HOLD = "HOLD"
SELL = "SELL"
STRONG_SELL = "STRONG_SELL"
@dataclass
class TradingSignal:
signal_type: SignalType
sentiment_score: float
price_momentum: float
confidence: float
reasoning: str
timestamp: str
asset: str
class SignalGenerator:
"""
Generate trading signals based on sentiment analysis and price data.
Combines multiple indicators for robust signal generation.
"""
def __init__(
self,
sentiment_threshold: float = 0.3,
momentum_threshold: float = 0.02,
volume_multiplier: float = 1.5
):
self.sentiment_threshold = sentiment_threshold
self.momentum_threshold = momentum_threshold
self.volume_multiplier = volume_multiplier
def generate_signal(
self,
current_sentiment: float,
sentiment_change: float,
price_momentum: float,
volume_ratio: float,
asset: str,
timestamp: str
) -> TradingSignal:
"""
Generate trading signal based on multiple factors.
Args:
current_sentiment: Current weighted sentiment score (-1 to 1)
sentiment_change: Change in sentiment over lookback period
price_momentum: Recent price return (e.g., 1-hour return)
volume_ratio: Current volume vs. average volume
asset: Trading pair (e.g., "BTC/USDT")
timestamp: Signal generation time
Returns:
TradingSignal with type, confidence, and reasoning
"""
# Initialize scoring
score = 0.0
factors = []
# Factor 1: Absolute sentiment level
if current_sentiment > self.sentiment_threshold:
score += current_sentiment * 0.4
factors.append(f"Strong sentiment ({current_sentiment:.2f})")
elif current_sentiment < -self.sentiment_threshold:
score += current_sentiment * 0.4
factors.append(f"Weak sentiment ({current_sentiment:.2f})")
else:
factors.append(f"Neutral sentiment ({current_sentiment:.2f})")
# Factor 2: Sentiment momentum
if sentiment_change > 0.1:
score += 0.2
factors.append("Improving sentiment trend")
elif sentiment_change < -0.1:
score -= 0.2
factors.append("Deteriorating sentiment trend")
# Factor 3: Price momentum alignment
if price_momentum > self.momentum_threshold and current_sentiment > 0: