Spaces:
Running
Running
File size: 1,997 Bytes
cafdd88 2750cce cafdd88 2750cce cafdd88 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 | from fastapi import FastAPI, HTTPException, Depends
from core.schema import OptimizationRequest, OptimizationResult
from main import QuantScaleSystem
import logging
app = FastAPI(title="QuantScale AI API", version="1.0.0")
logger = logging.getLogger("API")
# Singleton System
system = QuantScaleSystem()
from fastapi.responses import RedirectResponse
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
# Mount static files
app.mount("/static", StaticFiles(directory="api/static"), name="static")
@app.get("/")
def root():
"""Serves the AI Interface."""
return FileResponse('api/static/index.html')
@app.get("/health")
def health_check():
return {"status": "healthy", "service": "QuantScale AI Direct Indexing"}
def parse_constraints_with_llm(user_prompt: str) -> list:
"""
Dedicated parser function in the API layer.
Maps natural language to exact GICS sectors.
"""
return system.ai_reporter.parse_intent(user_prompt)
@app.post("/optimize", response_model=dict)
def optimize_portfolio(request: OptimizationRequest):
"""
Optimizes a portfolio based on exclusions and generates an AI Attribution report.
"""
try:
# If the request contains a raw prompt but no sectors, parse it here
if request.user_prompt and not request.excluded_sectors:
request.excluded_sectors = parse_constraints_with_llm(request.user_prompt)
result = system.run_pipeline(request)
if not result:
raise HTTPException(status_code=500, detail="Pipeline failed to execute.")
return {
"client_id": request.client_id,
"allocations": result['optimization'].weights,
"tracking_error": result['optimization'].tracking_error,
"attribution_narrative": result['commentary']
}
except Exception as e:
logger.error(f"API Error: {e}")
raise HTTPException(status_code=500, detail=str(e))
|