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:
Autonomous Tool Usage
tools = [get_live_stock_data, get_historical_data, get_fundamental_analysis]
agent = create_tool_calling_agent(groq_chat, tools, prompt)
Decision-Making Capability
def validate_stock(symbol: str) -> bool:
return all([...])Self-Monitoring & Error Recovery
except Exception as e:
logger.error(f"Historical data error: {str(e)}")
return f"Failed to fetch history: {str(e)}"Goal-Oriented Behavior
prompt = ChatPromptTemplate.from_messages([
("system", """1. Always reference tool data
2. Highlight data limitations
3. Compare to sector averages
4. Format with markdown tables""")
])Adaptive Reasoning
def try_convert(value, target_type, default=None):
return target_type(value) if value not in [None, 'N/A'] else default
Agentic Architecture Breakdown:
| Component | Agentic Feature | Implementation Example |
|---|
| Perception | Market data ingestion | yfinance API calls |
| Planning | Analysis strategy | Prompt template guidelines |
| Action | Tool execution | @tool decorated functions |
| Memory | Context preservation | AgentExecutor's state management |
| Learning | (Limited) Error adaptation | Error feedback in logging |
Why This is Agentic vs Regular AI:
Proactive Data Gathering
hist['50MA'] = hist['Close'].rolling(window=50, min_periods=1).mean()
hist['RSI'] = 100 - (100 / (1 + rs))
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:
| Date | Close | 50MA |
|---|
| 2023-11-01 | 150 | 148.2 |
| 2023-11-02 | 152 | 148.5 |
| ... | ... | ... |
| 2024-01-15 | 160 | 155.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:
delta = hist['Close'].diff(1) - Daily price changes
gain = delta.where(delta > 0, 0) - Positive changes
loss = -delta.where(delta < 0, 0) - Absolute negative changes
avg_gain = gain.rolling(14).mean() - 14-day average gain
avg_loss = loss.rolling(14).mean() - 14-day average loss
rs = avg_gain / avg_loss - Relative Strength
RSI = 100 - (100 / (1 + rs)) - Final formula
Example Output:
| Date | Close | RSI |
|---|
| 2023-11-01 | 150 | 58.7 |
| 2023-11-02 | 152 | 62.1 |
| 2023-11-03 | 149 | 54.3 |
Why This Matters for Agentic AI:
Automated Feature Engineering: The agent creates technical indicators without human intervention
Time-Series Analysis: Enables pattern recognition in price data
Decision Support: Provides quantitative metrics for buy/sell recommendations
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:
You can verify these calculations against platforms like TradingView or Yahoo Finance to ensure accuracy.
Multi-Step Reasoning
1. Validate symbol → 2. Fetch live data →
3. Get historical trends → 4. Analyze fundamentals →
5. Synthesize report
Context-Aware Decisions
"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:
True Agency Would Require
Market impact prediction models
Portfolio optimization suggestions
News sentiment integration
Adaptive learning from past analyses
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.
