""" Technical indicators computed from OHLCV data, calibrated to option horizon. Horizon calibration: <= 30 days : RSI(14), MA(20/50), BB(20), ATR(14) <= 90 days : RSI(21), MA(50/100), BB(50), ATR(21) > 90 days : RSI(28), MA(100/200),BB(100),ATR(28) """ from __future__ import annotations import math from typing import Dict, Optional try: import yfinance as yf import pandas as pd _YF_AVAILABLE = True except ImportError: _YF_AVAILABLE = False def _calibration(horizon_days: int) -> dict: if horizon_days <= 30: return {"rsi": 14, "ma_fast": 20, "ma_slow": 50, "bb": 20, "atr": 14, "label": "Court terme"} elif horizon_days <= 90: return {"rsi": 21, "ma_fast": 50, "ma_slow": 100, "bb": 50, "atr": 21, "label": "Moyen terme"} else: return {"rsi": 28, "ma_fast": 100, "ma_slow": 200, "bb": 100, "atr": 28, "label": "Long terme"} def _rsi(closes: "pd.Series", period: int) -> Optional[float]: delta = closes.diff() gain = delta.clip(lower=0).rolling(period).mean() loss = (-delta.clip(upper=0)).rolling(period).mean() rs = gain / loss.replace(0, float("nan")) rsi_series = 100 - (100 / (1 + rs)) val = rsi_series.iloc[-1] return float(val) if not math.isnan(val) else None def _bollinger(closes: "pd.Series", period: int) -> dict: ma = closes.rolling(period).mean() std = closes.rolling(period).std() upper = ma + 2 * std lower = ma - 2 * std price = closes.iloc[-1] u, l, m = float(upper.iloc[-1]), float(lower.iloc[-1]), float(ma.iloc[-1]) band_width = u - l bb_pct = ((price - l) / band_width * 100) if band_width > 0 else 50.0 return {"upper": round(u, 4), "lower": round(l, 4), "mid": round(m, 4), "bb_pct": round(bb_pct, 1)} def _atr(df: "pd.DataFrame", period: int) -> Optional[float]: high, low, close = df["High"], df["Low"], df["Close"] 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.rolling(period).mean().iloc[-1] return float(atr) if not math.isnan(atr) else None def _trend_signal(price: float, ma_fast: float, ma_slow: float) -> str: if price > ma_fast > ma_slow: return "uptrend" elif price < ma_fast < ma_slow: return "downtrend" elif ma_fast > ma_slow: return "bullish_bias" elif ma_fast < ma_slow: return "bearish_bias" return "sideways" def _rsi_label(rsi_val: float) -> str: if rsi_val >= 70: return "SURACHETÉ ⚠" elif rsi_val <= 30: return "SURVENDU 🔥" return "neutre" def _bb_label(bb_pct: float) -> str: if bb_pct >= 80: return "proche bande haute — potentiel retournement" elif bb_pct <= 20: return "proche bande basse — potentiel rebond" return "dans les bandes" def compute_indicators(ticker: str, horizon_days: int, enabled_indicators: Optional[list] = None) -> dict: """ Compute technical indicators for *ticker* calibrated to *horizon_days*. Returns a dict with computed values + a pre-formatted prompt_block string. Returns {"error": "..."} if data unavailable. """ if not _YF_AVAILABLE: return {"error": "yfinance not installed"} cal = _calibration(horizon_days) # Fetch enough history: need at least ma_slow + some buffer lookback = cal["ma_slow"] * 2 + 50 try: df = yf.download(ticker, period=f"{lookback}d", interval="1d", progress=False, auto_adjust=True) # yfinance ≥0.2 returns MultiIndex columns when group_by is not set — flatten if df is not None and isinstance(df.columns, pd.MultiIndex): df.columns = df.columns.droplevel(1) except Exception as e: return {"error": f"yfinance download failed: {e}"} if df is None or len(df) < cal["ma_slow"]: return {"error": f"Not enough data for {ticker} (got {len(df) if df is not None else 0} rows)"} closes = df["Close"].squeeze().dropna() price = float(closes.iloc[-1]) enabled = set(enabled_indicators) if enabled_indicators else {"rsi", "ma", "bollinger", "atr"} result: dict = { "ticker": ticker, "horizon_days": horizon_days, "calibration": cal["label"], "price": round(price, 4), "periods": cal, } if "rsi" in enabled: rsi_val = _rsi(closes, cal["rsi"]) result["rsi"] = round(rsi_val, 1) if rsi_val is not None else None result["rsi_label"] = _rsi_label(rsi_val) if rsi_val is not None else "N/A" ma_fast_val = ma_slow_val = None if "ma" in enabled: if len(closes) >= cal["ma_fast"]: ma_fast_val = float(closes.rolling(cal["ma_fast"]).mean().iloc[-1]) result["ma_fast"] = round(ma_fast_val, 4) if len(closes) >= cal["ma_slow"]: ma_slow_val = float(closes.rolling(cal["ma_slow"]).mean().iloc[-1]) result["ma_slow"] = round(ma_slow_val, 4) if ma_fast_val and ma_slow_val: result["trend"] = _trend_signal(price, ma_fast_val, ma_slow_val) if "bollinger" in enabled and len(closes) >= cal["bb"]: bb = _bollinger(closes, cal["bb"]) result["bollinger"] = bb result["bb_label"] = _bb_label(bb["bb_pct"]) if "atr" in enabled and len(df) >= cal["atr"]: atr_val = _atr(df, cal["atr"]) if atr_val is not None: result["atr"] = round(atr_val, 4) result["atr_pct"] = round(atr_val / price * 100, 2) # Build a human-readable prompt block lines = [f"📊 INDICATEURS TECHNIQUES — {ticker} (horizon {horizon_days}j, calibration {cal['label']}) :"] if "rsi" in result: lines.append(f" - RSI({cal['rsi']}): {result['rsi']} → {result['rsi_label']}") if "ma_fast" in result and "ma_slow" in result: above = "au-dessus ▲" if price > result["ma_slow"] else "en-dessous ▼" lines.append(f" - MA{cal['ma_fast']}/{cal['ma_slow']}: prix {above} MA{cal['ma_slow']} → trend {result.get('trend','N/A')}") if "bollinger" in result: bb = result["bollinger"] lines.append(f" - Bollinger({cal['bb']}): prix à {bb['bb_pct']}% des bandes → {result['bb_label']}") if "atr" in result: lines.append(f" - ATR({cal['atr']}): {result['atr']} ({result['atr_pct']}% du prix)") result["prompt_block"] = "\n".join(lines) return result def format_indicators_for_prompt(indicators: dict) -> str: """Return the pre-formatted prompt block, or empty string on error.""" if "error" in indicators: return "" return indicators.get("prompt_block", "")