994 lines
34 KiB
Python
994 lines
34 KiB
Python
#!/usr/bin/env python3
|
|
"""Vibe-Trading API Server - RESTful API for finance research and backtesting.
|
|
|
|
V5: ReAct Agent + async /run + CORS env + SSE tool events.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import signal
|
|
import time
|
|
import csv
|
|
import uuid
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from fastapi import BackgroundTasks, Depends, FastAPI, HTTPException, Query, Request, Security, UploadFile, status
|
|
from fastapi.responses import StreamingResponse
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
from pydantic import BaseModel, Field
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from rich.console import Console
|
|
|
|
from src.ui_services import build_run_analysis, load_run_context
|
|
|
|
# UTF-8 on Windows
|
|
import sys as _sys
|
|
for _s in ("stdout", "stderr"):
|
|
_r = getattr(getattr(_sys, _s, None), "reconfigure", None)
|
|
if callable(_r):
|
|
_r(encoding="utf-8", errors="replace")
|
|
|
|
RUNS_DIR = Path(__file__).resolve().parent / "runs"
|
|
SESSIONS_DIR = Path(__file__).resolve().parent / "sessions"
|
|
UPLOADS_DIR = Path(__file__).resolve().parent / "uploads"
|
|
|
|
MAX_UPLOAD_SIZE = 50 * 1024 * 1024 # 50 MB
|
|
|
|
# Rich console for colored logs
|
|
console = Console()
|
|
|
|
|
|
# ============================================================================
|
|
# Pydantic Models
|
|
# ============================================================================
|
|
|
|
class Artifact(BaseModel):
|
|
"""Artifact file metadata."""
|
|
name: str = Field(..., description="File name")
|
|
path: str = Field(..., description="File path")
|
|
type: str = Field(..., description="File type: csv, json, txt, etc.")
|
|
size: int = Field(..., description="Size in bytes")
|
|
exists: bool = Field(..., description="Whether the file exists")
|
|
|
|
|
|
class BacktestMetrics(BaseModel):
|
|
"""Backtest summary metrics."""
|
|
model_config = {"extra": "allow"}
|
|
|
|
final_value: float = Field(..., description="Ending portfolio value")
|
|
total_return: float = Field(..., description="Total return")
|
|
annual_return: float = Field(..., description="Annualized return")
|
|
max_drawdown: float = Field(..., description="Max drawdown")
|
|
sharpe: float = Field(..., description="Sharpe ratio")
|
|
win_rate: float = Field(..., description="Win rate")
|
|
trade_count: int = Field(..., description="Number of trades")
|
|
|
|
|
|
|
|
class RAGSelection(BaseModel):
|
|
"""RAG routing result."""
|
|
selected_api: str = Field(..., description="Selected API code")
|
|
selected_name: str = Field(..., description="Selected API name")
|
|
selected_score: float = Field(..., description="Match score")
|
|
|
|
|
|
class RunInfo(BaseModel):
|
|
"""Compact run row for list views."""
|
|
run_id: str
|
|
status: str
|
|
created_at: str
|
|
prompt: Optional[str] = None
|
|
total_return: Optional[float] = None
|
|
sharpe: Optional[float] = None
|
|
codes: List[str] = Field(default_factory=list)
|
|
start_date: Optional[str] = None
|
|
end_date: Optional[str] = None
|
|
|
|
|
|
class RunResponse(BaseModel):
|
|
"""API response payload for a single run."""
|
|
|
|
status: str = Field(..., description="Run status: success, failed, aborted")
|
|
run_id: str = Field(..., description="Run identifier")
|
|
elapsed_seconds: float = Field(..., description="Execution time in seconds")
|
|
reason: Optional[str] = Field(None, description="Failure reason when available")
|
|
|
|
planner_output: Optional[Dict[str, Any]] = Field(None, description="Planner output")
|
|
strategy_spec: Optional[Dict[str, Any]] = Field(None, description="Strategy specification")
|
|
rag_selection: Optional[RAGSelection] = Field(None, description="Selected RAG metadata")
|
|
|
|
metrics: Optional[BacktestMetrics] = Field(None, description="Backtest metrics")
|
|
artifacts: List[Artifact] = Field(default_factory=list, description="Run artifacts")
|
|
|
|
equity_curve: Optional[List[Dict[str, Any]]] = Field(None, description="Equity preview")
|
|
trade_log: Optional[List[Dict[str, Any]]] = Field(None, description="Trade preview")
|
|
|
|
artifacts_equity_csv: Optional[List[Dict[str, Any]]] = Field(None, description="Full equity rows")
|
|
artifacts_metrics_csv: Optional[List[Dict[str, Any]]] = Field(None, description="Full metrics rows")
|
|
artifacts_trades_csv: Optional[List[Dict[str, Any]]] = Field(None, description="Full trade rows")
|
|
|
|
run_directory: str = Field(..., description="Run directory path")
|
|
run_stage: Optional[str] = Field(None, description="UI-facing run stage")
|
|
run_context: Optional[Dict[str, Any]] = Field(None, description="Normalized request context")
|
|
price_series: Optional[Dict[str, List[Dict[str, Any]]]] = Field(None, description="Grouped OHLC series")
|
|
indicator_series: Optional[Dict[str, Dict[str, List[Dict[str, Any]]]]] = Field(
|
|
None,
|
|
description="Grouped indicator overlays",
|
|
)
|
|
trade_markers: Optional[List[Dict[str, Any]]] = Field(None, description="Trade markers for charts")
|
|
run_logs: Optional[List[Dict[str, Any]]] = Field(None, description="Structured stdout/stderr lines")
|
|
|
|
|
|
class HealthResponse(BaseModel):
|
|
"""Health check payload."""
|
|
status: str = Field(..., description="Service status")
|
|
service: str = Field(..., description="Service name")
|
|
timestamp: str = Field(..., description="Server timestamp")
|
|
|
|
|
|
# ---- V4 Session Models ----
|
|
|
|
class CreateSessionRequest(BaseModel):
|
|
"""Create session request body."""
|
|
title: str = Field("", description="Session title")
|
|
config: Optional[Dict[str, Any]] = Field(None, description="Session config")
|
|
|
|
|
|
class SessionResponse(BaseModel):
|
|
"""Session record."""
|
|
session_id: str
|
|
title: str
|
|
status: str
|
|
created_at: str
|
|
updated_at: str
|
|
last_attempt_id: Optional[str] = None
|
|
|
|
|
|
class SendMessageRequest(BaseModel):
|
|
"""Send chat message: natural-language strategy description."""
|
|
content: str = Field(..., description="Natural language strategy description", min_length=1, max_length=5000)
|
|
|
|
|
|
class MessageResponse(BaseModel):
|
|
"""Stored chat message."""
|
|
message_id: str
|
|
session_id: str
|
|
role: str
|
|
content: str
|
|
created_at: str
|
|
linked_attempt_id: Optional[str] = None
|
|
metadata: Optional[Dict[str, Any]] = None
|
|
|
|
|
|
|
|
# ============================================================================
|
|
# FastAPI Application
|
|
# ============================================================================
|
|
|
|
app = FastAPI(
|
|
title="Vibe-Trading API",
|
|
description="Vibe-Trading API: natural-language finance research, backtesting, and swarm workflows",
|
|
version="5.0.0",
|
|
docs_url="/docs",
|
|
redoc_url="/redoc"
|
|
)
|
|
|
|
# CORS: override with CORS_ORIGINS (comma-separated)
|
|
_CORS_ORIGINS = os.getenv(
|
|
"CORS_ORIGINS",
|
|
"http://localhost:3000,http://localhost:5173,http://localhost:8000,"
|
|
"http://127.0.0.1:3000,http://127.0.0.1:5173,http://127.0.0.1:8000",
|
|
).split(",")
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=_CORS_ORIGINS,
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
|
|
# ============================================================================
|
|
# API Key Authentication
|
|
# ============================================================================
|
|
|
|
_security = HTTPBearer(auto_error=False)
|
|
_API_KEY = os.getenv("API_AUTH_KEY")
|
|
|
|
|
|
async def require_auth(
|
|
cred: HTTPAuthorizationCredentials = Security(_security),
|
|
) -> None:
|
|
"""Validate Bearer token against API_AUTH_KEY environment variable.
|
|
|
|
If API_AUTH_KEY is not set, authentication is skipped (dev mode).
|
|
Only write endpoints (POST/PUT/DELETE/PATCH) use this dependency.
|
|
|
|
Args:
|
|
cred: HTTP Bearer credentials extracted from the Authorization header.
|
|
|
|
Raises:
|
|
HTTPException: 401 when API_AUTH_KEY is set but the token is missing or wrong.
|
|
"""
|
|
if not _API_KEY:
|
|
return
|
|
if not cred or cred.credentials != _API_KEY:
|
|
raise HTTPException(status_code=401, detail="Invalid or missing API key")
|
|
|
|
|
|
# ============================================================================
|
|
# Workflow Factory
|
|
# ============================================================================
|
|
|
|
# ============================================================================
|
|
# Helper Functions
|
|
# ============================================================================
|
|
|
|
def _load_json_file(path: Path) -> Optional[Dict[str, Any]]:
|
|
"""Load JSON from disk if present."""
|
|
try:
|
|
if path.exists():
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
|
|
def _load_csv_to_dict(path: Path, limit: Optional[int] = None) -> List[Dict[str, Any]]:
|
|
"""Load CSV rows into a list of dictionaries."""
|
|
try:
|
|
if not path.exists():
|
|
return []
|
|
with path.open("r", encoding="utf-8", newline="") as handle:
|
|
rows = [dict(row) for row in csv.DictReader(handle)]
|
|
if limit is not None:
|
|
rows = rows[:limit]
|
|
return rows
|
|
except Exception:
|
|
return []
|
|
|
|
|
|
|
|
def _build_response_from_run_dir(run_dir: Path, elapsed: float, *, include_analysis: bool = False) -> RunResponse:
|
|
"""Build a run response from a persisted run directory."""
|
|
run_id = run_dir.name
|
|
|
|
response = RunResponse(
|
|
status="unknown",
|
|
run_id=run_id,
|
|
elapsed_seconds=elapsed,
|
|
run_directory=str(run_dir),
|
|
)
|
|
|
|
state_data = _load_json_file(run_dir / "state.json")
|
|
if state_data:
|
|
state_status = str(state_data.get("status") or "").lower()
|
|
if state_status == "success":
|
|
response.status = "success"
|
|
elif state_status == "failed":
|
|
response.status = "failed"
|
|
response.reason = state_data.get("reason", "")
|
|
else:
|
|
response.status = state_status or "unknown"
|
|
else:
|
|
response.status = "unknown"
|
|
|
|
planner_path = run_dir / "planner_output.json"
|
|
response.planner_output = _load_json_file(planner_path)
|
|
|
|
design_path = run_dir / "design_spec.json"
|
|
response.strategy_spec = _load_json_file(design_path)
|
|
|
|
rag_path = run_dir / "rag_metadata.json"
|
|
rag_data = _load_json_file(rag_path)
|
|
if rag_data:
|
|
response.rag_selection = RAGSelection(
|
|
selected_api=rag_data.get("selected_api") or rag_data.get("api_code", ""),
|
|
selected_name=rag_data.get("selected_name") or rag_data.get("api_name", ""),
|
|
selected_score=float(rag_data.get("selected_score") or rag_data.get("score", 0.0)),
|
|
)
|
|
|
|
metrics_path = run_dir / "artifacts" / "metrics.csv"
|
|
if metrics_path.exists():
|
|
metrics_dict_list = _load_csv_to_dict(metrics_path, limit=1)
|
|
if metrics_dict_list:
|
|
row = metrics_dict_list[0]
|
|
try:
|
|
# Pass ALL CSV columns to BacktestMetrics (extra="allow")
|
|
parsed: dict = {}
|
|
for k, v in row.items():
|
|
if not k or not v:
|
|
continue
|
|
try:
|
|
parsed[k] = int(float(v)) if k == "trade_count" or k == "max_consecutive_loss" else float(v)
|
|
except (ValueError, TypeError):
|
|
continue
|
|
if "final_value" in parsed:
|
|
response.metrics = BacktestMetrics(**parsed)
|
|
except (ValueError, TypeError):
|
|
pass
|
|
|
|
|
|
artifacts_dir = run_dir / "artifacts"
|
|
if artifacts_dir.exists():
|
|
for file_path in artifacts_dir.iterdir():
|
|
if file_path.is_file():
|
|
file_type = file_path.suffix.lstrip(".")
|
|
response.artifacts.append(
|
|
Artifact(
|
|
name=file_path.name,
|
|
path=str(file_path),
|
|
type=file_type if file_type else "unknown",
|
|
size=file_path.stat().st_size,
|
|
exists=True,
|
|
)
|
|
)
|
|
|
|
equity_path = run_dir / "artifacts" / "equity.csv"
|
|
if equity_path.exists():
|
|
response.artifacts_equity_csv = _load_csv_to_dict(equity_path)
|
|
|
|
metrics_csv_path = run_dir / "artifacts" / "metrics.csv"
|
|
if metrics_csv_path.exists():
|
|
response.artifacts_metrics_csv = _load_csv_to_dict(metrics_csv_path)
|
|
|
|
trades_path = run_dir / "artifacts" / "trades.csv"
|
|
if trades_path.exists():
|
|
response.artifacts_trades_csv = _load_csv_to_dict(trades_path)
|
|
|
|
if response.artifacts_equity_csv:
|
|
filtered_equity = []
|
|
for row in response.artifacts_equity_csv[:1000]:
|
|
filtered_row: Dict[str, Any] = {}
|
|
if "timestamp" in row:
|
|
filtered_row["time"] = row["timestamp"]
|
|
if "equity" in row:
|
|
filtered_row["equity"] = row["equity"]
|
|
if "drawdown" in row:
|
|
filtered_row["drawdown"] = row["drawdown"]
|
|
filtered_equity.append(filtered_row)
|
|
response.equity_curve = filtered_equity
|
|
|
|
if response.artifacts_trades_csv:
|
|
response.trade_log = response.artifacts_trades_csv[:500]
|
|
|
|
if include_analysis:
|
|
analysis = build_run_analysis(run_dir)
|
|
response.run_stage = analysis.get("run_stage")
|
|
response.run_context = analysis.get("run_context")
|
|
response.price_series = analysis.get("price_series")
|
|
response.indicator_series = analysis.get("indicator_series")
|
|
response.trade_markers = analysis.get("trade_markers")
|
|
response.run_logs = analysis.get("run_logs")
|
|
|
|
return response
|
|
|
|
|
|
# ============================================================================
|
|
# API Endpoints
|
|
# ============================================================================
|
|
|
|
@app.get("/runs/{run_id}/code")
|
|
async def get_run_code(run_id: str):
|
|
"""Return strategy source files for a run.
|
|
|
|
Args:
|
|
run_id: Run identifier.
|
|
|
|
Returns:
|
|
Map filename -> source text.
|
|
"""
|
|
run_dir = RUNS_DIR / run_id / "code"
|
|
if not run_dir.exists():
|
|
raise HTTPException(status_code=404, detail=f"Code directory for run {run_id} not found")
|
|
result = {}
|
|
for f in ["signal_engine.py"]:
|
|
p = run_dir / f
|
|
if p.exists():
|
|
result[f] = p.read_text(encoding="utf-8")
|
|
return result
|
|
|
|
|
|
@app.get("/runs/{run_id}", response_model=RunResponse)
|
|
async def get_run_result(run_id: str):
|
|
"""Fetch full details for a historical run by ``run_id``."""
|
|
run_dir = RUNS_DIR / run_id
|
|
|
|
if not run_dir.exists():
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Run {run_id} not found"
|
|
)
|
|
|
|
response = _build_response_from_run_dir(run_dir, elapsed=0.0, include_analysis=True)
|
|
|
|
return response
|
|
|
|
|
|
@app.get("/runs", response_model=List[RunInfo])
|
|
async def list_runs(limit: int = 20):
|
|
"""List recent runs with summary fields."""
|
|
limit = min(max(1, limit), 100)
|
|
runs_dir = RUNS_DIR
|
|
|
|
if not runs_dir.exists():
|
|
return []
|
|
|
|
run_dirs = sorted(
|
|
[d for d in runs_dir.iterdir() if d.is_dir()],
|
|
key=lambda x: x.name,
|
|
reverse=True
|
|
)
|
|
|
|
results = []
|
|
for d in run_dirs[:limit]:
|
|
run_id = d.name
|
|
|
|
# Status from state.json or artifacts
|
|
status_val = "unknown"
|
|
state_file = _load_json_file(d / "state.json")
|
|
if state_file:
|
|
status_val = str(state_file.get("status") or "unknown").lower()
|
|
elif (d / "artifacts" / "equity.csv").exists():
|
|
status_val = "success"
|
|
elif (d / "review_report.json").exists():
|
|
status_val = "success"
|
|
|
|
# Parse created_at from run_id (YYYYMMDD_HHMMSS or run_YYYYMMDD_HHMMSS)
|
|
created_at = "Unknown"
|
|
if run_id.startswith("run_"):
|
|
parts = run_id.split('_')
|
|
if len(parts) >= 3:
|
|
d_str, t_str = parts[1], parts[2]
|
|
if len(d_str) == 8 and len(t_str) == 6:
|
|
created_at = f"{d_str[:4]}-{d_str[4:6]}-{d_str[6:8]} {t_str[:2]}:{t_str[2:4]}:{t_str[4:6]}"
|
|
elif "_" in run_id:
|
|
parts = run_id.split('_')
|
|
if len(parts) >= 2:
|
|
d_str, t_str = parts[0], parts[1]
|
|
if len(d_str) == 8 and len(t_str) == 6:
|
|
created_at = f"{d_str[:4]}-{d_str[4:6]}-{d_str[6:8]} {t_str[:2]}:{t_str[2:4]}:{t_str[4:6]}"
|
|
|
|
if created_at == "Unknown":
|
|
mtime = datetime.fromtimestamp(d.stat().st_mtime)
|
|
created_at = mtime.strftime("%Y-%m-%d %H:%M:%S")
|
|
|
|
prompt = None
|
|
req_file = d / "req.json"
|
|
planner_file = d / "planner_output.json"
|
|
if req_file.exists():
|
|
try:
|
|
req_data = json.loads(req_file.read_text(encoding="utf-8"))
|
|
prompt = req_data.get("prompt")
|
|
except: pass
|
|
|
|
if not prompt and planner_file.exists():
|
|
try:
|
|
planner_data = json.loads(planner_file.read_text(encoding="utf-8"))
|
|
prompt = planner_data.get("user_goal") or planner_data.get("goal")
|
|
except: pass
|
|
|
|
if not prompt:
|
|
prompt_file = d / "user_prompt.txt"
|
|
if prompt_file.exists():
|
|
prompt = prompt_file.read_text(encoding="utf-8").strip()
|
|
|
|
total_return = None
|
|
sharpe = None
|
|
metrics_file = d / "artifacts" / "metrics.csv"
|
|
if metrics_file.exists():
|
|
try:
|
|
import csv
|
|
with open(metrics_file, 'r', encoding='utf-8') as f:
|
|
reader = csv.DictReader(f)
|
|
for row in reader:
|
|
total_return = float(row.get('total_return', 0) or 0)
|
|
sharpe = float(row.get('sharpe', 0) or 0)
|
|
break
|
|
except:
|
|
pass
|
|
|
|
run_context = load_run_context(d)
|
|
results.append(RunInfo(
|
|
run_id=run_id,
|
|
status=status_val,
|
|
created_at=created_at,
|
|
prompt=prompt or "Manual Analysis",
|
|
total_return=total_return,
|
|
sharpe=sharpe,
|
|
codes=run_context.get("codes") or [],
|
|
start_date=run_context.get("start_date"),
|
|
end_date=run_context.get("end_date"),
|
|
))
|
|
|
|
return results
|
|
|
|
|
|
@app.get("/health", response_model=HealthResponse)
|
|
async def health_check():
|
|
"""Liveness probe."""
|
|
return HealthResponse(
|
|
status="healthy",
|
|
service="Vibe-Trading API",
|
|
timestamp=datetime.now().isoformat()
|
|
)
|
|
|
|
|
|
def _terminate_current_process() -> None:
|
|
"""Stop the current API process after the response has been sent."""
|
|
time.sleep(0.25)
|
|
os.kill(os.getpid(), signal.SIGTERM)
|
|
|
|
|
|
@app.post("/system/shutdown", dependencies=[Depends(require_auth)])
|
|
async def shutdown_local_api(background_tasks: BackgroundTasks, request: Request):
|
|
"""Shut down the local API server when requested from loopback clients."""
|
|
client_host = request.client.host if request.client else ""
|
|
if client_host not in {"127.0.0.1", "::1", "localhost"}:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Local access only")
|
|
|
|
background_tasks.add_task(_terminate_current_process)
|
|
return {
|
|
"status": "shutting-down",
|
|
"service": "Vibe-Trading API",
|
|
"timestamp": datetime.now().isoformat(),
|
|
}
|
|
|
|
|
|
@app.get("/skills")
|
|
async def list_skills():
|
|
"""List registered skills (name and description)."""
|
|
from src.agent.skills import SkillsLoader
|
|
|
|
loader = SkillsLoader()
|
|
return [
|
|
{
|
|
"name": s.name,
|
|
"description": s.description,
|
|
}
|
|
for s in loader.skills
|
|
]
|
|
|
|
|
|
@app.get("/api")
|
|
async def api_info():
|
|
"""Service metadata."""
|
|
return {
|
|
"service": "Vibe-Trading API",
|
|
"version": "5.0.0",
|
|
"docs": "/docs",
|
|
"health": "/health",
|
|
}
|
|
|
|
|
|
# ============================================================================
|
|
# Session API
|
|
# ============================================================================
|
|
|
|
_session_service = None
|
|
|
|
|
|
def _get_session_service():
|
|
"""Lazy-init session service when ENABLE_SESSION_RUNTIME=true."""
|
|
global _session_service
|
|
if _session_service is not None:
|
|
return _session_service
|
|
|
|
if os.getenv("ENABLE_SESSION_RUNTIME", "true").lower() != "true":
|
|
return None
|
|
|
|
import asyncio
|
|
from src.session.store import SessionStore
|
|
from src.session.events import EventBus
|
|
from src.session.service import SessionService
|
|
|
|
store = SessionStore(base_dir=SESSIONS_DIR)
|
|
event_bus = EventBus()
|
|
|
|
try:
|
|
loop = asyncio.get_event_loop()
|
|
event_bus.set_loop(loop)
|
|
except RuntimeError:
|
|
pass
|
|
|
|
_session_service = SessionService(
|
|
store=store,
|
|
event_bus=event_bus,
|
|
runs_dir=RUNS_DIR,
|
|
)
|
|
return _session_service
|
|
|
|
|
|
@app.post("/sessions", response_model=SessionResponse, status_code=status.HTTP_201_CREATED, dependencies=[Depends(require_auth)])
|
|
async def create_session(request: CreateSessionRequest):
|
|
"""Create a chat session."""
|
|
svc = _get_session_service()
|
|
if not svc:
|
|
raise HTTPException(status_code=501, detail="Session runtime not enabled")
|
|
session = svc.create_session(title=request.title, config=request.config)
|
|
return SessionResponse(
|
|
session_id=session.session_id,
|
|
title=session.title,
|
|
status=session.status.value,
|
|
created_at=session.created_at,
|
|
updated_at=session.updated_at,
|
|
last_attempt_id=session.last_attempt_id,
|
|
)
|
|
|
|
|
|
@app.get("/sessions", response_model=List[SessionResponse])
|
|
async def list_sessions(limit: int = Query(50, ge=1, le=200)):
|
|
"""List sessions."""
|
|
svc = _get_session_service()
|
|
if not svc:
|
|
raise HTTPException(status_code=501, detail="Session runtime not enabled")
|
|
sessions = svc.list_sessions(limit=limit)
|
|
return [
|
|
SessionResponse(
|
|
session_id=s.session_id,
|
|
title=s.title,
|
|
status=s.status.value,
|
|
created_at=s.created_at,
|
|
updated_at=s.updated_at,
|
|
last_attempt_id=s.last_attempt_id,
|
|
)
|
|
for s in sessions
|
|
]
|
|
|
|
|
|
@app.get("/sessions/{session_id}", response_model=SessionResponse)
|
|
async def get_session(session_id: str):
|
|
"""Get one session by id."""
|
|
svc = _get_session_service()
|
|
if not svc:
|
|
raise HTTPException(status_code=501, detail="Session runtime not enabled")
|
|
session = svc.get_session(session_id)
|
|
if not session:
|
|
raise HTTPException(status_code=404, detail=f"Session {session_id} not found")
|
|
return SessionResponse(
|
|
session_id=session.session_id,
|
|
title=session.title,
|
|
status=session.status.value,
|
|
created_at=session.created_at,
|
|
updated_at=session.updated_at,
|
|
last_attempt_id=session.last_attempt_id,
|
|
)
|
|
|
|
|
|
@app.delete("/sessions/{session_id}", dependencies=[Depends(require_auth)])
|
|
async def delete_session(session_id: str):
|
|
"""Delete a session."""
|
|
svc = _get_session_service()
|
|
if not svc:
|
|
raise HTTPException(status_code=501, detail="Session runtime not enabled")
|
|
deleted = svc.delete_session(session_id)
|
|
if not deleted:
|
|
raise HTTPException(status_code=404, detail=f"Session {session_id} not found")
|
|
return {"status": "deleted", "session_id": session_id}
|
|
|
|
|
|
class UpdateSessionRequest(BaseModel):
|
|
"""Session update fields."""
|
|
title: Optional[str] = None
|
|
|
|
|
|
@app.patch("/sessions/{session_id}", dependencies=[Depends(require_auth)])
|
|
async def update_session(session_id: str, req: UpdateSessionRequest):
|
|
"""Update session fields (e.g. title)."""
|
|
svc = _get_session_service()
|
|
if not svc:
|
|
raise HTTPException(status_code=501, detail="Session runtime not enabled")
|
|
session = svc.store.get_session(session_id)
|
|
if not session:
|
|
raise HTTPException(status_code=404, detail=f"Session {session_id} not found")
|
|
if req.title is not None:
|
|
session.title = req.title
|
|
from datetime import datetime
|
|
session.updated_at = datetime.now().isoformat()
|
|
svc.store.update_session(session)
|
|
return {"status": "updated", "session_id": session_id}
|
|
|
|
|
|
@app.post("/sessions/{session_id}/messages", dependencies=[Depends(require_auth)])
|
|
async def send_message(session_id: str, request: SendMessageRequest):
|
|
"""Send a user message and start the agent loop (natural language strategy)."""
|
|
svc = _get_session_service()
|
|
if not svc:
|
|
raise HTTPException(status_code=501, detail="Session runtime not enabled")
|
|
try:
|
|
result = await svc.send_message(session_id=session_id, content=request.content)
|
|
return result
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=404, detail=str(exc))
|
|
|
|
|
|
@app.post("/sessions/{session_id}/cancel", dependencies=[Depends(require_auth)])
|
|
async def cancel_session(session_id: str):
|
|
"""Cancel the in-flight agent loop for this session."""
|
|
svc = _get_session_service()
|
|
if not svc:
|
|
raise HTTPException(status_code=501, detail="Session runtime not enabled")
|
|
cancelled = svc.cancel_current(session_id)
|
|
if not cancelled:
|
|
return {"status": "no_active_loop"}
|
|
return {"status": "cancelled"}
|
|
|
|
|
|
@app.get("/sessions/{session_id}/messages", response_model=List[MessageResponse])
|
|
async def get_messages(session_id: str, limit: int = Query(100, ge=1, le=1000)):
|
|
"""List messages for a session."""
|
|
svc = _get_session_service()
|
|
if not svc:
|
|
raise HTTPException(status_code=501, detail="Session runtime not enabled")
|
|
messages = svc.get_messages(session_id, limit=limit)
|
|
return [
|
|
MessageResponse(
|
|
message_id=m.message_id,
|
|
session_id=m.session_id,
|
|
role=m.role,
|
|
content=m.content,
|
|
created_at=m.created_at,
|
|
linked_attempt_id=m.linked_attempt_id,
|
|
metadata=m.metadata if m.metadata else None,
|
|
)
|
|
for m in messages
|
|
]
|
|
|
|
|
|
@app.get("/sessions/{session_id}/events")
|
|
async def session_events(
|
|
session_id: str,
|
|
request: Request,
|
|
last_event_id: Optional[str] = Query(None, alias="Last-Event-ID"),
|
|
):
|
|
"""SSE stream for agent events."""
|
|
svc = _get_session_service()
|
|
if not svc:
|
|
raise HTTPException(status_code=501, detail="Session runtime not enabled")
|
|
session = svc.get_session(session_id)
|
|
if not session:
|
|
raise HTTPException(status_code=404, detail=f"Session {session_id} not found")
|
|
|
|
header_id = request.headers.get("Last-Event-ID")
|
|
event_id = header_id or last_event_id
|
|
|
|
async def event_generator():
|
|
async for event in svc.event_bus.subscribe(session_id, last_event_id=event_id):
|
|
if await request.is_disconnected():
|
|
break
|
|
yield event.to_sse()
|
|
|
|
return StreamingResponse(
|
|
event_generator(),
|
|
media_type="text/event-stream",
|
|
headers={
|
|
"Cache-Control": "no-cache",
|
|
"Connection": "keep-alive",
|
|
"X-Accel-Buffering": "no",
|
|
},
|
|
)
|
|
|
|
|
|
# ============================================================================
|
|
# File Upload
|
|
# ============================================================================
|
|
|
|
@app.post("/upload", dependencies=[Depends(require_auth)])
|
|
async def upload_file(file: UploadFile):
|
|
"""Upload a PDF; stored under uploads/ with a UUID name (max 50MB)."""
|
|
if not file.filename or not file.filename.lower().endswith(".pdf"):
|
|
raise HTTPException(status_code=400, detail="Only PDF files are allowed")
|
|
|
|
content = await file.read()
|
|
if len(content) > MAX_UPLOAD_SIZE:
|
|
raise HTTPException(
|
|
status_code=413,
|
|
detail=f"File too large (limit {MAX_UPLOAD_SIZE // (1024 * 1024)} MB)",
|
|
)
|
|
|
|
UPLOADS_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
safe_name = f"{uuid.uuid4().hex}.pdf"
|
|
dest = UPLOADS_DIR / safe_name
|
|
dest.write_bytes(content)
|
|
|
|
return {
|
|
"status": "ok",
|
|
"file_path": str(dest.resolve()),
|
|
"filename": file.filename,
|
|
}
|
|
|
|
|
|
# ============================================================================
|
|
# Swarm API
|
|
# ============================================================================
|
|
|
|
_swarm_runtime = None
|
|
|
|
|
|
def _get_swarm_runtime():
|
|
"""Lazy-init SwarmRuntime singleton."""
|
|
global _swarm_runtime
|
|
if _swarm_runtime is not None:
|
|
return _swarm_runtime
|
|
from src.swarm.store import SwarmStore
|
|
from src.swarm.runtime import SwarmRuntime
|
|
swarm_dir = Path(__file__).resolve().parent / ".swarm" / "runs"
|
|
store = SwarmStore(base_dir=swarm_dir)
|
|
_swarm_runtime = SwarmRuntime(store=store)
|
|
return _swarm_runtime
|
|
|
|
|
|
@app.get("/swarm/presets")
|
|
async def list_swarm_presets():
|
|
"""List Swarm YAML presets."""
|
|
from src.swarm.presets import list_presets
|
|
return list_presets()
|
|
|
|
|
|
@app.post("/swarm/runs", dependencies=[Depends(require_auth)])
|
|
async def create_swarm_run(request: dict):
|
|
"""Start a swarm run: body must include preset_name and user_vars."""
|
|
runtime = _get_swarm_runtime()
|
|
preset_name = request.get("preset_name", "")
|
|
user_vars = request.get("user_vars", {})
|
|
try:
|
|
run = runtime.start_run(preset_name, user_vars)
|
|
return {"id": run.id, "status": run.status.value, "preset_name": run.preset_name}
|
|
except FileNotFoundError as e:
|
|
raise HTTPException(status_code=404, detail=str(e))
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
|
|
@app.get("/swarm/runs")
|
|
async def list_swarm_runs(limit: int = Query(20, ge=1, le=100)):
|
|
"""List swarm runs (newest first)."""
|
|
runtime = _get_swarm_runtime()
|
|
runs = runtime._store.list_runs(limit=limit)
|
|
return [
|
|
{
|
|
"id": r.id,
|
|
"preset_name": r.preset_name,
|
|
"status": r.status.value,
|
|
"created_at": r.created_at,
|
|
"task_count": len(r.tasks),
|
|
"completed_count": sum(1 for t in r.tasks if t.status.value == "completed"),
|
|
}
|
|
for r in runs
|
|
]
|
|
|
|
|
|
@app.get("/swarm/runs/{run_id}")
|
|
async def get_swarm_run(run_id: str):
|
|
"""Swarm run detail including task statuses."""
|
|
from src.swarm.task_store import TaskStore
|
|
|
|
runtime = _get_swarm_runtime()
|
|
run = runtime._store.load_run(run_id)
|
|
if not run:
|
|
raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
|
|
|
|
# Merge real-time task statuses from task_store (updated during execution)
|
|
run_dir = runtime._store.run_dir(run_id)
|
|
tasks_dir = run_dir / "tasks"
|
|
if tasks_dir.exists():
|
|
task_store = TaskStore(run_dir)
|
|
live_tasks = task_store.load_all()
|
|
if live_tasks:
|
|
run.tasks = live_tasks
|
|
|
|
return {
|
|
"id": run.id,
|
|
"preset_name": run.preset_name,
|
|
"status": run.status.value,
|
|
"user_vars": run.user_vars,
|
|
"agents": [a.model_dump() for a in run.agents],
|
|
"tasks": [t.model_dump() for t in run.tasks],
|
|
"created_at": run.created_at,
|
|
"completed_at": run.completed_at,
|
|
"final_report": run.final_report,
|
|
}
|
|
|
|
|
|
@app.get("/swarm/runs/{run_id}/events")
|
|
async def swarm_run_events(run_id: str, request: Request, last_index: int = Query(0, ge=0)):
|
|
"""SSE stream for a swarm run."""
|
|
import asyncio
|
|
runtime = _get_swarm_runtime()
|
|
|
|
async def event_stream():
|
|
idx = last_index
|
|
while True:
|
|
if await request.is_disconnected():
|
|
break
|
|
events = runtime._store.read_events(run_id, after_index=idx)
|
|
for evt in events:
|
|
idx += 1
|
|
yield f"id: {idx}\nevent: {evt.type}\ndata: {json.dumps(evt.model_dump(), ensure_ascii=False)}\n\n"
|
|
run = runtime._store.load_run(run_id)
|
|
if run and run.status.value in ("completed", "failed", "cancelled"):
|
|
yield f"event: done\ndata: {{\"status\": \"{run.status.value}\"}}\n\n"
|
|
break
|
|
await asyncio.sleep(2)
|
|
|
|
return StreamingResponse(event_stream(), media_type="text/event-stream")
|
|
|
|
|
|
@app.post("/swarm/runs/{run_id}/cancel", dependencies=[Depends(require_auth)])
|
|
async def cancel_swarm_run(run_id: str):
|
|
"""Cancel an active swarm run."""
|
|
runtime = _get_swarm_runtime()
|
|
ok = runtime.cancel_run(run_id)
|
|
if not ok:
|
|
raise HTTPException(status_code=404, detail=f"No active run {run_id}")
|
|
return {"status": "cancelled"}
|
|
|
|
|
|
# ============================================================================
|
|
# Main Entry Point
|
|
# ============================================================================
|
|
|
|
def serve_main(argv: list[str] | None = None) -> int:
|
|
"""Start the API server from CLI-style arguments."""
|
|
import argparse
|
|
import subprocess
|
|
import uvicorn
|
|
from fastapi.staticfiles import StaticFiles
|
|
|
|
parser = argparse.ArgumentParser(description="Vibe-Trading Server")
|
|
parser.add_argument("--port", type=int, default=8000, help="Listen port (default 8000)")
|
|
parser.add_argument("--host", default="0.0.0.0", help="Bind address")
|
|
parser.add_argument("--dev", action="store_true", help="Dev mode: spawn Vite on :5173")
|
|
try:
|
|
args = parser.parse_args(argv)
|
|
except SystemExit as exc:
|
|
return int(exc.code) if isinstance(exc.code, int) else 2
|
|
|
|
frontend_dist = Path(__file__).resolve().parent.parent / "frontend" / "dist"
|
|
frontend_root = Path(__file__).resolve().parent.parent / "frontend"
|
|
|
|
vite_proc = None
|
|
if args.dev and frontend_root.exists():
|
|
print("[dev] Starting Vite dev server on :5173 ...")
|
|
vite_proc = subprocess.Popen(
|
|
["npx", "vite", "--host", "0.0.0.0"],
|
|
cwd=str(frontend_root),
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
)
|
|
print(f"[dev] Vite PID={vite_proc.pid}")
|
|
print("[dev] Frontend: http://localhost:5173")
|
|
print(f"[dev] API: http://localhost:{args.port}")
|
|
elif frontend_dist.exists():
|
|
if not any(route.path == "/" for route in app.routes):
|
|
app.mount("/", StaticFiles(directory=str(frontend_dist), html=True), name="frontend")
|
|
print(f"[prod] Frontend served from {frontend_dist}")
|
|
else:
|
|
print(f"[warn] No frontend build found at {frontend_dist}")
|
|
print("[warn] Run: cd frontend && npm run build")
|
|
|
|
print("=" * 50)
|
|
print(" Vibe-Trading Server")
|
|
print(f" http://127.0.0.1:{args.port}")
|
|
print("=" * 50)
|
|
|
|
try:
|
|
uvicorn.run(app, host=args.host, port=args.port, log_level="info")
|
|
finally:
|
|
if vite_proc:
|
|
vite_proc.terminate()
|
|
print("[dev] Vite stopped")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(serve_main())
|
|
|