""" 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, timedelta 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) _EMBED_SIM_THRESHOLD = 0.75 # cosine threshold to consider two patterns as duplicates — fallback default, see CYCLE_STEP_CATALOG["duplicate_detection"] def _is_duplicate_pattern( candidate: Dict, existing: List[Dict], api_key: str, jaccard_threshold: float = 0.30, embed_threshold: float = _EMBED_SIM_THRESHOLD, ) -> bool: """ Returns True if the candidate is too similar to an existing pattern. Tries cosine embeddings (Sprint 4.3) with Jaccard fallback. """ if not existing: return False # Tentative embedding cosinus if api_key: text = ((candidate.get("name") or "") + " " + (candidate.get("description") or "")).strip() if text: try: from services.database import max_cosine_similarity_vs_existing sim = max_cosine_similarity_vs_existing( candidate_text=text, existing_patterns=existing, api_key=api_key, candidate_id=candidate.get("id"), ) if sim > 0: # embedding worked return sim >= embed_threshold except Exception as _emb_err: logger.debug(f"[Cycle] Embedding failed, fallback Jaccard: {_emb_err}") # Fallback Jaccard sim = _max_similarity_vs_existing(candidate.get("keywords") or [], existing) return sim >= jaccard_threshold # ── Cycle step configuration — self-describing catalog + DB-backed overrides ── # Mirrors the SIGNAL_CATALOG pattern already used for AI Desks (routers/ai_desks.py): # one declarative entry per knob so Config.tsx can render a generic form instead # of hand-coded sliders. Stored as one JSON blob under config key # "cycle_step_config" (same convention as exit_defaults/analysis_config), merged # with these defaults at read time — adding a new knob here never requires a # DB migration. CYCLE_STEP_CATALOG: List[Dict[str, Any]] = [ {"id": "portfolio_monitor", "label": "Portfolio Monitor", "group": "portfolio", "description": "Alerte IA quand le risque du portefeuille simulé dépasse un seuil (conflits, concentration).", "params": {"enabled": {"type": "bool", "default": True}}}, {"id": "watchlist_auto_add", "label": "Ajout auto à la watchlist", "group": "portfolio", "description": "Ajoute automatiquement les nouveaux tickers détectés par le cycle à la watchlist.", "params": {"enabled": {"type": "bool", "default": True}}}, {"id": "regime_clustering", "label": "Clustering de régime macro", "group": "ai", "description": "Reclassifie le régime macro courant par clustering.", "params": { "enabled": {"type": "bool", "default": True}, "n_clusters": {"type": "int", "label": "Nb clusters", "default": 4, "min": 2, "max": 8}, "lookback_days": {"type": "int", "label": "Lookback (j)", "default": 180, "min": 30, "max": 365}, }}, {"id": "knowledge_synthesis", "label": "Synthèse de connaissances", "group": "ai", "description": "Résume les derniers rapports IA en enseignements exploitables.", "params": { "enabled": {"type": "bool", "default": True}, "min_hours_between": {"type": "int", "label": "Délai min entre synthèses (h)", "default": 6, "min": 1, "max": 48}, }}, {"id": "institutional_absorption", "label": "Absorption rapports institutionnels", "group": "data", "description": "Fenêtre de suivi de l'absorption marché d'un rapport institutionnel.", "params": {"lookback_days": {"type": "int", "label": "Lookback (j)", "default": 14, "min": 3, "max": 60}}}, {"id": "var_snapshot", "label": "Snapshot VaR", "group": "portfolio", "description": "Paramètres du calcul de Value-at-Risk dans le rapport de cycle.", "params": { "confidence": {"type": "float", "label": "Confiance", "default": 0.95, "min": 0.80, "max": 0.99}, "horizon_days": {"type": "int", "label": "Horizon (j)", "default": 1, "min": 1, "max": 30}, "lookback_days": {"type": "int", "label": "Lookback (j)", "default": 252, "min": 60, "max": 504}, "default_iv": {"type": "float", "label": "IV par défaut", "default": 0.20, "min": 0.05, "max": 1.0}, }}, {"id": "duplicate_detection", "label": "Détection de doublons de patterns", "group": "ai", "description": "Seuil de similarité cosinus (embeddings) pour éviter de recréer un pattern existant.", "params": {"embedding_threshold": {"type": "float", "label": "Seuil embedding", "default": 0.75, "min": 0.5, "max": 0.95}}}, {"id": "price_snapshot_retention", "label": "Rétention snapshots de prix", "group": "data", "description": "Durée de conservation des snapshots de prix intra-cycle.", "params": {"days": {"type": "int", "label": "Jours", "default": 14, "min": 1, "max": 90}}}, {"id": "iv_context", "label": "Contexte IV / mouvements récents", "group": "data", "description": "Fenêtre de trades utilisée pour le contexte IV (utilisée à 3 endroits du cycle).", "params": {"lookback_days": {"type": "int", "label": "Jours", "default": 90, "min": 7, "max": 365}}}, {"id": "absorption_detection", "label": "Détection d'absorption de prix", "group": "data", "description": "Fenêtre d'âge pour la détection d'absorption de prix (Phase 4B).", "params": { "min_age_minutes": {"type": "float", "label": "Âge min (min)", "default": 30.0, "min": 5, "max": 180}, "max_age_days": {"type": "int", "label": "Âge max (j)", "default": 7, "min": 1, "max": 30}, }}, {"id": "economic_surprises", "label": "Surprises économiques (FRED)", "group": "data", "description": "Fenêtre de lookback pour les surprises économiques récentes.", "params": {"lookback_days": {"type": "int", "label": "Jours", "default": 60, "min": 7, "max": 180}}}, {"id": "institutional_block", "label": "Bloc institutionnel (prompt)", "group": "data", "description": "Fenêtre des rapports institutionnels injectés dans le prompt de suggestion.", "params": {"lookback_days": {"type": "int", "label": "Jours", "default": 7, "min": 1, "max": 30}}}, {"id": "cycle_commentary", "label": "Commentaire de fin de cycle", "group": "ai", "description": "Fenêtre de trades récents utilisée pour le commentaire GPT-4o de fin de cycle.", "params": {"lookback_days": {"type": "int", "label": "Jours", "default": 7, "min": 1, "max": 30}}}, ] def _default_cycle_step_config() -> Dict[str, Dict[str, Any]]: return {step["id"]: {k: p["default"] for k, p in step["params"].items()} for step in CYCLE_STEP_CATALOG} def get_cycle_step_config() -> Dict[str, Dict[str, Any]]: """Merge saved overrides (config key "cycle_step_config") over the catalog defaults — missing keys/steps fall back to defaults, so adding a new knob to CYCLE_STEP_CATALOG is never a breaking change for existing saved config.""" import json from services.database import get_config defaults = _default_cycle_step_config() try: saved = json.loads(get_config("cycle_step_config") or "{}") except Exception: saved = {} return {step_id: {**step_defaults, **(saved.get(step_id) or {})} for step_id, step_defaults in defaults.items()} # ── Portfolio monitor agent ─────────────────────────────────────────────────── def _run_portfolio_monitor(risk: dict, cycle_id: str) -> dict: """ AI agent that analyses the open simulation portfolio and generates recommendations. Runs only when there are alerts (conflicts, concentration). Results saved to system_logs. """ from services.portfolio_risk import build_monitor_context from services.database import log_system_event from services.ai_analyzer import _chat import json as _json context = build_monitor_context(risk) n_conflicts = len(risk.get("conflicts", [])) n_alerts = len(risk.get("alerts", [])) system_prompt = ( "You are a risk manager specializing in geopolitical options portfolios. " "Analyze the state of the simulated portfolio below and generate concrete recommendations. " "Respond in JSON with this exact schema: " '{"assessment": "", ' '"actions": [{"priority": "high|medium", "type": "close_trade|rebalance|monitor", ' '"trade_id": , "underlying": "", "reason": ""}], ' '"rebalance_suggestion": ""}' ) user_prompt = context try: result = _chat(system_prompt, user_prompt, model="gpt-4o-mini", json_mode=True, max_tokens=600) if isinstance(result, dict) and "assessment" in result: details_str = _json.dumps(result, ensure_ascii=False) msg = f"[PortfolioMonitor] {n_conflicts} conflits, {n_alerts} alertes — {result.get('assessment', '')[:120]}" log_system_event("WARN" if n_conflicts else "INFO", "portfolio_monitor", msg, cycle_id=cycle_id, details=details_str) logger.info(f"[PortfolioMonitor] AI assessment saved — {len(result.get('actions', []))} actions") return result except Exception as e: logger.warning(f"[PortfolioMonitor] AI call failed: {e}") # Fallback: log raw alerts without AI fallback = { "assessment": f"{n_conflicts} conflits directionnels et {n_alerts} alertes détectées dans le portefeuille simulé.", "actions": [ {"priority": "high" if a["level"] == "danger" else "medium", "type": "close_trade" if a["type"] == "conflict" else "monitor", "trade_id": None, "underlying": a.get("underlying", ""), "reason": a["message"]} for a in risk.get("alerts", [])[:5] ], "rebalance_suggestion": "Revérifier manuellement les positions contradictoires.", } log_system_event("WARN", "portfolio_monitor", f"[PortfolioMonitor] {n_alerts} alertes (analyse IA indisponible)", cycle_id=cycle_id, details=_json.dumps(fallback, ensure_ascii=False)) return fallback # ── 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, log_system_event, get_last_completed_cycle_ts, save_cycle_context_snapshot, purge_old_price_snapshots, ) 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, ai_score_geo_risk, _chat, DEFAULT_ANALYSIS_TEMPLATE, ) from services.portfolio_context import ( get_open_trades_with_moves, get_portfolio_concentration, build_portfolio_context_block, ) # 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") step_cfg = get_cycle_step_config() add_cycle_run(run_id, trigger=trigger) _current_status["running"] = True _current_status["last_run_id"] = run_id # ── Cycle meta — timing context ─────────────────────────────────────── _now = datetime.utcnow() _last_cycle_ts_str = get_last_completed_cycle_ts() _delta_minutes: float = 180.0 # default 3h if no prior cycle if _last_cycle_ts_str: try: _last_dt = datetime.fromisoformat(_last_cycle_ts_str.replace("Z", "")) _delta_minutes = max(1.0, (_now - _last_dt).total_seconds() / 60) except Exception: pass _interval_hours = float(get_config("auto_cycle_hours") or "3") if _delta_minutes < 90: _calib_label = "Court terme (<2h) — signaux très récents" elif _delta_minutes < 720: _calib_label = f"Moyen terme ({_delta_minutes/60:.0f}h) — signaux potentiellement pas encore pricés" else: _calib_label = f"Long terme ({_delta_minutes/60:.0f}h) — marchés ont eu le temps d'intégrer" # ── Market calendar context ──────────────────────────────────────── from datetime import timezone as _tz _now_utc = _now.replace(tzinfo=_tz.utc) if _now.tzinfo is None else _now.astimezone(_tz.utc) _weekday = _now_utc.weekday() # 0=Mon … 6=Sun _day_names = ["Lundi", "Mardi", "Mercredi", "Jeudi", "Vendredi", "Samedi", "Dimanche"] _is_weekend = _weekday >= 5 # CME equity/energy/metals futures: closed Fri 17:00 ET → Sun 18:00 ET # Rough ET offset (UTC-5 winter / UTC-4 summer) — good enough for intent _et_hour = (_now_utc.hour - 5) % 24 if _is_weekend: _market_session = "closed" _market_note = f"{_day_names[_weekday]} — marchés US et CME fermés. Derniers prix disponibles: vendredi clôture." elif _et_hour < 4: _market_session = "overnight" _market_note = "Nuit US (overnight) — liquidité réduite, spreads larges." elif _et_hour < 9: _market_session = "pre_market" _market_note = "Pré-marché US — prix indicatifs, faible liquidité." elif _et_hour < 16: _market_session = "open" _market_note = "Marché US ouvert." else: _market_session = "after_hours" _market_note = "After-hours US — prix indicatifs, liquidité réduite." _budget_eur = float(get_config("trade_budget_eur") or "5000") _horizon_min = int(get_config("preferred_horizon_min") or "30") _horizon_max = int(get_config("preferred_horizon_max") or "180") cycle_meta = { "current_cycle_ts": _now.isoformat(), "last_cycle_ts": _last_cycle_ts_str, "delta_minutes": round(_delta_minutes, 1), "interval_hours": _interval_hours, "calibration_label": _calib_label, "day_of_week": _day_names[_weekday], "is_weekend": _is_weekend, "market_session": _market_session, "market_note": _market_note, "trade_budget_eur": _budget_eur, "preferred_horizon_min": _horizon_min, "preferred_horizon_max": _horizon_max, } logger.info( f"[Cycle {run_id[:16]}] Cycle meta: delta={_delta_minutes:.0f}min depuis dernier cycle" f" ({_last_cycle_ts_str[:16] if _last_cycle_ts_str else 'premier cycle'})" ) # ── 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 algo_geo_score_obj = compute_geo_risk_score(news) try: geo_score_obj = ai_score_geo_risk(news, algo_geo_score_obj, log_meta={"run_id": run_id, "call_type": "geo_risk_score"}) except Exception as _ge: logger.warning(f"[Cycle {run_id[:16]}] AI geo risk scoring failed, falling back to algo score: {_ge}") geo_score_obj = {**algo_geo_score_obj, "rationale": "Erreur IA — score algorithmique utilisé tel quel.", "top_risks": []} geo_score_obj.setdefault("breakdown", algo_geo_score_obj.get("breakdown", {})) geo_score_val = int(round(geo_score_obj.get("score") or 0)) summary["geo_score"] = geo_score_val from services.database import save_geo_risk_snapshot save_geo_risk_snapshot( run_id=run_id, score=geo_score_obj.get("score", geo_score_val), level=geo_score_obj.get("level", "low"), breakdown=geo_score_obj.get("breakdown", {}), top_risks=geo_score_obj.get("top_risks", []), rationale=geo_score_obj.get("rationale", ""), ) # ── 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() # ── Phase 4A: capture price snapshots for scored news ───────────────── try: from services.price_discovery import capture_price_snapshots as _cap_snap _quotes_flat: Dict[str, float] = {} for _qs in quotes.values(): for _q in _qs: if _q.get("symbol") and _q.get("price"): _quotes_flat[_q["symbol"]] = float(_q["price"]) _n_snaps = _cap_snap(news, _quotes_flat, run_id) if _n_snaps: logger.info(f"[Cycle {run_id[:16]}] Phase 4: {_n_snaps} price snapshots captured") purge_old_price_snapshots(older_than_days=step_cfg["price_snapshot_retention"]["days"]) except Exception as _pd_e: logger.warning(f"[Cycle] Price snapshot capture failed (non-blocking): {_pd_e}") 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 1.9: Pre-fetch IV context + snapshots (suggestion rules + gate) ─ iv_context = "" _iv_snapshots: Dict[str, Dict] = {} try: from services.iv_engine import get_iv_context_for_prompt, get_full_iv_snapshot, IV_WATCHLIST from services.database import get_mtm_trades_with_traces _mtm_pre = get_mtm_trades_with_traces(days=step_cfg["iv_context"]["lookback_days"]) _trade_tickers_pre = list({ (t.get("underlying") or "").upper() for t in _mtm_pre.get("all_trades", []) if t.get("underlying") }) _iv_tickers_pre = (_trade_tickers_pre + IV_WATCHLIST[:6])[:10] iv_context = get_iv_context_for_prompt(_iv_tickers_pre) # Also collect structured snapshots — used by IV gate before log_trade_entries for _tkr in _iv_tickers_pre: try: _iv_snapshots[_tkr] = get_full_iv_snapshot(_tkr) except Exception: pass if iv_context: logger.info( f"[Cycle {run_id[:16]}] IV context pre-fetched: {len(_iv_tickers_pre)} tickers, " f"{len(_iv_snapshots)} snapshots" ) except Exception as _e: logger.warning(f"[Cycle] IV context pre-fetch failed (non-blocking): {_e}") # ── 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}") # ── Phase 4B: compute price absorption from previous snapshots ─────── _absorptions = [] _price_discovery_block = "" try: from services.price_discovery import compute_absorptions, build_price_discovery_block _absorptions = compute_absorptions( min_age_minutes=step_cfg["absorption_detection"]["min_age_minutes"], max_age_days=step_cfg["absorption_detection"]["max_age_days"], ) _price_discovery_block = build_price_discovery_block(_absorptions) opps = sum(1 for a in _absorptions if a["opportunity"]) if _absorptions: logger.info(f"[Cycle {run_id[:16]}] Phase 4: {len(_absorptions)} absorptions, {opps} opportunities") except Exception as _abs_e: logger.warning(f"[Cycle] Price absorption failed (non-blocking): {_abs_e}") # ── Phase 2: FRED recent macro releases + economic surprise tracking ──── _fred_releases = [] _fred_block = "" _surprise_block = "" try: from services.fred_fetcher import ( get_fred_recent_releases, build_fred_context_block, save_fred_releases_to_db, build_economic_surprise_block, ) _fred_key = get_config("fred_api_key") or "" _fred_releases = get_fred_recent_releases() if _fred_releases: logger.info(f"[Cycle {run_id[:16]}] FRED: {len(_fred_releases)} releases fetched") # Persist to economic_events with z-score surprise scoring try: _saved = save_fred_releases_to_db(_fred_releases, api_key=_fred_key) if _saved: logger.info(f"[Cycle {run_id[:16]}] FRED: {_saved} releases saved to economic_events") except Exception as _dbe: logger.warning(f"[Cycle] FRED DB save failed (non-blocking): {_dbe}") # Add surprise scores to release dicts for richer prompt block if _fred_key: try: from services.database import get_recent_economic_surprises _db_surprises = {ev["series_id"]: ev for ev in get_recent_economic_surprises(days=step_cfg["economic_surprises"]["lookback_days"])} for rel in _fred_releases: sid = rel.get("series_id", "") if sid in _db_surprises: rel["surprise_zscore"] = _db_surprises[sid].get("surprise_zscore") except Exception: pass _fred_block = build_fred_context_block(_fred_releases) _surprise_block = build_economic_surprise_block(days=14) if _surprise_block: _fred_block = _fred_block + "\n\n" + _surprise_block if _fred_block else _surprise_block except Exception as _fe: logger.warning(f"[Cycle] FRED fetch failed (non-blocking): {_fe}") try: from services.data_fetcher import get_economic_calendar calendar = get_economic_calendar() # Apply decay to news before suggestion (adds decayed_score + age_hours) from services.ai_analyzer import apply_news_decay as _apply_decay news = _apply_decay(news) # ── Tech indicators for top tickers ────────────────────────────── _tech_block = "" try: _ti_enabled = (get_config("tech_indicators_enabled") or "true").lower() == "true" if _ti_enabled: from services.technical_indicators import compute_indicators, format_indicators_for_prompt _ti_list = [s.strip() for s in (get_config("tech_indicators_list") or "rsi,ma,bollinger,atr").split(",") if s.strip()] _horizon_days = 45 # default mid-range if no trade horizon context yet _ti_lines = [] # Pick up to 5 tickers from quotes (one per asset class) _seen_tickers: set = set() for _cls, _qs in quotes.items(): for _q in _qs[:1]: _sym = _q.get("symbol", "") if _sym and _sym not in _seen_tickers: _seen_tickers.add(_sym) _ind = compute_indicators(_sym, _horizon_days, enabled_indicators=_ti_list) _blk = format_indicators_for_prompt(_ind) if _blk: _ti_lines.append(_blk) if len(_seen_tickers) >= 5: break if len(_seen_tickers) >= 5: break if _ti_lines: _tech_block = "\n".join(_ti_lines) except Exception as _te: logger.warning(f"[Cycle] Tech indicators failed (non-blocking): {_te}") # ── Portfolio context (open positions + recent moves) ────────── _portfolio_block = "" _open_trades_snap = [] _portfolio_conc = {} try: _open_trades_snap = get_open_trades_with_moves() _portfolio_conc = get_portfolio_concentration(_open_trades_snap) _portfolio_block = build_portfolio_context_block(_open_trades_snap, _portfolio_conc) logger.info(f"[Cycle {run_id[:16]}] Portfolio context: {len(_open_trades_snap)} open trades") except Exception as _pe: logger.warning(f"[Cycle] Portfolio context failed (non-blocking): {_pe}") # ── Save full context snapshot ───────────────────────────────── try: from services.ai_analyzer import apply_news_decay as _apply_decay2, partition_news_by_age as _part _news_snap = _part(_apply_decay2(news), cycle_meta.get("delta_minutes", 180)) _context_snapshot = { "cycle_meta": cycle_meta, "macro_regime": { "dominant": (macro_regime or {}).get("scenarios", {}).get("dominant"), "scores": (macro_regime or {}).get("scenarios", {}).get("scores"), "gauges_summary": {k: v.get("value") for k, v in ((macro_regime or {}).get("gauges", {})).items()}, }, "geo_score": geo_score_obj, "news_partitioned": { "inter_cycle": [{"title": n.get("title"), "source": n.get("source"), "decayed_score": n.get("decayed_score"), "age_hours": n.get("age_hours")} for n in _news_snap["inter_cycle"][:10]], "recent_24h": [{"title": n.get("title"), "source": n.get("source"), "decayed_score": n.get("decayed_score")} for n in _news_snap["recent_24h"][:10]], "older": [{"title": n.get("title"), "source": n.get("source")} for n in _news_snap["older"][:5]], }, "fred_releases": _fred_releases, "price_discovery": [ {k: v for k, v in a.items() if k != "article_hash"} for a in _absorptions[:10] ], "tech_indicators_block": _tech_block, "iv_context_preview": iv_context[:500] if iv_context else "", "calendar": calendar[:8] if calendar else [], "quotes_summary": {cls: [{"symbol": q.get("symbol"), "price": q.get("price"), "change_pct": q.get("change_pct")} for q in qs[:3]] for cls, qs in quotes.items()}, "portfolio_open_positions": { "count": len(_open_trades_snap), "concentration": _portfolio_conc, "trades": [ {k: v for k, v in t.items() if k != "id"} for t in _open_trades_snap ], }, } save_cycle_context_snapshot(run_id, _context_snapshot) logger.info(f"[Cycle {run_id[:16]}] Context snapshot saved") 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}") # ── Build institutional reports block ────────────────────────── _institutional_block = "" try: from services.ai_analyzer import build_institutional_block _institutional_block = build_institutional_block(days=step_cfg["institutional_block"]["lookback_days"]) if _institutional_block: logger.info(f"[Cycle {run_id[:16]}] Institutional block injected") except Exception as _ibe: logger.warning(f"[Cycle] Institutional block failed (non-blocking): {_ibe}") # ── COT + Forward Curves (background refresh, non-blocking) ─────────────── try: from services.cot_fetcher import fetch_all_cot from services.forward_curve import fetch_forward_curves from services.database import save_cot_data, save_forward_curves as _sfc cot_data = fetch_all_cot() if cot_data: save_cot_data(cot_data) curve_data = fetch_forward_curves() if curve_data: _sfc(curve_data) except Exception as _e: import logging as _logging _logging.getLogger(__name__).warning(f"COT/Curve refresh failed (non-blocking): {_e}") 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, iv_context=iv_context, cycle_meta=cycle_meta, tech_indicators_block=_tech_block, 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: 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 (Embedding cosinus ou Jaccard fallback) ─ 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: if not _is_duplicate_pattern(s, existing, ai_key, jaccard_threshold=sim_threshold, embed_threshold=step_cfg["duplicate_detection"]["embedding_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}") # ── 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')}' — doublon détecté") 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: from services.database import get_risk_clusters as _grc _clusters = _grc() risk_cluster_context = _clusters.get("risk_prompt_context", "") if risk_cluster_context: logger.info( f"[Cycle {run_id[:16]}] Risk clusters: " f"{len(_clusters.get('clusters', []))} facteurs, " f"saturés={_clusters.get('saturated_factors', [])}" ) except Exception as _re: logger.warning(f"[Cycle] Risk cluster context failed (non-blocking): {_re}") # ── Step 3.6: Refresh IV context (with full trade+watchlist scope) ────── 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=step_cfg["iv_context"]["lookback_days"]) _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") # Invalidate the options_vol watchlist cache so the frontend sees fresh data try: from routers.options_vol import _iv_cache stale_keys = [k for k in list(_iv_cache.keys()) if k.startswith("watchlist:") or k.startswith("snapshot:")] for k in stale_keys: _iv_cache.pop(k, None) if stale_keys: logger.info(f"[Cycle] IV cache invalidated ({len(stale_keys)} entries)") except Exception: pass 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)") from services.database import get_analysis_config template = get_analysis_config().get("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, risk_context=risk_cluster_context, cycle_meta=cycle_meta, tech_indicators_block=_tech_block, fred_block=_fred_block, price_discovery_block=_price_discovery_block, portfolio_context_block=_portfolio_block, institutional_block=_institutional_block, run_id=run_id, ) 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", []) # 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") 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) _options_assessment = None # ── IV Gate: block ALERT-verdict trades before logging ──────────────── _iv_gate_blocked: List[Dict] = [] try: from services.database import get_config as _gc if (_gc("iv_gate_enabled") or "true").lower() == "true": # Augment _iv_snapshots with any new tickers from scored that weren't in the pre-fetch from services.iv_engine import get_full_iv_snapshot as _gfiv _scored_tickers = { (tr.get("underlying") or "").upper() for sp in scored for tr in (sp.get("trade_rankings") or sp.get("suggested_trades") or []) if tr.get("underlying") } for _stk in _scored_tickers - set(_iv_snapshots.keys()): try: _iv_snapshots[_stk] = _gfiv(_stk) except Exception: pass scored, _iv_gate_blocked = _apply_iv_gate(scored, _iv_snapshots) if _iv_gate_blocked: summary["iv_gate_blocked"] = len(_iv_gate_blocked) logger.warning( f"[IVGate] {len(_iv_gate_blocked)} trade(s) blocked by IV rules: " + ", ".join(f"{b['ticker']} {b['strategy']}" for b in _iv_gate_blocked) ) for _blk in _iv_gate_blocked: try: from services.database import log_skipped_trade log_skipped_trade( run_id=scoring_run_id, pattern_id=_blk.get("pattern_id", ""), pattern_name=_blk.get("pattern_name", ""), underlying=_blk.get("ticker", ""), strategy=_blk.get("strategy", ""), score=_blk.get("score", 0), expected_move_pct=0, skip_detail=( f"[IV_GATE] ALERT — {'; '.join(_blk.get('iv_issues', [])[:2])}. " f"Stratégie optimale: {_blk.get('optimal_strategy', '?')}" ), asset_class=_blk.get("asset_class", ""), ) except Exception: pass except Exception as _ige: logger.warning(f"[IVGate] Failed (non-blocking): {_ige}") log_trade_entries(run_id=scoring_run_id, scored_patterns=scored, quotes=quotes) # ── Step 5.2: Options Technical Agent — validate newly logged trades ── try: from services.options_technical_agent import assess_logged_trades, save_assessments_to_db _options_assessment = assess_logged_trades( scoring_run_id=scoring_run_id, scored=scored, ai_key=ai_key, ) if _options_assessment and _options_assessment.get("assessments"): save_assessments_to_db(_options_assessment["assessments"], scoring_run_id) summary["options_assessment"] = { "n_ok": _options_assessment.get("n_ok", 0), "n_warn": _options_assessment.get("n_warn", 0), "n_alert": _options_assessment.get("n_alert", 0), "global_score": _options_assessment.get("global_score"), } logger.info( f"[OptionsTech] Assessment done — OK={_options_assessment.get('n_ok')} " f"WARN={_options_assessment.get('n_warn')} ALERT={_options_assessment.get('n_alert')}" ) except Exception as _ota: logger.warning(f"[OptionsTech] Agent failed (non-blocking): {_ota}") # ── Step 5.1: Portfolio monitor — conflict & concentration check ────── try: from services.portfolio_risk import analyze_simulation_portfolio _risk = analyze_simulation_portfolio() if not step_cfg["portfolio_monitor"]["enabled"]: logger.info("[PortfolioMonitor] Disabled via cycle_step_config — skipping AI monitor") elif _risk.get("alerts"): logger.info(f"[PortfolioMonitor] {len(_risk['alerts'])} alerts ({len(_risk['conflicts'])} conflicts) — running AI monitor") _pm = _run_portfolio_monitor(_risk, scoring_run_id) if _pm: summary["portfolio_monitor"] = { "alerts": len(_risk["alerts"]), "conflicts": len(_risk["conflicts"]), "assessment": _pm.get("assessment", ""), "actions_count": len(_pm.get("actions", [])), } else: logger.info(f"[PortfolioMonitor] OK — {_risk['open_count']} positions, no alerts") except Exception as _pme: logger.warning(f"[PortfolioMonitor] Failed (non-blocking): {_pme}") # Auto-add any new underlying tickers to the IV watchlist if not step_cfg["watchlist_auto_add"]["enabled"]: logger.info("[Watchlist] Auto-add disabled via cycle_step_config — skipping") else: try: from services.database import _normalize_ticker, add_watchlist_ticker, get_watchlist_tickers from services.iv_engine import _resolve_ticker, bootstrap_iv_history existing = set(get_watchlist_tickers()) new_proxies = set() for sp in scored: for trade in (sp.get("trade_rankings") or sp.get("suggested_trades") or []): underlying = trade.get("underlying") or sp.get("underlying") or "" if underlying: normalized = _normalize_ticker(underlying.upper()) # WHEAT→ZW=F, EUR/USD→EURUSD=X proxy = _resolve_ticker(normalized) # ZW=F→WEAT, EURUSD=X→FXE if proxy not in existing: new_proxies.add(proxy) for proxy in new_proxies: if add_watchlist_ticker(proxy, added_by="cycle"): logger.info(f"[Watchlist] Auto-added new ticker: {proxy}") log_system_event("INFO", "auto_cycle", f"Nouveau ticker ajouté à la watchlist IV: {proxy}", cycle_id=scoring_run_id, ticker=proxy) if new_proxies: bootstrap_iv_history(tickers=list(new_proxies), min_existing=0) except Exception as _we: logger.warning(f"[Watchlist] Auto-add failed: {_we}") 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 5.5: Bayesian posterior update (Sprint 4.1) ───────────────── try: from services.database import update_bayesian_posteriors _n_updated = update_bayesian_posteriors() if _n_updated: logger.info(f"[Cycle {run_id[:16]}] Bayesian posteriors mis à jour : {_n_updated} patterns") except Exception as _be: logger.warning(f"[Cycle] Bayesian update failed (non-blocking): {_be}") # ── Step 5.6: Régime clustering (Sprint 4.2) ────────────────────────── if not step_cfg["regime_clustering"]["enabled"]: logger.info("[Cycle] Régime clustering disabled via cycle_step_config — skipping") else: try: from services.database import detect_and_save_regime_clusters _cluster_result = detect_and_save_regime_clusters( n_clusters=step_cfg["regime_clustering"]["n_clusters"], days=step_cfg["regime_clustering"]["lookback_days"], ) if "current_cluster" in _cluster_result: logger.info( f"[Cycle {run_id[:16]}] Régime cluster : {_cluster_result.get('current_label')} " f"(cluster {_cluster_result.get('current_cluster')}, " f"anomalie={_cluster_result.get('current_anomaly')})" ) except Exception as _ce: logger.warning(f"[Cycle] Régime clustering failed (non-blocking): {_ce}") # ── Step 5.7: Wavelet signal scan (watchlist instruments) ──────────── _wavelet_results: list = [] try: from services.wavelet_signals import compute_and_save_wavelet_signals _wavelet_results = compute_and_save_wavelet_signals(run_id) summary["wavelet_signals_count"] = len(_wavelet_results) logger.info(f"[Cycle {run_id[:16]}] Wavelet signals: {len(_wavelet_results)} detected") except Exception as _we: logger.warning(f"[Cycle] Wavelet signal scan failed (non-blocking): {_we}") summary["wavelet_signals_count"] = 0 # ── 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", wavelet_signals_count=summary.get("wavelet_signals_count", 0), ) _current_status["last_run_at"] = datetime.utcnow().isoformat() logger.info(f"[Cycle {run_id[:16]}] Completed — {added_count} new patterns, {len(scored)} scored") # ── Step 6.1: Institutional report absorption tracking ─────────────── try: from services.database import get_conn as _get_conn import json as _json _commentary_text = "" if commentary: try: _c = _json.loads(commentary) if isinstance(commentary, str) else commentary _commentary_text = " ".join([ str(_c.get("narrative", "")), str(_c.get("key_catalyst", "")), " ".join(str(x) for x in (_c.get("risks", []) or [])), ]).lower() except Exception: _commentary_text = str(commentary).lower() if _commentary_text: _conn = _get_conn() try: cutoff = (datetime.utcnow() - timedelta(days=step_cfg["institutional_absorption"]["lookback_days"])).strftime("%Y-%m-%d") _inst_rows = _conn.execute( "SELECT id, key_points_json FROM institutional_reports " "WHERE report_date >= ? AND (absorbed_score IS NULL OR absorbed_score = 0)", (cutoff,), ).fetchall() for _ir in _inst_rows: try: _kps = _json.loads(_ir["key_points_json"] or "[]") _kp_words = set( w for kp in _kps for w in kp.lower().split() if len(w) > 4 ) if not _kp_words: continue _overlap = sum(1 for w in _kp_words if w in _commentary_text) _score = min(1.0, _overlap / max(len(_kp_words), 1)) if _score > 0: _conn.execute( "UPDATE institutional_reports SET absorbed_score=?, injected_in_cycle_id=? WHERE id=?", (round(_score, 3), run_id, _ir["id"]), ) except Exception: pass _conn.commit() logger.info(f"[Cycle {run_id[:16]}] Absorption tracked for {len(_inst_rows)} institutional reports") finally: _conn.close() except Exception as _abs_e: logger.warning(f"[Cycle] Absorption tracking failed (non-blocking): {_abs_e}") # ── Step 6.5: Generate full cycle report ───────────────────────────── try: _cycle_report = _generate_cycle_report( run_id=run_id, scored=scored, dominant=dominant, scenarios=scenarios, geo_score_val=geo_score_val, news=news, gauges=gauges, ai_key=ai_key, added_patterns=[s for s in suggestions if s.get("id")], scoring_run_id=scoring_run_id, portfolio_monitor=summary.get("portfolio_monitor"), commentary=commentary, options_assessment=_options_assessment, wavelet_signals=_wavelet_results, ) if _cycle_report: from services.database import save_cycle_report save_cycle_report(run_id, _cycle_report) logger.info(f"[Cycle {run_id[:16]}] Cycle report saved") except Exception as _rpe: logger.warning(f"[Cycle] Cycle report failed (non-blocking): {_rpe}") # ── 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 entries = get_trade_entry_prices(get_cycle_step_config()["cycle_commentary"]["lookback_days"]) 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 # ── IV Gate ─────────────────────────────────────────────────────────────────── def _apply_iv_gate( scored: List[Dict], iv_snapshots: Dict[str, Dict], ) -> tuple: """ Filter ALERT-verdict trades from each scored pattern's suggested_trades/trade_rankings before they reach log_trade_entries. Returns (filtered_scored, blocked_trades_list). IVR thresholds are read from app_config (iv_gate_ivr_high, iv_gate_ivr_extreme, iv_gate_skew_threshold) so they can be tuned from the Config page. """ from services.options_technical_agent import assess_strategy_fit # Read thresholds from config (fall back to hardcoded defaults) try: from services.database import get_config as _gc _ivr_high = float(_gc("iv_gate_ivr_high") or 60) _ivr_extreme = float(_gc("iv_gate_ivr_extreme") or 80) _skew_thresh = float(_gc("iv_gate_skew_threshold") or 8) except Exception: _ivr_high, _ivr_extreme, _skew_thresh = 60.0, 80.0, 8.0 # Monkey-patch thresholds into the rule engine for this call import services.options_technical_agent as _ota_mod _orig_ivr_high = getattr(_ota_mod, "_IVR_HIGH", None) try: _ota_mod._IVR_HIGH = _ivr_high _ota_mod._IVR_EXTREME = _ivr_extreme _ota_mod._SKEW_THRESH = _skew_thresh except Exception: pass # Detect weekend: IVR=100 on weekends is artificial (weekend premium) — ignore rank from datetime import datetime as _dt_cls _weekend_now = _dt_cls.utcnow().weekday() >= 5 all_blocked: List[Dict] = [] for sp in scored: trade_key = "trade_rankings" if "trade_rankings" in sp else "suggested_trades" trades = sp.get(trade_key) or [] allowed = [] for trade in trades: ticker = (trade.get("underlying") or trade.get("ticker") or "").upper() strategy = trade.get("strategy") or "Long Call" snap = iv_snapshots.get(ticker, {}) if not snap: # No IV data → can't block; allow but don't judge allowed.append(trade) continue iv_rank = snap.get("iv_rank") # On weekends, ATM IV is inflated by weekend premium (market makers can't hedge). # IVR=99-100 on a weekend is almost always artificial — don't block on it. if _weekend_now and iv_rank is not None and iv_rank >= 99.0: iv_rank = None snap = {**snap, "iv_rank": None} iv_current_pct = snap.get("iv_current_pct") iv_min_52w = snap.get("iv_min_52w_pct") iv_max_52w = snap.get("iv_max_52w_pct") skew_pct = (snap.get("skew") or {}).get("skew_pct") term_structure = (snap.get("term_structure") or {}).get("structure") flow_bias = (snap.get("options_flow") or {}).get("flow_bias") result = assess_strategy_fit( strategy=strategy, iv_rank=iv_rank, iv_current_pct=iv_current_pct, iv_min_52w=iv_min_52w, iv_max_52w=iv_max_52w, skew_pct=skew_pct, term_structure=term_structure, flow_bias=flow_bias, ) if result["verdict"] == "ALERT": blocked_entry = { "ticker": ticker, "strategy": strategy, "pattern_id": sp.get("pattern_id"), "pattern_name": sp.get("geo_trigger") or sp.get("pattern_name") or "", "score": sp.get("score", 0), "asset_class": trade.get("asset_class") or sp.get("asset_class") or "", "iv_rank": iv_rank, "iv_issues": result["issues"], "optimal_strategy": result["optimal_strategy"], "fit_score": result["fit_score"], } all_blocked.append(blocked_entry) logger.warning( f"[IVGate] BLOCKED {ticker} {strategy} (IVR={iv_rank}, score={result['fit_score']}): " f"{result['issues'][0][:80] if result['issues'] else 'ALERT'}" ) else: allowed.append(trade) sp[trade_key] = allowed return scored, all_blocked # ── Cycle report generation ─────────────────────────────────────────────────── def _generate_cycle_report( run_id: str, scored: List[Dict], dominant: str, scenarios: Dict, geo_score_val: int, news: List[Dict], gauges: Dict, ai_key: str, added_patterns: List[Dict], scoring_run_id: str, portfolio_monitor: Optional[Dict], commentary: Optional[str], options_assessment: Optional[Dict] = None, wavelet_signals: Optional[List[Dict]] = None, ) -> Optional[Dict]: """ Build the full cycle report dict: - PnL + VaR snapshots (timestamped with this cycle) - Delta vs previous cycle (patterns added, trades logged, trades closed) - GPT-4o context narrative (what was in context, what it used, why) - Risk summary """ import json as _json try: from services.database import get_cycle_runs, get_trade_entries_by_run except ImportError: from services.database import get_cycle_runs get_trade_entries_by_run = None # ── Get previous cycle timestamp ────────────────────────────────────────── prev_cycle_at: Optional[str] = None try: history = get_cycle_runs(limit=5) prev_cycles = [r for r in history if r["run_id"] != run_id and r.get("status") == "completed"] if prev_cycles: prev_cycle_at = prev_cycles[0].get("started_at") except Exception: pass # ── PnL snapshot ────────────────────────────────────────────────────────── pnl_snapshot_id: Optional[int] = None pnl_summary: Dict = {} try: from services.var_service import save_pnl_snapshot, get_latest_pnl_snapshot pnl_snapshot_id = save_pnl_snapshot() snap = get_latest_pnl_snapshot() if snap: pnl_summary = { "total_pnl_pct": snap.get("total_pnl_pct"), "total_pnl_eur": snap.get("total_pnl_eur"), "total_capital_eur": snap.get("total_capital_eur"), "n_open": snap.get("n_open"), "n_closed": snap.get("n_closed"), } logger.info(f"[CycleReport] PnL snapshot saved id={pnl_snapshot_id}") except Exception as _e: logger.warning(f"[CycleReport] PnL snapshot failed: {_e}") # ── VaR snapshot ────────────────────────────────────────────────────────── var_snapshot_id: Optional[int] = None var_summary: Dict = {} try: from services.var_service import compute_var, save_var_snapshot _var_cfg = get_cycle_step_config()["var_snapshot"] var_result = compute_var( confidence=_var_cfg["confidence"], horizon_days=_var_cfg["horizon_days"], lookback_days=_var_cfg["lookback_days"], default_iv=_var_cfg["default_iv"], ) if "error" not in var_result: var_snapshot_id = save_var_snapshot(var_result, 0.95, 1, 252, 0.20) var_summary = { "hist_var_1d_pct": var_result.get("hist_var_1d_pct"), "hist_cvar_pct": var_result.get("hist_cvar_pct"), "mc_var_1d_pct": var_result.get("mc_var_1d_pct"), "n_positions": var_result.get("n_positions"), } logger.info(f"[CycleReport] VaR snapshot saved id={var_snapshot_id}") except Exception as _e: logger.warning(f"[CycleReport] VaR snapshot failed: {_e}") # ── Trades delta ────────────────────────────────────────────────────────── trades_logged: List[Dict] = [] trades_closed: List[Dict] = [] try: from services.database import get_conn _conn = get_conn() # Trades logged this cycle (by scoring_run_id) _rows = _conn.execute( """SELECT underlying, strategy, entry_date, pnl_pct, score_at_entry, pattern_name FROM trade_entry_prices WHERE run_id=? ORDER BY entry_date DESC""", (scoring_run_id,), ).fetchall() trades_logged = [dict(r) for r in _rows] # Trades closed since previous cycle if prev_cycle_at: _closed = _conn.execute( """SELECT underlying, strategy, closed_at, pnl_pct, pattern_name FROM trade_entry_prices WHERE status='closed' AND closed_at >= ? ORDER BY closed_at DESC""", (prev_cycle_at,), ).fetchall() trades_closed = [dict(r) for r in _closed] _conn.close() except Exception as _e: logger.warning(f"[CycleReport] Delta trades query failed: {_e}") # ── Context narrative (GPT-4o) ──────────────────────────────────────────── context_narrative: Dict = {} try: from services.ai_analyzer import _chat top_news_ctx = [ {"title": n.get("title", "")[:100], "impact": round(float(n.get("impact_score") or 0), 2)} for n in sorted(news, key=lambda x: -(float(x.get("impact_score") or 0)))[:8] ] def _gv(k: str) -> str: v = gauges.get(k, {}).get("value") return str(round(v, 2)) if v is not None else "N/A" def _gc(k: str) -> str: v = gauges.get(k, {}).get("change_pct") return f"{v:+.2f}%" if v is not None else "N/A" new_pattern_names = [p.get("name", "") for p in added_patterns] top_scored = sorted(scored, key=lambda x: -(x.get("score") or 0))[:5] ctx_prompt = f"""Tu es un analyste stratégique qui documente le RAISONNEMENT d'un cycle d'analyse géo-macro. CONTEXTE TRANSMIS À L'IA CE CYCLE: - Régime macro 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')} | S&P/200j: {_gv('spx_vs_200d')}% - Cuivre: {_gc('copper')} | Or: {_gc('gold')} | DXY: {_gc('dxy')} | Brent: {_gc('brent')} - Top 8 news géopolitiques transmises: {_json.dumps(top_news_ctx, ensure_ascii=False)} NOUVELLES IDÉES GÉNÉRÉES CE CYCLE ({len(new_pattern_names)} patterns ajoutés): {_json.dumps(new_pattern_names, ensure_ascii=False)} TOP 5 PATTERNS LES MIEUX SCORÉS MAINTENANT: {_json.dumps([{"name": s.get("geo_trigger","?"), "score": s.get("score"), "catalyst": s.get("key_catalyst","")[:80]} for s in top_scored], ensure_ascii=False)} Ta tâche: Explique le RAISONNEMENT de ce cycle. Pour chaque nouvelle idée générée: 1. Quels éléments SPÉCIFIQUES du contexte transmis (news, macro, gauges) l'ont motivée ? 2. Qu'est-ce qui vient de ta connaissance GÉNÉRALE des marchés (pas du contexte transmis) ? 3. Quels signaux étaient les plus déterminants ? Réponds en JSON avec ce schéma EXACT: {{ "narrative": "", "context_driven": ["", ...], "general_knowledge": ["", ...], "key_signals": ["<3-5 signaux les plus décisifs>"], "context_log": {{ "macro_regime": "{dominant}", "geo_score": {geo_score_val}, "dominant_news": ["", "", ""], "key_gauges": {{"vix": {_gv('vix')}, "slope": {_gv('slope_10y3m')}, "copper_chg": {_gc('copper')}, "gold_chg": {_gc('gold')}}} }} }}""" result = _chat( "Tu es un analyste macro-géopolitique. Documente le raisonnement du cycle en JSON.", ctx_prompt, model="gpt-4o", json_mode=True, max_tokens=800, ) if result and result.get("narrative"): context_narrative = result logger.info(f"[CycleReport] Context narrative generated ({len(result.get('narrative',''))} chars)") except Exception as _e: logger.warning(f"[CycleReport] Context narrative failed: {_e}") context_narrative = {"narrative": "", "context_driven": [], "general_knowledge": [], "key_signals": []} # ── Parse existing commentary ───────────────────────────────────────────── commentary_parsed: Dict = {} if commentary: try: commentary_parsed = _json.loads(commentary) if isinstance(commentary, str) else commentary except Exception: commentary_parsed = {"commentary": str(commentary)} # ── Calibration report ─────────────────────────────────────────────────── calibration_report: Dict = {} try: from services.database import get_calibration_summary as _get_calib calib_rows = _get_calib() if calib_rows: pure_ai = [r for r in calib_rows if r["source"] == "pure_ai"] early = [r for r in calib_rows if r["source"] == "early"] mixed = [r for r in calib_rows if r["source"] == "mixed"] driven = [r for r in calib_rows if r["source"] == "data_driven"] with_data = [r for r in calib_rows if (r["n_mature_trades"] or 0) > 0] avg_w = ( round(sum(r["calibration_weight"] for r in with_data) / len(with_data), 3) if with_data else 0.0 ) calibration_report = { "total_patterns": len(calib_rows), "pure_ai_count": len(pure_ai), "early_count": len(early), "mixed_count": len(mixed), "data_driven_count": len(driven), "avg_calibration_weight": avg_w, "avg_calibration_weight_pct": round(avg_w * 100, 1), "patterns_with_data": len(with_data), "detail": [ { "name": r["pattern_name"], "asset_class": r["asset_class"], "source": r["source"], "weight_pct": r["calibration_weight_pct"], "ai_estimate": r["ai_estimate"], "observed": r["observed_avg_win_pct"], "calibrated": r["calibrated_expected_move"], "n_trades": r["n_mature_trades"], "win_rate_pct": round((r["bayes_win_rate"] or 0) * 100, 1), } for r in calib_rows if (r["n_mature_trades"] or 0) > 0 ][:15], } logger.info( f"[CycleReport] Calibration: {len(pure_ai)} pure-AI, " f"{len(mixed)+len(driven)} with data, avg_weight={avg_w:.1%}" ) except Exception as _ce: logger.warning(f"[CycleReport] Calibration report failed: {_ce}") # ── Assemble full report ────────────────────────────────────────────────── report = { "run_id": run_id, "macro_dominant": dominant, "geo_score": geo_score_val, "macro_scores": scenarios.get("scores", {}), "macro_reasons": scenarios.get("reasons", {}).get(dominant, []), # Snapshots "pnl_snapshot_id": pnl_snapshot_id, "var_snapshot_id": var_snapshot_id, "pnl_summary": pnl_summary, "var_summary": var_summary, # Delta "patterns_added": len(added_patterns), "patterns_added_list": [ {"name": p.get("name"), "description": (p.get("description") or "")[:150], "asset_class": p.get("asset_class"), "macro_fit": p.get("macro_fit"), "triggers": p.get("triggers", [])[:3]} for p in added_patterns ], "trades_logged": len(trades_logged), "trades_logged_list": trades_logged[:10], "trades_closed": len(trades_closed), "trades_closed_list": trades_closed[:10], "prev_cycle_at": prev_cycle_at, # Context narrative "context_narrative": context_narrative, # Cycle commentary (existing) "commentary": commentary_parsed, # Risk "portfolio_monitor": portfolio_monitor, "risk_alerts": portfolio_monitor.get("alerts") if portfolio_monitor else None, # Wavelet signals detected this cycle (watchlist scan, see services.wavelet_signals) "wavelet_signals": wavelet_signals or [], # Top scored patterns "top_scored": [ {"name": s.get("geo_trigger"), "score": s.get("score"), "summary": (s.get("summary") or "")[:120], "key_catalyst": s.get("key_catalyst", "")} for s in sorted(scored, key=lambda x: -(x.get("score") or 0))[:5] ], # Options Technical Agent assessment "options_technical": { "global_assessment": (options_assessment or {}).get("global_assessment", ""), "global_score": (options_assessment or {}).get("global_score"), "n_ok": (options_assessment or {}).get("n_ok", 0), "n_warn": (options_assessment or {}).get("n_warn", 0), "n_alert": (options_assessment or {}).get("n_alert", 0), "assessments": (options_assessment or {}).get("assessments", []), } if options_assessment is not None else None, # Calibration: AI vs observed expected_move blend "calibration_report": calibration_report, } return report def generate_standalone_report() -> Dict[str, Any]: """Cycle Actions — standalone "generate-report" action. _generate_cycle_report() itself is NOT modified (too tightly coupled to run_cycle_once()'s in-memory state to safely change) — instead every one of its parameters is reconstructed from an independent, DB-or-live source: - scored <- get_last_scores() (config key "last_pattern_scores", filled by save_pattern_scores() every real cycle) - news/geo_score <- same live sequence as cycle Step 1 - dominant/scenarios/gauges <- same sequence as the update-regime action - commentary <- _generate_cycle_commentary(), independently callable - wavelet_signals <- compute_and_save_wavelet_signals(), independently callable - portfolio_monitor <- analyze_simulation_portfolio() (+ AI only if alerts) - added_patterns/options_assessment <- cycle-only artifacts, not reconstructible after the fact -> empty/None (report is honest, just thinner on these two fields than a live cycle's report).""" from services.database import get_config, get_pattern_scores, save_cycle_report from services.data_fetcher import fetch_geo_news, get_macro_gauges, score_macro_scenarios from services.geo_analyzer import compute_geo_risk_score from services.ai_analyzer import ai_score_news_batch, ai_score_geo_risk run_id = datetime.utcnow().isoformat() ai_key = get_config("openai_api_key") or "" last_scores = get_pattern_scores() scored = last_scores.get("scores") or [] scoring_run_id = last_scores.get("run_id") or run_id news = ai_score_news_batch(fetch_geo_news()) algo_geo = compute_geo_risk_score(news) try: geo_obj = ai_score_geo_risk(news, algo_geo, log_meta={"run_id": run_id, "call_type": "geo_risk_score"}) except Exception: geo_obj = algo_geo geo_score_val = int(round(geo_obj.get("score") or 0)) gauges = get_macro_gauges() scenarios = score_macro_scenarios(gauges) dominant = scenarios.get("dominant", "incertain") commentary = None try: commentary = _generate_cycle_commentary(scored, dominant, scenarios, geo_score_val, news, gauges) except Exception as e: logger.warning(f"[StandaloneReport] Commentary generation failed: {e}") wavelet_signals: List[Dict] = [] try: from services.wavelet_signals import compute_and_save_wavelet_signals wavelet_signals = compute_and_save_wavelet_signals(run_id) except Exception as e: logger.warning(f"[StandaloneReport] Wavelet scan failed (non-blocking): {e}") portfolio_monitor = None try: from services.portfolio_risk import analyze_simulation_portfolio risk = analyze_simulation_portfolio() if risk.get("alerts"): portfolio_monitor = _run_portfolio_monitor(risk, run_id) except Exception as e: logger.warning(f"[StandaloneReport] Portfolio monitor failed (non-blocking): {e}") report = _generate_cycle_report( run_id=run_id, scored=scored, dominant=dominant, scenarios=scenarios, geo_score_val=geo_score_val, news=news, gauges=gauges, ai_key=ai_key, added_patterns=[], scoring_run_id=scoring_run_id, portfolio_monitor=portfolio_monitor, commentary=commentary, options_assessment=None, wavelet_signals=wavelet_signals, ) if report: save_cycle_report(run_id, report) return {"run_id": run_id, "report": report} # ── 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=get_cycle_step_config()["iv_context"]["lookback_days"], 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, ) synth_cfg = get_cycle_step_config()["knowledge_synthesis"] if not synth_cfg["enabled"]: logger.info("[AutoSynth] Disabled via cycle_step_config — skipping") return # Skip if last synthesis is younger than the configured threshold min_hours = synth_cfg["min_hours_between"] 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 < min_hours: logger.info( f"[AutoSynth] Super Contexte is {age_h:.1f}h old — skipping re-synthesis (threshold: {min_hours}h)" ) return except Exception: pass reports = list_ai_reports(limit=10) mtm_data = get_mtm_trades_with_traces(days=get_cycle_step_config()["iv_context"]["lookback_days"]) 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 _parse_weekend_times(times_str: str): """Parse 'HH:MM,HH:MM' into list of (hour, minute) tuples.""" result = [] for t in times_str.split(","): t = t.strip() try: parts = t.split(":") result.append((int(parts[0]), int(parts[1]))) except Exception: pass return result or [(8, 0), (22, 0)] def _next_weekend_slot(now_utc: datetime, weekend_times): """Return (next_dt, wait_secs) for the next weekend cycle time (UTC).""" from datetime import timedelta today = now_utc.date() candidates = [] for day_offset in range(3): for h, m in weekend_times: candidate = datetime(today.year, today.month, today.day, h, m) + timedelta(days=day_offset) if candidate > now_utc + timedelta(minutes=5): candidates.append(candidate) if not candidates: fallback = now_utc + timedelta(hours=12) return fallback, 12 * 3600 next_slot = min(candidates) return next_slot, max(60, (next_slot - now_utc).total_seconds()) def _scheduler_loop(stop_event: threading.Event): """Background loop that runs the cycle at the configured interval. Weekday: fires every auto_cycle_hours. Weekend: fires only at weekend_cycle_times (UTC HH:MM list). If weekend_cycle_enabled=false, sleeps until Monday. """ from services.database import get_config from datetime import timedelta while not stop_event.is_set(): try: interval_hours = float(get_config("auto_cycle_hours") or "3") weekend_enabled = (get_config("weekend_cycle_enabled") or "true").lower() == "true" weekend_times_str = get_config("weekend_cycle_times") or "08:00,22:00" except Exception: interval_hours, weekend_enabled, weekend_times_str = 3.0, True, "08:00,22:00" weekend_times = _parse_weekend_times(weekend_times_str) now_utc = datetime.utcnow() weekday = now_utc.weekday() is_weekend = weekday >= 5 if is_weekend and not weekend_enabled: # Sleep until Monday 00:05 UTC days_to_monday = 7 - weekday # Sat→2, Sun→1 next_monday = datetime(now_utc.year, now_utc.month, now_utc.day, 0, 5) + timedelta(days=days_to_monday) wait_secs = max(60, (next_monday - now_utc).total_seconds()) _current_status["next_run_at"] = next_monday.isoformat() _current_status["interval_hours"] = interval_hours logger.info(f"[Scheduler] Weekend — cycles désactivés. Reprise lundi {next_monday.strftime('%Y-%m-%d %H:%M')} UTC") stop_event.wait(timeout=wait_secs) continue # don't run a cycle, loop back if is_weekend: next_slot, wait_secs = _next_weekend_slot(now_utc, weekend_times) _current_status["next_run_at"] = next_slot.isoformat() _current_status["interval_hours"] = interval_hours logger.info(f"[Scheduler] Weekend — prochain cycle à {next_slot.strftime('%H:%M')} UTC ({wait_secs/3600:.1f}h)") else: # Account for time already elapsed since last cycle so a restart # doesn't reset the countdown — fire as soon as the interval is due. last_run_iso = _current_status.get("last_run_at") if not last_run_iso: # Also check DB for the most recent completed run try: from services.database import get_cycle_runs recent = get_cycle_runs(limit=1) if recent: last_run_iso = recent[0].get("created_at") or recent[0].get("run_id", "")[:19].replace("T", " ") except Exception: pass elapsed_secs = 0.0 if last_run_iso: try: last_dt = datetime.fromisoformat(last_run_iso.replace(" ", "T").rstrip("Z")) elapsed_secs = (now_utc - last_dt).total_seconds() except Exception: pass wait_secs = max(60, interval_hours * 3600 - elapsed_secs) next_run = now_utc + timedelta(seconds=wait_secs) _current_status["next_run_at"] = next_run.isoformat() _current_status["interval_hours"] = interval_hours logger.info(f"[Scheduler] Prochain cycle dans {wait_secs/3600:.1f}h ({next_run.strftime('%H:%M')} UTC) — elapsed since last: {elapsed_secs/3600:.1f}h") stop_event.wait(timeout=wait_secs) 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") retention_days = int(get_config("journal_retention_days") or "90") maturity_pct = int(get_config("maturity_threshold_pct") or "35") trade_budget_eur = float(get_config("trade_budget_eur") or "5000") preferred_horizon_min = int(get_config("preferred_horizon_min") or "30") preferred_horizon_max = int(get_config("preferred_horizon_max") or "180") except Exception: interval_hours, enabled, sim_threshold, min_ev, min_score = 3.0, False, 0.30, 0.0, 0 retention_days, maturity_pct = 90, 35 trade_budget_eur, preferred_horizon_min, preferred_horizon_max = 5000.0, 30, 180 recent = get_cycle_runs(limit=1) last = recent[0] if recent else None # Enrich last_cycle with the scoring_run_id used for trade_entry_prices if last: try: from services.database import get_conn _conn = get_conn() started = (last.get("started_at") or "").replace(" ", "T") completed = last.get("completed_at") or "" if started and completed: row = _conn.execute( "SELECT run_id FROM trade_entry_prices WHERE run_id >= ? AND run_id <= ?" " ORDER BY run_id DESC LIMIT 1", (started, completed), ).fetchone() if row: last = dict(last) last["scoring_run_id"] = row["run_id"] except Exception: pass try: weekend_enabled = (get_config("weekend_cycle_enabled") or "true").lower() == "true" weekend_cycle_times = get_config("weekend_cycle_times") or "08:00,22:00" except Exception: weekend_enabled, weekend_cycle_times = True, "08:00,22:00" return { **_current_status, "enabled": enabled, "interval_hours": interval_hours, "similarity_threshold": sim_threshold, "min_ev_threshold": min_ev, "min_score_threshold": min_score, "journal_retention_days": retention_days, "maturity_threshold_pct": maturity_pct, "trade_budget_eur": trade_budget_eur, "preferred_horizon_min": preferred_horizon_min, "preferred_horizon_max": preferred_horizon_max, "weekend_cycle_enabled": weekend_enabled, "weekend_cycle_times": weekend_cycle_times, "cycle_step_config": get_cycle_step_config(), "last_cycle": last, "scheduler_alive": bool(_cycle_thread and _cycle_thread.is_alive()), }