Full source code
Copy the complete source code below. Install the required packages listed in the comments at the top.
# Agentic stock research system for Indian markets
# Uses LangChain agents + yfinance + news APIs
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
import yfinance as yf
import requests
import pandas as pd
import ta # Technical analysis library
# ── DEFINE RESEARCH TOOLS ────────────────────────────────────────────
@tool
def get_fundamentals(symbol: str) -> dict:
"""Get fundamental data for an NSE stock"""
ticker = yf.Ticker(f"{symbol}.NS")
info = ticker.info
return {
"company": info.get("longName"),
"sector": info.get("sector"),
"industry": info.get("industry"),
"market_cap_cr": round(info.get("marketCap", 0) / 1e7, 2),
"pe_ratio": info.get("trailingPE"),
"pb_ratio": info.get("priceToBook"),
"roe": info.get("returnOnEquity"),
"debt_to_equity": info.get("debtToEquity"),
"revenue_growth": info.get("revenueGrowth"),
"earnings_growth": info.get("earningsGrowth"),
"dividend_yield": info.get("dividendYield"),
"promoter_holding": "Check NSE/BSE directly", # Not in yfinance
"current_price": info.get("currentPrice"),
"target_price": info.get("targetMeanPrice"),
"analyst_recommendation": info.get("recommendationKey")
}
@tool
def get_technical_signals(symbol: str) -> dict:
"""Calculate technical indicators and signals"""
ticker = yf.Ticker(f"{symbol}.NS")
df = ticker.history(period="6mo", interval="1d")
if df.empty:
return {"error": "No data found"}
# Calculate indicators
df['ema20'] = ta.trend.ema_indicator(df['Close'], window=20)
df['ema50'] = ta.trend.ema_indicator(df['Close'], window=50)
df['rsi'] = ta.momentum.rsi(df['Close'], window=14)
df['macd'] = ta.trend.macd_diff(df['Close'])
# Bollinger Bands
bb = ta.volatility.BollingerBands(df['Close'])
df['bb_upper'] = bb.bollinger_hband()
df['bb_lower'] = bb.bollinger_lband()
latest = df.iloc[-1]
# Generate signals
signals = []
if latest['Close'] > latest['ema20'] > latest['ema50']:
signals.append("✅ Bullish: Price above EMA20 & EMA50")
if latest['rsi'] < 40:
signals.append("✅ RSI Oversold — potential buy zone")
elif latest['rsi'] > 70:
signals.append("⚠️ RSI Overbought — potential sell zone")
if latest['macd'] > 0:
signals.append("✅ MACD positive — bullish momentum")
# 52-week analysis
high_52w = df['High'].max()
low_52w = df['Low'].min()
position_pct = (latest['Close'] - low_52w) / (high_52w - low_52w) * 100
return {
"current_price": round(latest['Close'], 2),
"ema20": round(latest['ema20'], 2),
"ema50": round(latest['ema50'], 2),
"rsi": round(latest['rsi'], 1),
"52w_high": round(high_52w, 2),
"52w_low": round(low_52w, 2),
"position_in_52w_range": f"{round(position_pct, 1)}%",
"signals": signals,
"trend": "Uptrend" if latest['ema20'] > latest['ema50'] else "Downtrend"
}
@tool
def get_recent_news(symbol: str) -> str:
"""Get and analyse recent news for a stock"""
# Using NewsAPI or similar - replace with your API key
url = f"https://newsapi.org/v2/everything?q={symbol}+stock+NSE&language=en&sortBy=publishedAt&pageSize=5&apiKey={NEWS_API_KEY}"
try:
response = requests.get(url)
articles = response.json().get("articles", [])
news_text = ""
for article in articles[:5]:
news_text += f"- {article['title']} ({article['publishedAt'][:10]})
"
return news_text if news_text else "No recent news found"
except:
return "Could not fetch news"
@tool
def compare_with_peers(symbol: str) -> dict:
"""Compare stock with sector peers"""
# Map common Indian stocks to their peers
peer_map = {
"TCS": ["INFY", "WIPRO", "HCLTECH", "TECHM"],
"RELIANCE": ["ONGC", "IOC", "BPCL"],
"HDFC": ["ICICIBANK", "KOTAKBANK", "AXISBANK", "SBIN"],
"BAJFINANCE": ["HDFC", "ICICIGI", "SBILIFE"],
}
peers = peer_map.get(symbol.upper(), [])
if not peers:
return {"message": "Peer comparison not available for this stock"}
results = []
for peer in peers[:3]:
try:
info = yf.Ticker(f"{peer}.NS").info
results.append({
"symbol": peer,
"pe": info.get("trailingPE"),
"pb": info.get("priceToBook"),
"roe": info.get("returnOnEquity")
})
except:
pass
return {"peers": results}
# ── BUILD THE ANALYST AGENT ──────────────────────────────────────────
def create_stock_analyst():
llm = ChatOpenAI(model="gpt-4o", temperature=0)
prompt = ChatPromptTemplate.from_messages([
("system", """You are a senior equity analyst specialising in Indian stock markets (NSE/BSE).
Your analysis framework:
1. FUNDAMENTAL ANALYSIS — PE ratio vs sector, ROE, debt levels, growth
2. TECHNICAL ANALYSIS — trend, momentum, support/resistance
3. NEWS SENTIMENT — recent developments, management changes, regulatory news
4. PEER COMPARISON — is the stock cheap or expensive vs peers?
5. FINAL RECOMMENDATION — Buy/Hold/Sell with price target and risk factors
Always provide:
- Investment horizon (short-term 1-3 months / medium-term 6-12 months)
- Risk level (Low/Medium/High)
- Key risks that could invalidate the thesis
- DISCLAIMER: This is for educational purposes only, not financial advice
Think through each step before giving your recommendation."""),
("human", "{input}"),
MessagesPlaceholder("agent_scratchpad")
])
tools = [get_fundamentals, get_technical_signals, get_recent_news, compare_with_peers]
agent = create_openai_tools_agent(llm, tools, prompt)
return AgentExecutor(agent=agent, tools=tools, verbose=True, max_iterations=8)
# ── RUN THE ANALYST ──────────────────────────────────────────────────
if __name__ == "__main__":
analyst = create_stock_analyst()
# Analyse any NSE stock
result = analyst.invoke({
"input": "Analyse TITAN Company Limited (NSE: TITAN) and give me a buy/hold/sell recommendation with target price."
})
print("\n" + "="*60)
print("STOCK ANALYSIS REPORT")
print("="*60)
print(result["output"])