Upload app.py

app.py

@@ -0,0 +1,363 @@
+import streamlit as st
+import os
+import json
+import requests
+from google import genai
+from google.genai import types
+from dotenv import load_dotenv
+
+load_dotenv()
+
+# Set page configuration
+st.set_page_config(
+    page_title="Flaix - Financial Assistant",
+    page_icon="💰",
+    layout="centered"
+)
+
+# Initialize Gemini client
+api_key = st.secrets["GOOGLE_API_KEY"]
+client = genai.Client(api_key=api_key)
+
+# Indian Stock Market API base configuration
+INDIAN_API_KEY = st.secrets["FINANCE_KEY"]
+INDIAN_API_BASE_URL = "https://stock.indianapi.in"
+
+# Define API endpoints and their parameters
+API_ENDPOINTS = {
+    "get_stock_details": {
+        "endpoint": "/stock",
+        "required_params": ["stock_name"],
+        "param_mapping": {"stock_name": "name"},
+        "description": "Get details for a specific stock"
+    },
+    "get_trending_stocks": {
+        "endpoint": "/trending",
+        "required_params": [],
+        "param_mapping": {},
+        "description": "Get trending stocks in the market"
+    },
+    "get_market_news": {
+        "endpoint": "/news",
+        "required_params": [],
+        "param_mapping": {},
+        "description": "Get latest stock market news"
+    },
+    "get_mutual_funds": {
+        "endpoint": "/mutual_funds",
+        "required_params": [],
+        "param_mapping": {},
+        "description": "Get mutual funds data"
+    },
+    "get_ipo_data": {
+        "endpoint": "/ipo",
+        "required_params": [],
+        "param_mapping": {},
+        "description": "Get IPO data"
+    },
+    "get_bse_most_active": {
+        "endpoint": "/BSE_most_active",
+        "required_params": [],
+        "param_mapping": {},
+        "description": "Get BSE most active stocks"
+    },
+    "get_nse_most_active": {
+        "endpoint": "/NSE_most_active",
+        "required_params": [],
+        "param_mapping": {},
+        "description": "Get NSE most active stocks"
+    },
+    "get_historical_data": {
+        "endpoint": "/historical_data",
+        "required_params": ["stock_name"],
+        "optional_params": ["period"],
+        "default_values": {"period": "1m", "filter": "default"},
+        "param_mapping": {},
+        "description": "Get historical data for a stock"
+    }
+}
+
+# Unified API call function
+def call_indian_api(endpoint, params=None):
+    """
+    Generic function to call the Indian Stock Market API
+    
+    Args:
+        endpoint: API endpoint suffix (e.g., '/stock', '/trending')
+        params: Optional parameters for the API call
+        
+    Returns:
+        JSON response from the API
+    """
+    url = f"{INDIAN_API_BASE_URL}{endpoint}"
+    headers = {"X-Api-Key": INDIAN_API_KEY}
+    
+    try:
+        response = requests.get(url, headers=headers, params=params)
+        return response.json()
+    except Exception as e:
+        return {"error": str(e)}
+
+# Function to call API by name
+def call_api_by_name(api_name, **kwargs):
+    """
+    Call an API by its name from the API_ENDPOINTS dictionary
+    
+    Args:
+        api_name: Name of the API to call (key in API_ENDPOINTS)
+        **kwargs: Parameters to pass to the API
+        
+    Returns:
+        JSON response from the API
+    """
+    if api_name not in API_ENDPOINTS:
+        return {"error": f"Unknown API: {api_name}"}
+    
+    api_info = API_ENDPOINTS[api_name]
+    endpoint = api_info["endpoint"]
+    
+    # Check required parameters
+    for param in api_info.get("required_params", []):
+        if param not in kwargs:
+            return {"error": f"Missing required parameter: {param}"}
+    
+    # Apply parameter mapping
+    mapped_params = {}
+    for param, value in kwargs.items():
+        mapped_name = api_info.get("param_mapping", {}).get(param, param)
+        mapped_params[mapped_name] = value
+    
+    # Apply default values
+    for param, value in api_info.get("default_values", {}).items():
+        if param not in mapped_params:
+            mapped_params[param] = value
+    
+    return call_indian_api(endpoint, mapped_params)
+
+# Improved orchestrator function
+def orchestrator(query):
+    """
+    Determines if the query requires market data and which API to call
+    Returns: (needs_api, api_function, params)
+    """
+    # Create a more precise prompt for the orchestrator
+    orchestrator_prompt = """
+    You are an orchestrator for a financial assistant specialized in Indian markets. Your job is to analyze user queries and determine if they need real-time market data.
+
+    IMPORTANT: Be very precise in your analysis. Only return TRUE for "needs_api" when the query EXPLICITLY asks for current market data, stock prices, or listings.
+
+    Examples where needs_api should be TRUE:
+    - "Show me the most active stocks on NSE today" → get_nse_most_active
+    - "What is the current price of Reliance?" → get_stock_details with stock_name="Reliance"
+    - "Tell me about trending stocks" → get_trending_stocks
+    - "What are the latest IPOs?" → get_ipo_data
+
+    Examples where needs_api should be FALSE:
+    - "What is compound interest?"
+    - "How should I start investing?"
+    - "What are the tax benefits of PPF?"
+    - "Explain mutual funds to me"
+
+    Available API functions:
+    - get_stock_details(stock_name): Get details for a specific stock
+    - get_trending_stocks(): Get trending stocks in the market
+    - get_market_news(): Get latest stock market news
+    - get_mutual_funds(): Get mutual funds data
+    - get_ipo_data(): Get IPO data
+    - get_bse_most_active(): Get BSE most active stocks
+    - get_nse_most_active(): Get NSE most active stocks
+    - get_historical_data(stock_name, period="1m"): Get historical data for a stock
+
+    User query: """ + query + """
+
+    Respond in JSON format with the following structure:
+    {
+        "needs_api": true/false,
+        "function": "function_name_if_needed",
+        "params": {
+            "param1": "value1",
+            "param2": "value2"
+        }
+    }
+    """
+        
+    # Create content for the orchestrator
+    contents = [
+        types.Content(
+            role="user",
+            parts=[
+                types.Part.from_text(text=orchestrator_prompt)
+            ],
+        ),
+    ]
+    
+    # Configure generation parameters
+    generate_content_config = types.GenerateContentConfig(
+        temperature=0.2,
+        top_p=0.95,
+        top_k=40,
+        max_output_tokens=500,
+        response_mime_type="text/plain",
+    )
+    
+    # Generate content
+    response = client.models.generate_content(
+        model="gemini-1.5-flash",
+        contents=contents,
+        config=generate_content_config,
+    )
+    
+    # Parse the response
+    try:
+        decision_text = response.text
+        # Extract JSON from the response (it might be wrapped in markdown code blocks)
+        if "```json" in decision_text:
+            json_str = decision_text.split("```json")[1].split("```")[0].strip()
+        elif "```" in decision_text:
+            json_str = decision_text.split("```")[1].strip()
+        else:
+            json_str = decision_text.strip()
+        
+        decision = json.loads(json_str)
+        return decision
+    except Exception as e:
+        print(f"Error parsing orchestrator response: {e}")
+        return {"needs_api": False}
+
+# Language setting
+
+# Financial assistant system prompt
+SYSTEM_PROMPT = f"""You are Flaix, a helpful and knowledgeable financial assistant designed specifically for Indian users. Your purpose is to improve financial literacy and provide guidance on investments in the Indian market.
+
+Key responsibilities:
+1. Explain financial concepts in simple, easy-to-understand language
+2. Provide information about different investment options available in India (stocks, mutual funds, bonds, PPF, FDs, etc.)
+3. Help users understand investment risks and returns
+4. Explain tax implications of different investments in the Indian context
+5. Guide users on how to start investing based on their goals and risk tolerance
+6. Answer questions about market trends and financial news in India
+"""
+
+# Initialize session state for chat history
+if "messages" not in st.session_state:
+    st.session_state.messages = [
+        {"role": "user", "content": SYSTEM_PROMPT},
+        {"role": "model", "content": "Hello! I am Flaix, your financial assistant. You can ask me about investments, financial planning, or any other financial topic."}
+    ]
+
+# App title and description
+st.title("Flaix - Your Financial Assistant")
+st.markdown("Ask any questions about investing, financial planning, or the Indian financial market.")
+
+# Display chat messages
+for message in st.session_state.messages:
+    if message["role"] == "user" and message["content"] != SYSTEM_PROMPT:
+        with st.chat_message("user"):
+            st.write(message["content"])
+    elif message["role"] == "model":
+        with st.chat_message("assistant"):
+            st.write(message["content"])
+
+# Chat input
+if prompt := st.chat_input("Ask me anything about finance or investing..."):
+    # Add user message to chat history
+    st.session_state.messages.append({"role": "user", "content": prompt})
+    
+    # Display user message
+    with st.chat_message("user"):
+        st.write(prompt)
+    
+    # Display assistant response
+    with st.chat_message("assistant"):
+        message_placeholder = st.empty()
+        full_response = ""
+        
+        try:
+            # First, use the orchestrator to determine if we need to call an API
+            decision = orchestrator(prompt)
+            
+            # If we need to call an API, do so and add the result to the context
+            api_context = ""
+            if decision.get("needs_api", False):
+                function_name = decision.get("function", "")
+                params = decision.get("params", {})
+                
+                message_placeholder.write("Fetching real-time market data...")
+                
+                if function_name in API_ENDPOINTS:
+                    api_result = call_api_by_name(function_name, **params)
+                    api_context = f"\nHere is the real-time market data from the Indian Stock Market API:\n{json.dumps(api_result, indent=2)}\n\nPlease use this data to provide an informative response to the user's query."
+            
+            
+            # Prepare the user query with API context if available
+            user_query = prompt
+            if api_context:
+                user_query = f"{prompt}\n\n[SYSTEM NOTE: {api_context}]"
+            
+            # Prepare the system message
+            system_message = SYSTEM_PROMPT
+            if len(st.session_state.messages) > 2:  # If we have conversation history
+                # Extract previous conversation for context
+                conversation_history = ""
+                for i in range(1, min(5, len(st.session_state.messages) - 1)):  # Get up to 5 previous exchanges
+                    if st.session_state.messages[i]["role"] == "user" and st.session_state.messages[i]["content"] != SYSTEM_PROMPT:
+                        conversation_history += f"User: {st.session_state.messages[i]['content']}\n"
+                    elif st.session_state.messages[i]["role"] == "model":
+                        conversation_history += f"Assistant: {st.session_state.messages[i]['content']}\n"
+                
+                system_message += f"\n\nPrevious conversation:\n{conversation_history}"
+            
+            # Create content for the LLM
+            contents = [
+                types.Content(
+                    role="user",
+                    parts=[
+                        types.Part.from_text(text=system_message)
+                    ],
+                ),
+                types.Content(
+                    role="model",
+                    parts=[
+                        types.Part.from_text(text="I understand my role as Flaix, a financial assistant for Indian users. I'll provide helpful information about investing and financial planning in simple language.")
+                    ],
+                ),
+                types.Content(
+                    role="user",
+                    parts=[
+                        types.Part.from_text(text=user_query)
+                    ],
+                ),
+            ]
+            
+            # Configure generation parameters
+            generate_content_config = types.GenerateContentConfig(
+                temperature=0.7,
+                top_p=0.95,
+                top_k=40,
+                max_output_tokens=8192,
+                response_mime_type="text/plain",
+            )
+            
+            # Stream the response
+            response_stream = client.models.generate_content_stream(
+                model="gemini-1.5-flash",
+                contents=contents,
+                config=generate_content_config,
+            )
+            
+            # Process streaming response
+            for chunk in response_stream:
+                if hasattr(chunk, 'text'):
+                    full_response += chunk.text
+                    message_placeholder.write(full_response + "▌")
+            
+            # Final update without cursor
+            message_placeholder.write(full_response)
+        
+        except Exception as e:
+            st.error(f"Error: {str(e)}")
+            full_response = "I'm sorry, I encountered an error. Please try again later."
+            message_placeholder.write(full_response)
+        
+        # Add assistant response to chat history
+        st.session_state.messages.append({"role": "model", "content": full_response})