Portfolio Tracker — Automated Daily P&L and Position Monitoring

# positions.yaml
portfolio:
  name: "My Portfolio"
  currency: USD
  positions:
    - ticker: AAPL
      shares: 50
      avg_cost: 165.20
      sector: Technology
      account: taxable
    - ticker: MSFT
      shares: 30
      avg_cost: 310.50
      sector: Technology
      account: taxable
    - ticker: VTI
      shares: 100
      avg_cost: 218.00
      sector: ETF
      account: ira
    - ticker: BRK-B
      shares: 20
      avg_cost: 345.00
      sector: Financials
      account: taxable
    - ticker: GLD
      shares: 15
      avg_cost: 180.00
      sector: Commodities
      account: taxable

import yfinance as yf
import yaml
from datetime import datetime

def load_positions(path: str = "positions.yaml") -> list:
    with open(path) as f:
        data = yaml.safe_load(f)
    return data["portfolio"]["positions"]

def calculate_pnl(positions: list) -> list:
    tickers = [p["ticker"] for p in positions]
    prices = yf.download(tickers, period="2d", auto_adjust=True, progress=False)["Close"]
    results = []
    for p in positions:
        ticker = p["ticker"]
        try:
            current = float(prices[ticker].iloc[-1])
            prior = float(prices[ticker].iloc[-2])
        except (KeyError, IndexError):
            continue
        cost_basis = p["avg_cost"] * p["shares"]
        market_value = current * p["shares"]
        unrealized = market_value - cost_basis
        daily_pnl = (current - prior) * p["shares"]
        results.append({
            "ticker": ticker,
            "shares": p["shares"],
            "avg_cost": p["avg_cost"],
            "current_price": round(current, 2),
            "market_value": round(market_value, 2),
            "cost_basis": round(cost_basis, 2),
            "unrealized_pnl": round(unrealized, 2),
            "unrealized_pct": round((unrealized / cost_basis) * 100, 2),
            "daily_pnl": round(daily_pnl, 2),
            "daily_pct": round(((current - prior) / prior) * 100, 2),
        })
    return results

def portfolio_summary(results: list) -> dict:
    total_value = sum(r["market_value"] for r in results)
    total_cost = sum(r["cost_basis"] for r in results)
    total_unrealized = sum(r["unrealized_pnl"] for r in results)
    total_daily = sum(r["daily_pnl"] for r in results)
    return {
        "total_market_value": round(total_value, 2),
        "total_cost_basis": round(total_cost, 2),
        "total_unrealized_pnl": round(total_unrealized, 2),
        "total_unrealized_pct": round((total_unrealized / total_cost) * 100, 2),
        "total_daily_pnl": round(total_daily, 2),
        "total_daily_pct": round((total_daily / (total_cost)) * 100, 2),
        "position_count": len(results),
        "as_of": datetime.now().strftime("%Y-%m-%d %H:%M"),
    }

name: daily_portfolio_brief
schedule: "0 17 * * 1-5"
steps:
  - load_positions:
      file: positions.yaml
  - fetch_prices:
      provider: yfinance
      period: "2d"
  - calculate_pnl: {}
  - llm:
      prompt: |
        Generate a concise end-of-day portfolio brief:
        1. PORTFOLIO SUMMARY: Total value, daily P&L, unrealized gain/loss
        2. TOP MOVERS: Biggest daily gainers and losers
        3. NOTABLE: Any position up/down more than 3% today
        Data: {{ pnl_results }}
  - notify:
      channel: email
      subject: "📊 Portfolio Close: ${{ total_value }} | Daily: {{ daily_pnl }}"