Correlation & Concentration Risk — Are You More Concentrated Than You Think?

import yfinance as yf
import pandas as pd

def correlation_matrix(positions: list, period: str = "6mo") -> pd.DataFrame:
    tickers = [p["ticker"] for p in positions]
    prices = yf.download(tickers, period=period, auto_adjust=True, progress=False)["Close"]
    return prices.pct_change().dropna().corr().round(3)

def high_correlation_pairs(corr_matrix: pd.DataFrame, threshold: float = 0.75) -> list:
    tickers = corr_matrix.columns.tolist()
    pairs = []
    for i in range(len(tickers)):
        for j in range(i + 1, len(tickers)):
            val = corr_matrix.iloc[i, j]
            if abs(val) >= threshold:
                pairs.append({"ticker_a": tickers[i], "ticker_b": tickers[j],
                               "correlation": round(val, 3),
                               "direction": "positive" if val > 0 else "negative"})
    return sorted(pairs, key=lambda x: abs(x["correlation"]), reverse=True)

def sector_concentration(positions: list, current_prices: dict) -> dict:
    sector_values = {}
    total_value = 0
    for p in positions:
        value = p["shares"] * current_prices.get(p["ticker"], p["avg_cost"])
        sector = p.get("sector", "Unknown")
        sector_values[sector] = sector_values.get(sector, 0) + value
        total_value += value
    weights = {s: round(v / total_value * 100, 2) for s, v in sector_values.items()}
    weights_sorted = dict(sorted(weights.items(), key=lambda x: x[1], reverse=True))
    hhi = sum((w / 100) ** 2 for w in weights.values())
    return {
        "by_sector": weights_sorted,
        "hhi": round(hhi, 4),
        "concentration_score": "high" if hhi > 0.25 else "moderate" if hhi > 0.15 else "diversified",
        "top_sector": max(weights, key=weights.get),
        "top_sector_pct": max(weights.values()),
    }

def portfolio_beta(positions: list, benchmark: str = "SPY", period: str = "1y") -> dict:
    tickers = [p["ticker"] for p in positions] + [benchmark]
    prices = yf.download(tickers, period=period, auto_adjust=True, progress=False)["Close"]
    returns = prices.pct_change().dropna()
    bench = returns[benchmark]
    betas = {}
    for p in positions:
        t = p["ticker"]
        if t in returns.columns:
            cov = returns[t].cov(bench)
            betas[t] = round(cov / bench.var(), 3) if bench.var() > 0 else 1.0
    total_val = sum(p["shares"] * float(prices[p["ticker"]].iloc[-1]) for p in positions)
    port_beta = sum(
        betas.get(p["ticker"], 1.0) * (p["shares"] * float(prices[p["ticker"]].iloc[-1])) / total_val
        for p in positions
    )
    return {
        "individual_betas": betas,
        "portfolio_beta": round(port_beta, 3),
        "interpretation": "aggressive" if port_beta > 1.2 else "defensive" if port_beta < 0.8 else "market-like",
    }

name: correlation_monitor
schedule: "0 8 * * 1"
steps:
  - load_positions:
      file: positions.yaml
  - fetch_prices:
      period: "6mo"
  - correlation_matrix: {}
  - high_correlation_pairs:
      threshold: 0.75
  - sector_concentration: {}
  - portfolio_beta:
      benchmark: SPY
  - alert_if:
      conditions:
        - metric: top_sector_pct
          threshold: 40
          message: "Single sector exceeds 40% of portfolio"
  - notify:
      subject: "🔗 Weekly Risk: {{ top_sector }}={{ top_pct }}% | Beta={{ portfolio_beta }}"