#!/usr/bin/env python3 """Trading Arena read-only Alpaca paper connector. Python 3 standard library only.""" from __future__ import annotations import datetime as dt import hashlib import json import os import signal import ssl import sys import time import urllib.error import urllib.parse import urllib.request from typing import Any VERSION = "0.1.0" PAPER_ORIGIN = "https://paper-api.alpaca.markets" USER_AGENT = f"trading-arena-connector/{VERSION}" STOP = False class ConnectorError(Exception): """An actionable connector error safe to show to a participant.""" def utc_now() -> str: return dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z") def env(name: str) -> str: value = os.environ.get(name, "").strip() if not value: raise ConnectorError(f"Missing {name}. See the participation guide in README.md.") return value def number(value: Any, field: str) -> float: try: return float(value) except (TypeError, ValueError) as exc: raise ConnectorError(f"Alpaca returned an invalid {field}") from exc class Http: def __init__(self, timeout: int = 15): self.timeout = timeout self.context = ssl.create_default_context() def json(self, url: str, headers: dict[str, str], method: str = "GET", data: Any = None) -> Any: encoded = None if data is None else json.dumps(data, separators=(",", ":")).encode() request = urllib.request.Request(url, data=encoded, method=method, headers={"User-Agent": USER_AGENT, "Accept": "application/json", **headers}) if encoded is not None: request.add_header("Content-Type", "application/json") try: with urllib.request.urlopen(request, timeout=self.timeout, context=self.context) as response: raw = response.read(2_000_001) if len(raw) > 2_000_000: raise ConnectorError("Server response was unexpectedly large") return json.loads(raw or b"{}") except urllib.error.HTTPError as exc: detail = "" try: detail = json.loads(exc.read(8192)).get("error") or "" except Exception: pass if exc.code in (401, 403): raise ConnectorError(f"Authentication rejected by {urllib.parse.urlsplit(url).hostname}") from exc raise ConnectorError(f"HTTP {exc.code} from {urllib.parse.urlsplit(url).hostname}{': ' + detail if detail else ''}") from exc except (urllib.error.URLError, TimeoutError) as exc: raise ConnectorError(f"Could not reach {urllib.parse.urlsplit(url).hostname}: {exc.reason if hasattr(exc, 'reason') else exc}") from exc except json.JSONDecodeError as exc: raise ConnectorError(f"Invalid response from {urllib.parse.urlsplit(url).hostname}") from exc class Connector: def __init__(self) -> None: self.arena = env("ARENA_URL").rstrip("/") self.connector_key = env("ARENA_CONNECTOR_KEY") self.alpaca_key = env("APCA_API_KEY_ID") self.alpaca_secret = env("APCA_API_SECRET_KEY") self.alpaca = os.environ.get("APCA_API_BASE_URL", PAPER_ORIGIN).rstrip("/") parsed = urllib.parse.urlsplit(self.alpaca) if parsed.scheme != "https" or parsed.hostname != "paper-api.alpaca.markets" or parsed.path not in ("", "/"): raise ConnectorError(f"Refusing Alpaca endpoint {self.alpaca!r}; only {PAPER_ORIGIN} is allowed") arena_parsed = urllib.parse.urlsplit(self.arena) if arena_parsed.scheme != "https" and arena_parsed.hostname not in ("localhost", "127.0.0.1"): raise ConnectorError("ARENA_URL must use HTTPS (except localhost development)") self.http = Http(int(os.environ.get("ARENA_HTTP_TIMEOUT", "15"))) self.last_fill_after: str | None = None @property def alpaca_headers(self) -> dict[str, str]: return {"APCA-API-KEY-ID": self.alpaca_key, "APCA-API-SECRET-KEY": self.alpaca_secret} @property def arena_headers(self) -> dict[str, str]: return {"Authorization": f"Bearer {self.connector_key}"} def alpaca_get(self, path: str, query: dict[str, str] | None = None) -> Any: url = f"{self.alpaca}{path}" if query: url += "?" + urllib.parse.urlencode(query) return self.http.json(url, self.alpaca_headers) def arena_request(self, path: str, method: str = "GET", data: Any = None) -> Any: return self.http.json(f"{self.arena}/api/connector{path}", self.arena_headers, method, data) def observe(self) -> tuple[dict[str, Any], list[dict[str, Any]], int]: account = self.alpaca_get("/v2/account") positions = self.alpaca_get("/v2/positions") orders = self.alpaca_get("/v2/orders", {"status": "open", "limit": "500"}) normalized = [{ "symbol": str(p["symbol"]), "asset_class": str(p.get("asset_class", "unknown")), "side": str(p.get("side", "long")), "quantity": number(p["qty"], "position quantity"), "market_value": number(p["market_value"], "position market value"), "average_entry_price": number(p["avg_entry_price"], "average entry price"), "unrealized_pl": number(p.get("unrealized_pl", 0), "unrealized P/L"), } for p in positions] return account, normalized, len(orders) def fills(self) -> list[dict[str, Any]]: query = {"direction": "asc", "page_size": "100"} if self.last_fill_after: query["after"] = self.last_fill_after activities = self.alpaca_get("/v2/account/activities/FILL", query) fills = [] for fill in activities: executed_at = fill.get("transaction_time") or fill.get("date") if not executed_at: continue fills.append({"id": str(fill["id"]), "symbol": str(fill["symbol"]), "side": str(fill["side"]), "quantity": number(fill["qty"], "fill quantity"), "price": number(fill["price"], "fill price"), "executed_at": executed_at}) self.last_fill_after = max(self.last_fill_after or executed_at, executed_at) return fills def cycle(self) -> int: assignment = self.arena_request("/assignment") interval = max(10, min(3600, int(assignment.get("reporting_interval_seconds", 300)))) tournament = assignment.get("tournament") if not tournament: print(f"{utc_now()} IDLE — no current tournament") return interval observed_at = utc_now() account, positions, open_orders = self.observe() heartbeat = {"observed_at": observed_at, "paper": True, "equity": number(account["equity"], "equity"), "cash": number(account["cash"], "cash"), "open_positions": len(positions), "open_orders": open_orders} state = self.arena_request("/heartbeat", "POST", heartbeat) label = str(state.get("state", "unknown")).upper() print(f"{observed_at} {label} — {tournament['name']} — equity {heartbeat['equity']:.2f}" + (f" — {state['reason']}" if state.get("reason") and label != "ACTIVE" else "")) if label == "ACTIVE": fingerprint = hashlib.sha256(f"{tournament['id']}|{observed_at}|{account['equity']}|{account['cash']}".encode()).hexdigest() self.arena_request("/snapshot", "POST", {"idempotency_key": fingerprint, "observed_at": observed_at, "equity": heartbeat["equity"], "cash": heartbeat["cash"], "buying_power": number(account.get("buying_power", 0), "buying power"), "positions": positions, "executed_trades": self.fills()}) return max(10, min(3600, int(state.get("reporting_interval_seconds", interval)))) def run(self) -> None: print(f"Trading Arena connector {VERSION} — read-only Alpaca paper mode") failures = 0 while not STOP: try: interval = self.cycle(); failures = 0 except ConnectorError as exc: failures += 1; interval = min(300, 5 * (2 ** min(failures - 1, 6))) print(f"{utc_now()} ERROR — {exc}; retrying in {interval}s", file=sys.stderr) deadline = time.monotonic() + interval while not STOP and time.monotonic() < deadline: time.sleep(min(1, deadline - time.monotonic())) def stop(_signum: int, _frame: Any) -> None: global STOP STOP = True def main() -> int: signal.signal(signal.SIGINT, stop) if hasattr(signal, "SIGTERM"): signal.signal(signal.SIGTERM, stop) try: Connector().run() print("Connector stopped.") return 0 except ConnectorError as exc: print(f"CONFIGURATION ERROR — {exc}", file=sys.stderr) return 2 if __name__ == "__main__": raise SystemExit(main())