Spaces:
Sleeping
Sleeping
File size: 5,192 Bytes
a0c219b d872892 a0c219b d872892 1dece25 a0c219b a0a8d76 5d2f635 a0a8d76 d872892 a0c219b 1dece25 a0c219b 5d2f635 a0c219b 5d2f635 1dece25 a0c219b 5d2f635 a0c219b 5d2f635 1dece25 5d2f635 a0c219b 5d2f635 a0c219b 5d2f635 1dece25 a0c219b 5d2f635 a0c219b 1dece25 a0c219b 1dece25 a0c219b 1dece25 5d2f635 1dece25 a0c219b 5d2f635 a0c219b 1dece25 a0c219b 1dece25 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 |
from crewai.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Dict, Any, Type
# ---------- Input Schema ----------
class AnalyticsInput(BaseModel):
market_data: str | Dict[str, Any] = Field(..., description="Structured JSON from MarketDataTool")
historical_data: str | Dict[str, Any] = Field(..., description="Structured JSON from HistoricalDataTool")
sentiment_data: str | Dict[str, Any] = Field(..., description="Structured JSON from SentimentTool")
# ---------- Tool ----------
class AnalyticsTool(BaseTool):
name: str = "analytics_tool"
description: str = (
"Aggregates structured market, historical, and sentiment data to produce "
"quantitative indicators including pct_change, volatility, trend, sentiment, "
"sentiment_strength, confidence, alignment, and a composite score."
)
args_schema: Type[BaseModel] = AnalyticsInput
def _run(self, market_data: dict, historical_data: dict, sentiment_data: dict) -> dict:
try:
# ============================================================
# 1) Extract fields safely from structured tool outputs
# ============================================================
price = market_data.get("latest_price")
pct_change = historical_data.get("pct_change")
volatility = historical_data.get("volatility_pct")
trend = historical_data.get("trend")
sentiment = sentiment_data.get("sentiment")
if price is None or pct_change is None or trend is None or sentiment is None:
return {
"error": (
"Missing required fields in analytics input. "
"Ensure all tools returned structured JSON."
)
}
sentiment = sentiment.lower()
# ============================================================
# 2) Sentiment strength & confidence (new)
# ============================================================
# Pull from SentimentTool if present
sentiment_strength = sentiment_data.get("sentiment_strength")
sentiment_confidence = sentiment_data.get("confidence")
# ---- Backwards-compatible defaults ----
if sentiment_strength is None:
sentiment_strength = {
"bullish": 0.7,
"neutral": 0.0,
"bearish": -0.7
}.get(sentiment, 0.0)
if sentiment_confidence is None:
# Basic proxy confidence using number of headlines/comments
news_count = len(sentiment_data.get("news_headlines", []))
reddit_count = len(sentiment_data.get("reddit_comments", []))
sources = news_count + reddit_count
sentiment_confidence = min(1.0, 0.2 + 0.1 * sources)
# Effective weighted sentiment
effective_sentiment = sentiment_strength * sentiment_confidence
# ============================================================
# 3) Alignment logic (upgraded)
# ============================================================
aligned = (
(trend == "upward" and effective_sentiment > 0.2) or
(trend == "downward" and effective_sentiment < -0.2)
)
# ============================================================
# 4) Composite score (new formula)
# ============================================================
score = (
(pct_change / 10) + # Trend effect
(effective_sentiment * 1.5) - # Strong weight for sentiment
(volatility / 100 if volatility else 0) # Penalize volatility
)
# Bound between [-1, 1]
score = round(max(-1, min(1, score)), 2)
# ============================================================
# 5) Final structured output
# ============================================================
return {
"price": price,
"pct_change": pct_change,
"volatility_pct": volatility,
"trend": trend,
"sentiment": sentiment,
"sentiment_strength": round(sentiment_strength, 3),
"sentiment_confidence": round(sentiment_confidence, 3),
"effective_sentiment": round(effective_sentiment, 3),
"alignment": "aligned" if aligned else "divergent",
"composite_score": score,
"summary": (
f"Trend={trend}, Sentiment={sentiment}, "
f"Strength={round(sentiment_strength,3)}, "
f"Confidence={round(sentiment_confidence,3)}, "
f"Alignment={'aligned' if aligned else 'divergent'}, "
f"Score={score}"
),
}
except Exception as e:
return {"error": f"AnalyticsTool failed: {str(e)}"} |