feat: cycle

backend/routers/cycle.py

@@ -1,8 +1,8 @@
 from fastapi import APIRouter, HTTPException
 from pydantic import BaseModel
-from typing import Optional
+from typing import Any, Dict, Optional
 from services.database import get_cycle_runs, get_cycle_run, set_config, get_config, list_cycle_context_snapshots, get_cycle_context_snapshot, get_ai_call_logs
-from services.auto_cycle import get_status, trigger_manual, restart_scheduler
+from services.auto_cycle import get_status, trigger_manual, restart_scheduler, CYCLE_STEP_CATALOG
 
 router = APIRouter(prefix="/api/cycle", tags=["cycle"])
 
@@ -13,6 +13,14 @@ def cycle_status():
     return get_status()
 
 
+@router.get("/step-catalog")
+def cycle_step_catalog():
+    """Self-describing catalog of every configurable cycle step (mirrors
+    GET /api/ai-desks/signal-catalog) — lets Config.tsx render a generic
+    form instead of hand-coded sliders per knob."""
+    return {"steps": CYCLE_STEP_CATALOG}
+
+
 @router.get("/history")
 def cycle_history(limit: int = 20):
     """List recent cycle runs."""
@@ -51,6 +59,7 @@ class CycleConfigRequest(BaseModel):
     maturity_threshold_pct: Optional[int] = None
     weekend_cycle_enabled: Optional[bool] = None
     weekend_cycle_times: Optional[str] = None  # "HH:MM,HH:MM" in UTC
+    cycle_step_config: Optional[Dict[str, Dict[str, Any]]] = None  # {step_id: {param: value}}
 
 
 @router.post("/config")
@@ -102,6 +111,18 @@ def update_cycle_config(req: CycleConfigRequest):
         if not re.match(r'^(\d{2}:\d{2})(,\d{2}:\d{2})*$', req.weekend_cycle_times.strip()):
             raise HTTPException(400, "weekend_cycle_times must be 'HH:MM' or 'HH:MM,HH:MM,...'")
         set_config("weekend_cycle_times", req.weekend_cycle_times.strip())
+    if req.cycle_step_config is not None:
+        import json
+        known_ids = {step["id"] for step in CYCLE_STEP_CATALOG}
+        unknown = set(req.cycle_step_config.keys()) - known_ids
+        if unknown:
+            raise HTTPException(400, f"Unknown cycle step id(s): {sorted(unknown)}")
+        # Merge over whatever is already saved so a partial update from the UI
+        # never wipes out other steps' settings.
+        current = json.loads(get_config("cycle_step_config") or "{}")
+        for step_id, params in req.cycle_step_config.items():
+            current[step_id] = {**current.get(step_id, {}), **params}
+        set_config("cycle_step_config", json.dumps(current))
 
     # Restart scheduler to pick up changes
     restart_scheduler()

backend/routers/geopolitical.py

@@ -32,6 +32,14 @@ def geo_news(force_refresh: bool = False):
 
 @router.get("/risk-score")
 def risk_score():
+    """Frozen per-cycle AI-judged score (services.ai_analyzer.ai_score_geo_risk,
+    saved once per auto-cycle run in geo_risk_snapshots) — no longer
+    recomputed live on every request. Falls back to a live algorithmic
+    compute only if no cycle has ever run yet (e.g. fresh install)."""
+    from services.database import get_latest_geo_risk_snapshot
+    snapshot = get_latest_geo_risk_snapshot()
+    if snapshot:
+        return snapshot
     news = _news_cache["data"] or fetch_geo_news()
     return compute_geo_risk_score(news)
 

backend/services/ai_analyzer.py

