refactor: address python review findings
- Move FRED credential registration to FastAPI lifespan (was fragile import-order-dependent side-effect) - Add noqa E402 annotations for imports after curl_cffi patch - Fix all return type hints: bare dict -> dict[str, Any] - Move yfinance import to module level (was inline in functions) - Fix datetime.now() -> datetime.now(tz=timezone.utc) in openbb_service - Add try/except error handling to Group B service functions - Fix dict mutation in relative_rotation (immutable pattern) - Extract _classify_rrg_quadrant helper function - Fix type builtin shadow in routes_economy (type -> gdp_type) - Fix falsy int guard (if year: -> if year is not None:) - Remove user input echo from error messages
This commit is contained in:
@@ -411,13 +411,12 @@ async def get_relative_rotation(
|
||||
Returns RS-Ratio and RS-Momentum for each symbol, indicating
|
||||
which RRG quadrant they occupy (Leading/Weakening/Lagging/Improving).
|
||||
"""
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from datetime import datetime, timedelta, timezone as tz
|
||||
|
||||
start = (datetime.now(tz=timezone.utc) - timedelta(days=days)).strftime("%Y-%m-%d")
|
||||
start = (datetime.now(tz=tz.utc) - timedelta(days=days)).strftime("%Y-%m-%d")
|
||||
all_symbols = ",".join(symbols + [benchmark])
|
||||
|
||||
try:
|
||||
# Fetch multi-symbol historical data in one call
|
||||
hist = await asyncio.to_thread(
|
||||
obb.equity.price.historical,
|
||||
all_symbols,
|
||||
@@ -434,26 +433,17 @@ async def get_relative_rotation(
|
||||
study=study,
|
||||
)
|
||||
items = to_list(result)
|
||||
# Return the latest data point per symbol
|
||||
latest_by_symbol: dict[str, dict] = {}
|
||||
|
||||
latest_by_symbol: dict[str, dict[str, Any]] = {}
|
||||
for item in items:
|
||||
sym = item.get("symbol")
|
||||
if sym and sym != benchmark:
|
||||
latest_by_symbol[sym] = item
|
||||
|
||||
entries = list(latest_by_symbol.values())
|
||||
for entry in entries:
|
||||
rs_ratio = entry.get("rs_ratio")
|
||||
rs_momentum = entry.get("rs_momentum")
|
||||
if rs_ratio is not None and rs_momentum is not None:
|
||||
if rs_ratio > 100 and rs_momentum > 100:
|
||||
entry["quadrant"] = "Leading"
|
||||
elif rs_ratio > 100 and rs_momentum <= 100:
|
||||
entry["quadrant"] = "Weakening"
|
||||
elif rs_ratio <= 100 and rs_momentum <= 100:
|
||||
entry["quadrant"] = "Lagging"
|
||||
else:
|
||||
entry["quadrant"] = "Improving"
|
||||
entries = [
|
||||
{**item, "quadrant": _classify_rrg_quadrant(item)}
|
||||
for item in latest_by_symbol.values()
|
||||
]
|
||||
|
||||
return {
|
||||
"symbols": symbols,
|
||||
@@ -466,6 +456,21 @@ async def get_relative_rotation(
|
||||
return {"symbols": symbols, "error": "Failed to compute relative rotation"}
|
||||
|
||||
|
||||
def _classify_rrg_quadrant(item: dict[str, Any]) -> str | None:
|
||||
"""Classify RRG quadrant from RS-Ratio and RS-Momentum."""
|
||||
rs_ratio = item.get("rs_ratio")
|
||||
rs_momentum = item.get("rs_momentum")
|
||||
if rs_ratio is None or rs_momentum is None:
|
||||
return None
|
||||
if rs_ratio > 100 and rs_momentum > 100:
|
||||
return "Leading"
|
||||
if rs_ratio > 100:
|
||||
return "Weakening"
|
||||
if rs_momentum <= 100:
|
||||
return "Lagging"
|
||||
return "Improving"
|
||||
|
||||
|
||||
async def get_cones(symbol: str, days: int = 365) -> dict[str, Any]:
|
||||
"""Volatility Cones -- realized volatility quantiles for options analysis."""
|
||||
hist = await fetch_historical(symbol, days)
|
||||
|
||||
Reference in New Issue
Block a user