Below code is an example of Agent IA

from dotenv import load_dotenv

import yfinance as yf

from langchain_groq import ChatGroq

from langchain.agents import AgentExecutor, create_tool_calling_agent

from langchain_core.prompts import ChatPromptTemplate

from langchain_core.tools import tool

import pandas as pd

import logging


# Configure logging

logging.basicConfig(level=logging.INFO)

logger = logging.getLogger(__name__)


# Load environment variables

load_dotenv()


def format_table(df: pd.DataFrame) -> str:

    """Convert DataFrame to markdown table with safe formatting"""

    try:

        return df.fillna('N/A').to_markdown(floatfmt=".2f", index=False)

    except Exception as e:

        logger.error(f"Table formatting error: {str(e)}")

        return str(df)


def validate_stock(symbol: str) -> bool:

    """Enhanced stock validation with multiple checks"""

    try:

        stock = yf.Ticker(symbol)

        info = stock.info

        return all([

            not stock.history(period="1d").empty,

            info.get('currency') is not None,

            info.get('currentPrice') is not None

        ])

    except Exception as e:

        logger.error(f"Validation failed for {symbol}: {str(e)}")

        return False


@tool

def get_live_stock_data(symbol: str) -> str:

    """Get real-time stock data with enhanced reliability"""

    if not validate_stock(symbol):

        return f"Invalid symbol or data unavailable: {symbol}"

    

    try:

        stock = yf.Ticker(symbol)

        info = stock.info

        data = stock.history(period="1d")

        

        # Safely handle missing data

        current_price = data.iloc[-1].Close if not data.empty else info.get('currentPrice', 'N/A')

        market_cap = info.get('marketCap')

        

        stats = {

            "Currency": info.get('currency', 'USD'),

            "Current Price": current_price,

            "Open": info.get('open', 'N/A'),

            "Volume": f"{info.get('volume', 'N/A'):,}",

            "52W High": info.get('fiftyTwoWeekHigh', 'N/A'),

            "52W Low": info.get('fiftyTwoWeekLow', 'N/A'),

            "PE Ratio": info.get('trailingPE', 'N/A'),

            "Market Cap": (

                f"${market_cap/1e9:.2f}B" if market_cap and info.get('currency') == 'USD' else

                f"HK${market_cap/1e9:.2f}B" if market_cap and info.get('currency') == 'HKD' else

                f"₹{market_cap/1e7:.2f}Cr" if market_cap else 'N/A'

            )

        }

        

        return format_table(pd.DataFrame([stats]))

        

    except Exception as e:

        logger.error(f"Live data error: {str(e)}")

        return f"Failed to fetch live data: {str(e)}"


@tool

def get_historical_data(symbol: str, period: str = "1y") -> str:

    """Get historical data with fault-tolerant calculations"""

    if not validate_stock(symbol):

        return f"Invalid symbol: {symbol}"

    

    try:

        stock = yf.Ticker(symbol)

        hist = stock.history(period=period)

        

        if hist.empty:

            return "No historical data available"

        

        # Calculate with error handling

        hist['50MA'] = hist['Close'].rolling(window=50, min_periods=1).mean()

        hist['200MA'] = hist['Close'].rolling(window=200, min_periods=1).mean()

        

        # Safe RSI calculation

        delta = hist['Close'].diff(1)

        gain = delta.where(delta > 0, 0)

        loss = -delta.where(delta < 0, 0)

        avg_gain = gain.rolling(14).mean()

        avg_loss = loss.rolling(14).mean()

        rs = avg_gain / avg_loss

        hist['RSI'] = 100 - (100 / (1 + rs))

        

        return format_table(hist.tail(5)[['Close', '50MA', '200MA', 'RSI']])

        

    except Exception as e:

        logger.error(f"Historical data error: {str(e)}")

        return f"Failed to fetch history: {str(e)}"


@tool