@@ -1734,6 +1734,66 @@ JSON: {{"items": [{{"i":<int>,"impact_score":<float>,"dir_energy":"...","dir_met
     return news_items
 
 
+# ── Holistic geopolitical risk score (AI judgment, cycle-frozen) ──────────────
+
+def ai_score_geo_risk(news: List[Dict], algo_score: Dict, log_meta: Optional[Dict] = None) -> Dict:
+    """Holistic 0-100 geopolitical risk assessment by the AI. Takes the
+    already AI-scored news list plus compute_geo_risk_score()'s deterministic
+    category-weighted formula as a reference baseline (not a value to just
+    copy) — lets the AI actually judge severity/de-escalation/context instead
+    of a rigid weighted sum, while staying anchored enough not to swing
+    wildly cycle to cycle. Called once per auto-cycle (services/auto_cycle.py
+    Step 1); the result is frozen in DB until the next cycle runs."""
+    if not get_client() or not news:
+        return {
+            "score": algo_score.get("score", 0), "level": algo_score.get("level", "low"),
+            "rationale": "IA indisponible — score algorithmique utilisé tel quel.",
+            "top_risks": [],
+        }
+
+    top_news = sorted(news, key=lambda n: -(n.get("impact_score") or 0))[:15]
+    compact = [
+        {"title": n.get("title", ""), "category": n.get("category", ""), "impact": round(n.get("impact_score") or 0, 2)}
+        for n in top_news
+    ]
+
+    user = f"""Evalue le niveau de risque geopolitique global actuel pour les marches financiers, sur une echelle de 0 a 100.
+
+Score algorithmique de reference (formule ponderee par categorie, a titre indicatif seulement — exerce ton propre jugement, ne le recopie pas mecaniquement) : {algo_score.get('score')}/100 ({algo_score.get('level')}).
+Repartition par categorie : {json.dumps(algo_score.get('breakdown', {}), ensure_ascii=False)}
+
+Top actualites (triees par impact) :
+{json.dumps(compact, ensure_ascii=False)}
+
+Consignes :
+- Un score eleve doit refleter un risque REEL et actuel pour les marches (escalade militaire active, rupture commerciale majeure, crise politique...), pas juste un volume de news.
+- Un evenement de desescalade/resolution doit FAIRE BAISSER le score meme si son impact brut est eleve.
+- Ne t'ancre pas mecaniquement sur le score algorithmique si le contexte reel (titres) justifie un score different.
+- rationale : 2-3 phrases en francais expliquant precisement pourquoi ce score, en citant les evenements les plus determinants.
+- top_risks : 3 a 5 items les plus determinants pour ce score (titres courts).
+
+JSON: {{"score": <0-100 float>, "level": "low"|"medium"|"high"|"extreme", "rationale": "...", "top_risks": ["...", ...]}}"""
+
+    result = _chat(
+        "Tu es un analyste geopolitique senior qui evalue le risque marche global, pas evenement par evenement.",
+        user, model="gpt-4o", json_mode=True, max_tokens=700, log_meta=log_meta,
+    )
+    if not result:
+        return {
+            "score": algo_score.get("score", 0), "level": algo_score.get("level", "low"),
+            "rationale": "Erreur IA — score algorithmique utilisé tel quel.",
+            "top_risks": [],
+        }
+
+    score = max(0.0, min(100.0, float(result.get("score", algo_score.get("score", 0)))))
+    return {
+        "score": round(score, 1),
+        "level": result.get("level") or algo_score.get("level", "low"),
+        "rationale": result.get("rationale", ""),
+        "top_risks": result.get("top_risks", []) or [],
+    }
+
+
 # ── Re-score news batch with AI ───────────────────────────────────────────────
 
 def ai_rescore_news(news_items: List[Dict]) -> List[Dict]:

backend/services/auto_cycle.py

@@ -49,7 +49,7 @@ def _max_similarity_vs_existing(candidate_kws: List[str], existing: List[Dict])
     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
+_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(
@@ -57,6 +57,7 @@ def _is_duplicate_pattern(
     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.
@@ -78,7 +79,7 @@ def _is_duplicate_pattern(
                     candidate_id=candidate.get("id"),
                 )
                 if sim > 0:  # embedding worked
-                    return sim >= _EMBED_SIM_THRESHOLD
+                    return sim >= embed_threshold
             except Exception as _emb_err:
                 logger.debug(f"[Cycle] Embedding failed, fallback Jaccard: {_emb_err}")
 
@@ -87,6 +88,90 @@ def _is_duplicate_pattern(
     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:
@@ -181,7 +266,7 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
         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,
+            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,
@@ -206,6 +291,7 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
         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
@@ -222,7 +308,7 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
             except Exception:
                 pass
 
-        _interval_hours = float(get_config("auto_cycle_interval_hours") or "3")
+        _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:
@@ -329,10 +415,23 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
         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)
+        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
@@ -377,7 +476,7 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
             _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=14)
+            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}")
 
@@ -393,7 +492,7 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
         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=90)
+            _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", [])
@@ -430,7 +529,10 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
         _price_discovery_block = ""
         try:
             from services.price_discovery import compute_absorptions, build_price_discovery_block
-            _absorptions = compute_absorptions(min_age_minutes=30.0, max_age_days=7)
+            _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:
@@ -462,7 +564,7 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
                 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=60)}
+                        _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:
@@ -588,7 +690,7 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
             _institutional_block = ""
             try:
                 from services.ai_analyzer import build_institutional_block
-                _institutional_block = build_institutional_block(days=7)
+                _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:
@@ -634,7 +736,7 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
         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):
+            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)
@@ -713,7 +815,7 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
             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)
