feat: pattern convergence engine — categories, signal_direction, conviction scores

Phase 1 — Catégorisation:
- database.py: ADD COLUMN category + signal_direction on custom_patterns (migration);
  save_custom_pattern persists category/signal_direction; new helpers:
  get_unclassified_patterns(), update_pattern_classification(),
  get_patterns_with_last_score()
- ai_analyzer.py: PATTERN_CATEGORIES dict (8 categories: géopolitique, macro_monétaire,
  technique, commodités_supply, risk_off, flux_saisonnier, géo_économique, crédit_stress);
  classify_patterns_batch() → GPT-4o-mini batch classification
- suggest schema: added category + signal_direction fields so new patterns are
  classified from birth
- auto_cycle.py: Step 3.1 classifies all unclassified patterns after each suggestion

Phase 2 — Convergence layer (post-scoring, no extra AI call):
- ai_analyzer.py: _compute_convergence() groups scored patterns by (underlying, signal_direction);
  conviction_bonus = min(20, +5 per additional agreeing pattern); adds conviction_score,
  conviction_bonus, convergence_count, convergence_underlying, convergence_partners to each result;
  called at end of score_patterns_with_context(), re-sorts by conviction_score
- auto_cycle.py: logs convergence summary after scoring; propagates category/signal_direction
  to scored results for display

Phase optionnelle — Convergence in suggestion prompt:
- ai_analyzer.py: suggest_patterns_from_market_context() accepts convergence_block param;
  injected into prompt so AI knows which underlyings have multi-pattern agreement
- auto_cycle.py: before suggestion, loads last-cycle scores via get_patterns_with_last_score(),
  calls _compute_convergence() to build convergence block, passes to suggester

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

backend/services/ai_analyzer.py

@@ -880,9 +880,172 @@ TEMPLATE DE NOTATION:
             p["score"] = min(total, 100)
 
     all_scored.sort(key=lambda x: x.get("score", 0), reverse=True)
+
+    # ── Phase convergence: compute cross-pattern conviction bonuses ────────────
+    all_scored, _conv_block = _compute_convergence(all_scored, patterns)
+    all_scored.sort(key=lambda x: x.get("conviction_score", x.get("score", 0)), reverse=True)
+
     return all_scored[:top_n]
 
 
+# ── Pattern convergence ───────────────────────────────────────────────────────
+
+PATTERN_CATEGORIES = {
+    "géopolitique":     "Conflits armés, sanctions, élections, alliances militaires, tensions régionales",
+    "macro_monétaire":  "Inflation, taux directeurs Fed/BCE, dollar index, yield curve, QE/QT",
+    "technique":        "RSI, Bollinger, momentum, cassures de support/résistance, patterns chartistes",
+    "commodités_supply":"OPEC, stocks pétrole/gaz, récoltes, disruptions mines, logistique supply chain",
+    "risk_off":         "Catastrophes naturelles, crises bancaires, pandémies, chocs de volatilité",
+    "flux_saisonnier":  "COT flows, saisonnalité, rééquilibrages fin trimestre, options expiry",
+    "géo_économique":   "Chaînes d'approvisionnement, protectionnisme, dédollarisation, reshoring",
+    "crédit_stress":    "Spreads HY, CDS souverains, stress bancaire, risque de défaut",
+}
+
+
+def classify_patterns_batch(patterns: List[Dict]) -> List[Dict]:
+    """Classify unclassified patterns via GPT-4o-mini → [{id, category, signal_direction}]."""
+    if not get_client() or not patterns:
+        return []
+
+    compact = [
+        {
+            "id": p.get("id"),
+            "name": p.get("name", ""),
+            "description": (p.get("description") or "")[:200],
+            "triggers": (p.get("triggers") or [])[:3],
+            "asset_class": p.get("asset_class", ""),
+        }
+        for p in patterns
+    ]
+    cats_desc = "\n".join(f"- {c}: {d}" for c, d in PATTERN_CATEGORIES.items())
+    system = "Tu es un classificateur de patterns financiers géopolitiques. Réponds UNIQUEMENT en JSON valide."
+    user = f"""Classifie chaque pattern selon UNE de ces catégories:
+{cats_desc}
+
+signal_direction:
+- "bullish": anticipe une hausse du sous-jacent
+- "bearish": anticipe une baisse du sous-jacent
+- "volatility": anticipe un mouvement sans direction claire (straddle, vol play)
+- "neutral": direction incertaine / stratégie de range
+
+Patterns à classifier:
+{json.dumps(compact, ensure_ascii=False)}
+
+Retourne UNIQUEMENT ce JSON:
+{{"classifications": [{{"id": "<id>", "category": "<catégorie>", "signal_direction": "<direction>"}}]}}"""
+
+    try:
+        res = _chat(system, user, model="gpt-4o-mini", json_mode=True, max_tokens=2000)
+        return res.get("classifications", []) if res else []
+    except Exception as _e:
+        import logging as _log2
+        _log2.getLogger(__name__).warning(f"[Classify] Batch classification failed: {_e}")
+        return []
+
+
+def _compute_convergence(
+    scored_patterns: List[Dict],
+    original_patterns: List[Dict],
+) -> tuple:
+    """
+    Group scored patterns by (underlying, signal_direction).
+    Apply conviction_bonus = min(20, +5 per additional agreeing pattern).
+    Returns (enriched_list, convergence_summary_block_str).
+    """
+    from collections import defaultdict
+
+    pat_meta = {str(p.get("id", "")): p for p in original_patterns}
+
+    # Collect underlyings + direction per pattern
+    def _pat_underlyings(sp: Dict):
+        underlyings = set()
+        rec = sp.get("recommended_trade") or {}
+        if rec.get("underlying"):
+            underlyings.add(rec["underlying"])
+        for tr in (sp.get("trade_rankings") or []):
+            if tr.get("underlying"):
+                underlyings.add(tr["underlying"])
+        return underlyings
+
+    # Build (underlying, direction) → list of pattern entries
+    groups: Dict = defaultdict(list)
+    for sp in scored_patterns:
+        pid = str(sp.get("pattern_id", ""))
+        orig = pat_meta.get(pid, {})
+        direction = (orig.get("signal_direction") or "neutral").lower()
+        if direction not in ("bullish", "bearish", "volatility", "neutral"):
+            direction = "neutral"
+        category = orig.get("category") or ""
+        name = (orig.get("name") or sp.get("geo_trigger") or "")[:50]
+        score = sp.get("score", 0)
+
+        for und in _pat_underlyings(sp):
+            groups[(und, direction)].append({
+                "pid": pid, "name": name, "category": category, "score": score,
+            })
+
+    # Enrich each scored pattern
+    enriched = []
+    for sp in scored_patterns:
+        pid = str(sp.get("pattern_id", ""))
+        orig = pat_meta.get(pid, {})
+        direction = (orig.get("signal_direction") or "neutral").lower()
+        if direction not in ("bullish", "bearish", "volatility", "neutral"):
+            direction = "neutral"
+
+        best_count, best_und, best_partners = 0, "", []
+        for und in _pat_underlyings(sp):
+            group = groups.get((und, direction), [])
+            others = [x for x in group if x["pid"] != pid]
+            if len(others) > best_count:
+                best_count, best_und, best_partners = len(others), und, others
+
+        conviction_bonus = min(20, 5 * best_count)
+        conviction_score = min(100, sp.get("score", 0) + conviction_bonus)
+        enriched.append({
+            **sp,
+            "conviction_score": conviction_score,
+            "conviction_bonus": conviction_bonus,
+            "convergence_count": best_count,
+            "convergence_underlying": best_und,
+            "convergence_partners": [
+                {"name": p["name"], "category": p["category"], "score": p["score"]}
+                for p in best_partners[:5]
+            ],
+        })
+
+    # Build convergence block for injection into NEXT suggestion prompt
+    conv_lines = []
+    for (und, direction), pats in groups.items():
+        if len(pats) >= 2:
+            avg_sc = sum(p["score"] for p in pats) / len(pats)
+            cats = ", ".join(sorted({p["category"] for p in pats if p["category"]}))
+            conv_lines.append((len(pats), avg_sc, und, direction, pats, cats))
+    conv_lines.sort(key=lambda x: (-x[0], -x[1]))
+
+    if not conv_lines:
+        conv_block = ""
+    else:
+        _dir_sym = {"bullish": "▲", "bearish": "▼", "volatility": "⟷", "neutral": "↔"}
+        lines = [
+            "\n## 🎯 CONVERGENCE SIGNAUX — Underlyings à forte conviction inter-patterns",
+            "Ces sous-jacents sont ciblés par plusieurs patterns actifs dans la MÊME direction :",
+        ]
+        for count, avg_sc, und, direction, pats, cats in conv_lines[:8]:
+            sym = _dir_sym.get(direction, "↔")
+            pat_names = " | ".join(p["name"][:30] for p in pats[:4])
+            lines.append(f"  {sym} {und} {direction.upper()}: {count} patterns convergents (score moy. {avg_sc:.0f}/100)")
+            lines.append(f"     Catégories: {cats or 'N/A'} | Patterns: {pat_names}")
+        lines += [
+            "",
+            "⚠️ Ces convergences signalent une FORTE conviction collective — privilégie des patterns",
+            "complémentaires sur ces underlyings ou signale un contra-signal si le contexte l'inverse.",
+        ]
+        conv_block = "\n".join(lines)
+
+    return enriched, conv_block
+
+
 # ── Suggest new patterns from live market context ─────────────────────────────
 
 # ── Temporal news utilities ───────────────────────────────────────────────────
