feat: correlation-aware portfolio diversification

trading/worker.py

@@ -589,6 +589,86 @@ def _sync_to_supabase() -> None:
         logger.warning("Supabase sync failed: %s", e)
 
 
+def _diversified_selection(
+    candidates: List[Dict[str, Any]],
+    held_symbols: List[str],
+    slots: int,
+    market_id: str,
+    *,
+    max_corr: float = 0.80,
+    lookback: int = 60,
+) -> List[str]:
+    """Pick up to `slots` symbols (input already sorted best-first) while keeping
+    pairwise return-correlation with already-picked AND currently-held positions
+    below `max_corr`.
+
+    Diversification is the one mathematically-guaranteed 'free lunch': a basket of
+    less-correlated names has lower variance (and shallower drawdowns) for the same
+    expected return. The worker previously just took the top-N by confidence, which
+    can pile into correlated names (e.g. three banks) and amplify drawdowns — the
+    kind that trip the kill switch.
+
+    FAIL-OPEN: on any data/compute error this returns the plain top-`slots` list, so
+    it can never block trading or behave worse than the previous logic.
+    """
+    syms = [str(c["symbol"]) for c in candidates]
+    if slots <= 0 or not syms:
+        return []
+    top_fallback = syms[:slots]
+    try:
+        import pandas as pd
+        from data.stock_data_api import get_stock_data_for_api
+
+        all_syms = list(dict.fromkeys(syms + [str(s) for s in held_symbols]))
+        rets: Dict[str, Any] = {}
+        for s in all_syms:
+            try:
+                df = get_stock_data_for_api(
+                    to_provider_symbol(s, market_id=market_id),
+                    period="3mo", interval="1d", market_id=market_id,
+                )
+                if df is not None and not df.empty and len(df) > 25:
+                    rets[s] = df["Close"].pct_change().dropna().tail(lookback)
+            except Exception:
+                continue
+        if len(rets) < 2:
+            return top_fallback
+
+        corr = pd.DataFrame(rets).dropna(how="all").corr()
+        held = set(str(s) for s in held_symbols)
+
+        def _too_correlated(sym: str, chosen: List[str]) -> bool:
+            for other in list(chosen) + list(held):
+                if sym in corr.index and other in corr.columns:
+                    c = corr.loc[sym, other]
+                    if pd.notna(c) and abs(float(c)) > max_corr:
+                        return True
+            return False
+
+        picked: List[str] = []
+        for s in syms:                      # already sorted best-first
+            if len(picked) >= slots:
+                break
+            if s in held:
+                continue
+            if _too_correlated(s, picked):
+                continue
+            picked.append(s)
+
+        # If strict diversification underfilled the slots, top up with the
+        # next-best remaining names (relax the constraint rather than waste slots).
+        if len(picked) < slots:
+            for s in syms:
+                if len(picked) >= slots:
+                    break
+                if s not in picked and s not in held:
+                    picked.append(s)
+        return picked[:slots]
+    except Exception as e:
+        logger.warning("Diversified selection failed (%s) — using top-N fallback", e)
+        return top_fallback
+
+
 class TradingWorker:
     """Production trading worker with full safety stack.
 
@@ -1358,16 +1438,30 @@ class TradingWorker:
             for _, row in buy_signals.iterrows():
                 signal_actions[str(row["symbol"])] = "BUY_SKIPPED_NO_SLOT"
 
-        selected_buy_symbols = {
-            str(row["symbol"]) for _, row in buy_signals.head(slots).iterrows()
-        }
+        # Correlation-aware diversification: pick the best names that aren't too
+        # correlated with each other or with what we already hold. Reduces
+        # concentrated drawdowns vs. naive top-N-by-confidence. Fail-open.
+        _candidates = [
+            {"symbol": str(row["symbol"]), "confidence": float(row.get("confidence", 0))}
+            for _, row in buy_signals.iterrows()
+        ]
+        selected_buy_symbols = set(
+            _diversified_selection(
+                _candidates, list(broker_positions.keys()), slots, self.market_id,
+            )
+        )
+        if selected_buy_symbols:
+            logger.info("Diversified BUY selection (%d/%d slots): %s",
+                        len(selected_buy_symbols), slots, ", ".join(sorted(selected_buy_symbols)))
         for _, row in buy_signals.iterrows():
             sym = str(row["symbol"])
             if sym not in selected_buy_symbols and signal_actions.get(sym) == "BUY_SIGNAL":
-                signal_actions[sym] = "BUY_SKIPPED_LOW_PRIORITY"
+                signal_actions[sym] = "BUY_SKIPPED_DIVERSIFICATION"
 
-        for _, row in buy_signals.head(slots).iterrows():
+        for _, row in buy_signals.iterrows():
             sym = row["symbol"]
+            if str(sym) not in selected_buy_symbols:
+                continue
             if sym in broker_positions:
                 signal_actions[sym] = "BUY_SKIPPED_ALREADY_HELD"
                 continue