# import libraries, APis and LLMs from crewai import Agent, Task, Crew import os from tools.market_data import MarketDataTool from tools.sentiment_tool import SentimentTool from tools.historical_data_tool import HistoricalDataTool from tools.analytics_tool import AnalyticsTool import gradio as gr import warnings warnings.filterwarnings("ignore") # ---------------------- # ENVIRONMENT VARIABLES # ---------------------- OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") SERPER_API_KEY = os.getenv("SERPER_API_KEY") # ---------------------- # TOOLS (structured JSON) # ---------------------- market_data_tool = MarketDataTool() sentiment_tool = SentimentTool() historical_data_tool = HistoricalDataTool() analytics_tool = AnalyticsTool() # ---------------------- # AGENTS # ---------------------- market_agent = Agent( role="Crypto Market Analyst", goal="Fetch live market price using CoinGecko and produce structured market JSON.", backstory=( "A disciplined market analyst who retrieves clean market data " "including price and market liquidity." ), verbose=False, allow_delegations=True, tools=[market_data_tool], llm="gpt-4o-mini" ) sentiment_agent = Agent( role="Crypto Sentiment Analyst", goal="Analyze public sentiment on news & Reddit using Serper + OpenAI. Output structured sentiment JSON.", backstory="An expert in NLP-based crypto sentiment interpretation.", verbose=False, allow_delegations=True, tools=[sentiment_tool], llm="gpt-4.1" ) historical_agent = Agent( role="Crypto Historical Analyst", goal="Analyze long-term price trends, volatility, and movement patterns using clean structured JSON.", backstory="A quantitative analyst specializing in historical trends.", verbose=False, allow_delegations=True, tools=[historical_data_tool], llm="gpt-4o-mini" ) analytics_agent = Agent( role="Cryptocurrency Analytics", goal=( "Combine structured outputs from market, historical, and sentiment tools " "into final analytics including trend, volatility, alignment, and composite scoring." ), backstory="A senior quant strategist synthesizing multiple signals into coherent analytics.", verbose=False, allow_delegations=False, tools=[analytics_tool], llm="gpt-4o-mini" ) strategy_agent = Agent( role="Crypto Strategy Analyst", goal=( "Convert analytics into an actionable trading stance. Integrate sentiment, volatility, " "risk, opportunity, and alignment to produce a structured trading strategy." ), backstory="A crypto strategist experienced in forming actionable plans.", verbose=False, llm="gpt-4.1" ) reporting_agent = Agent( role="Crypto Reporting Analyst", goal=( "Produce a polished, professional, narrative-style market report written in clean Markdown. " "The report must use paragraphs, not bullet points, except where absolutely necessary. " "Each section must read like human financial analysis, with smooth transitions and explanatory context." ), backstory=( "You are a senior market strategist and financial writer for institutional clients. " "Your writing style is polished, narrative, and insight-driven — not bullet-point lists. " "You blend market data, sentiment, and strategy into clear, flowing paragraphs." ), llm="gpt-4.1", verbose=False, ) # ---------------------- # TASKS # ---------------------- market_data_task = Task( description=( "Use the market_data_tool to fetch the live market price for " "{cryptocurrency_selection} in {currency_selection}. " "Return structured JSON output only." ), expected_output="Structured JSON of current market metrics.", agent=market_agent, ) sentiment_task = Task( description=( "Fetch sentiment for {cryptocurrency_selection} using sentiment_tool. " "Return structured JSON with sentiment, reasoning, and the raw headlines used." ), expected_output="Structured sentiment JSON (sentiment, reasoning, headlines).", agent=sentiment_agent, ) historical_data_task = Task( description=( "Use the historical_data_tool to retrieve structured historical price data " "for {cryptocurrency_selection} over {days_selection} days." ), expected_output=( "Structured historical JSON with start_price, end_price, pct_change, " "volatility_pct, and trend." ), agent=historical_agent, ) analytics_task = Task( description=( "Use analytics_tool to combine market_data, historical_data, and sentiment_data " "into final structured analytics including trend, volatility, sentiment, alignment, " "and composite score." ), expected_output="Structured analytics JSON.", agent=analytics_agent, inputs={ "market_data": "{output_of_market_data_task}", "historical_data": "{output_of_historical_data_task}", "sentiment_data": "{output_of_sentiment_data_task}", } ) strategy_task = Task( description=( "Based on the structured analytics, produce an actionable trading stance " "including: bias, strategy, risk guidance, and rationale." ), expected_output="Structured strategy JSON.", agent=strategy_agent, ) reporting_task = Task( description=( "Using the outputs from all previous tasks, write a cohesive, high-quality market report " "in Markdown format. Each section must be written as narrative paragraphs, not bullet or numbered lists. " "You must synthesize insights, explain context, and avoid repeating raw field names. " "Required sections with H2 Markdown headings (##):\n\n" "## Market Overview\n" "Explain current price, recent movement, and key context in paragraph form.\n\n" "## Historical Performance\n" "Summarize recent price trajectory, volatility, and trend using smooth narrative sentences.\n\n" "## Sentiment Analysis\n" "Describe market sentiment, referencing sentiment sources in prose, not lists.\n\n" "## Analytical Summary\n" "Interpret composite metrics and explain what they mean for traders.\n\n" "## Strategy Outlook\n" "Provide clear strategy insights in paragraph format — no bullet points.\n\n" "## Final Takeaways\n" "Conclude with high-level insights in polished prose.\n\n" "IMPORTANT: Do NOT output bullet points or lists. Use flowing paragraphs." ), expected_output="A polished, narrative Markdown report with paragraphs only.", agent=reporting_agent, ) # ---------------------- # CREW EXECUTION PIPELINE # ---------------------- crypto_analysis_crew = Crew( agents=[ market_agent, historical_agent, sentiment_agent, analytics_agent, strategy_agent, reporting_agent ], tasks=[ market_data_task, historical_data_task, sentiment_task, analytics_task, strategy_task, reporting_task ], process="sequential", verbose=True ) # ---------------------- # GRADIO UI HANDLER # ---------------------- def generate_report(crypto_name, currency, days): crypto_inputs = { "cryptocurrency_selection": crypto_name.lower(), "currency_selection": currency.lower(), "days_selection": int(days) } # Disable button at start yield "", gr.update(interactive=False) # Run workflow result = crypto_analysis_crew.kickoff(inputs=crypto_inputs) final = result["final_output"] if isinstance(result, dict) and "final_output" in result else str(result) # Return report + re-enable button yield final, gr.update(interactive=True) # ---------------------- # GRADIO APP # ---------------------- with gr.Blocks(theme=gr.themes.Monochrome()) as app: gr.Markdown("# 🪙 Crypto Intelligence Dashboard") gr.Markdown("Run a full multi-agent crypto analysis using structured JSON tools.") with gr.Row(): crypto = gr.Textbox( label="Cryptocurrency (e.g. bitcoin, ethereum)", placeholder="Type cryptocurrency name...", value="bitcoin" ) currency = gr.Dropdown( ["usd", "eur", "gbp"], label="Currency", value="usd" ) days = gr.Slider( minimum=30, maximum=730, value=365, step=15, label="Historical Lookback (days)" ) # ← original working button run_button = gr.Button("🚀 Run Full Analysis", variant="primary") report_output = gr.Markdown(label="📊 Intelligence Report") run_button.click( fn=generate_report, inputs=[crypto, currency, days], outputs=[report_output, run_button] ) if __name__ == "__main__": app.launch(server_name="0.0.0.0", server_port=7860)