I Built an AI Agent to Monitor My Stock Portfolio
Problem
I had 15 stocks in my portfolio. Every week I spent 3+ hours checking news, reviewing earnings dates, and deciding what to buy or sell. I still missed things.
Last quarter, I realized I had completely forgotten about an earnings announcement. The stock dropped 12% overnight. I could have sold before earnings if I had a reminder.
I needed an automated system. One that would:
- Track all my stocks in one place
- Alert me on significant price changes
- Remind me about upcoming earnings
- Give me clear buy/sell recommendations
My First Attempt: Simple yfinance Script
I started with a basic Python script using yfinance:
import yfinance as yf
stocks = ['AAPL', 'MSFT', 'GOOGL', 'AMZN', 'TSLA']
for symbol in stocks: ticker = yf.Ticker(symbol) price = ticker.info.get('currentPrice') change = ticker.info.get('regularMarketChangePercent') print(f"{symbol}: ${price:.2f} ({change:.2f}%)")This worked. But it was just a snapshot. I had to run it manually every time I wanted to check prices. No alerts. No historical context. No recommendations.
Lesson: A data fetching script is not a monitoring system.
My Second Attempt: Adding Alerts
I added alert thresholds:
import yfinance as yfimport smtplibfrom email.mime.text import MIMEText
ALERT_THRESHOLD = 5.0 # 5% price change
def check_alerts(portfolio): alerts = [] for symbol in portfolio: ticker = yf.Ticker(symbol) change = ticker.info.get('regularMarketChangePercent', 0)
if abs(change) > ALERT_THRESHOLD: alerts.append(f"{symbol} moved {change:.2f}%")
return alerts
def send_email_alert(alerts): msg = MIMEText("\n".join(alerts)) msg['Subject'] = 'Stock Alert' # ... email sending logicBetter. But I was getting alerts for every 5% move. During volatile weeks, my inbox was flooded. I started ignoring the alerts entirely.
Lesson: Without severity filtering, alerts become noise.
My Third Attempt: Scoring System
I realized I needed a scoring system. Not just “price changed” but “is this stock worth my attention?”
def calculate_score(symbol, stock_data, hist_data): """ Score from 1-10 based on: - Momentum (technical) - Valuation (fundamentals) - Sentiment (news) """ score = 0
# Momentum: Is price above 20-day moving average? sma_20 = hist_data['Close'].rolling(20).mean().iloc[-1] current_price = hist_data['Close'].iloc[-1]
if current_price > sma_20: score += 2
# Valuation: Is P/E reasonable? pe_ratio = stock_data.get('trailingPE', 0) if 0 < pe_ratio < 25: score += 2 elif pe_ratio > 50: score -= 1
# Sentiment: Check recent news news = stock_data.get('news', []) # ... sentiment analysis
return min(10, max(1, score + 5)) # Scale to 1-10
def get_recommendation(score): if score >= 7: return 'BUY' elif score <= 3: return 'SELL' return 'HOLD'Now I had scores and recommendations. But I was still running everything manually.
The Complete System: Dual-Agent Architecture
The breakthrough came from a Reddit post about using two agents:
- Portfolio Agent: Monitors stocks, generates reports
- Management Agent: Checks system health, auto-repairs
+---------------------------+| Management Agent || - Health checks || - Security audits || - Auto-repair |+------------+--------------+ | v+---------------------------+| Portfolio Agent || +---------+ +---------+ || | yfinance| | Google | || | API | | Search | || +----+----+ +----+----+ || | | || v v || +---------------------+ || | Scoring Engine | || +----------+----------+ || v || +---------------------+ || | Alert / Report | || +---------------------+ |+---------------------------+The Portfolio Agent
import yfinance as yfimport scheduleimport timefrom dataclasses import dataclass
@dataclassclass StockAlert: symbol: str alert_type: str severity: str message: str
class PortfolioAgent: def __init__(self, config): self.portfolio = config['portfolio'] # {'AAPL': 100, 'MSFT': 50} self.watchlist = config['watchlist'] self.thresholds = config['thresholds']
def fetch_data(self, symbols): data = {} for symbol in symbols: ticker = yf.Ticker(symbol) data[symbol] = { 'price': ticker.info.get('currentPrice'), 'change': ticker.info.get('regularMarketChangePercent'), 'volume': ticker.info.get('volume'), 'pe_ratio': ticker.info.get('trailingPE'), 'earnings': ticker.calendar.get('Earnings Date', []), 'news': ticker.news[:5] } return data
def check_alerts(self, data): alerts = [] for symbol, info in data.items(): # Price alert with severity change = abs(info.get('change', 0)) if change > 10: alerts.append(StockAlert(symbol, 'PRICE', 'HIGH', f"{symbol} moved {info['change']:.1f}%")) elif change > 5: alerts.append(StockAlert(symbol, 'PRICE', 'MEDIUM', f"{symbol} moved {info['change']:.1f}%"))
# Earnings alert earnings = info.get('earnings', []) if earnings: days_to_earnings = (earnings[0] - datetime.now()).days if days_to_earnings <= 7: alerts.append(StockAlert(symbol, 'EARNINGS', 'HIGH', f"{symbol} earnings in {days_to_earnings} days"))
return alerts
def calculate_score(self, symbol, data): score = 5 # Start neutral
# Momentum check ticker = yf.Ticker(symbol) hist = ticker.history(period='2mo') sma_20 = hist['Close'].rolling(20).mean().iloc[-1] sma_50 = hist['Close'].rolling(50).mean().iloc[-1] current = hist['Close'].iloc[-1]
if current > sma_20 > sma_50: score += 2 # Bullish trend elif current < sma_20 < sma_50: score -= 2 # Bearish trend
# Valuation check pe = data.get('pe_ratio', 0) if 0 < pe < 20: score += 1 elif pe > 40: score -= 1
return max(1, min(10, score))
def generate_weekly_report(self): symbols = list(self.portfolio.keys()) + self.watchlist data = self.fetch_data(symbols)
report = { 'date': datetime.now().strftime('%Y-%m-%d'), 'positions': [], 'recommendations': {}, 'top_movers': {'gainers': [], 'losers': []} }
# Score each stock for symbol in symbols: score = self.calculate_score(symbol, data[symbol]) rec = 'BUY' if score >= 7 else ('SELL' if score <= 3 else 'HOLD') report['recommendations'][symbol] = {'score': score, 'rec': rec}
if symbol in self.portfolio: shares = self.portfolio[symbol] value = shares * data[symbol]['price'] report['positions'].append({ 'symbol': symbol, 'shares': shares, 'value': value, 'change': data[symbol]['change'] })
# Top movers sorted_data = sorted(data.items(), key=lambda x: x[1]['change'], reverse=True) report['top_movers']['gainers'] = sorted_data[:3] report['top_movers']['losers'] = sorted_data[-3:]
return report
def run_daily(self): symbols = list(self.portfolio.keys()) + self.watchlist data = self.fetch_data(symbols) alerts = self.check_alerts(data)
# Only send HIGH severity alerts immediately for alert in alerts: if alert.severity == 'HIGH': self.send_notification(alert)
# Daily digest for everything else self.send_daily_digest(data, [a for a in alerts if a.severity != 'HIGH'])
def start(self): schedule.every().day.at("09:30").do(self.run_daily) schedule.every().day.at("16:00").do(self.run_daily) schedule.every().sunday.at("10:00").do(self.generate_weekly_report)
while True: schedule.run_pending() time.sleep(60)The Management Agent
The management agent keeps everything running:
import subprocessimport loggingfrom datetime import datetime, timedelta
class ManagementAgent: def __init__(self): self.logger = logging.getLogger(__name__)
def health_check(self): """Verify all components work""" checks = { 'yfinance_api': self._check_yfinance(), 'data_freshness': self._check_data_freshness(), }
failures = [k for k, v in checks.items() if not v] if failures: self.logger.warning(f"Failures: {failures}") self.auto_repair(failures)
return checks
def _check_yfinance(self): try: ticker = yf.Ticker('AAPL') _ = ticker.info.get('currentPrice') return True except Exception as e: self.logger.error(f"yfinance failed: {e}") return False
def _check_data_freshness(self): # Check if data updated in last 24 hours last_update = self._get_last_update() return (datetime.now() - last_update) < timedelta(hours=24)
def auto_repair(self, failures): for failure in failures: if failure == 'yfinance_api': # Try upgrading package subprocess.run(['pip', 'install', '--upgrade', 'yfinance']) elif failure == 'data_freshness': # Trigger manual refresh self._trigger_refresh()
def security_audit(self): """Check for exposed credentials""" issues = []
# Check for hardcoded API keys import os env_keys = ['API_KEY', 'SECRET', 'PASSWORD'] for key in env_keys: if os.environ.get(key) and len(os.environ.get(key)) > 10: # Potential hardcoded secret in code issues.append(f"Check for exposed {key}")
return issuesPutting It Together
from portfolio_agent import PortfolioAgentfrom management_agent import ManagementAgentimport schedule
def main(): config = { 'portfolio': {'AAPL': 100, 'MSFT': 50, 'GOOGL': 25}, 'watchlist': ['TSLA', 'NVDA', 'META'], 'thresholds': {'price_change': 5, 'volume_spike': 2} }
portfolio_agent = PortfolioAgent(config) management_agent = ManagementAgent()
# Schedule portfolio tasks schedule.every().day.at("09:30").do(portfolio_agent.run_daily) schedule.every().sunday.at("10:00").do(portfolio_agent.generate_weekly_report)
# Schedule management tasks schedule.every().day.at("03:00").do(management_agent.health_check) schedule.every().week.do(management_agent.security_audit)
# Initial health check management_agent.health_check()
print("Stock monitoring agent started...") while True: schedule.run_pending() time.sleep(60)
if __name__ == '__main__': main()Results After 3 Months
| Metric | Before | After |
|---|---|---|
| Weekly research time | 3+ hours | 20 minutes |
| Missed earnings | 2 per quarter | 0 |
| Portfolio return | +8% | +12% |
| Alert fatigue | High (ignored most) | Low (relevant only) |
The 20 minutes I now spend is reviewing the weekly report and checking flagged items. The agent handles everything else.
What I Learned
1. Start simple, add complexity later
My first version was a 10-line script. It evolved into a full agent system over weeks. Don’t try to build everything at once.
2. Severity levels prevent alert fatigue
Not every 5% move needs immediate attention. I only get HIGH alerts now. Everything else goes in the daily digest.
3. Self-healing is essential
My yfinance API calls fail about once a week. The management agent auto-repairs. I didn’t notice until I checked the logs.
4. Scores are more useful than raw alerts
“Stock moved 6%” is noise. “Score dropped from 7 to 3: SELL” is actionable.
5. Weekly reports beat real-time monitoring
I used to check prices multiple times a day. Now I trust the weekly report. It’s actually better because it includes context (trends, recommendations).
Common Mistakes to Avoid
+-------------------------------+-------------------------------+| MISTAKE | SOLUTION |+-------------------------------+-------------------------------+| Alert on every price move | Use severity levels || | || Single data source | Combine yfinance + news || | || No self-healing | Add management agent || | || Complex ML for scoring | Simple rules work fine || | || Ignore security | Weekly credential audits |+-------------------------------+-------------------------------+Summary
I built an AI agent that monitors my stock portfolio. The key insight: use a dual-agent architecture where one agent handles portfolio monitoring and another handles system health.
The system uses yfinance for data, a simple scoring algorithm for recommendations, and severity-based alerts to avoid notification fatigue. The management agent handles self-repair and security audits.
Time savings are real: 3+ hours per week reduced to 20 minutes of report review. More importantly, I never miss earnings dates or significant moves anymore.
If you’re managing multiple stocks manually, this is a practical automation you can build in a weekend.
Final Words + More Resources
My intention with this article was to help others share my knowledge and experience. If you want to contact me, you can contact by email: Email me
Here are also the most important links from this article along with some further resources that will help you in this scope:
- 👨💻 Reddit: Real-world AI agent stock monitoring
- 👨💻 yfinance Documentation
- 👨💻 LangGraph Documentation
Oh, and if you found these resources useful, don’t forget to support me by starring the repo on GitHub!
Comments