Spaces:
Sleeping
Sleeping
| 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}) |