app.py

import gradio as gr
import requests
from transformers import pipeline
import json
import time
from typing import Dict, List
import os

class FinanceAgent:
    def __init__(self):
        # Initialize the summarization pipeline
        try:
            self.summarizer = pipeline(
                "summarization", 
                model="facebook/bart-large-cnn",
                max_length=512,
                min_length=100
            )
        except Exception as e:
            print(f"Error loading summarization model: {e}")
            self.summarizer = None
        
    def search_web(self, query: str, num_results: int = 5) -> List[Dict]:
        """
        Search the web using DuckDuckGo API (free alternative)
        """
        try:
            # Using DuckDuckGo Instant Answer API
            url = "https://api.duckduckgo.com/"
            params = {
                'q': f"{query} finance financial",
                'format': 'json',
                'no_html': '1',
                'skip_disambig': '1'
            }
            
            response = requests.get(url, params=params, timeout=10)
            data = response.json()
            
            results = []
            
            # Extract abstract if available
            if data.get('Abstract'):
                results.append({
                    'title': data.get('Heading', 'Financial Information'),
                    'content': data.get('Abstract', ''),
                    'url': data.get('AbstractURL', '')
                })
            
            # Extract related topics
            for topic in data.get('RelatedTopics', [])[:num_results-1]:
                if isinstance(topic, dict) and topic.get('Text'):
                    results.append({
                        'title': topic.get('Text', '')[:100] + '...',
                        'content': topic.get('Text', ''),
                        'url': topic.get('FirstURL', '')
                    })
            
            return results
            
        except Exception as e:
            print(f"Search error: {e}")
            return [{'title': 'Search Error', 'content': f'Unable to search: {str(e)}', 'url': ''}]
    
    def get_financial_context(self, topic: str) -> str:
        """
        Get basic financial context for common topics
        """
        financial_contexts = {
            'stock': 'Stocks represent ownership shares in publicly traded companies. Stock prices fluctuate based on company performance, market conditions, and investor sentiment.',
            'bond': 'Bonds are debt securities where investors lend money to entities for a defined period at a fixed interest rate.',
            'cryptocurrency': 'Cryptocurrencies are digital assets that use cryptography for security and operate on decentralized networks.',
            'inflation': 'Inflation is the rate at which the general level of prices for goods and services rises, eroding purchasing power.',
            'recession': 'A recession is a significant decline in economic activity across the economy lasting more than a few months.',
            'fed': 'The Federal Reserve is the central banking system of the United States, responsible for monetary policy.',
            'gdp': 'Gross Domestic Product (GDP) is the total monetary value of all goods and services produced within a country.',
            'market': 'Financial markets are platforms where buyers and sellers trade financial securities, commodities, and other assets.'
        }
        
        topic_lower = topic.lower()
        for key, context in financial_contexts.items():
            if key in topic_lower:
                return context
        
        return "This is a financial topic that may involve various economic, market, or investment-related concepts."
    
    def summarize_content(self, content: str, max_length: int = 300) -> str:
        """
        Summarize content using the transformer model
        """
        if not self.summarizer or not content.strip():
            return "Unable to generate summary."
        
        try:
            # Truncate content if too long
            if len(content) > 1000:
                content = content[:1000] + "..."
            
            summary = self.summarizer(content, max_length=max_length, min_length=50, do_sample=False)
            return summary[0]['summary_text']
        except Exception as e:
            print(f"Summarization error: {e}")
            return f"Summary unavailable: {str(e)}"
    
    def process_finance_query(self, prompt: str) -> str:
        """
        Main function to process finance queries
        """
        if not prompt.strip():
            return "Please enter a financial topic or question."
        
        # Add progress indicator
        progress_msg = f"🔍 Searching for information about: {prompt}\n\n"
        
        # Search for relevant information
        search_results = self.search_web(prompt)
        
        if not search_results:
            return progress_msg + "❌ No search results found. Please try a different query."
        
        # Combine search results
        combined_content = ""
        sources = []
        
        for i, result in enumerate(search_results[:3], 1):  # Limit to top 3 results
            if result.get('content'):
                combined_content += f"\n{result['content']}\n"
                if result.get('url'):
                    sources.append(f"{i}. {result['url']}")
        
        # Get financial context
        context = self.get_financial_context(prompt)
        
        # Create comprehensive content for summarization
        full_content = f"{context}\n\nCurrent Information:\n{combined_content}"
        
        # Generate summary
        summary = self.summarize_content(full_content)
        
        # Format final response
        response = f"""## 📊 Financial Summary: {prompt.title()}

### 🎯 Key Points:
{summary}

### 📋 Context:
{context}

### 🔗 Sources:
"""
        
        if sources:
            for source in sources:
                response += f"\n{source}"
        else:
            response += "\nInformation compiled from web search results."
        
        response += f"\n\n⏰ *Generated on: {time.strftime('%Y-%m-%d %H:%M:%S')}*"
        
        return response

# Initialize the agent
agent = FinanceAgent()

def finance_chat_interface(message, history):
    """
    Gradio chat interface function
    """
    try:
        response = agent.process_finance_query(message)
        return response
    except Exception as e:
        return f"❌ Error processing your request: {str(e)}\n\nPlease try again with a different query."

# Create Gradio interface
def create_interface():
    with gr.Blocks(title="Financial AI Agent", theme=gr.themes.Soft()) as demo:
        gr.Markdown("""
        # 🏦 Financial AI Agent
        
        Ask me about any financial topic and I'll search the web for current information and provide you with a comprehensive summary.
        
        **Examples:**
        - "What is the current state of cryptocurrency market?"
        - "Explain inflation and its impact on the economy"
        - "Tell me about recent stock market trends"
        - "What are bonds and how do they work?"
        """)
        
        chatbot = gr.Chatbot(
            height=500,
            placeholder="Financial AI Agent is ready to help with your finance questions!"
        )
        
        msg = gr.Textbox(
            label="Ask a financial question",
            placeholder="e.g., What is inflation and how does it affect the stock market?",
            lines=2
        )
        
        clear = gr.Button("Clear Chat")
        
        def user(user_message, history):
            return "", history + [[user_message, None]]
        
        def bot(history):
            if history and history[-1][1] is None:
                user_message = history[-1][0]
                bot_message = finance_chat_interface(user_message, history[:-1])
                history[-1][1] = bot_message
            return history
        
        msg.submit(user, [msg, chatbot], [msg, chatbot], queue=False).then(
            bot, chatbot, chatbot
        )
        
        clear.click(lambda: None, None, chatbot, queue=False)
        
        gr.Markdown("""
        ### 📝 Notes:
        - This agent searches the web for current financial information
        - Summaries are generated using AI and should be verified with official sources
        - For investment decisions, always consult with qualified financial advisors
        """)
    
    return demo

if __name__ == "__main__":
    demo = create_interface()
    demo.launch(share=True)