Fix error handling and improve crypto data retrieval

app.py

@@ -6,179 +6,65 @@ from utils import (
     evaluate_expertise,
     get_crypto_data,
 )
-import random
 
 
 def run_chat(crypto, data_span):
-    # This non-generator wrapper is needed to avoid Gradio issues
-    def _run_chat_gen(crypto, data_span):
-        # Validate inputs
-        valid_cryptos = ["btc", "eth", "virtual", "bnb"]
-        valid_spans = ["24h", "7d", "14d", "30d", "60d"]
-        if crypto not in valid_cryptos or data_span not in valid_spans:
-            yield [("🤖", "❌ Invalid crypto or data span selection")], []
-            return
-
-        try:
-            crypto_data = get_crypto_data(crypto, data_span)
-            chat_history = []
-            previous_qa = ""
-            qa_history = []
-
-            # Add an initial bot message to confirm it's working
-            chat_history.append(("Starting analysis for " + crypto + " over " + data_span + " timeframe...", ""))
-            yield chat_history, []
-            time.sleep(1)
-            
-            for i in range(5):
-                # --- QUESTION ---
-                question = get_question(previous_qa)
-                # append as a right-only message
-                chat_history.append(("", question.content))
-                yield chat_history, []  # show question
-                time.sleep(1)
-
-                # --- ANSWER ---
-                answer = get_answer(question.content, previous_qa, crypto_data)
-                # append as a left-only message
-                chat_history.append((answer.content, ""))
-                yield chat_history, []  # show answer
-
-                # track for final evaluation
-                qa_history.append({"question": question.content, "answer": answer.content})
-                previous_qa += f"\nQ: {question.content}\nA: {answer.content}\n"
-                time.sleep(1)
-
-            # --- FINAL EVALUATION ---
-            evaluation = evaluate_expertise(qa_history)
-            chat_history.append(("📊 Evaluation:\n" + evaluation.content, ""))
-            yield chat_history, chat_history
-            
-        except Exception as e:
-            import traceback
-            error_msg = f"Error: {str(e)}\n{traceback.format_exc()}"
-            chat_history = [("⚠️ An error occurred during the analysis: " + str(e), "")]
-            yield chat_history, []
-    
-    # Return the generator itself - this is the key fix
-    return _run_chat_gen(crypto, data_span)
-
-
-# Use a standard theme that's compatible with all Gradio versions
-custom_css = """
-.gradio-container {
-    background: linear-gradient(to bottom right, rgba(80, 80, 255, 0.05), rgba(0, 170, 170, 0.05));
-    font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif;
-}
-.gr-button-primary {
-    background-color: #6366f1 !important;
-    color: white !important;
-}
-.gr-box {
-    border-radius: 10px !important;
-    box-shadow: 0 4px 6px rgba(0, 0, 0, 0.1) !important;
-}
-"""
-
-# Create emoji indicators for cryptocurrencies
-crypto_emojis = {
-    "btc": "₿ Bitcoin",
-    "eth": "Ξ Ethereum",
-    "virtual": "🔮 Virtual",
-    "bnb": "🪙 BNB"
-}
-
-# Create time span emojis
-time_emojis = {
-    "24h": "⏰ 24 Hours",
-    "7d": "📅 7 Days",
-    "14d": "🗓️ 14 Days",
-    "30d": "📆 30 Days",
-    "60d": "📈 60 Days"
-}
-
-with gr.Blocks(css=custom_css) as demo:
-    with gr.Box(css="padding: 20px; border-radius: 10px; box-shadow: 0 4px 6px rgba(0, 0, 0, 0.1);"):
-        gr.Markdown(
-            """
-            # 🧠 TranchVest AI - Crypto Risk Assessment
-            ### Analyze cryptocurrency behavior with AI-powered insights
-            """
+    # Validate inputs
+    valid_cryptos = ["btc", "eth", "virtual", "bnb"]
+    valid_spans = ["24h", "7d", "14d", "30d", "60d"]
+    if crypto not in valid_cryptos or data_span not in valid_spans:
+        yield [("🤖", "❌ Invalid crypto or data span selection")], []
+        return
+
+    crypto_data = get_crypto_data(crypto, data_span)
+    chat_history = []
+    previous_qa = ""
+    qa_history = []
+
+    for _ in range(5):
+        # --- QUESTION ---
+        question = get_question(previous_qa)
+        # append as a right-only message
+        chat_history.append(("", question.content))
+        yield chat_history, []  # show question
+        time.sleep(1)
+
+        # --- ANSWER ---
+        answer = get_answer(question.content, previous_qa, crypto_data)
+        # append as a left-only message
+        chat_history.append((answer.content, ""))
+        yield chat_history, []  # show answer
+
+        # track for final evaluation
+        qa_history.append({"question": question.content, "answer": answer.content})
+        previous_qa += f"\nQ: {question.content}\nA: {answer.content}\n"
+        time.sleep(1)
+
+    # --- FINAL EVALUATION ---
+    evaluation = evaluate_expertise(qa_history)
+    chat_history.append(("📊 Evaluation:", evaluation.content))
+    yield chat_history, chat_history
+
+
+with gr.Blocks() as demo:
+    gr.Markdown("## 🧠 TranchVest AI - Crypto Behavior Analysis")
+
+    with gr.Row():
+        crypto_dropdown = gr.Dropdown(
+            ["btc", "eth", "virtual", "bnb"], label="Select Crypto"
         )
-        
-        with gr.Row(equal_height=True):
-            with gr.Column(scale=1):
-                gr.Markdown("### 🚀 Configuration")
-                with gr.Box(css="padding: 15px; border-radius: 8px; background: rgba(0, 0, 0, 0.03);"):
-                    crypto_dropdown = gr.Dropdown(
-                        list(crypto_emojis.values()), 
-                        label="Select Cryptocurrency",
-                        value="₿ Bitcoin",
-                        container=True,
-                        interactive=True
-                    )
-                    span_dropdown = gr.Dropdown(
-                        list(time_emojis.values()), 
-                        label="Select Time Frame",
-                        value="📅 7 Days",
-                        container=True,
-                        interactive=True
-                    )
-                    start_btn = gr.Button(
-                        "🔍 Start Analysis", 
-                        variant="primary",
-                        size="lg"
-                    )
-
-            with gr.Column(scale=2):
-                chatbot = gr.Chatbot(
-                    height=600,
-                    bubble_full_width=False,
-                    show_copy_button=True,
-                    avatar_images=("https://em-content.zobj.net/source/microsoft-teams/363/robot_1f916.png", None),
-                    css="border-radius: 12px; box-shadow: 0 2px 10px rgba(0, 0, 0, 0.05);"
-                )
-
-    # Tips and information section
-    with gr.Accordion("ℹ️ About This Tool", open=False):
-        gr.Markdown(
-            """
-            ### How to use this tool:
-            1. **Select a cryptocurrency** from the dropdown menu
-            2. **Choose a time frame** for analysis
-            3. **Click 'Start Analysis'** to begin the AI-driven assessment
-            4. The AI will ask a series of questions about risk factors
-            5. At the end, you'll receive a comprehensive risk evaluation
-            
-            This tool helps evaluate investment risk factors for different cryptocurrencies based on market behavior and trends.
-            """
+        span_dropdown = gr.Dropdown(
+            ["24h", "7d", "14d", "30d", "60d"], label="Select Data Span"
         )
-        
-    # Add hidden state
+        start_btn = gr.Button("Start Chat")
+
+    chatbot = gr.Chatbot(height=600)
     state = gr.State([])
 
-    # Process the input values from the dropdowns to match the expected format in run_chat
-    def preprocess_input(crypto, span):
-        # Extract the crypto code from the dropdown display value
-        crypto_code = next((code for code, display in crypto_emojis.items() if display == crypto), "btc")
-        # Extract the span code from the dropdown display value
-        span_code = next((code for code, display in time_emojis.items() if display == span), "7d")
-        return crypto_code, span_code
-    
-    # Connect the button click - with proper function pipeline
-    def process_and_run(crypto_display, span_display):
-        # First preprocess the display values to get the codes
-        crypto_code, span_code = preprocess_input(crypto_display, span_display)
-        # Then run the chat with those codes and return the generator directly
-        # This is the key fix - we're directly returning the generator
-        return run_chat(crypto_code, span_code)
-        
     start_btn.click(
-        fn=process_and_run,  # Use named parameter for clarity
-        inputs=[crypto_dropdown, span_dropdown],
-        outputs=[chatbot, state],
-        api_name="start_analysis",  # Add API name for better traceability
-        show_progress="minimal"  # Add a progress indicator while processing
+        run_chat,
+        [crypto_dropdown, span_dropdown],
+        [chatbot, state],
     )
 
 demo.launch()

utils.py

@@ -1,63 +1,35 @@
 import os
 from dotenv import load_dotenv
+from langchain_aws import ChatBedrock
 import time
 import requests
-import logging
-import traceback
 
-# Configure logging
-logging.basicConfig(level=logging.INFO)
-logger = logging.getLogger(__name__)
-
-# Create a simple class to mimic LLM response format
-class MockResponse:
-    def __init__(self, content):
-        self.content = content
-
-# Load environment variables
 load_dotenv()
 
-# Get AWS credentials from environment
 AWS_REGION = os.getenv("AWS_REGION")
 AWS_ACCESS_KEY_ID = os.getenv("AWS_ACCESS_KEY_ID")
 AWS_SECRET_ACCESS_KEY = os.getenv("AWS_SECRET_ACCESS_KEY")
 AWS_CHAT_MODEL_ID = os.getenv("AWS_CHAT_MODEL_ID")
 
-# Set up LLM with proper error handling
-try:
-    # Only import and configure if credentials are available
-    if all([AWS_REGION, AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_CHAT_MODEL_ID]):
-        # Configure AWS credentials
-        os.environ["AWS_ACCESS_KEY_ID"] = AWS_ACCESS_KEY_ID
-        os.environ["AWS_SECRET_ACCESS_KEY"] = AWS_SECRET_ACCESS_KEY
-        os.environ["AWS_DEFAULT_REGION"] = AWS_REGION
-        
-        from langchain_aws import ChatBedrock
-        
-        llm_questioner = ChatBedrock(
-            model_id=AWS_CHAT_MODEL_ID,
-            temperature=0.5,
-        )
-        
-        llm_answerer = ChatBedrock(
-            model_id=AWS_CHAT_MODEL_ID,
-            temperature=0.5,
-        )
-        
-        llm_evaluator = ChatBedrock(
-            model_id=AWS_CHAT_MODEL_ID,
-            temperature=0.5,
-        )
-        
-        logger.info("Successfully initialized AWS Bedrock LLMs")
-        USE_MOCK_RESPONSES = False
-    else:
-        logger.warning("Missing AWS credentials, using mock responses")
-        USE_MOCK_RESPONSES = True
-except Exception as e:
-    logger.error(f"Error initializing AWS Bedrock: {str(e)}")
-    logger.error(traceback.format_exc())
-    USE_MOCK_RESPONSES = True
+# Configure AWS credentials
+os.environ["AWS_ACCESS_KEY_ID"] = AWS_ACCESS_KEY_ID
+os.environ["AWS_SECRET_ACCESS_KEY"] = AWS_SECRET_ACCESS_KEY
+os.environ["AWS_DEFAULT_REGION"] = AWS_REGION
+
+llm_questioner = ChatBedrock(
+    model_id=AWS_CHAT_MODEL_ID,
+    temperature=0.5,
+)
+
+llm_answerer = ChatBedrock(
+    model_id=AWS_CHAT_MODEL_ID,
+    temperature=0.5,
+)
+
+llm_evaluator = ChatBedrock(
+    model_id=AWS_CHAT_MODEL_ID,
+    temperature=0.5,
+)
 
 crypto_token = {
     "btc": "bitcoin",
@@ -72,182 +44,105 @@ crypto_token = {
 
 
 def get_crypto_data(crypto_name, data_span):
-    try:
-        # Default mock data in case of errors
-        mock_data = f"""
-            Crypto Name: {crypto_name}
-            Data TimeSpan: {data_span}
-            Twitter up votes: 65.8
-            Twitter down votes: 34.2
-            Value change: 3.25
-            Value change in currency: 3.25
-        """
-        
-        if crypto_name not in crypto_token:
-            logger.warning(f"Invalid crypto name: {crypto_name}, using mock data")
-            return mock_data
-            
-        try:
-            crypto_token_name = crypto_token[crypto_name]
-            r = requests.get(
-                f"https://api.coingecko.com/api/v3/coins/{crypto_token_name}?localization=false",
-                timeout=10  # Add timeout to prevent hanging requests
-            )
-            
-            if r.status_code != 200:
-                logger.warning(f"API request failed with status code {r.status_code}, using mock data")
-                return mock_data
-                
-            data = r.json()
-            
-            # Safely extract data with fallbacks
-            tw_up_votes = data.get("sentiment_votes_up_percentage", 65.8)
-            tw_down_votes = data.get("sentiment_votes_down_percentage", 34.2)
-            
-            market_data = data.get("market_data", {})
-            
-            # Handle different data span formats safely
-            price_change_key = f"price_change_percentage_{data_span}"
-            value_change = market_data.get(price_change_key, 3.25)
-            
-            price_change_currency_key = f"price_change_percentage_{data_span}_in_currency"
-            currency_data = market_data.get(price_change_currency_key, {})
-            value_change_in_currency = currency_data.get("usd", 3.25)
-            
-            return f"""
-                Crypto Name: {crypto_token_name}
-                Data TimeSpan: {data_span}
-                Twitter up votes: {tw_up_votes}
-                Twitter down votes: {tw_down_votes}
-                Value change: {value_change}
-                Value change in currency: {value_change_in_currency}
-            """
-            
-        except requests.exceptions.RequestException as e:
-            logger.error(f"Request error: {str(e)}")
-            return mock_data
-            
-    except Exception as e:
-        logger.error(f"Error in get_crypto_data: {str(e)}")
-        logger.error(traceback.format_exc())
-        return f"""
-            Crypto Name: {crypto_name}
-            Data TimeSpan: {data_span}
-            Twitter up votes: 65.8
-            Twitter down votes: 34.2
-            Value change: 3.25
-            Value change in currency: 3.25
-        """
+    crypto_name = crypto_token[crypto_name]
+    r = requests.get(
+        f"https://api.coingecko.com/api/v3/coins/{crypto_name}?localization=false"
+    )
+    data = r.json()
+    tw_up_votes = data["sentiment_votes_up_percentage"]
+    tw_down_votes = data["sentiment_votes_down_percentage"]
+    value_change = data["market_data"][f"price_change_percentage_{data_span}"]
+    value_change_in_currency = data["market_data"][
+        f"price_change_percentage_{data_span}_in_currency"
+    ]["usd"]
+    return f"""
+        Crypto Name: {crypto_name}
+        Data TimeSpan: {data_span}
+        Twitter up votes: {tw_up_votes}
+        Twitter down votes: {tw_down_votes}
+        Value change: {value_change}
+        Value change in currency: {value_change_in_currency}
+    """
 
 
 def get_question(previous_qa=""):
     """Generate a crypto question using the questioner LLM"""
-    try:
-        if USE_MOCK_RESPONSES:
-            # Return a mock response for testing
-            return MockResponse(content="What percentage of your investment portfolio would you allocate to cryptocurrencies?\n\nA) Less than 5%\nB) 5-15%\nC) 16-30%\nD) More than 30%")
-            
-        prompt = f"""
-        You are a questioner and expert in the field of Cryptocurrency and Investments. 
-        
-        # Instructions
-        - Your task is to ask question to the user based on Cryptocurrency and Investments.
-        - You need to ask multiple choice questions so that the user can select the correct answer.
-        - Make sure the question is clear and the options are distinct.
-        - Just generate the question and options, do not include any other text.
-        - Your question will directly be given to the user, so make sure no starting or ending texts are added.
-        - The reason of asking the question is to evaluate the user's risk profiling.
-        - We need to build a risk profiling model based on the answers.
-        - You need to generate a new multiple choice question for risk profiling.
-        
-        # Types of Questions
-        - Risk Tolerance
-        - Investment Goals
-        - Time Horizon
-        - Risk Appetite
-        - Investment Sentiment
-        - Details about previous investments
-        - Percentage of investment in Cryptocurrency
-        
-        # previous questions with answers
-        {previous_qa}
-        
-        
-        """
-        return llm_questioner.invoke(prompt)
-        
-    except Exception as e:
-        logger.error(f"Error in get_question: {str(e)}")
-        logger.error(traceback.format_exc())
-        return MockResponse(content="What is your investment time horizon for cryptocurrency investments?\n\nA) Less than 1 year (short-term)\nB) 1-3 years (medium-term)\nC) 3-5 years (long-term)\nD) More than 5 years (very long-term)")
+    prompt = f"""
+    You are a questioner and expert in the field of Cryptocurrency and Investments. 
+    
+    # Instructions
+    - Your task is to ask question to the user based on Cryptocurrency and Investments.
+    - You need to ask multiple choice questions so that the user can select the correct answer.
+    - Make sure the question is clear and the options are distinct.
+    - Just generate the question and options, do not include any other text.
+    - Your question will directly be given to the user, so make sure no starting or ending texts are added.
+    - The reason of asking the question is to evaluate the user's risk profiling.
+    - We need to build a risk profiling model based on the answers.
+    - You need to generate a new multiple choice question for risk profiling.
+    
+    # Types of Questions
+    - Risk Tolerance
+    - Investment Goals
+    - Time Horizon
+    - Risk Appetite
+    - Investment Sentiment
+    - Details about previous investments
+    - Percentage of investment in Cryptocurrency
+    
+    # previous questions with answers
+    {previous_qa}
+    
+    
+    """
+    return llm_questioner.invoke(prompt)
 
 
 def get_answer(question, previous_qa, crypto_data):
     """Generate an answer explanation using the answerer LLM"""
-    try:
-        if USE_MOCK_RESPONSES:
-            # Return a mock response for testing
-            return MockResponse(content="Based on the data provided for this cryptocurrency and market trends, I would select option B as the most appropriate answer.")
-            
-        prompt = f"""
-        You are an expert in the field of Cryptocurrency and Investments.
-        
-        # Instructions
-        - Your task is to answer the question based on Cryptocurrency and Investments.
-        - The question will be the multiple choice question.
-        - Just select the correct answer from the options.
-        - Do not include any other text.
-        - Your answer should be consistent with previous answers and build upon them.
-        - Consider the user's previous responses to maintain a coherent risk profile.
-        
-        # Notes
-        - You need to give answers by looking at the following crypto data
-        
-        # Crypto Data
-        {crypto_data}
-        
-        # Previous Q&A History
-        {previous_qa}
-        
-        # Current Question with options
-        {question}
-        """
-        return llm_answerer.invoke(prompt)
-        
-    except Exception as e:
-        logger.error(f"Error in get_answer: {str(e)}")
-        logger.error(traceback.format_exc())
-        return MockResponse(content="Based on the current market trends and volatility, option C appears to be the most balanced choice for this cryptocurrency.")
+    prompt = f"""
+    You are an expert in the field of Cryptocurrency and Investments.
+    
+    # Instructions
+    - Your task is to answer the question based on Cryptocurrency and Investments.
+    - The question will be the multiple choice question.
+    - Just select the correct answer from the options.
+    - Do not include any other text.
+    - Your answer should be consistent with previous answers and build upon them.
+    - Consider the user's previous responses to maintain a coherent risk profile.
+    
+    # Notes
+    - You need to give answers by looking at the following crypto data
+    
+    # Crypto Data
+    {crypto_data}
+    
+    # Previous Q&A History
+    {previous_qa}
+    
+    # Current Question with options
+    {question}
+    """
+    return llm_answerer.invoke(prompt)
 
 
 def evaluate_expertise(qa_history):
     """Evaluate the user's crypto expertise level based on Q&A history"""
-    try:
-        if USE_MOCK_RESPONSES:
-            # Return a mock response for testing
-            return MockResponse(content="Risk Assessment Summary:\n\n1. Based on the responses, the investor demonstrates a moderate risk tolerance with a preference for balanced investment approaches.\n\n2. The time horizon indicated is medium-term, suggesting some patience for market cycles but not extremely long-term holding.\n\n3. The answers show familiarity with cryptocurrency market dynamics but not deep technical knowledge.\n\nTranche Recommendation: MEZZANINE\n\nThis middle-tier tranche is appropriate for investors who want some exposure to cryptocurrency growth potential while maintaining a reasonable risk profile.")
-            
-        prompt = f"""
-        You are an expert evaluator in the field of Cryptocurrency and Investments.
-        
-        # Instructions
-        - Analyze the following Q&A history to determine the user's risk profiling.
-        - Consider the complexity of questions asked and the quality of answers provided
-        - Include 2-3 specific observations about their risk profiling
-        - Just generate the evaluation, do not include any other text.
-        - You have to decide the tranche based on user's answer.
-        - Tranche can be Senior, Mezzanine and Junior.
-        - You need to give a final verdict based on the user's answer.
-        
-        # Q&A History
-        {qa_history}
-        """
-        return llm_evaluator.invoke(prompt)
-        
-    except Exception as e:
-        logger.error(f"Error in evaluate_expertise: {str(e)}")
-        logger.error(traceback.format_exc())
-        return MockResponse(content="Risk Assessment Summary:\n\n1. The investor shows a balanced approach to cryptocurrency investment, neither extremely conservative nor highly aggressive.\n\n2. There's an understanding of market volatility and a willingness to accept moderate risk for potential returns.\n\n3. Investment horizon appears to be medium to long-term, indicating patience with market fluctuations.\n\nTranche Recommendation: MEZZANINE\n\nThis investor fits well in the middle tranche, balancing potential growth with reasonable risk management.")
+    prompt = f"""
+    You are an expert evaluator in the field of Cryptocurrency and Investments.
+    
+    # Instructions
+    - Analyze the following Q&A history to determine the user's risk profiling.
+    - Consider the complexity of questions asked and the quality of answers provided
+    - Include 2-3 specific observations about their risk profiling
+    - Just generate the evaluation, do not include any other text.
+    - You have to decide the tranche based on user's answer.
+    - Tranche can be Senior, Mezzanine and Junior.
+    - You need to give a final verdict based on the user's answer.
+    
+    # Q&A History
+    {qa_history}
+    """
+    return llm_evaluator.invoke(prompt)
 
 
 def main():