feat: Phase 4+5 — price discovery status + replay historique

Phase 4 — Price Discovery Status (la pièce maîtresse) :
- price_discovery.py (nouveau) : capture_price_snapshots() sauve les prix des tickers
  liés à chaque news scorée (energy→BZ=F/NG=F, metals→GC=F/HG=F, indices→^GSPC/IWM)
- compute_absorptions() mesure combien du mouvement attendu s'est déjà produit
  (status: not_yet_priced <30% / partially_priced 30-80% / fully_priced >80%)
- build_price_discovery_block() → bloc prompt avec opportunités classées
- database.py : table news_price_snapshots + save/get/purge fonctions
- auto_cycle.py : capture après ai_score_news_batch, compute avant suggestions,
  block injecté dans suggestion + scoring prompts + context snapshot
- ai_analyzer.py : param price_discovery_block dans suggest + score

Phase 5 — Replay historique :
- cycle.py : POST /api/cycle/contexts/{run_id}/replay — recharge le snapshot historique
  et relance suggest_patterns_from_market_context avec le contexte original
- useApi.ts : hook useReplayCycle
- SystemLogs.tsx : bouton "Rejouer ce cycle" dans onglet Contexte IA avec champ
  notes, résultats inline (liste des patterns générés), section price_discovery
  ouverte par défaut en rouge

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

backend/routers/cycle.py

@@ -98,3 +98,96 @@ def get_context_snapshot(run_id: str):
     if not snap:
         raise HTTPException(404, "Snapshot non trouvé pour ce cycle")
     return snap
+
+
+class ReplayRequest(BaseModel):
+    override_notes: Optional[str] = None  # optional annotation added to the replay
+
+
+@router.post("/contexts/{run_id}/replay")
+def replay_cycle(run_id: str, req: ReplayRequest):
+    """
+    Phase 5 — Re-run AI suggestion with the historical context from a saved snapshot.
+    Returns new pattern suggestions based on the original context data.
+    """
+    snap = get_cycle_context_snapshot(run_id)
+    if not snap:
+        raise HTTPException(404, "Snapshot non trouvé pour ce cycle")
+
+    ctx = snap.get("context", {})
+    if not ctx:
+        raise HTTPException(400, "Snapshot vide — impossible de rejouer")
+
+    try:
+        from services.ai_analyzer import suggest_patterns_from_market_context, apply_news_decay, partition_news_by_age
+        import json
+
+        # Reconstruct minimal inputs from the snapshot
+        # news: merge inter_cycle + recent_24h + older from partitioned snapshot
+        news_part = ctx.get("news_partitioned", {})
+        news_flat = (
+            news_part.get("inter_cycle", [])
+            + news_part.get("recent_24h", [])
+            + news_part.get("older", [])
+        )
+
+        # quotes: reconstruct from quotes_summary
+        quotes_by_class = ctx.get("quotes_summary", {})
+
+        # calendar
+        calendar = ctx.get("calendar", [])
+
+        # macro regime (simplified for replay)
+        macro_regime = None
+        if ctx.get("macro_regime"):
+            macro_regime = {"scenarios": {"dominant": ctx["macro_regime"].get("dominant"), "scores": ctx["macro_regime"].get("scores"), "asset_bias": {}, "reasons": []}, "gauges": {}}
+
+        # geo score
+        geo_score = ctx.get("geo_score", {"score": 50, "level": "medium", "top_risks": []})
+
+        # preserved blocks
+        tech_block = ctx.get("tech_indicators_block", "")
+        iv_context = ctx.get("iv_context_preview", "")
+        cycle_meta = ctx.get("cycle_meta", {})
+
+        # Rebuild FRED block from saved releases
+        fred_block = ""
+        if ctx.get("fred_releases"):
+            from services.fred_fetcher import build_fred_context_block
+            fred_block = build_fred_context_block(ctx["fred_releases"])
+
+        # Rebuild price discovery block from saved absorptions
+        pd_block = ""
+        if ctx.get("price_discovery"):
+            from services.price_discovery import build_price_discovery_block
+            pd_block = build_price_discovery_block(ctx["price_discovery"])
+
+        # Add replay note to cycle_meta
+        if req.override_notes:
+            cycle_meta = {**cycle_meta, "replay_notes": req.override_notes}
+        cycle_meta["is_replay"] = True
+        cycle_meta["replayed_at"] = __import__("datetime").datetime.utcnow().isoformat()
+
+        suggestions = suggest_patterns_from_market_context(
+            news=news_flat,
+            quotes_by_class=quotes_by_class,
+            calendar=calendar,
+            macro_regime=macro_regime,
+            geo_score=geo_score,
+            iv_context=iv_context,
+            cycle_meta=cycle_meta,
+            tech_indicators_block=tech_block,
+            fred_block=fred_block,
+            price_discovery_block=pd_block,
+        )
+
+        return {
+            "run_id": run_id,
+            "original_ts": snap["ts"],
+            "replayed_at": cycle_meta["replayed_at"],
+            "override_notes": req.override_notes,
+            "suggestions_count": len(suggestions),
+            "suggestions": suggestions,
+        }
+    except Exception as e:
+        raise HTTPException(500, f"Replay failed: {str(e)}")

