diff --git a/backend/services/ai_analyzer.py b/backend/services/ai_analyzer.py index e8bd51e..27c7b8a 100644 --- a/backend/services/ai_analyzer.py +++ b/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": "", "category": "", "signal_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": ["", ""], "keywords": ["", "", ""], "asset_class": "", + "category": "", + "signal_direction": "", "expected_move_pct": , "probability": , "horizon_days": , diff --git a/backend/services/auto_cycle.py b/backend/services/auto_cycle.py index d186572..ef632e1 100644 --- a/backend/services/auto_cycle.py +++ b/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") diff --git a/backend/services/database.py b/backend/services/database.py index 3a28c79..eb9f855 100644 --- a/backend/services/database.py +++ b/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()