+            _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", [])
@@ -744,7 +846,8 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
             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
+        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,
@@ -941,7 +1044,9 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
         try:
             from services.portfolio_risk import analyze_simulation_portfolio
             _risk = analyze_simulation_portfolio()
-            if _risk.get("alerts"):
+            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:
@@ -957,27 +1062,30 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
             logger.warning(f"[PortfolioMonitor] Failed (non-blocking): {_pme}")
 
         # Auto-add any new underlying tickers to the IV watchlist
-        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}")
+        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")}
@@ -1004,17 +1112,23 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
             logger.warning(f"[Cycle] Bayesian update failed (non-blocking): {_be}")
 
         # ── Step 5.6: Régime clustering (Sprint 4.2) ──────────────────────────
-        try:
-            from services.database import detect_and_save_regime_clusters
-            _cluster_result = detect_and_save_regime_clusters(n_clusters=4, days=180)
-            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')})"
+        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"],
                 )
-        except Exception as _ce:
-            logger.warning(f"[Cycle] Régime clustering failed (non-blocking): {_ce}")
+                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 = []
@@ -1081,7 +1195,7 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
             if _commentary_text:
                 _conn = _get_conn()
                 try:
-                    cutoff = (datetime.utcnow() - timedelta(days=14)).strftime("%Y-%m-%d")
+                    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)",
@@ -1171,8 +1285,8 @@ def _generate_cycle_commentary(
         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)
+        # 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({
@@ -1416,7 +1530,11 @@ def _generate_cycle_report(
     var_summary: Dict = {}
     try:
         from services.var_service import compute_var, save_var_snapshot
-        var_result = compute_var(confidence=0.95, horizon_days=1, lookback_days=252, default_iv=0.20)
+        _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 = {
@@ -1654,7 +1772,7 @@ def _auto_portfolio_snapshot(ai_key: str) -> None:
         )
         from datetime import date as _date
 
-        data = get_mtm_trades_with_traces(days=90, limit_movers=10)
+        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)
 
@@ -1799,22 +1917,28 @@ def _auto_synthesize_knowledge(ai_key: str) -> None:
             list_ai_reports, get_mtm_trades_with_traces,
         )
 
-        # Skip if last synthesis < 6 hours ago
+        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 < 6:
+                if age_h < min_hours:
                     logger.info(
-                        f"[AutoSynth] Super Contexte is {age_h:.1f}h old — skipping re-synthesis (threshold: 6h)"
+                        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=90)
+        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()
 
@@ -2104,6 +2228,7 @@ def get_status() -> Dict[str, Any]:
         "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()),
     }

backend/services/database.py

