You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
40 lines
1.6 KiB
Python
40 lines
1.6 KiB
Python
"""FINRA short volume Oracle service methods."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from libs.oracle_client.client import OracleClient
|
|
from libs.oracle_client.models import (
|
|
ShortRatioPoint,
|
|
ShortRatioResponse,
|
|
ShortVolumeEntry,
|
|
ShortVolumeResponse,
|
|
)
|
|
|
|
|
|
class FinraService:
|
|
def __init__(self, client: OracleClient) -> None:
|
|
self._client = client
|
|
|
|
async def get_short_volume(self, symbol: str, days: int = 30) -> ShortVolumeResponse:
|
|
params: dict[str, str | int] = {"days": days}
|
|
data = await self._client.get(f"/api/v1/finra/short-volume/{symbol}", params=params)
|
|
# Real Oracle: {"symbol": ..., "entries": [{date, short_volume, ...}], "total_count": ...}
|
|
entries = [
|
|
ShortVolumeEntry(
|
|
date=e["date"],
|
|
short_volume=int(e["short_volume"]),
|
|
short_exempt_volume=int(e["short_exempt_volume"])
|
|
if e.get("short_exempt_volume") is not None
|
|
else None,
|
|
total_volume=int(e["total_volume"]) if e.get("total_volume") is not None else None,
|
|
)
|
|
for e in data.get("entries", [])
|
|
]
|
|
return ShortVolumeResponse(symbol=data.get("symbol", symbol), data=entries)
|
|
|
|
async def get_short_ratio(self, symbol: str, days: int = 30) -> ShortRatioResponse:
|
|
params: dict[str, str | int] = {"days": days}
|
|
data = await self._client.get(f"/api/v1/finra/short-ratio/{symbol}", params=params)
|
|
points = [ShortRatioPoint.model_validate(p) for p in data.get("data", [])]
|
|
return ShortRatioResponse(symbol=data.get("symbol", symbol), data=points)
|