File size: 1,780 Bytes
c174e7c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 | """DeepQuant — Alpaca client. PHASE 0: read-only.
Safety by construction: the base endpoint must be a PAPER host or every call refuses.
No order-placing functions exist in this phase. Later phases add execute() here, guardrailed.
"""
import json
import os
import ssl
import urllib.error
import urllib.parse
import urllib.request
try:
import certifi
_CTX = ssl.create_default_context(cafile=certifi.where())
except Exception:
_CTX = ssl.create_default_context()
_PAPER_HOST = "https://paper-api.alpaca.markets"
def _base():
ep = (os.environ.get("ALPACA_PAPER_ENDPOINT") or _PAPER_HOST).rstrip("/")
if ep.endswith("/v2"):
ep = ep[:-3].rstrip("/")
if "paper" not in ep: # hard refusal of any non-paper endpoint
raise RuntimeError("DeepQuant is PAPER-ONLY — ALPACA_PAPER_ENDPOINT must point at the paper API host.")
return ep
def _headers():
k, s = os.environ.get("ALPACA_API_KEY"), os.environ.get("ALPACA_SECRET_KEY")
if not (k and s):
raise RuntimeError("ALPACA_API_KEY / ALPACA_SECRET_KEY not set.")
return {"APCA-API-KEY-ID": k, "APCA-API-SECRET-KEY": s}
def _get(path, params=None):
url = _base() + path
if params:
url += "?" + urllib.parse.urlencode(params)
req = urllib.request.Request(url, headers=_headers())
with urllib.request.urlopen(req, timeout=25, context=_CTX) as r:
return json.load(r)
def get_account():
return _get("/v2/account")
def get_positions():
return _get("/v2/positions")
def get_clock():
return _get("/v2/clock")
def get_portfolio_history(period="1M", timeframe="1D"):
return _get("/v2/account/portfolio/history",
{"period": period, "timeframe": timeframe, "extended_hours": "false"})
|