@@ -341,6 +341,20 @@ def init_db():
         top_patterns_json TEXT NOT NULL DEFAULT '[]',
         news_count INTEGER DEFAULT 0
     )""")
+
+    # Geo risk score — one AI-judged snapshot per cycle run, insert-only, never
+    # mutated. Frontend reads only the latest row instead of recomputing live.
+    c.execute("""CREATE TABLE IF NOT EXISTS geo_risk_snapshots (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        run_id TEXT NOT NULL,
+        computed_at TEXT DEFAULT (datetime('now')),
+        score REAL NOT NULL,
+        level TEXT NOT NULL,
+        breakdown_json TEXT DEFAULT '{}',
+        top_risks_json TEXT DEFAULT '[]',
+        ai_rationale TEXT DEFAULT '',
+        source TEXT DEFAULT 'ai'
+    )""")
     try:
         c.execute("CREATE INDEX IF NOT EXISTS idx_gah_ts ON geo_alert_history(timestamp DESC)")
     except Exception:
@@ -1908,6 +1922,33 @@ def get_geo_alert_history(days: int = 30) -> List[Dict[str, Any]]:
     return result
 
 
+def save_geo_risk_snapshot(run_id: str, score: float, level: str, breakdown: Dict[str, Any],
+                            top_risks: List[Any], rationale: str, source: str = "ai") -> None:
+    conn = get_conn()
+    conn.execute(
+        """INSERT INTO geo_risk_snapshots (run_id, score, level, breakdown_json, top_risks_json, ai_rationale, source)
+           VALUES (?, ?, ?, ?, ?, ?, ?)""",
+        (run_id, score, level, json.dumps(breakdown or {}), json.dumps(top_risks or []), rationale or "", source),
+    )
+    conn.commit()
+    conn.close()
+
+
+def get_latest_geo_risk_snapshot() -> Optional[Dict[str, Any]]:
+    conn = get_conn()
+    row = conn.execute(
+        "SELECT * FROM geo_risk_snapshots ORDER BY computed_at DESC LIMIT 1"
+    ).fetchone()
+    conn.close()
+    if not row:
+        return None
+    d = dict(row)
+    d["breakdown"] = json.loads(d.pop("breakdown_json", "{}") or "{}")
+    d["top_risks"] = json.loads(d.pop("top_risks_json", "[]") or "[]")
+    d["rationale"] = d.pop("ai_rationale", "")
+    return d
+
+
 def _normalize_yf_ticker(ticker: str) -> str:
     """Normalize ticker for yfinance.
     - USD/KRW  → USDKRW=X  (slash-format forex pairs from GPT-4o)
@@ -2047,7 +2088,10 @@ def log_trade_entries(run_id: str, scored_patterns: List[Dict[str, Any]], quotes
     import logging as _logging
     _log = _logging.getLogger(__name__)
     profiles = get_risk_profiles(enabled_only=True)
-    _log.info(f"[TradeLog] run_id={run_id} scored_patterns={len(scored_patterns)} profiles={len(profiles)}")
+    min_score_threshold = int(get_config("min_score_threshold") or 0)
+    min_ev_threshold = float(get_config("min_ev_threshold") or 0.0)
+    _log.info(f"[TradeLog] run_id={run_id} scored_patterns={len(scored_patterns)} profiles={len(profiles)} "
+              f"min_score={min_score_threshold} min_ev={min_ev_threshold}")
 
     # Load original patterns as fallback for expected_move_pct
     # (GPT-4o scored output doesn't include this field)
@@ -2174,6 +2218,26 @@ def log_trade_entries(run_id: str, scored_patterns: List[Dict[str, Any]], quotes
 
             ev_gross, ev_net, trade_score = _compute_trade_score(eff_score, exp_move)
 
+            # Global floor — applies on top of the per-profile score/gain match above.
+            if eff_score < min_score_threshold or ev_net < min_ev_threshold:
+                skipped_no_profile += 1
+                _log.debug(
+                    f"[TradeLog] SKIP {underlying} score={eff_score} ev_net={ev_net:.2f} — "
+                    f"below global floor (min_score={min_score_threshold}, min_ev={min_ev_threshold})"
+                )
+                _trade_ac = trade.get("asset_class") or sp.get("asset_class") or _orig.get("asset_class") or ""
+                try:
+                    log_skipped_trade(
+                        run_id=run_id, pattern_id=pid, pattern_name=pattern_name,
+                        underlying=underlying, strategy=strategy, score=eff_score,
+                        expected_move_pct=exp_move,
+                        skip_detail=f"below global floor: score={eff_score}<{min_score_threshold} or ev_net={ev_net:.2f}<{min_ev_threshold}",
+                        asset_class=_trade_ac,
+                    )
+                except Exception:
+                    pass
+                continue
+
             ticker_key = _normalize_ticker(underlying.upper())
             entry_price = price_map.get(ticker_key)
             horizon = int(