def get_fundamental_analysis(symbol: str) -> str:

    """Robust fundamental analysis with type safety"""

    try:

        stock = yf.Ticker(symbol)

        info = stock.info

        

        metrics = {

            "ROE (%)": try_convert(info.get('returnOnEquity'), float, 0) * 100,

            "Debt/Equity": try_convert(info.get('debtToEquity'), float),

            "EPS": try_convert(info.get('trailingEps'), float),

            "Dividend Yield (%)": try_convert(info.get('dividendYield'), float, 0) * 100,

            "Beta": try_convert(info.get('beta'), float),

            "Institutional Ownership (%)": try_convert(info.get('heldPercentInstitutions'), float, 0) * 100

        }

        

        return format_table(pd.DataFrame([metrics]))

        

    except Exception as e:

        logger.error(f"Fundamental analysis error: {str(e)}")

        return f"Failed fundamental analysis: {str(e)}"


def try_convert(value, target_type, default=None):

    """Safe type conversion helper"""

    try:

        return target_type(value) if value not in [None, 'N/A'] else default

    except (ValueError, TypeError):

        return default


# Initialize Groq with error handling

try:

    groq_chat = ChatGroq(

        temperature=0.3,

        model_name="llama3-70b-8192",

        max_tokens=2000

    )

except Exception as e:

    logger.error(f"Groq initialization failed: {str(e)}")

    raise


# Create agent with comprehensive error handling

try:

    prompt = ChatPromptTemplate.from_messages([

        ("system", """You are Senior Financial Analyst. Use these guidelines:

         1. Always reference tool data

         2. Highlight data limitations

         3. Compare to sector averages

         4. Format with markdown tables"""),

        ("user", "{input}"),

        ("placeholder", "{agent_scratchpad}"),

    ])

    

    tools = [get_live_stock_data, get_historical_data, get_fundamental_analysis]

    agent = create_tool_calling_agent(groq_chat, tools, prompt)

    agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=False, handle_parsing_errors=True)

    

except Exception as e:

    logger.error(f"Agent setup failed: {str(e)}")

    raise


# Run analysis with full error protection

def analyze_stock(symbol: str):

    try:

        response = agent_executor.invoke({

            "input": f"""Analyze {symbol} stock with:

             - Current market data

             - Technical indicators

             - Financial health

             - Institutional activity

             Format with markdown tables"""

        })

        return response["output"]

    except Exception as e:

        return f"Analysis failed: {str(e)}"


# Example usage for Tencent (0700.HK)

if __name__ == "__main__":

    import sys

    symbol = sys.argv[1] if len(sys.argv) > 1 else "NVDA"

    print(f"\nAnalyzing {symbol}...\n")

    print(analyze_stock(symbol))


-------------------------------------------------------------

why it is an examaple of Agent AI


 This code is indeed a good example of agentic AI implementation, demonstrating several key characteristics of agentic systems. Let's break down why:

Key Agentic AI Features Present:

  1. Autonomous Tool Usage

    python
    Copy
    tools = [get_live_stock_data, get_historical_data, get_fundamental_analysis]
agent = create_tool_calling_agent(groq_chat, tools, prompt)

    • The agent automatically decides which tools to use based on the query

    • Orchestrates multiple data sources (Yahoo Finance API calls) without human intervention

  2. Decision-Making Capability

    python
    Copy
    def validate_stock(symbol: str) -> bool:
    # Complex validation logic checking multiple data points
    return all([...])

    • Makes decisions about data validity

    • Chooses appropriate error handling strategies

  3. Self-Monitoring & Error Recovery

    python
    Copy
    except Exception as e:
    logger.error(f"Historical data error: {str(e)}")
    return f"Failed to fetch history: {str(e)}"

    • Comprehensive error handling throughout

    • Automatic fallback mechanisms when data is unavailable

  4. Goal-Oriented Behavior

    python
    Copy
    prompt = ChatPromptTemplate.from_messages([
    ("system", """1. Always reference tool data
    2. Highlight data limitations
    3. Compare to sector averages
    4. Format with markdown tables""")
])

    • Maintains focus on financial analysis objectives

    • Enforces structured output requirements

  5. Adaptive Reasoning

    python
    Copy
    def try_convert(value, target_type, default=None):
    # Context-aware type conversion
    return target_type(value) if value not in [None, 'N/A'] else default

    • Adjusts data processing based on input types

    • Handles international stock formats and currencies

Agentic Architecture Breakdown:

ComponentAgentic FeatureImplementation Example
PerceptionMarket data ingestionyfinance API calls
PlanningAnalysis strategyPrompt template guidelines
ActionTool execution@tool decorated functions
MemoryContext preservationAgentExecutor's state management
Learning(Limited) Error adaptationError feedback in logging

Why This is Agentic vs Regular AI:

  1. Proactive Data Gathering

    • Not just responding to inputs, but actively:

    python
    Copy
    hist['50MA'] = hist['Close'].rolling(window=50, min_periods=1).mean()
hist['RSI'] = 100 - (100 / (1 + rs))  # Automatic technical analysis

What it does:

  • Calculates a rolling average of closing prices over 50 trading days

  • Creates a new column '50MA' in the DataFrame

  • window=50: Looks at the last 50 days' closing prices

  • min_periods=1: Starts calculating even if there's only 1 data point

  • .mean(): Averages the prices in the window

Example:

DateClose50MA
2023-11-01150148.2
2023-11-02152148.5
.........
2024-01-15160155.8


What it does:

  • Calculates the Relative Strength Index (RSI), a momentum indicator

  • Typically uses 14-period lookback (default in most implementations)

  • Measures speed/magnitude of recent price changes

  • Values range 0-100 (30=oversold, 70=overbought)

Key Components:

  1. delta = hist['Close'].diff(1) - Daily price changes

  2. gain = delta.where(delta > 0, 0) - Positive changes

  3. loss = -delta.where(delta < 0, 0) - Absolute negative changes

  4. avg_gain = gain.rolling(14).mean() - 14-day average gain

  5. avg_loss = loss.rolling(14).mean() - 14-day average loss

  6. rs = avg_gain / avg_loss - Relative Strength

  7. RSI = 100 - (100 / (1 + rs)) - Final formula

Example Output:

DateCloseRSI
2023-11-0115058.7
2023-11-0215262.1
2023-11-0314954.3

Why This Matters for Agentic AI:

  1. Automated Feature Engineering: The agent creates technical indicators without human intervention

  2. Time-Series Analysis: Enables pattern recognition in price data

  3. Decision Support: Provides quantitative metrics for buy/sell recommendations

  4. Adaptability: Adjusts calculations as new data arrives

Common Use Cases:

  • Identifying trend directions (using MA crossover strategies)

  • Spotting overbought/oversold conditions (RSI >70 or <30)

  • Generating trading signals

  • Risk management

Important Note: The min_periods=1 parameter allows the calculation to start immediately with available data, but meaningful analysis typically requires:

  • At least 50 days of data for reliable 50MA

  • At least 14 days of data for reliable RSI

You can verify these calculations against platforms like TradingView or Yahoo Finance to ensure accuracy. 

  1. Multi-Step Reasoning

    python
    Copy
    # Agent workflow:
1. Validate symbol → 2. Fetch live data → 
3. Get historical trends → 4. Analyze fundamentals → 
5. Synthesize report

  2. Context-Aware Decisions

    python
    Copy
    "Market Cap": (
    f"${market_cap/1e9:.2f}B" if USD else 
    f"HK${market_cap/1e9:.2f}B" if HKD else 
    f"₹{market_cap/1e7:.2f}Cr"
)

Limitations/Improvement Opportunities:

  1. True Agency Would Require

    • Market impact prediction models

    • Portfolio optimization suggestions

    • News sentiment integration

    • Adaptive learning from past analyses

  2. Current Constraints

    • Limited to predefined tools

    • No long-term memory between sessions

    • Reactive rather than proactive alerts

This implementation represents a strong foundation for agentic AI in financial analysis, particularly demonstrating tool orchestration and autonomous data processing capabilities. To make it fully agentic, you'd want to add features like goal-setting autonomy and continuous learning mechanisms.

No comments:

Post a Comment