# app.py
# NAVADA (Startup Viability Agent) - A Chainlit-powered AI agent for analyzing startup risk and failure patterns
# Features: Investor Mode, Founder Mode, UK Economist Mode with macroeconomic analysis
print("DEBUG: Starting app.py imports", flush=True)
# =============================
# IMPORTS
# =============================
import io # For in-memory file operations (byte streams)
import os # Operating system interface for environment variables and file operations
import time # Time-related functions for delays and timing operations
import re # Regular expressions for pattern matching and text processing
import requests # HTTP library for making API calls
from datetime import datetime # Date/time handling for timestamps
import datetime as dt # Full datetime module for compatibility
from typing import Dict, List, Optional, Any # Type hints for better code documentation
import uuid # UUID generation for unique identifiers
import math # Mathematical operations (currently unused but available)
import json # JSON parsing (currently unused but available)
import asyncio # Async/await support for concurrent operations
import logging # Logging system for error tracking and debugging
import traceback # Detailed error tracebacks for debugging
import functools # Function decorators and utilities
from document_templates import DocumentTemplateGenerator # Document template generation
from financial_data_integration import FinancialDataIntegrator # Real-time financial data
from document_intelligence import DocumentIntelligenceEngine # AI document intelligence
from advanced_analytics import AdvancedAnalyticsDashboard # Advanced analytics
from voice_streaming import VoiceStreamingManager, VoiceUIManager # OpenAI voice streaming
from startup_coach_personas import StartupCoachPersonas, PersonaUIManager # AI coaching personas
from market_intelligence_dashboard import MarketIntelligenceDashboard # Market intelligence
print("DEBUG: Importing chainlit", flush=True)
import chainlit as cl # Chainlit framework for building conversational AI interfaces
print("DEBUG: Chainlit imported", flush=True)
import pandas as pd # Data manipulation and analysis with DataFrames
import numpy as np # Numerical operations for calculations
print("DEBUG: Importing matplotlib", flush=True)
import matplotlib
matplotlib.use('Agg') # Use non-interactive backend for server environments
import matplotlib.pyplot as plt # Core plotting library for creating visualizations
print("DEBUG: Matplotlib imported", flush=True)
import seaborn as sns # Statistical data visualization built on matplotlib
import plotly.express as px # Interactive plotting library for dynamic visualizations
import plotly.graph_objects as go # Low-level plotly interface for custom charts
import plotly.io as pio # Plotly I/O utilities for saving/converting charts
from plotly.subplots import make_subplots # Create subplot layouts for dashboards
import requests # HTTP library for making web requests and scraping
from bs4 import BeautifulSoup # HTML/XML parser for web scraping
from urllib.parse import urlparse # URL validation and parsing utilities
import re # Regular expressions for text processing and validation
import scipy.stats as stats # Statistical functions for analysis
import random # Random number generation for Monte Carlo simulations
from typing import Dict, Any, List, Optional # Type hints for better code documentation
from dataclasses import dataclass, field # For structured data classes
from openai import OpenAI # OpenAI API client for GPT model interactions
from dotenv import load_dotenv # Load environment variables from .env file
import uuid # For generating unique thread/session IDs
# Optional ML imports - handle gracefully if not available (for Vercel size limits)
try:
from sklearn.model_selection import train_test_split # Split data for ML training
from sklearn.ensemble import RandomForestClassifier # Random Forest model for predictions
SKLEARN_AVAILABLE = True
except ImportError:
SKLEARN_AVAILABLE = False
print("WARNING: scikit-learn not available - ML features disabled")
# LangChain & LangSmith imports for hosting
print("DEBUG: Importing langchain_openai", flush=True)
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
print("DEBUG: langchain_openai imported", flush=True)
print("DEBUG: Importing langchain_core.documents", flush=True)
from langchain_core.documents import Document
print("DEBUG: langchain_core.documents imported", flush=True)
from langchain.text_splitter import RecursiveCharacterTextSplitter
# Optional vector store imports - handle gracefully if not available
try:
from langchain_chroma import Chroma
from langchain.chains import RetrievalQA
CHROMA_AVAILABLE = True
except ImportError:
CHROMA_AVAILABLE = False
print("WARNING: Chroma vector store not available - RAG features disabled")
from langsmith import traceable, Client as LangSmithClient
from langsmith.wrappers import wrap_openai
import langsmith as ls
print("DEBUG: All imports complete", flush=True)
# =============================
# ERROR HANDLING & LOGGING SETUP
# =============================
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(message)s',
handlers=[
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
def safe_api_call(func):
"""Decorator for safe API calls with comprehensive error handling."""
@functools.wraps(func)
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except requests.exceptions.RequestException as e:
logger.error(f"Network error in {func.__name__}: {e}")
return {"error": f"Network error: {str(e)}", "success": False}
except ValueError as e:
logger.error(f"Value error in {func.__name__}: {e}")
return {"error": f"Invalid data: {str(e)}", "success": False}
except KeyError as e:
logger.error(f"Missing key in {func.__name__}: {e}")
return {"error": f"Missing required field: {str(e)}", "success": False}
except Exception as e:
logger.error(f"Unexpected error in {func.__name__}: {e}")
logger.error(f"Traceback: {traceback.format_exc()}")
return {"error": f"Unexpected error: {str(e)}", "success": False}
return wrapper
def safe_async_api_call(func):
"""Decorator for safe async API calls with comprehensive error handling."""
@functools.wraps(func)
async def wrapper(*args, **kwargs):
try:
return await func(*args, **kwargs)
except requests.exceptions.RequestException as e:
logger.error(f"Network error in {func.__name__}: {e}")
return {"error": f"Network error: {str(e)}", "success": False}
except ValueError as e:
logger.error(f"Value error in {func.__name__}: {e}")
return {"error": f"Invalid data: {str(e)}", "success": False}
except KeyError as e:
logger.error(f"Missing key in {func.__name__}: {e}")
return {"error": f"Missing required field: {str(e)}", "success": False}
except Exception as e:
logger.error(f"Unexpected error in {func.__name__}: {e}")
logger.error(f"Traceback: {traceback.format_exc()}")
return {"error": f"Unexpected error: {str(e)}", "success": False}
return wrapper
def validate_environment():
"""Validate environment setup and dependencies."""
logger.info("Starting environment validation...")
issues = []
# Check critical environment variables
required_vars = ["OPENAI_API_KEY"]
for var in required_vars:
if not os.getenv(var) or os.getenv(var) == "your_openai_api_key_here":
issues.append(f"Missing or placeholder value for {var}")
# Check optional but recommended environment variables
optional_vars = {
"LANGSMITH_API_KEY": "LangSmith tracing",
"SEARCH_API_KEY": "web search functionality",
"TTS_PROMPT_ID": "text-to-speech features",
"LANGCHAIN_DATABASE_ID": "vector database features"
}
for var, feature in optional_vars.items():
value = os.getenv(var)
if not value or value.startswith("your_"):
logger.warning(f"Optional {var} not configured - {feature} will be disabled")
# Check critical imports
try:
import matplotlib.pyplot as plt
logger.info("✅ Matplotlib available")
except ImportError:
issues.append("Matplotlib not available - chart generation will fail")
try:
from langchain_chroma import Chroma
logger.info("✅ LangChain Chroma available")
except ImportError:
issues.append("LangChain Chroma not available - RAG features disabled")
try:
import openai
logger.info("✅ OpenAI library available")
except ImportError:
issues.append("OpenAI library not available - core functionality will fail")
if issues:
logger.error("Environment validation failed:")
for issue in issues:
logger.error(f" - {issue}")
return False, issues
else:
logger.info("✅ Environment validation passed")
return True, []
def create_startup_health_check():
"""Perform comprehensive health checks during startup."""
logger.info("🔍 Performing startup health checks...")
# Get environment variables for health checks
api_key_check = os.getenv("OPENAI_API_KEY")
langsmith_api_key_check = os.getenv("LANGSMITH_API_KEY")
search_api_key_check = os.getenv("SEARCH_API_KEY")
health_status = {
"overall": True,
"checks": {},
"warnings": [],
"errors": []
}
# Test matplotlib chart generation
try:
import matplotlib.pyplot as plt
fig, ax = plt.subplots(figsize=(1, 1))
ax.plot([1, 2], [1, 2])
buf = io.BytesIO()
fig.savefig(buf, format='png')
plt.close(fig)
health_status["checks"]["matplotlib"] = "✅ Working"
logger.info("✅ Matplotlib chart generation test passed")
except Exception as e:
health_status["checks"]["matplotlib"] = f"❌ Failed: {e}"
health_status["errors"].append(f"Matplotlib test failed: {e}")
health_status["overall"] = False
logger.error(f"❌ Matplotlib test failed: {e}")
# Test OpenAI client initialization
try:
if api_key_check and api_key_check != "your_openai_api_key_here":
# Don't make actual API call, just test client creation
from openai import OpenAI
test_client = OpenAI(api_key=api_key_check)
health_status["checks"]["openai"] = "✅ Client initialized"
logger.info("✅ OpenAI client initialization test passed")
else:
health_status["checks"]["openai"] = "⚠️ No valid API key"
health_status["warnings"].append("OpenAI API key not configured")
logger.warning("⚠️ OpenAI API key not configured")
except Exception as e:
health_status["checks"]["openai"] = f"❌ Failed: {e}"
health_status["errors"].append(f"OpenAI client test failed: {e}")
logger.error(f"❌ OpenAI client test failed: {e}")
# Test vector store functionality
try:
if CHROMA_AVAILABLE:
# Test basic Chroma functionality without creating actual store
from langchain_chroma import Chroma
health_status["checks"]["vector_store"] = "✅ Available"
logger.info("✅ Vector store (Chroma) available")
else:
health_status["checks"]["vector_store"] = "❌ Not available"
health_status["warnings"].append("Vector store not available - RAG features disabled")
logger.warning("⚠️ Vector store not available")
except Exception as e:
health_status["checks"]["vector_store"] = f"❌ Failed: {e}"
health_status["errors"].append(f"Vector store test failed: {e}")
logger.error(f"❌ Vector store test failed: {e}")
# Test search API configuration
if search_api_key_check and search_api_key_check != "your_brave_search_api_key_here":
health_status["checks"]["search_api"] = "✅ Configured"
logger.info("✅ Search API key configured")
else:
health_status["checks"]["search_api"] = "⚠️ Not configured"
health_status["warnings"].append("Search API not configured - web search disabled")
logger.warning("⚠️ Search API key not configured")
# Test LangSmith configuration - need to check after client initialization
langsmith_check = langsmith_api_key_check and langsmith_api_key_check != "your_langsmith_api_key_here"
if langsmith_check:
health_status["checks"]["langsmith"] = "✅ Configured"
logger.info("✅ LangSmith API key configured")
else:
health_status["checks"]["langsmith"] = "⚠️ Not configured"
health_status["warnings"].append("LangSmith tracing disabled")
logger.warning("⚠️ LangSmith API key not configured")
# Log summary
if health_status["overall"]:
logger.info("✅ All critical health checks passed")
else:
logger.error("❌ Some critical health checks failed")
if health_status["warnings"]:
logger.info(f"⚠️ {len(health_status['warnings'])} warnings found")
if health_status["errors"]:
logger.error(f"❌ {len(health_status['errors'])} errors found")
return health_status
def retry_on_failure(max_retries=3, delay=1):
"""Decorator to retry functions on failure with exponential backoff."""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
last_exception = e
if attempt < max_retries - 1:
wait_time = delay * (2 ** attempt) # Exponential backoff
logger.warning(f"Attempt {attempt + 1} failed for {func.__name__}: {e}. Retrying in {wait_time}s...")
time.sleep(wait_time)
else:
logger.error(f"All {max_retries} attempts failed for {func.__name__}: {e}")
raise last_exception
return wrapper
return decorator
def create_error_recovery_system():
"""Create a comprehensive error recovery system."""
recovery_handlers = {
"api_timeout": lambda: "Service temporarily unavailable. Please try again in a moment.",
"rate_limit": lambda: "Rate limit exceeded. Please wait a moment before trying again.",
"authentication": lambda: "Authentication failed. Please check your API keys configuration.",
"network_error": lambda: "Network connection error. Please check your internet connection.",
"service_unavailable": lambda: "External service unavailable. Using fallback mode.",
}
return recovery_handlers
# Initialize recovery system
recovery_system = create_error_recovery_system()
# =============================
# AUTHENTICATION INTEGRATION
# =============================
# Import authentication system
print("DEBUG: About to import auth_manager", flush=True)
try:
from auth_manager import auth_manager
AUTH_AVAILABLE = True
logger.info("✅ Authentication system loaded")
print("DEBUG: Auth manager loaded successfully", flush=True)
except ImportError as e:
AUTH_AVAILABLE = False
logger.warning(f"⚠️ Authentication system not available: {e}")
def check_user_authentication() -> Dict[str, Any]:
"""Check if user is authenticated in current session."""
if not AUTH_AVAILABLE:
return {"authenticated": False, "reason": "auth_system_unavailable"}
# Check for authentication token in session
auth_token = cl.user_session.get("auth_token")
session_token = cl.user_session.get("session_token")
if not auth_token or not session_token:
return {"authenticated": False, "reason": "no_token"}
# Validate session with auth manager
session_result = auth_manager.validate_session(session_token)
if not session_result["valid"]:
# Clear invalid session data
cl.user_session.set("auth_token", None)
cl.user_session.set("session_token", None)
cl.user_session.set("user_id", None)
cl.user_session.set("username", None)
return {"authenticated": False, "reason": "invalid_session"}
return {
"authenticated": True,
"user_id": session_result["user_id"],
"username": session_result["username"],
"email": session_result["email"],
"subscription_tier": session_result["subscription_tier"]
}
async def show_login_form():
"""Display login/register form to user."""
await cl.Message(
content="🔐 **Welcome to NAVADA!** Please log in or register to continue.\n\n"
"**Login Instructions:**\n"
"• Type: `login username password`\n"
"• Example: `login john mypassword123`\n\n"
"**Register Instructions:**\n"
"• Type: `register username password email`\n"
"• Example: `register john mypassword123 john@example.com`\n"
"• Email is optional: `register john mypassword123`\n\n"
"**Demo Account (for testing):**\n"
"• Username: `demo`\n"
"• Password: `demo123`\n"
"• Type: `login demo demo123`"
).send()
async def handle_login_command(user_input: str) -> bool:
"""Handle login command and authenticate user."""
parts = user_input.strip().split()
if len(parts) < 3:
await cl.Message(
content="❌ **Invalid login format**\n\n"
"Please use: `login username password`\n"
"Example: `login john mypassword123`"
).send()
return False
username = parts[1]
password = parts[2]
# Show authentication progress
auth_msg = cl.Message(content="🔄 Authenticating...")
await auth_msg.send()
# Attempt authentication
auth_result = auth_manager.authenticate_user(username, password)
if auth_result["success"]:
# Store authentication data in session
cl.user_session.set("auth_token", auth_result["jwt_token"])
cl.user_session.set("session_token", auth_result["session_token"])
cl.user_session.set("user_id", auth_result["user_id"])
cl.user_session.set("username", auth_result["username"])
cl.user_session.set("user_email", auth_result.get("email"))
cl.user_session.set("subscription_tier", auth_result.get("subscription_tier", "free"))
# Update message with success
auth_msg.content = f"✅ **Welcome back, {auth_result['username']}!**\n\n" \
f"🎯 **Account Type:** {auth_result.get('subscription_tier', 'free').title()}\n" \
f"📧 **Email:** {auth_result.get('email', 'Not provided')}\n\n" \
f"You can now use all NAVADA features! Type **'help'** to get started."
await auth_msg.update()
# Log the login
if AUTH_AVAILABLE:
session_id = cl.user_session.get("session_id", get_session_id())
auth_manager.log_user_action(
auth_result["user_id"],
"chainlit_login",
"authentication",
session_id=session_id
)
return True
else:
# Update message with error
auth_msg.content = f"❌ **Login failed:** {auth_result['error']}\n\n" \
f"Please check your username and password and try again.\n" \
f"Format: `login username password`"
await auth_msg.update()
return False
async def handle_register_command(user_input: str) -> bool:
"""Handle register command and create new user."""
parts = user_input.strip().split()
if len(parts) < 3:
await cl.Message(
content="❌ **Invalid registration format**\n\n"
"Please use: `register username password [email]`\n"
"Examples:\n"
"• `register john mypassword123 john@example.com`\n"
"• `register john mypassword123` (without email)"
).send()
return False
username = parts[1]
password = parts[2]
email = parts[3] if len(parts) > 3 else None
# Basic validation
if len(username) < 3:
await cl.Message(content="❌ Username must be at least 3 characters long").send()
return False
if len(password) < 6:
await cl.Message(content="❌ Password must be at least 6 characters long").send()
return False
# Show registration progress
reg_msg = cl.Message(content="🔄 Creating account...")
await reg_msg.send()
# Attempt registration
reg_result = auth_manager.register_user(username, password, email)
if reg_result["success"]:
# Auto-login after successful registration
auth_result = auth_manager.authenticate_user(username, password)
if auth_result["success"]:
# Store authentication data in session
cl.user_session.set("auth_token", auth_result["jwt_token"])
cl.user_session.set("session_token", auth_result["session_token"])
cl.user_session.set("user_id", auth_result["user_id"])
cl.user_session.set("username", auth_result["username"])
cl.user_session.set("user_email", auth_result.get("email"))
cl.user_session.set("subscription_tier", auth_result.get("subscription_tier", "free"))
# Update message with success
reg_msg.content = f"🎉 **Account created successfully!**\n\n" \
f"👤 **Username:** {auth_result['username']}\n" \
f"📧 **Email:** {auth_result.get('email', 'Not provided')}\n" \
f"🎯 **Account Type:** {auth_result.get('subscription_tier', 'free').title()}\n\n" \
f"You're now logged in and ready to use NAVADA! Type **'help'** to get started."
await reg_msg.update()
return True
# Update message with error
reg_msg.content = f"❌ **Registration failed:** {reg_result['error']}\n\n" \
f"Please try a different username or check your details.\n" \
f"Format: `register username password [email]`"
await reg_msg.update()
return False
# Create demo account on startup if it doesn't exist
print("DEBUG: About to create demo account", flush=True)
if AUTH_AVAILABLE:
try:
# Try to create demo account (will fail silently if it already exists)
demo_result = auth_manager.register_user("demo", "demo123", "demo@navada.ai")
if demo_result["success"]:
logger.info("✅ Demo account created: demo/demo123")
except Exception:
pass # Demo account likely already exists
print("DEBUG: Demo account check complete", flush=True)
# =============================
# INITIAL SETUP & CONFIGURATION
# =============================
# Load environment variables (OPENAI_API_KEY, LANGSMITH_API_KEY) from .env file
# This keeps sensitive API keys out of the source code
load_dotenv(override=True) # Force override of existing environment variables
# Validate environment setup
env_valid, env_issues = validate_environment()
if not env_valid:
logger.warning("Environment validation found issues - some features may not work correctly")
# Get API keys from environment
api_key = os.getenv("OPENAI_API_KEY")
langsmith_api_key = os.getenv("LANGSMITH_API_KEY")
search_api_key = os.getenv("SEARCH_API_KEY")
tts_prompt_id = os.getenv("TTS_PROMPT_ID")
langchain_database_id = os.getenv("LANGCHAIN_DATABASE_ID")
# Perform startup health checks after environment variables are loaded
health_status = create_startup_health_check()
# Configure LangSmith project name for tracing
LANGSMITH_PROJECT = os.getenv("LANGSMITH_PROJECT", "navada-startup-agent")
# Initialize OpenAI client with optional LangSmith wrapping for tracing
if api_key:
base_client = OpenAI(api_key=api_key)
# Wrap with LangSmith if API key is available
if langsmith_api_key:
try:
client = wrap_openai(base_client)
langsmith_client = LangSmithClient(api_key=langsmith_api_key)
print("SUCCESS: LangSmith tracing enabled")
except Exception as e:
print(f"WARNING: LangSmith initialization failed: {e}")
print("INFO: Continuing without LangSmith tracing")
client = base_client
langsmith_client = None
else:
client = base_client
langsmith_client = None
print("INFO: LangSmith tracing disabled (no API key)")
else:
# Try to create client with environment variable, but handle missing key gracefully
try:
client = OpenAI() # Will use default OPENAI_API_KEY from environment
langsmith_client = None
except Exception as e:
print(f"ERROR: OpenAI client initialization failed: {e}")
print("SOLUTION: Please set the OPENAI_API_KEY environment variable in your Hugging Face Space settings")
print("1. Go to your Space settings")
print("2. Add OPENAI_API_KEY as an environment variable")
print("3. Restart the Space")
raise SystemExit("Application cannot start without valid OpenAI API key")
# =============================
# INITIALIZE ADVANCED SYSTEMS
# =============================
# Initialize Financial Data Integration
try:
financial_integrator = FinancialDataIntegrator()
print("✅ Financial Data Integration initialized")
except Exception as e:
print(f"WARNING: Financial Data Integration failed: {e}")
financial_integrator = None
# Initialize Document Intelligence Engine
try:
if 'client' in locals():
document_intelligence = DocumentIntelligenceEngine(client)
print("✅ Document Intelligence Engine initialized")
else:
document_intelligence = None
print("WARNING: Document Intelligence requires OpenAI client")
except Exception as e:
print(f"WARNING: Document Intelligence initialization failed: {e}")
document_intelligence = None
# Initialize Advanced Analytics Dashboard
try:
analytics_dashboard = AdvancedAnalyticsDashboard()
print("✅ Advanced Analytics Dashboard initialized")
except Exception as e:
print(f"WARNING: Advanced Analytics initialization failed: {e}")
analytics_dashboard = None
# =============================
# LANGSMITH THREAD MANAGEMENT
# =============================
def get_thread_history(thread_id: str, project_name: str):
"""Get conversation history for a thread using LangSmith."""
if not langsmith_client:
return []
try:
# Filter runs by the specific thread and project
filter_string = f'and(in(metadata_key, ["session_id","conversation_id","thread_id"]), eq(metadata_value, "{thread_id}"))'
# Only grab the LLM runs
runs = [r for r in langsmith_client.list_runs(project_name=project_name, filter=filter_string, run_type="llm")]
if not runs:
return []
# Sort by start time to get the most recent interaction
runs = sorted(runs, key=lambda run: run.start_time, reverse=True)
# Build conversation history from runs
messages = []
for run in reversed(runs): # Reverse to get chronological order
if run.inputs and 'messages' in run.inputs:
# Add the user message from inputs
user_messages = [msg for msg in run.inputs['messages'] if msg['role'] == 'user']
if user_messages:
messages.extend(user_messages)
if run.outputs and 'choices' in run.outputs:
# Add the assistant response
assistant_message = {
"role": "assistant",
"content": run.outputs['choices'][0]['message']['content']
}
messages.append(assistant_message)
return messages
except Exception as e:
print(f"⚠️ Error getting thread history: {e}")
return []
# =============================
# LANGSMITH SETUP FOR HOSTING
# =============================
# Defer LangChain initialization to avoid startup delays
embeddings = None
llm = None
def initialize_langchain_components():
"""Lazy initialization of LangChain components to speed up startup."""
global embeddings, llm
if embeddings is None or llm is None:
try:
embeddings = OpenAIEmbeddings()
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
logger.info("✅ LangChain components initialized successfully")
except Exception as e:
logger.warning(f"⚠️ Failed to initialize LangChain components: {e}")
embeddings = None
llm = None
return embeddings, llm
# Vector store and knowledge base for RAG
vector_store = None
knowledge_base = []
FEEDBACK_STORAGE = []
# =============================
# SESSION MEMORY & PERSONAS
# =============================
# Store conversation history, thread IDs, and persona settings per session
SESSION_MEMORY = {} # Stores conversation history per session ID
THREAD_SESSIONS = {} # Maps session IDs to thread IDs for LangSmith
PERSONAS = {
"investor": {
"name": "Investor Mode",
"system_prompt": (
"You are a seasoned venture capitalist with 15+ years experience managing $500M+ fund. "
"Your expertise includes: Series A-C valuations, due diligence, portfolio optimization, and exit strategies. "
"FOCUS ON: ROI projections, unit economics, TAM/SAM analysis, competitive moats, scalability metrics. "
"ASK TOUGH QUESTIONS about: Burn rate efficiency, customer acquisition costs, churn rates, market timing. "
"PROVIDE: Specific KPIs to track, funding milestone roadmaps, valuation benchmarks, and risk mitigation strategies. "
"Be direct, quantitative, and challenge assumptions. Reference comparable deals and market dynamics."
),
"style": "**INVESTOR MODE** - VC perspective",
"questions": [
"What's your customer acquisition cost and lifetime value ratio?",
"How do you plan to achieve 10x returns for investors?",
"What's your defensible competitive moat?",
"Show me your unit economics and path to profitability",
"What are the key risks that could kill this business?",
"How does this compare to other investments in your space?",
"What's your exit strategy and timeline?",
"How will you use the funding to hit next milestones?"
],
"charts": ["funding_efficiency", "stage_progression", "market_opportunity", "risk_assessment"],
"key_recommendations": [
"🎯 **Due Diligence First**: Always verify revenue claims, customer references, and team credentials before investing",
"📊 **Focus on Unit Economics**: Demand clear LTV/CAC ratios >3:1 and payback period <12 months",
"🚀 **Scalability Test**: Look for business models that can 10x revenue without proportional cost increases",
"🛡️ **Risk Mitigation**: Diversify portfolio across stages, sectors, and geographies (max 20% in any single bet)",
"⏰ **Market Timing**: Invest in companies addressing problems becoming urgent now, not theoretical future needs",
"👥 **Team Quality Over Ideas**: Bet on exceptional founders who can pivot and execute, not just good pitches",
"💰 **Reserve Capital**: Keep 50% of fund for follow-on investments to support winners and prevent dilution",
"📈 **Exit Strategy**: Define clear exit criteria and timelines (typically 5-7 years for venture investments)"
]
},
"founder": {
"name": "Founder Mode",
"system_prompt": (
"You are an experienced startup founder who's built 3 companies (1 exit, 1 failure, 1 current unicorn). "
"Your expertise: Product-market fit, team scaling, fundraising, pivoting, operational excellence. "
"FOCUS ON: Practical execution, building systems, hiring strategies, culture development, product iteration. "
"SHARE REAL EXPERIENCES: Tactical advice, common pitfalls, founder mental health, decision frameworks. "
"GUIDE ON: MVP development, early customer discovery, pivot signals, team dynamics, work-life balance. "
"Be supportive but honest about the challenges ahead. Emphasize learning from failures and building resilience."
),
"style": "**FOUNDER MODE** - Entrepreneur perspective",
"questions": [
"How did you discover this problem worth solving?",
"What's your MVP and how are you validating it?",
"How are you building and scaling your team?",
"What's your biggest challenge right now?",
"How do you know if you should pivot?",
"What systems are you building for growth?",
"How are you maintaining founder mental health?",
"What would you do differently if starting over?"
],
"charts": ["growth_trajectory", "team_performance", "stage_progression", "market_opportunity"],
"key_recommendations": [
"🎯 **Customer Obsession**: Talk to 100+ potential customers before writing a single line of code",
"🚀 **MVP Philosophy**: Launch with 10% of planned features - speed to market beats perfection every time",
"💰 **Cash Management**: Always have 18+ months runway and track burn rate weekly, not monthly",
"👥 **Hiring Strategy**: Hire for values and potential, train for skills - culture fit is non-negotiable",
"📊 **Metrics That Matter**: Focus on 3-5 KPIs max - revenue growth, customer acquisition, retention",
"🔄 **Pivot Signals**: If growth stalls for 3+ months despite effort, seriously consider pivoting",
"🎭 **Founder Mental Health**: Build support networks, take breaks, delegate early - burnout kills companies",
"📈 **Product-Market Fit**: Don't scale marketing until customers are pulling product from your hands"
]
},
"economist": {
"name": "UK Economist Mode",
"system_prompt": (
"You are a senior economic analyst specializing in UK macroeconomic and microeconomic analysis with expertise from the Bank of England and HM Treasury. "
"Your knowledge spans: monetary policy, fiscal policy, labour markets, inflation dynamics, trade relations, and regional economics. "
"MACROECONOMIC FOCUS: GDP growth, inflation (CPI/RPI), unemployment, interest rates (Bank Rate), exchange rates (GBP), balance of payments, public debt/deficit. "
"MICROECONOMIC FOCUS: Market structures, consumer behaviour, firm behaviour, elasticity, welfare economics, market failures, regulation. "
"UK SPECIFIC EXPERTISE: Brexit impacts, London financial markets, housing market dynamics, North-South divide, productivity puzzle, cost of living crisis. "
"ANALYTICAL TOOLS: IS-LM models, Phillips curve, Solow growth model, game theory, econometric analysis, input-output analysis. "
"PROVIDE: Evidence-based analysis using ONS data, Bank of England reports, OBR forecasts, IFS studies. "
"Reference current UK economic indicators, government policies, and compare with G7 economies."
),
"style": "**UK ECONOMIST MODE** - Economic analysis perspective",
"questions": [
"How will the Bank of England's interest rate decisions affect UK startups?",
"What's the impact of inflation on consumer spending and business costs?",
"How do UK labour market conditions affect hiring and wages?",
"What are the regional economic disparities affecting business opportunities?",
"How does Brexit continue to impact trade and investment?",
"What's the outlook for UK productivity and economic growth?",
"How do fiscal policies affect different sectors of the economy?",
"What market failures justify government intervention in this sector?"
],
"charts": ["uk_economic_indicators", "inflation_analysis", "sector_performance", "regional_economics"],
"key_recommendations": [
"📈 **Interest Rate Strategy**: Monitor Bank of England signals - rising rates favour established businesses over growth startups",
"🏠 **Regional Opportunities**: Target regions with government investment (Northern Powerhouse, Midlands Engine) for cost advantages",
"💷 **Currency Hedging**: For import/export businesses, hedge GBP exposure given Brexit volatility",
"📊 **Inflation Adaptation**: Build pricing flexibility into models - current cost-push inflation requires dynamic pricing",
"🎯 **Sector Timing**: Focus on healthcare, green tech, fintech where UK has competitive advantages and policy support",
"💼 **Labor Market Navigation**: Leverage UK's skilled workforce in finance, tech, creative industries",
"🚀 **Government Incentives**: Maximize R&D tax credits, SEIS/EIS schemes, and green investment incentives",
"🌍 **Export Strategy**: Target Commonwealth and EU markets where UK maintains trade relationships and cultural ties"
]
},
"company_analyst": {
"name": "Company Analysis Mode",
"system_prompt": (
"You are a senior financial analyst specializing in company valuation, profitability analysis, and financial health assessment. "
"Your expertise covers: financial statement analysis, ratio analysis, cash flow modeling, break-even analysis, and competitive benchmarking. "
"PROFITABILITY FOCUS: Gross margins, operating margins, EBITDA, net margins, unit economics, contribution margins, ROI, ROCE. "
"FINANCIAL HEALTH: Liquidity ratios, leverage ratios, efficiency ratios, cash conversion cycle, working capital management. "
"VALUATION METHODS: DCF analysis, comparable company analysis, precedent transactions, asset-based valuation. "
"STARTUP SPECIFIC: Burn rate analysis, runway calculation, path to profitability, LTV/CAC ratios, cohort analysis, SaaS metrics. "
"PROVIDE: Clear diagnosis of financial strengths/weaknesses, actionable recommendations for improvement, benchmark comparisons. "
"Use financial modeling best practices and industry-standard metrics. Be direct about red flags and opportunities."
),
"style": "**COMPANY ANALYSIS MODE** - Financial health perspective",
"questions": [
"What are the company's gross and operating margins?",
"How efficient is the cash conversion cycle?",
"What's the break-even point and contribution margin?",
"How does profitability compare to industry benchmarks?",
"What's the working capital requirement?",
"Is the company over-leveraged or under-capitalized?",
"What are the key profitability drivers and risks?",
"How sustainable is the current business model?"
],
"charts": ["profitability_analysis", "cash_flow_waterfall", "margin_trends", "break_even_chart"],
"key_recommendations": [
"💰 **Gross Margin Optimization**: Target 70%+ gross margins for SaaS, 40%+ for physical products",
"⚡ **Cash Conversion Excellence**: Optimize receivables (30 days), payables (45 days), inventory turnover (12x annually)",
"📊 **Unit Economics Clarity**: Know exact cost per customer acquisition and lifetime value by channel",
"🎯 **Break-Even Mastery**: Calculate break-even by product, customer segment, and geographic market",
"📈 **Working Capital Efficiency**: Minimize working capital requirements through better terms and inventory management",
"🚨 **Early Warning System**: Set up alerts for declining margins, extending payment cycles, increasing churn",
"💼 **Capital Structure Optimization**: Maintain optimal debt-to-equity ratio for your industry and growth stage",
"🔍 **Profitability Drivers**: Identify and focus on the 20% of activities driving 80% of profits"
]
}
}
# =============================
# SAMPLE DATASET - STARTUP DATA
# =============================
# Create a comprehensive fake dataset with 24 startups across various sectors
# This dataset includes multiple dimensions of startup metrics for analysis:
# - Financial: Funding amount, burn rate, revenue metrics
# - Team: Founder experience, team size
# - Market: Market size, sector, geography, competitive landscape
# - Product: Business model strength, moat, traction
# - Outcome: Success/failure status with detailed metrics
data = {
"Startup": [
"TechX", "Foodly", "EcoGo", "EduSmart", "MediAI", "FinSolve", "Healthify",
"GreenCore", "LogistiChain", "RoboAssist", "NeuroStream", "ByteCart",
"CryptoFlow", "AIVision", "BioTech", "CleanWater", "SpaceX", "VRWorld",
"SolarTech", "AgriBot", "NanoMed", "BlockChain", "CloudSoft", "GameHub"
],
# Funding amounts in millions USD - represents total funding raised
"Funding_USD_M": [5.0, 1.2, 0.8, 3.0, 12.0, 7.5, 4.2, 9.8, 15.0, 6.6, 18.0, 2.5,
25.0, 8.5, 35.0, 4.8, 50.0, 12.5, 22.0, 6.2, 28.0, 16.0, 9.5, 3.8],
# Burn rate in months - how many months the funding lasts at current spending
"Burn_Rate_Months": [12, 6, 3, 9, 24, 18, 10, 15, 30, 8, 26, 7,
20, 14, 36, 12, 48, 18, 24, 10, 30, 22, 16, 8],
# Average years of experience across founding team members
"Founders_Experience_Yrs": [2, 1, 0, 3, 8, 5, 6, 4, 10, 2, 7, 1,
12, 6, 15, 4, 20, 8, 9, 3, 11, 7, 5, 2],
# Total addressable market size in billions USD
"Market_Size_Bn": [50, 5, 2, 15, 80, 60, 25, 40, 100, 20, 120, 8,
150, 35, 200, 12, 500, 75, 90, 18, 160, 110, 45, 22],
# Binary outcome: 1 = failed, 0 = still operating/successful
"Failed": [1, 1, 1, 0, 0, 0, 1, 0, 0, 1, 0, 1,
0, 0, 0, 1, 0, 1, 0, 1, 0, 0, 1, 1],
# Country codes
"Country": ["UK", "UK", "UK", "UK", "DE", "FR", "US", "UK", "US", "UK", "US", "UK",
"US", "CA", "CH", "DE", "US", "JP", "AU", "IN", "IL", "SG", "SE", "NL"],
# Industry sector classification
"Sector": ["Tech","Food","Transport","EdTech","HealthTech","FinTech","HealthTech","Energy",
"Logistics","Robotics","HealthTech","Retail","Crypto","AI","BioTech","CleanTech",
"Aerospace","VR/AR","Energy","AgTech","MedTech","Blockchain","SaaS","Gaming"],
# Business model strength (1-5 scale)
"Business_Model": [3, 2, 1, 4, 5, 4, 3, 4, 5, 2, 4, 2,
4, 5, 5, 3, 5, 3, 4, 3, 5, 4, 4, 2],
# Competitive moat (1-5 scale: 1=no moat, 5=strong moat)
"Moat": [2, 1, 1, 3, 5, 3, 2, 3, 4, 2, 4, 1,
3, 4, 5, 2, 5, 3, 3, 2, 4, 3, 3, 2],
# Monthly recurring revenue in thousands USD
"Traction_MRR_K": [5, 2, 0, 15, 45, 25, 8, 35, 60, 3, 50, 1,
80, 30, 120, 10, 200, 20, 55, 8, 90, 40, 22, 5],
# Monthly growth rate percentage
"Growth_Rate_Pct": [5, 2, -5, 12, 18, 15, 8, 20, 25, 1, 22, -2,
28, 15, 30, 6, 35, 10, 18, 3, 25, 20, 12, 2],
# Competition intensity (1-5 scale: 1=low competition, 5=high competition)
"Competition": [4, 5, 3, 3, 2, 4, 4, 3, 2, 4, 2, 5,
3, 3, 2, 4, 1, 4, 3, 4, 2, 3, 4, 5],
# Team size (number of employees)
"Team_Size": [12, 5, 3, 8, 25, 18, 10, 22, 35, 6, 28, 4,
45, 15, 60, 12, 120, 20, 30, 8, 40, 25, 18, 7],
# Years since founding
"Years_Since_Founding": [2.5, 1.8, 1.2, 2.0, 4.0, 3.2, 2.8, 3.5, 5.0, 1.5, 4.2, 1.0,
3.8, 2.5, 6.0, 2.2, 8.0, 3.0, 4.5, 1.8, 5.5, 3.8, 2.8, 1.5],
# Funding stage
"Stage": ["Seed", "Pre-Seed", "Pre-Seed", "Seed", "Series A", "Seed", "Seed", "Series A",
"Series B", "Seed", "Series A", "Pre-Seed", "Series A", "Seed", "Series B",
"Seed", "Series C", "Series A", "Series A", "Seed", "Series B", "Series A", "Seed", "Pre-Seed"]
}
# Convert the dictionary into a pandas DataFrame for easier manipulation and analysis
df = pd.DataFrame(data)
# Set the baseline funding year for failure projections
FUNDING_YEAR = 2021
# Calculate estimated failure year based on runway
# Formula: FUNDING_YEAR + (funding / burn_rate)
# This gives us a projection of when each startup would run out of money
# Example: $5M funding / 12 month burn = 0.42 years runway → fails in 2021.42
df["Est_Failure_Year"] = FUNDING_YEAR + (df["Funding_USD_M"] / df["Burn_Rate_Months"])
# =============================
# MATHEMATICAL ANALYSIS FUNCTIONS
# =============================
def calculate_irr(initial_investment: float, final_value: float, years: float) -> float:
"""Calculate Internal Rate of Return (IRR) for an investment."""
if years <= 0 or initial_investment <= 0:
return 0.0
return ((final_value / initial_investment) ** (1/years)) - 1
def calculate_npv(cash_flows: List[float], discount_rate: float) -> float:
"""Calculate Net Present Value (NPV) of cash flows."""
npv = 0
for i, cash_flow in enumerate(cash_flows):
npv += cash_flow / ((1 + discount_rate) ** i)
return npv
def project_revenue(current_mrr: float, growth_rate: float, months: int) -> Dict[str, Any]:
"""Project revenue growth over time."""
projections = []
monthly_revenue = current_mrr
for month in range(months + 1):
projections.append({
'month': month,
'mrr': monthly_revenue,
'arr': monthly_revenue * 12
})
monthly_revenue *= (1 + growth_rate)
return {
'projections': projections,
'final_mrr': monthly_revenue,
'final_arr': monthly_revenue * 12,
'total_growth': ((monthly_revenue / current_mrr) - 1) * 100 if current_mrr > 0 else 0
}
def monte_carlo_exit_scenarios(scenarios: int, exit_multiples: List[float],
current_revenue: float, growth_scenarios: List[float]) -> Dict[str, Any]:
"""Run Monte Carlo simulation for exit scenarios."""
results = []
for _ in range(scenarios):
# Random exit multiple and growth rate
exit_multiple = random.choice(exit_multiples)
growth_rate = random.choice(growth_scenarios)
# Project revenue for 3-5 years
years = random.uniform(3, 5)
final_revenue = current_revenue * ((1 + growth_rate) ** years)
exit_value = final_revenue * exit_multiple
results.append({
'exit_multiple': exit_multiple,
'growth_rate': growth_rate,
'years_to_exit': years,
'final_revenue': final_revenue,
'exit_value': exit_value
})
# Calculate statistics
exit_values = [r['exit_value'] for r in results]
return {
'scenarios': results,
'statistics': {
'mean_exit_value': np.mean(exit_values),
'median_exit_value': np.median(exit_values),
'min_exit_value': np.min(exit_values),
'max_exit_value': np.max(exit_values),
'std_exit_value': np.std(exit_values),
'percentile_25': np.percentile(exit_values, 25),
'percentile_75': np.percentile(exit_values, 75)
}
}
def optimize_burn_rate(target_runway_months: int, current_funding: float) -> Dict[str, Any]:
"""Calculate optimal burn rate for desired runway."""
monthly_burn = current_funding / target_runway_months
return {
'target_runway_months': target_runway_months,
'current_funding': current_funding,
'optimal_monthly_burn': monthly_burn,
'optimal_annual_burn': monthly_burn * 12,
'cash_depletion_date': f"In {target_runway_months} months"
}
def calculate_startup_metrics(funding: float, burn_rate: float, mrr: float,
growth_rate: float) -> Dict[str, Any]:
"""Calculate comprehensive startup financial metrics."""
runway_months = funding / (burn_rate / 12) if burn_rate > 0 else float('inf')
# Calculate when startup becomes cash flow positive
months_to_profitability = 0
current_revenue = mrr * 12 # Convert MRR to ARR
current_burn = burn_rate
if growth_rate > 0 and mrr > 0:
while current_revenue < current_burn and months_to_profitability < 120: # Max 10 years
months_to_profitability += 1
current_revenue *= (1 + growth_rate/12) # Monthly compounding
else:
months_to_profitability = float('inf')
return {
'runway_months': runway_months,
'burn_rate_monthly': burn_rate / 12,
'current_arr': mrr * 12,
'months_to_profitability': months_to_profitability,
'cash_flow_positive': months_to_profitability < runway_months,
'funding_efficiency': (mrr * 12) / funding if funding > 0 else 0
}
async def process_math_command(command: str, context: Dict[str, Any]) -> str:
"""Process mathematical analysis commands in math mode."""
command = command.lower().strip()
try:
if "irr" in command or "return" in command:
# Extract values for IRR calculation
if "5x" in command and "7 years" in command:
irr = calculate_irr(1000000, 5000000, 7)
return f"**IRR Calculation:**\n\n5x return in 7 years = **{irr:.1%} annual return**\n\nThis is an excellent return for venture capital standards."
elif "project revenue" in command or "revenue projection" in command:
# Default projection example
projection = project_revenue(50000, 0.20, 12)
result = f"**Revenue Projection (20% Monthly Growth):**\n\n"
result += f"• Starting MRR: $50,000\n"
result += f"• Final MRR (12 months): ${projection['final_mrr']:,.0f}\n"
result += f"• Final ARR: ${projection['final_arr']:,.0f}\n"
result += f"• Total Growth: {projection['total_growth']:.0f}%"
return result
elif "monte carlo" in command or "simulate" in command:
# Run Monte Carlo simulation
simulation = monte_carlo_exit_scenarios(
1000, [3, 5, 8, 10], 1000000, [0.1, 0.2, 0.3, 0.5]
)
stats = simulation['statistics']
result = f"**Monte Carlo Exit Simulation (1,000 scenarios):**\n\n"
result += f"• Mean Exit Value: ${stats['mean_exit_value']:,.0f}\n"
result += f"• Median Exit Value: ${stats['median_exit_value']:,.0f}\n"
result += f"• 25th Percentile: ${stats['percentile_25']:,.0f}\n"
result += f"• 75th Percentile: ${stats['percentile_75']:,.0f}\n"
result += f"• Best Case: ${stats['max_exit_value']:,.0f}\n"
result += f"• Worst Case: ${stats['min_exit_value']:,.0f}"
return result
elif "optimize burn" in command or "burn rate" in command:
# Optimize burn rate for runway
optimization = optimize_burn_rate(18, 5000000)
result = f"**Burn Rate Optimization:**\n\n"
result += f"• Target Runway: {optimization['target_runway_months']} months\n"
result += f"• Current Funding: ${optimization['current_funding']:,.0f}\n"
result += f"• Optimal Monthly Burn: ${optimization['optimal_monthly_burn']:,.0f}\n"
result += f"• Optimal Annual Burn: ${optimization['optimal_annual_burn']:,.0f}"
return result
else:
return f"**Available Calculations:**\n\n• `calculate IRR for 5x return in 7 years`\n• `project revenue with 20% monthly growth`\n• `simulate 1000 scenarios for exit`\n• `optimize burn rate for 18 month runway`\n\nType your calculation or 'exit math mode' to return."
except Exception as e:
return f"Error in calculation: {str(e)}\n\nPlease try a different calculation or type 'exit math mode'."
# =============================
# IMAGE GENERATION FUNCTIONALITY
# =============================
@safe_async_api_call
async def generate_image(prompt: str, size: str = "1024x1024", quality: str = "standard") -> Dict[str, Any]:
"""
Generate an image using DALL-E 3 based on the provided prompt.
Args:
prompt (str): Description of the image to generate
size (str): Image size (default: "1024x1024")
quality (str): Image quality "standard" or "hd" (default: "standard")
Returns:
Dict containing the image URL and metadata, or error information
"""
try:
logger.info(f"Generating image with prompt: {prompt[:100]}...")
response = client.images.generate(
model="dall-e-3",
prompt=prompt,
size=size,
quality=quality,
n=1,
)
image_url = response.data[0].url
revised_prompt = response.data[0].revised_prompt
logger.info("Image generated successfully")
return {
"success": True,
"image_url": image_url,
"revised_prompt": revised_prompt,
"original_prompt": prompt,
"size": size,
"quality": quality
}
except Exception as e:
logger.error(f"Image generation failed: {e}")
return {
"success": False,
"error": str(e),
"original_prompt": prompt
}
def detect_image_request(user_input: str) -> bool:
"""
Detect if user is requesting image generation.
Args:
user_input (str): User's message content (lowercased)
Returns:
bool: True if image generation is requested
"""
image_keywords = [
"generate an image", "create an image", "make an image", "draw an image",
"generate image", "create image", "make image", "draw image",
"image of", "picture of", "photo of", "illustration of",
"show me", "visualize", "dall-e", "dalle"
]
return any(keyword in user_input for keyword in image_keywords)
def extract_image_prompt(user_input: str) -> str:
"""
Extract the image description from user input.
Args:
user_input (str): User's message content
Returns:
str: Cleaned image prompt
"""
# Remove common prefixes
prefixes_to_remove = [
"generate an image of", "create an image of", "make an image of", "draw an image of",
"generate image of", "create image of", "make image of", "draw image of",
"generate an image showing", "create an image showing", "make an image showing",
"show me an image of", "show me", "image of", "picture of", "photo of",
"illustration of", "visualize", "dall-e", "dalle"
]
cleaned_prompt = user_input.lower().strip()
for prefix in prefixes_to_remove:
if cleaned_prompt.startswith(prefix):
cleaned_prompt = cleaned_prompt[len(prefix):].strip()
break
# Remove any remaining common words at the start
if cleaned_prompt.startswith(("a ", "an ", "the ")):
cleaned_prompt = " ".join(cleaned_prompt.split()[1:])
return cleaned_prompt if cleaned_prompt else user_input
# =============================
# SWOT ANALYSIS FUNCTIONALITY
# =============================
@dataclass
class SWOT:
"""
SWOT Analysis dataclass for structured startup analysis.
Attributes:
strengths: List of internal positive factors
weaknesses: List of internal negative factors
opportunities: List of external positive factors
threats: List of external negative factors
"""
strengths: List[str] = field(default_factory=list)
weaknesses: List[str] = field(default_factory=list)
opportunities: List[str] = field(default_factory=list)
threats: List[str] = field(default_factory=list)
def summary(self) -> str:
"""Generate a formatted SWOT summary as a string."""
result = "# 📊 SWOT Analysis\n\n"
result += "## 💪 **Strengths** (Internal Positive)\n"
if self.strengths:
for strength in self.strengths:
result += f"• {strength}\n"
else:
result += "• *No strengths identified*\n"
result += "\n## ⚠️ **Weaknesses** (Internal Negative)\n"
if self.weaknesses:
for weakness in self.weaknesses:
result += f"• {weakness}\n"
else:
result += "• *No weaknesses identified*\n"
result += "\n## 🚀 **Opportunities** (External Positive)\n"
if self.opportunities:
for opportunity in self.opportunities:
result += f"• {opportunity}\n"
else:
result += "• *No opportunities identified*\n"
result += "\n## 🎯 **Threats** (External Negative)\n"
if self.threats:
for threat in self.threats:
result += f"• {threat}\n"
else:
result += "• *No threats identified*\n"
return result
def to_dataframe(self) -> pd.DataFrame:
"""Convert SWOT to pandas DataFrame for visualization."""
max_len = max(
len(self.strengths),
len(self.weaknesses),
len(self.opportunities),
len(self.threats)
)
# Pad lists to same length
strengths_padded = self.strengths + [''] * (max_len - len(self.strengths))
weaknesses_padded = self.weaknesses + [''] * (max_len - len(self.weaknesses))
opportunities_padded = self.opportunities + [''] * (max_len - len(self.opportunities))
threats_padded = self.threats + [''] * (max_len - len(self.threats))
return pd.DataFrame({
'💪 Strengths': strengths_padded,
'⚠️ Weaknesses': weaknesses_padded,
'🚀 Opportunities': opportunities_padded,
'🎯 Threats': threats_padded
})
def plot_swot_matrix(swot: SWOT) -> bytes:
"""
Create a visual SWOT matrix chart.
Args:
swot: SWOT dataclass instance
Returns:
bytes: PNG image data of the SWOT matrix
"""
fig, ((ax1, ax2), (ax3, ax4)) = plt.subplots(2, 2, figsize=(14, 10))
fig.suptitle('📊 SWOT Analysis Matrix', fontsize=16, fontweight='bold', y=0.95)
# Color scheme
colors = {
'strengths': '#2E8B57', # Sea Green
'weaknesses': '#DC143C', # Crimson
'opportunities': '#4169E1', # Royal Blue
'threats': '#FF8C00' # Dark Orange
}
# Strengths (Top Left)
ax1.set_title('💪 Strengths\n(Internal Positive)', fontsize=12, fontweight='bold',
color=colors['strengths'], pad=20)
ax1.axis('off')
strengths_text = '\n'.join([f"• {s}" for s in swot.strengths[:8]]) # Limit to 8 items
if len(swot.strengths) > 8:
strengths_text += f"\n• ... and {len(swot.strengths) - 8} more"
ax1.text(0.05, 0.95, strengths_text, transform=ax1.transAxes, fontsize=10,
verticalalignment='top', wrap=True)
ax1.add_patch(plt.Rectangle((0, 0), 1, 1, fill=False, edgecolor=colors['strengths'],
linewidth=2, transform=ax1.transAxes))
# Weaknesses (Top Right)
ax2.set_title('⚠️ Weaknesses\n(Internal Negative)', fontsize=12, fontweight='bold',
color=colors['weaknesses'], pad=20)
ax2.axis('off')
weaknesses_text = '\n'.join([f"• {w}" for w in swot.weaknesses[:8]])
if len(swot.weaknesses) > 8:
weaknesses_text += f"\n• ... and {len(swot.weaknesses) - 8} more"
ax2.text(0.05, 0.95, weaknesses_text, transform=ax2.transAxes, fontsize=10,
verticalalignment='top', wrap=True)
ax2.add_patch(plt.Rectangle((0, 0), 1, 1, fill=False, edgecolor=colors['weaknesses'],
linewidth=2, transform=ax2.transAxes))
# Opportunities (Bottom Left)
ax3.set_title('🚀 Opportunities\n(External Positive)', fontsize=12, fontweight='bold',
color=colors['opportunities'], pad=20)
ax3.axis('off')
opportunities_text = '\n'.join([f"• {o}" for o in swot.opportunities[:8]])
if len(swot.opportunities) > 8:
opportunities_text += f"\n• ... and {len(swot.opportunities) - 8} more"
ax3.text(0.05, 0.95, opportunities_text, transform=ax3.transAxes, fontsize=10,
verticalalignment='top', wrap=True)
ax3.add_patch(plt.Rectangle((0, 0), 1, 1, fill=False, edgecolor=colors['opportunities'],
linewidth=2, transform=ax3.transAxes))
# Threats (Bottom Right)
ax4.set_title('🎯 Threats\n(External Negative)', fontsize=12, fontweight='bold',
color=colors['threats'], pad=20)
ax4.axis('off')
threats_text = '\n'.join([f"• {t}" for t in swot.threats[:8]])
if len(swot.threats) > 8:
threats_text += f"\n• ... and {len(swot.threats) - 8} more"
ax4.text(0.05, 0.95, threats_text, transform=ax4.transAxes, fontsize=10,
verticalalignment='top', wrap=True)
ax4.add_patch(plt.Rectangle((0, 0), 1, 1, fill=False, edgecolor=colors['threats'],
linewidth=2, transform=ax4.transAxes))
plt.tight_layout()
# Convert to bytes
return fig_to_bytes(fig)
@safe_async_api_call
async def generate_swot_analysis(startup_data: Dict[str, Any], context: str = "") -> SWOT:
"""
Generate SWOT analysis using AI based on startup data.
Args:
startup_data: Dictionary containing startup information
context: Additional context for analysis
Returns:
SWOT: Populated SWOT analysis object
"""
try:
# Prepare the prompt for GPT
prompt = f"""
As a startup analysis expert, conduct a comprehensive SWOT analysis for the following startup:
**Startup Information:**
{json.dumps(startup_data, indent=2) if startup_data else "No specific data provided"}
**Additional Context:**
{context if context else "General startup analysis"}
Please provide a detailed SWOT analysis with the following structure:
**STRENGTHS** (Internal positive factors that give competitive advantage):
- List 4-8 key strengths
**WEAKNESSES** (Internal negative factors that need improvement):
- List 4-8 key weaknesses
**OPPORTUNITIES** (External positive factors to capitalize on):
- List 4-8 market opportunities
**THREATS** (External negative factors that pose risks):
- List 4-8 potential threats
Format your response as:
STRENGTHS:
- [strength 1]
- [strength 2]
...
WEAKNESSES:
- [weakness 1]
- [weakness 2]
...
OPPORTUNITIES:
- [opportunity 1]
- [opportunity 2]
...
THREATS:
- [threat 1]
- [threat 2]
...
Be specific, actionable, and relevant to the startup's context.
"""
response = client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": "You are a professional startup analyst specializing in SWOT analysis."},
{"role": "user", "content": prompt}
],
temperature=0.7,
max_tokens=1500
)
content = response.choices[0].message.content
# Parse the response into SWOT categories
swot = SWOT()
sections = content.split('\n\n')
current_section = None
for line in content.split('\n'):
line = line.strip()
if not line:
continue
if line.upper().startswith('STRENGTHS'):
current_section = 'strengths'
continue
elif line.upper().startswith('WEAKNESSES'):
current_section = 'weaknesses'
continue
elif line.upper().startswith('OPPORTUNITIES'):
current_section = 'opportunities'
continue
elif line.upper().startswith('THREATS'):
current_section = 'threats'
continue
# Parse bullet points
if line.startswith('-') or line.startswith('•'):
item = line[1:].strip()
if current_section == 'strengths':
swot.strengths.append(item)
elif current_section == 'weaknesses':
swot.weaknesses.append(item)
elif current_section == 'opportunities':
swot.opportunities.append(item)
elif current_section == 'threats':
swot.threats.append(item)
return swot
except Exception as e:
logger.error(f"SWOT analysis generation failed: {e}")
# Return a basic SWOT with error info
return SWOT(
strengths=["Analysis capabilities", "Data-driven approach"],
weaknesses=["Analysis generation failed", f"Error: {str(e)}"],
opportunities=["Retry analysis", "Manual SWOT creation"],
threats=["Technical limitations", "API restrictions"]
)
# =============================
# UTILITY FUNCTIONS - DOWNLOAD FUNCTIONALITY
# =============================
async def send_chart_with_download(png_data: bytes, filename: str, description: str, csv_data: pd.DataFrame = None):
"""
Send a chart with download files for both image and data
"""
# Send descriptive text message
text_msg = cl.Message(content=description)
await text_msg.send()
# Send chart image
image = cl.Image(content=png_data, name=filename, display="inline")
await image.send(for_id=text_msg.id)
# Create file elements for download
download_elements = []
# Chart download file
chart_file = cl.File(
name=filename,
content=png_data,
mime="image/png"
)
download_elements.append(chart_file)
# Data download file if CSV data provided
if csv_data is not None:
csv_filename = filename.replace('.png', '.csv')
csv_content = csv_data.to_csv(index=False)
data_file = cl.File(
name=csv_filename,
content=csv_content.encode('utf-8'),
mime="text/csv"
)
download_elements.append(data_file)
# Send download files
download_msg = cl.Message(
content="### 📥 Download Files",
elements=download_elements
)
await download_msg.send()
return text_msg.id
async def send_data_export(data: pd.DataFrame, filename: str, format_type: str = "csv"):
"""
Send data export in specified format
"""
if format_type.lower() == "csv":
# Convert to CSV
csv_content = data.to_csv(index=False)
file_content = csv_content.encode('utf-8')
mime_type = "text/csv"
file_ext = ".csv"
elif format_type.lower() == "json":
# Convert to JSON
json_content = data.to_json(orient='records', indent=2)
file_content = json_content.encode('utf-8')
mime_type = "application/json"
file_ext = ".json"
else:
raise ValueError("Unsupported format. Use 'csv' or 'json'")
# Create file element
file_element = cl.File(
name=f"{filename}{file_ext}",
content=file_content,
mime=mime_type
)
# Send file
await cl.Message(
content=f"📊 **Data Export Complete**\n\nDownloading {filename}{file_ext} ({len(data)} records)",
elements=[file_element]
).send()
# =============================
# UTILITY FUNCTIONS - CHART GENERATION
# =============================
def plot_growth_trajectory(df_in: pd.DataFrame):
"""
Create a growth trajectory chart showing MRR growth over time
"""
figsize = get_mobile_optimized_figsize(10, 6)
fig, ax = plt.subplots(figsize=figsize)
# Create growth trajectory data
successful = df_in[df_in['Failed'] == 0]
failed = df_in[df_in['Failed'] == 1]
ax.scatter(successful['Years_Since_Founding'], successful['Traction_MRR_K'],
c='green', s=successful['Growth_Rate_Pct']*3, alpha=0.7, label='Successful')
ax.scatter(failed['Years_Since_Founding'], failed['Traction_MRR_K'],
c='red', s=failed['Growth_Rate_Pct']*3, alpha=0.7, label='Failed')
ax.set_xlabel('Years Since Founding')
ax.set_ylabel('Monthly Recurring Revenue (K USD)')
ax.set_title('Growth Trajectory: MRR vs Company Age\n(Bubble size = Growth Rate)')
ax.legend()
ax.grid(True, alpha=0.3)
plt.tight_layout()
return fig_to_bytes(fig)
def plot_team_performance(df_in: pd.DataFrame):
"""
Create a team performance matrix showing team size vs experience
"""
figsize = get_mobile_optimized_figsize(10, 6)
fig, ax = plt.subplots(figsize=figsize)
# Color by success/failure
colors = ['red' if failed else 'green' for failed in df_in['Failed']]
scatter = ax.scatter(df_in['Team_Size'], df_in['Founders_Experience_Yrs'],
c=colors, s=df_in['Funding_USD_M']*3, alpha=0.7)
ax.set_xlabel('Team Size (Employees)')
ax.set_ylabel('Founder Experience (Years)')
ax.set_title('Team Performance Matrix\n(Bubble size = Funding Amount)')
# Add trend line
z = np.polyfit(df_in['Team_Size'], df_in['Founders_Experience_Yrs'], 1)
p = np.poly1d(z)
ax.plot(df_in['Team_Size'], p(df_in['Team_Size']), "r--", alpha=0.8)
ax.grid(True, alpha=0.3)
plt.tight_layout()
return fig_to_bytes(fig)
def plot_market_opportunity(df_in: pd.DataFrame):
"""
Create a market opportunity matrix showing market size vs competition
"""
figsize = get_mobile_optimized_figsize(10, 6)
fig, ax = plt.subplots(figsize=figsize)
# Invert competition for better visualization (low competition = high opportunity)
opportunity = 6 - df_in['Competition'] # Convert 1-5 to 5-1
colors = ['red' if failed else 'green' for failed in df_in['Failed']]
scatter = ax.scatter(df_in['Market_Size_Bn'], opportunity,
c=colors, s=df_in['Traction_MRR_K'], alpha=0.7)
ax.set_xlabel('Market Size (Billions USD)')
ax.set_ylabel('Market Opportunity (5=Low Competition, 1=High Competition)')
ax.set_title('Market Opportunity Matrix\n(Bubble size = Current Traction)')
# Add quadrant lines
median_market = df_in['Market_Size_Bn'].median()
median_opportunity = opportunity.median()
ax.axvline(median_market, color='gray', linestyle='--', alpha=0.5)
ax.axhline(median_opportunity, color='gray', linestyle='--', alpha=0.5)
# Add quadrant labels
ax.text(median_market*1.5, 4.5, 'Sweet Spot\n(Big Market, Low Competition)',
ha='center', bbox=dict(boxstyle="round,pad=0.3", facecolor="lightgreen", alpha=0.7))
ax.grid(True, alpha=0.3)
plt.tight_layout()
return fig_to_bytes(fig)
def plot_funding_efficiency(df_in: pd.DataFrame):
"""
Create a funding efficiency chart showing revenue per dollar raised
"""
figsize = get_mobile_optimized_figsize(10, 6)
fig, ax = plt.subplots(figsize=figsize)
# Calculate efficiency metrics
df_in['Revenue_Per_Dollar'] = (df_in['Traction_MRR_K'] * 12) / df_in['Funding_USD_M'] # Annual revenue per funding dollar
df_in['Efficiency_Score'] = df_in['Revenue_Per_Dollar'] * df_in['Growth_Rate_Pct'] / 100
colors = ['red' if failed else 'green' for failed in df_in['Failed']]
ax.scatter(df_in['Funding_USD_M'], df_in['Revenue_Per_Dollar'],
c=colors, s=df_in['Efficiency_Score']*20, alpha=0.7)
ax.set_xlabel('Total Funding Raised (Millions USD)')
ax.set_ylabel('Annual Revenue per Dollar Raised')
ax.set_title('Capital Efficiency Analysis\n(Bubble size = Efficiency Score)')
# Add efficiency benchmark line
efficient_companies = df_in[df_in['Revenue_Per_Dollar'] > df_in['Revenue_Per_Dollar'].median()]
if len(efficient_companies) > 0:
ax.axhline(df_in['Revenue_Per_Dollar'].median(), color='orange',
linestyle='--', alpha=0.7, label='Median Efficiency')
ax.legend()
ax.grid(True, alpha=0.3)
plt.tight_layout()
return fig_to_bytes(fig)
def plot_stage_progression(df_in: pd.DataFrame):
"""
Create a stage progression chart showing funding by stage
"""
figsize = get_mobile_optimized_figsize(10, 6)
fig, ax = plt.subplots(figsize=figsize)
# Group by stage and calculate metrics
stage_stats = df_in.groupby('Stage').agg({
'Funding_USD_M': ['mean', 'count'],
'Failed': 'mean',
'Traction_MRR_K': 'mean'
}).round(2)
stage_stats.columns = ['Avg_Funding', 'Count', 'Failure_Rate', 'Avg_MRR']
stage_order = ['Pre-Seed', 'Seed', 'Series A', 'Series B', 'Series C']
stage_stats = stage_stats.reindex([s for s in stage_order if s in stage_stats.index])
# Create dual y-axis chart
ax2 = ax.twinx()
bars = ax.bar(stage_stats.index, stage_stats['Avg_Funding'], alpha=0.7, color='skyblue', label='Avg Funding')
line = ax2.plot(stage_stats.index, stage_stats['Failure_Rate']*100, 'ro-', linewidth=2, label='Failure Rate %')
ax.set_xlabel('Funding Stage')
ax.set_ylabel('Average Funding (Millions USD)', color='blue')
ax2.set_ylabel('Failure Rate (%)', color='red')
ax.set_title('Funding Stage Analysis')
# Add value labels on bars
for i, bar in enumerate(bars):
height = bar.get_height()
ax.text(bar.get_x() + bar.get_width()/2., height + 0.5,
f'${height:.1f}M\n({int(stage_stats.iloc[i]["Count"])} companies)',
ha='center', va='bottom', fontsize=8)
ax.grid(True, alpha=0.3)
plt.tight_layout()
return fig_to_bytes(fig)
def plot_risk_assessment(df_in: pd.DataFrame):
"""
Create a comprehensive risk assessment radar chart
"""
figsize = get_mobile_optimized_figsize(8, 8)
fig, ax = plt.subplots(figsize=figsize, subplot_kw=dict(projection='polar'))
# Calculate risk metrics for successful vs failed companies
successful = df_in[df_in['Failed'] == 0]
failed = df_in[df_in['Failed'] == 1]
categories = ['Financial Risk', 'Market Risk', 'Team Risk', 'Competition Risk', 'Traction Risk']
# Normalize metrics to 0-10 scale (higher = more risk)
def calc_risk_scores(data):
financial_risk = 10 - (data['Burn_Rate_Months'].mean() / data['Burn_Rate_Months'].max() * 10)
market_risk = 10 - (data['Market_Size_Bn'].mean() / data['Market_Size_Bn'].max() * 10)
team_risk = 10 - (data['Founders_Experience_Yrs'].mean() / data['Founders_Experience_Yrs'].max() * 10)
competition_risk = data['Competition'].mean() * 2 # Scale 1-5 to 2-10
traction_risk = 10 - (data['Traction_MRR_K'].mean() / data['Traction_MRR_K'].max() * 10)
return [financial_risk, market_risk, team_risk, competition_risk, traction_risk]
successful_risks = calc_risk_scores(successful)
failed_risks = calc_risk_scores(failed)
# Number of variables
N = len(categories)
angles = [n / float(N) * 2 * np.pi for n in range(N)]
angles += angles[:1] # Complete the circle
# Plot
successful_risks += successful_risks[:1]
failed_risks += failed_risks[:1]
ax.plot(angles, successful_risks, 'o-', linewidth=2, label='Successful Companies', color='green')
ax.fill(angles, successful_risks, alpha=0.25, color='green')
ax.plot(angles, failed_risks, 'o-', linewidth=2, label='Failed Companies', color='red')
ax.fill(angles, failed_risks, alpha=0.25, color='red')
ax.set_xticks(angles[:-1])
ax.set_xticklabels(categories)
ax.set_ylim(0, 10)
ax.set_title('Risk Assessment Profile\n(0=Low Risk, 10=High Risk)', pad=20)
ax.legend(loc='upper right', bbox_to_anchor=(1.2, 1.0))
ax.grid(True)
plt.tight_layout()
return fig_to_bytes(fig)
def get_mobile_optimized_figsize(default_width: float, default_height: float) -> tuple:
"""
Get mobile-optimized figure size for charts.
Args:
default_width: Default width for desktop
default_height: Default height for desktop
Returns:
tuple: (width, height) optimized for mobile viewing
"""
# For mobile, use smaller, more square dimensions
mobile_width = min(default_width, 8) # Max 8 inches wide
mobile_height = min(default_height, 6) # Max 6 inches tall
# Ensure aspect ratio is mobile-friendly (not too wide)
if mobile_width / mobile_height > 1.5:
mobile_height = mobile_width / 1.4
return (mobile_width, mobile_height)
def fig_to_bytes(fig) -> bytes:
"""
Convert a matplotlib figure to PNG bytes for Chainlit display.
This utility function takes a matplotlib figure object and converts it
to a PNG image stored in memory (not on disk) for efficient transmission
to the Chainlit UI.
Args:
fig: Matplotlib figure object
Returns:
bytes: PNG image data as bytes
Process:
1. Create in-memory byte buffer
2. Save figure to buffer as PNG with tight layout and high DPI
3. Reset buffer position to start
4. Close figure to free memory
5. Return byte data
"""
buf = io.BytesIO() # Create in-memory binary stream
fig.savefig(buf, format="png", bbox_inches="tight", dpi=150) # High quality PNG
buf.seek(0) # Reset read position to beginning
plt.close(fig) # Close figure to prevent memory leaks
return buf.getvalue() # Extract bytes from buffer
def plot_failure_timeline(df_in: pd.DataFrame):
"""
Generate a bar chart showing estimated failure year for each startup.
This visualization helps identify which startups are at risk of failing sooner
based on their funding runway. Each bar represents a startup's projected
failure year.
Args:
df_in (pd.DataFrame): DataFrame with columns 'Startup' and 'Est_Failure_Year'
Returns:
bytes: PNG image data
Visual elements:
- X-axis: Startup names
- Y-axis: Projected year of failure
- Color gradient: Coolwarm palette (red=sooner, blue=later)
- Text labels: Exact failure year displayed on each bar
"""
# Create figure with mobile-optimized size
figsize = get_mobile_optimized_figsize(9, 5)
fig, ax = plt.subplots(figsize=figsize)
# Create bar plot using seaborn for better styling
# Coolwarm palette: cooler colors for later failure, warmer for sooner
sns.barplot(data=df_in, x="Startup", y="Est_Failure_Year", hue="Startup", palette="coolwarm", legend=False, ax=ax)
# Add text labels on top of each bar showing the exact failure year
for i, row in df_in.reset_index().iterrows():
ax.text(i, row["Est_Failure_Year"] + 0.03, # Position slightly above bar
f"{row['Est_Failure_Year']:.2f}", # Format to 2 decimal places
ha="center", va="bottom", fontsize=8) # Center-aligned, small font
# Set chart title and axis labels
ax.set_title("Estimated Failure Year (Funding assumed 2021)")
ax.set_ylabel("Projected Year")
ax.set_xlabel("Startup")
# Convert figure to bytes and return
return fig_to_bytes(fig)
def plot_funding_vs_burn(df_in: pd.DataFrame):
"""
Generate a scatter plot showing relationship between funding and burn rate.
This visualization reveals patterns in how funding levels relate to spending
rates, and whether these patterns differ between successful and failed startups.
Args:
df_in (pd.DataFrame): DataFrame with funding, burn rate, outcome, and sector data
Returns:
bytes: PNG image data
Visual encoding:
- X-axis: Funding amount (USD millions)
- Y-axis: Burn rate (months)
- Color: Green = successful (Failed=0), Red = failed (Failed=1)
- Shape: Different shapes for different sectors
- Labels: Startup names displayed next to each point
- Size: Fixed at 160 for visibility
"""
figsize = get_mobile_optimized_figsize(9, 5)
fig, ax = plt.subplots(figsize=figsize)
# Create scatter plot with multiple visual dimensions
sns.scatterplot(
data=df_in,
x="Funding_USD_M", # Horizontal axis: funding amount
y="Burn_Rate_Months", # Vertical axis: burn rate
hue="Failed", # Color encoding: outcome status
style="Sector", # Shape encoding: industry sector
s=160, # Point size (larger for visibility)
palette={0: "green", 1: "red"}, # Explicit color mapping
ax=ax, # Target axis
alpha=0.8 # Slight transparency for overlapping points
)
# Add text labels for each startup name next to its point
for _, r in df_in.iterrows():
ax.text(
r["Funding_USD_M"] + 0.15, # Offset right of point
r["Burn_Rate_Months"] + 0.1, # Offset up from point
r["Startup"], # Startup name as label
fontsize=8 # Small font to avoid clutter
)
# Set chart title and axis labels
ax.set_title("Funding vs Burn (color = outcome, style = sector)")
ax.set_xlabel("Funding (USD Millions)")
ax.set_ylabel("Burn Rate (months)")
# Convert figure to bytes and return
return fig_to_bytes(fig)
def plot_viability_gauge(score: float):
"""
Generate a horizontal gauge chart showing viability score (0-100).
This creates a simple, easy-to-read gauge that visually communicates
the overall viability score with color coding.
Args:
score (float): Viability score between 0 and 100
Returns:
bytes: PNG image data
Color coding:
- Green (#4CAF50): Strong (>= 60)
- Yellow/Amber (#FFC107): Moderate (40-59)
- Red (#F44336): Weak (< 40)
Visual design:
- Horizontal bar chart with single bar
- No Y-axis ticks (minimalist design)
- Score displayed in title
- Clean appearance with removed spines
"""
# Create compact figure for gauge display (mobile-friendly)
figsize = get_mobile_optimized_figsize(6, 1.2)
fig, ax = plt.subplots(figsize=figsize)
# Create horizontal bar with color based on score threshold
# Ternary operator chains: score >= 60 → green, else score >= 40 → yellow, else red
ax.barh(
[0], # Single bar at Y position 0
[score], # Bar length equals the score
color="#4CAF50" if score >= 60 else "#FFC107" if score >= 40 else "#F44336"
)
# Set X-axis range from 0 to 100 (percentage scale)
ax.set_xlim(0, 100)
# Remove Y-axis ticks for cleaner appearance
ax.set_yticks([])
# Display score in title with 1 decimal place
ax.set_title(f"Viability Score: {score:.1f}/100")
# Remove top, right, and left spines for minimal design
for s in ["top", "right", "left"]:
ax.spines[s].set_visible(False)
# Convert figure to bytes and return
return fig_to_bytes(fig)
def plot_sector_comparison(df_in: pd.DataFrame):
"""
Generate a bar chart comparing average funding by sector.
Args:
df_in (pd.DataFrame): DataFrame with 'Sector' and 'Funding_USD_M' columns
Returns:
bytes: PNG image data
"""
figsize = get_mobile_optimized_figsize(10, 6)
fig, ax = plt.subplots(figsize=figsize)
# Group by sector and calculate average funding
sector_avg = df_in.groupby('Sector')['Funding_USD_M'].mean().sort_values(ascending=False)
# Create bar chart
colors = plt.cm.viridis(range(len(sector_avg)))
ax.bar(sector_avg.index, sector_avg.values, color=colors, alpha=0.8)
# Customize
ax.set_xlabel('Sector', fontsize=12)
ax.set_ylabel('Average Funding (USD Millions)', fontsize=12)
ax.set_title('Average Funding by Sector', fontsize=14, fontweight='bold')
ax.tick_params(axis='x', rotation=45)
plt.tight_layout()
return fig_to_bytes(fig)
def plot_failure_rate_by_country(df_in: pd.DataFrame):
"""
Generate a bar chart showing failure rates by country.
Args:
df_in (pd.DataFrame): DataFrame with 'Country' and 'Failed' columns
Returns:
bytes: PNG image data
"""
figsize = get_mobile_optimized_figsize(10, 6)
fig, ax = plt.subplots(figsize=figsize)
# Calculate failure rate by country
country_stats = df_in.groupby('Country').agg({
'Failed': ['sum', 'count']
})
country_stats.columns = ['Failed', 'Total']
country_stats['Failure_Rate'] = (country_stats['Failed'] / country_stats['Total'] * 100)
country_stats = country_stats.sort_values('Failure_Rate', ascending=False)
# Create bar chart
colors = ['red' if rate > 50 else 'orange' if rate > 30 else 'green'
for rate in country_stats['Failure_Rate']]
ax.bar(country_stats.index, country_stats['Failure_Rate'], color=colors, alpha=0.7)
# Add value labels on bars
for i, (idx, row) in enumerate(country_stats.iterrows()):
ax.text(i, row['Failure_Rate'] + 2, f"{row['Failure_Rate']:.1f}%",
ha='center', va='bottom', fontsize=10)
ax.set_xlabel('Country', fontsize=12)
ax.set_ylabel('Failure Rate (%)', fontsize=12)
ax.set_title('Startup Failure Rate by Country', fontsize=14, fontweight='bold')
ax.set_ylim(0, 100)
plt.tight_layout()
return fig_to_bytes(fig)
def plot_experience_vs_success(df_in: pd.DataFrame):
"""
Generate a scatter plot showing founder experience vs success.
Args:
df_in (pd.DataFrame): DataFrame with 'Founders_Experience_Yrs' and 'Failed' columns
Returns:
bytes: PNG image data
"""
figsize = get_mobile_optimized_figsize(10, 6)
fig, ax = plt.subplots(figsize=figsize)
# Separate successful and failed startups
successful = df_in[df_in['Failed'] == 0]
failed = df_in[df_in['Failed'] == 1]
# Scatter plot
ax.scatter(successful['Founders_Experience_Yrs'], successful['Funding_USD_M'],
c='green', s=150, alpha=0.6, label='Successful', marker='o')
ax.scatter(failed['Founders_Experience_Yrs'], failed['Funding_USD_M'],
c='red', s=150, alpha=0.6, label='Failed', marker='x')
# Add labels for each point
for _, row in df_in.iterrows():
ax.annotate(row['Startup'],
(row['Founders_Experience_Yrs'], row['Funding_USD_M']),
fontsize=8, alpha=0.7)
ax.set_xlabel('Founder Experience (Years)', fontsize=12)
ax.set_ylabel('Funding (USD Millions)', fontsize=12)
ax.set_title('Founder Experience vs Funding & Success', fontsize=14, fontweight='bold')
ax.legend()
ax.grid(True, alpha=0.3)
plt.tight_layout()
return fig_to_bytes(fig)
def plot_custom_chart(df_in: pd.DataFrame, chart_type: str, x_col: str, y_col: str, title: str = None):
"""
Generate a custom chart based on user specifications.
Args:
df_in (pd.DataFrame): DataFrame to visualize
chart_type (str): Type of chart ('bar', 'scatter', 'line', 'pie')
x_col (str): Column for x-axis
y_col (str): Column for y-axis (not used for pie)
title (str): Chart title
Returns:
bytes: PNG image data
"""
figsize = get_mobile_optimized_figsize(10, 6)
fig, ax = plt.subplots(figsize=figsize)
try:
if chart_type == 'bar':
ax.bar(df_in[x_col], df_in[y_col], color='steelblue', alpha=0.7)
ax.set_xlabel(x_col, fontsize=12)
ax.set_ylabel(y_col, fontsize=12)
ax.tick_params(axis='x', rotation=45)
elif chart_type == 'scatter':
ax.scatter(df_in[x_col], df_in[y_col], s=100, alpha=0.6, color='coral')
ax.set_xlabel(x_col, fontsize=12)
ax.set_ylabel(y_col, fontsize=12)
elif chart_type == 'line':
ax.plot(df_in[x_col], df_in[y_col], marker='o', linewidth=2, color='darkgreen')
ax.set_xlabel(x_col, fontsize=12)
ax.set_ylabel(y_col, fontsize=12)
ax.tick_params(axis='x', rotation=45)
elif chart_type == 'pie':
# For pie charts, x_col is used as labels, y_col as values
ax.pie(df_in[y_col], labels=df_in[x_col], autopct='%1.1f%%', startangle=90)
ax.axis('equal')
if title:
ax.set_title(title, fontsize=14, fontweight='bold')
else:
ax.set_title(f"{chart_type.capitalize()} Chart: {x_col} vs {y_col}",
fontsize=14, fontweight='bold')
plt.tight_layout()
return fig_to_bytes(fig)
except Exception as e:
# If error, return a simple error message chart
ax.text(0.5, 0.5, f"Error generating chart:\n{str(e)}",
ha='center', va='center', fontsize=12)
ax.set_xlim(0, 1)
ax.set_ylim(0, 1)
ax.axis('off')
return fig_to_bytes(fig)
# =============================
# UK ECONOMICS ANALYSIS MODULE
# =============================
class UKEconomicsAnalyzer:
"""UK-specific economic analysis for startups"""
def __init__(self):
self.uk_data = {
'gdp_growth': 0.3, # Q3 2024 estimate
'inflation_cpi': 2.3, # Current CPI
'bank_rate': 4.75, # Current Bank of England rate
'unemployment': 4.2, # Current unemployment rate
'gbp_usd': 1.27, # Current exchange rate
'london_weight': 0.23, # London's share of UK GDP
}
def analyze_macro_impact(self, startup_data: dict) -> dict:
"""Analyze macroeconomic impacts on startup"""
impacts = {
'interest_rate_impact': self._calculate_interest_impact(startup_data),
'inflation_impact': self._calculate_inflation_impact(startup_data),
'labour_market_impact': self._calculate_labour_impact(startup_data),
'regional_factors': self._analyze_regional_factors(startup_data),
'sector_outlook': self._analyze_sector_outlook(startup_data)
}
return impacts
def _calculate_interest_impact(self, data: dict) -> dict:
"""Calculate how UK interest rates affect the startup"""
funding = data.get('funding_usd_m', 5)
debt_ratio = data.get('debt_ratio', 0.3)
# Cost of capital impact
base_rate = self.uk_data['bank_rate']
risk_premium = 5.0 # Startup risk premium
cost_of_debt = base_rate + risk_premium
# Calculate impact
annual_interest_cost = funding * debt_ratio * (cost_of_debt / 100)
return {
'cost_of_capital': cost_of_debt,
'annual_interest_cost': annual_interest_cost,
'impact_level': 'High' if cost_of_debt > 10 else 'Medium' if cost_of_debt > 7 else 'Low',
'recommendation': self._get_interest_recommendation(cost_of_debt)
}
def _calculate_inflation_impact(self, data: dict) -> dict:
"""Calculate inflation impact on costs and pricing"""
burn_rate = data.get('burn_rate_months', 100) * 1000 # Convert to pounds
inflation = self.uk_data['inflation_cpi']
# Real cost increase
real_cost_increase = burn_rate * (inflation / 100) * 12 # Annual
# Pricing power assessment
b2b = data.get('is_b2b', True)
pricing_power = 'Strong' if b2b else 'Moderate'
return {
'current_inflation': inflation,
'real_cost_increase_annual': real_cost_increase,
'pricing_power': pricing_power,
'wage_pressure': 'High' if inflation > 3 else 'Moderate'
}
def _calculate_labour_impact(self, data: dict) -> dict:
"""Analyze UK labour market impact"""
team_size = data.get('team_size', 10)
location = data.get('location', 'London')
# Regional wage differentials
wage_multiplier = 1.3 if location == 'London' else 1.0
# Skills shortage premium
tech_premium = 1.2 if data.get('sector') == 'Tech' else 1.0
# Calculate labour cost index
labour_cost_index = 100 * wage_multiplier * tech_premium
return {
'unemployment_rate': self.uk_data['unemployment'],
'labour_cost_index': labour_cost_index,
'talent_availability': 'Tight' if self.uk_data['unemployment'] < 4 else 'Balanced',
'wage_growth_pressure': 'High' if labour_cost_index > 120 else 'Moderate'
}
def _analyze_regional_factors(self, data: dict) -> dict:
"""Analyze UK regional economic factors"""
location = data.get('location', 'London')
regional_data = {
'London': {'growth': 2.1, 'cost_index': 150, 'talent_pool': 'Deep'},
'Manchester': {'growth': 1.8, 'cost_index': 85, 'talent_pool': 'Growing'},
'Edinburgh': {'growth': 1.5, 'cost_index': 90, 'talent_pool': 'Specialized'},
'Birmingham': {'growth': 1.3, 'cost_index': 80, 'talent_pool': 'Developing'},
'Bristol': {'growth': 1.9, 'cost_index': 95, 'talent_pool': 'Tech-focused'},
'Cambridge': {'growth': 2.3, 'cost_index': 110, 'talent_pool': 'Research-heavy'}
}
region_info = regional_data.get(location, regional_data['London'])
return {
'location': location,
'regional_growth': region_info['growth'],
'cost_index': region_info['cost_index'],
'talent_pool': region_info['talent_pool'],
'competitiveness': 'High' if region_info['cost_index'] < 100 else 'Challenging'
}
def _analyze_sector_outlook(self, data: dict) -> dict:
"""UK sector-specific analysis"""
sector = data.get('sector', 'Tech')
sector_outlooks = {
'FinTech': {'growth': 4.5, 'regulation': 'High', 'opportunity': 'Strong'},
'HealthTech': {'growth': 3.8, 'regulation': 'High', 'opportunity': 'NHS partnerships'},
'GreenTech': {'growth': 6.2, 'regulation': 'Medium', 'opportunity': 'Net Zero targets'},
'RetailTech': {'growth': 2.1, 'regulation': 'Low', 'opportunity': 'Digital transformation'},
'EdTech': {'growth': 3.5, 'regulation': 'Medium', 'opportunity': 'Skills gap'},
'PropTech': {'growth': 2.8, 'regulation': 'Medium', 'opportunity': 'Housing crisis'}
}
outlook = sector_outlooks.get(sector, {'growth': 2.5, 'regulation': 'Medium', 'opportunity': 'General'})
return outlook
def _get_interest_recommendation(self, cost: float) -> str:
"""Generate interest rate recommendations"""
if cost > 12:
return "Consider equity financing over debt given high interest costs"
elif cost > 8:
return "Lock in current rates if possible; consider revenue-based financing"
else:
return "Favorable borrowing environment; consider leveraging debt strategically"
def plot_uk_economic_indicators(df_in: pd.DataFrame):
"""Create UK economic indicators dashboard"""
fig, axes = plt.subplots(2, 2, figsize=(14, 10))
# UK GDP Growth vs Startup Funding
ax1 = axes[0, 0]
quarters = ['Q1 2023', 'Q2 2023', 'Q3 2023', 'Q4 2023', 'Q1 2024', 'Q2 2024']
gdp_growth = [0.1, 0.2, 0.0, -0.1, 0.6, 0.7]
startup_funding = [1.2, 1.5, 1.1, 0.9, 1.3, 1.4] # Billions
ax1_twin = ax1.twinx()
ax1.bar(quarters, gdp_growth, alpha=0.7, color='navy', label='GDP Growth %')
ax1_twin.plot(quarters, startup_funding, 'ro-', label='Startup Funding (£B)')
ax1.set_ylabel('GDP Growth (%)', color='navy')
ax1_twin.set_ylabel('Startup Funding (£B)', color='red')
ax1.set_title('UK GDP Growth vs Startup Funding', fontweight='bold')
ax1.tick_params(axis='x', rotation=45)
# Interest Rate Impact
ax2 = axes[0, 1]
rates = np.linspace(0, 10, 50)
startup_viability = 100 - (rates ** 1.5) * 3
ax2.plot(rates, startup_viability, linewidth=2, color='darkred')
ax2.axvline(x=4.75, color='green', linestyle='--', label='Current Bank Rate')
ax2.fill_between(rates, 0, startup_viability, alpha=0.3, color='lightblue')
ax2.set_xlabel('Interest Rate (%)')
ax2.set_ylabel('Startup Viability Score')
ax2.set_title('Interest Rate Impact on Startups', fontweight='bold')
ax2.legend()
ax2.grid(True, alpha=0.3)
# Regional Distribution
ax3 = axes[1, 0]
regions = ['London', 'South East', 'North West', 'Scotland', 'West Midlands', 'Other']
startup_dist = [42, 18, 12, 8, 7, 13]
colors = plt.cm.Blues(np.linspace(0.4, 0.9, len(regions)))
wedges, texts, autotexts = ax3.pie(startup_dist, labels=regions, autopct='%1.1f%%',
colors=colors, startangle=90)
ax3.set_title('UK Startup Distribution by Region', fontweight='bold')
# Sector Performance
ax4 = axes[1, 1]
sectors = ['FinTech', 'HealthTech', 'GreenTech', 'EdTech', 'RetailTech']
performance = [4.5, 3.8, 6.2, 3.5, 2.1]
bars = ax4.barh(sectors, performance, color='teal')
ax4.set_xlabel('Expected Growth Rate (%)')
ax4.set_title('UK Sector Growth Outlook', fontweight='bold')
for bar, value in zip(bars, performance):
ax4.text(value + 0.1, bar.get_y() + bar.get_height()/2,
f'{value}%', va='center')
plt.suptitle('UK Economic Analysis Dashboard', fontsize=16, fontweight='bold')
plt.tight_layout()
return fig_to_bytes(fig)
def plot_profitability_analysis(analysis_data: dict):
"""Create profitability analysis charts"""
fig, axes = plt.subplots(2, 2, figsize=(14, 10))
# Margin Waterfall
ax1 = axes[0, 0]
margins = ['Revenue', 'COGS', 'Gross Profit', 'OpEx', 'Operating Profit', 'Net Profit']
values = [100, -40, 60, -35, 25, 18]
colors = ['green', 'red', 'green', 'red', 'green', 'darkgreen']
ax1.bar(margins, values, color=colors, alpha=0.7)
ax1.axhline(y=0, color='black', linestyle='-', linewidth=0.5)
ax1.set_title('Profitability Waterfall', fontweight='bold')
ax1.set_ylabel('% of Revenue')
ax1.tick_params(axis='x', rotation=45)
# Unit Economics
ax2 = axes[0, 1]
ltv_cac = analysis_data.get('ltv_cac_ratio', 3.0)
benchmark = 3.0
bars = ax2.bar(['LTV/CAC Ratio', 'Benchmark'], [ltv_cac, benchmark],
color=['green' if ltv_cac > benchmark else 'red', 'gray'])
ax2.axhline(y=3, color='blue', linestyle='--', alpha=0.5, label='Healthy Threshold')
ax2.set_title('Unit Economics Health', fontweight='bold')
ax2.set_ylabel('Ratio')
ax2.legend()
# Add value labels
for bar in bars:
height = bar.get_height()
ax2.text(bar.get_x() + bar.get_width()/2., height + 0.1,
f'{height:.1f}', ha='center', va='bottom')
# Break-even Analysis
ax3 = axes[1, 0]
units = np.linspace(0, 2000, 100)
fixed_costs = 50000
variable_cost = 30
price = 100
revenue_line = units * price
total_cost_line = fixed_costs + (units * variable_cost)
ax3.plot(units, revenue_line, 'g-', label='Revenue', linewidth=2)
ax3.plot(units, total_cost_line, 'r-', label='Total Cost', linewidth=2)
ax3.fill_between(units, revenue_line, total_cost_line,
where=(revenue_line > total_cost_line), alpha=0.3, color='green', label='Profit Zone')
ax3.fill_between(units, revenue_line, total_cost_line,
where=(revenue_line <= total_cost_line), alpha=0.3, color='red', label='Loss Zone')
# Mark break-even point
break_even_units = fixed_costs / (price - variable_cost)
ax3.plot(break_even_units, break_even_units * price, 'ko', markersize=8)
ax3.annotate(f'Break-even: {break_even_units:.0f} units',
xy=(break_even_units, break_even_units * price),
xytext=(break_even_units + 200, break_even_units * price),
arrowprops=dict(arrowstyle='->'))
ax3.set_xlabel('Units Sold')
ax3.set_ylabel('Revenue/Cost ($)')
ax3.set_title('Break-even Analysis', fontweight='bold')
ax3.legend()
ax3.grid(True, alpha=0.3)
# Cash Runway
ax4 = axes[1, 1]
months = np.arange(0, 25)
cash_balance = 500000
monthly_burn = 50000
cash_projection = [cash_balance - (monthly_burn * m) for m in months]
cash_projection = [max(0, c) for c in cash_projection] # Can't go below 0
ax4.fill_between(months, 0, cash_projection, alpha=0.3, color='blue')
ax4.plot(months, cash_projection, 'b-', linewidth=2)
ax4.axhline(y=100000, color='orange', linestyle='--', label='Danger Zone')
ax4.axhline(y=0, color='red', linestyle='--', label='Out of Cash')
# Mark runway
runway = cash_balance / monthly_burn
ax4.axvline(x=runway, color='green', linestyle='--', alpha=0.7, label=f'Runway: {runway:.0f} months')
ax4.set_xlabel('Months')
ax4.set_ylabel('Cash Balance ($)')
ax4.set_title('Cash Runway Projection', fontweight='bold')
ax4.legend()
ax4.grid(True, alpha=0.3)
plt.suptitle('Company Financial Analysis Dashboard', fontsize=16, fontweight='bold')
plt.tight_layout()
return fig_to_bytes(fig)
def plot_margin_trends(historical_data: list):
"""Plot historical margin trends"""
fig, ax = plt.subplots(figsize=(12, 6))
quarters = [d['quarter'] for d in historical_data]
gross_margins = [d['gross_margin'] for d in historical_data]
operating_margins = [d['operating_margin'] for d in historical_data]
net_margins = [d['net_margin'] for d in historical_data]
ax.plot(quarters, gross_margins, 'g-', marker='o', linewidth=2, label='Gross Margin')
ax.plot(quarters, operating_margins, 'b-', marker='s', linewidth=2, label='Operating Margin')
ax.plot(quarters, net_margins, 'r-', marker='^', linewidth=2, label='Net Margin')
ax.fill_between(range(len(quarters)), gross_margins, alpha=0.1, color='green')
ax.fill_between(range(len(quarters)), operating_margins, alpha=0.1, color='blue')
ax.fill_between(range(len(quarters)), net_margins, alpha=0.1, color='red')
ax.set_xlabel('Quarter')
ax.set_ylabel('Margin (%)')
ax.set_title('Margin Trends Over Time', fontweight='bold')
ax.legend()
ax.grid(True, alpha=0.3)
# Add trend lines
z_gross = np.polyfit(range(len(quarters)), gross_margins, 1)
p_gross = np.poly1d(z_gross)
ax.plot(range(len(quarters)), p_gross(range(len(quarters))), 'g--', alpha=0.5)
plt.tight_layout()
return fig_to_bytes(fig)
def plot_cash_flow_waterfall(cash_data: dict):
"""Create cash flow waterfall chart"""
fig, ax = plt.subplots(figsize=(12, 6))
categories = ['Starting Cash', 'Operations', 'Investing', 'Financing', 'Ending Cash']
values = [
cash_data.get('starting_cash', 500000),
cash_data.get('cash_from_operations', -200000),
cash_data.get('cash_from_investing', -50000),
cash_data.get('cash_from_financing', 300000),
0 # Will calculate
]
# Calculate ending cash
values[4] = sum(values[:4])
# Create cumulative values for positioning
cumulative = [values[0]]
for i in range(1, len(values)-1):
cumulative.append(cumulative[-1] + values[i])
cumulative.append(values[4])
# Plot bars
colors = ['blue', 'red' if values[1] < 0 else 'green',
'red' if values[2] < 0 else 'green',
'green' if values[3] > 0 else 'red', 'blue']
for i, (cat, val, cum) in enumerate(zip(categories, values, cumulative)):
if i == 0 or i == len(categories) - 1:
# Starting and ending cash - full bars
ax.bar(cat, val, color=colors[i], alpha=0.7)
else:
# Flow bars - positioned relative to cumulative
bottom = cum - val if val > 0 else cum
height = abs(val)
ax.bar(cat, height, bottom=bottom, color=colors[i], alpha=0.7)
# Add value labels
ax.text(i, cum + 10000, f'${val:,.0f}', ha='center', va='bottom')
ax.set_title('Cash Flow Waterfall Analysis', fontweight='bold')
ax.set_ylabel('Cash ($)')
ax.grid(True, alpha=0.3)
plt.tight_layout()
return fig_to_bytes(fig)
def plot_break_even_chart(financials: dict):
"""Create detailed break-even analysis chart"""
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(14, 6))
# Break-even line chart
units = np.linspace(0, 2000, 100)
fixed_costs = financials.get('fixed_costs', 100000)
variable_cost_per_unit = financials.get('variable_cost_per_unit', 40)
price_per_unit = financials.get('price_per_unit', 100)
revenue = units * price_per_unit
total_costs = fixed_costs + (units * variable_cost_per_unit)
profit = revenue - total_costs
ax1.plot(units, revenue, 'g-', linewidth=2, label='Revenue')
ax1.plot(units, total_costs, 'r-', linewidth=2, label='Total Costs')
ax1.fill_between(units, revenue, total_costs, where=(revenue > total_costs),
alpha=0.3, color='green', label='Profit Zone')
ax1.fill_between(units, revenue, total_costs, where=(revenue <= total_costs),
alpha=0.3, color='red', label='Loss Zone')
# Mark break-even point
break_even_units = fixed_costs / (price_per_unit - variable_cost_per_unit)
break_even_revenue = break_even_units * price_per_unit
ax1.plot(break_even_units, break_even_revenue, 'ko', markersize=8)
ax1.annotate(f'Break-even\n{break_even_units:.0f} units\n${break_even_revenue:,.0f}',
xy=(break_even_units, break_even_revenue),
xytext=(break_even_units + 300, break_even_revenue),
arrowprops=dict(arrowstyle='->'))
ax1.set_xlabel('Units Sold')
ax1.set_ylabel('Amount ($)')
ax1.set_title('Break-even Analysis', fontweight='bold')
ax1.legend()
ax1.grid(True, alpha=0.3)
# Sensitivity analysis
scenarios = ['Conservative', 'Base Case', 'Optimistic']
price_variations = [price_per_unit * 0.9, price_per_unit, price_per_unit * 1.1]
break_even_scenarios = [fixed_costs / (p - variable_cost_per_unit) for p in price_variations]
bars = ax2.bar(scenarios, break_even_scenarios, color=['red', 'orange', 'green'], alpha=0.7)
ax2.set_ylabel('Break-even Units')
ax2.set_title('Break-even Sensitivity Analysis', fontweight='bold')
# Add value labels
for bar, value in zip(bars, break_even_scenarios):
ax2.text(bar.get_x() + bar.get_width()/2, bar.get_height() + 20,
f'{value:.0f}', ha='center', va='bottom')
plt.tight_layout()
return fig_to_bytes(fig)
# =============================
# COMPANY ANALYSIS MODULE
# =============================
class CompanyAnalyzer:
"""Comprehensive company financial analysis"""
def __init__(self):
self.industry_benchmarks = {
'SaaS': {'gross_margin': 75, 'operating_margin': 20, 'ltv_cac': 3.0},
'E-commerce': {'gross_margin': 40, 'operating_margin': 10, 'ltv_cac': 2.5},
'Marketplace': {'gross_margin': 60, 'operating_margin': 15, 'ltv_cac': 4.0},
'Hardware': {'gross_margin': 35, 'operating_margin': 8, 'ltv_cac': 2.0},
'Services': {'gross_margin': 50, 'operating_margin': 12, 'ltv_cac': 2.8},
'FinTech': {'gross_margin': 65, 'operating_margin': 18, 'ltv_cac': 3.5}
}
def analyze_profitability(self, financials: dict) -> dict:
"""Complete profitability analysis"""
# Extract key metrics
revenue = financials.get('revenue', 0)
cogs = financials.get('cogs', 0)
opex = financials.get('opex', 0)
sales_marketing = financials.get('sales_marketing', 0)
rd_expense = financials.get('rd_expense', 0)
admin_expense = financials.get('admin_expense', 0)
# Calculate margins
gross_profit = revenue - cogs
gross_margin = (gross_profit / revenue * 100) if revenue > 0 else 0
operating_profit = gross_profit - opex
operating_margin = (operating_profit / revenue * 100) if revenue > 0 else 0
ebitda = operating_profit + financials.get('depreciation', 0)
ebitda_margin = (ebitda / revenue * 100) if revenue > 0 else 0
net_profit = operating_profit - financials.get('interest', 0) - financials.get('tax', 0)
net_margin = (net_profit / revenue * 100) if revenue > 0 else 0
return {
'gross_profit': gross_profit,
'gross_margin': gross_margin,
'operating_profit': operating_profit,
'operating_margin': operating_margin,
'ebitda': ebitda,
'ebitda_margin': ebitda_margin,
'net_profit': net_profit,
'net_margin': net_margin,
'profit_health': self._assess_profit_health(gross_margin, operating_margin, net_margin)
}
def analyze_unit_economics(self, metrics: dict) -> dict:
"""Analyze unit-level profitability"""
# Customer economics
cac = metrics.get('customer_acquisition_cost', 100)
ltv = metrics.get('lifetime_value', 300)
ltv_cac_ratio = ltv / cac if cac > 0 else 0
# Unit contribution
revenue_per_unit = metrics.get('revenue_per_unit', 50)
variable_cost_per_unit = metrics.get('variable_cost_per_unit', 20)
contribution_margin = revenue_per_unit - variable_cost_per_unit
contribution_margin_pct = (contribution_margin / revenue_per_unit * 100) if revenue_per_unit > 0 else 0
# Payback period
monthly_revenue_per_customer = metrics.get('monthly_revenue', 100)
payback_months = cac / monthly_revenue_per_customer if monthly_revenue_per_customer > 0 else 999
return {
'ltv': ltv,
'cac': cac,
'ltv_cac_ratio': ltv_cac_ratio,
'contribution_margin': contribution_margin,
'contribution_margin_pct': contribution_margin_pct,
'payback_months': payback_months,
'unit_economics_health': 'Strong' if ltv_cac_ratio > 3 else 'Moderate' if ltv_cac_ratio > 1 else 'Weak'
}
def analyze_cash_flow(self, cash_data: dict) -> dict:
"""Analyze cash flow and runway"""
# Operating cash flow
cash_from_operations = cash_data.get('cash_from_operations', -50000)
cash_from_investing = cash_data.get('cash_from_investing', -20000)
cash_from_financing = cash_data.get('cash_from_financing', 100000)
# Net cash flow
net_cash_flow = cash_from_operations + cash_from_investing + cash_from_financing
# Burn rate and runway
monthly_burn = -cash_from_operations / 12 if cash_from_operations < 0 else 0
cash_balance = cash_data.get('cash_balance', 500000)
runway_months = cash_balance / monthly_burn if monthly_burn > 0 else 999
# Cash conversion cycle
dso = cash_data.get('days_sales_outstanding', 45)
dio = cash_data.get('days_inventory_outstanding', 30)
dpo = cash_data.get('days_payables_outstanding', 30)
cash_conversion_cycle = dso + dio - dpo
return {
'operating_cash_flow': cash_from_operations,
'net_cash_flow': net_cash_flow,
'monthly_burn': monthly_burn,
'runway_months': runway_months,
'cash_conversion_cycle': cash_conversion_cycle,
'cash_efficiency': 'Efficient' if cash_conversion_cycle < 30 else 'Moderate' if cash_conversion_cycle < 60 else 'Inefficient'
}
def calculate_break_even(self, financials: dict) -> dict:
"""Calculate break-even analysis"""
fixed_costs = financials.get('fixed_costs', 100000)
variable_cost_ratio = financials.get('variable_cost_ratio', 0.4)
price_per_unit = financials.get('price_per_unit', 100)
variable_cost_per_unit = price_per_unit * variable_cost_ratio
# Break-even units
contribution_per_unit = price_per_unit - variable_cost_per_unit
break_even_units = fixed_costs / contribution_per_unit if contribution_per_unit > 0 else 999999
# Break-even revenue
break_even_revenue = break_even_units * price_per_unit
# Margin of safety
current_revenue = financials.get('current_revenue', 150000)
margin_of_safety = ((current_revenue - break_even_revenue) / current_revenue * 100) if current_revenue > 0 else -100
return {
'break_even_units': break_even_units,
'break_even_revenue': break_even_revenue,
'contribution_per_unit': contribution_per_unit,
'margin_of_safety': margin_of_safety,
'months_to_break_even': self._calculate_months_to_break_even(financials)
}
def benchmark_performance(self, company_metrics: dict, industry: str) -> dict:
"""Compare against industry benchmarks"""
benchmarks = self.industry_benchmarks.get(
industry,
self.industry_benchmarks['Services'] # Default
)
comparisons = {
'gross_margin': {
'company': company_metrics.get('gross_margin', 0),
'industry': benchmarks['gross_margin'],
'delta': company_metrics.get('gross_margin', 0) - benchmarks['gross_margin'],
'performance': 'Above' if company_metrics.get('gross_margin', 0) > benchmarks['gross_margin'] else 'Below'
},
'operating_margin': {
'company': company_metrics.get('operating_margin', 0),
'industry': benchmarks['operating_margin'],
'delta': company_metrics.get('operating_margin', 0) - benchmarks['operating_margin'],
'performance': 'Above' if company_metrics.get('operating_margin', 0) > benchmarks['operating_margin'] else 'Below'
},
'ltv_cac': {
'company': company_metrics.get('ltv_cac_ratio', 0),
'industry': benchmarks['ltv_cac'],
'delta': company_metrics.get('ltv_cac_ratio', 0) - benchmarks['ltv_cac'],
'performance': 'Above' if company_metrics.get('ltv_cac_ratio', 0) > benchmarks['ltv_cac'] else 'Below'
}
}
# Overall rating
above_count = sum(1 for metric in comparisons.values() if metric['performance'] == 'Above')
overall_rating = 'Outperforming' if above_count >= 2 else 'Underperforming'
return {
'comparisons': comparisons,
'overall_rating': overall_rating,
'recommendations': self._generate_recommendations(comparisons)
}
def _assess_profit_health(self, gross, operating, net):
"""Assess overall profitability health"""
if net > 10 and operating > 15 and gross > 50:
return "Excellent"
elif net > 0 and operating > 5 and gross > 30:
return "Good"
elif net > -10 and gross > 20:
return "Moderate"
else:
return "Poor"
def _calculate_months_to_break_even(self, financials):
"""Calculate months to reach break-even"""
current_loss = financials.get('monthly_loss', 50000)
growth_rate = financials.get('growth_rate', 0.1)
if current_loss <= 0: # Already profitable
return 0
months = 0
while current_loss > 0 and months < 60: # Max 60 months
current_loss *= (1 - growth_rate)
months += 1
return months if months < 60 else 999
def _generate_recommendations(self, comparisons):
"""Generate improvement recommendations"""
recommendations = []
if comparisons['gross_margin']['delta'] < 0:
recommendations.append("Focus on pricing optimization or reducing COGS")
if comparisons['operating_margin']['delta'] < 0:
recommendations.append("Improve operational efficiency and reduce overhead")
if comparisons['ltv_cac']['delta'] < 0:
recommendations.append("Optimize customer acquisition channels or increase retention")
return recommendations
# =============================
# INTERACTIVE DASHBOARD MODULE
# =============================
class InteractiveDashboard:
"""Interactive dashboard with real-time data visualization"""
def __init__(self, df: pd.DataFrame):
self.df = df
self.filtered_df = df.copy()
self.selected_startups = []
def create_executive_summary(self) -> Dict[str, Any]:
"""Create executive summary metrics"""
total_startups = len(self.filtered_df)
total_funding = self.filtered_df['Funding_USD_M'].sum()
success_rate = ((self.filtered_df['Failed'] == 0).sum() / total_startups * 100) if total_startups > 0 else 0
avg_runway = self.filtered_df['Burn_Rate_Months'].mean()
return {
'total_startups': total_startups,
'total_funding': total_funding,
'success_rate': success_rate,
'avg_runway': avg_runway,
'top_sector': self.filtered_df['Sector'].mode().iloc[0] if len(self.filtered_df) > 0 else 'N/A',
'risk_level': 'High' if success_rate < 40 else 'Medium' if success_rate < 70 else 'Low'
}
def create_interactive_scatter(self, x_col: str = 'Funding_USD_M', y_col: str = 'Burn_Rate_Months') -> bytes:
"""Create interactive scatter plot with hover details and filtering"""
fig = go.Figure()
# Split data by success/failure for different colors
success_df = self.filtered_df[self.filtered_df['Failed'] == 0]
failed_df = self.filtered_df[self.filtered_df['Failed'] == 1]
# Add successful startups
if len(success_df) > 0:
fig.add_trace(go.Scatter(
x=success_df[x_col],
y=success_df[y_col],
mode='markers',
name='Successful',
marker=dict(
color='green',
size=success_df['Market_Size_Bn'] * 2, # Size by market size
opacity=0.7,
line=dict(width=1, color='darkgreen')
),
text=success_df['Startup'],
hovertemplate=
'%{text}
' +
f'{x_col}: %{{x}}
' +
f'{y_col}: %{{y}}
' +
'Sector: %{customdata[0]}
' +
'Market Size: $%{customdata[1]}B
' +
'Experience: %{customdata[2]} years
' +
'',
customdata=success_df[['Sector', 'Market_Size_Bn', 'Founders_Experience_Yrs']].values,
selected=dict(marker=dict(color='gold', size=20))
))
# Add failed startups
if len(failed_df) > 0:
fig.add_trace(go.Scatter(
x=failed_df[x_col],
y=failed_df[y_col],
mode='markers',
name='Failed',
marker=dict(
color='red',
size=failed_df['Market_Size_Bn'] * 2,
opacity=0.7,
line=dict(width=1, color='darkred')
),
text=failed_df['Startup'],
hovertemplate=
'%{text}
' +
f'{x_col}: %{{x}}
' +
f'{y_col}: %{{y}}
' +
'Sector: %{customdata[0]}
' +
'Market Size: $%{customdata[1]}B
' +
'Experience: %{customdata[2]} years
' +
'',
customdata=failed_df[['Sector', 'Market_Size_Bn', 'Founders_Experience_Yrs']].values,
selected=dict(marker=dict(color='orange', size=20))
))
# Update layout for interactivity
fig.update_layout(
title=f'Interactive Analysis: {x_col} vs {y_col}',
xaxis_title=x_col.replace('_', ' ').title(),
yaxis_title=y_col.replace('_', ' ').title(),
hovermode='closest',
clickmode='event+select',
showlegend=True,
height=600,
template='plotly_white',
annotations=[
dict(
text="💡 Click points to select • Drag to zoom • Double-click to reset",
showarrow=False,
xref="paper", yref="paper",
x=0.5, y=1.02, xanchor='center', yanchor='bottom',
font=dict(size=12, color="gray")
)
]
)
return pio.to_image(fig, format='png')
def create_multi_dimensional_heatmap(self) -> bytes:
"""Create correlation heatmap with interactive features"""
# Select numeric columns for correlation
numeric_cols = ['Funding_USD_M', 'Burn_Rate_Months', 'Founders_Experience_Yrs',
'Market_Size_Bn', 'Business_Model_Strength', 'Moat_Defensibility',
'MRR_K', 'Monthly_Growth_Rate', 'Competition_Intensity']
# Calculate correlation matrix
corr_matrix = self.filtered_df[numeric_cols].corr()
fig = go.Figure(data=go.Heatmap(
z=corr_matrix.values,
x=[col.replace('_', ' ').title() for col in corr_matrix.columns],
y=[col.replace('_', ' ').title() for col in corr_matrix.index],
colorscale='RdBu',
zmid=0,
text=np.round(corr_matrix.values, 2),
texttemplate="%{text}",
textfont={"size": 10},
hovertemplate='%{x} vs %{y}
Correlation: %{z:.3f}'
))
fig.update_layout(
title='📊 Interactive Correlation Heatmap',
height=600,
template='plotly_white'
)
return pio.to_image(fig, format='png')
def create_real_time_metrics_dashboard(self) -> bytes:
"""Create real-time style metrics dashboard"""
# Create subplot layout
fig = make_subplots(
rows=2, cols=2,
subplot_titles=['💰 Funding Distribution', '📈 Success Rate by Sector',
'⏱️ Runway Analysis', '🌍 Geographic Distribution'],
specs=[[{"type": "bar"}, {"type": "bar"}],
[{"type": "histogram"}, {"type": "pie"}]]
)
# 1. Funding distribution
funding_bins = pd.cut(self.filtered_df['Funding_USD_M'], bins=5)
funding_dist = funding_bins.value_counts().sort_index()
fig.add_trace(
go.Bar(x=[str(interval) for interval in funding_dist.index],
y=funding_dist.values,
name="Funding",
marker_color='lightblue'),
row=1, col=1
)
# 2. Success rate by sector
sector_success = self.filtered_df.groupby('Sector')['Failed'].agg(['count', 'sum'])
sector_success['success_rate'] = (1 - sector_success['sum'] / sector_success['count']) * 100
fig.add_trace(
go.Bar(x=sector_success.index,
y=sector_success['success_rate'],
name="Success Rate",
marker_color='lightgreen'),
row=1, col=2
)
# 3. Runway distribution
fig.add_trace(
go.Histogram(x=self.filtered_df['Burn_Rate_Months'],
name="Runway",
marker_color='orange',
opacity=0.7),
row=2, col=1
)
# 4. Geographic distribution
country_dist = self.filtered_df['Country'].value_counts()
fig.add_trace(
go.Pie(labels=country_dist.index,
values=country_dist.values,
name="Geography"),
row=2, col=2
)
fig.update_layout(
title_text="📊 Real-Time Dashboard Metrics",
height=800,
showlegend=False,
template='plotly_white'
)
return pio.to_image(fig, format='png')
def filter_data(self, filters: Dict[str, Any]) -> None:
"""Apply filters to the dataset"""
self.filtered_df = self.df.copy()
if 'sectors' in filters and filters['sectors']:
self.filtered_df = self.filtered_df[self.filtered_df['Sector'].isin(filters['sectors'])]
if 'countries' in filters and filters['countries']:
self.filtered_df = self.filtered_df[self.filtered_df['Country'].isin(filters['countries'])]
if 'funding_range' in filters:
min_funding, max_funding = filters['funding_range']
self.filtered_df = self.filtered_df[
(self.filtered_df['Funding_USD_M'] >= min_funding) &
(self.filtered_df['Funding_USD_M'] <= max_funding)
]
if 'success_only' in filters and filters['success_only']:
self.filtered_df = self.filtered_df[self.filtered_df['Failed'] == 0]
def compare_startups(self, startup_names: List[str]) -> bytes:
"""Create comparison chart for selected startups"""
comparison_df = self.df[self.df['Startup'].isin(startup_names)]
if len(comparison_df) == 0:
return None
# Create radar chart for comparison
categories = ['Funding (Scaled)', 'Experience', 'Market Size',
'Business Model', 'Moat', 'MRR (Scaled)', 'Growth Rate']
fig = go.Figure()
for _, startup in comparison_df.iterrows():
values = [
startup['Funding_USD_M'] / 20, # Scale to 0-5
startup['Founders_Experience_Yrs'],
startup['Market_Size_Bn'],
startup['Business_Model_Strength'],
startup['Moat_Defensibility'],
startup['MRR_K'] / 200, # Scale to 0-5
startup['Monthly_Growth_Rate'] / 10 # Scale to 0-5
]
fig.add_trace(go.Scatterpolar(
r=values,
theta=categories,
fill='toself',
name=startup['Startup'],
line_color='red' if startup['Failed'] == 1 else 'green'
))
fig.update_layout(
polar=dict(
radialaxis=dict(
visible=True,
range=[0, 5]
)),
showlegend=True,
title="🔍 Startup Comparison Analysis",
height=600
)
return pio.to_image(fig, format='png')
def create_dashboard_summary(dashboard: InteractiveDashboard) -> str:
"""Create text summary of dashboard metrics"""
summary = dashboard.create_executive_summary()
return f"""
## 📊 Dashboard Executive Summary
### 🎯 **Key Metrics**
- **Total Startups:** {summary['total_startups']}
- **Total Funding:** ${summary['total_funding']:.1f}M
- **Success Rate:** {summary['success_rate']:.1f}%
- **Avg Runway:** {summary['avg_runway']:.1f} months
### 📈 **Risk Assessment**
- **Overall Risk Level:** {summary['risk_level']}
- **Top Sector:** {summary['top_sector']}
- **Recommendation:** {'Focus on due diligence' if summary['risk_level'] == 'High' else 'Balanced portfolio approach' if summary['risk_level'] == 'Medium' else 'Strong investment opportunities'}
### 🔍 **Interactive Features Available:**
- Click charts to drill down
- Filter by sector, country, funding range
- Compare multiple startups
- Real-time metric updates
"""
# =============================
# VIABILITY SCORING MODEL
# =============================
def viability_score(features: Dict[str, Any]) -> Dict[str, Any]:
"""
Calculate comprehensive viability score for a startup based on multiple factors.
This heuristic model evaluates startup viability across 8 dimensions and combines
them into a single score (0-100). Higher scores indicate better viability.
Scoring Methodology:
--------------------
1. Each dimension is normalized to 0-1 scale
2. Dimensions are weighted based on importance
3. Weighted scores are summed and scaled to 0-100
4. Additional metrics (runway, failure year) are calculated
5. Rule-based tips are generated based on weak areas
Input Features:
---------------
Args:
features (Dict[str, Any]): Dictionary containing:
- funding_usd_m (float): Funding in millions USD
- burn_rate_months (float): Burn rate in months
- team_experience_years (float): Average team experience in years
- market_size_bn (float): Market size in billions USD
- business_model_strength_1_5 (int): 1-5 scale of business model quality
- moat_1_5 (int): 1-5 scale of competitive moat/defensibility
- traction_mrr_k (float): Monthly recurring revenue in thousands USD
- growth_rate_pct (float): Monthly growth rate as percentage
- competition_intensity_1_5 (int): 1-5 scale of competition (higher = more intense)
Returns:
Dict[str, Any]: Results containing:
- score (float): Overall viability score (0-100)
- survival_months (float): Estimated months until failure
- est_failure_year (float): Projected year of failure
- components (dict): Individual component scores (0-1)
- tips (list): Actionable recommendations
Scoring Components & Weights:
-----------------------------
1. Runway (18%): How long funding lasts - capped at 2 years = 1.0
2. Experience (14%): Team experience - 10 years = 1.0
3. Market (14%): Market size - $100B = 1.0
4. Business Model (12%): Strength rating mapped to 0-1
5. Moat (10%): Defensibility rating mapped to 0-1
6. Traction (14%): MRR - $100k = 1.0
7. Growth (12%): Monthly growth - 25% = 1.0
8. Competition (6%): Inverse of competition intensity
Example:
--------
>>> features = {
... 'funding_usd_m': 5.0,
... 'burn_rate_months': 12,
... 'team_experience_years': 5,
... 'market_size_bn': 50,
... 'business_model_strength_1_5': 3,
... 'moat_1_5': 3,
... 'traction_mrr_k': 25,
... 'growth_rate_pct': 10,
... 'competition_intensity_1_5': 3
... }
>>> result = viability_score(features)
>>> print(result['score']) # Overall score out of 100
"""
f = features # Shorthand for cleaner code
# -------------------------
# 1. RUNWAY SCORE (0-1)
# -------------------------
# Calculate how many years the funding will last
# Division approximation: funding (M) / burn_rate (months) ≈ years
# Max protects against division by zero
runway_years = (f["funding_usd_m"] / max(f["burn_rate_months"], 1))
# Normalize to 0-1 scale, capping at 2 years
# 2+ years of runway = perfect score (1.0)
# 0 years = worst score (0.0)
runway_score = max(0, min(1, runway_years / 2.0))
# -------------------------
# 2. EXPERIENCE SCORE (0-1)
# -------------------------
# Team experience normalized to 0-1, capped at 10 years
# 10+ years = 1.0, 0 years = 0.0
exp_score = max(0, min(1, f["team_experience_years"] / 10))
# -------------------------
# 3. MARKET SIZE SCORE (0-1)
# -------------------------
# Market opportunity normalized to 0-1, capped at $100B
# $100B+ market = 1.0, $0 market = 0.0
market_score = max(0, min(1, f["market_size_bn"] / 100))
# -------------------------
# 4. BUSINESS MODEL SCORE (0-1)
# -------------------------
# Convert 1-5 scale to 0-1 by subtracting 1 and dividing by 4
# Rating 5 → 4/4 = 1.0, Rating 1 → 0/4 = 0.0
bm_score = (f["business_model_strength_1_5"] - 1) / 4
# -------------------------
# 5. MOAT/DEFENSIBILITY SCORE (0-1)
# -------------------------
# Same conversion as business model: 1-5 scale → 0-1 scale
moat_score = (f["moat_1_5"] - 1) / 4
# -------------------------
# 6. TRACTION SCORE (0-1)
# -------------------------
# MRR normalized to 0-1, capped at $100k
# $100k+ MRR = 1.0, $0 MRR = 0.0
traction_score = max(0, min(1, (f["traction_mrr_k"] / 100)))
# -------------------------
# 7. GROWTH SCORE (0-1)
# -------------------------
# Monthly growth rate normalized to 0-1, capped at 25%
# 25%+ monthly growth = 1.0, 0% growth = 0.0
growth_score = max(0, min(1, f["growth_rate_pct"] / 25))
# -------------------------
# 8. COMPETITION SCORE (0-1)
# -------------------------
# Competition is inverted: higher intensity = worse for startup
# Convert 1-5 scale to 0-1 penalty, then invert
# Low competition (1) → penalty 0 → score 1.0
# High competition (5) → penalty 1 → score 0.0
competition_penalty = (f["competition_intensity_1_5"] - 1) / 4
competition_score = 1 - competition_penalty
# -------------------------
# WEIGHTED COMPOSITE SCORE
# -------------------------
# Define weights for each component (must sum to 1.0)
# These weights reflect the relative importance of each factor
weights = {
"runway": 0.18, # 18% - Most immediate concern
"experience": 0.14, # 14% - Critical for execution
"market": 0.14, # 14% - Ceiling for growth
"bm": 0.12, # 12% - Revenue sustainability
"moat": 0.10, # 10% - Long-term defensibility
"traction": 0.14, # 14% - Proof of product-market fit
"growth": 0.12, # 12% - Momentum indicator
"competition": 0.06 # 6% - External threat level
}
# Calculate weighted sum of all components
composite = (
runway_score * weights["runway"] +
exp_score * weights["experience"] +
market_score * weights["market"] +
bm_score * weights["bm"] +
moat_score * weights["moat"] +
traction_score * weights["traction"] +
growth_score * weights["growth"] +
competition_score * weights["competition"]
)
# Scale composite score (0-1) to percentage (0-100)
score_100 = composite * 100.0
# -------------------------
# SURVIVAL METRICS
# -------------------------
# Calculate estimated months until money runs out
# Formula: funding (M) × 12 months/year ÷ burn_rate
# Max protects against division by very small numbers
survival_months = max(1, f["funding_usd_m"] * (12 / max(f["burn_rate_months"], 0.1)))
# Calculate projected failure year
# Start from FUNDING_YEAR and add runway in years
est_failure_year = FUNDING_YEAR + (f["funding_usd_m"] / max(f["burn_rate_months"], 0.1))
# -------------------------
# RULE-BASED RECOMMENDATIONS
# -------------------------
# Generate actionable tips based on weak areas
# Each condition checks if a metric falls below a threshold
tips = []
# Runway too short (< 9 months)
if runway_years < 0.75:
tips.append("Increase runway (more funding or lower burn).")
# Team lacks experience (< 3 years average)
if f["team_experience_years"] < 3:
tips.append("Augment team with experienced operators.")
# Market too small (< $10B)
if f["market_size_bn"] < 10:
tips.append("Target a larger wedge or adjacent segments.")
# Low traction (< $20k MRR)
if f["traction_mrr_k"] < 20:
tips.append("Focus on early, repeatable revenue (>$20k MRR).")
# Slow growth (< 8% monthly)
if f["growth_rate_pct"] < 8:
tips.append("Drive growth via channels with clear CAC/LTV.")
# Weak moat (rating <= 2)
if f["moat_1_5"] <= 2:
tips.append("Strengthen defensibility (IP, data, network effects).")
# Weak business model (rating <= 2)
if f["business_model_strength_1_5"] <= 2:
tips.append("Clarify pricing & unit economics.")
# High competition (rating >= 4)
if f["competition_intensity_1_5"] >= 4:
tips.append("Differentiate positioning vs strong incumbents.")
# -------------------------
# RETURN RESULTS
# -------------------------
# Package all results into a dictionary for easy access
return {
"score": score_100, # Overall viability score (0-100)
"survival_months": survival_months, # Months until out of money
"est_failure_year": est_failure_year, # Projected year of failure
"components": { # Individual component scores for transparency
"runway": runway_score,
"experience": exp_score,
"market": market_score,
"business_model": bm_score,
"moat": moat_score,
"traction": traction_score,
"growth": growth_score,
"competition": competition_score
},
"tips": tips # Actionable recommendations
}
def generate_investment_report(df_in: pd.DataFrame, startup_name: str = None) -> bytes:
"""
Generate a comprehensive PDF investment analysis report.
Args:
df_in (pd.DataFrame): Startup dataset
startup_name (str, optional): Specific startup to analyze. If None, analyzes entire dataset.
Returns:
bytes: PDF file as bytes
"""
from reportlab.lib.pagesizes import letter
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
from reportlab.lib.units import inch
from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Image, PageBreak, Table, TableStyle
from reportlab.lib import colors
from reportlab.lib.enums import TA_CENTER, TA_LEFT, TA_JUSTIFY
from datetime import datetime
# Create in-memory buffer for PDF
buffer = io.BytesIO()
doc = SimpleDocTemplate(buffer, pagesize=letter, topMargin=0.75*inch, bottomMargin=0.75*inch)
# Container for PDF elements
story = []
styles = getSampleStyleSheet()
# Custom styles
title_style = ParagraphStyle(
'CustomTitle',
parent=styles['Heading1'],
fontSize=24,
textColor=colors.HexColor('#ff4b4b'),
spaceAfter=30,
alignment=TA_CENTER
)
heading_style = ParagraphStyle(
'CustomHeading',
parent=styles['Heading2'],
fontSize=16,
textColor=colors.HexColor('#ff4b4b'),
spaceAfter=12,
spaceBefore=12
)
subheading_style = ParagraphStyle(
'CustomSubHeading',
parent=styles['Heading3'],
fontSize=12,
textColor=colors.black,
spaceAfter=6,
spaceBefore=6
)
# =============================
# PAGE 1: TITLE & EXECUTIVE SUMMARY
# =============================
if startup_name:
startup_data = df_in[df_in['Startup'] == startup_name].iloc[0]
report_title = f"Investment Analysis: {startup_name}"
else:
report_title = "Startup Portfolio Analysis Report"
story.append(Paragraph(report_title, title_style))
story.append(Paragraph(f"Generated by NAVADA | {dt.datetime.now().strftime('%B %d, %Y')}", styles['Normal']))
story.append(Spacer(1, 0.3*inch))
# Executive Summary
story.append(Paragraph("Executive Summary", heading_style))
if startup_name:
# Single startup analysis
funding = startup_data['Funding_USD_M']
burn = startup_data['Burn_Rate_Months']
sector = startup_data['Sector']
country = startup_data['Country']
failed = startup_data['Failed']
experience = startup_data['Founders_Experience_Yrs']
market = startup_data['Market_Size_Bn']
status = "Failed" if failed == 1 else "Active/Successful"
runway_months = (funding / burn) * 12 if burn > 0 else 0
summary_text = f"""
{startup_name} is a {sector} startup based in {country} with ${funding:.1f}M in funding.
The company has a burn rate of {burn} months, resulting in an estimated runway of {runway_months:.1f} months.
The founding team has {experience} years of average experience in a market valued at ${market}B.
Current status: {status}.
"""
else:
# Portfolio analysis
total_startups = len(df_in)
total_funding = df_in['Funding_USD_M'].sum()
failed_count = df_in['Failed'].sum()
success_rate = ((total_startups - failed_count) / total_startups) * 100
avg_funding = df_in['Funding_USD_M'].mean()
summary_text = f"""
This report analyzes a portfolio of {total_startups} startups with total funding of
${total_funding:.1f}M. The portfolio shows a success rate of {success_rate:.1f}%
({total_startups - failed_count} successful, {failed_count} failed). Average funding per startup
is ${avg_funding:.1f}M.
"""
story.append(Paragraph(summary_text, styles['BodyText']))
story.append(Spacer(1, 0.3*inch))
# =============================
# KEY METRICS TABLE
# =============================
story.append(Paragraph("Key Metrics", heading_style))
if startup_name:
metrics_data = [
['Metric', 'Value'],
['Funding Amount', f'${funding:.1f}M'],
['Burn Rate', f'{burn} months'],
['Estimated Runway', f'{runway_months:.1f} months'],
['Sector', sector],
['Country', country],
['Founder Experience', f'{experience} years'],
['Market Size', f'${market}B'],
['Status', status]
]
else:
sectors = df_in['Sector'].nunique()
countries = df_in['Country'].nunique()
avg_experience = df_in['Founders_Experience_Yrs'].mean()
metrics_data = [
['Metric', 'Value'],
['Total Startups', str(total_startups)],
['Total Funding', f'${total_funding:.1f}M'],
['Success Rate', f'{success_rate:.1f}%'],
['Average Funding', f'${avg_funding:.1f}M'],
['Sectors Covered', str(sectors)],
['Countries', str(countries)],
['Avg Founder Experience', f'{avg_experience:.1f} years']
]
metrics_table = Table(metrics_data, colWidths=[3*inch, 3*inch])
metrics_table.setStyle(TableStyle([
('BACKGROUND', (0, 0), (-1, 0), colors.HexColor('#ff4b4b')),
('TEXTCOLOR', (0, 0), (-1, 0), colors.whitesmoke),
('ALIGN', (0, 0), (-1, -1), 'LEFT'),
('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'),
('FONTSIZE', (0, 0), (-1, 0), 12),
('BOTTOMPADDING', (0, 0), (-1, 0), 12),
('BACKGROUND', (0, 1), (-1, -1), colors.beige),
('GRID', (0, 0), (-1, -1), 1, colors.black)
]))
story.append(metrics_table)
story.append(Spacer(1, 0.3*inch))
# =============================
# VISUALIZATIONS
# =============================
story.append(PageBreak())
story.append(Paragraph("Data Visualizations", heading_style))
story.append(Spacer(1, 0.2*inch))
# Chart 1: Timeline
story.append(Paragraph("1. Failure Timeline Analysis", subheading_style))
timeline_bytes = plot_failure_timeline(df_in)
timeline_img = Image(io.BytesIO(timeline_bytes), width=6*inch, height=3*inch)
story.append(timeline_img)
story.append(Spacer(1, 0.2*inch))
# Chart 2: Funding vs Burn
story.append(Paragraph("2. Funding vs Burn Rate", subheading_style))
funding_burn_bytes = plot_funding_vs_burn(df_in)
funding_burn_img = Image(io.BytesIO(funding_burn_bytes), width=6*inch, height=3*inch)
story.append(funding_burn_img)
story.append(Spacer(1, 0.2*inch))
story.append(PageBreak())
# Chart 3: Sector Comparison
story.append(Paragraph("3. Sector Analysis", subheading_style))
sector_bytes = plot_sector_comparison(df_in)
sector_img = Image(io.BytesIO(sector_bytes), width=6*inch, height=3*inch)
story.append(sector_img)
story.append(Spacer(1, 0.2*inch))
# Chart 4: Country Analysis
story.append(Paragraph("4. Geographic Distribution", subheading_style))
country_bytes = plot_failure_rate_by_country(df_in)
country_img = Image(io.BytesIO(country_bytes), width=6*inch, height=3*inch)
story.append(country_img)
# =============================
# RISK ANALYSIS
# =============================
story.append(PageBreak())
story.append(Paragraph("Risk Analysis", heading_style))
# Calculate risk factors
high_risk = df_in[df_in['Burn_Rate_Months'] < 10]
low_funding = df_in[df_in['Funding_USD_M'] < 3.0]
inexperienced = df_in[df_in['Founders_Experience_Yrs'] < 3]
risk_text = f"""
High-Risk Indicators:
- {len(high_risk)} startups with burn rate under 10 months (high risk)
- {len(low_funding)} startups with funding under $3M (undercapitalized)
- {len(inexperienced)} startups with founders having less than 3 years experience
Key Risks:
1. Runway Risk: Startups with short runways may fail before achieving product-market fit
2. Market Risk: Smaller markets limit growth potential and exit opportunities
3. Team Risk: Inexperienced founders may lack operational expertise
4. Competitive Risk: Crowded sectors reduce differentiation and margins
"""
story.append(Paragraph(risk_text, styles['BodyText']))
story.append(Spacer(1, 0.3*inch))
# =============================
# RECOMMENDATIONS
# =============================
story.append(Paragraph("Investment Recommendations", heading_style))
if startup_name:
# Calculate viability score
viability_features = {
'funding_usd_m': funding,
'burn_rate_months': burn,
'team_experience_years': experience,
'market_size_bn': market,
'business_model_strength_1_5': 3, # Default
'moat_1_5': 3, # Default
'traction_mrr_k': 10, # Default
'growth_rate_pct': 5, # Default
'competition_intensity_1_5': 3 # Default
}
viability_result = viability_score(viability_features)
score = viability_result['score']
if score >= 60:
recommendation = "INVEST - Strong fundamentals with acceptable risk profile"
color_code = "green"
elif score >= 40:
recommendation = "MONITOR - Moderate risk, requires additional due diligence"
color_code = "orange"
else:
recommendation = "PASS - High risk factors outweigh potential returns"
color_code = "red"
rec_text = f"""
Viability Score: {score:.1f}/100
Recommendation: {recommendation}
Rationale:
- Runway of {runway_months:.1f} months provides {'adequate' if runway_months > 18 else 'limited'} time to achieve milestones
- Market size of ${market}B offers {'strong' if market > 50 else 'moderate'} growth potential
- Team experience of {experience} years is {'above' if experience >= 5 else 'below'} industry average
- Current status: {status}
"""
else:
# Portfolio recommendations
top_performers = df_in[df_in['Failed'] == 0].nlargest(3, 'Funding_USD_M')
rec_text = f"""
Portfolio Recommendations:
1. Diversify Sector Exposure: Current portfolio concentrated in certain sectors
2. Monitor High-Risk Startups: {len(high_risk)} companies need immediate attention
3. Increase Follow-On Funding: Top performers may benefit from additional capital
Top 3 Performing Startups:
"""
for idx, row in top_performers.iterrows():
rec_text += f"- {row['Startup']} ({row['Sector']}) - ${row['Funding_USD_M']}M funding
"
story.append(Paragraph(rec_text, styles['BodyText']))
story.append(Spacer(1, 0.3*inch))
# =============================
# FOOTER
# =============================
story.append(Spacer(1, 0.5*inch))
footer_text = """
This report was automatically generated by NAVADA (Startup Viability Agent).
All analysis is based on provided data and should be supplemented with additional due diligence.
Past performance does not guarantee future results.
"""
story.append(Paragraph(footer_text, styles['Italic']))
# Build PDF
doc.build(story)
# Get PDF bytes
buffer.seek(0)
return buffer.getvalue()
def train_ml_model(df: pd.DataFrame):
"""
Train a Random Forest classifier to predict startup failure.
Args:
df (pd.DataFrame): Startup dataset with features and 'Failed' column
Returns:
RandomForestClassifier: Trained model
Features used:
- Funding_USD_M: Total funding in millions
- Burn_Rate_Months: Burn rate in months
- Founders_Experience_Yrs: Founder experience in years
- Market_Size_Bn: Market size in billions
Target:
- Failed: 0 = success, 1 = failed
"""
# Select feature columns
X = df[["Funding_USD_M", "Burn_Rate_Months", "Founders_Experience_Yrs", "Market_Size_Bn"]]
# Target variable
y = df["Failed"]
# Train Random Forest model
model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X, y)
return model
# =============================
# INTERACTIVE DASHBOARD FUNCTIONS
# =============================
def create_interactive_scatter(df_in: pd.DataFrame, title: str = "Interactive Startup Analysis") -> str:
"""
Create an interactive Plotly scatter plot with hover details and click functionality.
This function generates a comprehensive interactive visualization that allows users to:
- Hover over points to see detailed startup information
- Zoom and pan to explore specific regions of the data
- Filter by clicking legend items
- Export the chart in various formats
The scatter plot uses multiple visual encodings:
- X-axis: Funding amount (USD Millions)
- Y-axis: Burn rate (Months of runway)
- Size: Market size (larger bubble = bigger market)
- Color: Success/Failure status (green = success, red = failure)
- Symbol: Sector (different shapes for different industries)
Args:
df_in (pd.DataFrame): Input startup dataset with required columns:
- Funding_USD_M: Funding in millions USD
- Burn_Rate_Months: Burn rate in months
- Market_Size_Bn: Market size in billions
- Failed: 0 = success, 1 = failed
- Sector: Industry sector
- Country: Country of operation
title (str): Chart title for display
Returns:
str: Complete HTML string containing the interactive chart with embedded Plotly.js
"""
# Create the base scatter plot with multiple visual dimensions
# Each data point represents one startup with 6+ attributes encoded visually
fig = px.scatter(
df_in, # Input DataFrame
x="Funding_USD_M", # X-axis: funding amount
y="Burn_Rate_Months", # Y-axis: burn rate (runway)
size="Market_Size_Bn", # Bubble size: market opportunity
color="Failed", # Color coding: success (0) vs failure (1)
symbol="Sector", # Shape coding: different sectors
hover_data={
"Startup": True, # Show startup name on hover
"Founders_Experience_Yrs": True, # Show founder experience years
"Country": True, # Show country of operation
"Market_Size_Bn": ":.1f", # Format market size to 1 decimal
"Funding_USD_M": ":.1f", # Format funding to 1 decimal
"Burn_Rate_Months": ":.1f" # Format burn rate to 1 decimal
},
title=title, # Dynamic title from parameter
labels={
"Funding_USD_M": "Funding (USD Millions)", # User-friendly axis label
"Burn_Rate_Months": "Burn Rate (Months)", # User-friendly axis label
"Failed": "Status" # User-friendly legend label
},
color_discrete_map={0: "green", 1: "red"}, # Explicit color mapping
width=800, # Fixed width for consistency
height=600 # Fixed height for consistency
)
# Customize layout for enhanced user experience and interactivity
fig.update_layout(
hovermode="closest", # Show hover info for nearest point only
showlegend=True, # Display legend for color/symbol mapping
legend=dict( # Position legend horizontally at top
orientation="h", # Horizontal orientation saves space
yanchor="bottom", # Anchor to bottom of legend box
y=1.02, # Position slightly above chart
xanchor="right", # Align to right side
x=1 # Full width positioning
),
margin=dict( # Mobile-optimized margins
t=40, # Reduced top margin
b=40, # Reduced bottom margin
l=40, # Reduced left margin
r=40 # Reduced right margin
),
autosize=True, # Enable responsive sizing for mobile
font=dict(size=10) # Smaller font size for mobile readability
)
# Add user instruction annotation for better UX
# This helps users understand they can interact with the chart
fig.add_annotation(
text="Click and drag to zoom, hover for details, double-click to reset", # Clear instructions
showarrow=False, # No arrow pointing to anything
xref="paper", yref="paper", # Use paper coordinates (0-1 range)
x=0.5, y=-0.1, # Center horizontally, below chart
xanchor='center', yanchor='top', # Center the text anchor point
font=dict(size=12, color="gray") # Subtle gray color, readable size
)
# Convert Plotly figure to standalone HTML string
# include_plotlyjs='cdn' loads Plotly.js from CDN (smaller file size)
# div_id provides unique identifier for multiple charts on same page
return fig.to_html(include_plotlyjs='cdn', div_id="interactive-chart")
def create_interactive_timeline(df_in: pd.DataFrame) -> str:
"""
Create an interactive timeline showing failure progression over time.
Args:
df_in (pd.DataFrame): Input startup dataset
Returns:
str: HTML string of the interactive timeline
"""
# Calculate failure timeline (same logic as static version)
timeline_data = []
for _, row in df_in.iterrows():
if row["Funding_USD_M"] > 0 and row["Burn_Rate_Months"] > 0:
failure_time = row["Funding_USD_M"] / (12 / row["Burn_Rate_Months"])
timeline_data.append({
"Startup": row["Startup"],
"Failure_Time_Years": failure_time,
"Sector": row["Sector"],
"Funding": row["Funding_USD_M"],
"Status": "Failed" if row["Failed"] else "Active"
})
timeline_df = pd.DataFrame(timeline_data).sort_values("Failure_Time_Years")
# Create interactive bar chart
fig = px.bar(
timeline_df,
x="Failure_Time_Years",
y="Startup",
color="Status",
hover_data=["Sector", "Funding"],
title="📈 Interactive Failure Timeline - Hover for Details",
labels={
"Failure_Time_Years": "Estimated Failure Time (Years)",
"Startup": "Startup Name"
},
color_discrete_map={"Failed": "red", "Active": "green"},
orientation="h",
width=None, # Responsive width for mobile
height=400, # Reduced height for mobile
autosize=True # Enable auto-sizing
)
fig.update_layout(
hovermode="y unified",
yaxis={'categoryorder':'total ascending'}
)
return fig.to_html(include_plotlyjs='cdn', div_id="interactive-timeline")
def create_sector_dashboard(df_in: pd.DataFrame) -> str:
"""
Create an interactive multi-chart dashboard for sector analysis.
Args:
df_in (pd.DataFrame): Input startup dataset
Returns:
str: HTML string of the interactive dashboard
"""
from plotly.subplots import make_subplots
# Create subplot figure with secondary y-axis
fig = make_subplots(
rows=2, cols=2,
subplot_titles=(
"Average Funding by Sector",
"Failure Rate by Sector",
"Experience vs Funding",
"Market Size Distribution"
),
specs=[
[{"secondary_y": False}, {"secondary_y": False}],
[{"secondary_y": False}, {"secondary_y": False}]
]
)
# Chart 1: Average funding by sector
sector_avg = df_in.groupby("Sector")["Funding_USD_M"].mean().reset_index()
fig.add_trace(
go.Bar(x=sector_avg["Sector"], y=sector_avg["Funding_USD_M"],
name="Avg Funding", marker_color="lightblue"),
row=1, col=1
)
# Chart 2: Failure rate by sector
sector_failure = df_in.groupby("Sector")["Failed"].mean().reset_index()
fig.add_trace(
go.Bar(x=sector_failure["Sector"], y=sector_failure["Failed"],
name="Failure Rate", marker_color="salmon"),
row=1, col=2
)
# Chart 3: Experience vs Funding scatter
fig.add_trace(
go.Scatter(
x=df_in["Founders_Experience_Yrs"],
y=df_in["Funding_USD_M"],
mode="markers",
marker=dict(
size=8,
color=df_in["Failed"],
colorscale="RdYlGn_r",
showscale=True
),
name="Startups",
text=df_in["Startup"],
hovertemplate="%{text}
Experience: %{x} years
Funding: $%{y}M"
),
row=2, col=1
)
# Chart 4: Market size distribution
fig.add_trace(
go.Histogram(x=df_in["Market_Size_Bn"], nbinsx=10,
name="Market Size", marker_color="lightgreen"),
row=2, col=2
)
# Update layout
fig.update_layout(
height=600, # Reduced height for mobile
title_text="🏭 Interactive Sector Dashboard - Click and Zoom to Explore",
showlegend=True
)
return fig.to_html(include_plotlyjs='cdn', div_id="sector-dashboard")
# =============================
# SESSION MEMORY FUNCTIONS
# =============================
def get_session_id() -> str:
"""Get or create session ID for memory tracking."""
session = cl.user_session.get("session_id")
if not session:
import uuid
session = str(uuid.uuid4())[:8]
cl.user_session.set("session_id", session)
return session
def add_to_memory(session_id: str, role: str, content: str):
"""Add a message to session memory."""
if session_id not in SESSION_MEMORY:
SESSION_MEMORY[session_id] = []
SESSION_MEMORY[session_id].append({
"role": role,
"content": content,
"timestamp": pd.Timestamp.now()
})
# Keep only last 20 messages to avoid token limits
if len(SESSION_MEMORY[session_id]) > 20:
SESSION_MEMORY[session_id] = SESSION_MEMORY[session_id][-20:]
def get_memory_context(session_id: str) -> str:
"""Get formatted conversation history for context."""
if session_id not in SESSION_MEMORY:
return ""
history = SESSION_MEMORY[session_id][-10:] # Last 10 messages
context = "Recent conversation history:\n"
for msg in history:
context += f"- {msg['role']}: {msg['content'][:100]}...\n"
return context
def get_current_persona() -> Dict[str, str]:
"""Get current persona settings from session."""
persona_name = cl.user_session.get("persona", "founder")
return PERSONAS.get(persona_name, PERSONAS["founder"])
def format_persona_recommendations(persona_name: str) -> str:
"""Format key recommendations for a persona mode."""
persona = PERSONAS.get(persona_name, {})
recommendations = persona.get("key_recommendations", [])
if not recommendations:
return ""
formatted = f"\n\n🎯 **Key {persona.get('name', persona_name)} Recommendations:**\n\n"
for rec in recommendations:
formatted += f"• {rec}\n\n"
return formatted
# =============================
# LANGSMITH THREAD MANAGEMENT
# =============================
def get_thread_history(thread_id: str, project_name: str) -> List[Dict[str, str]]:
"""
Gets a history of all LLM calls in the thread to construct conversation history
Args:
thread_id (str): The thread/session ID to retrieve history for
project_name (str): LangSmith project name
Returns:
List[Dict[str, str]]: List of message objects with role and content
"""
if not langsmith_client:
return []
try:
# Filter runs by the specific thread and project
filter_string = f'and(in(metadata_key, ["session_id","conversation_id","thread_id"]), eq(metadata_value, "{thread_id}"))'
# Only grab the LLM runs
runs = [r for r in langsmith_client.list_runs(
project_name=project_name,
filter=filter_string,
run_type="llm"
)]
# Sort by start time to get chronological order
runs = sorted(runs, key=lambda run: run.start_time)
# Extract conversation history
messages = []
for run in runs:
if hasattr(run, 'inputs') and 'messages' in run.inputs:
# Add input messages
messages.extend(run.inputs['messages'])
# Add assistant response
if hasattr(run, 'outputs') and 'choices' in run.outputs:
assistant_msg = run.outputs['choices'][0]['message']
messages.append(assistant_msg)
return messages
except Exception as e:
print(f"Error retrieving thread history: {str(e)}")
return []
@traceable(
name="NAVADA Chat Pipeline",
run_type="chain",
tags=["navada", "startup-analysis", "conversational-ai"],
metadata={
"app_name": "navada",
"app_version": "2.0.0",
"environment": "production"
}
)
def navada_chat_pipeline(question: str, session_id: str, persona: str, get_chat_history: bool = False) -> str:
"""
Enhanced chat pipeline with LangSmith thread management for NAVADA
Args:
question (str): User's question/input
session_id (str): Unique session identifier for thread tracking
persona (str): Current persona mode (investor/founder)
get_chat_history (bool): Whether to retrieve conversation history
Returns:
str: AI response content
"""
try:
# Get current run tree for dynamic metadata and tags
current_run = ls.get_current_run_tree()
# Add dynamic metadata based on current context
if current_run:
current_run.metadata.update({
"session_id": session_id,
"persona_mode": persona,
"conversation_type": "thread_continuation" if get_chat_history else "new_conversation",
"question_length": len(question),
"timestamp": pd.Timestamp.now().isoformat()
})
# Add dynamic tags based on persona and question type
dynamic_tags = [f"persona-{persona}"]
# Detect question type and add relevant tags
question_lower = question.lower()
if "funding" in question_lower or "investment" in question_lower:
dynamic_tags.append("funding-analysis")
if "market" in question_lower or "competition" in question_lower:
dynamic_tags.append("market-analysis")
if "team" in question_lower or "founder" in question_lower:
dynamic_tags.append("team-analysis")
if "chart" in question_lower or "plot" in question_lower or "visualization" in question_lower:
dynamic_tags.append("data-visualization")
current_run.tags.extend(dynamic_tags)
# Set up LangSmith metadata for thread tracking
langsmith_extra = {
"project_name": LANGSMITH_PROJECT,
"metadata": {
"session_id": session_id,
"persona": persona,
"app": "navada",
"trace_type": "chat_pipeline"
},
"tags": [f"session-{session_id[:8]}", f"persona-{persona}"]
}
# Build conversation context
if get_chat_history and langsmith_client:
# Get LangSmith thread history
thread_messages = get_thread_history(session_id, LANGSMITH_PROJECT)
# Combine with new user question
messages = thread_messages + [{"role": "user", "content": question}]
else:
# Start fresh conversation
messages = [{"role": "user", "content": question}]
# Get current persona information
current_persona = PERSONAS.get(persona, PERSONAS["investor"])
# Add persona system message if starting fresh or no history
if not get_chat_history or not messages:
system_msg = {
"role": "system",
"content": current_persona["system_prompt"]
}
messages = [system_msg] + messages
# Create chat completion with LangSmith metadata
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
max_tokens=800,
temperature=0.7
)
return response.choices[0].message.content
except Exception as e:
print(f"Error in NAVADA chat pipeline: {str(e)}")
return f"I apologize, but I encountered an error processing your request: {str(e)}"
def get_session_metadata(session_id: str) -> Dict[str, Any]:
"""
Get metadata for a session including conversation count and current persona
Args:
session_id (str): Session identifier
Returns:
Dict[str, Any]: Session metadata
"""
if not langsmith_client:
return {"conversation_count": 0, "persona": "investor"}
try:
filter_string = f'and(in(metadata_key, ["session_id"]), eq(metadata_value, "{session_id}"))'
runs = list(langsmith_client.list_runs(
project_name=LANGSMITH_PROJECT,
filter=filter_string,
run_type="llm"
))
return {
"conversation_count": len(runs),
"persona": runs[-1].extra.get("metadata", {}).get("persona", "investor") if runs else "investor",
"last_interaction": runs[-1].start_time if runs else None
}
except Exception as e:
print(f"Error getting session metadata: {str(e)}")
return {"conversation_count": 0, "persona": "investor"}
# =============================
# AUTO-GENERATED INSIGHTS FUNCTIONS
# =============================
def generate_insights(df_in: pd.DataFrame, analysis_type: str = "general") -> Dict[str, List[str]]:
"""
Generate automated insights based on data analysis.
Args:
df_in (pd.DataFrame): Input dataset
analysis_type (str): Type of analysis performed
Returns:
Dict with risks, opportunities, and recommendations
"""
insights = {
"risks": [],
"opportunities": [],
"recommendations": []
}
# Calculate key metrics
avg_funding = df_in["Funding_USD_M"].mean()
failure_rate = df_in["Failed"].mean()
avg_burn = df_in["Burn_Rate_Months"].mean()
avg_experience = df_in["Founders_Experience_Yrs"].mean()
# Risk Detection
if failure_rate > 0.5:
insights["risks"].append(f"🔴 High failure rate detected: {failure_rate:.0%} of startups failed")
if avg_burn < 6:
insights["risks"].append(f"🔴 Short runway alert: Average burn rate is only {avg_burn:.1f} months")
if avg_experience < 3:
insights["risks"].append(f"🔴 Inexperienced teams: Average founder experience is {avg_experience:.1f} years")
# Opportunity Detection
high_funding_sectors = df_in.groupby("Sector")["Funding_USD_M"].mean().sort_values(ascending=False).head(2)
for sector, funding in high_funding_sectors.items():
if funding > avg_funding * 1.5:
insights["opportunities"].append(f"🟢 Hot sector identified: {sector} (avg funding ${funding:.1f}M)")
successful_patterns = df_in[df_in["Failed"] == 0]
if len(successful_patterns) > 0:
success_funding = successful_patterns["Funding_USD_M"].mean()
insights["opportunities"].append(f"🟢 Success pattern: Successful startups raised avg ${success_funding:.1f}M")
# Recommendations
if avg_burn < 12:
insights["recommendations"].append("💡 Extend runway: Focus on increasing funding or reducing burn rate")
if failure_rate > 0.4:
insights["recommendations"].append("💡 De-risk strategy: Consider pivot to sectors with lower failure rates")
insights["recommendations"].append("💡 Track metrics: Monitor burn rate, customer acquisition, and team experience")
return insights
def format_insights_message(insights: Dict[str, List[str]]) -> str:
"""Format insights into a readable message."""
message = "## 🤖 Auto-Generated Insights\n\n"
if insights["risks"]:
message += "### ⚠️ Top Risks Detected:\n"
for risk in insights["risks"][:3]: # Top 3 risks
message += f"- {risk}\n"
message += "\n"
if insights["opportunities"]:
message += "### 🎯 Opportunities Identified:\n"
for opp in insights["opportunities"][:3]: # Top 3 opportunities
message += f"- {opp}\n"
message += "\n"
if insights["recommendations"]:
message += "### 💡 Next Steps:\n"
for rec in insights["recommendations"][:3]: # Top 3 recommendations
message += f"- {rec}\n"
return message
# =============================
# WEB SCRAPING FUNCTIONS
# =============================
def validate_url(url: str) -> bool:
"""
Validate URL format and check for security concerns.
Args:
url (str): URL to validate
Returns:
bool: True if URL is valid and safe, False otherwise
"""
try:
# Parse URL to check structure
parsed = urlparse(url)
# Must have scheme (http/https) and netloc (domain)
if not all([parsed.scheme, parsed.netloc]):
return False
# Only allow HTTP/HTTPS protocols for security
if parsed.scheme not in ['http', 'https']:
return False
# Block dangerous or inappropriate domains
blocked_domains = [
'localhost', '127.0.0.1', '0.0.0.0', # Local addresses
'file://', 'ftp://', # Non-web protocols
]
netloc_lower = parsed.netloc.lower()
for blocked in blocked_domains:
if blocked in netloc_lower:
return False
return True
except Exception:
return False
def scrape_site(url: str, selector: str = "p") -> Dict[str, Any]:
"""
Scrape text content from a website with comprehensive safety measures and error handling.
This function performs web scraping with multiple safeguards:
- URL validation and security checks
- Request timeouts and size limits
- Content filtering and cleaning
- Structured error reporting
Args:
url (str): Website URL to scrape (must be http/https)
selector (str): CSS selector for content extraction
- "p" = paragraphs (default)
- "h1, h2, h3" = headings
- ".class-name" = by CSS class
- "#id-name" = by element ID
Returns:
Dict[str, Any]: Scraping results containing:
- success (bool): Whether scraping succeeded
- data (pd.DataFrame): Scraped content (if successful)
- url (str): Original URL
- count (int): Number of items scraped
- error (str): Error message (if failed)
- size_mb (float): Content size in megabytes
"""
result = {
"success": False,
"data": pd.DataFrame(),
"url": url,
"count": 0,
"error": "",
"size_mb": 0.0
}
try:
# Step 1: Validate URL for security and format
if not validate_url(url):
result["error"] = "Invalid or unsafe URL. Use http/https URLs only."
return result
# Step 2: Configure HTTP request with safety limits
headers = {
'User-Agent': 'NAVADA-Bot/1.0 (Educational Web Scraping Tool)',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
'Accept-Language': 'en-US,en;q=0.5',
'Accept-Encoding': 'gzip, deflate',
'Connection': 'keep-alive',
'Upgrade-Insecure-Requests': '1',
}
# Step 3: Make HTTP request with timeout and size limits
response = requests.get(
url,
headers=headers,
timeout=15, # 15 second timeout
stream=True, # Stream to check size before downloading
allow_redirects=True, # Follow redirects (max 30 by default)
verify=True # Verify SSL certificates
)
# Step 4: Check response size before processing (max 5MB)
content_length = response.headers.get('content-length')
if content_length and int(content_length) > 5 * 1024 * 1024: # 5MB limit
result["error"] = "Content too large (>5MB). Choose a smaller page."
return result
# Step 5: Check HTTP status code
response.raise_for_status() # Raises exception for 4xx/5xx status codes
# Step 6: Get content and check actual size
content = response.text
content_size_mb = len(content.encode('utf-8')) / (1024 * 1024)
result["size_mb"] = round(content_size_mb, 2)
if content_size_mb > 5: # Double-check size after download
result["error"] = f"Content too large ({content_size_mb:.1f}MB). Choose a smaller page."
return result
# Step 7: Parse HTML with BeautifulSoup
soup = BeautifulSoup(content, "html.parser")
# Step 8: Remove script and style elements (they contain non-content)
for script in soup(["script", "style", "nav", "footer", "header"]):
script.decompose()
# Step 9: Extract content using CSS selector
elements = soup.select(selector)
# Step 10: Clean and filter extracted text
scraped_content = []
for element in elements:
text = element.get_text(strip=True)
# Filter out empty content and very short text
if text and len(text) > 10:
# Clean whitespace and normalize
text = re.sub(r'\s+', ' ', text) # Replace multiple whitespace with single space
text = text.strip()
# Limit individual text length to prevent spam
if len(text) <= 2000:
scraped_content.append(text)
# Step 11: Create structured DataFrame
if scraped_content:
result["data"] = pd.DataFrame({
"content": scraped_content,
"length": [len(text) for text in scraped_content],
"source": [url] * len(scraped_content)
})
result["count"] = len(scraped_content)
result["success"] = True
else:
result["error"] = f"No content found using selector '{selector}'. Try different selectors like 'h1', 'div', or 'span'."
except requests.exceptions.Timeout:
result["error"] = "Request timed out. The website may be slow or unresponsive."
except requests.exceptions.ConnectionError:
result["error"] = "Could not connect to the website. Check the URL and internet connection."
except requests.exceptions.HTTPError as e:
result["error"] = f"HTTP error {e.response.status_code}: {e.response.reason}"
except requests.exceptions.RequestException as e:
result["error"] = f"Request failed: {str(e)}"
except Exception as e:
result["error"] = f"Scraping failed: {str(e)}"
return result
def analyze_scraped_content(scraped_data: pd.DataFrame, url: str, persona: Dict[str, str]) -> str:
"""
Use GPT to analyze scraped website content with persona-specific focus.
Args:
scraped_data (pd.DataFrame): DataFrame containing scraped content
url (str): Original URL for context
persona (Dict[str, str]): Current user persona (investor/founder)
Returns:
str: AI analysis of the scraped content
"""
if scraped_data.empty:
return "No content available for analysis."
# Prepare content for GPT analysis
# Take top 20 content items to stay within token limits
content_items = scraped_data.head(20)["content"].tolist()
content_text = "\n\n".join([f"Section {i+1}: {text}" for i, text in enumerate(content_items)])
# Truncate if too long (approximately 3000 tokens = 12000 characters)
if len(content_text) > 10000:
content_text = content_text[:10000] + "\n\n[Content truncated for analysis...]"
# Create persona-specific analysis prompt
persona_focus = persona.get('system_prompt', '')
analysis_style = ""
if 'investor' in persona.get('name', '').lower():
analysis_style = (
"Focus on investment opportunities, market analysis, business models, "
"competitive landscape, and financial indicators. Identify potential risks and ROI factors."
)
else: # founder mode
analysis_style = (
"Focus on actionable insights, operational strategies, market positioning, "
"customer needs, and execution opportunities. Provide tactical recommendations."
)
try:
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "system",
"content": (
f"You are analyzing web content from {url}. "
f"{analysis_style}\n\n"
"Provide a structured analysis with:\n"
"1. Key insights (3-5 bullet points)\n"
"2. Notable patterns or trends\n"
"3. Actionable recommendations\n"
"4. Potential concerns or red flags\n\n"
"Keep analysis concise but insightful."
)
},
{
"role": "user",
"content": f"Analyze this website content:\n\nURL: {url}\n\nContent:\n{content_text}"
}
],
max_tokens=600,
temperature=0.7
)
return response.choices[0].message.content
except Exception as e:
return f"Analysis failed: {str(e)}"
# =============================
# INTERNET SEARCH FUNCTIONALITY
# =============================
@traceable(
name="NAVADA Internet Search",
run_type="tool",
tags=["navada", "search", "brave-api", "market-intelligence"],
metadata={"tool_type": "internet_search", "api_provider": "brave_search"}
)
def search_internet(query: str, count: int = 5) -> Dict[str, Any]:
"""
Search the internet using Brave Search API
Args:
query (str): Search query
count (int): Number of results to return (default 5)
Returns:
dict: Search results with titles, descriptions, and URLs
"""
# Add dynamic metadata to current run
current_run = ls.get_current_run_tree()
if current_run:
current_run.metadata.update({
"search_query": query,
"requested_count": count,
"query_length": len(query),
"timestamp": pd.Timestamp.now().isoformat()
})
# Add query-specific tags
query_lower = query.lower()
search_tags = []
if "startup" in query_lower:
search_tags.append("startup-search")
if "funding" in query_lower or "investment" in query_lower:
search_tags.append("funding-search")
if "market" in query_lower:
search_tags.append("market-research")
if "competition" in query_lower:
search_tags.append("competitive-analysis")
current_run.tags.extend(search_tags)
if not search_api_key or search_api_key == "your_brave_search_api_key_here":
logger.warning("Search API key not configured or using placeholder value")
return {
"success": False,
"error": "Search API key not configured. Please set SEARCH_API_KEY in your .env file with a valid Brave Search API key.",
"results": [],
"fallback_available": True
}
try:
# Brave Search API endpoint
url = "https://api.search.brave.com/res/v1/web/search"
headers = {
"Accept": "application/json",
"Accept-Encoding": "gzip",
"X-Subscription-Token": search_api_key
}
params = {
"q": query,
"count": count,
"offset": 0,
"mkt": "en-US",
"safesearch": "moderate",
"freshness": "pw", # Past week for fresh results
"text_decorations": False,
"search_lang": "en"
}
response = requests.get(url, headers=headers, params=params, timeout=10)
response.raise_for_status()
data = response.json()
# Extract relevant information from search results
results = []
if "web" in data and "results" in data["web"]:
for result in data["web"]["results"][:count]:
results.append({
"title": result.get("title", ""),
"description": result.get("description", ""),
"url": result.get("url", ""),
"age": result.get("age", ""),
"language": result.get("language", "en")
})
return {
"success": True,
"query": query,
"results": results,
"total_results": len(results)
}
except requests.exceptions.HTTPError as e:
logger.error(f"HTTP error during search: {e}")
if e.response.status_code == 422:
return {
"success": False,
"error": "Search API authentication failed. Please check your SEARCH_API_KEY in .env file.",
"results": [],
"fallback_available": True,
"status_code": 422
}
else:
return {
"success": False,
"error": f"Search request failed: HTTP {e.response.status_code}",
"results": [],
"fallback_available": True
}
except requests.exceptions.RequestException as e:
logger.error(f"Network error during search: {e}")
return {
"success": False,
"error": f"Search request failed: {str(e)}",
"results": [],
"fallback_available": True
}
except Exception as e:
logger.error(f"Unexpected error during search: {e}")
logger.error(f"Traceback: {traceback.format_exc()}")
return {
"success": False,
"error": f"Search error: {str(e)}",
"results": [],
"fallback_available": True
}
def analyze_search_results(search_data: Dict[str, Any], persona: Dict[str, str], context: str = "", session_id: str = None) -> str:
"""
Analyze search results using AI based on current persona mode
Args:
search_data (dict): Search results from search_internet()
persona (dict): Current persona (investor/founder mode)
context (str): Additional context for analysis
Returns:
str: AI analysis of search results
"""
if not search_data["success"] or not search_data["results"]:
return "No search results to analyze or search failed."
try:
# Format search results for analysis
results_text = f"Search Query: {search_data['query']}\n\n"
results_text += f"Found {search_data['total_results']} results:\n\n"
for i, result in enumerate(search_data['results'], 1):
results_text += f"{i}. **{result['title']}**\n"
results_text += f" URL: {result['url']}\n"
results_text += f" Description: {result['description']}\n"
if result.get('age'):
results_text += f" Age: {result['age']}\n"
results_text += "\n"
# Create persona-specific analysis prompt
analysis_prompt = f"{persona['system_prompt']}\n\n"
analysis_prompt += f"Analyze these search results from a {persona['name']} perspective.\n\n"
if context:
analysis_prompt += f"Context: {context}\n\n"
analysis_prompt += "Provide insights, opportunities, risks, and actionable recommendations based on the search results."
# Include LangSmith metadata if session_id is available
if langsmith_client and session_id:
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "system",
"content": analysis_prompt
},
{
"role": "user",
"content": f"Analyze these search results:\n\n{results_text}"
}
],
max_tokens=800,
temperature=0.7,
)
else:
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "system",
"content": analysis_prompt
},
{
"role": "user",
"content": f"Analyze these search results:\n\n{results_text}"
}
],
max_tokens=800,
temperature=0.7
)
return response.choices[0].message.content
except Exception as e:
return f"Analysis failed: {str(e)}"
def generate_search_query(user_question: str, persona: str) -> str:
"""
Generate intelligent search queries based on user question and persona mode
Args:
user_question (str): User's original question
persona (str): Current persona mode (investor/founder)
Returns:
str: Optimized search query for Brave Search API
"""
try:
# Persona-specific search query templates
investor_keywords = {
"market": "startup market trends valuation 2025",
"competition": "startup competitive landscape industry analysis",
"funding": "venture capital funding trends startup investment 2025",
"exit": "startup exit strategies IPO acquisition trends",
"valuation": "startup valuation metrics Series A B C funding",
"growth": "startup growth metrics scaling strategies",
"roi": "startup ROI investment returns venture capital",
"due diligence": "startup due diligence checklist investment"
}
founder_keywords = {
"market": "startup market validation product-market fit",
"competition": "startup competitor analysis differentiation",
"funding": "startup fundraising tips pitch deck Series A",
"growth": "startup growth hacking scaling strategies",
"team": "startup team building hiring strategies",
"product": "startup product development MVP strategies",
"customer": "startup customer acquisition retention strategies",
"pivot": "startup pivot strategies when to pivot"
}
# Select keyword set based on persona
keywords = investor_keywords if persona == "investor" else founder_keywords
# Extract key topics from user question
question_lower = user_question.lower()
# Find matching keywords and build search query
search_terms = []
for topic, search_template in keywords.items():
if topic in question_lower:
search_terms.append(search_template)
break # Use first match to avoid overly complex queries
# Add year context for recent information
current_year = "2025"
if current_year not in user_question:
search_terms.append(current_year)
# If no specific keywords found, use general startup search
if not search_terms:
if persona == "investor":
search_terms = ["startup investment trends 2025", "venture capital market"]
else:
search_terms = ["startup trends 2025", "founder advice entrepreneurship"]
# Combine and return the search query
return " ".join(search_terms[:2]) # Limit to 2 main search terms
except Exception as e:
print(f"Error generating search query: {str(e)}")
return f"startup {persona} trends 2025"
# =============================
# TEXT-TO-SPEECH FUNCTIONALITY
# =============================
def generate_speech(text: str, voice: str = "alloy") -> bytes:
"""
Generate speech from text using OpenAI TTS API
Args:
text (str): Text to convert to speech
voice (str): Voice to use (alloy, echo, fable, onyx, nova, shimmer)
Returns:
bytes: Audio data in MP3 format
"""
try:
if not api_key:
raise Exception("OpenAI API key not configured")
response = client.audio.speech.create(
model="tts-1",
voice=voice,
input=text,
speed=1.0
)
return response.content
except Exception as e:
print(f"TTS generation failed: {str(e)}")
return b""
def create_audio_message(content: str, voice: str = "alloy") -> cl.Audio:
"""
Create Chainlit audio message with TTS
Args:
content (str): Text content to convert to speech
voice (str): Voice to use for TTS
Returns:
cl.Audio: Chainlit audio element
"""
try:
# Limit text length for TTS (OpenAI has character limits)
max_length = 4000
if len(content) > max_length:
# Truncate but try to end at a sentence
truncated = content[:max_length]
last_period = truncated.rfind('.')
if last_period > max_length * 0.8: # If period is reasonably close to end
content = truncated[:last_period + 1]
else:
content = truncated + "..."
# Generate speech
audio_data = generate_speech(content, voice)
if not audio_data:
return None
# Create audio element
audio = cl.Audio(
content=audio_data,
name="navada_response.mp3",
display="inline",
auto_play=False
)
return audio
except Exception as e:
print(f"Audio message creation failed: {str(e)}")
return None
async def collect_startup_information(template_type: str, original_message) -> dict:
"""
Collect personalized startup information through interactive questionnaire
"""
try:
# Define template-specific questions
question_sets = {
"business_case": [
{"key": "company_name", "question": "🏢 What is your company/startup name?", "default": "Your Startup"},
{"key": "industry", "question": "🏭 What industry are you in? (e.g., fintech, healthcare, e-commerce)", "default": "Technology"},
{"key": "problem_statement", "question": "❗ What specific problem does your startup solve?", "default": "A significant market problem"},
{"key": "solution", "question": "💡 What is your solution/product?", "default": "An innovative solution"},
{"key": "target_market", "question": "🎯 Who is your target market/customer?", "default": "B2B/B2C customers"},
{"key": "funding_need", "question": "💰 How much funding do you need? (e.g., $500K, $2M)", "default": "$1M"},
{"key": "funding_use", "question": "📊 How will you use the funding? (e.g., product development, marketing)", "default": "Product development and marketing"}
],
"pitch_deck": [
{"key": "company_name", "question": "🏢 What is your company name?", "default": "Your Startup"},
{"key": "tagline", "question": "📢 What's your company tagline or one-liner?", "default": "Revolutionizing the industry"},
{"key": "industry", "question": "🏭 What industry/market are you in?", "default": "Technology"},
{"key": "team_size", "question": "👥 How many team members do you have?", "default": "3-5 people"},
{"key": "stage", "question": "🚀 What stage is your startup? (idea, MVP, early revenue, growth)", "default": "MVP"},
{"key": "funding_round", "question": "💼 What funding round are you raising? (pre-seed, seed, Series A)", "default": "Seed"}
],
"financial_model": [
{"key": "company_name", "question": "🏢 Company name?", "default": "Your Startup"},
{"key": "business_model", "question": "💼 What's your business model? (SaaS, marketplace, e-commerce, etc.)", "default": "SaaS"},
{"key": "revenue_streams", "question": "💰 What are your main revenue streams?", "default": "Subscription fees"},
{"key": "pricing", "question": "💵 What's your pricing model? (e.g., $50/month, $500/year)", "default": "$50/month"},
{"key": "projected_customers", "question": "📈 How many customers do you project in Year 1?", "default": "1000"},
{"key": "growth_rate", "question": "📊 What's your projected monthly growth rate? (e.g., 10%, 20%)", "default": "15%"}
],
"investor_memo": [
{"key": "company_name", "question": "🏢 Company name?", "default": "Your Startup"},
{"key": "industry", "question": "🏭 Industry?", "default": "Technology"},
{"key": "funding_round", "question": "💼 Funding round?", "default": "Series A"},
{"key": "funding_amount", "question": "💰 Funding amount needed?", "default": "$5M"},
{"key": "valuation", "question": "📈 Pre-money valuation?", "default": "$20M"},
{"key": "key_metrics", "question": "📊 Key traction metrics? (e.g., ARR, users, growth rate)", "default": "$1M ARR, 50% YoY growth"}
],
"executive_summary": [
{"key": "company_name", "question": "🏢 Company name?", "default": "Your Startup"},
{"key": "industry", "question": "🏭 Industry sector?", "default": "Technology"},
{"key": "mission", "question": "🎯 What is your company's mission?", "default": "To revolutionize the industry"},
{"key": "target_market", "question": "👥 Who is your target market?", "default": "SMB businesses"},
{"key": "competitive_advantage", "question": "🚀 What's your key competitive advantage?", "default": "Proprietary technology"}
],
"market_analysis": [
{"key": "company_name", "question": "🏢 Company name?", "default": "Your Startup"},
{"key": "industry", "question": "🏭 Target industry?", "default": "Technology"},
{"key": "geographic_market", "question": "🌍 Geographic market focus? (e.g., US, Europe, Global)", "default": "North America"},
{"key": "market_size", "question": "📏 Estimated market size? (TAM)", "default": "$50B"},
{"key": "growth_rate", "question": "📈 Market growth rate? (annual %)", "default": "15%"}
],
"term_sheet": [
{"key": "company_name", "question": "🏢 Company name?", "default": "Your Startup"},
{"key": "funding_amount", "question": "💰 Investment amount?", "default": "$2M"},
{"key": "valuation", "question": "📈 Post-money valuation?", "default": "$10M"},
{"key": "investor_name", "question": "🤝 Lead investor name?", "default": "Venture Capital Partners"},
{"key": "equity_percentage", "question": "📊 Equity percentage offered?", "default": "20%"}
]
}
# Get questions for the template type
questions = question_sets.get(template_type, question_sets["business_case"])
# Initialize data dictionary
startup_data = {}
# Send introduction message
intro_msg = await cl.Message(
content=f"📋 **Let's personalize your {template_type.replace('_', ' ').title()}!**\n\n"
f"I'll ask you {len(questions)} quick questions to tailor the document to your startup.\n\n"
f"**First question:**"
).send()
# Ask each question and collect responses
for i, q in enumerate(questions, 1):
# Ask the question
question_msg = await cl.Message(
content=f"**{i}/{len(questions)}** {q['question']}\n\n"
f"💡 *Example: {q['default']}*"
).send()
# Wait for user response
user_response = await cl.AskUserMessage(
content=f"Please provide your answer for: {q['question']}",
timeout=120
).send()
if user_response:
# Store the response or use default if empty
answer = user_response.content.strip() if user_response.content.strip() else q['default']
startup_data[q['key']] = answer
# Acknowledge the response
await cl.Message(
content=f"✅ Got it: **{answer}**"
).send()
else:
# User didn't respond, use default
startup_data[q['key']] = q['default']
await cl.Message(
content=f"⏱️ Using default: **{q['default']}**"
).send()
# Add some calculated defaults for missing fields
startup_data.update({
'executive_summary': f"Executive summary for {startup_data.get('company_name', 'Your Startup')}",
'tam': '$50B',
'sam': '$5B',
'som': '$100M',
'target_irr': '35%',
'exit_timeline': '5-7 years',
'exit_multiple': '10x'
})
# Confirmation message
await cl.Message(
content=f"🎉 **Perfect! I have all the information needed.**\n\n"
f"Generating your personalized **{template_type.replace('_', ' ').title()}** for **{startup_data['company_name']}**..."
).send()
return startup_data
except Exception as e:
await cl.Message(
content=f"❌ Error collecting information: {str(e)}\n\n"
f"Using default template instead..."
).send()
# Return basic defaults if questionnaire fails
return {
'company_name': 'Your Startup',
'industry': 'Technology',
'problem_statement': 'A significant market problem',
'solution': 'An innovative solution',
'target_market': 'B2B/B2C customers',
'funding_need': '$1M',
'executive_summary': 'Executive summary',
'tam': '$50B',
'sam': '$5B',
'som': '$100M',
'target_irr': '35%',
'exit_timeline': '5-7 years',
'exit_multiple': '10x'
}
def benchmark_founder_idea(features: Dict[str, Any], df: pd.DataFrame) -> Dict[str, Any]:
"""
Benchmark a founder's idea against dataset averages and percentiles.
Args:
features (dict): Founder's startup metrics
df (pd.DataFrame): Dataset to benchmark against
Returns:
dict: Benchmarking results with percentiles and recommendations
"""
results = {
"metrics": {},
"insights": [],
"risk_level": "",
"recommendations": []
}
# Calculate dataset statistics
stats = {
"funding": {
"median": df["Funding_USD_M"].median(),
"mean": df["Funding_USD_M"].mean(),
"p20": df["Funding_USD_M"].quantile(0.2),
"p80": df["Funding_USD_M"].quantile(0.8)
},
"burn_rate": {
"median": df["Burn_Rate_Months"].median(),
"mean": df["Burn_Rate_Months"].mean(),
"p20": df["Burn_Rate_Months"].quantile(0.2),
"p80": df["Burn_Rate_Months"].quantile(0.8)
},
"experience": {
"median": df["Founders_Experience_Yrs"].median(),
"mean": df["Founders_Experience_Yrs"].mean(),
"p20": df["Founders_Experience_Yrs"].quantile(0.2),
"p80": df["Founders_Experience_Yrs"].quantile(0.8)
},
"market": {
"median": df["Market_Size_Bn"].median(),
"mean": df["Market_Size_Bn"].mean(),
"p20": df["Market_Size_Bn"].quantile(0.2),
"p80": df["Market_Size_Bn"].quantile(0.8)
}
}
# Benchmark funding
funding = features.get('funding_usd_m', 3.0)
funding_percentile = (df["Funding_USD_M"] < funding).mean() * 100
results["metrics"]["funding"] = {
"value": funding,
"percentile": funding_percentile,
"vs_median": funding / stats["funding"]["median"] if stats["funding"]["median"] > 0 else 0,
"vs_mean": funding / stats["funding"]["mean"] if stats["funding"]["mean"] > 0 else 0
}
# Benchmark burn rate
burn = features.get('burn_rate_months', 9.0)
burn_percentile = (df["Burn_Rate_Months"] < burn).mean() * 100
results["metrics"]["burn_rate"] = {
"value": burn,
"percentile": burn_percentile,
"vs_median": burn / stats["burn_rate"]["median"] if stats["burn_rate"]["median"] > 0 else 0,
"vs_mean": burn / stats["burn_rate"]["mean"] if stats["burn_rate"]["mean"] > 0 else 0
}
# Benchmark experience
experience = features.get('team_experience_years', 3.0)
exp_percentile = (df["Founders_Experience_Yrs"] < experience).mean() * 100
results["metrics"]["experience"] = {
"value": experience,
"percentile": exp_percentile,
"vs_median": experience / stats["experience"]["median"] if stats["experience"]["median"] > 0 else 0,
"vs_mean": experience / stats["experience"]["mean"] if stats["experience"]["mean"] > 0 else 0
}
# Benchmark market size
market = features.get('market_size_bn', 10.0)
market_percentile = (df["Market_Size_Bn"] < market).mean() * 100
results["metrics"]["market_size"] = {
"value": market,
"percentile": market_percentile,
"vs_median": market / stats["market"]["median"] if stats["market"]["median"] > 0 else 0,
"vs_mean": market / stats["market"]["mean"] if stats["market"]["mean"] > 0 else 0
}
# Generate insights
if burn_percentile < 30:
results["insights"].append(f"⚠️ Your burn rate ({burn} months) is in the **bottom 30%** - HIGH RISK! Most startups have longer runways.")
results["recommendations"].append("Reduce burn rate or secure additional funding urgently")
if burn_percentile > 70:
results["insights"].append(f"✅ Your burn rate ({burn} months) is in the **top 30%** - well-managed cash flow")
if funding_percentile < 30:
results["insights"].append(f"⚠️ Your funding (${funding}M) is in the **bottom 30%** - may need more capital")
results["recommendations"].append("Consider raising a larger round to extend runway")
if funding_percentile > 70:
results["insights"].append(f"✅ Your funding (${funding}M) is in the **top 30%** - strong financial position")
if exp_percentile < 30:
results["insights"].append(f"⚠️ Your team experience ({experience} years) is **below dataset median** - consider adding senior advisors")
results["recommendations"].append("Add experienced advisors or co-founders to the team")
if exp_percentile > 70:
results["insights"].append(f"✅ Your team experience ({experience} years) is in the **top 30%** - strong foundation")
if results["metrics"]["market_size"]["vs_median"] > 2:
results["insights"].append(f"🚀 Your market size (${market}B) is **{results['metrics']['market_size']['vs_median']:.1f}× bigger** than average!")
# Calculate risk level
risk_score = 0
if burn_percentile < 30: risk_score += 2
if funding_percentile < 30: risk_score += 2
if exp_percentile < 30: risk_score += 1
if risk_score >= 3:
results["risk_level"] = "HIGH RISK 🔴"
elif risk_score >= 1:
results["risk_level"] = "MODERATE RISK 🟡"
else:
results["risk_level"] = "LOW RISK 🟢"
return results
def create_portfolio_heatmap(portfolio_df: pd.DataFrame) -> bytes:
"""
Create a heatmap visualization of multiple startups' viability scores.
Args:
portfolio_df (pd.DataFrame): DataFrame with startup names and viability metrics
Returns:
bytes: PNG image of heatmap
"""
# Calculate viability scores for each startup
scores_data = []
for _, row in portfolio_df.iterrows():
# Calculate viability score for this startup
features = {
'funding_usd_m': row.get('Funding_USD_M', 3.0),
'burn_rate_months': row.get('Burn_Rate_Months', 9.0),
'team_experience_years': row.get('Founders_Experience_Yrs', 3.0),
'market_size_bn': row.get('Market_Size_Bn', 10.0),
'business_model_strength_1_5': row.get('Business_Model', 3),
'moat_1_5': row.get('Moat', 3),
'traction_mrr_k': row.get('Traction_MRR_K', 10),
'growth_rate_pct': row.get('Growth_Rate_Pct', 5),
'competition_intensity_1_5': row.get('Competition', 3)
}
result = viability_score(features)
scores_data.append({
'Startup': row['Startup'],
'Overall Score': result['score'],
'Runway': result['components']['runway'] * 100,
'Experience': result['components']['experience'] * 100,
'Market': result['components']['market'] * 100,
'Traction': result['components']['traction'] * 100,
'Growth': result['components']['growth'] * 100
})
# Create DataFrame for heatmap
heatmap_df = pd.DataFrame(scores_data)
heatmap_matrix = heatmap_df.set_index('Startup')[['Overall Score', 'Runway', 'Experience', 'Market', 'Traction', 'Growth']]
# Create heatmap with mobile-optimized size
default_height = max(6, len(heatmap_df) * 0.5)
figsize = get_mobile_optimized_figsize(10, default_height)
fig, ax = plt.subplots(figsize=figsize)
# Create heatmap with custom colormap
sns.heatmap(
heatmap_matrix.T,
annot=True,
fmt='.1f',
cmap='RdYlGn',
vmin=0,
vmax=100,
cbar_kws={'label': 'Score (0-100)'},
linewidths=0.5,
linecolor='gray',
ax=ax
)
ax.set_title('Portfolio Viability Heatmap\n🔴 Poor (0-40) | 🟡 Moderate (40-60) | 🟢 Strong (60-100)',
fontsize=14, fontweight='bold')
ax.set_xlabel('Startups', fontsize=12)
ax.set_ylabel('Metrics', fontsize=12)
# Rotate x-axis labels for readability
plt.xticks(rotation=45, ha='right')
plt.tight_layout()
return fig_to_bytes(fig)
# =============================
# THREAD MANAGEMENT FOR LANGSMITH
# =============================
def get_thread_history(thread_id: str, project_name: str) -> List[Dict[str, Any]]:
"""
Retrieve conversation history from LangSmith for a specific thread.
Args:
thread_id: Unique identifier for the conversation thread
project_name: LangSmith project name
Returns:
List of message dictionaries representing the conversation history
"""
if not langsmith_client:
return []
try:
# Filter runs by the specific thread and project
filter_string = f'and(in(metadata_key, ["session_id","conversation_id","thread_id"]), eq(metadata_value, "{thread_id}"))'
# Only grab the LLM runs
runs = list(langsmith_client.list_runs(
project_name=project_name,
filter=filter_string,
run_type="llm"
))
if not runs:
return []
# Sort by start time to get chronological order
runs = sorted(runs, key=lambda run: run.start_time)
# Extract messages from runs
messages = []
for run in runs:
if run.inputs and 'messages' in run.inputs:
messages.extend(run.inputs['messages'])
if run.outputs and 'choices' in run.outputs:
if run.outputs['choices'] and run.outputs['choices'][0].get('message'):
messages.append(run.outputs['choices'][0]['message'])
return messages
except Exception as e:
print(f"Error retrieving thread history: {e}")
return []
@traceable(name="NAVADA Chat Pipeline")
def process_with_thread_context(
question: str,
session_id: str,
get_chat_history: bool = True,
persona: Optional[Dict[str, str]] = None
) -> str:
"""
Process user message with thread context for continuity.
Args:
question: User's current question
session_id: Thread/session identifier
get_chat_history: Whether to retrieve and use conversation history
persona: Current persona configuration
Returns:
AI response as string
"""
langsmith_extra = {
"project_name": LANGSMITH_PROJECT,
"metadata": {"session_id": session_id}
}
messages = []
# Retrieve conversation history if requested
if get_chat_history and langsmith_client:
try:
historical_messages = get_thread_history(session_id, LANGSMITH_PROJECT)
if historical_messages:
messages.extend(historical_messages)
except Exception as e:
print(f"Could not retrieve history: {e}")
# Add system prompt based on persona
if persona:
messages.insert(0, {
"role": "system",
"content": persona.get('system_prompt', '')
})
# Add current user question
messages.append({"role": "user", "content": question})
# Make API call with thread metadata
try:
if langsmith_client:
# Use LangSmith thread pattern from documentation
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
max_tokens=800,
temperature=0.7,
)
else:
# Standard OpenAI call without LangSmith
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
max_tokens=800,
temperature=0.7
)
return response.choices[0].message.content
except Exception as e:
return f"Error processing message: {str(e)}"
# =============================
# PASSKEY AUTHENTICATION SYSTEM
# =============================
# Global variables for authentication state
PASSKEY = "54321" # Temporary passkey for testing
AUTHENTICATED_SESSIONS = set() # Track authenticated session IDs
async def check_authentication():
"""Check if user is authenticated."""
session_id = cl.user_session.get("session_id")
if not session_id:
session_id = str(uuid.uuid4())
cl.user_session.set("session_id", session_id)
# Check if already authenticated
if session_id in AUTHENTICATED_SESSIONS:
return True
# Show NAVADA with padlock icon
await cl.Message(content="**NAVADA** 🔐").send()
return False
# =============================
# CHAINLIT EVENT HANDLERS
# =============================
# Action handlers are not needed for this implementation
# Downloads will be handled through direct file sending
@cl.on_chat_start
async def start():
"""
Initialize the chat session when a user first connects.
This function runs once at the start of each new chat session and:
1. Sets up chat settings with About section and quick actions
2. Initializes thread/session tracking for LangSmith
3. Sends a brief welcome message
The settings panel (burger menu) contains detailed information
about NAVADA's capabilities.
Chainlit decorator: @cl.on_chat_start
- Automatically called when a new chat begins
- Async function for non-blocking UI operations
"""
# -------------------------
# PASSKEY AUTHENTICATION CHECK
# -------------------------
# Generate a unique session ID for this conversation thread
session_id = str(uuid.uuid4())
cl.user_session.set("session_id", session_id)
# Authenticate user with passkey
if not await check_authentication():
return # Exit if authentication fails
# -------------------------
# INITIALIZE THREAD/SESSION TRACKING
# -------------------------
# Store session info in Chainlit user session for persistence
cl.user_session.set("session_id", session_id)
cl.user_session.set("conversation_history", [])
cl.user_session.set("thread_metadata", {
"session_id": session_id,
"project_name": LANGSMITH_PROJECT,
"start_time": pd.Timestamp.now().isoformat()
})
# Store in global thread sessions mapping
THREAD_SESSIONS[session_id] = {
"start_time": pd.Timestamp.now(),
"messages": [],
"persona": "founder" # Default persona
}
# Log thread initialization
if langsmith_client:
print(f"🧵 Thread initialized: {session_id[:8]}...")
# -------------------------
# SETUP CHAT SETTINGS WITH ABOUT SECTION
# -------------------------
# Create chat settings with an "About" section accessible via burger menu
about_content = (
"# 👋 Welcome to NAVADA\n\n"
"**NAVADA** (Startup Viability Agent) helps you analyze startup risk, funding, and failure patterns "
"with **interactive charts and AI analysis**.\n\n"
"## 🎭 Analysis Modes:\n\n"
"💼 **Investor Mode** - VC perspective focused on ROI and exit strategies\n"
"🚀 **Founder Mode** - Entrepreneur perspective focused on execution and growth\n"
"🇬🇧 **UK Economist Mode** - Economic analysis perspective for UK markets\n\n"
"## 📊 Advanced Charts:\n\n"
"🔹 **Growth Trajectory** - MRR growth patterns vs company age\n"
"🔹 **Team Performance** - Team size vs founder experience matrix\n"
"🔹 **Market Opportunity** - Market size vs competition analysis\n"
"🔹 **Funding Efficiency** - Capital efficiency and ROI analysis\n"
"🔹 **Stage Progression** - Funding stages vs failure rates\n"
"🔹 **Risk Assessment** - Comprehensive risk radar chart\n"
"🔹 **UK Economic Dashboard** - Macroeconomic indicators and regional analysis\n\n"
"## 📈 Interactive Tools:\n\n"
"🔹 **Interactive Scatter** - Dynamic correlations and filtering\n"
"🔹 **Sector Dashboard** - Multi-dimensional sector analysis\n"
"🔹 **Interactive Timeline** - Failure patterns over time\n\n"
"## 🤖 AI-Powered Features:\n\n"
"🔹 **assess idea** - Interactive viability scoring with 24 data points\n"
"🔹 **benchmark** - Compare your startup against 24 successful companies\n"
"🔹 **portfolio** - Analyze multiple startups with heatmap visualization\n"
"🔹 **insights** - AI-powered risk assessment and opportunities\n"
"🔹 **questions** - Guided questions based on your current mode\n"
"🔹 **macro analysis** - UK macroeconomic impact assessment\n\n"
"## 🔍 Internet Search:\n\n"
"🔹 **search [query]** - Get up-to-date market intelligence and trends\n"
"🔹 **latest news** - Current developments in startup ecosystem\n"
"🔹 **current trends** - Market shifts and opportunities\n"
"🔹 Auto-triggered for questions about recent events or market updates\n\n"
"## 📥 Download & Export:\n\n"
"🔹 **Charts** - Download any generated chart as PNG with built-in download buttons\n"
"🔹 **export data** / **download csv** - Export complete dataset as CSV\n"
"🔹 **export json** / **download json** - Export complete dataset as JSON\n"
"🔹 **Data Tables** - Download chart data as CSV alongside visualizations\n\n"
"## 💬 Get Started:\n\n"
"• Type **'investor mode'**, **'founder mode'**, or **'economist mode'** to set your perspective\n"
"• Type **'questions'** to get guided analysis questions\n"
"• Ask: \"Which chart should I look at first?\"\n"
"• Try: \"Show me funding efficiency\" or \"Risk assessment\"\n\n"
"---\n\n"
"**Ready to start?** Choose your mode and dive into comprehensive startup analysis!"
)
# Store TTS setting in user session (default off)
cl.user_session.set("tts_enabled", False)
# -------------------------
# SEND SIMPLE WELCOME MESSAGE
# -------------------------
# Send basic welcome message
await cl.Message(content="**NAVADA**").send()
# Try to inject the widget via Chainlit's HTML element
try:
await cl.Html(content=elevenlabs_widget, display="page").send()
except Exception as e:
print(f"⚠️ Could not inject ElevenLabs widget: {e}")
# Alternative: provide instructions to user
await cl.Message(
content="**Manual Setup:** To enable voice chat, add this HTML to your page:\n```html\n" + elevenlabs_widget + "\n```"
).send()
async def generate_speech(text: str) -> cl.Audio:
"""Generate speech from text using OpenAI TTS."""
try:
# Generate speech using OpenAI TTS
response = base_client.audio.speech.create(
model="tts-1",
voice="alloy",
input=text[:4000] # Limit text length
)
# Save audio to bytes
audio_bytes = response.content
# Create Chainlit audio element
audio_element = cl.Audio(
content=audio_bytes,
mime="audio/mpeg",
display="inline"
)
return audio_element
except Exception as e:
print(f"⚠️ TTS Error: {e}")
return None
# Voice commands are handled in the main message handler
# =============================
# INPUT HELPER FUNCTIONS
# =============================
async def ask_float(prompt: str, default: float) -> float:
"""
Prompt user for a floating-point number input with a default value.
This helper function handles the complexity of asking for numeric input
in Chainlit, including error handling and default values.
Args:
prompt (str): Question to ask the user
default (float): Default value if user presses Enter or input is invalid
Returns:
float: The user's input as a float, or the default value
Error handling:
- Empty input → returns default
- Invalid input (non-numeric) → returns default
- Timeout (600 seconds) → returns default
Example:
>>> funding = await ask_float("Enter funding amount:", 3.0)
# User sees: "Enter funding amount: (default 3.0):"
# If user enters "5.5" → returns 5.5
# If user presses Enter → returns 3.0
"""
# Send question to user with timeout of 600 seconds (10 minutes)
msg = await cl.AskUserMessage(
content=f"{prompt} (default {default}):",
timeout=600
).send()
# Try to parse the response as a float
try:
# Check if message exists and has 'output' field
if msg and msg.get('output'):
return float(msg['output']) # Convert string to float
return float(default) # No input provided, use default
except Exception:
# If any error occurs (ValueError, AttributeError, etc.), use default
return float(default)
async def ask_int(prompt: str, default: int, mi: int = 1, ma: int = 5) -> int:
"""
Prompt user for an integer input within a specified range.
This helper function asks for integer input and enforces min/max bounds.
Commonly used for rating scales (1-5).
Args:
prompt (str): Question to ask the user
default (int): Default value if input is invalid or empty
mi (int): Minimum allowed value (default: 1)
ma (int): Maximum allowed value (default: 5)
Returns:
int: The user's input clamped to [mi, ma], or default value
Behavior:
- Input is automatically clamped to the min/max range
- Example: If range is 1-5 and user enters 7, returns 5
- Example: If range is 1-5 and user enters 0, returns 1
Example:
>>> rating = await ask_int("Rate your moat:", 3, mi=1, ma=5)
# User sees: "Rate your moat: [1..5] (default 3):"
# If user enters "4" → returns 4
# If user enters "10" → returns 5 (clamped to max)
# If user enters "0" → returns 1 (clamped to min)
"""
# Send question to user showing the valid range
msg = await cl.AskUserMessage(
content=f"{prompt} [{mi}..{ma}] (default {default}):",
timeout=600
).send()
# Try to parse the response as an integer and clamp to range
try:
if msg and msg.get('output'):
val = int(msg['output']) # Convert string to int
else:
val = default # No input provided, use default
# Clamp value to [mi, ma] range
# max(mi, val) ensures val >= mi
# min(ma, ...) ensures result <= ma
return max(mi, min(ma, val))
except Exception:
# If any error occurs, use default value
return default
# =============================
# CSV UPLOAD HANDLER
# =============================
async def handle_csv_upload():
"""
Handle CSV file upload from user to replace the current dataset.
This function manages the entire CSV upload workflow:
1. Prompts user to upload a CSV file
2. Validates file type and size
3. Parses CSV into DataFrame
4. Calculates derived columns if possible
5. Returns the new DataFrame or None on failure
Returns:
pd.DataFrame or None: New dataset if upload succeeds, None otherwise
File requirements:
- Format: CSV (comma-separated values)
- Max size: 5 MB
- Recommended columns: Startup, Funding_USD_M, Burn_Rate_Months,
Failed, Country, Sector, Founders_Experience_Yrs, Market_Size_Bn
Automatic processing:
- If Funding_USD_M and Burn_Rate_Months exist, Est_Failure_Year
is automatically calculated
- Missing columns are allowed but may affect functionality
User feedback:
- Confirmation message showing filename and row count
- Error message if upload fails or is cancelled
"""
# -------------------------
# PROMPT FOR FILE UPLOAD
# -------------------------
# Request CSV file from user with constraints
files = await cl.AskFileMessage(
content="Upload a CSV with columns like Startup, Funding_USD_M, Burn_Rate_Months, Failed, Country, Sector",
accept=["text/csv"], # Only allow CSV files
max_size_mb=5, # Limit file size to 5 MB
timeout=600 # 10 minute timeout
).send()
# -------------------------
# HANDLE CANCELLATION
# -------------------------
# If user cancels or no file is provided
if not files:
await cl.Message(content="No file received. Keeping current dataset.").send()
return None # Return None to indicate no change
# -------------------------
# PARSE CSV FILE
# -------------------------
# Extract first file from the list (usually only one file uploaded)
f = files[0]
# Read CSV from bytes into pandas DataFrame
# f.content is bytes, so wrap in BytesIO for pandas
new_df = pd.read_csv(io.BytesIO(f.content))
# -------------------------
# CALCULATE DERIVED COLUMNS
# -------------------------
# Check if required columns exist for failure year calculation
# Using set intersection to check if both columns are present
if {"Funding_USD_M", "Burn_Rate_Months"}.issubset(new_df.columns):
# Calculate estimated failure year using same formula as original data
new_df["Est_Failure_Year"] = FUNDING_YEAR + (
new_df["Funding_USD_M"] / new_df["Burn_Rate_Months"]
)
# -------------------------
# CONFIRM UPLOAD SUCCESS
# -------------------------
# Send confirmation message showing filename and row count
await cl.Message(
content=f"Loaded `{f.name}` with {len(new_df)} rows."
).send()
# Return the new DataFrame to replace the global dataset
return new_df
# =============================
# MAIN MESSAGE HANDLER
# =============================
@cl.on_message
@traceable(
name="NAVADA Message Handler",
run_type="chain",
tags=["navada", "message-handler", "conversation-entry"],
metadata={"handler_type": "chainlit_message", "app_version": "2.0.0"}
)
async def main(message: cl.Message):
"""
Process every user message and route to appropriate handler.
This is the main message processing function that runs whenever a user
sends a message. It implements a routing pattern to handle different
types of requests:
1. Chart generation commands (timeline, funding vs burn)
2. CSV upload requests
3. Viability assessment requests
4. General Q&A using GPT-4
Message routing logic:
----------------------
- Show thinking indicator for user feedback
- Lowercase the input for case-insensitive matching
- Check for specific keywords to determine intent
- Execute corresponding handler and return
- If no specific pattern matches, fallback to AI chat
Args:
message (cl.Message): Chainlit message object containing:
- content: User's text input
- author: User identifier
- Other metadata
Global variables:
df (pd.DataFrame): Can be modified by CSV upload handler
Chainlit decorator: @cl.on_message
- Automatically called for every user message
- Async function for non-blocking operations
"""
global df # Allow modification of global dataset
# -------------------------
# PASSKEY AUTHENTICATION CHECK
# -------------------------
session_id = cl.user_session.get("session_id", "unknown")
# Handle passkey authentication
user_input_raw = message.content.strip()
# Check if user is trying to authenticate with the passkey
if user_input_raw == PASSKEY and session_id not in AUTHENTICATED_SESSIONS:
# Authenticate the user
AUTHENTICATED_SESSIONS.add(session_id)
await cl.Message(content="✅ **Access Granted!**").send()
return
# Check if user is authenticated
if session_id not in AUTHENTICATED_SESSIONS:
await cl.Message(content="🔐").send()
return
# Set authenticated user info
auth_status = {
"authenticated": True,
"username": "authenticated_user",
"user_id": f"user_{session_id[:8]}",
"email": "user@navada.ai",
"subscription_tier": "free"
}
# Parse user input (already extracted above in authentication check)
user_input = user_input_raw.lower()
# Handle authentication commands (commented out for testing)
# if user_input.startswith("login "):
# await handle_login_command(user_input_raw)
# return
# if user_input.startswith("register "):
# await handle_register_command(user_input_raw)
# return
# If not authenticated and not using auth commands, show login form (disabled)
# if not auth_status["authenticated"]:
# await show_login_form()
# return
# -------------------------
# THINKING INDICATOR & TIMESTAMP
# -------------------------
# Show thinking indicator to provide immediate user feedback
import datetime
timestamp = datetime.datetime.now().strftime("%H:%M:%S")
# Show brief thinking message (will be removed after processing)
thinking_msg = cl.Message(content=f"🤔 Thinking... ({timestamp})")
await thinking_msg.send()
# -------------------------
# NORMALIZE INPUT
# -------------------------
# Convert to lowercase and strip whitespace for consistent matching
user_input = message.content.strip().lower()
# =============================
# ROUTE 0: VOICE COMMANDS
# =============================
if user_input in ["voice on", "voice enable", "tts on", "speech on", "audio on"]:
cl.user_session.set("tts_enabled", True)
await cl.Message(content="🔊 **Voice enabled!** AI responses will now include audio.").send()
return
if user_input in ["voice off", "voice disable", "tts off", "speech off", "audio off"]:
cl.user_session.set("tts_enabled", False)
await cl.Message(content="🔇 **Voice disabled.** AI responses will be text only.").send()
return
# =============================
# ROUTE 0.5: MATHEMATICAL ANALYSIS MODE
# =============================
if "math mode" in user_input or "analysis mode" in user_input:
await thinking_msg.remove()
# Create math-enabled analysis environment
math_context = {
'df': df,
'np': np,
'stats': stats,
'current_startup': cl.user_session.get('selected_startup')
}
await cl.Message(
content="## 🧮 Mathematical Analysis Mode\n\n"
"You can now perform complex calculations:\n\n"
"**Examples:**\n"
"• `calculate IRR for 5x return in 7 years`\n"
"• `project revenue with 20% monthly growth`\n"
"• `simulate 1000 scenarios for exit`\n"
"• `optimize burn rate for 18 month runway`\n\n"
"Type your calculation or 'exit math mode' to return."
).send()
cl.user_session.set("math_mode", True)
return
# Check if user is in math mode
if cl.user_session.get("math_mode", False):
await thinking_msg.remove()
if user_input in ["exit math mode", "exit", "return", "back"]:
cl.user_session.set("math_mode", False)
await cl.Message(content="🔄 **Exited math mode.** Back to regular analysis.").send()
return
# Process math command
math_context = {
'df': df,
'np': np,
'stats': stats,
'current_startup': cl.user_session.get('selected_startup')
}
result = await process_math_command(user_input, math_context)
await cl.Message(content=result).send()
return
# =============================
# ROUTE 0.7: IMAGE GENERATION
# =============================
if detect_image_request(user_input):
await thinking_msg.remove()
# Extract the image prompt from user input
image_prompt = extract_image_prompt(user_input_raw)
# Show image generation message
generation_msg = cl.Message(content=f"🎨 **Generating image:** {image_prompt}\n\nThis may take a few moments...")
await generation_msg.send()
# Generate the image
result = await generate_image(image_prompt)
if result["success"]:
# Create image element from URL
image_element = cl.Image(
url=result["image_url"],
name=f"generated_image_{timestamp.replace(':', '')}.png",
display="inline"
)
# Update message with success
success_content = f"✅ **Image generated successfully!**\n\n"
success_content += f"**Original prompt:** {result['original_prompt']}\n"
if result.get('revised_prompt') and result['revised_prompt'] != result['original_prompt']:
success_content += f"**DALL-E revised prompt:** {result['revised_prompt']}\n"
success_content += f"**Size:** {result['size']} | **Quality:** {result['quality']}"
await cl.Message(content=success_content, elements=[image_element]).send()
# Log for conversation tracking
auth_manager.save_conversation(
user_id=auth_status.get("user_id", "test_123"),
chainlit_session_id=cl.user_session.get("session_id", "unknown"),
role="user",
content=f"Image generation request: {image_prompt}",
metadata={"action_type": "image_generation", "prompt": image_prompt}
)
auth_manager.save_conversation(
user_id=auth_status.get("user_id", "test_123"),
chainlit_session_id=cl.user_session.get("session_id", "unknown"),
role="assistant",
content=f"Generated image: {result['image_url']}",
metadata={"action_type": "image_generated", "image_url": result["image_url"]}
)
else:
# Handle error
error_content = f"❌ **Image generation failed**\n\n"
error_content += f"**Error:** {result['error']}\n"
error_content += f"**Prompt:** {result['original_prompt']}\n\n"
error_content += "Please try again with a different prompt or check your API configuration."
await cl.Message(content=error_content).send()
return
# =============================
# ROUTE 0.8: SWOT ANALYSIS
# =============================
if "swot" in user_input or ("swot mode" in user_input) or ("analyze swot" in user_input):
await thinking_msg.remove()
# Show SWOT analysis message
analysis_msg = cl.Message(content="📊 **Generating SWOT Analysis...**\n\nAnalyzing strengths, weaknesses, opportunities, and threats...")
await analysis_msg.send()
# Gather startup context from user session or create generic context
startup_context = {
"persona": cl.user_session.get("persona", "founder"),
"selected_startup": cl.user_session.get("selected_startup"),
"session_id": cl.user_session.get("session_id", "unknown")
}
# Add any available startup data from the dataframe
startup_data = {}
if df is not None and not df.empty:
# Get a sample of startups for context
sample_data = df.head(3).to_dict('records')
startup_data["sample_startups"] = sample_data
startup_data["total_startups"] = len(df)
startup_data["sectors"] = df['Sector'].value_counts().head(5).to_dict() if 'Sector' in df.columns else {}
# Generate SWOT analysis using AI
swot_analysis = await generate_swot_analysis(startup_data, user_input_raw)
# Create SWOT visualization
try:
swot_chart_png = plot_swot_matrix(swot_analysis)
# Create SWOT DataFrame for download
swot_df = swot_analysis.to_dataframe()
# Send the comprehensive SWOT analysis
swot_summary = swot_analysis.summary()
await cl.Message(content=swot_summary).send()
# Send the visual SWOT matrix
await send_chart_with_download(
png_data=swot_chart_png,
filename=f"swot_analysis_{timestamp.replace(':', '')}.png",
description="📊 **SWOT Analysis Matrix** - Visual representation of your startup analysis",
csv_data=swot_df
)
# Log for conversation tracking
auth_manager.save_conversation(
user_id=auth_status.get("user_id", "test_123"),
chainlit_session_id=cl.user_session.get("session_id", "unknown"),
role="user",
content=f"SWOT analysis request: {user_input_raw}",
metadata={"action_type": "swot_analysis", "context": startup_context}
)
auth_manager.save_conversation(
user_id=auth_status.get("user_id", "test_123"),
chainlit_session_id=cl.user_session.get("session_id", "unknown"),
role="assistant",
content="Generated comprehensive SWOT analysis with matrix visualization",
metadata={
"action_type": "swot_generated",
"strengths_count": len(swot_analysis.strengths),
"weaknesses_count": len(swot_analysis.weaknesses),
"opportunities_count": len(swot_analysis.opportunities),
"threats_count": len(swot_analysis.threats)
}
)
except Exception as e:
# Fallback to text-only SWOT if visualization fails
error_msg = f"⚠️ **SWOT Analysis Generated** (Chart error: {str(e)})\n\n"
error_msg += swot_analysis.summary()
await cl.Message(content=error_msg).send()
return
# =============================
# ROUTE 1: FAILURE TIMELINE CHART
# =============================
if "timeline" in user_input:
# Show loading message while generating chart
msg = cl.Message(content="📊 Generating failure timeline chart...")
await msg.send()
# Generate the chart PNG
png = plot_failure_timeline(df)
# Remove loading message
await msg.remove()
# Send chart with download options
# Select available columns for CSV export
preferred_cols = ['Company', 'Total Funding', 'Monthly Burn Rate', 'Estimated Runway (Months)']
available_cols = [col for col in preferred_cols if col in df.columns]
# If no preferred columns exist, use first 4 columns or all if less than 4
if not available_cols:
available_cols = df.columns.tolist()[:4]
await send_chart_with_download(
png_data=png,
filename="failure_timeline.png",
description=(
"### 📈 Estimated Failure Timeline\n\n"
"This chart shows when each startup is projected to fail "
"based on their funding and burn rate."
),
csv_data=df[available_cols] if available_cols else df
)
return # Exit handler, don't process further
# =============================
# ROUTE 2: FUNDING VS BURN CHART
# =============================
# Check for multiple possible phrasings
if "funding vs burn" in user_input or (
"funding" in user_input and "burn" in user_input and "vs" in user_input
):
# Show loading message
msg = cl.Message(content="📊 Generating funding vs burn chart...")
await msg.send()
# Generate scatter plot
png = plot_funding_vs_burn(df)
# Remove loading message
await msg.remove()
# Send descriptive text with legend explanation
text_msg = cl.Message(
content=(
"### 💰 Funding vs Burn Rate Analysis\n\n"
"**Green** = Successful | **Red** = Failed\n"
"Each shape represents a different sector."
)
)
await text_msg.send()
# Attach chart image
image = cl.Image(content=png, name="funding_vs_burn.png", display="inline")
await image.send(for_id=text_msg.id)
return # Exit handler
# =============================
# ROUTE 3: ADDITIONAL CHART COMMANDS
# =============================
# Sector comparison chart
if "sector" in user_input and ("compare" in user_input or "comparison" in user_input or "chart" in user_input):
msg = cl.Message(content="📊 Generating sector comparison chart...")
await msg.send()
png = plot_sector_comparison(df)
await msg.remove()
text_msg = cl.Message(content="### 🏭 Sector Comparison\n\nAverage funding by industry sector.")
await text_msg.send()
image = cl.Image(content=png, name="sector_comparison.png", display="inline")
await image.send(for_id=text_msg.id)
return
# Failure rate by country
if ("failure" in user_input or "fail" in user_input) and "country" in user_input:
msg = cl.Message(content="📊 Generating failure rate by country chart...")
await msg.send()
png = plot_failure_rate_by_country(df)
await msg.remove()
text_msg = cl.Message(content="### 🌍 Failure Rate by Country\n\nPercentage of failed startups per country.")
await text_msg.send()
image = cl.Image(content=png, name="failure_rate_country.png", display="inline")
await image.send(for_id=text_msg.id)
return
# Experience vs success
if "experience" in user_input and ("success" in user_input or "funding" in user_input or "chart" in user_input):
msg = cl.Message(content="📊 Generating experience vs success chart...")
await msg.send()
png = plot_experience_vs_success(df)
await msg.remove()
text_msg = cl.Message(content="### 👥 Experience vs Success\n\nRelationship between founder experience, funding, and outcome.")
await text_msg.send()
image = cl.Image(content=png, name="experience_success.png", display="inline")
await image.send(for_id=text_msg.id)
return
# =============================
# ROUTE 4: CSV UPLOAD
# =============================
if "upload csv" in user_input or "load csv" in user_input:
# Call upload handler and get new DataFrame (or None)
new_df = await handle_csv_upload()
# If upload succeeded, replace global dataset
if new_df is not None:
df = new_df
return # Exit handler
# =============================
# ROUTE 4: VIABILITY ASSESSMENT
# =============================
# Check for various phrasings of assessment request
if "assess idea" in user_input or "new idea" in user_input or "viability" in user_input:
# -------------------------
# INTRODUCTION
# -------------------------
# Explain the assessment process
await cl.Message(
content=(
"## 🎯 Startup Viability Assessment\n\n"
"Great! Let's evaluate your startup idea. "
"I'll ask you **9 quick questions** to calculate a "
"comprehensive viability score.\n\n"
"*Press Enter to use default values, or type your answer.*"
)
).send()
# -------------------------
# COLLECT INPUT DATA
# -------------------------
# Ask 9 questions to gather all required features
# Each question has a sensible default value
funding = await ask_float("Funding (USD Millions)", 3.0)
burn = await ask_float(
"Burn rate (months of runway if spending 1M/year ≈ 83k/month)", 9.0
)
expy = await ask_float("Team experience (years, average)", 3.0)
market = await ask_float("Market size (Billions USD)", 15.0)
bm = await ask_int("Business model strength (1=weak..5=excellent)", 3)
moat = await ask_int("Moat/defensibility (1..5)", 3)
mrrk = await ask_float("Current MRR (in $k)", 10.0)
growth = await ask_float("Monthly growth rate (%)", 6.0)
comp = await ask_int("Competition intensity (1=low..5=very high)", 3)
# -------------------------
# PACKAGE FEATURES
# -------------------------
# Create dictionary matching the viability_score function signature
feats = {
"funding_usd_m": funding,
"burn_rate_months": burn,
"team_experience_years": expy,
"market_size_bn": market,
"business_model_strength_1_5": bm,
"moat_1_5": moat,
"traction_mrr_k": mrrk,
"growth_rate_pct": growth,
"competition_intensity_1_5": comp
}
# -------------------------
# CALCULATE SCORE
# -------------------------
# Call viability scoring model
result = viability_score(feats)
# -------------------------
# DISPLAY GAUGE CHART
# -------------------------
# Generate and display visual score gauge
gauge_png = plot_viability_gauge(result["score"])
gauge_msg = cl.Message(content="\n## 📊 Your Viability Score")
await gauge_msg.send()
# Attach gauge image to message
gauge_image = cl.Image(
content=gauge_png, name="viability_score.png", display="inline"
)
await gauge_image.send(for_id=gauge_msg.id)
# -------------------------
# DISPLAY DETAILED RESULTS
# -------------------------
# Interpret score with color-coded assessment
score_interpretation = (
"🟢 Strong" if result['score'] >= 60
else "🟡 Moderate" if result['score'] >= 40
else "🔴 Weak"
)
# Format comprehensive summary with all metrics
summary = (
f"### Overall Assessment: {score_interpretation}\n\n"
f"**Final Score:** {result['score']:.1f}/100\n\n"
f"#### 📈 Key Metrics:\n"
f"• **Estimated Runway:** ~{result['survival_months']:.1f} months\n"
f"• **Projected Failure Year:** {result['est_failure_year']:.2f} "
f"(funded {FUNDING_YEAR})\n\n"
f"#### 🔍 Score Breakdown (0-1 scale):\n"
f"• Runway: {result['components']['runway']:.2f} | "
f"Experience: {result['components']['experience']:.2f}\n"
f"• Market: {result['components']['market']:.2f} | "
f"Business Model: {result['components']['business_model']:.2f}\n"
f"• Moat: {result['components']['moat']:.2f} | "
f"Traction: {result['components']['traction']:.2f}\n"
f"• Growth: {result['components']['growth']:.2f} | "
f"Competition: {result['components']['competition']:.2f}\n"
)
await cl.Message(content=summary).send()
# -------------------------
# DISPLAY RECOMMENDATIONS
# -------------------------
# Show actionable tips based on weaknesses
if result["tips"]:
# Format tips as bullet list
tips_text = (
"### 💡 Recommended Actions\n\n" +
"\n".join([f"• {tip}" for tip in result["tips"]])
)
await cl.Message(content=tips_text).send()
else:
# If no tips, startup looks strong
await cl.Message(
content=(
"### ✅ Looking Good!\n\n"
"• Keep executing—your foundations look solid.\n"
"• Focus on consistent growth and customer acquisition."
)
).send()
return # Exit handler
# =============================
# ROUTE 6: BENCHMARK IDEA
# =============================
if "benchmark idea" in user_input or "benchmark my idea" in user_input or "compare my idea" in user_input:
await cl.Message(content="## 🎯 Benchmark Your Startup Idea\n\nI'll compare your metrics against our dataset to see how you stack up!").send()
# Collect startup metrics from user
funding = await ask_float("Your funding amount (USD millions)", 3.0)
burn = await ask_float("Your burn rate (months)", 9.0)
expy = await ask_float("Your team's average experience (years)", 3.0)
market = await ask_float("Your target market size (billions USD)", 10.0)
# Package features for benchmarking
features = {
"funding_usd_m": funding,
"burn_rate_months": burn,
"team_experience_years": expy,
"market_size_bn": market
}
# Run benchmarking analysis
benchmark_results = benchmark_founder_idea(features, df)
# Display benchmarking results
results_text = f"## 📊 Benchmarking Results\n\n"
results_text += f"**Risk Level:** {benchmark_results['risk_level']}\n\n"
results_text += "### 📈 How You Compare:\n\n"
# Display key insights
for insight in benchmark_results['insights']:
results_text += f"• {insight}\n"
results_text += "\n### 📊 Detailed Metrics:\n\n"
# Funding percentile
funding_data = benchmark_results['metrics']['funding']
results_text += f"**Funding:** ${funding_data['value']}M (percentile: {funding_data['percentile']:.0f}%)\n"
results_text += f"• {funding_data['vs_median']:.1f}× the median startup\n\n"
# Burn rate percentile
burn_data = benchmark_results['metrics']['burn_rate']
results_text += f"**Burn Rate:** {burn_data['value']} months (percentile: {burn_data['percentile']:.0f}%)\n"
results_text += f"• {burn_data['vs_median']:.1f}× the median startup\n\n"
# Experience percentile
exp_data = benchmark_results['metrics']['experience']
results_text += f"**Experience:** {exp_data['value']} years (percentile: {exp_data['percentile']:.0f}%)\n"
results_text += f"• {exp_data['vs_median']:.1f}× the median startup\n\n"
# Market size percentile
market_data = benchmark_results['metrics']['market_size']
results_text += f"**Market Size:** ${market_data['value']}B (percentile: {market_data['percentile']:.0f}%)\n"
results_text += f"• {market_data['vs_median']:.1f}× the median startup\n\n"
# Recommendations
if benchmark_results['recommendations']:
results_text += "### 💡 Recommendations:\n\n"
for rec in benchmark_results['recommendations']:
results_text += f"• {rec}\n"
await cl.Message(content=results_text).send()
return # Exit handler
# =============================
# ROUTE 7: PORTFOLIO MODE
# =============================
if "portfolio" in user_input and ("mode" in user_input or "analysis" in user_input or "analyze" in user_input or user_input.strip() == "portfolio"):
# Remove thinking indicator
await thinking_msg.remove()
await cl.Message(content="## 📊 Portfolio Analysis Mode\n\nI'll create a comprehensive heatmap of all startups in your dataset!").send()
# Show loading message
msg = cl.Message(content="🔥 Generating portfolio heatmap... Calculating viability scores for all startups.")
await msg.send()
try:
# Generate portfolio heatmap
heatmap_bytes = create_portfolio_heatmap(df)
# Remove loading message
await msg.remove()
# Send heatmap description
desc_msg = cl.Message(
content="### 🔥 Portfolio Viability Heatmap\n\n"
f"**Analysis of {len(df)} startups across 6 key metrics:**\n"
"• **Overall Score** - Combined viability (0-100)\n"
"• **Runway** - Financial sustainability\n"
"• **Experience** - Team expertise\n"
"• **Market** - Market opportunity\n"
"• **Traction** - Current momentum\n"
"• **Growth** - Growth trajectory\n\n"
"**Color Guide:** 🔴 Poor (0-40) | 🟡 Moderate (40-60) | 🟢 Strong (60-100)"
)
await desc_msg.send()
# Attach heatmap image
heatmap_image = cl.Image(
content=heatmap_bytes, name="portfolio_heatmap.png", display="inline"
)
await heatmap_image.send(for_id=desc_msg.id)
# Calculate and show investment recommendations
portfolio_scores = []
for _, row in df.iterrows():
features = {
'funding_usd_m': row['Funding_USD_M'],
'burn_rate_months': row['Burn_Rate_Months'],
'team_experience_years': row['Founders_Experience_Yrs'],
'market_size_bn': row['Market_Size_Bn'],
'business_model_strength_1_5': 3,
'moat_1_5': 3,
'traction_mrr_k': 10,
'growth_rate_pct': 5,
'competition_intensity_1_5': 3
}
score = viability_score(features)['score']
portfolio_scores.append((row['Startup'], score))
# Sort by score
portfolio_scores.sort(key=lambda x: x[1], reverse=True)
# Create recommendations
reco_text = "### 🎯 Investment Recommendations:\n\n"
reco_text += "**🟢 INVEST (Score ≥ 60):**\n"
invest_list = [f"• {name} ({score:.1f})" for name, score in portfolio_scores if score >= 60]
if invest_list:
reco_text += "\n".join(invest_list) + "\n\n"
else:
reco_text += "• None in this category\n\n"
reco_text += "**🟡 MONITOR (Score 40-59):**\n"
monitor_list = [f"• {name} ({score:.1f})" for name, score in portfolio_scores if 40 <= score < 60]
if monitor_list:
reco_text += "\n".join(monitor_list) + "\n\n"
else:
reco_text += "• None in this category\n\n"
reco_text += "**🔴 PASS (Score < 40):**\n"
pass_list = [f"• {name} ({score:.1f})" for name, score in portfolio_scores if score < 40]
if pass_list:
reco_text += "\n".join(pass_list) + "\n\n"
else:
reco_text += "• None in this category\n\n"
# Portfolio statistics
avg_score = sum(score for _, score in portfolio_scores) / len(portfolio_scores)
high_performers = len([s for _, s in portfolio_scores if s >= 60])
reco_text += f"**📊 Portfolio Stats:**\n"
reco_text += f"• Average Score: {avg_score:.1f}/100\n"
reco_text += f"• High Performers: {high_performers}/{len(df)} ({high_performers/len(df)*100:.1f}%)\n"
reco_text += f"• Success Rate: {((df['Failed'] == 0).sum()/len(df)*100):.1f}%"
await cl.Message(content=reco_text).send()
except Exception as e:
await msg.remove()
await cl.Message(
content=f"❌ Error generating portfolio analysis: {str(e)}\n\n"
f"Please try again or contact support if the issue persists."
).send()
return # Exit handler
# =============================
# ROUTE: UK MACRO ANALYSIS
# =============================
if "macro analysis" in user_input or "uk analysis" in user_input:
await thinking_msg.remove()
analyzer = UKEconomicsAnalyzer()
await cl.Message(content="## 🇬🇧 UK Macroeconomic Impact Analysis\n\nI'll analyze how UK economic conditions affect your startup.").send()
# Collect startup data
funding = await ask_float("Funding (£ millions)", 4.0)
location = await cl.AskUserMessage(content="Location (London/Manchester/Edinburgh/Birmingham/Bristol/Cambridge):").send()
sector = await cl.AskUserMessage(content="Sector (FinTech/HealthTech/GreenTech/EdTech/RetailTech):").send()
team_size = await ask_int("Team size", 10, mi=1, ma=500)
startup_data = {
'funding_usd_m': funding * 1.27, # Convert to USD
'location': location.get('output', 'London'),
'sector': sector.get('output', 'Tech'),
'team_size': team_size,
'burn_rate_months': 12, # Default
'debt_ratio': 0.3, # Default 30% debt
'is_b2b': True
}
# Run analysis
macro_impacts = analyzer.analyze_macro_impact(startup_data)
# Generate chart
chart = plot_uk_economic_indicators(df)
# Display results
content = f"""
## 🇬🇧 UK Economic Impact Assessment
### Interest Rate Environment
- **Cost of Capital:** {macro_impacts['interest_rate_impact']['cost_of_capital']:.1f}%
- **Annual Interest Cost:** £{macro_impacts['interest_rate_impact']['annual_interest_cost']*0.79:.0f}k
- **Impact Level:** {macro_impacts['interest_rate_impact']['impact_level']}
- **Recommendation:** {macro_impacts['interest_rate_impact']['recommendation']}
### Inflation Impact
- **Current CPI:** {macro_impacts['inflation_impact']['current_inflation']}%
- **Annual Cost Increase:** £{macro_impacts['inflation_impact']['real_cost_increase_annual']*0.79:.0f}
- **Pricing Power:** {macro_impacts['inflation_impact']['pricing_power']}
- **Wage Pressure:** {macro_impacts['inflation_impact']['wage_pressure']}
### Labour Market Conditions
- **UK Unemployment:** {macro_impacts['labour_market_impact']['unemployment_rate']}%
- **Labour Cost Index:** {macro_impacts['labour_market_impact']['labour_cost_index']:.0f}
- **Talent Availability:** {macro_impacts['labour_market_impact']['talent_availability']}
- **Wage Growth Pressure:** {macro_impacts['labour_market_impact']['wage_growth_pressure']}
### Regional Factors ({startup_data['location']})
- **Regional Growth:** {macro_impacts['regional_factors']['regional_growth']}%
- **Cost Index:** {macro_impacts['regional_factors']['cost_index']} (100 = UK average)
- **Talent Pool:** {macro_impacts['regional_factors']['talent_pool']}
- **Competitiveness:** {macro_impacts['regional_factors']['competitiveness']}
### Sector Outlook ({startup_data['sector']})
- **Expected Growth:** {macro_impacts['sector_outlook']['growth']}%
- **Regulatory Burden:** {macro_impacts['sector_outlook']['regulation']}
- **Key Opportunity:** {macro_impacts['sector_outlook']['opportunity']}
### Strategic Recommendations:
1. {'Consider debt financing while rates stabilize' if macro_impacts['interest_rate_impact']['cost_of_capital'] < 10 else 'Focus on equity financing'}
2. {'Build inflation adjustments into contracts' if macro_impacts['inflation_impact']['current_inflation'] > 2 else 'Lock in current pricing'}
3. {'Invest in talent retention' if macro_impacts['labour_market_impact']['talent_availability'] == 'Tight' else 'Opportunity to hire quality talent'}
"""
text_msg = await cl.Message(content=content).send()
# Send chart
image = cl.Image(content=chart, name="uk_economic_dashboard.png", display="inline")
await image.send(for_id=text_msg.id)
return
# =============================
# ROUTE: INTERACTIVE DASHBOARD
# =============================
if "dashboard" in user_input or "interactive dashboard" in user_input:
await thinking_msg.remove()
dashboard = InteractiveDashboard(df)
await cl.Message(content="## 📊 Interactive Dashboard Mode\n\nGenerating real-time analytics dashboard with interactive features...").send()
# Create executive summary
summary_text = create_dashboard_summary(dashboard)
summary_msg = await cl.Message(content=summary_text).send()
# Generate and send real-time metrics dashboard
dashboard_chart = dashboard.create_real_time_metrics_dashboard()
dashboard_image = cl.Image(content=dashboard_chart, name="interactive_dashboard.png", display="inline")
await dashboard_image.send(for_id=summary_msg.id)
await cl.Message(
content="## 🎯 Dashboard Commands Available:\n\n"
"• **'interactive scatter'** - Dynamic scatter plot with hover details\n"
"• **'correlation heatmap'** - Multi-dimensional relationship analysis\n"
"• **'filter dashboard'** - Apply filters (sector, country, funding)\n"
"• **'compare startups'** - Side-by-side startup analysis\n"
"• **'export dashboard'** - Generate professional PDF report\n\n"
"🔍 **Pro Tip:** Use filters to drill down into specific segments!"
).send()
return
# =============================
# ROUTE: INTERACTIVE SCATTER PLOT
# =============================
if "interactive scatter" in user_input:
await thinking_msg.remove()
dashboard = InteractiveDashboard(df)
await cl.Message(content="## 📊 Interactive Scatter Plot Analysis\n\nGenerating dynamic visualization with hover details and selection capabilities...").send()
# Get axis preferences from user input
x_axis = 'Funding_USD_M'
y_axis = 'Burn_Rate_Months'
if 'funding' in user_input and 'experience' in user_input:
x_axis, y_axis = 'Funding_USD_M', 'Founders_Experience_Yrs'
elif 'market' in user_input and 'funding' in user_input:
x_axis, y_axis = 'Market_Size_Bn', 'Funding_USD_M'
elif 'mrr' in user_input and 'growth' in user_input:
x_axis, y_axis = 'MRR_K', 'Monthly_Growth_Rate'
# Generate interactive scatter plot
scatter_chart = dashboard.create_interactive_scatter(x_axis, y_axis)
content = f"""
## 📊 Interactive Scatter Analysis: {x_axis.replace('_', ' ')} vs {y_axis.replace('_', ' ')}
### 🎯 **Key Insights:**
- **Green dots** = Successful startups
- **Red dots** = Failed startups
- **Dot size** = Market size (larger = bigger market)
### 🔍 **Interactive Features:**
- **Hover** over points for detailed information
- **Click** points to select for comparison
- **Drag** to zoom into specific areas
- **Double-click** to reset zoom
### 📈 **Analysis Options:**
Try these variations:
• "interactive scatter funding vs experience"
• "interactive scatter market vs funding"
• "interactive scatter mrr vs growth"
"""
text_msg = await cl.Message(content=content).send()
scatter_image = cl.Image(content=scatter_chart, name="interactive_scatter.png", display="inline")
await scatter_image.send(for_id=text_msg.id)
return
# =============================
# ROUTE: CORRELATION HEATMAP
# =============================
if "correlation heatmap" in user_input or "heatmap" in user_input:
await thinking_msg.remove()
dashboard = InteractiveDashboard(df)
await cl.Message(content="## 📊 Multi-Dimensional Correlation Analysis\n\nGenerating interactive correlation heatmap...").send()
# Generate correlation heatmap
heatmap_chart = dashboard.create_multi_dimensional_heatmap()
content = """
## 📊 Interactive Correlation Heatmap
### 🎯 **How to Read:**
- **Blue** = Positive correlation (variables move together)
- **Red** = Negative correlation (variables move opposite)
- **White** = No correlation
- **Numbers** = Correlation strength (-1 to +1)
### 🔍 **Key Relationships to Explore:**
- Funding vs Market Size
- Experience vs Success Rate
- MRR vs Growth Rate
- Competition vs Moat Strength
### 📈 **Insights:**
Strong correlations (>0.5 or <-0.5) indicate important relationships for investment decisions.
"""
text_msg = await cl.Message(content=content).send()
heatmap_image = cl.Image(content=heatmap_chart, name="correlation_heatmap.png", display="inline")
await heatmap_image.send(for_id=text_msg.id)
return
# =============================
# ROUTE: STARTUP COMPARISON
# =============================
if "compare startups" in user_input or "startup comparison" in user_input:
await thinking_msg.remove()
dashboard = InteractiveDashboard(df)
await cl.Message(content="## 🔍 Startup Comparison Analysis\n\nSelect startups to compare side-by-side...").send()
# Extract startup names from user input or ask user
startup_names = []
for startup in df['Startup'].values:
if startup.lower() in user_input.lower():
startup_names.append(startup)
# If no startups found in input, ask user to specify
if len(startup_names) < 2:
available_startups = ", ".join(df['Startup'].head(10).values)
await cl.Message(
content=f"Please specify 2-4 startup names to compare.\n\n"
f"**Available startups:** {available_startups}...\n\n"
f"**Example:** \"Compare TechFlow and DataCorp and AIStart\""
).send()
return
# Generate comparison radar chart
comparison_chart = dashboard.compare_startups(startup_names[:4]) # Limit to 4 for readability
if comparison_chart is None:
await cl.Message(content="❌ No valid startups found for comparison. Please check the names.").send()
return
content = f"""
## 🔍 Startup Comparison: {', '.join(startup_names[:4])}
### 📊 **Radar Chart Analysis:**
- **Green lines** = Successful startups
- **Red lines** = Failed startups
- **Outer edge** = Better performance (scale 0-5)
### 🎯 **Comparison Dimensions:**
- **Funding** (scaled to 0-5)
- **Founder Experience** (years)
- **Market Size** (billions)
- **Business Model Strength** (1-5)
- **Competitive Moat** (1-5)
- **MRR** (scaled to 0-5)
- **Growth Rate** (scaled to 0-5)
### 💡 **Investment Insights:**
Look for startups with larger radar areas and balanced performance across dimensions.
"""
text_msg = await cl.Message(content=content).send()
comparison_image = cl.Image(content=comparison_chart, name="startup_comparison.png", display="inline")
await comparison_image.send(for_id=text_msg.id)
return
# =============================
# ROUTE: DASHBOARD FILTERING
# =============================
if "filter dashboard" in user_input or "apply filters" in user_input:
await thinking_msg.remove()
await cl.Message(content="## 🔍 Dashboard Filtering Options\n\nApply filters to focus your analysis...").send()
# Parse filters from user input
filters = {}
# Sector filtering
if 'fintech' in user_input.lower():
filters['sectors'] = ['FinTech']
elif 'healthtech' in user_input.lower():
filters['sectors'] = ['HealthTech']
elif 'ai' in user_input.lower() or 'artificial intelligence' in user_input.lower():
filters['sectors'] = ['AI']
# Country filtering
if 'uk' in user_input.lower() or 'united kingdom' in user_input.lower():
filters['countries'] = ['UK']
elif 'us' in user_input.lower() or 'usa' in user_input.lower():
filters['countries'] = ['US']
# Success filtering
if 'successful' in user_input.lower() or 'success only' in user_input.lower():
filters['success_only'] = True
# Apply filters and generate filtered dashboard
dashboard = InteractiveDashboard(df)
if filters:
dashboard.filter_data(filters)
# Generate filtered dashboard
filtered_summary = create_dashboard_summary(dashboard)
filtered_chart = dashboard.create_real_time_metrics_dashboard()
filter_description = ""
if 'sectors' in filters:
filter_description += f"**Sectors:** {', '.join(filters['sectors'])}\n"
if 'countries' in filters:
filter_description += f"**Countries:** {', '.join(filters['countries'])}\n"
if 'success_only' in filters:
filter_description += "**Filter:** Successful startups only\n"
content = f"""
## 🔍 Filtered Dashboard Analysis
### 📊 **Applied Filters:**
{filter_description if filter_description else "**No specific filters detected.** Try: 'filter dashboard fintech uk successful'"}
{filtered_summary}
### 🎯 **Available Filter Commands:**
• "filter dashboard fintech" - FinTech startups only
• "filter dashboard uk successful" - Successful UK startups
• "filter dashboard healthtech us" - US HealthTech companies
• "filter dashboard ai" - AI/ML startups
"""
text_msg = await cl.Message(content=content).send()
filtered_image = cl.Image(content=filtered_chart, name="filtered_dashboard.png", display="inline")
await filtered_image.send(for_id=text_msg.id)
return
# =============================
# ROUTE: EXPORT DASHBOARD
# =============================
if "export dashboard" in user_input or "dashboard report" in user_input:
await thinking_msg.remove()
await cl.Message(content="## 📄 Exporting Interactive Dashboard Report\n\nGenerating comprehensive PDF with all dashboard analytics...").send()
dashboard = InteractiveDashboard(df)
# Create a comprehensive dashboard export
export_content = f"""
# 📊 NAVADA Interactive Dashboard Report
## Executive Summary
{create_dashboard_summary(dashboard)}
## 📈 Dashboard Analytics
### Key Insights:
- **Real-time Metrics:** Multi-dimensional analysis across 4 key areas
- **Interactive Features:** Hover details, drill-down capabilities, filtering
- **Comparison Tools:** Side-by-side startup analysis with radar charts
- **Correlation Analysis:** Relationship mapping between key variables
### Available Commands:
1. **dashboard** - Launch main interactive dashboard
2. **interactive scatter** - Dynamic scatter plots with selection
3. **correlation heatmap** - Multi-dimensional correlation matrix
4. **compare startups** - Radar chart comparisons
5. **filter dashboard [criteria]** - Apply smart filters
### Professional Features:
- ✅ Real-time data visualization
- ✅ Interactive hover details
- ✅ Custom filtering and drill-down
- ✅ Multi-startup comparisons
- ✅ Export capabilities
- ✅ Mobile-responsive design
---
**Generated by NAVADA Interactive Dashboard Suite**
*Next-generation startup analytics with real-time intelligence*
"""
# Send the export summary
await cl.Message(content=export_content).send()
await cl.Message(
content="## 🎯 Dashboard Export Complete!\n\n"
"**What's Included:**\n"
"• Executive summary with key metrics\n"
"• Interactive feature documentation\n"
"• Command reference guide\n"
"• Professional formatting\n\n"
"**Next Steps:**\n"
"• Use 'dashboard' to launch interactive mode\n"
"• Try 'interactive scatter' for dynamic analysis\n"
"• Explore 'compare startups [names]' for detailed comparisons"
).send()
return
# =============================
# ROUTE 8: GENERATE DOCUMENT TEMPLATES
# =============================
if any(template in user_input for template in
["business case", "pitch deck", "financial model", "investor memo",
"executive summary", "term sheet", "generate template", "create template"]):
# Initialize document generator
doc_generator = DocumentTemplateGenerator()
# Determine template type
template_type = None
if "business case" in user_input:
template_type = "business_case"
elif "pitch deck" in user_input:
template_type = "pitch_deck"
elif "financial model" in user_input:
template_type = "financial_model"
elif "investor memo" in user_input:
template_type = "investor_memo"
elif "executive summary" in user_input:
template_type = "executive_summary"
elif "term sheet" in user_input:
template_type = "term_sheet"
else:
# Show available templates
available = doc_generator.list_available_templates()
template_list = "\n".join([f"• **{t.replace('_', ' ').title()}**" for t in available])
await cl.Message(
content=f"📄 **Available Document Templates:**\n\n{template_list}\n\n"
f"Please specify which template you'd like to generate.\n"
f"Example: 'Generate business case template'"
).send()
return
# Start interactive questionnaire to gather personalized data
startup_data = await collect_startup_information(template_type, message)
if not startup_data:
return # User cancelled or error occurred
# Check if we have scraped or analyzed data to customize
if cl.user_session.get("last_analysis"):
analysis_data = cl.user_session.get("last_analysis")
if isinstance(analysis_data, dict):
startup_data.update(analysis_data)
# Generate loading message
gen_msg = cl.Message(
content=f"📝 Generating **{template_type.replace('_', ' ').title()}** template..."
)
await gen_msg.send()
try:
# Generate the document
pdf_bytes = doc_generator.generate_document(template_type, startup_data)
# Remove loading message
await gen_msg.remove()
# Send success message with details
await cl.Message(
content=f"✅ **{template_type.replace('_', ' ').title()}** Generated!\n\n"
f"This template includes:\n"
f"• Professional formatting\n"
f"• Industry-standard sections\n"
f"• Customizable content areas\n"
f"• Data visualization elements\n\n"
f"💡 **Pro Tip:** Click on the document below to view it in the split-screen viewer, "
f"where you can edit and add comments!"
).send()
# Send the PDF file with personalized filename
safe_company_name = startup_data.get('company_name', 'Your_Startup').replace(' ', '_')
filename = f"{safe_company_name}_{template_type}_{dt.datetime.now().strftime('%Y%m%d_%H%M%S')}.pdf"
pdf_element = cl.File(
name=filename,
content=pdf_bytes,
display="inline"
)
# Create a message to attach the file to
file_message = cl.Message(
content=f"📄 **Your Personalized {template_type.replace('_', ' ').title()} is Ready!**\n\n"
f"**Company:** {startup_data.get('company_name', 'Your Startup')}\n"
f"**Industry:** {startup_data.get('industry', 'Technology')}\n"
f"**Filename:** {filename}\n\n"
f"👁️ **Click the PDF below to open it in the Document Viewer**"
)
await file_message.send()
await pdf_element.send(for_id=file_message.id)
# Add additional guidance message about the viewer
await cl.Message(
content="📋 **Document Viewer Tips:**\n\n"
"• **Red 'View' button** should appear on the PDF above\n"
"• **Click the PDF or button** to open the split-screen viewer\n"
"• **Black panel** will appear on the right with editing features\n"
"• **Multiple tabs** supported for viewing several documents\n\n"
"If you don't see the viewer button, refresh the page and try again."
).send()
except Exception as e:
await gen_msg.remove()
await cl.Message(
content=f"❌ Error generating template: {str(e)}\n\n"
f"Please try again or contact support."
).send()
return
# =============================
# ROUTE 8.1: REAL-TIME FINANCIAL DATA & COMPETITOR ANALYSIS
# =============================
if any(keyword in user_input for keyword in
["competitor analysis", "market data", "financial data", "stock analysis",
"market sentiment", "funding trends", "economic indicators", "stock data",
"stock price", "stock info", "company stock", "compare stocks"]):
if not financial_integrator:
await cl.Message(content="❌ Financial data integration not available").send()
return
# Extract sector from user input
sector = 'technology' # Default
sectors = ['technology', 'fintech', 'healthcare', 'ecommerce', 'saas', 'biotech', 'cybersecurity', 'ai_ml']
for s in sectors:
if s.replace('_', ' ') in user_input.lower() or s in user_input.lower():
sector = s
break
loading_msg = cl.Message(content="🔄 Fetching real-time market data and competitor analysis...")
await loading_msg.send()
try:
if "competitor analysis" in user_input:
# Get comprehensive competitor analysis
analysis = financial_integrator.get_competitor_analysis(sector)
if 'error' not in analysis:
competitor_count = analysis.get('competitor_count', 0)
sector_averages = analysis.get('sector_averages', {})
content = f"## 📊 **{sector.title()} Sector Competitor Analysis**\n\n"
content += f"**Analysis Summary:**\n"
content += f"• Analyzed {competitor_count} public companies\n"
content += f"• Sector: {analysis.get('sector', sector).title()}\n\n"
if sector_averages:
content += "**Key Sector Metrics:**\n"
for metric, data in sector_averages.items():
if isinstance(data, dict) and 'median' in data:
content += f"• {metric.replace('_', ' ').title()}: {data['median']:.2f} (median)\n"
if analysis.get('market_insights'):
content += "\n**Market Insights:**\n"
for insight in analysis['market_insights'][:3]:
content += f"• {insight}\n"
if analysis.get('valuation_multiples'):
vm = analysis['valuation_multiples']
content += "\n**Valuation Benchmarks:**\n"
if vm.get('price_to_sales'):
ps = vm['price_to_sales']
content += f"• Price-to-Sales: {ps.get('median', 0):.1f}x median ({ps.get('range', 'N/A')})\n"
await loading_msg.remove()
await cl.Message(content=content).send()
else:
await loading_msg.remove()
await cl.Message(content=f"❌ Error: {analysis.get('error', 'Unknown error')}").send()
elif "market sentiment" in user_input:
# Get market sentiment analysis
sentiment = financial_integrator.get_market_sentiment(sector)
content = f"## 📰 **Market Sentiment Analysis - {sector.title()}**\n\n"
if 'error' not in sentiment:
overall = sentiment.get('overall_sentiment', 'neutral')
confidence = sentiment.get('confidence', 0)
content += f"**Overall Sentiment:** {overall.title()} (Confidence: {confidence:.0%})\n\n"
if sentiment.get('trending_topics'):
content += "**Trending Topics:**\n"
for topic in sentiment['trending_topics'][:4]:
content += f"• {topic['topic']}: {topic['sentiment'].replace('_', ' ').title()} ({topic['mentions']} mentions)\n"
await loading_msg.remove()
await cl.Message(content=content).send()
elif any(keyword in user_input for keyword in ["stock data", "stock price", "stock info", "company stock"]):
# Handle individual stock data requests
# Extract company name or symbol from user input
stock_symbol = None
# Look for company names or symbols in the input
words = user_input.lower().split()
for i, word in enumerate(words):
if word in ["stock", "price", "data", "info"] and i + 1 < len(words):
potential_symbol = words[i + 1]
if len(potential_symbol) <= 10: # Reasonable symbol/company name length
stock_symbol = potential_symbol
break
# Check for common company names in the input
if not stock_symbol:
for company, symbol in financial_integrator.common_symbols.items():
if company in user_input.lower():
stock_symbol = symbol
break
if not stock_symbol:
# Ask user for the stock symbol
await cl.Message(
content="📊 **Stock Data Analysis**\n\n"
"Please specify which company stock you'd like to analyze.\n\n"
"**Examples:**\n"
"• 'Apple stock data'\n"
"• 'TSLA stock price'\n"
"• 'Microsoft stock info'\n"
"• 'GOOGL stock analysis'\n\n"
"**Supported companies include:** Apple, Microsoft, Google, Amazon, Tesla, Meta, Netflix, Nvidia, and many more!"
).send()
return
# Get stock data
stock_data = financial_integrator.get_stock_data(stock_symbol)
if 'error' in stock_data:
await loading_msg.remove()
await cl.Message(content=f"❌ {stock_data['error']}").send()
return
# Format stock data response
company_name = stock_data.get('company_name', stock_symbol)
symbol = stock_data.get('symbol', stock_symbol)
current_data = stock_data.get('current_data', {})
technical = stock_data.get('technical_analysis', {})
fundamentals = stock_data.get('fundamental_metrics', {})
historical = stock_data.get('historical_data', {})
content = f"## 📈 **{company_name} ({symbol}) Stock Analysis**\n\n"
# Current Price & Performance
price = current_data.get('price', 'N/A')
change = current_data.get('change', 0)
change_pct = current_data.get('change_percent', 0)
volume = current_data.get('volume', 'N/A')
market_cap = current_data.get('market_cap')
change_indicator = "📈" if change > 0 else "📉" if change < 0 else "➡️"
content += f"**Current Price & Performance:**\n"
content += f"• Price: ${price}\n"
content += f"• Change: ${change:+.2f} ({change_pct:+.2f}%) {change_indicator}\n"
content += f"• Volume: {volume:,}\n"
if market_cap:
content += f"• Market Cap: ${market_cap:,.0f}\n"
content += f"• Last Updated: {current_data.get('last_updated', 'N/A')}\n\n"
# Technical Analysis
content += "**Technical Analysis:**\n"
rsi = technical.get('rsi')
volatility = technical.get('volatility_annual')
trend = technical.get('trend', 'Unknown')
ma_50 = technical.get('ma_50')
ma_200 = technical.get('ma_200')
if rsi:
content += f"• RSI (14-day): {rsi:.1f}\n"
content += f"• Trend: {trend}\n"
if ma_50:
content += f"• 50-day MA: ${ma_50:.2f}\n"
if ma_200:
content += f"• 200-day MA: ${ma_200:.2f}\n"
if volatility:
content += f"• Annual Volatility: {volatility:.1f}%\n"
content += "\n"
# Fundamental Metrics
content += "**Fundamental Metrics:**\n"
pe_ratio = fundamentals.get('pe_ratio')
pb_ratio = fundamentals.get('pb_ratio')
roe = fundamentals.get('roe')
revenue_growth = fundamentals.get('revenue_growth')
profit_margins = fundamentals.get('profit_margins')
if pe_ratio:
content += f"• P/E Ratio: {pe_ratio:.2f}\n"
if pb_ratio:
content += f"• P/B Ratio: {pb_ratio:.2f}\n"
if roe:
content += f"• ROE: {roe:.1f}%\n"
if revenue_growth:
content += f"• Revenue Growth: {revenue_growth:.1f}%\n"
if profit_margins:
content += f"• Profit Margin: {profit_margins:.1f}%\n"
content += "\n"
# 52-Week Range
week_high = historical.get('52_week_high')
week_low = historical.get('52_week_low')
if week_high and week_low:
content += f"**52-Week Range:** ${week_low:.2f} - ${week_high:.2f}\n\n"
# Investment Analysis
analysis = stock_data.get('investment_analysis', 'Analysis unavailable')
content += f"**Investment Analysis:**\n{analysis}\n\n"
content += f"**Company Info:**\n"
content += f"• Sector: {stock_data.get('sector', 'Unknown')}\n"
content += f"• Industry: {stock_data.get('industry', 'Unknown')}\n\n"
content += "💡 **Try these commands:**\n"
content += "• 'Compare AAPL MSFT GOOGL' - Compare multiple stocks\n"
content += "• 'Apple stock 5 year data' - Get longer historical data\n"
content += "• 'Tesla technical analysis' - Focus on technical indicators"
await loading_msg.remove()
await cl.Message(content=content).send()
elif "compare stocks" in user_input:
# Handle stock comparison
# Extract stock symbols from input
symbols = []
words = user_input.upper().split()
# Look for ticker symbols (3-5 character uppercase words)
for word in words:
if 2 <= len(word) <= 5 and word.isalpha() and word not in ['COMPARE', 'STOCKS', 'STOCK', 'VS']:
symbols.append(word)
# Also check for company names
for company, symbol in financial_integrator.common_symbols.items():
if company in user_input.lower():
if symbol not in symbols:
symbols.append(symbol)
if len(symbols) < 2:
await cl.Message(
content="📊 **Stock Comparison**\n\n"
"Please specify 2-5 stocks to compare.\n\n"
"**Examples:**\n"
"• 'Compare AAPL MSFT GOOGL'\n"
"• 'Compare Apple Microsoft Google'\n"
"• 'Compare TSLA NIO RIVN' (EV stocks)\n"
"• 'Compare CRM NOW TEAM' (SaaS stocks)"
).send()
return
# Get comparison data
comparison = financial_integrator.compare_stocks(symbols[:5]) # Limit to 5 stocks
if 'error' in comparison:
await loading_msg.remove()
await cl.Message(content=f"❌ {comparison['error']}").send()
return
# Format comparison response
symbols_analyzed = comparison.get('symbols_analyzed', [])
summary = comparison.get('comparison_summary', {})
content = f"## 📊 **Stock Comparison Analysis**\n\n"
content += f"**Stocks Analyzed:** {', '.join(symbols_analyzed)}\n\n"
# Performance Summary
best_performer = summary.get('best_performer')
worst_performer = summary.get('worst_performer')
avg_performance = summary.get('avg_performance')
avg_volatility = summary.get('avg_volatility')
content += "**Performance Summary:**\n"
if best_performer:
content += f"• Best Performer: {best_performer[0]} ({best_performer[1]:+.2f}%)\n"
if worst_performer:
content += f"• Worst Performer: {worst_performer[0]} ({worst_performer[1]:+.2f}%)\n"
if avg_performance:
content += f"• Average Performance: {avg_performance:+.2f}%\n"
if avg_volatility:
content += f"• Average Volatility: {avg_volatility:.1f}%\n"
content += "\n"
# Detailed comparison table
detailed_data = comparison.get('detailed_data', {})
content += "**Detailed Comparison:**\n\n"
content += "| Symbol | Price | Change % | P/E | Trend |\n"
content += "|--------|--------|----------|-----|-------|\n"
for symbol, data in detailed_data.items():
price = data.get('current_data', {}).get('price', 'N/A')
change_pct = data.get('current_data', {}).get('change_percent', 0)
pe_ratio = data.get('fundamental_metrics', {}).get('pe_ratio', 'N/A')
trend = data.get('technical_analysis', {}).get('trend', 'Unknown')
content += f"| {symbol} | ${price} | {change_pct:+.1f}% | {pe_ratio} | {trend} |\n"
content += f"\n**Analysis Date:** {comparison.get('analysis_date', 'N/A')}\n\n"
content += "💡 **Next Steps:** Type the individual stock symbol for detailed analysis"
await loading_msg.remove()
await cl.Message(content=content).send()
elif "economic indicators" in user_input:
# Get economic indicators
indicators = financial_integrator.get_economic_indicators()
content = "## 📈 **Economic Indicators Dashboard**\n\n"
if 'error' not in indicators:
summary = indicators.get('summary', 'Economic conditions analysis')
content += f"**Summary:** {summary}\n\n"
if indicators.get('indicators'):
content += "**Key Indicators:**\n"
for key, data in list(indicators['indicators'].items())[:5]:
if isinstance(data, dict):
name = data.get('name', key)
value = data.get('current_value', 'N/A')
change = data.get('monthly_change', 0)
trend = "📈" if change > 0 else "📉" if change < 0 else "➡️"
content += f"• {name}: {value} {trend} ({change:+.1f}%)\n"
if indicators.get('startup_impact'):
content += "\n**Startup Impact:**\n"
for impact in indicators['startup_impact'][:3]:
content += f"• {impact}\n"
await loading_msg.remove()
await cl.Message(content=content).send()
except Exception as e:
await loading_msg.remove()
await cl.Message(content=f"❌ Error fetching financial data: {str(e)}").send()
return
# =============================
# ROUTE 8.2: ADVANCED ANALYTICS & PREDICTIVE MODELING
# =============================
if any(keyword in user_input for keyword in
["advanced analytics", "predictive model", "success probability",
"interactive dashboard", "cohort analysis", "ab test"]):
if not analytics_dashboard:
await cl.Message(content="❌ Advanced analytics not available").send()
return
if "predictive model" in user_input or "success probability" in user_input:
loading_msg = cl.Message(content="🤖 Training predictive model...")
await loading_msg.send()
try:
# Train model on current dataset
model_results = analytics_dashboard.train_success_prediction_model(df)
if 'error' not in model_results:
best_model = model_results.get('best_model', 'Unknown')
accuracy = model_results.get('best_accuracy', 0)
content = "## 🤖 **Startup Success Prediction Model**\n\n"
content += f"**Model Performance:**\n"
content += f"• Best Model: {best_model}\n"
content += f"• Accuracy: {accuracy:.1%}\n"
content += f"• Training Samples: {model_results.get('training_samples', 0)}\n\n"
if model_results.get('feature_importance'):
content += "**Most Important Factors:**\n"
for feature, importance in model_results['feature_importance'][:5]:
content += f"• {feature.replace('_', ' ').title()}: {importance:.3f}\n"
content += "\n💡 **How to use:** Type 'predict my startup success' with your startup details"
await loading_msg.remove()
await cl.Message(content=content).send()
else:
await loading_msg.remove()
await cl.Message(content=f"❌ Model training failed: {model_results.get('error')}").send()
except Exception as e:
await loading_msg.remove()
await cl.Message(content=f"❌ Error training model: {str(e)}").send()
elif "interactive dashboard" in user_input or "advanced analytics" in user_input:
loading_msg = cl.Message(content="📊 Creating advanced analytics dashboard...")
await loading_msg.send()
try:
dashboard_html = analytics_dashboard.create_interactive_exploration_dashboard(df)
await loading_msg.remove()
await cl.Message(
content="## 🔍 **Advanced Interactive Analytics Dashboard**\n\n"
"**Features:**\n"
"• Click on charts to drill down into details\n"
"• Interactive filters and exploration\n"
"• Real-time data correlation analysis\n"
"• Multi-dimensional visualizations\n\n"
"**📊 Dashboard ready below** ⬇️"
).send()
# Save dashboard
dashboard_file = cl.File(
path="advanced_analytics_dashboard.html",
name="Advanced Analytics Dashboard",
display="inline"
)
with open("advanced_analytics_dashboard.html", "w", encoding='utf-8') as f:
f.write(dashboard_html)
await dashboard_file.send(for_id=loading_msg.id)
except Exception as e:
await loading_msg.remove()
await cl.Message(content=f"❌ Error creating dashboard: {str(e)}").send()
elif "cohort analysis" in user_input:
loading_msg = cl.Message(content="📈 Generating cohort analysis...")
await loading_msg.send()
try:
cohort_html = analytics_dashboard.create_cohort_analysis(df)
await loading_msg.remove()
await cl.Message(
content="## 📈 **Cohort Analysis Dashboard**\n\n"
"Track startup performance across different founding years\n"
"and identify trends in success rates over time."
).send()
# Save cohort analysis
with open("cohort_analysis.html", "w", encoding='utf-8') as f:
f.write(cohort_html)
cohort_file = cl.File(
path="cohort_analysis.html",
name="Cohort Analysis",
display="inline"
)
await cohort_file.send(for_id=loading_msg.id)
except Exception as e:
await loading_msg.remove()
await cl.Message(content=f"❌ Error creating cohort analysis: {str(e)}").send()
return
# =============================
# ROUTE 8.3: ORIGINAL PDF REPORT
# =============================
if "generate report" in user_input or "create report" in user_input or "investment report" in user_input:
# Show loading message
msg = cl.Message(content="📄 Generating comprehensive PDF report... This may take 10-15 seconds.")
await msg.send()
# Check if user specified a startup name
startup_name = None
for startup in df['Startup'].values:
if startup.lower() in user_input:
startup_name = startup
break
try:
# Generate PDF report
pdf_bytes = generate_investment_report(df, startup_name)
# Remove loading message
await msg.remove()
# Send description message
if startup_name:
desc_msg = cl.Message(
content=f"### 📊 Investment Analysis Report: {startup_name}\n\n"
f"**Report Contents:**\n"
f"- Executive Summary\n"
f"- Key Metrics & Financials\n"
f"- 4 Data Visualizations\n"
f"- Risk Analysis\n"
f"- Investment Recommendations\n\n"
f"**Viability score and actionable insights included.**"
)
else:
desc_msg = cl.Message(
content=f"### 📊 Portfolio Analysis Report\n\n"
f"**Report Contents:**\n"
f"- Portfolio Overview ({len(df)} startups)\n"
f"- Success Rate Analysis\n"
f"- 4 Data Visualizations\n"
f"- Risk Assessment\n"
f"- Strategic Recommendations\n\n"
f"**Download the PDF below for the complete analysis.**"
)
await desc_msg.send()
# Send PDF file as downloadable attachment
report_filename = f"{startup_name}_Analysis.pdf" if startup_name else "Portfolio_Analysis.pdf"
pdf_element = cl.File(
name=report_filename,
content=pdf_bytes,
display="inline"
)
await pdf_element.send(for_id=desc_msg.id)
except Exception as e:
await msg.remove()
await cl.Message(
content=f"❌ Error generating report: {str(e)}\n\n"
f"Please try again or contact support if the issue persists."
).send()
return # Exit handler
# =============================
# ROUTE 7: INTELLIGENT CHART DETECTION
# =============================
# Check if user is asking for a chart/graph/visualization
chart_keywords = ["chart", "graph", "plot", "visualize", "visualization", "show me", "display"]
is_chart_request = any(keyword in user_input for keyword in chart_keywords)
if is_chart_request:
# -------------------------
# DETECT CHART INTENT WITH AI
# -------------------------
df_str = df.to_string(index=False)
available_columns = list(df.columns)
# Get session ID for LangSmith tracking
session_id = cl.user_session.get("session_id", get_session_id())
# Include LangSmith metadata if available
if langsmith_client and session_id:
intent_response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "system",
"content": (
"You are a data visualization assistant. Analyze the user's request and determine:\n"
"1. What type of chart they want (bar, scatter, line, pie, or 'analysis' for text response)\n"
"2. Which columns to use (x-axis and y-axis)\n"
"3. A descriptive title\n\n"
f"Available columns: {', '.join(available_columns)}\n\n"
"Respond ONLY in this JSON format:\n"
'{"chart_type": "bar/scatter/line/pie/analysis", "x_col": "column_name", "y_col": "column_name", "title": "Chart Title"}\n\n'
"If the request doesn't make sense for a chart, use chart_type: 'analysis'."
)
},
{
"role": "user",
"content": f"Dataset columns: {available_columns}\n\nUser request: {message.content}"
}
],
max_tokens=150,
)
else:
intent_response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "system",
"content": (
"You are a data visualization assistant. Analyze the user's request and determine:\n"
"1. What type of chart they want (bar, scatter, line, pie, or 'analysis' for text response)\n"
"2. Which columns to use (x-axis and y-axis)\n"
"3. A descriptive title\n\n"
f"Available columns: {', '.join(available_columns)}\n\n"
"Respond ONLY in this JSON format:\n"
'{"chart_type": "bar/scatter/line/pie/analysis", "x_col": "column_name", "y_col": "column_name", "title": "Chart Title"}\n\n'
"If the request doesn't make sense for a chart, use chart_type: 'analysis'."
)
},
{
"role": "user",
"content": f"Dataset columns: {available_columns}\n\nUser request: {message.content}"
}
],
max_tokens=150
)
try:
# Parse AI response to get chart parameters
import json
intent_text = intent_response.choices[0].message.content.strip()
# Extract JSON from response (handle markdown code blocks)
if "```" in intent_text:
intent_text = intent_text.split("```")[1]
if intent_text.startswith("json"):
intent_text = intent_text[4:]
intent = json.loads(intent_text.strip())
# If AI says to do analysis instead of chart, fall through to Q&A
if intent.get("chart_type") == "analysis":
raise ValueError("Analysis requested instead of chart")
# Generate the requested chart
msg = cl.Message(content=f"📊 Generating {intent['chart_type']} chart...")
await msg.send()
png = plot_custom_chart(
df,
intent.get("chart_type", "bar"),
intent.get("x_col", "Startup"),
intent.get("y_col", "Funding_USD_M"),
intent.get("title")
)
await msg.remove()
text_msg = cl.Message(content=f"### 📊 {intent.get('title', 'Custom Chart')}")
await text_msg.send()
image = cl.Image(content=png, name="custom_chart.png", display="inline")
await image.send(for_id=text_msg.id)
return
except Exception as e:
# If chart generation fails, fall through to regular Q&A
print(f"Chart generation failed: {e}")
pass
# =============================
# ROUTE 8: INTERACTIVE DASHBOARDS
# =============================
# Handle general "interactive" request - default to scatter plot dashboard
if user_input.strip() == "interactive" or (user_input.strip() == "proceed" and thinking_msg.content and "interactive" in thinking_msg.content):
msg = cl.Message(content="🎯 Creating interactive scatter plot dashboard...")
await msg.send()
html_content = create_interactive_scatter(df, "Interactive Startup Analysis")
await msg.remove()
# Send the interactive chart as HTML
await cl.Message(
content="## 🎯 Interactive Startup Dashboard\n\n"
"**Features:**\n"
"- 🖱️ Hover for detailed startup information\n"
"- 🔍 Zoom and pan to explore data\n"
"- 📊 Size indicates market size, color shows success/failure\n\n"
).send()
# Create an HTML file and send it
with open("interactive_dashboard.html", "w") as f:
f.write(html_content)
dashboard_file = cl.File(
path="interactive_dashboard.html",
name="Interactive Dashboard",
display="inline"
)
await dashboard_file.send()
return
if "interactive" in user_input and ("dashboard" in user_input or "scatter" in user_input):
msg = cl.Message(content="🎯 Creating interactive scatter plot dashboard...")
await msg.send()
html_content = create_interactive_scatter(df, "Interactive Startup Analysis")
await msg.remove()
# Send the interactive chart as HTML
await cl.Message(
content="## 🎯 Interactive Startup Dashboard\n\n"
"**Features:**\n"
"- 🖱️ Hover for detailed startup information\n"
"- 🔍 Zoom and pan to explore data\n"
"- 📊 Size indicates market size, color shows success/failure\n\n"
).send()
# Create an HTML file and send it
with open("interactive_dashboard.html", "w") as f:
f.write(html_content)
dashboard_file = cl.File(
path="interactive_dashboard.html",
name="Interactive Dashboard",
display="inline"
)
await dashboard_file.send()
return
if "interactive timeline" in user_input:
msg = cl.Message(content="📈 Creating interactive failure timeline...")
await msg.send()
html_content = create_interactive_timeline(df)
await msg.remove()
await cl.Message(
content="## 📈 Interactive Failure Timeline\n\n"
"**Features:**\n"
"- 🖱️ Hover to see startup details\n"
"- 📊 Interactive bars with sector and funding info\n"
"- 🔍 Zoom to focus on specific time ranges\n\n"
).send()
with open("interactive_timeline.html", "w") as f:
f.write(html_content)
timeline_file = cl.File(
path="interactive_timeline.html",
name="Interactive Timeline",
display="inline"
)
await timeline_file.send()
return
if "sector dashboard" in user_input or ("interactive" in user_input and "sector" in user_input):
msg = cl.Message(content="🏭 Creating interactive sector dashboard...")
await msg.send()
html_content = create_sector_dashboard(df)
await msg.remove()
await cl.Message(
content="## 🏭 Interactive Sector Dashboard\n\n"
"**Features:**\n"
"- 📊 Four interconnected charts\n"
"- 🖱️ Hover and zoom on each panel\n"
"- 💡 Compare sectors across multiple dimensions\n\n"
).send()
with open("sector_dashboard.html", "w") as f:
f.write(html_content)
sector_file = cl.File(
path="sector_dashboard.html",
name="Sector Dashboard",
display="inline"
)
await sector_file.send()
return
# =============================
# ROUTE 9: NEW ADVANCED CHARTS
# =============================
if "growth trajectory" in user_input or ("growth" in user_input and "chart" in user_input):
msg = cl.Message(content="📈 Generating growth trajectory analysis...")
await msg.send()
png = plot_growth_trajectory(df)
await msg.remove()
text_msg = cl.Message(
content="### 📈 Growth Trajectory Analysis\n\n"
"This chart shows **MRR growth vs company age** with bubble sizes representing growth rates.\n"
"- **Green dots**: Successful companies\n"
"- **Red dots**: Failed companies\n"
"- **Bubble size**: Monthly growth rate percentage"
)
await text_msg.send()
image = cl.Image(content=png, name="growth_trajectory.png", display="inline")
await image.send(for_id=text_msg.id)
return
if "team performance" in user_input or ("team" in user_input and "matrix" in user_input):
msg = cl.Message(content="👥 Generating team performance matrix...")
await msg.send()
png = plot_team_performance(df)
await msg.remove()
text_msg = cl.Message(
content="### 👥 Team Performance Matrix\n\n"
"This chart analyzes **team size vs founder experience** correlation.\n"
"- **Bubble size**: Total funding raised\n"
"- **Red trend line**: Shows correlation between team size and experience\n"
"- **Colors**: Green = successful, Red = failed"
)
await text_msg.send()
image = cl.Image(content=png, name="team_performance.png", display="inline")
await image.send(for_id=text_msg.id)
return
if "market opportunity" in user_input or ("market" in user_input and "competition" in user_input):
msg = cl.Message(content="🎯 Generating market opportunity matrix...")
await msg.send()
png = plot_market_opportunity(df)
await msg.remove()
text_msg = cl.Message(
content="### 🎯 Market Opportunity Matrix\n\n"
"This chart identifies **sweet spots** in market size vs competition landscape.\n"
"- **X-axis**: Market size (bigger = better)\n"
"- **Y-axis**: Market opportunity (higher = less competition)\n"
"- **Bubble size**: Current traction (MRR)\n"
"- **Sweet Spot**: Large market + low competition"
)
await text_msg.send()
image = cl.Image(content=png, name="market_opportunity.png", display="inline")
await image.send(for_id=text_msg.id)
return
if "funding efficiency" in user_input or ("capital" in user_input and "efficiency" in user_input):
msg = cl.Message(content="💰 Generating capital efficiency analysis...")
await msg.send()
png = plot_funding_efficiency(df)
await msg.remove()
text_msg = cl.Message(
content="### 💰 Capital Efficiency Analysis\n\n"
"This chart shows **revenue generated per dollar invested**.\n"
"- **X-axis**: Total funding raised\n"
"- **Y-axis**: Annual revenue per dollar of funding\n"
"- **Bubble size**: Efficiency score (revenue × growth rate)\n"
"- **Orange line**: Median efficiency benchmark"
)
await text_msg.send()
image = cl.Image(content=png, name="funding_efficiency.png", display="inline")
await image.send(for_id=text_msg.id)
return
if "stage progression" in user_input or ("funding" in user_input and "stage" in user_input):
msg = cl.Message(content="🚀 Generating funding stage analysis...")
await msg.send()
png = plot_stage_progression(df)
await msg.remove()
text_msg = cl.Message(
content="### 🚀 Funding Stage Analysis\n\n"
"This chart tracks **funding amounts and failure rates by stage**.\n"
"- **Blue bars**: Average funding amount per stage\n"
"- **Red line**: Failure rate percentage\n"
"- **Labels**: Show funding amount and company count\n"
"- **Insight**: Later stages typically have lower failure rates"
)
await text_msg.send()
image = cl.Image(content=png, name="stage_progression.png", display="inline")
await image.send(for_id=text_msg.id)
return
if "risk assessment" in user_input or ("risk" in user_input and "radar" in user_input):
msg = cl.Message(content="🎯 Generating risk assessment radar...")
await msg.send()
png = plot_risk_assessment(df)
await msg.remove()
text_msg = cl.Message(
content="### 🎯 Risk Assessment Profile\n\n"
"This **radar chart** compares risk profiles between successful and failed companies.\n"
"- **Green area**: Successful companies' average risk profile\n"
"- **Red area**: Failed companies' average risk profile\n"
"- **Scale**: 0 = low risk, 10 = high risk\n"
"- **Categories**: Financial, Market, Team, Competition, and Traction risks"
)
await text_msg.send()
image = cl.Image(content=png, name="risk_assessment.png", display="inline")
await image.send(for_id=text_msg.id)
return
# =============================
# ROUTE 10: GUIDED QUESTIONS
# =============================
if "questions" in user_input or "guide me" in user_input or "what should i ask" in user_input:
await thinking_msg.remove()
persona = get_current_persona()
questions = persona.get("questions", [])
content = f"### 🎯 {persona['name']} - Guided Questions\n\n"
content += "Here are key questions to explore based on your current mode:\n\n"
for i, question in enumerate(questions, 1):
content += f"**{i}.** {question}\n"
content += "\n📊 **Recommended Charts:**\n"
for chart in persona.get("charts", []):
chart_name = chart.replace("_", " ").title()
content += f"- Type **'{chart_name}'** for {chart_name} analysis\n"
content += "\n💡 **Pro Tip:** Copy any question above and I'll provide detailed analysis!"
await cl.Message(content=content).send()
return
# =============================
# ROUTE 11: INTERNET SEARCH
# =============================
if user_input.startswith("search ") or \
("search" in user_input and ("internet" in user_input or "web" in user_input)) or \
("search" in user_input and any(word in user_input for word in ["latest", "recent", "current", "news", "trends", "2024", "2025"])) or \
("what's happening" in user_input) or \
("latest news" in user_input) or \
("current trends" in user_input) or \
("find" in user_input and any(word in user_input for word in ["latest", "recent", "current", "new"])):
await thinking_msg.remove()
# Extract search query
if user_input.startswith("search "):
query = user_input[7:].strip()
elif "search" in user_input:
# Extract query after "search"
parts = user_input.split("search", 1)
if len(parts) > 1:
query = parts[1].strip()
else:
query = user_input
else:
# For phrases like "what's happening with AI startups"
query = user_input
if not query:
await cl.Message(
content="Please provide a search query. For example:\n"
"- **search AI startups 2024**\n"
"- **latest trends in fintech**\n"
"- **current venture capital news**"
).send()
return
msg = cl.Message(content=f"🔍 Searching for: **{query}**...")
await msg.send()
# Perform search
search_results = search_internet(query, count=5)
if not search_results["success"]:
# Provide fallback response with helpful guidance
fallback_message = f"❌ Search failed: {search_results['error']}\n\n"
if search_results.get("fallback_available"):
fallback_message += "💡 **Alternative approach**: I can still help you with:\n"
fallback_message += "• General startup advice and best practices\n"
fallback_message += "• Analysis based on my training data\n"
fallback_message += "• Startup failure pattern analysis\n"
fallback_message += "• Business model evaluation\n\n"
fallback_message += "🔧 **To enable web search**: Please configure a valid Brave Search API key in your .env file.\n"
fallback_message += "Visit https://brave.com/search/api/ to get an API key."
msg.content = fallback_message
await msg.update()
return
# Get current persona for analysis
persona = get_current_persona()
# Get session ID for LangSmith tracking
session_id = cl.user_session.get("session_id", get_session_id())
# Analyze results with persona context
analysis = analyze_search_results(search_results, persona, "startup and investment context", session_id)
# Format and send results
content = f"## 🔍 Search Results: {query}\n\n"
content += f"**Found {search_results['total_results']} results:**\n\n"
for i, result in enumerate(search_results['results'], 1):
content += f"**{i}. {result['title']}**\n"
content += f"🔗 {result['url']}\n"
content += f"📝 {result['description'][:200]}{'...' if len(result['description']) > 200 else ''}\n"
if result.get('age'):
content += f"⏰ {result['age']}\n"
content += "\n"
content += f"---\n\n## 🤖 {persona['name']} Analysis:\n\n{analysis}"
msg.content = content
await msg.update()
return
# =============================
# ROUTE 12: TEXT-TO-SPEECH
# =============================
if user_input.startswith("speak ") or user_input.startswith("say ") or \
("audio" in user_input and "response" in user_input) or \
("read aloud" in user_input) or ("voice" in user_input):
await thinking_msg.remove()
# Extract text to speak
if user_input.startswith("speak "):
text_to_speak = user_input[6:].strip()
elif user_input.startswith("say "):
text_to_speak = user_input[4:].strip()
else:
text_to_speak = "Welcome to NAVADA, your AI-powered startup viability agent. I can analyze startup risks, generate charts, and provide investment insights in both investor and founder modes."
if not text_to_speak:
await cl.Message(
content="Please provide text to convert to speech:\n"
"- **speak [your text]**\n"
"- **say [your text]**\n"
"- **read aloud [your text]**"
).send()
return
# Generate audio
msg = cl.Message(content=f"🔊 Generating speech: **{text_to_speak[:100]}{'...' if len(text_to_speak) > 100 else ''}**")
await msg.send()
try:
# Create audio message
audio = create_audio_message(text_to_speak, voice="alloy")
if audio:
# Send text message with audio
content = f"🔊 **Audio Response:**\n\n{text_to_speak}"
text_msg = cl.Message(content=content)
await text_msg.send()
# Send audio
await audio.send(for_id=text_msg.id)
else:
msg.content = "❌ Failed to generate audio. Please try again."
await msg.update()
except Exception as e:
msg.content = f"❌ Audio generation error: {str(e)}"
await msg.update()
return
# =============================
# ROUTE 13: PERSONA MANAGEMENT
# =============================
if "investor mode" in user_input or "switch to investor" in user_input:
# Remove thinking indicator
await thinking_msg.remove()
cl.user_session.set("persona", "investor")
persona = get_current_persona()
recommendations = format_persona_recommendations("investor")
await cl.Message(
content=f"{persona['style']}\n\n"
"I'm now analyzing from a **venture capitalist perspective**. "
"I'll focus on ROI, market size, competitive analysis, and exit strategies.\n\n"
f"{recommendations}"
"**What would you like to analyze today?**\n\n"
"💰 **Quick Analysis:**\n"
"• Type **'portfolio'** - Investment recommendations across all startups\n"
"• Type **'insights'** - AI-powered risk assessment and opportunities\n"
"• Type **'benchmark'** - Compare new startup ideas against our dataset\n"
"• Type **'questions'** - Get guided investor-focused questions\n\n"
"📊 **Advanced Charts:**\n"
"• **'Funding Efficiency'** - Capital efficiency and ROI analysis\n"
"• **'Stage Progression'** - Funding stages vs failure rates\n"
"• **'Market Opportunity'** - Market size vs competition matrix\n"
"• **'Risk Assessment'** - Comprehensive risk radar chart\n\n"
"📈 **Interactive Tools:**\n"
"• **'Sector Dashboard'** - Multi-dimensional sector analysis\n"
"• **'Interactive'** - Dynamic scatter plots and correlations\n\n"
"🔍 **Internet Search:**\n"
"• **'search latest VC trends'** - Get up-to-date market intelligence\n"
"• **'current startup news'** - Recent developments in startup ecosystem\n"
"• **'search [company name] funding'** - Research specific companies\n\n"
"🎯 **Ask me directly:**\n"
"• \"Which startups have the best ROI potential?\"\n"
"• \"What are the red flags in our portfolio?\"\n"
"• \"Search for latest AI startup trends\""
).send()
return
if "founder mode" in user_input or "switch to founder" in user_input:
# Remove thinking indicator
await thinking_msg.remove()
cl.user_session.set("persona", "founder")
persona = get_current_persona()
recommendations = format_persona_recommendations("founder")
await cl.Message(
content=f"{persona['style']}\n\n"
"I'm now analyzing from an **experienced founder perspective**. "
"I'll focus on practical execution, team building, product development, and tactical advice.\n\n"
f"{recommendations}"
"**What challenges can I help you tackle today?**\n\n"
"🚀 **Quick Assessment:**\n"
"• Type **'assess idea'** - Get viability score for your startup concept\n"
"• Type **'benchmark'** - Compare your metrics to successful startups\n"
"• Type **'insights'** - Get tactical recommendations to reduce risk\n"
"• Type **'questions'** - Get guided founder-focused questions\n\n"
"📊 **Growth Analysis:**\n"
"• **'Growth Trajectory'** - MRR growth patterns and success factors\n"
"• **'Team Performance'** - Team size vs experience optimization\n"
"• **'Market Opportunity'** - Find your competitive sweet spot\n"
"• **'Stage Progression'** - Funding stage benchmarks and expectations\n\n"
"📈 **Tactical Tools:**\n"
"• **'Timeline'** - Failure patterns to avoid common pitfalls\n"
"• **'Interactive'** - Explore data patterns affecting your sector\n"
"• **'Portfolio'** - Study successful companies in your space\n\n"
"🔍 **Market Intelligence:**\n"
"• **'search competitor analysis'** - Research competitive landscape\n"
"• **'latest startup challenges'** - Current industry challenges\n"
"• **'search [your sector] trends'** - Stay ahead of market shifts\n\n"
"💡 **Ask me directly:**\n"
"• \"How can I extend my runway and reduce burn?\"\n"
"• \"What team size is optimal for my stage?\"\n"
"• \"Search for current SaaS pricing trends\""
).send()
return
# =============================
# ROUTE: DISPLAY RECOMMENDATIONS
# =============================
if "recommendations" in user_input or "best practices" in user_input:
await thinking_msg.remove()
current_persona_name = cl.user_session.get("persona", "founder")
current_persona = PERSONAS[current_persona_name]
recommendations = format_persona_recommendations(current_persona_name)
await cl.Message(
content=f"{current_persona['style']}\n\n"
f"Here are the key recommendations for {current_persona['name']}:"
f"{recommendations}"
"💡 **Want more specific advice?** Ask me about any of these areas, or switch to a different mode:\n"
"• **'investor mode'** - VC investment criteria\n"
"• **'founder mode'** - Tactical execution advice\n"
"• **'economist mode'** - UK economic analysis\n"
"• **'company analyst mode'** - Financial performance focus"
).send()
return
# =============================
# ROUTE: USER DASHBOARD & HISTORY
# =============================
if "dashboard" in user_input or "my conversations" in user_input or "history" in user_input:
await thinking_msg.remove()
if not AUTH_AVAILABLE or not auth_status["authenticated"]:
await cl.Message(content="❌ Please log in to view your dashboard.").send()
return
# Get user's conversation history
conversations = auth_manager.get_user_conversations(auth_status["user_id"], limit=10)
if not conversations:
await cl.Message(
content=f"📊 **Welcome to your NAVADA Dashboard, {auth_status['username']}!**\n\n"
"🎯 **Account Info:**\n"
f"• Username: {auth_status['username']}\n"
f"• Email: {auth_status.get('email', 'Not provided')}\n"
f"• Subscription: {auth_status.get('subscription_tier', 'free').title()}\n\n"
"📝 **Conversation History:** No conversations yet\n\n"
"Start chatting to build your conversation history!"
).send()
else:
dashboard_content = f"📊 **Welcome to your NAVADA Dashboard, {auth_status['username']}!**\n\n"
dashboard_content += "🎯 **Account Info:**\n"
dashboard_content += f"• Username: {auth_status['username']}\n"
dashboard_content += f"• Email: {auth_status.get('email', 'Not provided')}\n"
dashboard_content += f"• Subscription: {auth_status.get('subscription_tier', 'free').title()}\n\n"
dashboard_content += f"📝 **Recent Conversations ({len(conversations)}):**\n\n"
for i, conv in enumerate(conversations, 1):
dashboard_content += f"**{i}. {conv['title']}**\n"
dashboard_content += f"• Mode: {conv['persona_mode'].title()}\n"
dashboard_content += f"• Messages: {conv['message_count']}\n"
dashboard_content += f"• Updated: {conv['updated_at'][:19].replace('T', ' ')}\n"
dashboard_content += f"• Session ID: `{conv['session_id'][:8]}...`\n\n"
dashboard_content += "💡 **Tip:** Type 'logout' to end your session securely."
await cl.Message(content=dashboard_content).send()
return
if user_input == "logout":
await thinking_msg.remove()
if not AUTH_AVAILABLE or not auth_status["authenticated"]:
await cl.Message(content="❌ You are not logged in.").send()
return
# Logout user
session_token = cl.user_session.get("session_token")
if session_token:
auth_manager.logout_user(session_token)
# Clear session data
cl.user_session.set("auth_token", None)
cl.user_session.set("session_token", None)
cl.user_session.set("user_id", None)
cl.user_session.set("username", None)
cl.user_session.set("user_email", None)
cl.user_session.set("subscription_tier", None)
await cl.Message(
content=f"👋 **Goodbye, {auth_status['username']}!**\n\n"
"You have been logged out successfully.\n"
"Type `login username password` to log back in."
).send()
return
# =============================
# ROUTE: UK ECONOMIST MODE
# =============================
if "economist mode" in user_input or "economics mode" in user_input or "uk economy" in user_input:
await thinking_msg.remove()
cl.user_session.set("persona", "economist")
persona = get_current_persona()
recommendations = format_persona_recommendations("economist")
await cl.Message(
content=f"{persona['style']}\n\n"
"I'm now analyzing from a **UK economics perspective**, combining macroeconomic trends with startup viability.\n\n"
f"{recommendations}"
"**Current UK Economic Context:**\n"
"• Bank Rate: 4.75% (affecting cost of capital)\n"
"• CPI Inflation: 2.3% (near BoE target)\n"
"• Unemployment: 4.2% (tight labour market)\n"
"• GDP Growth: 0.3% quarterly (sluggish growth)\n"
"• GBP/USD: 1.27 (currency impacts)\n\n"
"**Economic Analysis Tools:**\n"
"• Type **'macro analysis'** - UK macroeconomic impact assessment\n"
"• Type **'sector outlook'** - UK sector-specific opportunities\n"
"• Type **'regional analysis'** - Location-based economic factors\n"
"• Type **'policy impact'** - Government policy effects\n\n"
"**Key Questions I Can Answer:**\n"
"• How do interest rates affect your funding strategy?\n"
"• What's the inflation impact on your cost structure?\n"
"• How does UK productivity affect your scaling plans?\n"
"• Which UK regions offer the best opportunities?\n"
"• How do fiscal policies impact your sector?\n\n"
"**Ask me about:**\n"
"• Brexit impacts on your market\n"
"• London vs regional economics\n"
"• UK labour market conditions\n"
"• Sector-specific regulations\n"
"• Currency exposure and hedging"
).send()
return
if "persona" in user_input or "mode" in user_input:
current_persona = get_current_persona()
await cl.Message(
content=f"## 🎭 Current Mode: {current_persona['style']}\n\n"
"**Available modes:**\n\n"
"💼 **Investor Mode** - VC perspective focused on ROI and exit strategies\n"
"• Best for: Portfolio analysis, due diligence, investment decisions\n"
"• Commands: portfolio, insights, sector dashboard\n\n"
"🚀 **Founder Mode** - Entrepreneur perspective focused on execution\n"
"• Best for: Startup assessment, risk reduction, tactical advice\n"
"• Commands: assess idea, benchmark, timeline\n\n"
"🇬🇧 **UK Economist Mode** - Economic analysis perspective for UK markets\n"
"• Best for: Macroeconomic impacts, regional analysis, policy effects\n"
"• Commands: macro analysis, sector outlook, regional analysis\n\n"
"**Ready to switch?**\n"
"• Type **'investor mode'** for VC analysis\n"
"• Type **'founder mode'** for founder guidance\n"
"• Type **'economist mode'** for UK economic analysis\n\n"
"**Or continue in current mode - what would you like to analyze?**"
).send()
return
# =============================
# ROUTE: COMPANY ANALYST MODE
# =============================
if "company analyst" in user_input or "company analysis" in user_input or "financial analysis" in user_input:
await thinking_msg.remove()
cl.user_session.set("persona", "company_analyst")
persona = get_current_persona()
await cl.Message(
content=f"{persona['style']}\n\n"
"I'm now analyzing from a **company financial health perspective**, focusing on profitability, unit economics, and financial sustainability.\n\n"
"**Financial Analysis Focus:**\n"
"• Profitability: Gross, operating, and net margins\n"
"• Unit Economics: LTV/CAC ratios, payback periods\n"
"• Cash Flow: Runway analysis, working capital\n"
"• Break-even: Path to profitability analysis\n"
"• Benchmarking: Industry performance comparisons\n\n"
"**Analysis Tools:**\n"
"• Type **'analyze company'** - Comprehensive financial analysis\n"
"• Type **'profitability analysis'** - Full margin and profitability assessment\n"
"• Type **'unit economics'** - Customer economics and LTV/CAC analysis\n"
"• Type **'cash flow analysis'** - Runway and cash management assessment\n"
"• Type **'break even analysis'** - Path to profitability calculation\n\n"
"**Ready to dive deep into financial health?**"
).send()
return
# =============================
# ROUTE: COMPANY ANALYSIS EXECUTION
# =============================
if "analyze company" in user_input or "profitability analysis" in user_input or "financial analysis" in user_input:
await thinking_msg.remove()
analyzer = CompanyAnalyzer()
await cl.Message(
content="## 💼 Company Financial Analysis\n\n"
"I'll perform a comprehensive profitability and financial health analysis.\n\n"
"Choose analysis type:\n"
"1. **Quick Analysis** - Key metrics only\n"
"2. **Full Analysis** - Complete financial deep dive\n"
"3. **Upload Financials** - Analyze from CSV/Excel"
).send()
analysis_type = await cl.AskUserMessage(content="Enter choice (1/2/3):").send()
if "2" in analysis_type.get('output', '') or "full" in analysis_type.get('output', '').lower():
# Full Analysis
await cl.Message(content="### 📊 Full Company Analysis\n\nI'll need detailed financial information.").send()
# Collect comprehensive data
company_name = await cl.AskUserMessage(content="Company name:").send()
industry = await cl.AskUserMessage(content="Industry (SaaS/E-commerce/Marketplace/FinTech/Services):").send()
# Revenue metrics
revenue = await ask_float("Annual revenue ($ millions)", 10.0)
growth_rate = await ask_float("Revenue growth rate (%)", 20.0)
# Cost structure
cogs_pct = await ask_float("COGS as % of revenue", 40.0)
opex_pct = await ask_float("Operating expenses as % of revenue", 35.0)
sales_marketing_pct = await ask_float("Sales & Marketing as % of revenue", 15.0)
# Unit economics
cac = await ask_float("Customer Acquisition Cost ($)", 100.0)
ltv = await ask_float("Customer Lifetime Value ($)", 400.0)
monthly_churn = await ask_float("Monthly churn rate (%)", 5.0)
# Cash metrics
cash_balance = await ask_float("Current cash balance ($ millions)", 5.0)
monthly_burn = await ask_float("Monthly burn rate ($ thousands)", 200.0)
# Prepare financial data
revenue_amount = revenue * 1_000_000
financials = {
'revenue': revenue_amount,
'cogs': revenue_amount * (cogs_pct / 100),
'opex': revenue_amount * (opex_pct / 100),
'sales_marketing': revenue_amount * (sales_marketing_pct / 100),
'fixed_costs': revenue_amount * 0.2,
'variable_cost_ratio': cogs_pct / 100,
'price_per_unit': 100,
'current_revenue': revenue_amount / 12 # Monthly
}
metrics = {
'customer_acquisition_cost': cac,
'lifetime_value': ltv,
'monthly_revenue': ltv / 24, # Assume 24-month lifetime
'revenue_per_unit': 100,
'variable_cost_per_unit': 40
}
cash_data = {
'cash_from_operations': -monthly_burn * 1000 * 12,
'cash_from_investing': -revenue_amount * 0.05,
'cash_from_financing': 0,
'cash_balance': cash_balance * 1_000_000,
'days_sales_outstanding': 45,
'days_inventory_outstanding': 0 if industry.get('output', '') == 'SaaS' else 30,
'days_payables_outstanding': 30
}
# Run analyses
msg = cl.Message(content="🔍 Analyzing financial health...")
await msg.send()
profitability = analyzer.analyze_profitability(financials)
unit_economics = analyzer.analyze_unit_economics(metrics)
cash_flow = analyzer.analyze_cash_flow(cash_data)
break_even = analyzer.calculate_break_even(financials)
# Benchmark analysis
company_metrics = {
'gross_margin': profitability['gross_margin'],
'operating_margin': profitability['operating_margin'],
'ltv_cac_ratio': unit_economics['ltv_cac_ratio']
}
benchmarks = analyzer.benchmark_performance(
company_metrics,
industry.get('output', 'Services')
)
# Generate visualization
analysis_data = {
'ltv_cac_ratio': unit_economics['ltv_cac_ratio']
}
chart = plot_profitability_analysis(analysis_data)
await msg.remove()
# Display comprehensive results
content = f"""
## 📊 Financial Analysis: {company_name.get('output', 'Company')}
### 💰 Profitability Analysis
- **Gross Margin:** {profitability['gross_margin']:.1f}% {'✅' if profitability['gross_margin'] > 50 else '⚠️' if profitability['gross_margin'] > 30 else '❌'}
- **Operating Margin:** {profitability['operating_margin']:.1f}% {'✅' if profitability['operating_margin'] > 15 else '⚠️' if profitability['operating_margin'] > 0 else '❌'}
- **EBITDA Margin:** {profitability['ebitda_margin']:.1f}%
- **Net Margin:** {profitability['net_margin']:.1f}%
- **Overall Health:** {profitability['profit_health']}
### 📈 Unit Economics
- **LTV/CAC Ratio:** {unit_economics['ltv_cac_ratio']:.2f} {'✅ Healthy' if unit_economics['ltv_cac_ratio'] > 3 else '⚠️ Concerning' if unit_economics['ltv_cac_ratio'] > 1 else '❌ Unsustainable'}
- **Payback Period:** {unit_economics['payback_months']:.1f} months
- **Contribution Margin:** {unit_economics['contribution_margin_pct']:.1f}%
- **Unit Economics:** {unit_economics['unit_economics_health']}
### 💵 Cash Flow & Runway
- **Monthly Burn:** ${cash_flow['monthly_burn']:,.0f}
- **Runway:** {cash_flow['runway_months']:.1f} months {'✅' if cash_flow['runway_months'] > 18 else '⚠️' if cash_flow['runway_months'] > 12 else '❌'}
- **Cash Conversion Cycle:** {cash_flow['cash_conversion_cycle']:.0f} days
- **Cash Efficiency:** {cash_flow['cash_efficiency']}
### 🎯 Break-even Analysis
- **Break-even Revenue:** ${break_even['break_even_revenue']:,.0f}
- **Margin of Safety:** {break_even['margin_of_safety']:.1f}%
- **Months to Break-even:** {break_even['months_to_break_even'] if break_even['months_to_break_even'] < 60 else '60+'}
### 📊 Industry Benchmarks ({industry.get('output', 'Services')})
- **Gross Margin:** Company {benchmarks['comparisons']['gross_margin']['company']:.1f}% vs Industry {benchmarks['comparisons']['gross_margin']['industry']:.1f}% ({benchmarks['comparisons']['gross_margin']['performance']})
- **Operating Margin:** Company {benchmarks['comparisons']['operating_margin']['company']:.1f}% vs Industry {benchmarks['comparisons']['operating_margin']['industry']:.1f}% ({benchmarks['comparisons']['operating_margin']['performance']})
- **Overall Rating:** {benchmarks['overall_rating']}
### 💡 Key Recommendations
1. {'✅ Maintain strong margins' if profitability['gross_margin'] > 50 else '⚠️ Improve gross margins through pricing or cost reduction'}
2. {'✅ Unit economics are healthy' if unit_economics['ltv_cac_ratio'] > 3 else '⚠️ Optimize CAC or increase LTV'}
3. {'✅ Adequate runway' if cash_flow['runway_months'] > 18 else '❌ Consider fundraising or reducing burn'}
4. {'✅ Near profitability' if break_even['margin_of_safety'] > 0 else '⚠️ Focus on path to profitability'}
### 🎬 Action Items
{chr(10).join(['• ' + rec for rec in benchmarks['recommendations']])}
"""
text_msg = cl.Message(content=content)
await text_msg.send()
# Send chart
image = cl.Image(content=chart, name="company_analysis.png", display="inline")
await image.send(for_id=text_msg.id)
else: # Quick Analysis
await cl.Message(content="### ⚡ Quick Profitability Check\n\nProvide key metrics for rapid assessment.").send()
revenue = await ask_float("Monthly revenue ($ thousands)", 100.0)
costs = await ask_float("Monthly costs ($ thousands)", 120.0)
customers = await ask_int("Number of customers", 100, mi=1, ma=100000)
# Quick calculations
profit = revenue - costs
margin = (profit / revenue * 100) if revenue > 0 else -100
revenue_per_customer = (revenue * 1000) / customers if customers > 0 else 0
status = "📈 Profitable" if profit > 0 else "📉 Not Yet Profitable"
health = "Strong" if margin > 20 else "Moderate" if margin > 0 else "Needs Improvement"
content = f"""
### ⚡ Quick Analysis Results
**Status:** {status}
**Net Margin:** {margin:.1f}%
**Monthly Profit/Loss:** ${profit*1000:,.0f}
**Revenue per Customer:** ${revenue_per_customer:.2f}
**Financial Health:** {health}
**Quick Insights:**
- {'Focus on achieving profitability' if profit < 0 else 'Maintain positive trajectory'}
- {'Reduce costs or increase pricing' if margin < 0 else 'Consider scaling'}
- {'Improve customer monetization' if revenue_per_customer < 100 else 'Good customer value'}
"""
await cl.Message(content=content).send()
return
# =============================
# ROUTE 10: WEB SCRAPING
# =============================
if user_input.startswith("scrape "):
# Remove thinking indicator
await thinking_msg.remove()
# Parse scrape command: "scrape " or "scrape "
parts = message.content.strip().split()
if len(parts) < 2:
await cl.Message(
content="⚠️ **Invalid scrape command**\n\n"
"**Usage:** `scrape ` or `scrape `\n\n"
"**Examples:**\n"
"• `scrape https://example.com` - Scrape paragraphs\n"
"• `scrape https://news.site h1,h2` - Scrape headlines\n"
"• `scrape https://blog.com .article` - Scrape articles by class"
).send()
return
url = parts[1]
selector = parts[2] if len(parts) > 2 else "p" # Default to paragraphs
# Show loading message with URL
loading_msg = cl.Message(content=f"🔍 Scraping {url}...")
await loading_msg.send()
# Perform scraping
scrape_result = scrape_site(url, selector)
# Remove loading message
await loading_msg.remove()
if not scrape_result["success"]:
# Handle scraping failure
await cl.Message(
content=f"❌ **Scraping failed**\n\n"
f"**URL:** {url}\n"
f"**Error:** {scrape_result['error']}\n\n"
f"**Suggestions:**\n"
f"• Check if the URL is accessible in your browser\n"
f"• Try a different CSS selector (h1, div, span)\n"
f"• Some sites block automated requests"
).send()
return
# Scraping successful - get results
scraped_data = scrape_result["data"]
count = scrape_result["count"]
size_mb = scrape_result["size_mb"]
# Show scraping summary
summary_msg = cl.Message(
content=f"✅ **Scraping successful!**\n\n"
f"**URL:** {url}\n"
f"**Selector:** `{selector}`\n"
f"**Items scraped:** {count}\n"
f"**Content size:** {size_mb}MB\n\n"
f"**Preview (first 3 items):**"
)
await summary_msg.send()
# Show preview of scraped content
preview_items = scraped_data.head(3)
preview_text = ""
for i, row in preview_items.iterrows():
content = row["content"]
if len(content) > 200:
content = content[:200] + "..."
preview_text += f"**{i+1}.** {content}\n\n"
preview_msg = cl.Message(content=preview_text)
await preview_msg.send()
# Store scraped data in session memory for later reference
session_id = get_session_id()
add_to_memory(session_id, "scraped_data", f"Scraped {count} items from {url}")
# Store the actual data in session for follow-up questions
cl.user_session.set("last_scraped_data", scraped_data)
cl.user_session.set("last_scraped_url", url)
# Generate AI analysis of scraped content
analysis_msg = cl.Message(content="🤖 **Analyzing scraped content...**")
await analysis_msg.send()
# Get current persona for contextual analysis
persona = get_current_persona()
analysis = analyze_scraped_content(scraped_data, url, persona)
# Remove analysis loading message
await analysis_msg.remove()
# Send AI analysis with persona indicator
analysis_response = f"**Website Analysis**\n\n{analysis}"
await cl.Message(content=analysis_response).send()
return
# =============================
# ROUTE 11: AUTO-GENERATED INSIGHTS
# =============================
if "insights" in user_input or "auto insights" in user_input or "generate insights" in user_input:
msg = cl.Message(content="🤖 Analyzing data and generating insights...")
await msg.send()
insights = generate_insights(df, "general")
insights_message = format_insights_message(insights)
await msg.remove()
await cl.Message(content=insights_message).send()
return
# =============================
# ROUTE 12: DATA EXPORT & DOWNLOAD
# =============================
if any(keyword in user_input for keyword in ["export data", "download data", "export csv", "download csv", "export json", "download json"]):
await thinking_msg.remove()
# Determine export format
if "json" in user_input:
format_type = "json"
else:
format_type = "csv" # Default to CSV
# Send data export
await send_data_export(df, "startup_dataset", format_type)
return
# =============================
# ROUTE 14: VOICE STREAMING & CONVERSATION
# =============================
if any(keyword in user_input for keyword in ["voice chat", "start voice", "voice streaming", "talk to", "voice conversation"]):
# Initialize voice streaming interface
voice_ui = VoiceUIManager()
voice_interface = await voice_ui.create_voice_interface()
await cl.Message(
content="🎤 **Voice Conversation Mode**\n\n"
"Choose your AI coaching persona and start a voice conversation!\n\n"
"**Available Features:**\n"
"• Real-time voice conversation with OpenAI\n"
"• Multiple specialized coaching personas\n"
"• Audio visualization and recording\n"
"• Conversation summaries\n\n"
"Click the microphone button below to start:",
elements=[cl.Html(content=voice_interface)]
).send()
return
# Handle voice-related messages from JavaScript interface
if hasattr(message, 'type') and message.type in ['voice_audio', 'switch_persona', 'get_summary']:
voice_ui = VoiceUIManager()
response = await voice_ui.handle_voice_message(message.type, message.content)
if response.get('status') == 'success':
await cl.Message(content=f"✅ {response.get('message', 'Voice command processed')}").send()
else:
await cl.Message(content=f"❌ {response.get('message', 'Voice command failed')}").send()
return
# =============================
# ROUTE 15: AI STARTUP COACHING PERSONAS
# =============================
if any(keyword in user_input for keyword in ["coaching", "advisor", "persona", "coach", "startup coach", "choose coach"]):
# Initialize persona manager
persona_ui = PersonaUIManager()
coach_personas = StartupCoachPersonas()
if "switch" in user_input or "change" in user_input:
# Handle persona switching
personas = coach_personas.get_available_personas()
persona_list = "\n".join([f"• **{p['name']}** - {', '.join(p['expertise'][:2])}"
for p in personas.values()])
await cl.Message(
content=f"🎯 **Available Startup Coaches:**\n\n{persona_list}\n\n"
"Which coach would you like to work with? Just mention their name or expertise area."
).send()
return
# Show persona selector interface
persona_interface = persona_ui.create_persona_selector()
await cl.Message(
content="🎯 **AI-Powered Startup Coaching**\n\n"
"Choose from our specialized AI coaches, each with deep expertise in different areas of startup development:",
elements=[cl.Html(content=persona_interface)]
).send()
return
# Handle persona selection from interface
if hasattr(message, 'type') and message.type == 'select_persona':
coach_personas = StartupCoachPersonas()
persona_id = message.content.get('persona_id')
startup_context = message.content.get('startup_context', {})
session_result = await coach_personas.initialize_coaching_session(persona_id, startup_context)
if session_result.get('status') == 'success':
persona = session_result['persona']
welcome = session_result['welcome_message']
# Store session in Chainlit user session
cl.user_session.set("coaching_session", session_result['session_id'])
cl.user_session.set("current_persona", persona_id)
await cl.Message(
content=f"✅ **{persona['avatar']} {persona['name']} is now your coach!**\n\n"
f"{welcome}\n\n"
f"**Expertise:** {', '.join(persona['expertise'])}\n\n"
"Ask me anything about your startup - I'm here to help!"
).send()
else:
await cl.Message(content=f"❌ Error initializing coaching session: {session_result.get('message')}").send()
return
# =============================
# ROUTE 16: MARKET INTELLIGENCE DASHBOARD
# =============================
if any(keyword in user_input for keyword in ["market intelligence", "market dashboard", "competitive analysis",
"market trends", "sector analysis", "market data"]):
# Initialize market intelligence dashboard
market_dashboard = MarketIntelligenceDashboard()
dashboard_interface = await market_dashboard.create_dashboard_interface()
await cl.Message(
content="📊 **Market Intelligence Dashboard**\n\n"
"Access real-time market data, competitive intelligence, and trend analysis:\n\n"
"**Features:**\n"
"• Sector performance analysis\n"
"• Competitive landscape mapping\n"
"• Market trend identification\n"
"• Investment theme analysis\n"
"• Real-time data visualization\n\n"
"Use the controls below to analyze your market:",
elements=[cl.Html(content=dashboard_interface)]
).send()
return
# Handle dashboard requests from interface
if hasattr(message, 'type') and message.type == 'dashboard_request':
market_dashboard = MarketIntelligenceDashboard()
request_type = message.content.get('request_type')
data = message.content.get('data', {})
response = await market_dashboard.handle_dashboard_request(request_type, data)
if response.get('status') == 'success':
# Send the response back to the dashboard
await cl.Message(content="📊 Dashboard updated with latest data.").send()
else:
await cl.Message(content=f"❌ Dashboard error: {response.get('message')}").send()
return
# =============================
# ROUTE 17: AI-POWERED Q&A WITH MEMORY & PERSONA (DEFAULT)
# =============================
# If no specific pattern matched, use GPT-4 for natural language response
# -------------------------
# SESSION MEMORY & PERSONA INTEGRATION
# -------------------------
session_id = cl.user_session.get("session_id", get_session_id())
persona = get_current_persona()
memory_context = get_memory_context(session_id)
# Add user message to memory
add_to_memory(session_id, "user", message.content)
# Save user message to database if authenticated
if AUTH_AVAILABLE and auth_status["authenticated"]:
auth_manager.save_conversation(
user_id=auth_status["user_id"],
chainlit_session_id=session_id,
role="user",
content=message.content,
persona_mode=get_current_persona()["name"].lower().replace(" mode", ""),
metadata={"timestamp": timestamp, "raw_input": user_input_raw}
)
# -------------------------
# PREPARE ENHANCED CONTEXT
# -------------------------
# Convert DataFrame to string for inclusion in prompt
df_str = df.to_string(index=False)
# Auto-search enhancement for relevant queries
search_results = None
search_context = ""
current_persona_name = cl.user_session.get("persona", "founder")
# -------------------------
# USE THREAD-AWARE PROCESSING FOR LANGSMITH
# -------------------------
# Check if LangSmith is enabled and use thread context
if langsmith_client:
# Use the thread-aware processing function with LangSmith tracing
enhanced_question = f"Dataset:\n{df_str}\n\nUser question: {message.content}"
ai_response = process_with_thread_context(
question=enhanced_question,
session_id=session_id,
get_chat_history=True, # Always use history for continuity
persona=persona
)
else:
# Fallback to standard processing without LangSmith tracing
enhanced_system_prompt = (
f"{persona['system_prompt']}\n\n"
"Available commands you can suggest:\n"
"- 'timeline' - failure timeline\n"
"- 'funding vs burn' - funding vs burn rate\n"
"- 'interactive dashboard' - interactive scatter plot\n"
"- 'interactive timeline' - interactive failure timeline\n"
"- 'sector dashboard' - multi-chart sector analysis\n"
"- 'benchmark' - compare founder idea to dataset\n"
"- 'portfolio' - analyze multiple startups with heatmap\n"
"- 'insights' - auto-generate risks and recommendations\n"
"- 'investor mode' / 'founder mode' - switch analysis perspective\n\n"
"Remember conversation history when relevant. "
"If the user asks a question that would be better answered with a visualization, "
"suggest they try one of these commands."
)
# Use LangSmith thread management for conversation continuity
current_persona = PERSONAS[current_persona_name]
# Check if we should use conversation history (after first message in session)
session_metadata = get_session_metadata(session_id)
use_history = session_metadata["conversation_count"] > 0
# Determine if this query would benefit from real-time search
search_triggers = [
"market", "competition", "trends", "latest", "recent", "current", "2024", "2025",
"startup", "funding", "investment", "industry", "valuation", "growth", "exit"
]
should_search = any(trigger in message.content.lower() for trigger in search_triggers)
if should_search and search_api_key:
# Generate persona-specific search query
search_query = generate_search_query(message.content, current_persona_name)
if search_query:
# Show search indicator
search_msg = await cl.Message(content="🔍 Searching for latest market intelligence...").send()
# Perform search
search_results = search_internet(search_query, count=3)
# Remove search indicator
await search_msg.remove()
if search_results["success"]:
search_context = f"\n\nRECENT MARKET INTELLIGENCE:\n"
for i, result in enumerate(search_results["results"], 1):
search_context += f"\n{i}. **{result['title']}**\n"
search_context += f" {result['description']}\n"
search_context += f" Source: {result['url']}\n"
# Enhance user question with dataset and search context
enhanced_question = f"Dataset:\n{df_str}\n\nUser question: {message.content}"
if search_context:
enhanced_question += search_context
# Use the LangSmith chat pipeline for thread-aware responses
ai_response = navada_chat_pipeline(
question=enhanced_question,
session_id=session_id,
persona=current_persona_name,
get_chat_history=use_history
)
# -------------------------
# SEND AI RESPONSE WITH PERSONA INDICATOR
# -------------------------
# Add AI response to memory
add_to_memory(session_id, "assistant", ai_response)
# Save AI response to database if authenticated
if AUTH_AVAILABLE and auth_status["authenticated"]:
auth_manager.save_conversation(
user_id=auth_status["user_id"],
chainlit_session_id=session_id,
role="assistant",
content=ai_response,
persona_mode=get_current_persona()["name"].lower().replace(" mode", ""),
metadata={
"search_used": bool(search_context),
"search_results_count": len(search_results.get("results", [])) if search_results else 0,
"persona": current_persona_name,
"enhanced_with_search": bool(search_context and search_results and search_results["success"])
}
)
# Add persona indicator to response (with safety check)
if not ai_response or ai_response.strip() == "":
ai_response = "I apologize, but I encountered an issue generating a response. Please try again."
response_with_persona = ai_response
# Add search intelligence indicator if search was used
if search_context and search_results and search_results["success"]:
response_with_persona += f"\n\n---\n\n*🔍 Enhanced with real-time market intelligence from {len(search_results['results'])} sources*"
# Generate and append auto-insights for analysis-type responses
if any(keyword in user_input for keyword in ["analyze", "analysis", "compare", "evaluate"]):
insights = generate_insights(df, "analysis")
if insights["risks"] or insights["opportunities"] or insights["recommendations"]:
response_with_persona += "\n\n" + format_insights_message(insights)
# Remove thinking indicator before sending final response
await thinking_msg.remove()
# Send the text response
message = await cl.Message(content=response_with_persona).send()
# -------------------------
# AUTO TEXT-TO-SPEECH
# -------------------------
# Check if TTS is enabled in user settings
tts_enabled = cl.user_session.get("tts_enabled", False)
if tts_enabled and response_with_persona:
try:
# Clean response text for TTS (remove markdown, emojis, etc.)
clean_text = clean_text_for_tts(response_with_persona)
if clean_text.strip():
# Generate and send audio
audio_element = await generate_speech(clean_text)
if audio_element:
await cl.Message(
content="🔊 Audio version:",
elements=[audio_element]
).send()
except Exception as e:
print(f"⚠️ Auto-TTS failed: {e}")
def clean_text_for_tts(text: str) -> str:
"""Clean text for text-to-speech by removing markdown and special characters."""
import re
# Remove markdown formatting
text = re.sub(r'\*\*(.*?)\*\*', r'\1', text) # Bold
text = re.sub(r'\*(.*?)\*', r'\1', text) # Italic
text = re.sub(r'`(.*?)`', r'\1', text) # Code
text = re.sub(r'#+ ', '', text) # Headers
text = re.sub(r'\[.*?\]\(.*?\)', '', text) # Links
text = re.sub(r'---+', '', text) # Horizontal rules
# Remove emojis and special characters
text = re.sub(r'[🔍🚀💼📊🔹⚡📈🎯💡🔧📚⏰📝🔗✅⚠️🎙️🔊📢]', '', text)
# Clean up multiple spaces and newlines
text = re.sub(r'\n+', '. ', text)
text = re.sub(r'\s+', ' ', text)
# Limit length for TTS
return text.strip()[:1000]
# =============================
# LANGSMITH PLATFORM OPTIMIZATION
# =============================
@traceable
def initialize_knowledge_base():
"""Initialize vector store with external LangChain database or fallback to local."""
global vector_store
if not vector_store:
# Try to connect to external LangChain database first
if langchain_database_id and CHROMA_AVAILABLE:
try:
# Initialize LangChain components if not already done
embeddings_func, _ = initialize_langchain_components()
# Connect to external LangChain database using the provided ID
vector_store = Chroma(
embedding_function=embeddings_func,
persist_directory=f"./langchain_db_{langchain_database_id}"
)
print(f"✅ Connected to LangChain database: {langchain_database_id[:8]}...")
except Exception as e:
print(f"⚠️ Failed to connect to LangChain database: {e}")
# Fallback to local knowledge base
vector_store = _create_local_knowledge_base()
else:
# Fallback to local knowledge base
vector_store = _create_local_knowledge_base()
return vector_store
def _create_local_knowledge_base():
"""Create local knowledge base as fallback."""
# Startup knowledge optimized for LangSmith platform
startup_knowledge = [
"Successful startups show product-market fit within 18-24 months",
"SaaS startups should aim for 20% month-over-month growth",
"B2B startups need longer sales cycles but higher LTV",
"Consumer apps require viral growth and strong engagement",
"Hardware startups need more capital and longer dev cycles",
"Fintech faces regulatory challenges but high market opportunity",
"AI/ML startups need strong technical teams and data advantages",
"E-commerce should focus on unit economics and CAC",
"Database ID: " + (langchain_database_id or "local-fallback")
]
documents = [Document(page_content=text) for text in startup_knowledge]
return Chroma.from_documents(
documents=documents,
embedding=embeddings,
persist_directory="./chroma_db"
)
@traceable
def enhanced_rag_response(user_query: str, context: str) -> str:
"""Generate enhanced response using RAG system optimized for LangSmith."""
if not vector_store:
initialize_knowledge_base()
# Query knowledge base
docs = vector_store.similarity_search(user_query, k=3)
relevant_knowledge = [doc.page_content for doc in docs]
# Enhanced prompt with RAG context
rag_prompt = f"""
Based on startup knowledge and context, provide actionable insights:
Relevant Knowledge:
{chr(10).join(relevant_knowledge)}
Context: {context}
Query: {user_query}
Provide detailed, actionable response with specific recommendations.
"""
response = llm.invoke(rag_prompt)
return response.content
# Health check endpoint for LangSmith platform
@cl.on_settings_update
async def health_check():
"""Health check endpoint for LangSmith monitoring."""
return {"status": "healthy", "app": "NAVADA", "version": "1.0.0"}
# =============================
# LANGGRAPH AGENT EXPORT
# =============================
# Export agent for LangGraph deployment