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 <noreply@anthropic.com>

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"])
+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"])

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()

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'.",
+    )

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"))

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():
                 <li><code>GET /filings/exhibit/{{accession_number}}</code> - Extract exhibit content (e.g., EX-99.1 press releases)</li>
             </ul>
 
+            <h3>Alpaca Market Data <span class="new-badge">NEW</span></h3>
+            <ul>
+                <li><code>GET /alpaca/status</code> - Alpaca connection status and key validity</li>
+                <li><code>GET /alpaca/bars/{{ticker}}</code> - Fetch bars directly from Alpaca (raw, no DB)</li>
+            </ul>
+
+            <h3>FINRA Short Volume <span class="new-badge">NEW</span></h3>
+            <ul>
+                <li><code>GET /finra/short-volume/{{symbol}}</code> - Short sale volume data for a symbol</li>
+                <li><code>GET /finra/short-ratio/{{symbol}}</code> - Short ratio history (aggregated)</li>
+                <li><code>POST /finra/admin/ingest</code> - Manually ingest FINRA data for a date/range</li>
+            </ul>
+
             <h3>ETF Holdings</h3>
             <ul>
                 <li><code>GET /etf/holdings/{{ticker}}</code> - ETF holdings at date or most recent</li>
@@ -242,10 +255,29 @@ news = client.get_news_social_data("AAPL", days_back=7)</code></pre>
                 <li><strong>Investment Metrics</strong> - P/E, ROE, debt ratios, growth metrics</li>
             </ul>
 
+            <h3>Alpaca Market Data <span class="new-badge">NEW</span></h3>
+            <pre><code># 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"</code></pre>
+
+            <h3>FINRA Short Volume <span class="new-badge">NEW</span></h3>
+            <pre><code># 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"</code></pre>
+
             <h2>🔧 Data Sources</h2>
             <ul>
                 <li><strong>SEC EDGAR</strong> - Official company filings and ETF holdings</li>
                 <li><strong>Yahoo Finance</strong> - Price data and financial news (via yfinance_plus)</li>
+                <li><strong>Alpaca</strong> - Market price data (OHLCV) with optional API key</li>
+                <li><strong>FINRA</strong> - RegSHO short sale volume data (public CDN, no API key)</li>
                 <li><strong>NewsAPI</strong> - Professional news aggregation</li>
                 <li><strong>Reddit API</strong> - Social media sentiment from investing subreddits</li>
             </ul>
@@ -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.)"

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",
 ]

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'),
+    )

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"),
+    )

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]

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

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,
+            }

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

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

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

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: