From bf6932ca2f2b02b85c49a536b84a5fc3de7174ce Mon Sep 17 00:00:00 2001 From: I Luk Kim Date: Thu, 12 Mar 2026 08:42:10 -0700 Subject: [PATCH] Add Alpaca-specific DB table and Redis caching for Alpaca/FINRA endpoints Separate Alpaca price data into dedicated AlpacaPriceData table to avoid UniqueConstraint('ticker', 'date') conflicts with Yahoo Finance PriceData. Add Redis caching (build_cache_key/get_cached_response/set_cached_response) to 3 Alpaca endpoints and 2 FINRA query endpoints with appropriate TTLs. Co-Authored-By: Claude Sonnet 4.6 --- app/api/v1/api.py | 6 +- app/api/v1/endpoints/alpaca.py | 301 +++++++++++++++++++++ app/api/v1/endpoints/finra.py | 174 ++++++++++++ app/core/config.py | 5 + app/main.py | 42 ++- app/models/__init__.py | 4 + app/models/alpaca_price.py | 36 +++ app/models/finra_short_volume.py | 32 +++ app/schemas/financial.py | 21 ++ app/schemas/finra.py | 63 +++++ app/services/alpaca_client.py | 233 ++++++++++++++++ app/services/alpaca_price_service.py | 131 +++++++++ app/services/finra_short_volume_service.py | 292 ++++++++++++++++++++ app/services/price_data_service.py | 3 +- docker-compose.yml | 2 + 15 files changed, 1340 insertions(+), 5 deletions(-) create mode 100644 app/api/v1/endpoints/alpaca.py create mode 100644 app/api/v1/endpoints/finra.py create mode 100644 app/models/alpaca_price.py create mode 100644 app/models/finra_short_volume.py create mode 100644 app/schemas/finra.py create mode 100644 app/services/alpaca_client.py create mode 100644 app/services/alpaca_price_service.py create mode 100644 app/services/finra_short_volume_service.py diff --git a/app/api/v1/api.py b/app/api/v1/api.py index 6147364..f1a64b3 100644 --- a/app/api/v1/api.py +++ b/app/api/v1/api.py @@ -3,7 +3,7 @@ API v1 router """ from fastapi import APIRouter -from app.api.v1.endpoints import financial, price, catalog, health, migration, database, error_logs, request_logs, news, etf, stocks, fred, filings +from app.api.v1.endpoints import financial, price, catalog, health, migration, database, error_logs, request_logs, news, etf, stocks, fred, filings, alpaca, finra api_router = APIRouter() @@ -21,4 +21,6 @@ api_router.include_router(migration.router, prefix="/admin", tags=["admin"]) # Removed docs.router - documentation now served at root path api_router.include_router(database.router, prefix="/database", tags=["database"]) api_router.include_router(error_logs.router, prefix="/admin/errors", tags=["error-logs"]) -api_router.include_router(request_logs.router, prefix="/admin/requests", tags=["request-logs"]) \ No newline at end of file +api_router.include_router(request_logs.router, prefix="/admin/requests", tags=["request-logs"]) +api_router.include_router(alpaca.router, prefix="/alpaca", tags=["alpaca"]) +api_router.include_router(finra.router, prefix="/finra", tags=["finra"]) \ No newline at end of file diff --git a/app/api/v1/endpoints/alpaca.py b/app/api/v1/endpoints/alpaca.py new file mode 100644 index 0000000..326a682 --- /dev/null +++ b/app/api/v1/endpoints/alpaca.py @@ -0,0 +1,301 @@ +""" +Alpaca Market Data endpoints — standalone price data via Alpaca API +""" + +from datetime import date, datetime, timezone +from typing import Optional + +from fastapi import APIRouter, Depends, HTTPException, Query +from fastapi.responses import Response +from starlette.responses import JSONResponse +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy import select, and_ + +from app.core.config import settings +from app.core.database import get_db +from app.models.alpaca_price import AlpacaPriceData +from app.schemas.financial import ( + PriceDataResponse, + AlpacaPriceDataPoint, + ErrorType, +) +from app.services.alpaca_client import AlpacaClient +from app.services.alpaca_price_service import AlpacaPriceService +from app.utils.cache import build_cache_key, get_cached_response, set_cached_response + +router = APIRouter() + +INTRADAY_CACHE_TTL = 300 # 5 minutes for intraday data + + +def _require_alpaca() -> AlpacaPriceService: + svc = AlpacaPriceService() + if not svc.is_available(): + raise HTTPException( + status_code=503, + detail="Alpaca API keys not configured. Set ALPACA_API_KEY and ALPACA_SECRET_KEY.", + ) + return svc + + +# ------------------------------------------------------------------ +# Status (no caching — always real-time) +# ------------------------------------------------------------------ + +@router.get( + "/status", + summary="Alpaca connection status", + description="Check Alpaca API key validity and connection health.", +) +async def alpaca_status(): + client = AlpacaClient() + if not client.is_configured(): + return { + "configured": False, + "message": "ALPACA_API_KEY / ALPACA_SECRET_KEY not set", + } + status = await client.check_connection() + await client.close() + return {"configured": True, **status} + + +# ------------------------------------------------------------------ +# Raw bars (no DB) — with Redis caching +# ------------------------------------------------------------------ + +@router.get( + "/bars/{ticker}", + summary="Get Alpaca bars (raw, no DB)", + description="Fetch historical bars directly from Alpaca without storing in DB.", +) +async def get_alpaca_bars( + ticker: str, + response: Response, + interval: str = Query("1d", description="Interval: 1m, 5m, 15m, 1h, 1d, 1w, 1mo"), + start_date: Optional[date] = Query(None, description="Start date (YYYY-MM-DD)"), + end_date: Optional[date] = Query(None, description="End date (YYYY-MM-DD)"), + limit: int = Query(1000, ge=1, le=10000, description="Max bars to return"), + force_refresh: bool = Query(False, description="Bypass cache"), +): + svc = _require_alpaca() + + # Check cache + cache_key = build_cache_key("alpaca:bars", ticker.upper(), interval, start_date, end_date, limit) + if not force_refresh: + cached = await get_cached_response(cache_key) + if cached: + cached_body, etag = cached + response.headers["X-Cache"] = "HIT" + response.headers["Cache-Control"] = f"public, max-age={settings.CACHE_TTL}" + response.headers["ETag"] = etag + response.headers["X-Data-Source"] = "redis-cache" + return cached_body + + start_dt = datetime.combine(start_date, datetime.min.time()).replace(tzinfo=timezone.utc) if start_date else None + end_dt = datetime.combine(end_date, datetime.min.time()).replace(tzinfo=timezone.utc) if end_date else None + + try: + bars = await svc.fetch_bars_raw( + ticker=ticker.upper(), + interval=interval, + start_date=start_dt, + end_date=end_dt, + ) + bars = bars[:limit] + body_dict = { + "ticker": ticker.upper(), + "interval": interval, + "count": len(bars), + "bars": bars, + } + + etag = await set_cached_response(cache_key, body_dict, ttl_seconds=settings.CACHE_TTL) + response.headers["X-Cache"] = "MISS" + response.headers["Cache-Control"] = f"public, max-age={settings.CACHE_TTL}" + response.headers["ETag"] = etag + response.headers["X-Data-Source"] = "alpaca-api" + return body_dict + except Exception as e: + raise HTTPException(status_code=502, detail=f"Alpaca API error: {e}") + finally: + await svc.client.close() + + +# ------------------------------------------------------------------ +# Price data (DB storage) — with Redis caching +# ------------------------------------------------------------------ + +@router.get( + "/data/{ticker}", + response_model=PriceDataResponse, + summary="Get price data via Alpaca (with DB storage)", + description=""" + Fetch OHLCV price data from Alpaca, store in AlpacaPriceData table, and return + in PriceDataResponse format. Includes vwap and trade_count in metadata. + + - Requires `ALPACA_API_KEY` / `ALPACA_SECRET_KEY` + - Uses `data_source = "ALPACA"` to distinguish from Yahoo data + - Supports: 1m, 5m, 15m, 1h, 1d, 1w, 1mo intervals + """, +) +async def get_alpaca_price_data( + ticker: str, + response: Response, + interval: str = Query("1d", description="Interval: 1m, 5m, 15m, 1h, 1d, 1w, 1mo"), + start_date: date = Query(..., description="Start date (YYYY-MM-DD)"), + end_date: date = Query(..., description="End date (YYYY-MM-DD)"), + force_refresh: bool = Query(False, description="Re-fetch even if data exists in DB"), + db: AsyncSession = Depends(get_db), +): + svc = _require_alpaca() + + # Check cache + cache_key = build_cache_key("alpaca:data", ticker.upper(), interval, start_date, end_date) + if not force_refresh: + cached = await get_cached_response(cache_key) + if cached: + cached_body, etag = cached + response.headers["X-Cache"] = "HIT" + response.headers["Cache-Control"] = f"public, max-age={settings.CACHE_TTL}" + response.headers["ETag"] = etag + response.headers["X-Data-Source"] = "redis-cache" + return cached_body + + start_dt = datetime.combine(start_date, datetime.min.time()).replace(tzinfo=timezone.utc) + end_dt = datetime.combine(end_date, datetime.min.time()).replace(tzinfo=timezone.utc) + + if start_dt >= end_dt: + raise HTTPException(status_code=400, detail="start_date must be before end_date") + + try: + count = await svc.fetch_and_store_bars( + db, ticker, start_dt, end_dt, interval + ) + + # Read back from AlpacaPriceData table + result = await db.execute( + select(AlpacaPriceData) + .where( + and_( + AlpacaPriceData.ticker == ticker.upper(), + AlpacaPriceData.date >= start_dt, + AlpacaPriceData.date <= end_dt, + ) + ) + .order_by(AlpacaPriceData.date) + ) + rows = result.scalars().all() + + alpaca_points = [AlpacaPriceDataPoint.model_validate(r) for r in rows] + + # Build PriceDataResponse-compatible data with vwap/trade_count in metadata + from app.schemas.financial import PriceDataPoint + price_points = [ + PriceDataPoint( + date=p.date, + open=p.open, + high=p.high, + low=p.low, + close=p.close, + volume=p.volume, + adjusted_close=p.vwap, # Map vwap → adjusted_close for compatibility + data_source=p.data_source, + ) + for p in alpaca_points + ] + + body = PriceDataResponse( + ticker=ticker.upper(), + interval=interval, + data=price_points, + metadata={ + "source": "ALPACA", + "data_points": len(price_points), + "new_bars_inserted": count, + "date_range": { + "start": start_date.isoformat(), + "end": end_date.isoformat(), + }, + "alpaca_fields": [ + {"date": p.date.isoformat(), "vwap": p.vwap, "trade_count": p.trade_count} + for p in alpaca_points + ], + }, + ) + + body_dict = body.model_dump() + etag = await set_cached_response(cache_key, body_dict, ttl_seconds=settings.CACHE_TTL) + response.headers["X-Cache"] = "MISS" + response.headers["Cache-Control"] = f"public, max-age={settings.CACHE_TTL}" + response.headers["ETag"] = etag + response.headers["X-Data-Source"] = "alpaca-api" + return body + except HTTPException: + raise + except Exception as e: + raise HTTPException(status_code=502, detail=f"Alpaca error: {e}") + finally: + await svc.client.close() + + +# ------------------------------------------------------------------ +# Intraday (raw — no DB) — with Redis caching (short TTL) +# ------------------------------------------------------------------ + +@router.get( + "/intraday/{ticker}", + summary="Get intraday candles from Alpaca", + description="Fetch intraday bars (1m, 5m, 15m, 1h) directly from Alpaca. Not stored in DB.", +) +async def get_alpaca_intraday( + ticker: str, + response: Response, + interval: str = Query("1m", description="Interval: 1m, 5m, 15m, 1h"), + start_date: Optional[date] = Query(None, description="Start date"), + end_date: Optional[date] = Query(None, description="End date"), + limit: int = Query(1000, ge=1, le=10000, description="Max candles"), + force_refresh: bool = Query(False, description="Bypass cache"), +): + svc = _require_alpaca() + + # Check cache (short TTL for intraday) + cache_key = build_cache_key("alpaca:intraday", ticker.upper(), interval, start_date, end_date) + if not force_refresh: + cached = await get_cached_response(cache_key) + if cached: + cached_body, etag = cached + response.headers["X-Cache"] = "HIT" + response.headers["Cache-Control"] = f"public, max-age={INTRADAY_CACHE_TTL}" + response.headers["ETag"] = etag + response.headers["X-Data-Source"] = "redis-cache" + return cached_body + + start_dt = datetime.combine(start_date, datetime.min.time()).replace(tzinfo=timezone.utc) if start_date else None + end_dt = datetime.combine(end_date, datetime.min.time()).replace(tzinfo=timezone.utc) if end_date else None + + try: + bars = await svc.fetch_bars_raw( + ticker=ticker.upper(), + interval=interval, + start_date=start_dt, + end_date=end_dt, + ) + bars = bars[:limit] + body_dict = { + "ticker": ticker.upper(), + "interval": interval, + "source": "ALPACA", + "count": len(bars), + "candles": bars, + } + + etag = await set_cached_response(cache_key, body_dict, ttl_seconds=INTRADAY_CACHE_TTL) + response.headers["X-Cache"] = "MISS" + response.headers["Cache-Control"] = f"public, max-age={INTRADAY_CACHE_TTL}" + response.headers["ETag"] = etag + response.headers["X-Data-Source"] = "alpaca-api" + return body_dict + except Exception as e: + raise HTTPException(status_code=502, detail=f"Alpaca API error: {e}") + finally: + await svc.client.close() diff --git a/app/api/v1/endpoints/finra.py b/app/api/v1/endpoints/finra.py new file mode 100644 index 0000000..5c90c19 --- /dev/null +++ b/app/api/v1/endpoints/finra.py @@ -0,0 +1,174 @@ +""" +FINRA Short Sale Volume endpoints +""" + +from datetime import date, datetime, timedelta, timezone +from typing import Optional + +from fastapi import APIRouter, Depends, HTTPException, Query +from fastapi.responses import Response +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.config import settings +from app.core.database import get_db +from app.schemas.finra import ( + ShortVolumeEntry, + ShortVolumeResponse, + ShortRatioHistoryResponse, + ShortRatioPoint, + IngestResponse, +) +from app.services.finra_short_volume_service import FinraShortVolumeService +from app.utils.cache import build_cache_key, get_cached_response, set_cached_response + +router = APIRouter() + + +@router.get( + "/short-volume/{symbol}", + response_model=ShortVolumeResponse, + summary="Get short volume data for a symbol", + description="Query FINRA RegSHO short sale volume. Auto-ingests if data is missing.", +) +async def get_short_volume( + symbol: str, + response: Response, + days: int = Query(30, ge=1, le=365, description="Number of days to look back"), + limit: int = Query(100, ge=1, le=1000, description="Max entries to return"), + force_refresh: bool = Query(False, description="Bypass cache"), + db: AsyncSession = Depends(get_db), +): + # Check cache + cache_key = build_cache_key("finra:short-volume", symbol.upper(), days, limit) + if not force_refresh: + cached = await get_cached_response(cache_key) + if cached: + cached_body, etag = cached + response.headers["X-Cache"] = "HIT" + response.headers["Cache-Control"] = f"public, max-age={settings.CACHE_TTL}" + response.headers["ETag"] = etag + response.headers["X-Data-Source"] = "redis-cache" + return cached_body + + svc = FinraShortVolumeService() + end = datetime.now(timezone.utc).date() + start = end - timedelta(days=days) + + rows, total = await svc.get_short_volume( + db, symbol=symbol, start_date=start, end_date=end, limit=limit + ) + + entries = [ShortVolumeEntry.from_orm_obj(r) for r in rows] + + body = ShortVolumeResponse( + symbol=symbol.upper(), + entries=entries, + total_count=total, + metadata={ + "days_requested": days, + "start_date": start.isoformat(), + "end_date": end.isoformat(), + }, + ) + + body_dict = body.model_dump() + etag = await set_cached_response(cache_key, body_dict, ttl_seconds=settings.CACHE_TTL) + response.headers["X-Cache"] = "MISS" + response.headers["Cache-Control"] = f"public, max-age={settings.CACHE_TTL}" + response.headers["ETag"] = etag + response.headers["X-Data-Source"] = "finra-db" + return body + + +@router.get( + "/short-ratio/{symbol}", + response_model=ShortRatioHistoryResponse, + summary="Get short ratio history for a symbol", + description="Return daily short_ratio (aggregated across markets) for the last N days.", +) +async def get_short_ratio( + symbol: str, + response: Response, + days: int = Query(60, ge=1, le=365, description="Number of days"), + force_refresh: bool = Query(False, description="Bypass cache"), + db: AsyncSession = Depends(get_db), +): + # Check cache + cache_key = build_cache_key("finra:short-ratio", symbol.upper(), days) + if not force_refresh: + cached = await get_cached_response(cache_key) + if cached: + cached_body, etag = cached + response.headers["X-Cache"] = "HIT" + response.headers["Cache-Control"] = f"public, max-age={settings.CACHE_TTL}" + response.headers["ETag"] = etag + response.headers["X-Data-Source"] = "redis-cache" + return cached_body + + svc = FinraShortVolumeService() + history = await svc.get_short_ratio_history(db, symbol=symbol, days=days) + + points = [ShortRatioPoint(**h) for h in history] + avg = ( + sum(p.short_ratio for p in points) / len(points) if points else None + ) + + body = ShortRatioHistoryResponse( + symbol=symbol.upper(), + history=points, + avg_short_ratio=round(avg, 6) if avg is not None else None, + metadata={"days_requested": days, "data_points": len(points)}, + ) + + body_dict = body.model_dump() + etag = await set_cached_response(cache_key, body_dict, ttl_seconds=settings.CACHE_TTL) + response.headers["X-Cache"] = "MISS" + response.headers["Cache-Control"] = f"public, max-age={settings.CACHE_TTL}" + response.headers["ETag"] = etag + response.headers["X-Data-Source"] = "finra-db" + return body + + +@router.post( + "/admin/ingest", + response_model=IngestResponse, + summary="Manually ingest FINRA short volume data", + description="Download and ingest FINRA short volume file(s) for a specific date or date range.", +) +async def ingest_short_volume( + date_str: Optional[str] = Query(None, alias="date", description="Single date (YYYY-MM-DD)"), + start_date: Optional[date] = Query(None, description="Range start (YYYY-MM-DD)"), + end_date: Optional[date] = Query(None, description="Range end (YYYY-MM-DD)"), + force: bool = Query(False, description="Re-ingest even if data exists"), + db: AsyncSession = Depends(get_db), +): + svc = FinraShortVolumeService() + + # Single date + if date_str: + try: + target = datetime.strptime(date_str, "%Y-%m-%d").date() + except ValueError: + raise HTTPException(status_code=400, detail="Invalid date format. Use YYYY-MM-DD.") + count = await svc.ingest_date(db, target, force_refresh=force) + return IngestResponse( + date=target.isoformat(), + records_ingested=count, + status="completed", + ) + + # Date range + if start_date and end_date: + if start_date > end_date: + raise HTTPException(status_code=400, detail="start_date must be <= end_date") + count = await svc.ingest_date_range(db, start_date, end_date, force_refresh=force) + return IngestResponse( + date_range={"start": start_date.isoformat(), "end": end_date.isoformat()}, + records_ingested=count, + status="completed", + ) + + raise HTTPException( + status_code=400, + detail="Provide either 'date' (single date) or both 'start_date' and 'end_date'.", + ) diff --git a/app/core/config.py b/app/core/config.py index 8caec28..cda15a1 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -102,6 +102,11 @@ class Settings(BaseSettings): return {str(k).strip().upper(): str(val).strip() for k, val in v.items() if str(val).strip()} return {} + # Alpaca Market Data + ALPACA_API_KEY: str = os.getenv("ALPACA_API_KEY", "") + ALPACA_SECRET_KEY: str = os.getenv("ALPACA_SECRET_KEY", "") + ALPACA_BASE_URL: str = os.getenv("ALPACA_BASE_URL", "https://data.alpaca.markets") + # Server API_PORT: int = int(os.getenv("API_PORT", "18000")) DB_PORT: int = int(os.getenv("DB_PORT", "15432")) diff --git a/app/main.py b/app/main.py index 0f305e4..19afc9d 100644 --- a/app/main.py +++ b/app/main.py @@ -12,7 +12,7 @@ from app.core.config import settings from app.api.v1.api import api_router from app.core.database import engine, Base from app.middleware.error_logger import ErrorLoggingMiddleware -from app.models import error_log, request_log, fred_data, filing # Import to register models +from app.models import error_log, request_log, fred_data, filing, finra_short_volume, alpaca_price # Import to register models # Create database tables @asynccontextmanager @@ -159,6 +159,19 @@ async def root_documentation():
  • GET /filings/exhibit/{{accession_number}} - Extract exhibit content (e.g., EX-99.1 press releases)
    • +

    Alpaca Market Data NEW

    + + +

    FINRA Short Volume NEW

    + +

    ETF Holdings

    +

    Alpaca Market Data NEW

    + 
    # Check Alpaca connection status
+curl "http://localhost:18001/api/v1/alpaca/status"
+
+# Get daily bars from Alpaca
+curl "http://localhost:18001/api/v1/alpaca/bars/AAPL?interval=1d&start_date=2025-01-01&end_date=2025-01-31"
    + +

    FINRA Short Volume NEW

    + 
    # Ingest FINRA data for a specific date
+curl -X POST "http://localhost:18001/api/v1/finra/admin/ingest?date=2025-03-10"
+
+# Get short volume for AAPL (last 30 days)
+curl "http://localhost:18001/api/v1/finra/short-volume/AAPL?days=30"
+
+# Get short ratio history
+curl "http://localhost:18001/api/v1/finra/short-ratio/AAPL?days=60"
    +

    🔧 Data Sources

    @@ -295,6 +327,14 @@ app.openapi_tags = [ "name": "etf", "description": "ETF holdings endpoints" }, + { + "name": "alpaca", + "description": "Alpaca Market Data endpoints (OHLCV bars, connection status)" + }, + { + "name": "finra", + "description": "FINRA RegSHO short sale volume data (ingest, query, ratio history)" + }, { "name": "admin", "description": "Administrative endpoints (migration, etc.)" diff --git a/app/models/__init__.py b/app/models/__init__.py index 6f734c7..9edd4ad 100644 --- a/app/models/__init__.py +++ b/app/models/__init__.py @@ -1,6 +1,8 @@ from app.models.financial import Company, FinancialData, CalculatedMetrics, PriceData, DataUpdateLog from app.models.etf import CusipMap, ETFCIKMap, ETFSeriesMap, ETFHoldingsSnapshot, ETFHolding from app.models.filing import SECFiling +from app.models.finra_short_volume import FinraShortVolume +from app.models.alpaca_price import AlpacaPriceData __all__ = [ "Company", @@ -14,4 +16,6 @@ __all__ = [ "ETFHoldingsSnapshot", "ETFHolding", "SECFiling", + "FinraShortVolume", + "AlpacaPriceData", ] \ No newline at end of file diff --git a/app/models/alpaca_price.py b/app/models/alpaca_price.py new file mode 100644 index 0000000..961d21e --- /dev/null +++ b/app/models/alpaca_price.py @@ -0,0 +1,36 @@ +""" +Alpaca-specific price data model — separate from Yahoo Finance PriceData +to avoid UniqueConstraint('ticker', 'date') conflicts. +""" + +from sqlalchemy import Column, String, Float, Integer, UniqueConstraint, Index +from sqlalchemy.dialects.postgresql import UUID, TIMESTAMP +from datetime import datetime, timezone +import uuid + +from app.core.database import Base + + +class AlpacaPriceData(Base): + __tablename__ = "alpaca_price_data" + + id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + ticker = Column(String(10), nullable=False, index=True) + date = Column(TIMESTAMP(timezone=True), nullable=False) + open = Column(Float, nullable=True) + high = Column(Float, nullable=True) + low = Column(Float, nullable=True) + close = Column(Float, nullable=False) + volume = Column(Float, nullable=True) + vwap = Column(Float, nullable=True) # Volume-weighted average price (Alpaca-specific) + trade_count = Column(Integer, nullable=True) # Number of trades (Alpaca-specific) + + # Metadata + data_source = Column(String(50), default='ALPACA') + created_at = Column(TIMESTAMP(timezone=True), default=lambda: datetime.now(timezone.utc)) + updated_at = Column(TIMESTAMP(timezone=True), default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc)) + + __table_args__ = ( + UniqueConstraint('ticker', 'date', name='uq_alpaca_price_data'), + Index('idx_alpaca_price_ticker_date', 'ticker', 'date'), + ) diff --git a/app/models/finra_short_volume.py b/app/models/finra_short_volume.py new file mode 100644 index 0000000..6db91c1 --- /dev/null +++ b/app/models/finra_short_volume.py @@ -0,0 +1,32 @@ +""" +FINRA RegSHO Short Sale Volume model +""" + +from sqlalchemy import Column, String, Float, Index, UniqueConstraint +from sqlalchemy.dialects.postgresql import UUID, TIMESTAMP +from datetime import datetime, timezone +import uuid + +from app.core.database import Base + + +class FinraShortVolume(Base): + __tablename__ = "finra_short_volume" + + id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + date = Column(TIMESTAMP(timezone=True), nullable=False) + symbol = Column(String(10), nullable=False, index=True) + short_volume = Column(Float, nullable=False) + short_exempt_volume = Column(Float, default=0) + total_volume = Column(Float, nullable=False) + market = Column(String(10), nullable=True) # e.g. "B", "Q", "N" + short_ratio = Column(Float, nullable=True) # short_volume / total_volume + created_at = Column( + TIMESTAMP(timezone=True), default=lambda: datetime.now(timezone.utc) + ) + + __table_args__ = ( + UniqueConstraint("symbol", "date", "market", name="uq_finra_short_volume"), + Index("idx_finra_symbol_date", "symbol", "date"), + Index("idx_finra_date", "date"), + ) diff --git a/app/schemas/financial.py b/app/schemas/financial.py index 942aa25..0adbf7e 100644 --- a/app/schemas/financial.py +++ b/app/schemas/financial.py @@ -23,6 +23,7 @@ class PeriodType(str, Enum): class DataSource(str, Enum): SEC_EDGAR = "SEC_EDGAR" YAHOO_FINANCE = "YAHOO_FINANCE" + ALPACA = "ALPACA" ALPHA_VANTAGE = "ALPHA_VANTAGE" MOCK = "MOCK" @@ -391,6 +392,26 @@ class PriceDataPoint(BaseModel): return v +class AlpacaPriceDataPoint(BaseModel): + model_config = ConfigDict(from_attributes=True) + + date: date + open: Optional[float] = None + high: Optional[float] = None + low: Optional[float] = None + close: float + volume: Optional[float] = None + vwap: Optional[float] = None + trade_count: Optional[int] = None + data_source: str = "ALPACA" + + @validator('date', pre=True) + def convert_datetime_to_date(cls, v): + if isinstance(v, datetime): + return v.date() + return v + + class FinancialDataResponse(BaseModel): company: CompanyInfo financial_data: List[FinancialDataPoint] diff --git a/app/schemas/finra.py b/app/schemas/finra.py new file mode 100644 index 0000000..78a0beb --- /dev/null +++ b/app/schemas/finra.py @@ -0,0 +1,63 @@ +""" +Pydantic schemas for FINRA short volume data +""" + +from datetime import date, datetime +from typing import Dict, List, Any, Optional +from pydantic import BaseModel, Field, ConfigDict + + +class ShortVolumeEntry(BaseModel): + model_config = ConfigDict(from_attributes=True) + + date: date + symbol: str + short_volume: float + short_exempt_volume: float = 0.0 + total_volume: float + market: Optional[str] = None + short_ratio: Optional[float] = None + + @classmethod + def from_orm_obj(cls, obj): + d = obj.date + if isinstance(d, datetime): + d = d.date() + return cls( + date=d, + symbol=obj.symbol, + short_volume=obj.short_volume, + short_exempt_volume=obj.short_exempt_volume or 0.0, + total_volume=obj.total_volume, + market=obj.market, + short_ratio=obj.short_ratio, + ) + + +class ShortVolumeResponse(BaseModel): + symbol: str + entries: List[ShortVolumeEntry] + total_count: int + metadata: Dict[str, Any] = Field(default_factory=dict) + + +class ShortRatioPoint(BaseModel): + date: date + short_volume: float + short_exempt_volume: float + total_volume: float + short_ratio: float + + +class ShortRatioHistoryResponse(BaseModel): + symbol: str + history: List[ShortRatioPoint] + avg_short_ratio: Optional[float] = None + metadata: Dict[str, Any] = Field(default_factory=dict) + + +class IngestResponse(BaseModel): + date: Optional[str] = None + date_range: Optional[Dict[str, str]] = None + records_ingested: int + status: str diff --git a/app/services/alpaca_client.py b/app/services/alpaca_client.py new file mode 100644 index 0000000..3783746 --- /dev/null +++ b/app/services/alpaca_client.py @@ -0,0 +1,233 @@ +""" +Alpaca Market Data API client (v2) +""" + +import asyncio +import time +import logging +from typing import Dict, List, Optional + +import httpx + +from app.core.config import settings + +logger = logging.getLogger(__name__) + +# Interval mapping: internal format → Alpaca API format +INTERVAL_MAP = { + "1m": "1Min", + "2m": "2Min", + "5m": "5Min", + "15m": "15Min", + "30m": "30Min", + "1h": "1Hour", + "1d": "1Day", + "1w": "1Week", + "1mo": "1Month", +} + + +class AlpacaClient: + """Alpaca Market Data API client (v2)""" + + def __init__( + self, + api_key: Optional[str] = None, + secret_key: Optional[str] = None, + base_url: Optional[str] = None, + max_requests_per_min: int = 200, + ): + self.api_key = api_key or settings.ALPACA_API_KEY + self.secret_key = secret_key or settings.ALPACA_SECRET_KEY + self.base_url = (base_url or settings.ALPACA_BASE_URL).rstrip("/") + self._semaphore = asyncio.Semaphore(max_requests_per_min) + self._request_times: List[float] = [] + self._max_rpm = max_requests_per_min + self._client: Optional[httpx.AsyncClient] = None + + # ------------------------------------------------------------------ + # Lifecycle + # ------------------------------------------------------------------ + + def is_configured(self) -> bool: + """Return True if both API key and secret are set.""" + return bool(self.api_key) and bool(self.secret_key) + + async def _get_client(self) -> httpx.AsyncClient: + if self._client is None or self._client.is_closed: + self._client = httpx.AsyncClient( + base_url=self.base_url, + headers={ + "APCA-API-KEY-ID": self.api_key, + "APCA-API-SECRET-KEY": self.secret_key, + }, + timeout=30.0, + ) + return self._client + + async def close(self): + if self._client and not self._client.is_closed: + await self._client.aclose() + + # ------------------------------------------------------------------ + # Rate limiting + # ------------------------------------------------------------------ + + async def _wait_for_rate_limit(self): + """Simple token-bucket style rate limiter (200 req/min).""" + now = time.monotonic() + # Remove entries older than 60 seconds + self._request_times = [t for t in self._request_times if now - t < 60] + if len(self._request_times) >= self._max_rpm: + sleep_for = 60 - (now - self._request_times[0]) + 0.1 + if sleep_for > 0: + logger.debug(f"Alpaca rate limit reached, sleeping {sleep_for:.1f}s") + await asyncio.sleep(sleep_for) + self._request_times.append(time.monotonic()) + + # ------------------------------------------------------------------ + # Core HTTP + # ------------------------------------------------------------------ + + async def _request( + self, method: str, path: str, params: Optional[Dict] = None, retries: int = 3 + ) -> Dict: + """Make an authenticated request with retry + rate limiting.""" + await self._wait_for_rate_limit() + client = await self._get_client() + + last_exc: Optional[Exception] = None + for attempt in range(retries): + try: + resp = await client.request(method, path, params=params) + if resp.status_code == 429: + wait = 2 ** attempt + logger.warning(f"Alpaca 429 – retrying in {wait}s (attempt {attempt + 1})") + await asyncio.sleep(wait) + continue + if resp.status_code >= 500: + wait = 2 ** attempt + logger.warning(f"Alpaca {resp.status_code} – retrying in {wait}s (attempt {attempt + 1})") + await asyncio.sleep(wait) + continue + resp.raise_for_status() + return resp.json() + except httpx.HTTPStatusError as exc: + last_exc = exc + if attempt < retries - 1 and exc.response.status_code in (429, 500, 502, 503, 504): + await asyncio.sleep(2 ** attempt) + continue + raise + except (httpx.ConnectError, httpx.ReadTimeout) as exc: + last_exc = exc + if attempt < retries - 1: + await asyncio.sleep(2 ** attempt) + continue + raise + + raise last_exc # type: ignore[misc] + + # ------------------------------------------------------------------ + # Public API methods + # ------------------------------------------------------------------ + + async def get_bars( + self, + symbol: str, + timeframe: str = "1d", + start: Optional[str] = None, + end: Optional[str] = None, + limit: int = 10000, + ) -> List[Dict]: + """ + Fetch bars for a single symbol with automatic pagination. + + Args: + symbol: Ticker symbol (e.g. "AAPL") + timeframe: Internal interval string (e.g. "1d", "1h", "1m") + start: RFC-3339 date/datetime (e.g. "2024-01-01") + end: RFC-3339 date/datetime + limit: Max bars per page (Alpaca max 10000) + + Returns: + List of bar dicts with keys: t, o, h, l, c, v, n, vw + """ + alpaca_tf = INTERVAL_MAP.get(timeframe, timeframe) + params: Dict = {"timeframe": alpaca_tf, "limit": min(limit, 10000)} + if start: + params["start"] = start + if end: + params["end"] = end + + all_bars: List[Dict] = [] + path = f"/v2/stocks/{symbol.upper()}/bars" + + while True: + data = await self._request("GET", path, params=params) + bars = data.get("bars") or [] + all_bars.extend(bars) + next_token = data.get("next_page_token") + if not next_token or not bars: + break + params["page_token"] = next_token + + logger.info(f"Alpaca: fetched {len(all_bars)} bars for {symbol} ({alpaca_tf})") + return all_bars + + async def get_multi_bars( + self, + symbols: List[str], + timeframe: str = "1d", + start: Optional[str] = None, + end: Optional[str] = None, + limit: int = 10000, + ) -> Dict[str, List[Dict]]: + """ + Fetch bars for multiple symbols in one request with auto-pagination. + + Returns: + Dict mapping symbol → list of bar dicts + """ + alpaca_tf = INTERVAL_MAP.get(timeframe, timeframe) + params: Dict = { + "symbols": ",".join(s.upper() for s in symbols), + "timeframe": alpaca_tf, + "limit": min(limit, 10000), + } + if start: + params["start"] = start + if end: + params["end"] = end + + result: Dict[str, List[Dict]] = {s.upper(): [] for s in symbols} + path = "/v2/stocks/bars" + + while True: + data = await self._request("GET", path, params=params) + bars_map = data.get("bars") or {} + for sym, bars in bars_map.items(): + result.setdefault(sym, []).extend(bars) + next_token = data.get("next_page_token") + if not next_token: + break + params["page_token"] = next_token + + total = sum(len(v) for v in result.values()) + logger.info(f"Alpaca: fetched {total} bars for {len(symbols)} symbols ({alpaca_tf})") + return result + + async def check_connection(self) -> Dict: + """Verify API key validity by requesting a small amount of data.""" + try: + bars = await self.get_bars("AAPL", timeframe="1d", limit=1) + return { + "connected": True, + "bars_returned": len(bars), + "base_url": self.base_url, + } + except Exception as e: + return { + "connected": False, + "error": str(e), + "base_url": self.base_url, + } diff --git a/app/services/alpaca_price_service.py b/app/services/alpaca_price_service.py new file mode 100644 index 0000000..6f07027 --- /dev/null +++ b/app/services/alpaca_price_service.py @@ -0,0 +1,131 @@ +""" +Alpaca bars → AlpacaPriceData conversion and DB storage +""" + +import logging +from datetime import datetime, timezone +from typing import Dict, List, Optional + +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy import select, and_ + +from app.models.alpaca_price import AlpacaPriceData +from app.services.alpaca_client import AlpacaClient + +logger = logging.getLogger(__name__) + + +class AlpacaPriceService: + """Alpaca bars → AlpacaPriceData conversion and DB storage""" + + def __init__(self, client: Optional[AlpacaClient] = None): + self.client = client or AlpacaClient() + + def is_available(self) -> bool: + return self.client.is_configured() + + async def fetch_and_store_bars( + self, + db: AsyncSession, + ticker: str, + start_date: datetime, + end_date: datetime, + interval: str = "1d", + ) -> int: + """ + Fetch bars from Alpaca and upsert into AlpacaPriceData table. + + Returns: + Number of newly inserted records. + """ + ticker = ticker.upper() + start_str = start_date.strftime("%Y-%m-%d") + end_str = end_date.strftime("%Y-%m-%d") + + bars = await self.client.get_bars( + symbol=ticker, + timeframe=interval, + start=start_str, + end=end_str, + ) + + if not bars: + logger.warning(f"Alpaca returned 0 bars for {ticker}") + return 0 + + # Fetch existing dates for dedup + result = await db.execute( + select(AlpacaPriceData.date).where(AlpacaPriceData.ticker == ticker) + ) + existing_dates = {row[0].date() for row in result.fetchall()} + + inserted = 0 + for bar in bars: + bar_dt = _parse_bar_timestamp(bar["t"]) + if bar_dt.date() in existing_dates: + continue + + record = AlpacaPriceData( + ticker=ticker, + date=bar_dt, + open=float(bar.get("o", 0)), + high=float(bar.get("h", 0)), + low=float(bar.get("l", 0)), + close=float(bar.get("c", 0)), + volume=float(bar.get("v", 0)), + vwap=float(bar["vw"]) if bar.get("vw") else None, + trade_count=int(bar["n"]) if bar.get("n") else None, + data_source="ALPACA", + ) + db.add(record) + existing_dates.add(bar_dt.date()) + inserted += 1 + + if inserted: + await db.commit() + logger.info(f"Alpaca: inserted {inserted} bars for {ticker}") + + return inserted + + async def fetch_bars_raw( + self, + ticker: str, + interval: str = "1d", + start_date: Optional[datetime] = None, + end_date: Optional[datetime] = None, + ) -> List[Dict]: + """ + Return raw bar dicts without touching the DB (useful for intraday / non-persistent use). + """ + start_str = start_date.strftime("%Y-%m-%d") if start_date else None + end_str = end_date.strftime("%Y-%m-%d") if end_date else None + + bars = await self.client.get_bars( + symbol=ticker.upper(), + timeframe=interval, + start=start_str, + end=end_str, + ) + + return [ + { + "timestamp": bar["t"], + "open": bar.get("o"), + "high": bar.get("h"), + "low": bar.get("l"), + "close": bar.get("c"), + "volume": bar.get("v"), + "vwap": bar.get("vw"), + "trade_count": bar.get("n"), + } + for bar in bars + ] + + +def _parse_bar_timestamp(ts_str: str) -> datetime: + """Parse Alpaca bar timestamp (RFC-3339) into a timezone-aware datetime.""" + # Alpaca returns e.g. "2024-01-02T05:00:00Z" + dt = datetime.fromisoformat(ts_str.replace("Z", "+00:00")) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt diff --git a/app/services/finra_short_volume_service.py b/app/services/finra_short_volume_service.py new file mode 100644 index 0000000..4cf6a9b --- /dev/null +++ b/app/services/finra_short_volume_service.py @@ -0,0 +1,292 @@ +""" +FINRA RegSHO daily short sale volume data service +""" + +import logging +from datetime import date, datetime, timedelta, timezone +from typing import Dict, List, Optional, Tuple + +import aiohttp +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy import select, and_, func, desc + +from app.models.finra_short_volume import FinraShortVolume + +logger = logging.getLogger(__name__) + +FINRA_CDN_BASE = "https://cdn.finra.org/equity/regsho/daily" + + +class FinraShortVolumeService: + """FINRA RegSHO daily short sale volume data""" + + # ------------------------------------------------------------------ + # Fetch & Parse + # ------------------------------------------------------------------ + + async def fetch_daily_file(self, target_date: date) -> Optional[str]: + """ + Download the CNMS short volume file for *target_date*. + + Returns the raw text or None if 404 (weekend/holiday). + """ + filename = f"CNMSshvol{target_date:%Y%m%d}.txt" + url = f"{FINRA_CDN_BASE}/{filename}" + + try: + async with aiohttp.ClientSession() as session: + async with session.get(url, timeout=aiohttp.ClientTimeout(total=30)) as resp: + if resp.status == 404: + logger.debug(f"FINRA file not found (holiday/weekend): {filename}") + return None + resp.raise_for_status() + text = await resp.text() + logger.info(f"FINRA: downloaded {filename} ({len(text)} bytes)") + return text + except aiohttp.ClientError as e: + logger.error(f"FINRA download error for {filename}: {e}") + return None + + def parse_short_volume_file(self, text: str) -> List[Dict]: + """ + Parse pipe-delimited FINRA short volume text. + + Expected header: + Date|Symbol|ShortVolume|ShortExemptVolume|TotalVolume|Market + + Returns list of dicts ready for DB insertion. + """ + lines = text.strip().splitlines() + if len(lines) < 2: + return [] + + records: List[Dict] = [] + for line in lines[1:]: + parts = line.split("|") + if len(parts) < 6: + continue + raw_date, symbol, short_vol, exempt_vol, total_vol, market = ( + parts[0].strip(), + parts[1].strip(), + parts[2].strip(), + parts[3].strip(), + parts[4].strip(), + parts[5].strip(), + ) + if not symbol or not total_vol: + continue + + try: + dt = datetime.strptime(raw_date, "%Y%m%d").replace(tzinfo=timezone.utc) + sv = float(short_vol) + sev = float(exempt_vol) if exempt_vol else 0.0 + tv = float(total_vol) + ratio = sv / tv if tv > 0 else 0.0 + except (ValueError, ZeroDivisionError): + continue + + records.append( + { + "date": dt, + "symbol": symbol.upper(), + "short_volume": sv, + "short_exempt_volume": sev, + "total_volume": tv, + "market": market if market else None, + "short_ratio": round(ratio, 6), + } + ) + + return records + + # ------------------------------------------------------------------ + # Ingest + # ------------------------------------------------------------------ + + async def ingest_date( + self, db: AsyncSession, target_date: date, force_refresh: bool = False + ) -> int: + """ + Download, parse, and bulk-upsert data for a single date. + + Returns number of records inserted. + """ + if not force_refresh: + # Check if already ingested + count = await self._count_for_date(db, target_date) + if count > 0: + logger.info(f"FINRA date {target_date} already ingested ({count} rows)") + return 0 + + text = await self.fetch_daily_file(target_date) + if text is None: + return 0 + + records = self.parse_short_volume_file(text) + if not records: + return 0 + + # Bulk insert – skip duplicates + inserted = 0 + for rec in records: + existing = await db.execute( + select(FinraShortVolume.id).where( + and_( + FinraShortVolume.symbol == rec["symbol"], + FinraShortVolume.date == rec["date"], + FinraShortVolume.market == rec["market"], + ) + ) + ) + if existing.first(): + continue + + db.add(FinraShortVolume(**rec)) + inserted += 1 + + if inserted: + await db.commit() + logger.info(f"FINRA: ingested {inserted} records for {target_date}") + + return inserted + + async def ingest_date_range( + self, + db: AsyncSession, + start_date: date, + end_date: date, + force_refresh: bool = False, + ) -> int: + """Ingest short volume data for a date range (weekdays only).""" + total = 0 + current = start_date + while current <= end_date: + # Skip weekends + if current.weekday() < 5: + count = await self.ingest_date(db, current, force_refresh=force_refresh) + total += count + current += timedelta(days=1) + + logger.info(f"FINRA range ingest: {total} records from {start_date} to {end_date}") + return total + + # ------------------------------------------------------------------ + # Query + # ------------------------------------------------------------------ + + async def get_short_volume( + self, + db: AsyncSession, + symbol: str, + start_date: Optional[date] = None, + end_date: Optional[date] = None, + limit: int = 100, + ) -> Tuple[List[FinraShortVolume], int]: + """ + Query short volume for a symbol within a date range. + + If no data found, attempts auto-ingest for the requested range. + """ + symbol = symbol.upper() + + conditions = [FinraShortVolume.symbol == symbol] + if start_date: + conditions.append( + FinraShortVolume.date >= datetime.combine(start_date, datetime.min.time()).replace(tzinfo=timezone.utc) + ) + if end_date: + conditions.append( + FinraShortVolume.date <= datetime.combine(end_date, datetime.min.time()).replace(tzinfo=timezone.utc) + ) + + # Count + count_q = await db.execute( + select(func.count(FinraShortVolume.id)).where(and_(*conditions)) + ) + total_count = count_q.scalar() or 0 + + # If no data, try auto-ingest + if total_count == 0 and start_date and end_date: + ingested = await self.ingest_date_range(db, start_date, end_date) + if ingested > 0: + count_q = await db.execute( + select(func.count(FinraShortVolume.id)).where(and_(*conditions)) + ) + total_count = count_q.scalar() or 0 + + # Fetch + result = await db.execute( + select(FinraShortVolume) + .where(and_(*conditions)) + .order_by(desc(FinraShortVolume.date)) + .limit(limit) + ) + rows = result.scalars().all() + + return rows, total_count + + async def get_short_ratio_history( + self, + db: AsyncSession, + symbol: str, + days: int = 30, + ) -> List[Dict]: + """ + Return short_ratio history for the last *days* trading days. + + Groups by date (aggregates across markets). + """ + symbol = symbol.upper() + cutoff = datetime.now(timezone.utc) - timedelta(days=days) + + result = await db.execute( + select( + FinraShortVolume.date, + func.sum(FinraShortVolume.short_volume).label("short_volume"), + func.sum(FinraShortVolume.short_exempt_volume).label("short_exempt_volume"), + func.sum(FinraShortVolume.total_volume).label("total_volume"), + ) + .where( + and_( + FinraShortVolume.symbol == symbol, + FinraShortVolume.date >= cutoff, + ) + ) + .group_by(FinraShortVolume.date) + .order_by(FinraShortVolume.date) + ) + + history = [] + for row in result.fetchall(): + dt, sv, sev, tv = row + ratio = sv / tv if tv and tv > 0 else 0.0 + history.append( + { + "date": dt.date() if isinstance(dt, datetime) else dt, + "short_volume": sv, + "short_exempt_volume": sev, + "total_volume": tv, + "short_ratio": round(ratio, 6), + } + ) + + # Auto-ingest if empty + if not history: + start = (datetime.now(timezone.utc) - timedelta(days=days)).date() + end = datetime.now(timezone.utc).date() + ingested = await self.ingest_date_range(db, start, end) + if ingested > 0: + return await self.get_short_ratio_history(db, symbol, days) + + return history + + # ------------------------------------------------------------------ + # Helpers + # ------------------------------------------------------------------ + + async def _count_for_date(self, db: AsyncSession, target_date: date) -> int: + dt = datetime.combine(target_date, datetime.min.time()).replace(tzinfo=timezone.utc) + result = await db.execute( + select(func.count(FinraShortVolume.id)).where(FinraShortVolume.date == dt) + ) + return result.scalar() or 0 diff --git a/app/services/price_data_service.py b/app/services/price_data_service.py index fb3c1ac..8625bf3 100644 --- a/app/services/price_data_service.py +++ b/app/services/price_data_service.py @@ -30,7 +30,6 @@ except ImportError: logger.error("yfinance-plus not available for price data") YFINANCE_AVAILABLE = False - class PriceDataService: def __init__(self): self.yf_available = YFINANCE_AVAILABLE @@ -70,7 +69,7 @@ class PriceDataService: if missing_periods or force_refresh: if not self.yf_available: raise ValueError("Yahoo Finance (yfinance-plus) data source not available") - + # Fetch data from Yahoo Finance using yfinance-plus await self._fetch_and_store_price_data( db, ticker, start_date, end_date, interval diff --git a/docker-compose.yml b/docker-compose.yml index 5ef0d98..b5137d6 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -48,6 +48,8 @@ services: - DEBUG=True - API_PORT=18000 - SEC_EMAIL=example@example.com + - ALPACA_API_KEY=${ALPACA_API_KEY:-} + - ALPACA_SECRET_KEY=${ALPACA_SECRET_KEY:-} ports: - "18001:18000" # External:Internal port mapping depends_on: