862 lines
33 KiB
Python
862 lines
33 KiB
Python
"""
|
|
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__})."
|