Spaces:
Running
Running
| import asyncio | |
| import time | |
| import re | |
| import json | |
| from crewai import Crew, Process | |
| from textwrap import dedent | |
| from crypto_analysis_agents import CryptoAnalysisAgents | |
| from crypto__analysis_tasks import CryptoAnalysisTasks | |
| class CryptoCrew: | |
| def __init__(self, crypto): | |
| self.crypto = crypto | |
| self.agents_instance = CryptoAnalysisAgents() | |
| self.tasks_instance = CryptoAnalysisTasks() | |
| def run(self): | |
| return asyncio.run(self.run_async()) | |
| async def run_async(self): | |
| start_time = time.time() | |
| try: | |
| # Create agents | |
| market_analyst = self.agents_instance.market_analyst() | |
| technical_analyst = self.agents_instance.technical_analyst() | |
| advisor = self.agents_instance.crypto_advisor() | |
| # Create tasks | |
| tasks = [ | |
| self.tasks_instance.market_research(market_analyst, self.crypto), | |
| self.tasks_instance.technical_analysis(technical_analyst, self.crypto), | |
| self.tasks_instance.sentiment_analysis(advisor, self.crypto), | |
| self.tasks_instance.recommend(advisor, self.crypto) | |
| ] | |
| # Enhanced crew configuration | |
| crew = Crew( | |
| agents=[market_analyst, technical_analyst, advisor], | |
| tasks=tasks, | |
| verbose=True, # Enable for better debugging | |
| process=Process.sequential, | |
| max_iterations=8, | |
| task_timeout=90 | |
| ) | |
| # Run crew | |
| result = await asyncio.to_thread(crew.kickoff) | |
| end_time = time.time() | |
| return self.parse_result(result, end_time - start_time) | |
| except Exception as e: | |
| execution_time = time.time() - start_time | |
| return { | |
| "summary": f"Analysis failed: {str(e)}", | |
| "market_data": self._get_fallback_market_data(), | |
| "technical_data": self._get_fallback_technical_data(), | |
| "sentiment": self._get_fallback_sentiment(), | |
| "recommendation": {"action": "HOLD", "confidence": "Low", "reasoning": "Analysis incomplete"}, | |
| "execution_time": execution_time, | |
| "risk_assessment": "High - Analysis failed" | |
| } | |
| def parse_result(self, result, execution_time): | |
| """Enhanced parsing to extract structured data from LLM responses""" | |
| result_str = str(result) | |
| # Extract market data | |
| market_data = self._extract_market_data(result_str) | |
| # Extract technical analysis | |
| technical_data = self._extract_technical_data(result_str) | |
| # Extract detailed sentiment analysis | |
| sentiment_analysis = self._extract_sentiment_analysis(result_str) | |
| # Extract recommendation | |
| recommendation = self._extract_recommendation(result_str) | |
| # Extract risk assessment | |
| risk_assessment = self._extract_risk_assessment(result_str) | |
| return { | |
| "summary": self._clean_summary(result_str), | |
| "market_data": market_data, | |
| "technical_data": technical_data, | |
| "sentiment": sentiment_analysis, | |
| "recommendation": recommendation, | |
| "risk_assessment": risk_assessment, | |
| "execution_time": execution_time, | |
| "last_updated": time.strftime("%Y-%m-%d %H:%M:%S UTC", time.gmtime()) | |
| } | |
| def _extract_market_data(self, text): | |
| """Extract market metrics from analysis""" | |
| market_data = { | |
| "current_price": "N/A", | |
| "market_cap": "N/A", | |
| "price_change_24h": "N/A", | |
| "price_change_7d": "N/A", | |
| "volume_24h": "N/A", | |
| "market_dominance": "N/A" | |
| } | |
| # Extract price | |
| price_match = re.search(r'\$([0-9,]+\.?[0-9]*)', text) | |
| if price_match: | |
| market_data["current_price"] = f"${price_match.group(1)}" | |
| # Extract market cap | |
| mcap_match = re.search(r'market cap[:\s]+\$([0-9,]+\.?[0-9]*[BMK]?)', text, re.IGNORECASE) | |
| if mcap_match: | |
| market_data["market_cap"] = f"${mcap_match.group(1)}" | |
| # Extract percentage changes | |
| change_24h = re.search(r'24h?[:\s]*([+-]?[0-9]+\.?[0-9]*%)', text, re.IGNORECASE) | |
| if change_24h: | |
| market_data["price_change_24h"] = change_24h.group(1) | |
| change_7d = re.search(r'7d?[:\s]*([+-]?[0-9]+\.?[0-9]*%)', text, re.IGNORECASE) | |
| if change_7d: | |
| market_data["price_change_7d"] = change_7d.group(1) | |
| return market_data | |
| def _extract_technical_data(self, text): | |
| """Extract technical indicators""" | |
| technical_data = { | |
| "rsi": "N/A", | |
| "rsi_signal": "Neutral", | |
| "moving_average_7d": "N/A", | |
| "moving_average_50d": "N/A", | |
| "trend": "Neutral", | |
| "support_level": "N/A", | |
| "resistance_level": "N/A" | |
| } | |
| # Extract RSI | |
| rsi_match = re.search(r'RSI[:\s]*([0-9]+\.?[0-9]*)', text, re.IGNORECASE) | |
| if rsi_match: | |
| rsi_value = float(rsi_match.group(1)) | |
| technical_data["rsi"] = str(rsi_value) | |
| if rsi_value > 70: | |
| technical_data["rsi_signal"] = "Overbought" | |
| elif rsi_value < 30: | |
| technical_data["rsi_signal"] = "Oversold" | |
| else: | |
| technical_data["rsi_signal"] = "Neutral" | |
| # Extract moving averages | |
| ma_match = re.search(r'(?:7-day )?MA[:\s]*\$([0-9,]+\.?[0-9]*)', text, re.IGNORECASE) | |
| if ma_match: | |
| technical_data["moving_average_7d"] = f"${ma_match.group(1)}" | |
| # Determine trend | |
| if "bullish" in text.lower() or "uptrend" in text.lower(): | |
| technical_data["trend"] = "Bullish" | |
| elif "bearish" in text.lower() or "downtrend" in text.lower(): | |
| technical_data["trend"] = "Bearish" | |
| return technical_data | |
| def _extract_sentiment_analysis(self, text): | |
| """Extract differentiated sentiment analysis""" | |
| # Default to varied sentiments for demonstration | |
| sentiment_data = { | |
| "overall": "Neutral", | |
| "social_media": "Neutral", | |
| "news": "Neutral", | |
| "community": "Neutral" | |
| } | |
| # Extract overall sentiment | |
| if re.search(r'overall.*positive|positive.*overall', text, re.IGNORECASE): | |
| sentiment_data["overall"] = "Positive" | |
| elif re.search(r'overall.*negative|negative.*overall', text, re.IGNORECASE): | |
| sentiment_data["overall"] = "Negative" | |
| elif "bullish" in text.lower(): | |
| sentiment_data["overall"] = "Positive" | |
| elif "bearish" in text.lower(): | |
| sentiment_data["overall"] = "Negative" | |
| # Extract social media sentiment | |
| if re.search(r'social.*positive|twitter.*positive|reddit.*positive', text, re.IGNORECASE): | |
| sentiment_data["social_media"] = "Positive" | |
| elif re.search(r'social.*negative|twitter.*negative|reddit.*negative', text, re.IGNORECASE): | |
| sentiment_data["social_media"] = "Negative" | |
| elif re.search(r'social.*bullish|community.*optimistic', text, re.IGNORECASE): | |
| sentiment_data["social_media"] = "Positive" | |
| # Extract news sentiment | |
| if re.search(r'news.*positive|headlines.*positive|media.*positive', text, re.IGNORECASE): | |
| sentiment_data["news"] = "Positive" | |
| elif re.search(r'news.*negative|headlines.*negative|regulatory.*concern', text, re.IGNORECASE): | |
| sentiment_data["news"] = "Negative" | |
| # Extract community sentiment | |
| if re.search(r'community.*positive|development.*active|adoption.*growing', text, re.IGNORECASE): | |
| sentiment_data["community"] = "Positive" | |
| elif re.search(r'community.*negative|development.*slow|adoption.*declining', text, re.IGNORECASE): | |
| sentiment_data["community"] = "Negative" | |
| elif re.search(r'institutional.*adoption|enterprise.*adoption', text, re.IGNORECASE): | |
| sentiment_data["community"] = "Positive" | |
| return sentiment_data | |
| def _extract_recommendation(self, text): | |
| """Extract investment recommendation with reasoning""" | |
| recommendation = { | |
| "action": "HOLD", | |
| "confidence": "Medium", | |
| "reasoning": "Standard analysis completed", | |
| "time_horizon": "Medium-term", | |
| "risk_level": "Moderate" | |
| } | |
| # Extract recommendation | |
| if re.search(r'recommendation[:\s]*BUY|BUY.*recommendation', text, re.IGNORECASE): | |
| recommendation["action"] = "BUY" | |
| elif re.search(r'recommendation[:\s]*SELL|SELL.*recommendation', text, re.IGNORECASE): | |
| recommendation["action"] = "SELL" | |
| # Extract confidence | |
| if re.search(r'confidence[:\s]*high|high.*confidence', text, re.IGNORECASE): | |
| recommendation["confidence"] = "High" | |
| elif re.search(r'confidence[:\s]*low|low.*confidence', text, re.IGNORECASE): | |
| recommendation["confidence"] = "Low" | |
| # Extract reasoning | |
| reason_match = re.search(r'(?:reason|reasoning)[:\s]*([^.]+)', text, re.IGNORECASE) | |
| if reason_match: | |
| recommendation["reasoning"] = reason_match.group(1).strip() | |
| return recommendation | |
| def _extract_risk_assessment(self, text): | |
| """Extract risk assessment""" | |
| if re.search(r'high.*risk|risk.*high|volatile|risky', text, re.IGNORECASE): | |
| return "High Risk" | |
| elif re.search(r'low.*risk|risk.*low|stable|conservative', text, re.IGNORECASE): | |
| return "Low Risk" | |
| else: | |
| return "Moderate Risk" | |
| def _clean_summary(self, text): | |
| """Clean and format the summary""" | |
| # Remove excess whitespace and format | |
| summary = re.sub(r'\s+', ' ', text).strip() | |
| # Truncate if too long | |
| if len(summary) > 1000: | |
| summary = summary[:1000] + "..." | |
| return summary | |
| def _get_fallback_market_data(self): | |
| return { | |
| "current_price": "N/A", | |
| "market_cap": "N/A", | |
| "price_change_24h": "N/A", | |
| "price_change_7d": "N/A", | |
| "volume_24h": "N/A", | |
| "market_dominance": "N/A" | |
| } | |
| def _get_fallback_technical_data(self): | |
| return { | |
| "rsi": "N/A", | |
| "rsi_signal": "Neutral", | |
| "moving_average_7d": "N/A", | |
| "moving_average_50d": "N/A", | |
| "trend": "Neutral", | |
| "support_level": "N/A", | |
| "resistance_level": "N/A" | |
| } | |
| def _get_fallback_sentiment(self): | |
| return { | |
| "overall": "Neutral", | |
| "social_media": "Neutral", | |
| "news": "Neutral", | |
| "community": "Neutral" | |
| } | |
| if __name__ == "__main__": | |
| print("## Welcome to Enhanced Crypto Analysis Crew") | |
| print('-------------------------------') | |
| crypto = input(dedent(""" | |
| What is the cryptocurrency you want to analyze? | |
| """)) | |
| crypto_crew = CryptoCrew(crypto) | |
| result = crypto_crew.run() | |
| print("\n\n########################") | |
| print("## Here is the Enhanced Report") | |
| print("########################\n") | |
| print(json.dumps(result, indent=2)) | |