c6178c14d5
- 2 nouvelles tables SQLite : market_events + timeline_context
- 32 événements historiques seedés (long/medium/short de feb 2020 à juin 2026)
- timeline_service.py : bootstrap, get_events_for_date, génération commentaires GPT-4o-mini
- /api/timeline router : GET /day/{date}, GET /events, POST /generate/{date}, POST /bootstrap
- Timeline.tsx : navigateur date avec strip visuel, 3 panneaux contextuels, catalogue d'événements
- Sidebar : entrée Timeline avec icône Layers
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
114 lines
3.8 KiB
Python
114 lines
3.8 KiB
Python
"""
|
|
Timeline Navigator — historical market context browser.
|
|
Prefix: /api/timeline
|
|
"""
|
|
import logging
|
|
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
|
|
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
|
|
new_id = save_market_event(body.dict())
|
|
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.get("/day/{ref_date}")
|
|
def get_day_context(ref_date: str) -> Dict[str, Any]:
|
|
"""Return cached or freshly generated timeline context for a date (YYYY-MM-DD)."""
|
|
from services.database import get_events_for_date, get_timeline_context
|
|
import json
|
|
|
|
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]:
|
|
"""Generate (or regenerate) AI commentary for a specific date and cache it."""
|
|
from services.timeline_service import get_or_generate_context
|
|
import json
|
|
|
|
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": json.loads(ctx.get("evidence_headlines", "[]") or "[]") if isinstance(ctx.get("evidence_headlines"), str) else 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]:
|
|
"""Seed historical market events (idempotent — only runs if table is empty)."""
|
|
from services.timeline_service import bootstrap_events as _bootstrap
|
|
count = _bootstrap()
|
|
return {"seeded": count, "status": "ok" if count > 0 else "already_seeded"}
|