537fea8148
- 20 instruments configured (equity indices, metals, energy, bonds, FX, stocks, crypto) each with custom drivers, regime labels, MA periods, event keywords, ai_context - InstrumentChart: TradingView lightweight-charts candlesticks + MA lines + Bollinger + volume histogram + macro event markers overlaid on price - InstrumentDashboard: regime detection card (scores + signals), trend indicators (RSI gauge, MA slopes, momentum, 52W range), events card (links to Timeline), AI narrative via GPT-4o-mini (cached by day) - Backend: instrument_service (OHLCV fetch, indicators, regime scoring, GPT narrative) + /api/instruments router (3 endpoints) - Route: /instruments/:id with selector dropdown, period buttons (3M/6M/1Y/2Y/5Y) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
66 lines
2.0 KiB
Python
66 lines
2.0 KiB
Python
"""
|
|
Instrument Dashboard Router.
|
|
Exposes per-instrument snapshot (price, indicators, regime, trend, events) and AI narrative.
|
|
"""
|
|
from fastapi import APIRouter, HTTPException, Query
|
|
from typing import List, Dict, Any, Optional
|
|
|
|
from services.instrument_service import (
|
|
get_all_instruments,
|
|
get_instrument,
|
|
get_snapshot,
|
|
get_narrative,
|
|
)
|
|
|
|
router = APIRouter(prefix="/api/instruments", tags=["instruments"])
|
|
|
|
|
|
@router.get("", response_model=List[Dict[str, Any]])
|
|
def list_instruments() -> List[Dict[str, Any]]:
|
|
"""
|
|
Return all instrument configurations (no price data).
|
|
"""
|
|
return get_all_instruments()
|
|
|
|
|
|
@router.get("/{instrument_id}/snapshot")
|
|
async def instrument_snapshot(
|
|
instrument_id: str,
|
|
period: str = Query(default="1y", description="yfinance period string (e.g. 1y, 6mo, 3mo)"),
|
|
interval: str = Query(default="1d", description="yfinance interval string (e.g. 1d, 1wk)"),
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Full snapshot for a single instrument: price data, indicators, regime, trend, events.
|
|
"""
|
|
config = get_instrument(instrument_id)
|
|
if not config:
|
|
raise HTTPException(status_code=404, detail=f"Instrument '{instrument_id}' not found")
|
|
|
|
snapshot = await get_snapshot(instrument_id, period=period, interval=interval)
|
|
|
|
if "error" in snapshot:
|
|
raise HTTPException(status_code=500, detail=snapshot["error"])
|
|
|
|
return snapshot
|
|
|
|
|
|
@router.post("/{instrument_id}/narrative")
|
|
async def generate_narrative(
|
|
instrument_id: str,
|
|
force: bool = Query(default=False, description="Force regeneration, bypass cache"),
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Generate (or return cached) a French AI narrative for the instrument.
|
|
"""
|
|
config = get_instrument(instrument_id)
|
|
if not config:
|
|
raise HTTPException(status_code=404, detail=f"Instrument '{instrument_id}' not found")
|
|
|
|
narrative = await get_narrative(instrument_id, force=force)
|
|
|
|
return {
|
|
"instrument_id": instrument_id.upper(),
|
|
"instrument_name": config.get("name", instrument_id),
|
|
"narrative": narrative,
|
|
}
|