costajohnt
diff --git a/‎client.py‎
Lines changed: 29 additions & 1 deletion b/‎client.py‎
Lines changed: 29 additions & 1 deletion
diff --git a/‎strategies/bar_utils.py‎
Lines changed: 249 additions & 0 deletions b/‎strategies/bar_utils.py‎
Lines changed: 249 additions & 0 deletions
@@ -1,9 +1,10 @@
-"""Alpaca client setup -- loads credentials from .env and exposes trading and data clients."""
+"""Alpaca client setup -- loads credentials from .env and exposes trading, data, and streaming clients."""
 
 import os
 from pathlib import Path
 
 from alpaca.data.historical import StockHistoricalDataClient
+from alpaca.data.live import StockDataStream
 from alpaca.trading.client import TradingClient
 from dotenv import load_dotenv
 
@@ -34,3 +35,30 @@ def get_data_client() -> StockHistoricalDataClient:
         api_key=os.environ["ALPACA_API_KEY"],
         secret_key=os.environ["ALPACA_SECRET_KEY"],
     )
+
+
+def get_stream_client() -> StockDataStream:
+    """Create an Alpaca WebSocket streaming client for real-time price data.
+
+    Returns a StockDataStream configured for the appropriate environment
+    (paper or live). The caller must subscribe to symbols and call .run()
+    to start receiving data.
+
+    Example usage:
+        stream = get_stream_client()
+
+        @stream.on("bar")
+        async def on_bar(bar):
+            print(f"{bar.symbol}: ${bar.close}")
+
+        stream.subscribe_bars(on_bar, "AAPL", "MSFT")
+        stream.run()
+    """
+    _load_env()
+    is_paper = os.environ.get("ALPACA_PAPER", "true").lower() == "true"
+    feed = "iex" if is_paper else "sip"
+    return StockDataStream(
+        api_key=os.environ["ALPACA_API_KEY"],
+        secret_key=os.environ["ALPACA_SECRET_KEY"],
+        feed=feed,
+    )
@@ -0,0 +1,249 @@
+"""
+Bar data batching utilities — reduce API calls by fetching multiple symbols at once.
+
+Instead of calling Alpaca's bar data API one symbol at a time (3+ calls per symbol
+in the pipeline), this module batches multi-symbol requests to dramatically reduce
+API call volume.
+
+For a screener scanning 50 candidates:
+  Before: 150+ individual API calls (3 per symbol: daily bars, weekly bars, ATR bars)
+  After:  ~3 batched requests (one per timeframe)
+
+Usage:
+    from strategies.bar_utils import fetch_bars_batch, extract_symbol_bars
+
+    # Fetch daily bars for many symbols at once
+    all_bars_df = fetch_bars_batch(data_client, symbols, timeframe=TimeFrame.Day, days=50)
+
+    # Extract a single symbol's bars from the batch result
+    aapl_df = extract_symbol_bars(all_bars_df, "AAPL")
+"""
+
+from datetime import datetime, timedelta
+
+import pandas as pd
+from alpaca.data.requests import StockBarsRequest
+from alpaca.data.timeframe import TimeFrame
+
+from strategies.log import get_logger
+
+logger = get_logger(__name__)
+
+# Alpaca API limit for multi-symbol bar requests
+DEFAULT_BATCH_SIZE = 500
+
+
+def extract_symbol_bars(df: pd.DataFrame, symbol: str) -> pd.DataFrame | None:
+    """Extract and sort bars for a single symbol from a multi-symbol DataFrame.
+
+    Args:
+        df: Multi-index DataFrame from Alpaca's get_stock_bars() with
+            (symbol, timestamp) index levels.
+        symbol: The ticker symbol to extract.
+
+    Returns:
+        A sorted DataFrame of bars for the symbol, or None if the symbol
+        is not found in the data.
+    """
+    if df is None or df.empty:
+        return None
+
+    try:
+        if symbol in df.index.get_level_values(0):
+            sdf = df.loc[symbol].sort_index()
+            if len(sdf) > 0:
+                return sdf
+    except (KeyError, TypeError):
+        pass
+
+    return None
+
+
+def fetch_bars_batch(
+    data_client,
+    symbols: list[str],
+    timeframe: TimeFrame = TimeFrame.Day,
+    days: int = 50,
+    batch_size: int = DEFAULT_BATCH_SIZE,
+) -> pd.DataFrame:
+    """Fetch bar data for multiple symbols in batched API calls.
+
+    Alpaca supports multi-symbol bar requests, so instead of one API call per
+    symbol, we batch them into groups of batch_size (default 500) and fetch
+    all bars in a handful of requests.
+
+    Note: retry_api is NOT applied here because per-batch failures are handled
+    internally with individual symbol fallbacks. The function never raises on
+    API errors -- it degrades gracefully to an empty DataFrame.
+
+    Args:
+        data_client: Alpaca StockHistoricalDataClient instance.
+        symbols: List of ticker symbols to fetch bars for.
+        timeframe: Bar timeframe (TimeFrame.Day, TimeFrame.Week, etc.).
+        days: Number of calendar days of history to fetch.
+        batch_size: Maximum symbols per API request (Alpaca limit).
+
+    Returns:
+        A multi-index DataFrame with (symbol, timestamp) index containing
+        all bar data. May be empty if all requests fail.
+    """
+    if not symbols:
+        return pd.DataFrame()
+
+    # Deduplicate while preserving order to avoid duplicate index entries
+    seen: set[str] = set()
+    unique_symbols: list[str] = []
+    for s in symbols:
+        if s not in seen:
+            seen.add(s)
+            unique_symbols.append(s)
+    symbols = unique_symbols
+
+    start = datetime.now() - timedelta(days=days)
+    all_frames: list[pd.DataFrame] = []
+
+    for i in range(0, len(symbols), batch_size):
+        batch = symbols[i:i + batch_size]
+        try:
+            bars_request = StockBarsRequest(
+                symbol_or_symbols=batch,
+                timeframe=timeframe,
+                start=start,
+            )
+            bars = data_client.get_stock_bars(bars_request)
+            if bars.df is not None and not bars.df.empty:
+                all_frames.append(bars.df)
+        except Exception as e:
+            logger.warning(
+                "Batch bar fetch failed for %d symbols (batch %d-%d): %s",
+                len(batch), i, i + len(batch), e,
+            )
+            # Fall back to individual fetches for this batch
+            for symbol in batch:
+                try:
+                    single_request = StockBarsRequest(
+                        symbol_or_symbols=[symbol],
+                        timeframe=timeframe,
+                        start=start,
+                    )
+                    single_bars = data_client.get_stock_bars(single_request)
+                    if single_bars.df is not None and not single_bars.df.empty:
+                        all_frames.append(single_bars.df)
+                except Exception as sym_err:
+                    logger.debug("Individual bar fetch failed for %s: %s", symbol, sym_err)
+
+    if not all_frames:
+        return pd.DataFrame()
+
+    return pd.concat(all_frames)
+
+
+def fetch_daily_bars(
+    data_client,
+    symbols: list[str],
+    days: int = 50,
+    batch_size: int = DEFAULT_BATCH_SIZE,
+) -> pd.DataFrame:
+    """Convenience wrapper: fetch daily bars for multiple symbols.
+
+    This is the most common use case -- daily bars are used for:
+      - SMA deviation (mean reversion signals)
+      - ATR calculation (volatility normalization)
+      - Weekly SMA (computed from daily bars with a wider window)
+      - Momentum calculation (10d/50d returns)
+
+    By fetching once with a wide enough window, all of these can share the
+    same data without redundant API calls.
+
+    Args:
+        data_client: Alpaca StockHistoricalDataClient instance.
+        symbols: List of ticker symbols.
+        days: Number of calendar days of history (default 50 covers 20-day SMA
+              + 14-day ATR + buffer for weekends/holidays).
+        batch_size: Maximum symbols per API request.
+
+    Returns:
+        Multi-index DataFrame with (symbol, timestamp) index.
+    """
+    return fetch_bars_batch(
+        data_client,
+        symbols,
+        timeframe=TimeFrame.Day,
+        days=days,
+        batch_size=batch_size,
+    )
+
+
+def compute_atr_from_bars(df: pd.DataFrame, period: int = 14) -> tuple[float, float]:
+    """Compute ATR and ATR% from a pre-fetched DataFrame.
+
+    This avoids a separate API call for ATR calculation — reuses bars that
+    were already fetched for other purposes (SMA deviation, etc.).
+
+    Args:
+        df: Single-symbol DataFrame with 'high', 'low', 'close' columns.
+            Must be sorted by time.
+        period: ATR lookback period (default 14).
+
+    Returns:
+        (atr_dollar, atr_pct) tuple. Returns (0.0, 0.0) if insufficient data.
+    """
+    from strategies.volatility import compute_atr, compute_atr_pct
+
+    atr = compute_atr(df, period)
+    atr_pct = compute_atr_pct(df, period)
+    return (atr, atr_pct)
+
+
+def check_weekly_trend_from_bars(
+    symbol: str,
+    df: pd.DataFrame,
+    weekly_sma_period: int = 10,
+) -> tuple[bool, str]:
+    """Check weekly trend from pre-fetched daily bars.
+
+    This avoids a separate API call for weekly trend checking — reuses
+    daily bars that were already fetched for other purposes.
+
+    Requires daily bars with at least (weekly_sma_period * 5) data points.
+
+    Args:
+        symbol: Ticker symbol (for logging).
+        df: Single-symbol DataFrame with 'close' column, sorted by time.
+        weekly_sma_period: Weekly SMA period (default 10 = ~50 trading days).
+
+    Returns:
+        (confirmed, reason) tuple. If confirmed=False, the dip may be structural.
+    """
+    from strategies.constants import (
+        WEEKLY_TREND_LOOKBACK_BARS,
+        WEEKLY_TREND_SEVERE_BELOW_SMA_PCT,
+        WEEKLY_TREND_SMA_DECLINE_FRAC,
+    )
+
+    try:
+        weekly_sma_days = weekly_sma_period * 5
+        if len(df) < weekly_sma_days:
+            return True, ""  # not enough data, don't block
+
+        df = df.copy()
+        df["weekly_sma"] = df["close"].rolling(window=weekly_sma_days).mean()
+        latest_close = df["close"].iloc[-1]
+        latest_weekly_sma = df["weekly_sma"].iloc[-1]
+
+        if pd.isna(latest_weekly_sma) or abs(latest_weekly_sma) < 1e-9:
+            return True, ""
+
+        weekly_deviation = ((latest_close - latest_weekly_sma) / latest_weekly_sma) * 100
+        if weekly_deviation < -WEEKLY_TREND_SEVERE_BELOW_SMA_PCT:
+            return False, f"weekly trend down {weekly_deviation:.0f}% (structural)"
+
+        if len(df) >= weekly_sma_days + WEEKLY_TREND_LOOKBACK_BARS - 1:
+            prev_weekly_sma_3w = df["weekly_sma"].iloc[-WEEKLY_TREND_LOOKBACK_BARS]
+            if not pd.isna(prev_weekly_sma_3w) and latest_weekly_sma < prev_weekly_sma_3w * (1 - WEEKLY_TREND_SMA_DECLINE_FRAC):
+                return False, "weekly SMA declining >2% over 3 weeks"
+
+        return True, ""
+    except Exception as e:
+        logger.warning("Weekly trend check from bars failed for %s: %s", symbol, e)
+        return True, ""