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)
The agent automatically decides which tools to use based on the query
Orchestrates multiple data sources (Yahoo Finance API calls) without human intervention
Decision-Making Capability
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
Self-Monitoring & Error Recovery
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
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""") ])
Maintains focus on financial analysis objectives
Enforces structured output requirements
Adaptive Reasoning
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:
| 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
Not just responding to inputs, but actively:
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 pricesmin_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 changesgain = delta.where(delta > 0, 0)- Positive changesloss = -delta.where(delta < 0, 0)- Absolute negative changesavg_gain = gain.rolling(14).mean()- 14-day average gainavg_loss = loss.rolling(14).mean()- 14-day average lossrs = avg_gain / avg_loss- Relative StrengthRSI = 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:
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.
Multi-Step Reasoning
# Agent workflow: 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.
No comments:
Post a Comment