feat: Phase 3 — indicateurs techniques calibrés par horizon option
- technical_indicators.py (nouveau) : compute_indicators() calcule RSI, MA fast/slow, Bollinger Bands, ATR — périodes calibrées automatiquement selon horizon_days - config.py : endpoints GET/PUT /config/tech-indicators (activé, liste, auto-calibration) - useApi.ts : useTechIndicatorsConfig + useSaveTechIndicatorsConfig hooks - Config.tsx : carte "Indicateurs techniques" dans Options—Paramètres avec toggles - auto_cycle.py : compute top-5 tickers à chaque cycle si tech_indicators_enabled=true - ai_analyzer.py : tech_indicators_block injecté dans suggestion + scoring prompts Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
OpenSquared
@@ -119,3 +119,31 @@ def save_options_gate(req: OptionsGateRequest):
|
||||
if req.iv_gate_skew_threshold is not None:
|
||||
set_config("iv_gate_skew_threshold", str(req.iv_gate_skew_threshold))
|
||||
return {"status": "ok"}
|
||||
|
||||
|
||||
# ── Tech Indicators config ──────────────────────────────────────────────────
|
||||
|
||||
class TechIndicatorsRequest(BaseModel):
|
||||
tech_indicators_enabled: Optional[bool] = None
|
||||
tech_indicators_list: Optional[str] = None # comma-separated: "rsi,ma,bollinger,atr"
|
||||
tech_indicators_auto_calibrate: Optional[bool] = None
|
||||
|
||||
|
||||
@router.get("/tech-indicators")
|
||||
def get_tech_indicators():
|
||||
return {
|
||||
"tech_indicators_enabled": (get_config("tech_indicators_enabled") or "true").lower() == "true",
|
||||
"tech_indicators_list": get_config("tech_indicators_list") or "rsi,ma,bollinger,atr",
|
||||
"tech_indicators_auto_calibrate": (get_config("tech_indicators_auto_calibrate") or "true").lower() == "true",
|
||||
}
|
||||
|
||||
|
||||
@router.put("/tech-indicators")
|
||||
def save_tech_indicators(req: TechIndicatorsRequest):
|
||||
if req.tech_indicators_enabled is not None:
|
||||
set_config("tech_indicators_enabled", "true" if req.tech_indicators_enabled else "false")
|
||||
if req.tech_indicators_list is not None:
|
||||
set_config("tech_indicators_list", req.tech_indicators_list)
|
||||
if req.tech_indicators_auto_calibrate is not None:
|
||||
set_config("tech_indicators_auto_calibrate", "true" if req.tech_indicators_auto_calibrate else "false")
|
||||
return {"status": "ok"}
|
||||
|
||||
@@ -353,6 +353,7 @@ def score_patterns_with_context(
|
||||
iv_context: str = "",
|
||||
risk_context: str = "",
|
||||
cycle_meta: Optional[Dict] = None,
|
||||
tech_indicators_block: str = "",
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""Score all patterns with rich context (news, prices, IV, risk clusters) using GPT-4o."""
|
||||
if not get_client():
|
||||
@@ -571,11 +572,14 @@ Instructions de notation:
|
||||
f"⚠️ Tiens compte du délai depuis le dernier cycle pour évaluer si les news sont déjà intégrées.\n"
|
||||
)
|
||||
|
||||
_tech_sc_section = f"\n{tech_indicators_block}\n" if tech_indicators_block else ""
|
||||
|
||||
user = f"""CONTEXTE GLOBAL:
|
||||
- Score risque géopolitique: {geo_score.get('score', 50)}/100 ({geo_score.get('level', 'medium')})
|
||||
- Top risques: {geo_score.get('top_risks', [])}
|
||||
{temporal_section_sc}
|
||||
{macro_section}
|
||||
{_tech_sc_section}
|
||||
TEMPLATE DE NOTATION:
|
||||
{scoring_template}
|
||||
|
||||
@@ -964,6 +968,7 @@ def suggest_patterns_from_market_context(
|
||||
reliability_map: Optional[Dict] = None,
|
||||
iv_context: str = "",
|
||||
cycle_meta: Optional[Dict] = None,
|
||||
tech_indicators_block: str = "",
|
||||
) -> List[Dict]:
|
||||
"""Ask GPT-4o to propose new patterns based on current geo/market + macro regime context."""
|
||||
_cycle_meta = cycle_meta or {}
|
||||
@@ -1109,12 +1114,14 @@ Règles supplémentaires:
|
||||
"## Actualités géopolitiques du moment (triées par impact)\n" + news_block
|
||||
)
|
||||
|
||||
tech_block_section = f"\n{tech_indicators_block}\n" if tech_indicators_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}
|
||||
{temporal_news_block}
|
||||
## Prix des marchés (variation J-1)
|
||||
{market_block}
|
||||
|
||||
{tech_block_section}
|
||||
## Calendrier économique à venir
|
||||
{cal_block}
|
||||
|
||||
|
||||
@@ -376,12 +376,43 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
|
||||
# Apply decay to news before suggestion (adds decayed_score + age_hours)
|
||||
from services.ai_analyzer import apply_news_decay as _apply_decay
|
||||
news = _apply_decay(news)
|
||||
|
||||
# ── Tech indicators for top tickers ──────────────────────────────
|
||||
_tech_block = ""
|
||||
try:
|
||||
_ti_enabled = (get_config("tech_indicators_enabled") or "true").lower() == "true"
|
||||
if _ti_enabled:
|
||||
from services.technical_indicators import compute_indicators, format_indicators_for_prompt
|
||||
_ti_list = [s.strip() for s in (get_config("tech_indicators_list") or "rsi,ma,bollinger,atr").split(",") if s.strip()]
|
||||
_horizon_days = 45 # default mid-range if no trade horizon context yet
|
||||
_ti_lines = []
|
||||
# Pick up to 5 tickers from quotes (one per asset class)
|
||||
_seen_tickers: set = set()
|
||||
for _cls, _qs in quotes.items():
|
||||
for _q in _qs[:1]:
|
||||
_sym = _q.get("symbol", "")
|
||||
if _sym and _sym not in _seen_tickers:
|
||||
_seen_tickers.add(_sym)
|
||||
_ind = compute_indicators(_sym, _horizon_days, enabled_indicators=_ti_list)
|
||||
_blk = format_indicators_for_prompt(_ind)
|
||||
if _blk:
|
||||
_ti_lines.append(_blk)
|
||||
if len(_seen_tickers) >= 5:
|
||||
break
|
||||
if len(_seen_tickers) >= 5:
|
||||
break
|
||||
if _ti_lines:
|
||||
_tech_block = "\n".join(_ti_lines)
|
||||
except Exception as _te:
|
||||
logger.warning(f"[Cycle] Tech indicators failed (non-blocking): {_te}")
|
||||
|
||||
suggestions = suggest_patterns_from_market_context(
|
||||
news, quotes, calendar, macro_regime=macro_regime, geo_score=geo_score_obj,
|
||||
portfolio_lessons=portfolio_lessons,
|
||||
reliability_map=_reliability_map or None,
|
||||
iv_context=iv_context,
|
||||
cycle_meta=cycle_meta,
|
||||
tech_indicators_block=_tech_block,
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(f"[Cycle] Suggestion step failed: {e}")
|
||||
@@ -505,6 +536,7 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
|
||||
iv_context=iv_context,
|
||||
risk_context=risk_cluster_context,
|
||||
cycle_meta=cycle_meta,
|
||||
tech_indicators_block=_tech_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")]
|
||||
|
||||
174
backend/services/technical_indicators.py
Normal file
174
backend/services/technical_indicators.py
Normal file
@@ -0,0 +1,174 @@
|
||||
"""
|
||||
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)
|
||||
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"].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", "")
|
||||
Reference in New Issue
Block a user