Skip to content

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:

stock_monitor_v1.py
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:

stock_monitor_v2.py
import yfinance as yf
import smtplib
from 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 logic

Better. 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?”

stock_scorer.py
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:

  1. Portfolio Agent: Monitors stocks, generates reports
  2. Management Agent: Checks system health, auto-repairs
Architecture
+---------------------------+
| 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

portfolio_agent.py
import yfinance as yf
import schedule
import time
from dataclasses import dataclass
@dataclass
class 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:

management_agent.py
import subprocess
import logging
from 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 issues

Putting It Together

main.py
from portfolio_agent import PortfolioAgent
from management_agent import ManagementAgent
import 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

MetricBeforeAfter
Weekly research time3+ hours20 minutes
Missed earnings2 per quarter0
Portfolio return+8%+12%
Alert fatigueHigh (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

Mistakes vs Solutions
+-------------------------------+-------------------------------+
| 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:

Oh, and if you found these resources useful, don’t forget to support me by starring the repo on GitHub!

Comments