backend/services/ai_analyzer.py

@@ -355,6 +355,7 @@ def score_patterns_with_context(
     cycle_meta: Optional[Dict] = None,
     tech_indicators_block: str = "",
     fred_block: str = "",
+    price_discovery_block: str = "",
 ) -> List[Dict[str, Any]]:
     """Score all patterns with rich context (news, prices, IV, risk clusters) using GPT-4o."""
     if not get_client():
@@ -575,6 +576,7 @@ Instructions de notation:
 
     _tech_sc_section = f"\n{tech_indicators_block}\n" if tech_indicators_block else ""
     _fred_sc_section = f"\n{fred_block}\n" if fred_block else ""
+    _pd_sc_section = f"\n{price_discovery_block}\n" if price_discovery_block else ""
 
     user = f"""CONTEXTE GLOBAL:
 - Score risque géopolitique: {geo_score.get('score', 50)}/100 ({geo_score.get('level', 'medium')})
@@ -582,6 +584,7 @@ Instructions de notation:
 {temporal_section_sc}
 {macro_section}
 {_fred_sc_section}
+{_pd_sc_section}
 {_tech_sc_section}
 TEMPLATE DE NOTATION:
 {scoring_template}
@@ -973,6 +976,7 @@ def suggest_patterns_from_market_context(
     cycle_meta: Optional[Dict] = None,
     tech_indicators_block: str = "",
     fred_block: str = "",
+    price_discovery_block: str = "",
 ) -> List[Dict]:
     """Ask GPT-4o to propose new patterns based on current geo/market + macro regime context."""
     _cycle_meta = cycle_meta or {}
@@ -1120,6 +1124,7 @@ Règles supplémentaires:
 
     tech_block_section = f"\n{tech_indicators_block}\n" if tech_indicators_block else ""
     fred_section = f"\n{fred_block}\n" if fred_block else ""
+    pd_section = f"\n{price_discovery_block}\n" if price_discovery_block else ""
 
     user = f"""Tu es un stratège géopolitique et financier senior, expert en options.
 {macro_block}{geo_block}{lessons_block}{reliability_block}{iv_block}
@@ -1127,6 +1132,7 @@ Règles supplémentaires:
 ## Prix des marchés (variation J-1)
 {market_block}
 {fred_section}
+{pd_section}
 {tech_block_section}
 ## Calendrier économique à venir
 {cal_block}

backend/services/auto_cycle.py

@@ -175,6 +175,7 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
             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
@@ -326,6 +327,21 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
 
         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=14)
+        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}
@@ -370,6 +386,19 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
                 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=30.0, max_age_days=7)
+            _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 ──────────────────────────────
         _fred_releases = []
         _fred_block = ""
@@ -436,6 +465,10 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
                         "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 [],
@@ -454,6 +487,7 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
                 cycle_meta=cycle_meta,
                 tech_indicators_block=_tech_block,
                 fred_block=_fred_block,
+                price_discovery_block=_price_discovery_block,
             )
         except Exception as e:
             logger.warning(f"[Cycle] Suggestion step failed: {e}")
