path: root/scripts/stock_screener.py

#!/usr/bin/env python3
"""Claude-powered daily stock screener for MOC strategy.

Analyzes market conditions and recommends stocks to buy at close.
Uses Anthropic Claude API for fundamental/news analysis.

Usage:
    python scripts/stock_screener.py
    python scripts/stock_screener.py --top 5

Requires: ANTHROPIC_API_KEY environment variable
"""

import argparse
import asyncio
import json
import os
import sys
from datetime import UTC, datetime
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "shared" / "src"))

from shared.alpaca import AlpacaClient  # noqa: E402


async def get_market_data(alpaca: AlpacaClient, symbols: list[str]) -> list[dict]:
    """Fetch latest market data for screening."""
    results = []
    for symbol in symbols:
        try:
            bars = await alpaca.get_bars(symbol, timeframe="1Day", limit=5)
            if not bars:
                continue

            latest = bars[-1]
            prev = bars[-2] if len(bars) > 1 else latest

            # Calculate basic metrics
            close = float(latest["c"])
            prev_close = float(prev["c"])
            change_pct = (close - prev_close) / prev_close * 100 if prev_close else 0
            volume = int(latest["v"])

            # 5-day average volume
            avg_vol = sum(int(b["v"]) for b in bars) / len(bars)
            vol_ratio = volume / avg_vol if avg_vol > 0 else 1.0

            # 5-day price range
            highs = [float(b["h"]) for b in bars]
            lows = [float(b["l"]) for b in bars]
            range_pct = (max(highs) - min(lows)) / close * 100 if close > 0 else 0

            results.append(
                {
                    "symbol": symbol,
                    "close": close,
                    "change_pct": round(change_pct, 2),
                    "volume": volume,
                    "vol_ratio": round(vol_ratio, 2),
                    "range_5d_pct": round(range_pct, 2),
                    "is_bullish": close > float(latest["o"]),  # Today bullish?
                }
            )
        except Exception as exc:
            print(f"  Warning: Failed to fetch {symbol}: {exc}")

    return results


def build_claude_prompt(market_data: list[dict]) -> str:
    """Build a prompt for Claude to analyze stocks."""
    data_str = json.dumps(market_data, indent=2)

    return f"""You are a stock analyst. Analyze these US stocks for a Market-on-Close (MOC) overnight trading strategy.

The strategy buys stocks at market close (3:50-4:00 PM ET) and sells at market open next day (9:30-10:00 AM ET).

We want stocks that are likely to gap UP overnight. Good candidates have:
1. Bullish daily candle (closed above open)
2. Volume above average (vol_ratio > 1.0)
3. Positive momentum (change_pct > 0)
4. Not overextended (change_pct < 5% — avoid chasing)
5. Moderate volatility (range_5d_pct between 3-15%)

Market data:
{data_str}

Analyze each stock and return a JSON array of your top recommendations, ranked by confidence.
Format:
```json
[
  {{
    "symbol": "AAPL",
    "score": 85,
    "reason": "Bullish candle on above-average volume, moderate momentum"
  }}
]
```

Only include stocks with score >= 60. If no stocks qualify, return an empty array.
Return ONLY the JSON array, no other text."""


async def analyze_with_claude(prompt: str, api_key: str) -> list[dict]:
    """Send prompt to Claude and parse response."""
    try:
        import anthropic

        client = anthropic.Anthropic(api_key=api_key)
        message = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            messages=[{"role": "user", "content": prompt}],
        )

        response_text = message.content[0].text.strip()

        # Extract JSON from response (handle markdown code blocks)
        if "```json" in response_text:
            response_text = response_text.split("```json")[1].split("```")[0].strip()
        elif "```" in response_text:
            response_text = response_text.split("```")[1].split("```")[0].strip()

        return json.loads(response_text)

    except ImportError:
        print("Error: anthropic package not installed. Run: pip install anthropic")
        return []
    except json.JSONDecodeError as e:
        print(f"Error: Failed to parse Claude response: {e}")
        print(f"Response was: {response_text[:500]}")
        return []
    except Exception as e:
        print(f"Error: Claude API call failed: {e}")
        return []


