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
128 lines
3.8 KiB
Python
128 lines
3.8 KiB
Python
"""Routes for calendar events, screening, ownership, and estimates."""
|
|
|
|
from fastapi import APIRouter, Path, Query
|
|
|
|
from models import ApiResponse
|
|
from route_utils import safe, validate_symbol
|
|
import calendar_service
|
|
|
|
router = APIRouter(prefix="/api/v1")
|
|
|
|
DATE_PATTERN = r"^\d{4}-\d{2}-\d{2}$"
|
|
|
|
|
|
# --- Calendar Events ---
|
|
|
|
|
|
@router.get("/calendar/earnings", response_model=ApiResponse)
|
|
@safe
|
|
async def earnings_calendar(
|
|
start_date: str | None = Query(
|
|
default=None, pattern=DATE_PATTERN, description="YYYY-MM-DD"
|
|
),
|
|
end_date: str | None = Query(
|
|
default=None, pattern=DATE_PATTERN, description="YYYY-MM-DD"
|
|
),
|
|
):
|
|
"""Get upcoming earnings announcements."""
|
|
data = await calendar_service.get_earnings_calendar(start_date, end_date)
|
|
return ApiResponse(data=data)
|
|
|
|
|
|
@router.get("/calendar/dividends", response_model=ApiResponse)
|
|
@safe
|
|
async def dividend_calendar(
|
|
start_date: str | None = Query(
|
|
default=None, pattern=DATE_PATTERN, description="YYYY-MM-DD"
|
|
),
|
|
end_date: str | None = Query(
|
|
default=None, pattern=DATE_PATTERN, description="YYYY-MM-DD"
|
|
),
|
|
):
|
|
"""Get upcoming dividend dates."""
|
|
data = await calendar_service.get_dividend_calendar(start_date, end_date)
|
|
return ApiResponse(data=data)
|
|
|
|
|
|
@router.get("/calendar/ipo", response_model=ApiResponse)
|
|
@safe
|
|
async def ipo_calendar(
|
|
start_date: str | None = Query(
|
|
default=None, pattern=DATE_PATTERN, description="YYYY-MM-DD"
|
|
),
|
|
end_date: str | None = Query(
|
|
default=None, pattern=DATE_PATTERN, description="YYYY-MM-DD"
|
|
),
|
|
):
|
|
"""Get upcoming IPOs."""
|
|
data = await calendar_service.get_ipo_calendar(start_date, end_date)
|
|
return ApiResponse(data=data)
|
|
|
|
|
|
@router.get("/calendar/splits", response_model=ApiResponse)
|
|
@safe
|
|
async def splits_calendar(
|
|
start_date: str | None = Query(
|
|
default=None, pattern=DATE_PATTERN, description="YYYY-MM-DD"
|
|
),
|
|
end_date: str | None = Query(
|
|
default=None, pattern=DATE_PATTERN, description="YYYY-MM-DD"
|
|
),
|
|
):
|
|
"""Get upcoming stock splits."""
|
|
data = await calendar_service.get_splits_calendar(start_date, end_date)
|
|
return ApiResponse(data=data)
|
|
|
|
|
|
# --- Analyst Estimates ---
|
|
|
|
|
|
@router.get("/stock/{symbol}/estimates", response_model=ApiResponse)
|
|
@safe
|
|
async def stock_estimates(symbol: str = Path(..., min_length=1, max_length=20)):
|
|
"""Get analyst consensus estimates."""
|
|
symbol = validate_symbol(symbol)
|
|
data = await calendar_service.get_analyst_estimates(symbol)
|
|
return ApiResponse(data=data)
|
|
|
|
|
|
@router.get("/stock/{symbol}/share-statistics", response_model=ApiResponse)
|
|
@safe
|
|
async def stock_share_stats(symbol: str = Path(..., min_length=1, max_length=20)):
|
|
"""Get share statistics: float, outstanding, short interest."""
|
|
symbol = validate_symbol(symbol)
|
|
data = await calendar_service.get_share_statistics(symbol)
|
|
return ApiResponse(data=data)
|
|
|
|
|
|
# --- Ownership (SEC, free) ---
|
|
|
|
|
|
@router.get("/stock/{symbol}/sec-insider", response_model=ApiResponse)
|
|
@safe
|
|
async def stock_sec_insider(symbol: str = Path(..., min_length=1, max_length=20)):
|
|
"""Get insider trading data from SEC (Form 4)."""
|
|
symbol = validate_symbol(symbol)
|
|
data = await calendar_service.get_insider_trading(symbol)
|
|
return ApiResponse(data=data)
|
|
|
|
|
|
@router.get("/stock/{symbol}/institutional", response_model=ApiResponse)
|
|
@safe
|
|
async def stock_institutional(symbol: str = Path(..., min_length=1, max_length=20)):
|
|
"""Get institutional holders from SEC 13F filings."""
|
|
symbol = validate_symbol(symbol)
|
|
data = await calendar_service.get_institutional_holders(symbol)
|
|
return ApiResponse(data=data)
|
|
|
|
|
|
# --- Screener ---
|
|
|
|
|
|
@router.get("/screener", response_model=ApiResponse)
|
|
@safe
|
|
async def stock_screener():
|
|
"""Screen stocks using available filters."""
|
|
data = await calendar_service.screen_stocks()
|
|
return ApiResponse(data=data)
|