@@ -579,6 +613,7 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
                 cycle_meta=cycle_meta,
                 tech_indicators_block=_tech_block,
                 fred_block=_fred_block,
+                price_discovery_block=_price_discovery_block,
             )
             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")]

backend/services/database.py

@@ -385,6 +385,19 @@ def init_db():
         context_json TEXT NOT NULL
     )""")
 
+    c.execute("""CREATE TABLE IF NOT EXISTS news_price_snapshots (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        article_hash TEXT NOT NULL,
+        article_title TEXT,
+        ticker TEXT NOT NULL,
+        expected_direction TEXT,
+        expected_impact_score REAL,
+        price_at_capture REAL,
+        captured_at TEXT NOT NULL DEFAULT (datetime('now')),
+        capture_cycle_id TEXT,
+        UNIQUE(article_hash, ticker)
+    )""")
+
     c.execute("""CREATE TABLE IF NOT EXISTS skipped_trades (
         id INTEGER PRIMARY KEY AUTOINCREMENT,
         run_id TEXT,
@@ -2130,6 +2143,57 @@ def list_cycle_context_snapshots(limit: int = 30) -> list:
     return [{"run_id": r["run_id"], "ts": r["ts"]} for r in rows]
 
 
+# ── News Price Snapshots (Phase 4 — Price Discovery) ─────────────────────────
+
+def save_news_price_snapshot(
+    article_hash: str,
+    article_title: str,
+    ticker: str,
+    expected_direction: str,
+    expected_impact_score: float,
+    price_at_capture: float,
+    cycle_id: str,
+) -> None:
+    conn = get_conn()
+    conn.execute(
+        """INSERT OR IGNORE INTO news_price_snapshots
+           (article_hash, article_title, ticker, expected_direction,
+            expected_impact_score, price_at_capture, capture_cycle_id)
+           VALUES (?, ?, ?, ?, ?, ?, ?)""",
+        (article_hash, article_title, ticker, expected_direction,
+         expected_impact_score, price_at_capture, cycle_id),
+    )
+    conn.commit()
+    conn.close()
+
+
+def get_news_price_snapshots(max_age_days: int = 7, min_age_minutes: float = 30.0) -> list:
+    conn = get_conn()
+    rows = conn.execute(
+        """SELECT article_hash, article_title, ticker, expected_direction,
+                  expected_impact_score, price_at_capture, captured_at, capture_cycle_id
+           FROM news_price_snapshots
+           WHERE captured_at >= datetime('now', ? || ' days')
+             AND captured_at <= datetime('now', ? || ' minutes')
+           ORDER BY captured_at DESC""",
+        (f"-{max_age_days}", f"-{int(min_age_minutes)}"),
+    ).fetchall()
+    conn.close()
+    return [dict(r) for r in rows]
+
+
+def purge_old_price_snapshots(older_than_days: int = 14) -> int:
+    conn = get_conn()
+    conn.execute(
+        "DELETE FROM news_price_snapshots WHERE captured_at < datetime('now', ? || ' days')",
+        (f"-{older_than_days}",),
+    )
+    deleted = conn.total_changes
+    conn.commit()
+    conn.close()
+    return deleted
+
+
 # ── Knowledge Base Decay ──────────────────────────────────────────────────────
 
 def decay_kb_confidence() -> int:

backend/services/price_discovery.py

@@ -0,0 +1,230 @@
+"""
+Phase 4 — Price Discovery Status
+
+For each scored news article, we track the prices of related tickers at capture time.
+On subsequent cycles we measure how much of the expected move has already happened
+("absorption"), and flag opportunities where price has NOT yet moved.
+
+Direction mapping from ai_score_news_batch fields:
+  ai_dir_energy  → BZ=F (Brent), NG=F (Natural Gas)
+  ai_dir_metals  → GC=F (Gold), HG=F (Copper)
+  ai_dir_indices → ^GSPC (S&P 500), IWM (Russell 2000)
+  ai_dir_forex   → DX-Y.NYB (DXY) [if present]
+"""
+from __future__ import annotations
+
+import hashlib
+import logging
+import math
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Optional
+
+logger = logging.getLogger("price_discovery")
+
+# Map AI direction fields → related tickers
+DIR_FIELD_TO_TICKERS: Dict[str, list] = {
+    "ai_dir_energy":  ["BZ=F", "NG=F"],
+    "ai_dir_metals":  ["GC=F", "HG=F"],
+    "ai_dir_indices": ["^GSPC", "IWM"],
+}
+
+# Minimum impact score to bother tracking
+MIN_IMPACT_SCORE = 0.4
+
+# Absorption thresholds
+FULLY_PRICED_THRESHOLD   = 0.80   # >80% absorbed
+PARTIALLY_PRICED         = 0.30   # 30-80%
+# <30% → not_yet_priced
+
+# Expected move % per asset class per unit of impact score
+# e.g. energy with impact 0.8 → expected ~2.0% move
+EXPECTED_MOVE_PCT: Dict[str, float] = {
+    "BZ=F":   2.5,
+    "NG=F":   3.5,
+    "GC=F":   1.2,
+    "HG=F":   2.0,
+    "^GSPC":  1.5,
+    "IWM":    2.0,
+    "DX-Y.NYB": 0.8,
+}
+
+
+def _article_hash(article: Dict) -> str:
+    key = (article.get("title", "") + article.get("source", "")).encode()
+    return hashlib.md5(key).hexdigest()[:16]
+
+
+def capture_price_snapshots(news: List[Dict], quotes_flat: Dict[str, float], cycle_id: str) -> int:
+    """
+    For each scored news article, save a price snapshot for related tickers.
+
+    quotes_flat: {ticker_symbol: price} — pre-flattened from get_all_quotes()
+    Returns number of snapshots saved.
+    """
+    from services.database import save_news_price_snapshot
+    saved = 0
+    for article in news:
+        impact = float(article.get("impact_score") or 0)
+        if impact < MIN_IMPACT_SCORE:
+            continue
+        ah = _article_hash(article)
+        title = article.get("title", "")[:200]
+
+        for dir_field, tickers in DIR_FIELD_TO_TICKERS.items():
+            direction = article.get(dir_field, "neutral")
+            if direction == "neutral":
+                continue
+            for ticker in tickers:
+                price = quotes_flat.get(ticker)
+                if price is None:
+                    continue
+                try:
+                    save_news_price_snapshot(
+                        article_hash=ah,
+                        article_title=title,
+                        ticker=ticker,
+                        expected_direction=direction,
+                        expected_impact_score=round(impact, 3),
+                        price_at_capture=price,
+                        cycle_id=cycle_id,
+                    )
+                    saved += 1
+                except Exception as e:
+                    logger.debug(f"[PD] snapshot save failed {ticker}: {e}")
+    return saved
+
+
+def _fetch_current_prices(tickers: List[str]) -> Dict[str, float]:
+    """Fetch latest prices via yfinance for a list of tickers."""
+    if not tickers:
+        return {}
+    try:
+        import yfinance as yf
+        data = yf.download(tickers, period="2d", interval="1d", progress=False, auto_adjust=True)
+        prices: Dict[str, float] = {}
+        if hasattr(data.columns, "levels"):
+            # MultiIndex: (field, ticker)
+            close = data["Close"] if "Close" in data else data
+            for tkr in tickers:
+                try:
+                    val = float(close[tkr].dropna().iloc[-1])
+                    prices[tkr] = val
+                except Exception:
+                    pass
+        else:
+            # Single ticker
+            try:
+                val = float(data["Close"].dropna().iloc[-1])
+                prices[tickers[0]] = val
+            except Exception:
+                pass
+        return prices
+    except Exception as e:
+        logger.warning(f"[PD] yfinance price fetch failed: {e}")
+        return {}
+
+
+def compute_absorptions(min_age_minutes: float = 30.0, max_age_days: int = 7) -> List[Dict]:
+    """
+    For all snapshots in the correct age window, compute absorption.
+
+    Returns list of dicts with absorption data, sorted by opportunity (lowest absorption first).
+    """
+    from services.database import get_news_price_snapshots
+    snapshots = get_news_price_snapshots(max_age_days=max_age_days, min_age_minutes=min_age_minutes)
+    if not snapshots:
+        return []
+
+    tickers = list({s["ticker"] for s in snapshots})
+    current_prices = _fetch_current_prices(tickers)
+
+    results = []
+    for snap in snapshots:
+        ticker = snap["ticker"]
+        price_now = current_prices.get(ticker)
+        if price_now is None:
+            continue
+
+        price_cap = snap["price_at_capture"]
+        if not price_cap or price_cap == 0:
+            continue
+
+        direction = snap["expected_direction"]  # bullish | bearish
+        impact = snap["expected_impact_score"]
+        expected_move = EXPECTED_MOVE_PCT.get(ticker, 1.5) * impact
+
+        actual_move_pct = (price_now - price_cap) / price_cap * 100
+        # Align actual move with expected direction
+        signed_move = actual_move_pct if direction == "bullish" else -actual_move_pct
+
+        absorption = max(0.0, signed_move / expected_move) if expected_move > 0 else 0.0
+
+        if absorption >= FULLY_PRICED_THRESHOLD:
+            status = "fully_priced"
+        elif absorption >= PARTIALLY_PRICED:
+            status = "partially_priced"
+        else:
+            status = "not_yet_priced"
+
+        results.append({
+            "ticker": ticker,
+            "article_title": snap["article_title"],
+            "article_hash": snap["article_hash"],
+            "expected_direction": direction,
+            "expected_impact_score": impact,
+            "price_at_capture": round(price_cap, 4),
+            "price_now": round(price_now, 4),
+            "actual_move_pct": round(actual_move_pct, 3),
+            "signed_move_pct": round(signed_move, 3),
+            "expected_move_pct": round(expected_move, 3),
+            "absorption_pct": round(min(absorption * 100, 200), 1),
+            "status": status,
+            "opportunity": status == "not_yet_priced",
+            "captured_at": snap["captured_at"],
+        })
+
+    # Sort: opportunities first (lowest absorption), then partially priced
+    results.sort(key=lambda r: r["absorption_pct"])
+    return results
+
+
+def build_price_discovery_block(absorptions: List[Dict]) -> str:
+    """Build the prompt block from computed absorptions."""
+    if not absorptions:
+        return ""
+
+    opportunities = [a for a in absorptions if a["status"] == "not_yet_priced"]
+    partial = [a for a in absorptions if a["status"] == "partially_priced"]
+    priced = [a for a in absorptions if a["status"] == "fully_priced"]
+
+    lines = ["## ⚡ PRICE DISCOVERY STATUS — Signaux déjà dans les prix ou pas ?"]
+
+    if opportunities:
+        lines.append(f"\n🔥 NON ENCORE PRICÉS ({len(opportunities)}) — OPPORTUNITÉS POTENTIELLES :")
+        for a in opportunities[:5]:
+            lines.append(
+                f"  {a['ticker']} | \"{a['article_title'][:55]}...\" "
+                f"→ dir {a['expected_direction'].upper()} | mouvement actuel {a['actual_move_pct']:+.2f}% "
+                f"vs attendu {a['expected_move_pct']:+.2f}% → {a['absorption_pct']:.0f}% absorbé"
+            )
+        lines.append("  ⚠️ Ces tickers ont reçu un signal fort mais le marché n'a pas encore bougé.")
+
+    if partial:
+        lines.append(f"\n⚠️ PARTIELLEMENT PRICÉS ({len(partial)}) — FENÊTRE EN COURS :")
+        for a in partial[:3]:
+            lines.append(
+                f"  {a['ticker']} → {a['absorption_pct']:.0f}% absorbé ({a['actual_move_pct']:+.2f}% / {a['expected_move_pct']:+.2f}% attendu)"
+            )
+
+    if priced:
+        lines.append(f"\n✅ DÉJÀ PRICÉS ({len(priced)}) — éviter de chasser :")
+        for a in priced[:3]:
+            lines.append(
+                f"  {a['ticker']} → {a['absorption_pct']:.0f}% absorbé — marché a déjà intégré"
+            )
+
+    lines.append(
+        "\n⚠️ CONSIGNE : Pour les tickers NON ENCORE PRICÉS, tu peux être plus agressif sur l'expected_move_pct. "
+        "Pour les tickers DÉJÀ PRICÉS, évite les positions dans le sens de la news (momentum tardif)."
+    )
+    return "\n".join(lines)