""" Impact evaluation service. Given a market_event (calendar or geopolitical), calls the AI to estimate instrument-level impacts (score 0-1, direction, rationale, confidence). Results are persisted in instrument_impacts table. """ import json import logging import os from typing import Any, Dict, List, Optional logger = logging.getLogger(__name__) ALL_INSTRUMENTS = [ "SPY","QQQ","IWM","EEM","EFA", "GLD","SLV","USO","UNG","TLT", "HYG","EURUSD=X","USDJPY=X","GBPUSD=X", "VXX","AAPL","NVDA","GS","XOM","BTC-USD", ] INSTRUMENT_NAMES = { "SPY": "S&P 500 ETF", "QQQ": "Nasdaq 100 ETF", "IWM": "Russell 2000 ETF", "EEM": "Emerging Markets ETF", "EFA": "MSCI EAFE ETF", "GLD": "Gold ETF", "SLV": "Silver ETF", "USO": "WTI Oil ETF", "UNG": "Natural Gas ETF", "TLT": "US 20Y Treasury ETF", "HYG": "High Yield Credit ETF", "EURUSD=X": "EUR/USD", "USDJPY=X": "USD/JPY", "GBPUSD=X": "GBP/USD", "VXX": "VIX Tracker", "AAPL": "Apple Inc", "NVDA": "NVIDIA Corp", "GS": "Goldman Sachs", "XOM": "ExxonMobil", "BTC-USD": "Bitcoin", } _INST_LIST_STR = ", ".join(f"{k} ({v})" for k, v in INSTRUMENT_NAMES.items()) def _get_api_key() -> str: key = os.environ.get("OPENAI_API_KEY", "") if not key: from services.database import get_config key = get_config("openai_api_key") or "" return key _KNOWN_TYPES = {"geopolitical", "fundamental", "event_calendar", "report", "sentiment", "technical"} def _build_prompt( event: Dict[str, Any], all_cats: List[Dict], preselected_defaults: Optional[List[Dict]] = None, ) -> str: # ── Category selection block ────────────────────────────────────────────── if all_cats: # Number each category so the AI can refer to it unambiguously cat_lines = "\n".join( f" CAT_{i+1}: \"{c['name']}\"" + (f" ({c.get('description','')[:70]})" if c.get('description') else "") for i, c in enumerate(all_cats) ) # Build the valid names list for the JSON instruction valid_names = " | ".join(f'"{c["name"]}"' for c in all_cats[:6]) + " | ..." cat_selection_block = f""" LISTE DES CATÉGORIES DISPONIBLES (utilise uniquement ces noms) : {cat_lines} → Dans le JSON, "matched_category" DOIT être l'un des noms entre guillemets ci-dessus (ex: {valid_names}), ou null si vraiment aucune ne convient. NE retourne PAS un type générique comme "geopolitical" ou "fundamental" — ces valeurs sont invalides ici. """ else: cat_selection_block = "" # ── Default impacts from pre-selected category ──────────────────────────── defaults_block = "" if preselected_defaults: top = sorted(preselected_defaults, key=lambda x: x.get("sensitivity", 0), reverse=True) lines = [ f" • {d['instrument_id']}: sensibilité {d.get('sensitivity',0):.0%}, " f"direction usuelle = {d.get('typical_direction','?')}" + (f", note: {d.get('notes','')[:60]}" if d.get("notes") else "") for d in top ] defaults_block = ( "\nImpacts par défaut de la catégorie — COUVRE TOUS CES INSTRUMENTS dans ta réponse " "(ajuste scores/directions selon le contexte spécifique de l'événement) :\n" + "\n".join(lines) ) return f"""Tu es un analyste macro senior spécialisé en options et trading multi-actifs. ÉVÉNEMENT À ANALYSER : - Nom : {event.get('name', '')} - Date : {event.get('start_date', '')} - Famille : {event.get('category', '')} - Description : {event.get('description', '')} - Impact attendu : {event.get('market_impact', '')} {cat_selection_block}{defaults_block} MISSION — IMPACTS PAR INSTRUMENT : Choisis d'abord la catégorie la plus proche dans la liste ci-dessus, puis évalue l'impact sur chacun des instruments. Si une catégorie est choisie, utilise ses impacts par défaut comme point de départ. Instruments : {_INST_LIST_STR} Pour chaque instrument IMPACTÉ (score >= 0.15) : - instrument_id : identifiant exact (ex: "SPY", "EURUSD=X") - impact_score : 0.0 → 1.0 - direction : "bullish" | "bearish" | "neutral" | "depends_on_outcome" - rationale : 1 phrase MAX - confidence : 0.0-1.0 FORMAT JSON STRICT : {{ "matched_category": "Nom exact d'une CAT ci-dessus, ou null", "impacts": [ {{"instrument_id": "SPY", "impact_score": 0.85, "direction": "bearish", "rationale": "...", "confidence": 0.90}} ], "summary": "Contexte macro en 1 phrase", "key_driver": "Instrument le plus impacté et pourquoi" }}""" def evaluate_event_impacts( event_id: int, force: bool = False, ) -> Dict[str, Any]: """ Evaluate instrument impacts for a market_event via AI. Saves results to instrument_impacts table. Returns the list of impacts + AI summary. """ from services.database import ( get_all_market_events, get_event_category, get_impacts_for_source, delete_impacts_for_source, save_instrument_impacts, ) # Fetch event events = get_all_market_events() event = next((e for e in events if e["id"] == event_id), None) if not event: return {"error": f"Event {event_id} not found"} # Check if already evaluated existing = get_impacts_for_source("event", event_id) if existing and not force: return { "event_id": event_id, "impacts": existing, "summary": "", "from_cache": True, "n_instruments": len(existing), } if force and existing: delete_impacts_for_source("event", event_id) from services.database import get_all_event_categories all_cats_raw = get_all_event_categories() # Parse default_impacts JSON for each category all_cats: List[Dict] = [] for c in all_cats_raw: try: c["default_impacts"] = json.loads(c.get("default_impacts") or "[]") except Exception: c["default_impacts"] = [] all_cats.append(c) # Heuristic pre-selection: if sub_type matches a category exactly, pass its # defaults as a hint (the AI will still pick the best category independently) sub = (event.get("sub_type") or "").strip() preselected_defaults: Optional[List[Dict]] = None if sub: pre = next((c for c in all_cats if c["name"].lower() == sub.lower()), None) if pre: preselected_defaults = pre["default_impacts"] api_key = _get_api_key() if not api_key: return {"error": "No OpenAI API key configured"} prompt = _build_prompt(event, all_cats, preselected_defaults) matched_cat_name: Optional[str] = None try: from openai import OpenAI client = OpenAI(api_key=api_key) response = client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}], response_format={"type": "json_object"}, temperature=0.25, max_tokens=2000, ) raw = json.loads(response.choices[0].message.content) except Exception as e: logger.error(f"[impact_service] AI call failed for event {event_id}: {e}") return {"error": str(e)} ai_impacts = raw.get("impacts", []) summary = raw.get("summary", "") key_driver = raw.get("key_driver", "") # ── Use AI-chosen category ──────────────────────────────────────────────── import unicodedata, re as _re def _normalize(s: str) -> str: """Lowercase, strip accents, collapse all dash/space variants.""" s = unicodedata.normalize("NFD", s) s = "".join(c for c in s if unicodedata.category(c) != "Mn") # strip accents s = _re.sub(r"[–—―\-–—]", "-", s) # normalize dashes return s.lower().strip() ai_cat_name = (raw.get("matched_category") or "").strip() logger.info(f"[impact_service] AI returned matched_category='{ai_cat_name}' for event {event_id}") if ai_cat_name and ai_cat_name.lower() not in ("none", "null", "") and ai_cat_name.lower() not in _KNOWN_TYPES: # Exact match first, then normalized fuzzy match matched = ( next((c for c in all_cats if c["name"].lower() == ai_cat_name.lower()), None) or next((c for c in all_cats if _normalize(c["name"]) == _normalize(ai_cat_name)), None) # Partial: AI name is contained in category name or vice versa or next((c for c in all_cats if _normalize(ai_cat_name) in _normalize(c["name"]) or _normalize(c["name"]) in _normalize(ai_cat_name)), None) ) if matched: matched_cat_name = matched["name"] logger.info(f"[impact_service] Matched event {event_id} → '{matched_cat_name}'") current_sub = (event.get("sub_type") or "").strip() if not current_sub: try: from services.database import update_market_event, get_all_market_events # Re-fetch full event to get proper JSON fields full_ev = next((e for e in get_all_market_events() if e["id"] == event_id), None) if full_ev: refs = full_ev.get("source_refs") or "[]" if isinstance(refs, str): refs = json.loads(refs) update_market_event(event_id, {**full_ev, "sub_type": matched_cat_name, "source_refs": refs}) logger.info(f"[impact_service] Auto-set sub_type='{matched_cat_name}' on event {event_id}") except Exception as _e: logger.warning(f"[impact_service] Could not auto-set sub_type: {_e}") else: logger.warning(f"[impact_service] No category match for AI response '{ai_cat_name}' (event {event_id})") cat_label = matched_cat_name or (event.get("sub_type") or "").strip() or (event.get("category") or "") # ── Merge: category defaults (baseline) + AI adjustments ───────────────── # 1. Start with ALL category defaults base: Dict[str, Dict] = {} if matched_cat_name: cat_obj = next((c for c in all_cats if c["name"] == matched_cat_name), None) if cat_obj: for d in cat_obj.get("default_impacts", []): tid = d.get("instrument_id") if tid and tid in ALL_INSTRUMENTS: base[tid] = { "source_type": "event", "source_id": event_id, "source_name": event["name"], "source_date": event["start_date"], "category_name": cat_label, "instrument_id": tid, "impact_score": float(d.get("sensitivity", 0.5)), "direction": d.get("typical_direction", "neutral"), "rationale": d.get("notes", "") or f"Défaut catégorie {cat_label}", "confidence": 0.70, "ai_generated": 0, } # 2. AI impacts override / extend the baseline for imp in ai_impacts: tid = imp.get("instrument_id") if not tid or tid not in ALL_INSTRUMENTS: continue base[tid] = { "source_type": "event", "source_id": event_id, "source_name": event["name"], "source_date": event["start_date"], "category_name": cat_label, "instrument_id": tid, "impact_score": float(imp.get("impact_score", 0.5)), "direction": imp.get("direction", "neutral"), "rationale": imp.get("rationale", ""), "confidence": float(imp.get("confidence", 0.7)), "ai_generated": 1, } to_save = list(base.values()) save_instrument_impacts(to_save) saved = get_impacts_for_source("event", event_id) return { "event_id": event_id, "event_name": event["name"], "matched_category": matched_cat_name, "impacts": saved, "summary": summary, "key_driver": key_driver, "n_instruments": len(saved), "from_cache": False, } def get_impact_monitor_data(days: int = 7, min_score: float = 0.3) -> Dict[str, Any]: """ Return all evaluated sources (events + news) from the last N days, filtered by min_score, with their instrument impacts. Also includes unevaluated events for quick launch. """ from services.database import ( get_weekly_impact_sources, get_all_market_events, get_impacts_for_source, ) from datetime import datetime, timedelta cutoff = (datetime.utcnow() - timedelta(days=days)).strftime("%Y-%m-%d") # Evaluated sources evaluated = get_weekly_impact_sources(days=days, min_score=min_score) # All recent events not yet evaluated all_events = get_all_market_events() evaluated_ids = {(s["source_type"], s.get("source_id")) for s in evaluated} unevaluated = [] for ev in all_events: if ev.get("start_date", "") < cutoff: continue if ("event", ev["id"]) not in evaluated_ids: unevaluated.append({ "id": ev["id"], "name": ev["name"], "start_date": ev["start_date"], "category": ev.get("category",""), "sub_type": ev.get("sub_type",""), "impact_score": ev.get("impact_score", 0.5), }) # Top impacted instruments this period from collections import defaultdict inst_scores: Dict[str, List[float]] = defaultdict(list) for source in evaluated: for imp in source.get("impacts", []): score = imp.get("adjusted_score") or imp.get("impact_score", 0) if score >= min_score: inst_scores[imp["instrument_id"]].append(score) top_instruments = sorted( [{"instrument_id": k, "avg_score": sum(v)/len(v), "n_events": len(v)} for k, v in inst_scores.items()], key=lambda x: x["avg_score"] * x["n_events"], reverse=True )[:10] return { "evaluated": evaluated, "unevaluated": sorted(unevaluated, key=lambda x: x["start_date"], reverse=True), "top_instruments": top_instruments, "period_days": days, "min_score": min_score, }