#!/usr/bin/env python3 """Fetch NYSE trade data from Alpaca and display like an old stock ticker tape.""" import argparse import os import sys from datetime import datetime, timedelta import requests # NYSE-listed stocks (tape A). AAPL, MSFT, INTC etc are Nasdaq-listed (tape C). DEFAULT_SYMBOLS = "JPM,JNJ,V,PG,UNH,HD,BAC,XOM,WMT,DIS,KO,PFE,VZ,MRK,T,GE,F,GM,CVX,CL" BASE_URL = "https://data.alpaca.markets/v2/stocks/trades" def fetch_trades(api_key, api_secret, symbols, start, end, limit=1000, page_token=None, debug=False): headers = { "APCA-API-KEY-ID": api_key, "APCA-API-SECRET-KEY": api_secret, } params = { "symbols": symbols, "start": start, "end": end, "feed": "sip", "limit": limit, "sort": "asc", } if page_token: params["page_token"] = page_token if debug: print(f"[DEBUG] GET {BASE_URL}", file=sys.stderr) print(f"[DEBUG] params: {params}", file=sys.stderr) resp = requests.get(BASE_URL, headers=headers, params=params) if debug: print(f"[DEBUG] HTTP {resp.status_code}", file=sys.stderr) if not resp.ok: print(f"[DEBUG] response body: {resp.text[:500]}", file=sys.stderr) resp.raise_for_status() data = resp.json() if debug: top_keys = list(data.keys()) symbols_returned = list(data.get("trades", {}).keys()) if isinstance(data.get("trades"), dict) else "N/A" trade_count = sum(len(v) for v in data.get("trades", {}).values()) if isinstance(data.get("trades"), dict) else 0 print(f"[DEBUG] response keys: {top_keys}", file=sys.stderr) print(f"[DEBUG] symbols in response: {symbols_returned}", file=sys.stderr) print(f"[DEBUG] total trades returned: {trade_count}", file=sys.stderr) print(f"[DEBUG] next_page_token: {data.get('next_page_token', 'none')}", file=sys.stderr) return data def flatten_and_sort(data): """Flatten the per-symbol trade dict into a single time-sorted list.""" trades = [] for symbol, symbol_trades in data.get("trades", {}).items(): for t in symbol_trades: trades.append({"sym": symbol, **t}) trades.sort(key=lambda t: t["t"]) return trades def format_tape_line(trade): """Format a trade like an old ticker tape: SYM shares price""" sym = trade["sym"] size = trade["s"] price = trade["p"] # Old tickers showed round lots (100s). "2s" meant 200 shares. # A bare price meant 100 shares (1 round lot). if size < 100: vol = f"{size}sh" elif size % 100 == 0: lots = size // 100 vol = f"{lots}s" if lots > 1 else "" else: vol = f"{size}sh" parts = [sym] if vol: parts.append(vol) parts.append(f"{price:.2f}") return " ".join(parts) def main(): parser = argparse.ArgumentParser(description="NYSE ticker tape via Alpaca market data") parser.add_argument("--symbols", default=DEFAULT_SYMBOLS, help=f"Comma-separated symbols (default: {DEFAULT_SYMBOLS})") parser.add_argument("--start", help="Start datetime, YYYY-MM-DD or RFC-3339 (default: previous trading day)") parser.add_argument("--end", help="End datetime (default: same day as start)") parser.add_argument("--limit", type=int, default=1000, help="Max trades to fetch per API call (default: 1000, max: 10000)") parser.add_argument("--all", action="store_true", help="Paginate through ALL trades in the range (many API calls)") parser.add_argument("--tape", default="A", help="Filter by SIP tape: A=NYSE-listed, B=regional/Arca, C=Nasdaq-listed, all=no filter (default: A)") parser.add_argument("--no-filter", action="store_true", help="Show all trades regardless of tape/exchange") parser.add_argument("--raw", action="store_true", help="Output raw CSV instead of tape format: time,symbol,lots,price") parser.add_argument("--debug", action="store_true", help="Print debug info: request URL/params, response structure, exchange codes seen") args = parser.parse_args() api_key = os.environ.get("APCA_API_KEY_ID", "") api_secret = os.environ.get("APCA_API_SECRET_KEY", "") if not api_key or not api_secret: print("Set APCA_API_KEY_ID and APCA_API_SECRET_KEY environment variables.", file=sys.stderr) sys.exit(1) # Default to last Friday if today is weekend, else yesterday if not args.start: today = datetime.now() days_back = 1 if today.weekday() > 0 else 3 # Monday->Friday if today.weekday() == 6: # Sunday days_back = 2 args.start = (today - timedelta(days=days_back)).strftime("%Y-%m-%d") # Default start to market open (9:30 ET = 13:30 UTC) if only date given if "T" not in args.start: args.start = args.start + "T13:30:00Z" if not args.end: # Default end = start day at market close (4pm ET = 20:00 UTC) args.end = args.start[:10] + "T20:00:00Z" tape_filter = None if args.no_filter or args.tape.lower() == "all" else args.tape.upper() if args.debug: print(f"[DEBUG] start={args.start} end={args.end}", file=sys.stderr) print(f"[DEBUG] symbols={args.symbols}", file=sys.stderr) print(f"[DEBUG] tape_filter={tape_filter} (A=NYSE-listed, B=regional, C=Nasdaq)", file=sys.stderr) page_token = None total = 0 exchanges_seen = {} while True: data = fetch_trades(api_key, api_secret, args.symbols, args.start, args.end, args.limit, page_token, debug=args.debug) trades = flatten_and_sort(data) if args.debug: for trade in trades: ex = trade.get("x", "?") tape = trade.get("z", "?") exchanges_seen[ex] = exchanges_seen.get(ex, 0) + 1 exchanges_seen[f"tape:{tape}"] = exchanges_seen.get(f"tape:{tape}", 0) + 1 for trade in trades: if tape_filter and trade.get("z") != tape_filter: continue if trade["s"] < 100: continue lots = trade["s"] // 100 if args.raw: print(f"{trade['t']},{trade['sym']},{lots},{trade['p']:.2f}") else: ts = trade["t"][:19].replace("T", " ") line = format_tape_line(trade) print(f"{ts} {line}") total += 1 page_token = data.get("next_page_token") if not page_token or not args.all: break if args.debug and exchanges_seen: print(f"[DEBUG] exchange codes seen: {exchanges_seen}", file=sys.stderr) print(f"[DEBUG] (exchanges: N=NYSE, Q=Nasdaq, P=Arca, D=FINRA, K=EDGX...)", file=sys.stderr) print(f"[DEBUG] (tapes: A=NYSE-listed, B=regional, C=Nasdaq-listed)", file=sys.stderr) if total == 0: print("No trades found. Try --no-filter or different --tape/symbols/dates.", file=sys.stderr) elif not args.all and data.get("next_page_token"): print(f"\n--- {total} trades shown. More available; use --all to paginate. ---", file=sys.stderr) if __name__ == "__main__": main()