198 lines
7.4 KiB
Python
198 lines
7.4 KiB
Python
"""
|
|
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", "")
|
|
|
|
|
|
def compute_and_save_indicators(horizon_days: int = 45) -> dict:
|
|
"""Cycle Actions — standalone "compute-indicators" action. compute_indicators()
|
|
itself is pure (no persistence) — this loops the watchlist and persists each
|
|
result into instrument_indicators (which nothing else reads from yet; this is
|
|
an inspection snapshot, not a cache other steps depend on)."""
|
|
from services.database import get_instruments_watchlist, save_instrument_indicators
|
|
|
|
tickers = [w["ticker"] for w in get_instruments_watchlist()]
|
|
computed = 0
|
|
failed = []
|
|
for ticker in tickers:
|
|
result = compute_indicators(ticker, horizon_days=horizon_days)
|
|
if "error" in result:
|
|
failed.append(ticker)
|
|
continue
|
|
save_instrument_indicators(ticker, horizon_days, result)
|
|
computed += 1
|
|
return {"tickers_computed": computed, "failed": failed, "horizon_days": horizon_days}
|