003c1d6ffc
- Extract shared route_utils.py (validate_symbol, safe decorator)
removing duplication from 6 route files
- Extract shared obb_utils.py (to_list, extract_single, safe_last)
removing duplication from calendar_service and market_service
- Fix _to_list dict mutation during iteration (use comprehension)
- Fix double vars() call and live __dict__ mutation risk
- Fix route ordering: /etf/search and /crypto/search now registered
before /{symbol} path params to prevent shadowing
- Add date format validation (YYYY-MM-DD pattern) on calendar routes
- Use timezone-aware datetime.now(tz=timezone.utc) in all services
- Add explicit type annotation for asyncio.gather results
125 lines
4.8 KiB
Python
125 lines
4.8 KiB
Python
"""Quantitative analysis: risk metrics, performance, CAPM, normality tests."""
|
|
|
|
import asyncio
|
|
import logging
|
|
from datetime import datetime, timezone, timedelta
|
|
from typing import Any
|
|
|
|
from openbb import obb
|
|
|
|
from obb_utils import extract_single, safe_last
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
PROVIDER = "yfinance"
|
|
|
|
# Need 252+ trading days for default window; 730 calendar days is safe
|
|
PERF_DAYS = 730
|
|
TARGET = "close"
|
|
|
|
|
|
async def get_performance_metrics(symbol: str, days: int = 365) -> dict[str, Any]:
|
|
"""Calculate Sharpe ratio, summary stats, and volatility for a symbol."""
|
|
# Need at least 252 trading days for Sharpe window
|
|
fetch_days = max(days, PERF_DAYS)
|
|
start = (datetime.now(tz=timezone.utc) - timedelta(days=fetch_days)).strftime("%Y-%m-%d")
|
|
|
|
try:
|
|
hist = await asyncio.to_thread(
|
|
obb.equity.price.historical, symbol, start_date=start, provider=PROVIDER
|
|
)
|
|
if not hist or not hist.results:
|
|
return {"symbol": symbol, "error": "No historical data"}
|
|
|
|
results: tuple[Any, ...] = await asyncio.gather(
|
|
asyncio.to_thread(
|
|
obb.quantitative.performance.sharpe_ratio,
|
|
data=hist.results, target=TARGET,
|
|
),
|
|
asyncio.to_thread(
|
|
obb.quantitative.summary, data=hist.results, target=TARGET
|
|
),
|
|
asyncio.to_thread(
|
|
obb.quantitative.stats.stdev, data=hist.results, target=TARGET
|
|
),
|
|
return_exceptions=True,
|
|
)
|
|
sharpe_result, summary_result, stdev_result = results
|
|
|
|
sharpe = safe_last(sharpe_result) if not isinstance(sharpe_result, BaseException) else None
|
|
summary = extract_single(summary_result) if not isinstance(summary_result, BaseException) else {}
|
|
stdev = safe_last(stdev_result) if not isinstance(stdev_result, BaseException) else None
|
|
|
|
return {
|
|
"symbol": symbol,
|
|
"period_days": days,
|
|
"sharpe_ratio": sharpe,
|
|
"summary": summary,
|
|
"stdev": stdev,
|
|
}
|
|
except Exception:
|
|
logger.warning("Performance metrics failed for %s", symbol, exc_info=True)
|
|
return {"symbol": symbol, "error": "Failed to compute performance metrics"}
|
|
|
|
|
|
async def get_capm(symbol: str) -> dict[str, Any]:
|
|
"""Calculate CAPM metrics: beta, alpha, systematic/idiosyncratic risk."""
|
|
start = (datetime.now(tz=timezone.utc) - timedelta(days=PERF_DAYS)).strftime("%Y-%m-%d")
|
|
|
|
try:
|
|
hist = await asyncio.to_thread(
|
|
obb.equity.price.historical, symbol, start_date=start, provider=PROVIDER
|
|
)
|
|
if not hist or not hist.results:
|
|
return {"symbol": symbol, "error": "No historical data"}
|
|
|
|
capm = await asyncio.to_thread(
|
|
obb.quantitative.capm, data=hist.results, target=TARGET
|
|
)
|
|
return {"symbol": symbol, **extract_single(capm)}
|
|
except Exception:
|
|
logger.warning("CAPM failed for %s", symbol, exc_info=True)
|
|
return {"symbol": symbol, "error": "Failed to compute CAPM"}
|
|
|
|
|
|
async def get_normality_test(symbol: str, days: int = 365) -> dict[str, Any]:
|
|
"""Run normality tests (Jarque-Bera, Shapiro-Wilk, etc.) on returns."""
|
|
fetch_days = max(days, PERF_DAYS)
|
|
start = (datetime.now(tz=timezone.utc) - timedelta(days=fetch_days)).strftime("%Y-%m-%d")
|
|
|
|
try:
|
|
hist = await asyncio.to_thread(
|
|
obb.equity.price.historical, symbol, start_date=start, provider=PROVIDER
|
|
)
|
|
if not hist or not hist.results:
|
|
return {"symbol": symbol, "error": "No historical data"}
|
|
|
|
norm = await asyncio.to_thread(
|
|
obb.quantitative.normality, data=hist.results, target=TARGET
|
|
)
|
|
return {"symbol": symbol, **extract_single(norm)}
|
|
except Exception:
|
|
logger.warning("Normality test failed for %s", symbol, exc_info=True)
|
|
return {"symbol": symbol, "error": "Failed to compute normality tests"}
|
|
|
|
|
|
async def get_unitroot_test(symbol: str, days: int = 365) -> dict[str, Any]:
|
|
"""Run unit root tests (ADF, KPSS) for stationarity."""
|
|
fetch_days = max(days, PERF_DAYS)
|
|
start = (datetime.now(tz=timezone.utc) - timedelta(days=fetch_days)).strftime("%Y-%m-%d")
|
|
|
|
try:
|
|
hist = await asyncio.to_thread(
|
|
obb.equity.price.historical, symbol, start_date=start, provider=PROVIDER
|
|
)
|
|
if not hist or not hist.results:
|
|
return {"symbol": symbol, "error": "No historical data"}
|
|
|
|
ur = await asyncio.to_thread(
|
|
obb.quantitative.unitroot_test, data=hist.results, target=TARGET
|
|
)
|
|
return {"symbol": symbol, **extract_single(ur)}
|
|
except Exception:
|
|
logger.warning("Unit root test failed for %s", symbol, exc_info=True)
|
|
return {"symbol": symbol, "error": "Failed to compute unit root test"}
|