70 lines
2.1 KiB
Python
70 lines
2.1 KiB
Python
|
"""
|
||
|
|
Drawdown API endpoints for portfolio risk monitoring.
|
||
|
|
"""
|
||
|
|
from fastapi import APIRouter, Depends
|
||
|
|
from sqlalchemy.orm import Session
|
||
|
|
|
||
|
|
from app.core.database import get_db
|
||
|
|
from app.api.deps import CurrentUser
|
||
|
|
from app.api.snapshot import _get_portfolio
|
||
|
|
from app.schemas.drawdown import (
|
||
|
|
DrawdownResponse,
|
||
|
|
DrawdownHistoryResponse,
|
||
|
|
DrawdownSettingsUpdate,
|
||
|
|
)
|
||
|
|
from app.services.drawdown import (
|
||
|
|
calculate_drawdown,
|
||
|
|
calculate_rolling_drawdown,
|
||
|
|
get_alert_threshold,
|
||
|
|
set_alert_threshold,
|
||
|
|
)
|
||
|
|
|
||
|
|
router = APIRouter(prefix="/api/drawdown", tags=["drawdown"])
|
||
|
|
|
||
|
|
|
||
|
|
@router.get("/{portfolio_id}", response_model=DrawdownResponse)
|
||
|
|
async def get_drawdown(
|
||
|
|
portfolio_id: int,
|
||
|
|
current_user: CurrentUser,
|
||
|
|
db: Session = Depends(get_db),
|
||
|
|
):
|
||
|
|
"""Get current drawdown metrics for a portfolio."""
|
||
|
|
_get_portfolio(db, portfolio_id, current_user.id)
|
||
|
|
data = calculate_drawdown(db, portfolio_id)
|
||
|
|
return DrawdownResponse(portfolio_id=portfolio_id, **data)
|
||
|
|
|
||
|
|
|
||
|
|
@router.get("/{portfolio_id}/history", response_model=DrawdownHistoryResponse)
|
||
|
|
async def get_drawdown_history(
|
||
|
|
portfolio_id: int,
|
||
|
|
current_user: CurrentUser,
|
||
|
|
db: Session = Depends(get_db),
|
||
|
|
):
|
||
|
|
"""Get rolling drawdown time series for a portfolio."""
|
||
|
|
_get_portfolio(db, portfolio_id, current_user.id)
|
||
|
|
rolling = calculate_rolling_drawdown(db, portfolio_id)
|
||
|
|
summary = calculate_drawdown(db, portfolio_id)
|
||
|
|
|
||
|
|
return DrawdownHistoryResponse(
|
||
|
|
portfolio_id=portfolio_id,
|
||
|
|
data=rolling,
|
||
|
|
max_drawdown_pct=summary["max_drawdown_pct"],
|
||
|
|
current_drawdown_pct=summary["current_drawdown_pct"],
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
@router.put("/settings/{portfolio_id}")
|
||
|
|
async def update_drawdown_settings(
|
||
|
|
portfolio_id: int,
|
||
|
|
body: DrawdownSettingsUpdate,
|
||
|
|
current_user: CurrentUser,
|
||
|
|
db: Session = Depends(get_db),
|
||
|
|
):
|
||
|
|
"""Update drawdown alert threshold for a portfolio."""
|
||
|
|
_get_portfolio(db, portfolio_id, current_user.id)
|
||
|
|
set_alert_threshold(portfolio_id, body.alert_threshold_pct)
|
||
|
|
return {
|
||
|
|
"portfolio_id": portfolio_id,
|
||
|
|
"alert_threshold_pct": float(body.alert_threshold_pct),
|
||
|
|
}
|