""" Timeline Navigator — historical market context browser. Prefix: /api/timeline """ import json import logging import os from typing import Any, Dict, List, Optional from fastapi import APIRouter, HTTPException from pydantic import BaseModel logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/timeline", tags=["timeline"]) class EventCreate(BaseModel): name: str start_date: str end_date: Optional[str] = None level: str # long | medium | short category: str = "macro" description: str = "" market_impact: str = "" affected_assets: List[str] = [] impact_score: float = 0.5 absorption_pct: Optional[float] = None relevant_indicators: List[Dict[str, Any]] = [] parent_event_id: Optional[int] = None class EventUpdate(EventCreate): pass @router.get("/events") def list_events() -> List[Dict[str, Any]]: from services.database import get_all_market_events return get_all_market_events() @router.post("/events") def create_event(body: EventCreate) -> Dict[str, Any]: from services.database import save_market_event data = body.dict() data.setdefault("origin", "manual") new_id = save_market_event(data) return {"id": new_id, "status": "created"} @router.put("/events/{event_id}") def update_event(event_id: int, body: EventUpdate) -> Dict[str, Any]: from services.database import update_market_event ok = update_market_event(event_id, body.dict()) if not ok: raise HTTPException(404, "Event not found") return {"status": "updated"} @router.delete("/events/{event_id}") def delete_event(event_id: int) -> Dict[str, Any]: from services.database import delete_market_event delete_market_event(event_id) return {"status": "deleted"} @router.post("/events/{event_id}/ai-enrich") def ai_enrich_event(event_id: int) -> Dict[str, Any]: """Ask GPT-4o-mini to suggest absorption_pct + relevant_indicators for this event.""" from services.database import get_all_market_events, update_market_event from datetime import datetime api_key = os.environ.get("OPENAI_API_KEY", "") if not api_key: raise HTTPException(400, "OpenAI API key not configured") # Fetch event all_evs = get_all_market_events() ev = next((e for e in all_evs if e["id"] == event_id), None) if not ev: raise HTTPException(404, "Event not found") from openai import OpenAI client = OpenAI(api_key=api_key) today = datetime.utcnow().strftime("%Y-%m-%d") days_elapsed = max(0, (datetime.strptime(today, "%Y-%m-%d") - datetime.strptime(ev["start_date"][:10], "%Y-%m-%d")).days) is_ongoing = not ev.get("end_date") prompt = f"""Tu es un analyste macro senior spécialisé en options géopolitiques. Événement à analyser: - Nom: {ev['name']} - Niveau: {ev['level']} (long=structurel, medium=régime, short=catalyseur) - Catégorie: {ev['category']} - Début: {ev['start_date']} - Fin: {ev.get('end_date') or 'en cours'} - Jours écoulés depuis début: {days_elapsed}j - Description: {ev['description']} - Impact marché: {ev.get('market_impact', '')} - Assets affectés: {ev.get('affected_assets', '[]')} Ta mission: 1. Estime le taux d'absorption (0-100%) de cet événement par le marché à ce jour. Absorption = à quel point l'impact attendu est déjà "pricé" par les marchés. Exemples: 0%=pas encore pricé, 50%=à moitié absorbé, 100%=fully priced in. 2. Suggère 3-5 indicateurs techniques PERTINENTS pour suivre cet événement dans le contexte d'un trader d'options. Pour chaque indicateur: - symbol: ticker yfinance (ex: "CL=F", "GC=F", "SPY", "^VIX") - indicator: type ("MA10", "MA20", "MA50", "MA100", "price", "RSI14") - label: nom affiché (ex: "Brent MA100", "Gold prix spot", "VIX 20j avg") - rationale: 1 phrase pourquoi cet indicateur est pertinent POUR CET ÉVÉNEMENT PRÉCIS Règle: choisis les indicateurs selon le niveau temporel: - Long terme → préfère MA100, MA50 (tendances structurelles) - Moyen terme → MA20, MA50 - Court terme → MA10, MA20, prix spot FORMAT JSON STRICT: {{ "absorption_pct": <0-100>, "absorption_rationale": "<1 phrase expliquant pourquoi ce niveau>", "relevant_indicators": [ {{"symbol": "...", "indicator": "...", "label": "...", "rationale": "..."}} ] }}""" try: response = client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}], response_format={"type": "json_object"}, temperature=0.3, max_tokens=600, ) result = json.loads(response.choices[0].message.content) # Update event in DB ev_update = dict(ev) ev_update["affected_assets"] = json.loads(ev.get("affected_assets", "[]") or "[]") ev_update["relevant_indicators"] = result.get("relevant_indicators", []) ev_update["absorption_pct"] = result.get("absorption_pct") update_market_event(event_id, ev_update) return { "event_id": event_id, "absorption_pct": result.get("absorption_pct"), "absorption_rationale": result.get("absorption_rationale", ""), "relevant_indicators": result.get("relevant_indicators", []), "status": "enriched", } except Exception as e: logger.error(f"[Timeline] ai_enrich failed for event {event_id}: {e}") raise HTTPException(500, str(e)) @router.post("/bootstrap-macro") def bootstrap_macro(force: bool = False) -> Dict[str, Any]: """ Seed the market_events table with predefined historical macro + geopolitical events. Pass ?force=true to re-run even if events already exist. """ try: from services.macro_events_bootstrap import bootstrap_macro_events result = bootstrap_macro_events(force=force) return result except Exception as e: logger.error(f"[Timeline] bootstrap_macro failed: {e}") raise HTTPException(500, str(e)) @router.get("/day/{ref_date}") def get_day_context(ref_date: str) -> Dict[str, Any]: from services.database import get_events_for_date, get_timeline_context events = get_events_for_date(ref_date) cached = get_timeline_context(ref_date) result: Dict[str, Any] = { "ref_date": ref_date, "events": events, "long_commentary": "", "medium_commentary": "", "short_commentary": "", "evidence_headlines": [], "generated_at": None, "has_commentary": False, } if cached: result.update({ "long_commentary": cached.get("long_commentary", ""), "medium_commentary": cached.get("medium_commentary", ""), "short_commentary": cached.get("short_commentary", ""), "evidence_headlines": json.loads(cached.get("evidence_headlines", "[]") or "[]"), "generated_at": cached.get("generated_at"), "has_commentary": bool(cached.get("long_commentary")), }) return result @router.post("/generate/{ref_date}") def generate_commentary(ref_date: str) -> Dict[str, Any]: from services.timeline_service import get_or_generate_context try: ctx = get_or_generate_context(ref_date) return { "ref_date": ref_date, "long_commentary": ctx.get("long_commentary", ""), "medium_commentary": ctx.get("medium_commentary", ""), "short_commentary": ctx.get("short_commentary", ""), "evidence_headlines": ctx.get("evidence_headlines", []), "events": ctx.get("events", {}), "has_commentary": True, } except Exception as e: logger.error(f"[Timeline] generate_commentary failed for {ref_date}: {e}") raise HTTPException(500, str(e)) @router.post("/bootstrap") def bootstrap_events() -> Dict[str, Any]: from services.timeline_service import bootstrap_events as _bootstrap count = _bootstrap() return {"seeded": count, "status": "ok" if count > 0 else "already_seeded"} @router.post("/bootstrap-eco") def bootstrap_eco(force: bool = False) -> Dict[str, Any]: """ Seed the market_events table with economic calendar events (2020-2026). Includes FOMC, NFP, CPI, GDP, ISM, BOJ, ECB, BOE with expected/actual/surprise fields. Pass ?force=true to delete existing eco events and re-insert. """ try: from services.eco_calendar_bootstrap import bootstrap_eco_events result = bootstrap_eco_events(force=force) return result except Exception as e: logger.error(f"[Timeline] bootstrap_eco failed: {e}") raise HTTPException(500, str(e)) @router.post("/bootstrap-ma") async def bootstrap_ma() -> Dict[str, Any]: """Detect MA ruptures on 5 key underlyings and generate historical market events via GPT.""" from services.ma_analyzer import bootstrap_ma_events try: result = await bootstrap_ma_events() return {**result, "status": "ok"} except Exception as e: logger.error(f"[Timeline] bootstrap-ma failed: {e}") raise HTTPException(500, str(e))