@@ -1045,6 +1208,7 @@ def suggest_patterns_from_market_context(
     fred_block: str = "",
     price_discovery_block: str = "",
     portfolio_context_block: str = "",
+    convergence_block: str = "",
     run_id: str = "",
 ) -> List[Dict]:
     """Ask GPT-4o to propose new patterns based on current geo/market + macro regime context."""
@@ -1195,6 +1359,7 @@ Règles supplémentaires:
     fred_section = f"\n{fred_block}\n" if fred_block else ""
     pd_section = f"\n{price_discovery_block}\n" if price_discovery_block else ""
     portfolio_section = f"\n{portfolio_context_block}\n" if portfolio_context_block else ""
+    convergence_section = f"\n{convergence_block}\n" if convergence_block else ""
 
     user = f"""Tu es un stratège géopolitique et financier senior, expert en options.
 {macro_block}{geo_block}{lessons_block}{reliability_block}{iv_block}
@@ -1205,6 +1370,7 @@ Règles supplémentaires:
 {pd_section}
 {tech_block_section}
 {portfolio_section}
+{convergence_section}
 ## Calendrier économique à venir
 {cal_block}
 
@@ -1233,6 +1399,8 @@ Retourne UNIQUEMENT ce JSON:
       "triggers": ["<trigger1>", "<trigger2>"],
       "keywords": ["<kw1>", "<kw2>", "<kw3>"],
       "asset_class": "<energy|metals|agriculture|indices|equities|forex>",
+      "category": "<géopolitique|macro_monétaire|technique|commodités_supply|risk_off|flux_saisonnier|géo_économique|crédit_stress>",
+      "signal_direction": "<bullish|bearish|volatility|neutral>",
       "expected_move_pct": <float, RENDEMENT OPTION MOYEN en % pour ce pattern, levier inclus. Typiquement 50-300%.>,
       "probability": <float 0-1>,
       "horizon_days": <int>,

backend/services/auto_cycle.py

@@ -531,6 +531,26 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
             except Exception as _snap_e:
                 logger.warning(f"[Cycle] Context snapshot save failed (non-blocking): {_snap_e}")
 
+            # ── Build convergence block from previous cycle's scored patterns ──
+            _convergence_block_for_suggest = ""
+            try:
+                from services.database import get_patterns_with_last_score
+                from services.ai_analyzer import _compute_convergence
+                _prev_scored = get_patterns_with_last_score()
+                _prev_with_scores = [
+                    {"pattern_id": p.get("id"), "score": p.get("last_score") or 0,
+                     "recommended_trade": {"underlying": (p.get("suggested_trades") or [{}])[0].get("underlying", "")} if p.get("suggested_trades") else {},
+                     "trade_rankings": [{"underlying": t.get("underlying", "")} for t in (p.get("suggested_trades") or [])],
+                     "geo_trigger": p.get("name", "")}
+                    for p in _prev_scored if p.get("last_score") is not None
+                ]
+                if _prev_with_scores:
+                    _, _convergence_block_for_suggest = _compute_convergence(_prev_with_scores, _prev_scored)
+                    if _convergence_block_for_suggest:
+                        logger.info(f"[Cycle {run_id[:16]}] Convergence block built from {len(_prev_with_scores)} previously scored patterns")
+            except Exception as _cbe:
+                logger.warning(f"[Cycle] Convergence block build failed (non-blocking): {_cbe}")
+
             suggestions = suggest_patterns_from_market_context(
                 news, quotes, calendar, macro_regime=macro_regime, geo_score=geo_score_obj,
                 portfolio_lessons=portfolio_lessons,
@@ -541,6 +561,7 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
                 fred_block=_fred_block,
                 price_discovery_block=_price_discovery_block,
                 portfolio_context_block=_portfolio_block,
+                convergence_block=_convergence_block_for_suggest,
                 run_id=run_id,
             )
         except Exception as e:
@@ -599,6 +620,21 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
 
         summary["patterns_added"] = added_count
 
+        # ── Step 3.1: Classify unclassified patterns (category + signal_direction) ─
+        try:
+            from services.database import get_unclassified_patterns, update_pattern_classification
+            from services.ai_analyzer import classify_patterns_batch
+            _unclassified = get_unclassified_patterns()
+            if _unclassified:
+                logger.info(f"[Cycle {run_id[:16]}] Classifying {len(_unclassified)} unclassified patterns")
+                _classifications = classify_patterns_batch(_unclassified)
+                for _cl in _classifications:
+                    if _cl.get("id") and _cl.get("category"):
+                        update_pattern_classification(_cl["id"], _cl["category"], _cl.get("signal_direction", "neutral"))
+                logger.info(f"[Cycle {run_id[:16]}] Classified {len(_classifications)} patterns")
+        except Exception as _cle:
+            logger.warning(f"[Cycle] Pattern classification failed (non-blocking): {_cle}")
+
         # ── Step 3.5: Collect risk cluster context ───────────────────────────
         risk_cluster_context = ""
         try:
@@ -694,6 +730,22 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
                     logger.debug(f"[Cycle] Enriched '{orig.get('name')}' expected_move_pct={orig['expected_move_pct']}")
                 if not s.get("trade_rankings") and not s.get("suggested_trades"):
                     s["suggested_trades"] = orig.get("suggested_trades", [])
+                # Propagate category + signal_direction to scored result for display
+                if orig.get("category") and not s.get("category"):
+                    s["category"] = orig["category"]
+                if orig.get("signal_direction") and not s.get("signal_direction"):
+                    s["signal_direction"] = orig["signal_direction"]
+
+        # ── Log convergence summary ───────────────────────────────────────────
+        try:
+            _top_conv = [s for s in scored if s.get("convergence_count", 0) > 0]
+            if _top_conv:
+                logger.info(
+                    f"[Cycle {run_id[:16]}] Convergence: {len(_top_conv)} patterns with cross-pattern boost "
+                    f"— top: {_top_conv[0].get('convergence_underlying','?')} ×{_top_conv[0].get('convergence_count',0)+1}"
+                )
+        except Exception:
+            pass
 
         # ── Step 5: Log everything ────────────────────────────────────────────
         logger.info(f"[Cycle {run_id[:16]}] Step 5: logging")

backend/services/database.py

@@ -78,6 +78,9 @@ def init_db():
         "ALTER TABLE custom_patterns ADD COLUMN bayesian_win_rate REAL",
         "ALTER TABLE custom_patterns ADD COLUMN bayesian_updated_at TEXT",
         "ALTER TABLE custom_patterns ADD COLUMN bayesian_sample_size INTEGER DEFAULT 0",
+        # Convergence Phase 1 — thematic classification
+        "ALTER TABLE custom_patterns ADD COLUMN category TEXT",
+        "ALTER TABLE custom_patterns ADD COLUMN signal_direction TEXT",
     ]:
         try:
             c.execute(_sql)
@@ -863,8 +866,9 @@ def save_custom_pattern(pattern: Dict[str, Any]) -> str:
         suggested_trades, asset_class, expected_move_pct, probability,
         horizon_days, ai_quality_score, ai_evaluation, source,
         counter_thesis, invalidation_trigger, invalidation_probability,
+        category, signal_direction,
         is_active, updated_at
-    ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,1,datetime('now'))""", (
+    ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,1,datetime('now'))""", (
         pat_id,
         pattern.get("name", ""),
         pattern.get("description", ""),
@@ -882,6 +886,8 @@ def save_custom_pattern(pattern: Dict[str, Any]) -> str:
         pattern.get("counter_thesis"),
         pattern.get("invalidation_trigger"),
         pattern.get("invalidation_probability"),
+        pattern.get("category"),
+        pattern.get("signal_direction"),
     ))
     conn.commit()
     conn.close()
@@ -903,6 +909,57 @@ def get_custom_patterns() -> List[Dict[str, Any]]:
     return result
 
 
+def get_unclassified_patterns() -> List[Dict[str, Any]]:
+    """Return active patterns where category IS NULL (need AI classification)."""
+    conn = get_conn()
+    rows = conn.execute(
+        "SELECT * FROM custom_patterns WHERE is_active=1 AND (category IS NULL OR category='')"
+    ).fetchall()
+    conn.close()
+    result = []
+    for r in rows:
+        d = dict(r)
+        for f in ["triggers", "keywords", "historical_instances", "suggested_trades", "ai_evaluation"]:
+            d[f] = json.loads(d.get(f) or "[]")
+        result.append(d)
+    return result
+
+
+def update_pattern_classification(pat_id: str, category: str, signal_direction: str) -> None:
+    """Persist category + signal_direction for a pattern."""
+    conn = get_conn()
+    conn.execute(
+        "UPDATE custom_patterns SET category=?, signal_direction=?, updated_at=datetime('now') WHERE id=?",
+        (category, signal_direction, pat_id),
+    )
+    conn.commit()
+    conn.close()
+
+
+def get_patterns_with_last_score() -> List[Dict[str, Any]]:
+    """Each active pattern + its most recent score from pattern_score_history."""
+    conn = get_conn()
+    rows = conn.execute("""
+        SELECT cp.*, psh.score AS last_score, psh.summary AS last_summary, psh.scored_at AS last_scored_at
+        FROM custom_patterns cp
+        LEFT JOIN (
+            SELECT pattern_id, score, summary, scored_at,
+                   ROW_NUMBER() OVER (PARTITION BY pattern_id ORDER BY scored_at DESC) AS rn
+            FROM pattern_score_history
+        ) psh ON psh.pattern_id = cp.id AND psh.rn = 1
+        WHERE cp.is_active=1
+        ORDER BY cp.created_at DESC
+    """).fetchall()
+    conn.close()
+    result = []
+    for r in rows:
+        d = dict(r)
+        for f in ["triggers", "keywords", "historical_instances", "suggested_trades", "ai_evaluation"]:
+            d[f] = json.loads(d.get(f) or "[]")
+        result.append(d)
+    return result
+
+
 def toggle_pattern_active(pat_id: str) -> bool:
     """Toggle is_active for a pattern. Returns the new state."""
     conn = get_conn()