*What to look for: We are looking for companies that consistently grow their revenue year after year, ideally. Consistent growth indicates market acceptance and business expansion.*
### Free Cash Flow per Share*What to look for: Look for companies with consistently high and growing free cash flow. High FCF indicates a company has strong financial health and flexibility for reinvestment, debt reduction, or shareholder returns.*
### Shares Outstanding*What to look for: Ideally, look for a declining trend in shares outstanding. This suggests the company is buying back its own shares, which can increase shareholder value by reducing the number of shares in circulation.*
""" chart_insights = gr.Markdown(value=chart_insights_text, visible=False, label="Chart Insights") download_button = gr.DownloadButton( label="Download Analysis as TXT", visible=False, variant="secondary" ) # Hidden state to store analysis text for download analysis_text_state = gr.State() # Event handlers generate_btn.click( fn=self.generate_report, inputs=[ticker_input], outputs=[ output, revenue_plot, fcf_plot, shares_plot, analysis_text_state, download_button, loading_status, chart_insights ] ) ticker_input.submit( fn=self.generate_report, inputs=[ticker_input], outputs=[ output, revenue_plot, fcf_plot, shares_plot, analysis_text_state, download_button, loading_status, chart_insights ] ) download_button.click( fn=self._save_and_return_analysis_file, inputs=[analysis_text_state, ticker_input], outputs=[download_button], show_progress="hidden" ) # Event handler for connection button connection_btn.click( fn=self._test_mcp_connection, inputs=[], outputs=[connection_status] ) vllm_connection_btn.click( fn=self._test_vllm_connection, inputs=[], outputs=[vllm_connection_status] ) return app def _save_and_return_analysis_file(self, analysis_text: str, ticker: str): """Saves the analysis text to a file and returns the path for download.""" if not analysis_text: return None # No file to download if no analysis text # Ensure reports directory exists reports_dir = "reports" os.makedirs(reports_dir, exist_ok=True) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") file_path = os.path.join(reports_dir, f"{ticker}_analysis_{timestamp}.txt") with open(file_path, "w") as f: f.write(analysis_text) return file_path def generate_report(self, ticker): """Generate stock analysis report.""" if not ticker or not ticker.strip(): return ( gr.update(value="Error: Please enter a valid stock ticker", visible=True), gr.update(value=None, visible=False), gr.update(value=None, visible=False), gr.update(value=None, visible=False), None, gr.update(visible=False), gr.update(value="❌ Error: Please enter a valid stock ticker", visible=True), gr.update(visible=False) ) try: # Show loading state yield ( gr.update(value="⏳ Generating report... Please wait.", visible=True), gr.update(value=None, visible=False), gr.update(value=None, visible=False), gr.update(value=None, visible=False), None, gr.update(visible=False), gr.update(value="⏳ Analyzing stock data and generating insights...", visible=True), gr.update(visible=False) ) # Call the Modal MCP server for analysis headers = {'Content-Type': 'application/json'} payload = {"ticker": ticker} response = requests.post(f"{self.mcp_server_url}/analyze", headers=headers, json=payload) response.raise_for_status() analysis_data = response.json() # Extract data from the response analysis_results = { "analysis": analysis_data["analysis"], "revenue_chart": pio.from_json(analysis_data["revenue_chart"]) if analysis_data["revenue_chart"] else None, "fcf_chart": pio.from_json(analysis_data["fcf_chart"]) if analysis_data["fcf_chart"] else None, "shares_chart": pio.from_json(analysis_data["shares_chart"]) if analysis_data["shares_chart"] else None } # Format the analysis text with a timestamp timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") formatted_analysis = f"## Analysis Report for {ticker.upper()}\n*Generated on {timestamp}*\n\n{analysis_results['analysis']}" yield ( gr.update(value=formatted_analysis, visible=True), gr.update(value=analysis_results["revenue_chart"], visible=True), gr.update(value=analysis_results["fcf_chart"], visible=True), gr.update(value=analysis_results["shares_chart"], visible=True), formatted_analysis, gr.update(label=f"Download {ticker.upper()} Report", visible=True), gr.update(value="✅ Analysis complete!", visible=True), gr.update(visible=True) ) except requests.exceptions.RequestException as e: error_msg = f"Error analyzing stock: {str(e)}" if "429" in str(e): error_msg = "Rate limit exceeded. Please try again in a few minutes." elif "500" in str(e): error_msg = "Server error. Please try again later." yield ( gr.update(value=error_msg, visible=True), gr.update(value=None, visible=False), gr.update(value=None, visible=False), gr.update(value=None, visible=False), None, gr.update(visible=False), gr.update(value=f"❌ {error_msg}", visible=True), gr.update(visible=False) ) def _test_mcp_connection(self): """Test connection to Modal MCP server health endpoint.""" try: response = requests.get(f"{self.mcp_server_url}/health", timeout=10) if response.status_code == 200: return f"✅ Connected to Modal MCP Server: {response.json().get('status', 'OK')}" else: return f"❌ Connection failed (HTTP {response.status_code}): {response.text}" except requests.exceptions.RequestException as e: return f"❌ Connection error: {str(e)}" def _test_vllm_connection(self): """Tests the connection to the Modal vLLM service via the MCP server.""" try: response = requests.get(f"{self.mcp_server_url}/test_llm_connection") response.raise_for_status() # Raise an HTTPError for bad responses (4xx or 5xx) status_data = response.json() return f"Status: {status_data.get('status', 'Unknown')}" except requests.exceptions.RequestException as e: return f"Error calling vLLM test endpoint: {e}" def main(): """Main entry point for the application.""" app_instance = AgenticStockAdvisorApp() app = app_instance.create_ui() app.launch(server_name="0.0.0.0", server_port=7860, mcp_server=True) if __name__ == "__main__": main()