""" Auto-cycle orchestration — runs every N hours (configurable). Cycle steps: 1. Fetch current context (news, quotes, macro, geo) 2. Ask GPT-4o to suggest new patterns 3. Filter: keep only those with Jaccard keyword similarity < threshold vs existing 4. Save filtered patterns to DB 5. Score ALL patterns (existing + new) 6. Log: pattern scores, trade entry prices, geo alert, macro snapshot 7. Generate GPT-4o commentary: why are top/bottom trades performing this way? 8. Update cycle_runs with results + commentary """ import logging import threading import uuid from datetime import datetime from typing import Any, Dict, List, Optional logger = logging.getLogger(__name__) # ── Global scheduler state ──────────────────────────────────────────────────── _stop_event = threading.Event() _cycle_thread: Optional[threading.Thread] = None _cycle_lock = threading.Lock() # prevents concurrent cycles _current_status: Dict[str, Any] = { "running": False, "last_run_id": None, "last_run_at": None, "next_run_at": None, "enabled": False, "interval_hours": 3, } # ── Helpers ─────────────────────────────────────────────────────────────────── def _jaccard(a: List[str], b: List[str]) -> float: sa = {x.lower() for x in (a or [])} sb = {x.lower() for x in (b or [])} if not sa and not sb: return 0.0 union = sa | sb return len(sa & sb) / len(union) if union else 0.0 def _max_similarity_vs_existing(candidate_kws: List[str], existing: List[Dict]) -> float: return max((_jaccard(candidate_kws, p.get("keywords") or []) for p in existing), default=0.0) # ── Core cycle logic ────────────────────────────────────────────────────────── def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]: """ Execute one full auto-cycle. Thread-safe (skips if already running). Returns a summary dict. """ if not _cycle_lock.acquire(blocking=False): logger.info("Auto-cycle skipped — another cycle is already running") return {"skipped": True, "reason": "already_running"} run_id = datetime.utcnow().isoformat() summary: Dict[str, Any] = { "run_id": run_id, "trigger": trigger, "patterns_suggested": 0, "patterns_added": 0, "patterns_scored": 0, "geo_score": None, "dominant_regime": None, "commentary": None, "status": "error", } try: from services.database import ( get_config, get_custom_patterns, save_custom_pattern, save_pattern_scores, log_macro_regime, log_geo_alert, log_trade_entries, add_cycle_run, update_cycle_run, save_reasoning_trace, get_latest_portfolio_lessons, ) from services.data_fetcher import fetch_geo_news, get_all_quotes, get_macro_gauges, score_macro_scenarios from services.geo_analyzer import compute_geo_risk_score from services.ai_analyzer import ( suggest_patterns_from_market_context, score_patterns_with_context, ai_score_news_batch, _chat, DEFAULT_ANALYSIS_TEMPLATE, ) # KB confidence decay (non-blocking) try: from services.database import decay_kb_confidence _decayed = decay_kb_confidence() if _decayed: logger.info(f"[Cycle {run_id[:16]}] KB decay: {_decayed} entrée(s) mise(s) à jour") except Exception as _e: logger.warning(f"[Cycle] KB decay failed (non-blocking): {_e}") # Check AI key ai_key = get_config("openai_api_key") or "" if not ai_key: logger.warning("Auto-cycle: no OpenAI key configured, skipping AI steps") return {**summary, "status": "no_ai_key"} import os os.environ["OPENAI_API_KEY"] = ai_key sim_threshold = float(get_config("auto_cycle_similarity_threshold") or "0.30") add_cycle_run(run_id, trigger=trigger) _current_status["running"] = True _current_status["last_run_id"] = run_id # ── Step 0: Load portfolio lessons + Super Contexte ────────────────── portfolio_lessons = get_latest_portfolio_lessons() # Load Super Contexte (accumulated knowledge base) try: from services.database import get_latest_reasoning_state _reasoning_state = get_latest_reasoning_state() if _reasoning_state: if portfolio_lessons is None: portfolio_lessons = {} portfolio_lessons["super_context"] = ( f"[Super Contexte v{_reasoning_state.get('version',1)} " f"du {(_reasoning_state.get('created_at','')[:16])}]\n" + _reasoning_state.get("narrative", "")[:800] ) _synth = _reasoning_state.get("synthesis") or {} portfolio_lessons["strategic_priorities"] = _synth.get("strategic_priorities", []) portfolio_lessons["recurring_mistakes"] = [ m.get("mistake", "") for m in _synth.get("recurring_mistakes", [])[:3] ] logger.info( f"[Cycle {run_id[:16]}] Super Contexte v{_reasoning_state.get('version')} loaded " f"({_reasoning_state.get('reports_used',0)} rapports, " f"{_reasoning_state.get('trades_analyzed',0)} trades)" ) except Exception as _e: logger.warning(f"[Cycle] Could not load Super Contexte: {_e}") if portfolio_lessons: age_hours = 0 try: from datetime import datetime as _dt created = _dt.fromisoformat(portfolio_lessons["created_at"]) age_hours = (_dt.utcnow() - created).total_seconds() / 3600 except Exception: pass logger.info( f"[Cycle {run_id[:16]}] Portfolio lessons loaded " f"(report from {portfolio_lessons.get('created_at','?')[:10]}, " f"{age_hours:.0f}h ago, avg_pnl={portfolio_lessons.get('stats', {}).get('avg_pnl_pct')}%)" ) else: logger.info(f"[Cycle {run_id[:16]}] No portfolio report yet — cycle runs without performance feedback") # ── Step 1: Fetch context ───────────────────────────────────────────── logger.info(f"[Cycle {run_id[:16]}] Step 1: fetching context") from routers.geopolitical import _news_cache # type: ignore news = _news_cache.get("data") or fetch_geo_news() news = ai_score_news_batch(news) _news_cache["data"] = news geo_score_obj = compute_geo_risk_score(news) geo_score_val = int(geo_score_obj.get("score") or 0) summary["geo_score"] = geo_score_val # ── Invalidation trigger detection ──────────────────────────────────── try: from services.database import get_custom_patterns as _gcp _active_patterns = _gcp() _news_headlines = " ".join( (n.get("title", "") + " " + (n.get("summary", "") or "")).lower() for n in news[:15] ) _triggered = [] for _pat in _active_patterns: _trigger = (_pat.get("invalidation_trigger") or "").lower().strip() if not _trigger: continue # Simple keyword matching: split trigger into words and check majority match _words = [w for w in _trigger.split() if len(w) > 3] if _words and sum(1 for w in _words if w in _news_headlines) >= max(1, len(_words) // 2): _triggered.append({ "pattern_id": _pat["id"], "pattern_name": _pat["name"], "trigger": _pat["invalidation_trigger"], "probability": _pat.get("invalidation_probability"), }) if _triggered: logger.warning( f"[Cycle {run_id[:16]}] ⚠ INVALIDATION TRIGGERS FIRED for " f"{len(_triggered)} pattern(s): {[t['pattern_name'] for t in _triggered]}" ) summary["invalidation_alerts"] = _triggered except Exception as _ie: logger.debug(f"[Cycle] Invalidation check failed (non-blocking): {_ie}") quotes = get_all_quotes() gauges = get_macro_gauges() scenarios = score_macro_scenarios(gauges) macro_regime = {"gauges": gauges, "scenarios": scenarios} dominant = scenarios.get("dominant", "incertain") summary["dominant_regime"] = dominant # ── Step 2: Suggest new patterns ────────────────────────────────────── logger.info(f"[Cycle {run_id[:16]}] Step 2: suggesting patterns") _reliability_map = {} try: from services.database import get_all_pattern_reliability_map _reliability_map = get_all_pattern_reliability_map() if _reliability_map: logger.info(f"[Cycle {run_id[:16]}] Reliability map: {len(_reliability_map)} patterns") except Exception as _re: logger.warning(f"[Cycle] Reliability map failed (non-blocking): {_re}") try: from services.data_fetcher import get_economic_calendar calendar = get_economic_calendar() suggestions = suggest_patterns_from_market_context( news, quotes, calendar, macro_regime=macro_regime, geo_score=geo_score_obj, portfolio_lessons=portfolio_lessons, reliability_map=_reliability_map or None, ) except Exception as e: logger.warning(f"[Cycle] Suggestion step failed: {e}") suggestions = [] summary["patterns_suggested"] = len(suggestions) logger.info(f"[Cycle {run_id[:16]}] Suggested {len(suggestions)} patterns from AI") # ── Step 3: Filter by similarity ────────────────────────────────────── existing = get_custom_patterns() logger.info(f"[Cycle {run_id[:16]}] Step 3: {len(existing)} existing patterns, threshold={sim_threshold}") added_count = 0 for s in suggestions: kws = s.get("keywords") or [] sim = _max_similarity_vs_existing(kws, existing) if sim < sim_threshold: # Capture returned ID so the pattern has a valid id for scoring assigned_id = save_custom_pattern(s) s["id"] = assigned_id existing.append(s) added_count += 1 logger.info(f"[Cycle] Added pattern '{s.get('name')}' id={assigned_id} (sim={sim:.2f})") # ── Save suggestion reasoning trace ─────────────────────────── _top_news_ctx = [ {"title": n.get("title", "")[:120], "impact": round(float(n.get("impact_score") or 0), 2), "source": n.get("source", "")} for n in sorted(news, key=lambda x: -(float(x.get("impact_score") or 0)))[:8] ] save_reasoning_trace( run_id=run_id, trace_type="suggestion", pattern_id=assigned_id, input_context={ "geo_score": geo_score_val, "macro_dominant": dominant, "macro_scores": scenarios.get("scores", {}), "top_news": _top_news_ctx, "cycle_run_id": run_id, }, output={ "name": s.get("name"), "description": s.get("description"), "macro_fit": s.get("macro_fit"), "expected_move_pct": s.get("expected_move_pct"), "probability": s.get("probability"), "horizon_days": s.get("horizon_days"), "suggested_trades": s.get("suggested_trades", []), "keywords": s.get("keywords", []), "triggers": s.get("triggers", []), }, reasoning_summary=(s.get("macro_fit") or s.get("description") or "")[:300], geo_score=geo_score_val, macro_dominant=dominant, ) else: logger.debug(f"[Cycle] Filtered '{s.get('name')}' — sim={sim:.2f} >= {sim_threshold}") summary["patterns_added"] = added_count # ── Step 3.5: Collect IV context ───────────────────────────────────── iv_context = "" try: from services.iv_engine import get_iv_context_for_prompt, IV_WATCHLIST # Collect underlyings from current trade journal + default watchlist from services.database import get_mtm_trades_with_traces _mtm = get_mtm_trades_with_traces(days=90) _trade_tickers = list({ (t.get("underlying") or "").upper() for t in _mtm.get("all_trades", []) if t.get("underlying") }) _iv_tickers = (_trade_tickers + IV_WATCHLIST[:6])[:10] iv_context = get_iv_context_for_prompt(_iv_tickers) if iv_context: logger.info(f"[Cycle {run_id[:16]}] IV context collected for {len(_iv_tickers)} tickers") except Exception as _e: logger.warning(f"[Cycle] IV context collection failed (non-blocking): {_e}") # ── Step 4: Score ALL patterns ──────────────────────────────────────── # Verify all patterns have IDs before scoring (guard against stale data) patterns_with_id = [p for p in existing if p.get("id")] patterns_without_id = [p.get("name", "?") for p in existing if not p.get("id")] if patterns_without_id: logger.warning(f"[Cycle] {len(patterns_without_id)} patterns have no id, skipping: {patterns_without_id}") logger.info(f"[Cycle {run_id[:16]}] Step 4: scoring {len(patterns_with_id)} patterns (of {len(existing)} total)") template = get_config("analysis_template") or DEFAULT_ANALYSIS_TEMPLATE try: scored = score_patterns_with_context( patterns=patterns_with_id, recent_news=news, quotes_by_class=quotes, geo_score=geo_score_obj, template=template, top_n=len(patterns_with_id), category_filter=None, macro_regime=macro_regime, portfolio_lessons=portfolio_lessons, iv_context=iv_context, ) scored_with_id = [s for s in scored if s.get("pattern_id")] scored_without_id = [s for s in scored if not s.get("pattern_id")] if scored_without_id: logger.warning(f"[Cycle] {len(scored_without_id)} scored results have no pattern_id — they will NOT be saved to history") logger.info(f"[Cycle {run_id[:16]}] Scoring returned {len(scored)} results ({len(scored_with_id)} with valid id)") except Exception as e: logger.error(f"[Cycle] Scoring failed: {e}", exc_info=True) scored = [] summary["patterns_scored"] = len(scored) # ── Enrich scored patterns with original data not in GPT-4o response ─ # GPT-4o scoring doesn't return expected_move_pct or suggested_trades — # copy from the original pattern so log_trade_entries can use them. _pmap = {p.get("id"): p for p in patterns_with_id} for s in scored: orig = _pmap.get(s.get("pattern_id", ""), {}) if orig: if not s.get("expected_move_pct") and orig.get("expected_move_pct"): s["expected_move_pct"] = orig["expected_move_pct"] 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", []) # ── Step 5: Log everything ──────────────────────────────────────────── logger.info(f"[Cycle {run_id[:16]}] Step 5: logging") scoring_run_id = save_pattern_scores(scored, meta={ "geo_score": geo_score_val, "total": len(scored), "cycle_run_id": run_id, "trigger": trigger, }) # ── Save scoring reasoning traces (one per scored pattern) ──────────── _macro_scores_ctx = scenarios.get("scores", {}) _asset_bias_ctx = scenarios.get("asset_bias", {}).get(dominant, {}) for sp in scored: pid = sp.get("pattern_id", "") if not pid: continue orig = _pmap.get(pid, {}) save_reasoning_trace( run_id=scoring_run_id, trace_type="scoring", pattern_id=pid, input_context={ "geo_score": geo_score_val, "macro_dominant": dominant, "macro_scores": _macro_scores_ctx, "asset_class": sp.get("asset_class") or orig.get("asset_class"), "asset_bias": _asset_bias_ctx.get(sp.get("asset_class") or orig.get("asset_class", ""), "neutral"), "expected_move_pct": sp.get("expected_move_pct") or orig.get("expected_move_pct"), "cycle_run_id": run_id, }, output={ "score": sp.get("score"), "confidence": sp.get("confidence"), "buckets": sp.get("buckets", []), "key_catalyst": sp.get("key_catalyst"), "summary": sp.get("summary"), "trade_rankings": sp.get("trade_rankings", []), "recommended_trade": sp.get("recommended_trade", {}), "has_strong_contra": sp.get("has_strong_contra", False), "geo_trigger": sp.get("geo_trigger"), }, reasoning_summary=((sp.get("key_catalyst") or "") + " | " + (sp.get("summary") or ""))[:400], geo_score=geo_score_val, macro_dominant=dominant, ) logger.info(f"[Cycle {run_id[:16]}] Saved {len([s for s in scored if s.get('pattern_id')])} reasoning traces") top_patterns_log = sorted( [{"pattern_id": sp.get("pattern_id"), "name": sp.get("geo_trigger"), "score": sp.get("score", 0)} for sp in scored if sp.get("score", 0) > 0], key=lambda x: -x["score"] )[:10] log_geo_alert(geo_score=geo_score_val, top_patterns=top_patterns_log, news_count=len(news), run_id=scoring_run_id) log_trade_entries(run_id=scoring_run_id, scored_patterns=scored, quotes=quotes) gauges_summary = { k: {"value": v.get("value"), "change_pct": v.get("change_pct"), "label": v.get("label")} for k, v in gauges.items() if v.get("value") is not None or v.get("change_pct") is not None } log_macro_regime(dominant=dominant, scores=scenarios.get("scores", {}), reasons=scenarios.get("reasons", {}), gauges_summary=gauges_summary) # Update macro cache so the UI sees fresh data immediately from routers.market_data import _macro_cache # type: ignore import datetime as _dt _macro_cache["data"] = {"gauges": gauges, "scenarios": scenarios, "fetched_at": datetime.utcnow().isoformat(), "cached": False} _macro_cache["ts"] = _dt.datetime.utcnow() # ── Step 6: GPT-4o cycle commentary ────────────────────────────────── logger.info(f"[Cycle {run_id[:16]}] Step 6: generating commentary") commentary = _generate_cycle_commentary( scored=scored, dominant=dominant, scenarios=scenarios, geo_score_val=geo_score_val, news=news, gauges=gauges, ) # Attach lessons metadata to commentary so UI can display it if commentary and portfolio_lessons: try: import json as _json c = _json.loads(commentary) if isinstance(commentary, str) else commentary c["lessons_from_report"] = portfolio_lessons.get("created_at", "")[:16].replace("T", " ") c["lessons_headline"] = portfolio_lessons.get("headline", "")[:100] commentary = _json.dumps(c, ensure_ascii=False) except Exception: pass summary["commentary"] = commentary # ── Finalize ────────────────────────────────────────────────────────── summary["status"] = "completed" update_cycle_run( run_id, completed_at=datetime.utcnow().isoformat(), patterns_suggested=summary["patterns_suggested"], patterns_added=summary["patterns_added"], patterns_scored=summary["patterns_scored"], geo_score=geo_score_val, dominant_regime=dominant, commentary=commentary, status="completed", ) _current_status["last_run_at"] = datetime.utcnow().isoformat() logger.info(f"[Cycle {run_id[:16]}] Completed — {added_count} new patterns, {len(scored)} scored") # ── Step 7: Auto portfolio snapshot ────────────────────────────────── # Generate (or refresh) the portfolio report so the NEXT cycle has # fresh performance lessons. Runs in background to not block the cycle. import threading threading.Thread( target=_auto_portfolio_snapshot, args=(ai_key,), daemon=True, name=f"portfolio-snapshot-{run_id[:8]}", ).start() except Exception as e: logger.error(f"[Cycle {run_id[:16]}] Fatal error: {e}", exc_info=True) try: from services.database import update_cycle_run update_cycle_run(run_id, status="error", completed_at=datetime.utcnow().isoformat()) except Exception: pass finally: _current_status["running"] = False _cycle_lock.release() return summary def _generate_cycle_commentary( scored: List[Dict], dominant: str, scenarios: Dict, geo_score_val: int, news: List[Dict], gauges: Dict, ) -> Optional[str]: """Ask GPT-4o to explain current performance of top/bottom patterns.""" try: from services.database import get_trade_entry_prices from services.ai_analyzer import _chat # Get recent trade P&L for context (last 7 days) entries = get_trade_entry_prices(7) trade_summary = [] for e in entries[:15]: trade_summary.append({ "pattern": e.get("pattern_name", ""), "underlying": e.get("underlying", ""), "strategy": e.get("strategy", ""), "score_at_entry": e.get("score_at_entry", 0), "entry_date": e.get("entry_date", ""), }) # Top 5 scored patterns now top_scored = sorted(scored, key=lambda x: -(x.get("score") or 0))[:5] top_scored_summary = [ {"name": s.get("geo_trigger"), "score": s.get("score"), "summary": s.get("summary", "")[:120]} for s in top_scored ] # Top recent news headlines top_news = [{"title": n.get("title", ""), "impact": n.get("impact_score", 0)} for n in sorted(news, key=lambda x: -(x.get("impact_score") or 0))[:5]] import json def gv(key: str) -> str: v = gauges.get(key, {}).get("value") return str(round(v, 2)) if v is not None else "N/A" def gc(key: str) -> str: v = gauges.get(key, {}).get("change_pct") return f"{v:+.2f}%" if v is not None else "N/A" prompt = f"""Tu es un stratège macro-géopolitique senior qui analyse la performance de notre système de détection de patterns. CONTEXTE DU CYCLE (maintenant): - Régime dominant: {dominant.upper()} (score: {scenarios.get('scores', {}).get(dominant, 0)}%) - Score risque géopolitique: {geo_score_val}/100 - VIX: {gv('vix')} | Pente 10Y-3M: {gv('slope_10y3m')}% | DXY: {gc('dxy')} | Brent: {gc('brent')} - Cuivre: {gc('copper')} | Or: {gc('gold')} | S&P vs 200j: {gv('spx_vs_200d')}% TOP 5 PATTERNS ACTUELLEMENT LES MIEUX SCORÉS: {json.dumps(top_scored_summary, ensure_ascii=False, indent=2)} TRADES LOGUÉS CES 7 DERNIERS JOURS: {json.dumps(trade_summary, ensure_ascii=False, indent=2)} NEWS GÉOPOLITIQUES À FORT IMPACT: {json.dumps(top_news, ensure_ascii=False, indent=2)} Écris un COMMENTAIRE DE CYCLE concis (4-6 phrases) pour un trader options: 1. Le régime macro confirme-t-il les patterns qui scorent le mieux ? 2. Y a-t-il des news ou événements qui expliquent un écart avec nos prévisions ? 3. Quels patterns/trades méritent attention (confirmation ou invalidation) ? 4. Une recommandation tactique pour le prochain cycle (3h) Réponds UNIQUEMENT en JSON: {{"commentary": "", "key_risk": "", "top_pattern": ""}}""" result = _chat( "Tu es un stratège macro senior. Analyse concise et actionnable. JSON uniquement.", prompt, model="gpt-4o", json_mode=True, max_tokens=500, ) if result and result.get("commentary"): return json.dumps(result, ensure_ascii=False) except Exception as e: logger.warning(f"[Cycle] Commentary generation failed: {e}") return None # ── Auto portfolio snapshot ─────────────────────────────────────────────────── def _auto_portfolio_snapshot(ai_key: str) -> None: """ Called in a background thread at the end of each cycle. Fetches live prices, checks if enough trades are priced (P&L ≠ 0), and if so generates a GPT-4o portfolio report so the NEXT cycle has fresh performance lessons. Skipped silently if not enough data. """ try: import os os.environ["OPENAI_API_KEY"] = ai_key from services.database import ( get_mtm_trades_with_traces, save_ai_report, _trade_maturity, ) from datetime import date as _date data = get_mtm_trades_with_traces(days=90, limit_movers=10) all_trades = data.get("all_trades", []) priced = data.get("priced_count", 0) # Classify each trade by maturity def _with_maturity(t: Dict) -> Dict: try: dh = (_date.today() - _date.fromisoformat(t["entry_date"])).days except Exception: dh = 0 mat = _trade_maturity(dh, t.get("horizon_days") or 90) return {**t, "days_held": dh, "maturity": mat} enriched = [_with_maturity(t) for t in all_trades if t.get("pnl_pct") is not None] trop_tot = [t for t in enriched if t["maturity"]["status"] == "trop_tot"] en_cours = [t for t in enriched if t["maturity"]["status"] == "debut"] matures = [t for t in enriched if t["maturity"]["status"] in ("mature", "fin_horizon")] # Only learn lessons from mature trades (weight > 0.9) meaningful_mature = [t for t in matures if abs(t.get("pnl_pct", 0)) > 0.1] if len(meaningful_mature) < 2: logger.info( f"[AutoSnapshot] Skipping GPT-4o report: only {len(meaningful_mature)} mature trades " f"with meaningful P&L (need ≥ 2). " f"Immatures={len(trop_tot)}, en cours={len(en_cours)}, matures={len(matures)}" ) # Still try to refresh Super Contexte (independent of portfolio report) _auto_synthesize_knowledge(ai_key) return # Sort matures by P&L for the report winners = sorted(matures, key=lambda t: t.get("pnl_pct", 0), reverse=True)[:5] losers = sorted(matures, key=lambda t: t.get("pnl_pct", 0))[:5] avg_pnl = (sum(t.get("pnl_pct", 0) for t in matures) / len(matures)) if matures else None stats = { "total_trades": len(all_trades), "priced_count": priced, "avg_pnl_pct": avg_pnl, "mature_count": len(matures), "early_count": len(trop_tot) + len(en_cours), } # Build maturity context blocks for the prompt def _trade_line(t: Dict) -> str: mat = t["maturity"] return ( f" {t.get('underlying','?')} {t.get('strategy','?')} " f"P&L={t.get('pnl_pct',0):+.1f}% " f"[{mat['readable']}] " f"score_entrée={t.get('score_at_entry','?')}" ) mature_wins_block = "\n".join(_trade_line(t) for t in winners) or " Aucun" mature_losses_block = "\n".join(_trade_line(t) for t in losers) or " Aucun" early_block = "\n".join( f" {t.get('underlying','?')} {t.get('strategy','?')} " f"P&L={t.get('pnl_pct',0):+.1f}% [{t['maturity']['readable']}]" for t in (trop_tot + en_cours)[:6] ) or " Aucun" avg_str = f"{avg_pnl:+.1f}%" if avg_pnl is not None else "N/A" from services.ai_analyzer import _chat prompt = f"""Tu es un stratège macro-géopolitique senior. Rapport post-cycle automatique. ⚠️ RÈGLE FONDAMENTALE DE TIMING : Les options ont des horizons de 30-90 jours. Un trade de 3 jours ne dit RIEN sur sa performance finale. Tu dois UNIQUEMENT tirer des leçons des trades MATURES (≥35% de l'horizon écoulé). Les trades immatures sont listés pour information seulement — n'en tire AUCUNE conclusion de performance. ═══ STATISTIQUES GLOBALES ═══ Période analysée : 90 jours | Trades total: {len(all_trades)} | Pricés: {priced} Trades MATURES (signal fiable): {len(matures)} | P&L moyen matures: {avg_str} Trades IMMATURES (trop tôt pour juger): {len(trop_tot) + len(en_cours)} ═══ MATURES — TOP GAINS (signal fiable) ═══ {mature_wins_block} ═══ MATURES — TOP PERTES (signal fiable) ═══ {mature_losses_block} ═══ EN COURS — NE PAS ÉVALUER (trop tôt) ═══ {early_block} Génère un rapport JSON basé UNIQUEMENT sur les trades matures : {{ "headline": "<1 phrase résumant la performance des trades matures>", "regime_assessment": "", "winners_analysis": "", "losers_analysis": "", "key_lessons": ["", "", ""], "blind_spots": "", "next_cycle_priorities": "<3 priorités pour le prochain cycle>", "risk_watch": "<1-2 risques à surveiller>", "timing_note": "" }}""" result = _chat( "Tu es un stratège macro senior. Rapport post-cycle concis. JSON uniquement.", prompt, model="gpt-4o", json_mode=True, max_tokens=800, ) if not result: logger.warning("[AutoSnapshot] GPT-4o returned empty response") return report_id = save_ai_report( days=30, stats=stats, winners=winners, losers=losers, report=result, report_type="portfolio", ) logger.info( f"[AutoSnapshot] Portfolio report #{report_id} saved automatically " f"({len(meaningful_mature)} meaningful mature trades, avg P&L {avg_str})" ) # ── Auto-synthesize Super Contexte if stale (>6h or never generated) ── _auto_synthesize_knowledge(ai_key) except Exception as e: logger.error(f"[AutoSnapshot] Failed: {e}", exc_info=True) def _auto_synthesize_knowledge(ai_key: str) -> None: """ Synthesize the Super Contexte knowledge base after a portfolio report is generated. Skipped if the last synthesis is < 6 hours old, to avoid redundant GPT-4o calls. """ try: import os, json as _json from datetime import datetime as _dt, timedelta as _td import openai os.environ["OPENAI_API_KEY"] = ai_key from services.database import ( get_latest_reasoning_state, save_reasoning_state, get_all_kb_entries, save_kb_entry, list_ai_reports, get_mtm_trades_with_traces, ) # Skip if last synthesis < 6 hours ago last_state = get_latest_reasoning_state() if last_state: try: last_at = _dt.fromisoformat(last_state["created_at"]) age_h = (_dt.utcnow() - last_at).total_seconds() / 3600 if age_h < 6: logger.info( f"[AutoSynth] Super Contexte is {age_h:.1f}h old — skipping re-synthesis (threshold: 6h)" ) return except Exception: pass reports = list_ai_reports(limit=10) mtm_data = get_mtm_trades_with_traces(days=90) trades = mtm_data.get("all_trades", []) if isinstance(mtm_data, dict) else [] kb_entries = get_all_kb_entries() logger.info( f"[AutoSynth] Starting Super Contexte synthesis " f"({len(reports)} rapports, {len(trades)} trades, {len(kb_entries)} KB entries)" ) # Build synthesis prompt (reuse router logic inline to avoid import cycle) from routers.knowledge import _build_synthesis_prompt system_msg, user_msg = _build_synthesis_prompt(reports, trades, kb_entries) client = openai.OpenAI(api_key=ai_key) resp = client.chat.completions.create( model="gpt-4o", messages=[ {"role": "system", "content": system_msg}, {"role": "user", "content": user_msg}, ], temperature=0.3, max_tokens=2500, response_format={"type": "json_object"}, ) raw = resp.choices[0].message.content or "{}" synthesis = _json.loads(raw) narrative = synthesis.pop("narrative", "Synthèse non disponible.") state_id = save_reasoning_state( narrative=narrative, synthesis=synthesis, sources_count=len(reports) + len(trades), reports_used=len(reports), trades_analyzed=len(trades), ) # Auto-persist new KB entries from synthesis added = 0 for regime in synthesis.get("regime_insights", []): if regime.get("observation"): save_kb_entry("régimes", f"Régime: {regime.get('regime','?')}", regime["observation"], regime.get("confidence", 50), "auto-synth") added += 1 for pattern in synthesis.get("pattern_insights", []): if pattern.get("observation"): save_kb_entry("patterns", f"Pattern: {pattern.get('pattern','?')}", pattern["observation"], pattern.get("confidence", 50), "auto-synth") added += 1 for mistake in synthesis.get("recurring_mistakes", []): if mistake.get("mistake"): save_kb_entry("erreurs", mistake["mistake"][:80], f"{mistake.get('mistake','')} → {mistake.get('mitigation','')}", 70, "auto-synth") added += 1 logger.info( f"[AutoSynth] Super Contexte v{state_id} saved automatically " f"({added} KB entries added)" ) except Exception as e: logger.error(f"[AutoSynth] Failed: {e}", exc_info=True) # ── Scheduler ───────────────────────────────────────────────────────────────── def _scheduler_loop(stop_event: threading.Event): """Background loop that runs the cycle at the configured interval.""" import time from services.database import get_config while not stop_event.is_set(): try: interval_hours = float(get_config("auto_cycle_hours") or "3") except Exception: interval_hours = 3.0 _current_status["interval_hours"] = interval_hours from datetime import timedelta next_run = (datetime.utcnow() + timedelta(hours=interval_hours)).isoformat() _current_status["next_run_at"] = next_run logger.info(f"[Scheduler] Next cycle in {interval_hours}h (at {next_run[:16]} UTC)") # Wait for the interval (or until stop is signalled) stop_event.wait(timeout=interval_hours * 3600) if stop_event.is_set(): break # Check if still enabled try: enabled = (get_config("auto_cycle_enabled") or "false").lower() == "true" except Exception: enabled = False if enabled: logger.info("[Scheduler] Running scheduled auto-cycle") try: run_cycle_once(trigger="auto") except Exception as e: logger.error(f"[Scheduler] Cycle error: {e}", exc_info=True) def start_scheduler(): """Start the background scheduler thread if auto_cycle is enabled.""" global _cycle_thread, _stop_event from services.database import get_config enabled = (get_config("auto_cycle_enabled") or "false").lower() == "true" _current_status["enabled"] = enabled if not enabled: logger.info("[Scheduler] Auto-cycle disabled — skipping scheduler start") return if _cycle_thread and _cycle_thread.is_alive(): logger.info("[Scheduler] Already running") return _stop_event = threading.Event() _cycle_thread = threading.Thread( target=_scheduler_loop, args=(_stop_event,), daemon=True, name="auto-cycle-scheduler", ) _cycle_thread.start() logger.info("[Scheduler] Auto-cycle scheduler started") def stop_scheduler(): """Stop the background scheduler thread.""" global _cycle_thread _stop_event.set() if _cycle_thread: _cycle_thread.join(timeout=5) _current_status["enabled"] = False logger.info("[Scheduler] Stopped") def restart_scheduler(): """Restart the scheduler — call after config changes.""" stop_scheduler() _stop_event.clear() start_scheduler() def trigger_manual(): """Run one cycle immediately in a background thread (non-blocking).""" t = threading.Thread(target=run_cycle_once, args=("manual",), daemon=True, name="auto-cycle-manual") t.start() return t def get_status() -> Dict[str, Any]: from services.database import get_config, get_cycle_runs try: interval_hours = float(get_config("auto_cycle_hours") or "3") enabled = (get_config("auto_cycle_enabled") or "false").lower() == "true" sim_threshold = float(get_config("auto_cycle_similarity_threshold") or "0.30") min_ev = float(get_config("min_ev_threshold") or "0.0") min_score = int(get_config("min_score_threshold") or "0") except Exception: interval_hours, enabled, sim_threshold, min_ev, min_score = 3.0, False, 0.30, 0.0, 0 recent = get_cycle_runs(limit=1) last = recent[0] if recent else None return { **_current_status, "enabled": enabled, "interval_hours": interval_hours, "similarity_threshold": sim_threshold, "min_ev_threshold": min_ev, "min_score_threshold": min_score, "last_cycle": last, "scheduler_alive": bool(_cycle_thread and _cycle_thread.is_alive()), }