# Default universe of liquid US stocks
DEFAULT_UNIVERSE = [
    # Tech
    "AAPL",
    "MSFT",
    "GOOGL",
    "AMZN",
    "META",
    "NVDA",
    "TSLA",
    "AMD",
    "INTC",
    "CRM",
    # Finance
    "JPM",
    "BAC",
    "GS",
    "MS",
    "V",
    "MA",
    # Healthcare
    "JNJ",
    "UNH",
    "PFE",
    "ABBV",
    # Consumer
    "WMT",
    "KO",
    "PEP",
    "MCD",
    "NKE",
    # Energy
    "XOM",
    "CVX",
    # Industrial
    "CAT",
    "BA",
    "GE",
    # ETFs
    "SPY",
    "QQQ",
    "IWM",
]


async def main_async(top_n: int = 5, universe: list[str] | None = None):
    api_key = os.environ.get("ANTHROPIC_API_KEY", "")
    alpaca_key = os.environ.get("ALPACA_API_KEY", "")
    alpaca_secret = os.environ.get("ALPACA_API_SECRET", "")

    if not alpaca_key:
        print("Error: ALPACA_API_KEY not set")
        sys.exit(1)

    symbols = universe or DEFAULT_UNIVERSE

    print("=" * 60)
    print("Daily Stock Screener — MOC Strategy")
    print(f"Date: {datetime.now(UTC).strftime('%Y-%m-%d %H:%M UTC')}")
    print(f"Universe: {len(symbols)} stocks")
    print("=" * 60)

    # Fetch market data
    print("\nFetching market data from Alpaca...")
    alpaca = AlpacaClient(api_key=alpaca_key, api_secret=alpaca_secret, paper=True)

    try:
        market_data = await get_market_data(alpaca, symbols)
    finally:
        await alpaca.close()

    if not market_data:
        print("No market data available. Market may be closed.")
        return

    print(f"Got data for {len(market_data)} stocks\n")

    # Pre-filter obvious rejects
    candidates = [
        s for s in market_data if s["is_bullish"] and s["vol_ratio"] >= 0.8 and s["change_pct"] > -2
    ]
    print(f"Pre-filter: {len(candidates)} candidates (bullish, decent volume)\n")

    if not candidates:
        print("No candidates passed pre-filter.")
        return

    # Claude analysis
    if api_key:
        print("Analyzing with Claude...")
        prompt = build_claude_prompt(candidates)
        recommendations = await analyze_with_claude(prompt, api_key)

        if recommendations:
            print(f"\nClaude's Top {min(top_n, len(recommendations))} Picks:")
            print("-" * 60)
            for i, rec in enumerate(recommendations[:top_n], 1):
                print(f"\n#{i} {rec['symbol']} (Score: {rec['score']})")
                print(f"   {rec['reason']}")
                # Find market data for this symbol
                md = next((m for m in market_data if m["symbol"] == rec["symbol"]), None)
                if md:
                    print(
                        f"   Close: ${md['close']:.2f} | Change: {md['change_pct']:+.2f}% | Vol Ratio: {md['vol_ratio']:.1f}x"
                    )
        else:
            print("Claude found no qualifying stocks today.")
    else:
        print("ANTHROPIC_API_KEY not set — showing technical screening only\n")
        # Fallback: simple technical screening
        scored = []
        for s in candidates:
            score = 0
            if s["change_pct"] > 0:
                score += 30
            if s["vol_ratio"] > 1.2:
                score += 25
            if s["is_bullish"]:
                score += 20
            if 3 < s["range_5d_pct"] < 15:
                score += 15
            if s["change_pct"] < 5:
                score += 10
            scored.append((s, score))

        scored.sort(key=lambda x: x[1], reverse=True)
        print(f"Technical Screening Top {top_n}:")
        print("-" * 60)
        for i, (s, score) in enumerate(scored[:top_n], 1):
            print(f"#{i} {s['symbol']} (Score: {score})")
            print(
                f"   Close: ${s['close']:.2f} | Change: {s['change_pct']:+.2f}% | Vol: {s['vol_ratio']:.1f}x"
            )

    print("\n" + "=" * 60)


def main():
    parser = argparse.ArgumentParser(description="Daily stock screener for MOC strategy")
    parser.add_argument("--top", type=int, default=5, help="Number of top picks")
    args = parser.parse_args()
    asyncio.run(main_async(top_n=args.top))


if __name__ == "__main__":
    main()