OpenFin/backend/services/impact_service.py

"""
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


def _build_prompt(
    event: Dict[str, Any],
    all_cats: List[Dict],
    preselected_defaults: Optional[List[Dict]] = None,
) -> str:
    # ── Category selection block ──────────────────────────────────────────────
    if all_cats:
        cat_lines = "\n".join(
            f"  {i+1}. [{c['type']}] {c['name']}"
            + (f" — {c.get('sub_type','')}" if c.get('sub_type') else "")
            + (f" : {c.get('description','')[:80]}" if c.get('description') else "")
            for i, c in enumerate(all_cats)
        )
        cat_selection_block = f"""
ÉTAPE 1 — CHOIX DE CATÉGORIE :
Voici toutes les catégories disponibles. Choisis celle qui correspond le mieux à cet événement (ou "none" si aucune ne convient) :
{cat_lines}

Retourne le nom EXACT de la catégorie choisie dans le champ "matched_category" du JSON.
"""
    else:
        cat_selection_block = ""

    # ── Default impacts from pre-selected category (fallback heuristic) ───────
    defaults_block = ""
    if preselected_defaults:
        top = sorted(preselected_defaults, key=lambda x: x.get("sensitivity", 0), reverse=True)[:8]
        lines = [
            f"  • {d['instrument_id']}: sensibilité {d.get('sensitivity',0):.0%}, "
            f"direction usuelle = {d.get('typical_direction','?')}"
            for d in top
        ]
        defaults_block = "\nImpacts par défaut (ajuste si l'événement est atypique) :\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', '')}
- Type : {event.get('category', '')} / {event.get('sub_type', '')}
- Description : {event.get('description', '')}
- Impact attendu (résumé) : {event.get('market_impact', '')}
{cat_selection_block}{defaults_block}

ÉTAPE 2 — IMPACTS PAR INSTRUMENT :
Évalue l'impact potentiel de cet événement sur chacun des 20 instruments.
Si tu as choisi une catégorie à l'étape 1, utilise ses impacts par défaut comme point de départ et ajuste selon le contexte spécifique de l'événement.
Instruments disponibles : {_INST_LIST_STR}

Pour chaque instrument IMPACTÉ (score >= 0.15), retourne :
- instrument_id : identifiant exact (ex: "SPY", "EURUSD=X", "BTC-USD")
- impact_score : 0.0 (aucun) → 1.0 (impact majeur, mouvement > 2%)
- direction : "bullish" | "bearish" | "neutral" | "depends_on_outcome"
- rationale : 1 phrase MAX expliquant le mécanisme
- confidence : 0.0-1.0 (ta certitude sur l'estimation)

Critères de scoring :
- 0.0-0.2 : négligeable (bruit)
- 0.2-0.4 : faible (< 0.5% mouvement)
- 0.4-0.6 : modéré (0.5-1% mouvement)
- 0.6-0.8 : fort (1-2% mouvement)
- 0.8-1.0 : majeur (> 2%, event driver primaire)

FORMAT JSON STRICT :
{{
  "matched_category": "Nom exact de la catégorie choisie ou null",
  "impacts": [
    {{"instrument_id": "SPY", "impact_score": 0.85, "direction": "bearish", "rationale": "...", "confidence": 0.90}}
  ],
  "summary": "Contexte macro de cet événement 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=1200,
        )
        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 ────────────────────────────────────────────────
    ai_cat_name = (raw.get("matched_category") or "").strip()
    if ai_cat_name and ai_cat_name.lower() != "none":
        # Validate against known categories (case-insensitive)
        matched = next((c for c in all_cats if c["name"].lower() == ai_cat_name.lower()), None)
        if matched:
            matched_cat_name = matched["name"]
            logger.info(f"[impact_service] AI matched event {event_id} → category '{matched_cat_name}'")
            # Auto-update event sub_type if not already set
            current_sub = (event.get("sub_type") or "").strip()
            if not current_sub:
                try:
                    from services.database import update_market_event
                    update_market_event(event_id, {**event, "sub_type": matched_cat_name,
                                                   "source_refs": event.get("source_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] AI returned unknown category '{ai_cat_name}' for event {event_id}")

    cat_label = matched_cat_name or (event.get("sub_type") or "").strip() or (event.get("category") or "")

    # Persist
    to_save = []
    for imp in ai_impacts:
        if imp.get("instrument_id") not in ALL_INSTRUMENTS:
            continue
        to_save.append({
            "source_type":   "event",
            "source_id":     event_id,
            "source_name":   event["name"],
            "source_date":   event["start_date"],
            "category_name": cat_label,
            "instrument_id": imp["instrument_id"],
            "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,
        })

    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,
    }