""" Instrument Dashboard Service. Provides per-instrument snapshots with price data, technical indicators, regime detection, trend summary, event filtering, and AI narrative. """ import json import os import re import logging import numpy as np import pandas as pd from pathlib import Path from typing import Dict, Any, List, Optional, Tuple from datetime import datetime, date def _base_ticker(t: str) -> str: """Normalize Yahoo Finance tickers for comparison: EURUSD=X → EURUSD, BZ=F → BZ, ^GSPC → GSPC.""" return re.sub(r'(=X|=F|=RR|-USD|\^)$', '', t.strip().upper()) logger = logging.getLogger(__name__) # ── Config loading ───────────────────────────────────────────────────────────── CONFIG_PATH = Path(__file__).parent.parent / "config" / "instruments.json" _configs: Optional[Dict[str, Any]] = None # In-memory narrative cache: key = (instrument_id, iso_date) → str _narrative_cache: Dict[Tuple[str, str], str] = {} def _load_configs() -> None: global _configs with open(CONFIG_PATH, "r", encoding="utf-8") as f: data = json.load(f) _configs = {inst["id"]: inst for inst in data["instruments"]} logger.info(f"[instrument_service] Loaded {len(_configs)} instrument configs") def get_all_instruments() -> List[Dict]: if _configs is None: _load_configs() return list(_configs.values()) def get_instrument(instrument_id: str) -> Optional[Dict]: if _configs is None: _load_configs() return _configs.get(instrument_id.upper()) def update_instrument_drivers(instrument_id: str, drivers: List[Dict]) -> None: """Persist updated drivers to instruments.json and refresh in-memory config.""" global _configs if _configs is None: _load_configs() uid = instrument_id.upper() if uid not in _configs: raise ValueError(f"Instrument {uid} not found") # Load raw JSON, update the matching instrument, save back with open(CONFIG_PATH, "r", encoding="utf-8") as f: raw = json.load(f) for inst in raw["instruments"]: if inst["id"] == uid: inst["drivers"] = drivers break with open(CONFIG_PATH, "w", encoding="utf-8") as f: json.dump(raw, f, ensure_ascii=False, indent=2) # Refresh in-memory cache _configs[uid]["drivers"] = drivers logger.info(f"[instrument_service] Updated drivers for {uid} ({len(drivers)} drivers)") # ── DataFrame helpers ────────────────────────────────────────────────────────── def _ohlcv_to_df(records: List[Dict]) -> pd.DataFrame: """Convert list of OHLCV dicts (with 'date' key) to a DataFrame indexed by date.""" if not records: return pd.DataFrame() df = pd.DataFrame(records) # Normalise date column — may come as ISO string with or without time component df["date"] = pd.to_datetime(df["date"], utc=True, errors="coerce") df = df.dropna(subset=["date"]) df = df.sort_values("date") df = df.set_index("date") for col in ("open", "high", "low", "close"): if col in df.columns: df[col] = pd.to_numeric(df[col], errors="coerce") if "volume" in df.columns: df["volume"] = pd.to_numeric(df["volume"], errors="coerce").fillna(0) return df def _safe_float(val) -> Optional[float]: """Return a Python float or None for NaN/inf values.""" try: f = float(val) return f if np.isfinite(f) else None except (TypeError, ValueError): return None def _round(val, decimals: int = 4) -> Optional[float]: f = _safe_float(val) return round(f, decimals) if f is not None else None # ── Technical indicators ─────────────────────────────────────────────────────── def _compute_indicators(df: pd.DataFrame, config: Dict) -> Dict[str, Any]: """ Compute moving averages, Bollinger Bands, RSI14, ATR14, volume MA20. Returns a dict with time-series lists and scalar latest values. """ chart_cfg = config.get("chart", {}) ma_periods = chart_cfg.get("ma_periods", [20, 50, 200]) bb_period = chart_cfg.get("bollinger_period", 20) bb_std = chart_cfg.get("bollinger_std", 2) result: Dict[str, Any] = {} if df.empty or "close" not in df.columns: return result close = df["close"] high = df["high"] if "high" in df.columns else close low = df["low"] if "low" in df.columns else close volume = df["volume"] if "volume" in df.columns else pd.Series(dtype=float) # Moving averages — time-series format for charting for period in ma_periods: if len(close) >= period: ma = close.rolling(period).mean() valid = ma.dropna() result[f"ma{period}"] = [ {"time": idx.strftime("%Y-%m-%d"), "value": _round(val)} for idx, val in valid.items() ] else: result[f"ma{period}"] = [] # Bollinger Bands if len(close) >= bb_period: bb_ma = close.rolling(bb_period).mean() bb_sigma = close.rolling(bb_period).std() bb_upper = bb_ma + bb_std * bb_sigma bb_lower = bb_ma - bb_std * bb_sigma valid_idx = bb_ma.dropna().index result["bb_upper"] = [ {"time": idx.strftime("%Y-%m-%d"), "value": _round(bb_upper[idx])} for idx in valid_idx ] result["bb_lower"] = [ {"time": idx.strftime("%Y-%m-%d"), "value": _round(bb_lower[idx])} for idx in valid_idx ] result["bb_mid"] = [ {"time": idx.strftime("%Y-%m-%d"), "value": _round(bb_ma[idx])} for idx in valid_idx ] else: result["bb_upper"] = [] result["bb_lower"] = [] result["bb_mid"] = [] # Realized volatility (annualized %, rolling 20d stddev of log returns) — a "how # turbulent has price action actually been" overlay, distinct from the market-implied # vol computed elsewhere (vol_surface.py) for the options Strategy Builder. vol_window = chart_cfg.get("volatility_window", 20) if len(close) >= vol_window + 1: log_ret = np.log(close / close.shift(1)) realized_vol = log_ret.rolling(vol_window).std() * np.sqrt(252) * 100 valid_vol = realized_vol.dropna() result["volatility"] = [ {"time": idx.strftime("%Y-%m-%d"), "value": _round(val, 2)} for idx, val in valid_vol.items() ] else: result["volatility"] = [] # RSI 14 if len(close) >= 15: delta = close.diff() gain = delta.clip(lower=0) loss = (-delta).clip(lower=0) avg_gain = gain.ewm(com=13, adjust=False).mean() avg_loss = loss.ewm(com=13, adjust=False).mean() rs = avg_gain / avg_loss.replace(0, np.nan) rsi = 100 - (100 / (1 + rs)) valid_rsi = rsi.dropna() result["rsi14"] = [ {"time": idx.strftime("%Y-%m-%d"), "value": _round(val, 2)} for idx, val in valid_rsi.items() ] else: result["rsi14"] = [] # ATR 14 if len(close) >= 15: prev_close = close.shift(1) tr = pd.concat([ high - low, (high - prev_close).abs(), (low - prev_close).abs(), ], axis=1).max(axis=1) atr = tr.ewm(span=14, adjust=False).mean() valid_atr = atr.dropna() result["atr14"] = [ {"time": idx.strftime("%Y-%m-%d"), "value": _round(val)} for idx, val in valid_atr.items() ] else: result["atr14"] = [] # Volume MA20 if len(volume) >= 20 and volume.sum() > 0: vol_ma = volume.rolling(20).mean() valid_vma = vol_ma.dropna() result["volume_ma20"] = [ {"time": idx.strftime("%Y-%m-%d"), "value": int(val) if np.isfinite(val) else None} for idx, val in valid_vma.items() ] else: result["volume_ma20"] = [] return result # ── Regime detection ─────────────────────────────────────────────────────────── def _detect_regime(df: pd.DataFrame, config: Dict) -> Dict[str, Any]: """ Score-based regime detection mapped to config.regime_labels. Labels index: 0=bullish, 1=bearish, 2=transition, 3=volatile, 4=late cycle / consolidation. """ regime_labels = config.get("regime_labels", ["Bull", "Bear", "Transition", "Volatile", "Consolidation"]) default = { "current": regime_labels[0] if regime_labels else "Unknown", "confidence": 0.0, "scores": {label: 0.0 for label in regime_labels}, "signals": {}, } if df.empty or len(df) < 20 or "close" not in df.columns: return default close = df["close"] high = df["high"] if "high" in df.columns else close low = df["low"] if "low" in df.columns else close # Moving averages ma50 = close.rolling(50).mean() if len(close) >= 50 else pd.Series(dtype=float) ma200 = close.rolling(200).mean() if len(close) >= 200 else pd.Series(dtype=float) current_price = _safe_float(close.iloc[-1]) current_ma50 = _safe_float(ma50.iloc[-1]) if not ma50.empty else None current_ma200 = _safe_float(ma200.iloc[-1]) if not ma200.empty else None # MA slopes (% over n bars) def _slope_pct(series: pd.Series, lookback: int) -> Optional[float]: s = series.dropna() if len(s) < lookback + 1: return None v_now = _safe_float(s.iloc[-1]) v_old = _safe_float(s.iloc[-lookback - 1]) if v_now is None or v_old is None or v_old == 0: return None return (v_now - v_old) / abs(v_old) * 100 ma50_slope = _slope_pct(ma50, 5) if not ma50.empty else None ma200_slope = _slope_pct(ma200, 20) if not ma200.empty else None # Momentum 20d momentum_20d = None if len(close) >= 21: p0 = _safe_float(close.iloc[-21]) p1 = _safe_float(close.iloc[-1]) if p0 and p0 != 0: momentum_20d = (p1 - p0) / abs(p0) * 100 # Distance from MA200 dist_ma200 = None if current_price and current_ma200 and current_ma200 != 0: dist_ma200 = (current_price - current_ma200) / abs(current_ma200) * 100 # ATR vs price (volatility ratio) atr_pct = None if len(close) >= 15: prev_close = close.shift(1) tr = pd.concat([ high - low, (high - prev_close).abs(), (low - prev_close).abs(), ], axis=1).max(axis=1) atr14 = tr.ewm(span=14, adjust=False).mean().iloc[-1] if current_price and current_price != 0: atr_pct = _safe_float(atr14) / current_price * 100 if _safe_float(atr14) else None # MA50 above MA200 flag ma50_above_ma200 = None if current_ma50 is not None and current_ma200 is not None: ma50_above_ma200 = current_ma50 > current_ma200 signals = { "ma50_above_ma200": ma50_above_ma200, "ma50_slope_pct": _round(ma50_slope, 3), "ma200_slope_pct": _round(ma200_slope, 3), "momentum_20d_pct": _round(momentum_20d, 3), "dist_ma200_pct": _round(dist_ma200, 3), "vol_ratio_pct": _round(atr_pct, 3), } # ── Score computation ────────────────────────────────────────────────────── # bull_score aggregates trend-following signals bull_score = 0.0 if ma50_above_ma200 is True: bull_score += 0.4 elif ma50_above_ma200 is False: bull_score -= 0.4 if ma50_slope is not None: bull_score += 0.2 if ma50_slope > 0 else -0.2 if momentum_20d is not None: bull_score += 0.2 if momentum_20d > 0 else -0.2 if ma200_slope is not None: bull_score += 0.1 if ma200_slope > 0 else -0.1 # Extra penalty/boost for distance extremes if dist_ma200 is not None: if dist_ma200 > 15: bull_score += 0.1 # strong uptrend extension elif dist_ma200 < -15: bull_score -= 0.1 # Volatile flag: ATR/price > 3% is_volatile = (atr_pct is not None and atr_pct > 3.0) # Transition flag: weak slope + weak momentum is_transition = ( (ma50_slope is not None and abs(ma50_slope) < 0.1) and (momentum_20d is not None and abs(momentum_20d) < 1.0) ) # Late-bull flag: strongly extended above MA200 — risk of reversal is_late_bull = ( bull_score > 0.6 and dist_ma200 is not None and dist_ma200 > 20 ) # Correction-in-bull flag: price above MA200 but momentum turning is_correction_in_bull = ( ma50_above_ma200 is True and momentum_20d is not None and momentum_20d < -3.0 ) # ── Map to 5 regime slots ────────────────────────────────────────────────── # Slot 0 = bullish, 1 = bearish, 2 = transition, 3 = volatile, 4 = late/consolidation raw_scores = [0.0] * 5 if is_correction_in_bull: # Price still above MA200 but momentum rolling over — partial weight to transition raw_scores[0] = max(0.0, bull_score * 0.5) raw_scores[2] = 0.6 elif is_late_bull: # Extended above MA200: some probability we're in late / consolidation regime raw_scores[0] = max(0.0, bull_score * 0.55) raw_scores[4] = bull_score * 0.5 else: # Bullish raw_scores[0] = max(0.0, bull_score) # Bearish raw_scores[1] = max(0.0, -bull_score) # Transition (overrides if flagged) if not is_correction_in_bull: raw_scores[2] = 0.6 if is_transition else 0.0 # Volatile raw_scores[3] = 0.7 if is_volatile else 0.0 # Late cycle / consolidation — moderate bull_score but high dist_ma200 if not is_late_bull and not is_correction_in_bull: if 0.0 < bull_score < 0.3 and dist_ma200 is not None and dist_ma200 > 5: raw_scores[4] = 0.5 else: raw_scores[4] = max(0.0, 0.3 - abs(bull_score)) if not is_transition else 0.0 # If volatile, suppress the others a bit if is_volatile: raw_scores[0] *= 0.5 raw_scores[1] *= 0.5 raw_scores[2] *= 0.5 # Normalise to sum=1 total = sum(raw_scores) if total > 0: norm_scores = [s / total for s in raw_scores] else: norm_scores = [1.0 / 5] * 5 # Pad / truncate to match the number of provided labels n_labels = len(regime_labels) while len(norm_scores) < n_labels: norm_scores.append(0.0) norm_scores = norm_scores[:n_labels] best_idx = int(np.argmax(norm_scores)) # Cap confidence: max 85% for technical regime (always some model uncertainty) confidence = _round(min(norm_scores[best_idx], 0.85), 3) or 0.0 scores_dict = {} for i, label in enumerate(regime_labels): scores_dict[label] = _round(norm_scores[i], 3) or 0.0 return { "current": regime_labels[best_idx], "confidence": confidence, "scores": scores_dict, "signals": signals, } # ── Trend summary ────────────────────────────────────────────────────────────── def _get_trend_summary(df: pd.DataFrame) -> Dict[str, Any]: """ Return key trend metrics as a flat dict of scalars. """ result: Dict[str, Any] = {} if df.empty or "close" not in df.columns: return result close = df["close"] high = df["high"] if "high" in df.columns else close low = df["low"] if "low" in df.columns else close # Current price result["current_price"] = _round(close.iloc[-1]) # 52-week high / low n_252 = min(252, len(close)) result["high_52w"] = _round(close.tail(n_252).max()) result["low_52w"] = _round(close.tail(n_252).min()) # MA slopes def _slope_pct(series: pd.Series, lookback: int) -> Optional[float]: s = series.dropna() if len(s) < lookback + 1: return None v_now = _safe_float(s.iloc[-1]) v_old = _safe_float(s.iloc[-lookback - 1]) if v_now is None or v_old is None or v_old == 0: return None return round((v_now - v_old) / abs(v_old) * 100, 4) ma50 = close.rolling(50).mean() if len(close) >= 50 else pd.Series(dtype=float) ma200 = close.rolling(200).mean() if len(close) >= 200 else pd.Series(dtype=float) result["ma50_slope_5d"] = _slope_pct(ma50, 5) result["ma200_slope_20d"] = _slope_pct(ma200, 20) # RSI 14 current if len(close) >= 15: delta = close.diff() gain = delta.clip(lower=0) loss = (-delta).clip(lower=0) avg_gain = gain.ewm(com=13, adjust=False).mean() avg_loss = loss.ewm(com=13, adjust=False).mean() rs = avg_gain / avg_loss.replace(0, np.nan) rsi = 100 - (100 / (1 + rs)) result["rsi14_current"] = _round(rsi.iloc[-1], 2) else: result["rsi14_current"] = None # ATR 14 current and vs 3-month average if len(close) >= 15: prev_close = close.shift(1) tr = pd.concat([ high - low, (high - prev_close).abs(), (low - prev_close).abs(), ], axis=1).max(axis=1) atr_series = tr.ewm(span=14, adjust=False).mean() atr_current = _safe_float(atr_series.iloc[-1]) result["atr14_current"] = _round(atr_current) # ATR vs 63-day average if len(atr_series) >= 63: atr_3m_avg = _safe_float(atr_series.tail(63).mean()) if atr_3m_avg and atr_3m_avg != 0: result["atr_vs_3m_avg_pct"] = _round((atr_current - atr_3m_avg) / atr_3m_avg * 100, 2) else: result["atr_vs_3m_avg_pct"] = None else: result["atr_vs_3m_avg_pct"] = None else: result["atr14_current"] = None result["atr_vs_3m_avg_pct"] = None # Momentum def _momentum(lookback: int) -> Optional[float]: if len(close) <= lookback: return None p0 = _safe_float(close.iloc[-lookback - 1]) p1 = _safe_float(close.iloc[-1]) if p0 and p0 != 0: return _round((p1 - p0) / abs(p0) * 100, 3) return None result["momentum_1m_pct"] = _momentum(21) result["momentum_3m_pct"] = _momentum(63) # Distance from MAs current_price = _safe_float(close.iloc[-1]) if not ma50.empty and current_price: ma50_val = _safe_float(ma50.iloc[-1]) if ma50_val and ma50_val != 0: result["dist_ma50_pct"] = _round((current_price - ma50_val) / abs(ma50_val) * 100, 3) else: result["dist_ma50_pct"] = None else: result["dist_ma50_pct"] = None if not ma200.empty and current_price: ma200_val = _safe_float(ma200.iloc[-1]) if ma200_val and ma200_val != 0: result["dist_ma200_pct"] = _round((current_price - ma200_val) / abs(ma200_val) * 100, 3) else: result["dist_ma200_pct"] = None else: result["dist_ma200_pct"] = None return result # ── Event filtering ──────────────────────────────────────────────────────────── def _get_relevant_events( config: Dict, from_date: Optional[str] = None, to_date: Optional[str] = None, instrument_id: Optional[str] = None, ) -> List[Dict]: """ Filter market_events DB rows relevant to the instrument, by date range and keyword/asset match. Also includes events that have a causal analysis for this instrument (regardless of keywords). Returns max 30 events sorted by start_date asc. """ try: from services.database import get_conn from services.causal_graphs import init_tables conn = get_conn() init_tables(conn) # ensure causal_event_analyses exists rows = conn.execute(""" SELECT e.*, (SELECT GROUP_CONCAT(DISTINCT a.instrument) FROM causal_event_analyses a WHERE a.market_event_id = e.id) AS analyzed_instruments, (SELECT a.template_id FROM causal_event_analyses a WHERE a.market_event_id = e.id ORDER BY a.id DESC LIMIT 1) AS cea_template_id, (SELECT a.id FROM causal_event_analyses a WHERE a.market_event_id = e.id ORDER BY a.id DESC LIMIT 1) AS analysis_id, (SELECT a.prediction_json FROM causal_event_analyses a WHERE a.market_event_id = e.id ORDER BY a.id DESC LIMIT 1) AS cea_prediction_json, (SELECT a.actual_json FROM causal_event_analyses a WHERE a.market_event_id = e.id ORDER BY a.id DESC LIMIT 1) AS cea_actual_json, (SELECT a.activation_score FROM causal_event_analyses a WHERE a.market_event_id = e.id ORDER BY a.id DESC LIMIT 1) AS cea_activation_score FROM market_events e ORDER BY e.start_date DESC """).fetchall() conn.close() # Use cea_template_id (from causal_event_analyses) explicitly to avoid # any collision with an e.* column named template_id all_events = [ {**dict(r), "template_id": r["cea_template_id"], "prediction_json": r["cea_prediction_json"], "actual_json": r["cea_actual_json"], "activation_score": r["cea_activation_score"]} for r in rows ] except Exception as e: logger.warning(f"[instrument_service] Could not load market events: {e}") return [] keywords = [kw.lower() for kw in config.get("event_keywords", [])] related = [ra.lower() for ra in config.get("related_assets", [])] inst_upper = (instrument_id or "").upper() filtered = [] for ev in all_events: ev_start = str(ev.get("start_date", "") or "") ev_end = str(ev.get("end_date", "") or "") # Date range overlap filter: # Include if event overlaps with [from_date, to_date] # An event overlaps if: ev_start <= to_date AND (ev_end >= from_date OR ev_end is empty) if to_date and ev_start and ev_start > to_date: continue if from_date and ev_end and ev_end < from_date: continue # If ev_end is empty (point event), include if start is within [-6 months, to_date] if from_date and not ev_end: # Allow events whose start is up to 6 months before the chart window import datetime try: start_dt = datetime.date.fromisoformat(ev_start) from_dt = datetime.date.fromisoformat(from_date) if (from_dt - start_dt).days > 180: continue except Exception: pass # Keyword / asset relevance ev_name = (ev.get("name") or ev.get("event_name") or "").lower() ev_desc = (ev.get("description") or "").lower() ev_assets = (ev.get("affected_assets") or "").lower() keyword_hit = any(kw in ev_name or kw in ev_desc for kw in keywords) asset_hit = any(ra in ev_assets for ra in related) # Always include events that have a causal analysis for this instrument # Normalize tickers: strip Yahoo Finance suffixes (=X, =F, ^, -USD, =RR) for comparison analyzed = ev.get("analyzed_instruments") or "" analyzed_bases = [_base_ticker(s) for s in analyzed.split(",") if s.strip()] analysis_hit = bool(inst_upper and _base_ticker(inst_upper) in analyzed_bases) if keyword_hit or asset_hit or analysis_hit: filtered.append({ "id": ev.get("id"), "template_id": ev.get("template_id"), "analyzed_instruments": ev.get("analyzed_instruments"), # comma-sep, e.g. "EURUSD,SP500" "date": ev_start, "end_date": ev.get("end_date") or None, "title": ev.get("name") or ev.get("event_name") or "", "level": ev.get("level", "medium"), "category": ev.get("category", ""), "sub_type": ev.get("sub_type") or None, "description": (ev.get("description") or "")[:120], "impact_score": float(ev.get("impact_score") or 0.5), "expected_value": ev.get("expected_value") or None, "actual_value": ev.get("actual_value") or None, "surprise_pct": ev.get("surprise_pct"), "unit": ev.get("unit") or None, "absorption_pct": ev.get("absorption_pct"), "prediction_json": ev.get("prediction_json") or None, "actual_json": ev.get("actual_json") or None, "activation_score": ev.get("activation_score"), }) # Sort by start_date asc for timeline display, cap at 30 filtered.sort(key=lambda e: str(e.get("date", "") or "")) return filtered[:30] # ── Main snapshot ────────────────────────────────────────────────────────────── async def get_snapshot( instrument_id: str, period: str = "1y", interval: str = "1d", ) -> Dict[str, Any]: """ Build the full instrument snapshot: price_data + indicators + regime + trend + events + current price/change. """ config = get_instrument(instrument_id) if not config: return {"error": f"Unknown instrument: {instrument_id}"} yf_ticker = config.get("yf_ticker", instrument_id) # Fetch OHLCV try: from services.data_fetcher import get_historical records = get_historical(yf_ticker, period=period, interval=interval) except Exception as e: logger.error(f"[instrument_service] Data fetch failed for {instrument_id}: {e}") records = [] df = _ohlcv_to_df(records) # Compute everything indicators = _compute_indicators(df, config) if not df.empty else {} regime = _detect_regime(df, config) if not df.empty else {} trend = _get_trend_summary(df) if not df.empty else {} # Current price and 1-day change current_price: Optional[float] = None change_pct: Optional[float] = None change_abs: Optional[float] = None if not df.empty and "close" in df.columns and len(df) >= 2: last = _safe_float(df["close"].iloc[-1]) prev = _safe_float(df["close"].iloc[-2]) current_price = _round(last) if last is not None and prev is not None and prev != 0: change_abs = _round(last - prev) change_pct = _round((last - prev) / abs(prev) * 100, 3) # Price data for chart (time + ohlcv) price_data = [] if records: for r in records: price_data.append({ "time": str(r.get("date", ""))[:10], "open": r.get("open"), "high": r.get("high"), "low": r.get("low"), "close": r.get("close"), "volume": r.get("volume", 0), }) # Events spanning the chart period (built after price_data so dates are known) try: chart_start = price_data[0]["time"] if price_data else None chart_end = price_data[-1]["time"] if price_data else None events = _get_relevant_events(config, from_date=chart_start, to_date=chart_end, instrument_id=instrument_id) except Exception as e: logger.warning(f"[instrument_service] Event filtering error: {e}") events = [] # Macro regime (global cycle: goldilocks / stagflation / recession / etc.) macro_regime: Dict[str, Any] = {} try: from services.data_fetcher import get_macro_gauges, score_macro_scenarios gauges = get_macro_gauges() macro_raw = score_macro_scenarios(gauges) dominant = macro_raw.get("dominant", "incertain") meta = macro_raw.get("meta", {}) ranked = macro_raw.get("ranked", []) macro_regime = { "dominant": dominant, "label": meta.get(dominant, {}).get("label", dominant) if isinstance(meta, dict) else dominant, "color": meta.get(dominant, {}).get("color", "#6b7280") if isinstance(meta, dict) else "#6b7280", "emoji": meta.get(dominant, {}).get("emoji", "🌍") if isinstance(meta, dict) else "🌍", "scores": macro_raw.get("scores", {}), "ranked": ranked[:5], "asset_bias": (macro_raw.get("asset_bias") or {}).get(dominant, {}), } except Exception as _me: logger.warning(f"[instrument_service] Macro regime fetch error: {_me}") macro_regime = {"dominant": "incertain", "label": "Incertain", "scores": {}} return { "instrument": config, "price_data": price_data, "indicators": indicators, "regime": regime, "macro_regime": macro_regime, "trend": trend, "events": events, "current_price": current_price, "change_pct": change_pct, "change_abs": change_abs, "period": period, } # ── AI Narrative ─────────────────────────────────────────────────────────────── async def get_narrative( instrument_id: str, snapshot_data: Optional[Dict] = None, force: bool = False, ) -> str: """ Generate (or return cached) a 3-4 sentence French narrative for the instrument. Uses gpt-4o-mini via the existing OpenAI client pattern. Cache key: (instrument_id, today_iso_date). """ today_str = date.today().isoformat() cache_key = (instrument_id.upper(), today_str) if not force and cache_key in _narrative_cache: return _narrative_cache[cache_key] config = get_instrument(instrument_id) if not config: return f"Instrument {instrument_id} non reconnu." # Fetch snapshot if not provided if snapshot_data is None: snapshot_data = await get_snapshot(instrument_id) trend = snapshot_data.get("trend", {}) regime = snapshot_data.get("regime", {}) drivers = config.get("drivers", []) ai_ctx = config.get("ai_context", "") # Build concise prompt data drivers_str = ", ".join( f"{d['label']} (poids {d['weight']})" for d in drivers[:5] ) regime_str = regime.get("current", "N/A") regime_conf = regime.get("confidence", 0) rsi = trend.get("rsi14_current") dist50 = trend.get("dist_ma50_pct") dist200 = trend.get("dist_ma200_pct") mom1m = trend.get("momentum_1m_pct") mom3m = trend.get("momentum_3m_pct") current_price = trend.get("current_price") or snapshot_data.get("current_price") change_pct = snapshot_data.get("change_pct") atr = trend.get("atr14_current") atr_vs_avg = trend.get("atr_vs_3m_avg_pct") system_prompt = ( "Tu es un analyste financier quantitatif senior spécialisé en options et macro. " "Tu génères des narratives d'analyse courtes, précises et actionnables en français." ) user_prompt = f"""Analyse l'instrument {config['name']} ({instrument_id}) et génère une narrative de 3-4 phrases en français. ## Contexte instrument {ai_ctx} ## Données techniques actuelles - Prix actuel : {current_price} | Variation 1j : {change_pct}% - Momentum 1 mois : {mom1m}% | Momentum 3 mois : {mom3m}% - Distance MA50 : {dist50}% | Distance MA200 : {dist200}% - RSI14 : {rsi} - ATR14 : {atr} | ATR vs moy. 3 mois : {atr_vs_avg}% ## Régime détecté - Régime : {regime_str} (confiance : {regime_conf:.0%}) ## Drivers principaux {drivers_str} ## Format requis (4 phrases en français, sans bullet points) : 1. Situation technique actuelle (tendance, niveaux clés, momentum) 2. Contexte macro dominant et driver principal 3. Niveaux clés à surveiller (support, résistance, MA critique) 4. Implication pour le trading d'options (vol implicite, stratégie suggérée)""" # Call OpenAI try: import os as _os from openai import OpenAI api_key = _os.environ.get("OPENAI_API_KEY", "") if not api_key: from services.database import get_config api_key = get_config("openai_api_key") or "" if not api_key: return "Clé API OpenAI non configurée — narrative indisponible." client = OpenAI(api_key=api_key) resp = client.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, ], temperature=0.3, max_tokens=400, ) narrative = resp.choices[0].message.content.strip() _narrative_cache[cache_key] = narrative return narrative except Exception as e: logger.error(f"[instrument_service] Narrative generation failed for {instrument_id}: {e}") return f"Narrative indisponible ({type(e).__name__})."