Rebalancing & Alert System — Automate the Discipline Your Portfolio Needs

# targets.yaml
target_allocation:
  by_position:
    AAPL: 12
    MSFT: 10
    VTI: 20
    BRK-B: 8
    GLD: 10
  by_sector:
    Technology: 30
    Healthcare: 15
    Financials: 15
    Consumer: 10
    ETF: 20
    Commodities: 10
  rebalance_threshold_pct: 5.0
  stop_loss_pct: -15.0
  tax_loss_threshold_pct: -8.0

import yaml

def load_targets(path: str = "targets.yaml") -> dict:
    with open(path) as f:
        return yaml.safe_load(f)["target_allocation"]

def detect_drift(current_weights: dict, target_weights: dict, threshold: float = 5.0) -> list:
    drifts = []
    for ticker, target_pct in target_weights.items():
        current_pct = current_weights.get(ticker, 0)
        drift = current_pct - target_pct
        if abs(drift) >= threshold:
            drifts.append({
                "ticker": ticker,
                "target_pct": target_pct,
                "current_pct": round(current_pct, 2),
                "drift_pct": round(drift, 2),
                "action": "TRIM" if drift > 0 else "ADD",
            })
    return sorted(drifts, key=lambda x: abs(x["drift_pct"]), reverse=True)

def check_stop_losses(pnl_results: list, stop_loss_pct: float = -15.0) -> list:
    return [
        {
            "ticker": r["ticker"],
            "unrealized_pct": r["unrealized_pct"],
            "unrealized_pnl": r["unrealized_pnl"],
            "severity": "critical" if r["unrealized_pct"] <= stop_loss_pct * 1.5 else "warning",
        }
        for r in pnl_results if r["unrealized_pct"] <= stop_loss_pct
    ]

def tax_loss_candidates(pnl_results: list, threshold_pct: float = -8.0) -> list:
    return [
        {
            "ticker": r["ticker"],
            "unrealized_pct": r["unrealized_pct"],
            "harvestable_loss": r["unrealized_pnl"],
            "note": "Wait 31 days before repurchasing to avoid wash sale rule",
        }
        for r in pnl_results if r["unrealized_pct"] <= threshold_pct
    ]

name: rebalancing_alerts
schedule: "0 17 * * 5"
steps:
  - load_positions:
      file: positions.yaml
  - load_targets:
      file: targets.yaml
  - fetch_prices:
      period: "2d"
  - calculate_current_weights: {}
  - detect_drift:
      threshold_pct: 5.0
  - check_stop_losses:
      threshold_pct: -15.0
  - tax_loss_candidates:
      threshold_pct: -8.0
  - llm:
      prompt: |
        Weekly portfolio maintenance brief:
        1. REBALANCING: Positions to trim and add with drift amounts
        2. STOP-LOSS ALERTS: Positions triggering stop-loss thresholds
        3. TAX-LOSS OPPORTUNITIES: Harvestable losses worth capturing
        4. RECOMMENDED ACTIONS: Top 3 prioritized actions this week
        Data: {{ drift_data }} {{ stop_loss_data }} {{ tax_loss_data }}
  - notify:
      subject: "⚖️ Weekly Rebalancing | {{ drift_count }} drifts | {{ alert_count }} alerts"