""" Smoke test de la fontaneria Alpaca paper. Confirma end-to-end que Leonex puede: 1. Conectar con tu cuenta paper. 2. Enviar una orden BUY MARKET real (a paper). 3. Ver la posicion abierta resultante. 4. (opcional) Cerrar la posicion automaticamente. Es un test de bajo riesgo: usa 1 sola accion de SPY (~$580 sobre tus $100K paper = 0.6%). Como es paper trading, no hay riesgo de capital real. Si el mercado US esta cerrado (fuera de 9:30-16:00 ET / 15:30-22:00 hora ES), la orden se queda "accepted" y se ejecuta al abrir. La validacion del flujo funciona igual. Uso: python tools/smoke_test_alpaca.py --round-trip # abre 1 SPY y luego lo cierra (recomendado) python tools/smoke_test_alpaca.py --open # abre 1 SPY (deja la posicion abierta) python tools/smoke_test_alpaca.py --close SPY # cierra la posicion de SPY python tools/smoke_test_alpaca.py --status # solo lee estado de cuenta y posiciones python tools/smoke_test_alpaca.py --ticker AAPL --round-trip # cambiar ticker / qty python tools/smoke_test_alpaca.py --qty 2 --open Requiere que .env tenga ALPACA_API_KEY y ALPACA_SECRET_KEY validas. """ from __future__ import annotations import argparse import sys import time from pathlib import Path # Permitir importar agents/alpaca_client.py ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "agents")) from alpaca_client import ( # noqa: E402 fetch_snapshot, submit_market_order, get_latest_quote, ) def is_market_open() -> tuple[bool, str]: """Pregunta a Alpaca si el mercado esta abierto ahora. Devuelve (abierto, descripcion).""" import os api_key = os.environ.get("ALPACA_API_KEY", "").strip() secret = os.environ.get("ALPACA_SECRET_KEY", "").strip() if not api_key or not secret: return False, "sin credenciales" try: from alpaca.trading.client import TradingClient client = TradingClient(api_key, secret, paper=True) clock = client.get_clock() return bool(clock.is_open), f"abierto={clock.is_open}, next_open={clock.next_open}" except Exception as exc: return False, f"error consultando clock: {exc!r}" def print_account(snap) -> None: if not snap.configured: print(f"[X] Alpaca no configurado: {snap.error}") return if snap.error and not snap.account: print(f"[X] Error Alpaca: {snap.error}") return a = snap.account print(f"[OK] Cuenta {'PAPER' if a.is_paper else 'LIVE'} {a.account_number}") print(f" equity=${a.equity:,.2f} | cash=${a.cash:,.2f} | " f"buying_power=${a.buying_power:,.2f}") print(f" posiciones abiertas: {len(snap.positions)}") for p in snap.positions: print(f" {p.symbol:<6} {p.qty:>7.2f} @ ${p.avg_entry_price:.2f} " f"-> ${p.current_price:.2f} PnL ${p.unrealized_pl:+.2f} " f"({p.unrealized_plpc*100:+.2f}%)") def open_position(ticker: str, qty: float) -> bool: quote = get_latest_quote(ticker) if quote: price = quote["mid"] or max(quote.get("ask", 0), quote.get("bid", 0)) notional = price * qty print(f" Quote {ticker}: bid=${quote['bid']:.2f} ask=${quote['ask']:.2f} -> notional ~${notional:.2f}") print(f" Enviando BUY {qty} {ticker} (market, TIF=day)...") result = submit_market_order(ticker, qty, side="buy", time_in_force="day") if result["ok"]: print(f" [OK] Orden enviada. id={result['order_id']} status={result['status']}") return True print(f" [X] Fallo enviando orden: {result['error']}") return False def close_position(ticker: str, qty: float) -> bool: print(f" Enviando SELL {qty} {ticker} (market, TIF=day) para cerrar...") result = submit_market_order(ticker, qty, side="sell", time_in_force="day") if result["ok"]: print(f" [OK] Orden de cierre enviada. id={result['order_id']} status={result['status']}") return True print(f" [X] Fallo cerrando posicion: {result['error']}") return False def main() -> int: parser = argparse.ArgumentParser(description="Smoke test fontaneria Alpaca paper") parser.add_argument("--ticker", type=str, default="SPY") parser.add_argument("--qty", type=float, default=1.0, help="Cantidad de acciones (default 1)") parser.add_argument("--open", action="store_true", help="Solo abre la posicion") parser.add_argument("--close", type=str, default=None, help="Cierra la posicion del ticker indicado (ej. --close SPY)") parser.add_argument("--round-trip", action="store_true", help="Abre la posicion y la cierra tras --wait segundos") parser.add_argument("--wait", type=int, default=8, help="Segundos a esperar entre BUY y SELL en --round-trip") parser.add_argument("--status", action="store_true", help="Solo muestra estado de cuenta y posiciones") args = parser.parse_args() sep = "=" * 60 print(sep) print(f"Leonex - Smoke test Alpaca | ticker={args.ticker} qty={args.qty}") print(sep) snap = fetch_snapshot() print_account(snap) print() if args.status: return 0 if not snap.configured or not snap.account: print("[X] Conexion con Alpaca no operativa. Revisa .env y vuelve a probar.") return 1 open_state, clock_desc = is_market_open() print(f"[CLOCK] Mercado US: {clock_desc}") if not open_state: print(" Mercado cerrado: la orden se aceptara pero se ejecutara al abrir.") print() if args.close: ok = close_position(args.close, args.qty) print() print("Refrescando estado tras cierre...") time.sleep(2) print_account(fetch_snapshot()) return 0 if ok else 1 # --open o --round-trip if not (args.open or args.round_trip): print("[i] Sin accion. Usa --open, --close TICKER, --round-trip o --status.") return 0 ok_open = open_position(args.ticker, args.qty) if not ok_open: return 1 if args.round_trip: print() print(f"[WAIT] Esperando {args.wait}s para que la BUY se procese...") time.sleep(args.wait) print() # Solo cerrar si el mercado esta abierto — si no, la BUY no llenara # todavia y la SELL fallara porque no hay posicion if open_state: ok_close = close_position(args.ticker, args.qty) print() print("Refrescando estado tras round-trip...") time.sleep(2) print_account(fetch_snapshot()) return 0 if ok_close else 1 else: print("[!] Mercado cerrado: la BUY esta encolada pero no llenada,") print(" asi que NO envio SELL ahora (no hay nada que vender).") print(" Cuando abra el mercado y la BUY se llene, ejecuta:") print(f" python tools/smoke_test_alpaca.py --close {args.ticker} --qty {args.qty}") return 0 print() print("Refrescando estado tras BUY...") time.sleep(3) print_account(fetch_snapshot()) return 0 if __name__ == "__main__": raise SystemExit(main())