"""Minimal Tradier REST client mirroring AlpacaPaperClient's trading surface. Stdlib only. Every trading method the live loop touches — account, clock, positions, orders (list / by-client-id), submit (incl. bracket via OTOCO), cancel, replace, close — has the SAME name, signature and Alpaca-shaped return dict as AlpacaPaperClient, so all downstream consumers run unchanged. Key mappings (Tradier -> Alpaca shape): order: exec_quantity->filled_qty, avg_fill_price->filled_avg_price, tag->client_order_id, create_date->submitted_at, leg->legs, price->limit_price, stop_price/stop->stop_price status: open->new, pending->pending_new, submitted->accepted, error->rejected position: cost_basis/quantity->avg_entry_price; current_price/market_value/ unrealized_pl derived from /markets/quotes (Tradier positions carry no mark) account: balances.total_equity->equity, total_cash->cash, margin/cash/pdt buying power -> buying_power bracket: Alpaca order_class=bracket -> Tradier class=otoco (entry + OCO take-profit limit / stop); oto -> class=oto Market data calls delegate to an injected AlpacaPaperClient (Alpaca data keys work without a live brokerage account; Tradier sandbox quotes are 15-min delayed, so they are only used to mark positions). """ from __future__ import annotations import json import logging import threading import time import urllib.error import urllib.parse import urllib.request # importing the alpaca client also installs its process-wide DNS cache, which # covers the Tradier hostnames too (same flaky-DNS host) from alpaca_paper.client import AlpacaApiError, AlpacaPaperClient from tradier_broker.settings import TradierSettings logger = logging.getLogger(__name__) class TradierApiError(AlpacaApiError): """Subclass so AlpacaPaperExecutor's `except AlpacaApiError` (genuine broker rejection vs transport error — the lost-submit reconcile fork) works unchanged with this client injected.""" def __init__(self, status: int, body: str): super().__init__(status, body) # more honest message than the parent's "Alpaca API error" self.args = (f"Tradier API error {status}: {body[:300]}",) # -- pure translation helpers (unit-testable, no network) ----------------- _STATUS_MAP = { "open": "new", "pending": "pending_new", "submitted": "accepted", "partially_filled": "partially_filled", "filled": "filled", "canceled": "canceled", "rejected": "rejected", "expired": "expired", "error": "rejected", "held": "held", "calculated": "accepted", "ok": "accepted", # POST /orders success body carries status=ok } def _s(v) -> str | None: """Tradier returns numbers; Alpaca returns strings. Stringify, None-safe.""" return None if v is None else str(v) def _one_or_many(node, key: str) -> list[dict]: """Tradier wraps lists oddly: {"orders": "null"} when empty, a bare dict for a single item, a list otherwise. Normalize to a list of dicts.""" if not node or node == "null": return [] inner = node.get(key) if isinstance(node, dict) else None if inner is None or inner == "null": return [] return inner if isinstance(inner, list) else [inner] def translate_order(o: dict) -> dict: status = _STATUS_MAP.get(str(o.get("status", "")).lower(), str(o.get("status", "")).lower()) legs = [translate_order(leg) for leg in _one_or_many(o, "leg")] \ if isinstance(o.get("leg"), (list, dict)) else None filled_at = o.get("transaction_date") if status == "filled" else None if legs and not o.get("symbol"): # Tradier OTOCO/OTO PARENT orders carry no top-level symbol/side/qty — # only the legs do. Inherit identity fields from the entry leg (leg[0]) # so symbol-scoped lookups (stop-leg search, naked-stop guard) can see # bracket orders; without this the guard reads a protected position as # naked and force-exits it (2026-07-28 sandbox incident). # Fill fields are NOT inherited — the legs carry the fills, and a # parent echoing them double-counts in journal reconstruction # (storage.journal_backfill.flatten_fills adds parent AND legs). entry_leg = legs[0] o = {**o, "symbol": entry_leg.get("symbol"), "side": o.get("side") or entry_leg.get("side"), "quantity": o.get("quantity") or entry_leg.get("qty"), "create_date": o.get("create_date") or entry_leg.get("submitted_at"), "transaction_date": o.get("transaction_date") or entry_leg.get("updated_at")} return { "id": _s(o.get("id")), "client_order_id": o.get("tag") or "", "symbol": o.get("symbol"), "qty": _s(o.get("quantity") or 0), "filled_qty": _s(o.get("exec_quantity") or 0), "filled_avg_price": _s(o.get("avg_fill_price")), "side": o.get("side"), "type": o.get("type"), "order_type": o.get("type"), "status": status, "limit_price": _s(o.get("price")), "stop_price": _s(o.get("stop_price", o.get("stop"))), "time_in_force": o.get("duration"), "submitted_at": o.get("create_date"), "created_at": o.get("create_date"), "updated_at": o.get("transaction_date"), "filled_at": filled_at, "order_class": "bracket" if o.get("class") == "otoco" else ( "oto" if o.get("class") == "oto" else ""), "legs": legs, "_tradier": {"class": o.get("class"), "status": o.get("status")}, } def translate_position(p: dict, last_price: float | None) -> dict: qty = float(p.get("quantity") or 0) cost = float(p.get("cost_basis") or 0) avg = cost / qty if qty else 0.0 mv = (last_price * qty) if last_price is not None else None upl = (mv - cost) if mv is not None else None return { "symbol": p.get("symbol"), "qty": str(int(qty)) if float(qty).is_integer() else str(qty), "side": "long" if qty >= 0 else "short", "avg_entry_price": str(round(avg, 4)), "cost_basis": str(cost), "current_price": _s(last_price), "market_value": _s(round(mv, 2) if mv is not None else None), "unrealized_pl": _s(round(upl, 2) if upl is not None else None), "unrealized_plpc": _s(round(upl / cost, 6) if upl is not None and cost else None), } def translate_balances(b: dict) -> dict: margin = b.get("margin") or {} cash_bp = b.get("cash") or {} pdt = b.get("pdt") or {} buying_power = ( margin.get("stock_buying_power") or pdt.get("stock_buying_power") or cash_bp.get("cash_available") or b.get("total_cash") or 0 ) return { "equity": _s(b.get("total_equity") or 0), # Tradier has NO prior-session-close equity field. Do NOT fake it with # current equity: the dashboard's realized P&L is the broker equity # delta (equity - last_equity), which then reads $0 forever. Absent # last_equity makes the projection fall back to summing # position_closed events — honest, and correct now that fills/closes # are journaled reliably. "last_equity": None, "cash": _s(b.get("total_cash") or 0), "buying_power": _s(buying_power), "status": "ACTIVE", "account_number": _s(b.get("account_number")), "currency": "USD", } def build_order_form( symbol: str, qty: int, side: str, order_type: str, time_in_force: str, limit_price: float | None, stop_price: float | None, stop_loss_price: float | None, take_profit_price: float | None, client_order_id: str | None, ) -> dict: """Alpaca submit_order args -> Tradier POST /orders form fields.""" duration = "gtc" if time_in_force == "gtc" else "day" exit_side = "sell" if side == "buy" else "buy_to_cover" if stop_loss_price is not None and take_profit_price is not None: # Alpaca bracket == Tradier OTOCO: entry triggers an OCO pair form = { "class": "otoco", "duration": duration, "symbol[0]": symbol, "side[0]": side, "quantity[0]": str(qty), "type[0]": order_type, "symbol[1]": symbol, "side[1]": exit_side, "quantity[1]": str(qty), "type[1]": "limit", "price[1]": str(round(take_profit_price, 2)), "symbol[2]": symbol, "side[2]": exit_side, "quantity[2]": str(qty), "type[2]": "stop", "stop[2]": str(round(stop_loss_price, 2)), } if limit_price is not None: form["price[0]"] = str(round(limit_price, 2)) if stop_price is not None: form["stop[0]"] = str(round(stop_price, 2)) elif stop_loss_price is not None or take_profit_price is not None: # Alpaca oto == Tradier OTO: entry triggers one exit form = { "class": "oto", "duration": duration, "symbol[0]": symbol, "side[0]": side, "quantity[0]": str(qty), "type[0]": order_type, "symbol[1]": symbol, "side[1]": exit_side, "quantity[1]": str(qty), } if limit_price is not None: form["price[0]"] = str(round(limit_price, 2)) if stop_price is not None: form["stop[0]"] = str(round(stop_price, 2)) if take_profit_price is not None: form["type[1]"] = "limit" form["price[1]"] = str(round(take_profit_price, 2)) else: form["type[1]"] = "stop" form["stop[1]"] = str(round(stop_loss_price, 2)) # type: ignore[arg-type] else: form = { "class": "equity", "symbol": symbol, "side": side, "quantity": str(qty), "type": order_type, "duration": duration, } if limit_price is not None: form["price"] = str(round(limit_price, 2)) if stop_price is not None: form["stop"] = str(round(stop_price, 2)) if client_order_id: # Tradier tag: letters/numbers/dashes, <=255 chars — our uuid4 ids fit form["tag"] = client_order_id[:255] return form # -- client ---------------------------------------------------------------- _OPEN_STATUSES = {"new", "pending_new", "accepted", "partially_filled", "held"} class TradierClient: broker_name = "tradier" def __init__( self, settings: TradierSettings | None = None, data_client: AlpacaPaperClient | None = None, timeout: int = 15, max_retries: int = 2, ): self.settings = settings or TradierSettings.from_env() self.data_client = data_client self.timeout = timeout self.max_retries = max_retries # same tiny TTL cache as the Alpaca client: collapses the redundant # account/positions reads the breaker/sync/guard make every pass self._cache: dict = {} self._cache_lock = threading.Lock() self.cache_ttl = 2.5 def _cached(self, key: str, fn): now = time.monotonic() with self._cache_lock: hit = self._cache.get(key) if hit and (now - hit[1]) < self.cache_ttl: return hit[0] val = fn() with self._cache_lock: self._cache[key] = (val, time.monotonic()) return val def _invalidate_cache(self): with self._cache_lock: self._cache.clear() # -- low level --------------------------------------------------------- def _headers(self) -> dict[str, str]: return { "Authorization": f"Bearer {self.settings.access_token}", "Accept": "application/json", "Content-Type": "application/x-www-form-urlencoded", } def _request(self, method: str, path: str, params: dict | None = None, form: dict | None = None): url = f"{self.settings.base_url}{path}" if params: qs = urllib.parse.urlencode( {k: v for k, v in params.items() if v is not None}) url = f"{url}?{qs}" data = urllib.parse.urlencode(form).encode() if form is not None else None req = urllib.request.Request(url, data=data, headers=self._headers(), method=method) for attempt in range(self.max_retries + 1): try: with urllib.request.urlopen(req, timeout=self.timeout) as resp: raw = resp.read().decode() payload = json.loads(raw) if raw else {} # Tradier can return 200 with an errors envelope if isinstance(payload, dict) and payload.get("errors"): raise TradierApiError(resp.status, raw) return payload except urllib.error.HTTPError as exc: # 429/5xx transient — retry with backoff (mirrors Alpaca client) if exc.code in (429, 500, 502, 503, 504) and attempt < self.max_retries: retry_after = exc.headers.get("Retry-After") if exc.headers else None try: delay = float(retry_after) if retry_after else 0.0 except (TypeError, ValueError): delay = 0.0 time.sleep(max(delay, 0.5 * (2 ** attempt))) continue raise TradierApiError(exc.code, exc.read().decode(errors="replace")) from exc except urllib.error.URLError: # transient network/DNS. NOT for POST/PUT: the request may have # LANDED with the response lost — re-sending rests a duplicate # order. Caller reconciles via get_order_by_client_order_id # (same contract as the Alpaca client). if method in ("POST", "PUT") or attempt >= self.max_retries: raise time.sleep(0.4 * (attempt + 1)) raise RuntimeError("request retries exhausted") # unreachable def _acct(self, method: str, path: str = "", **kw): return self._request( method, f"/accounts/{self.settings.account_id}{path}", **kw) # -- trading API (Alpaca-shaped) ---------------------------------------- def get_account(self) -> dict: def fetch(): payload = self._acct("GET", "/balances") or {} return translate_balances(payload.get("balances") or {}) return self._cached("account", fetch) def get_clock(self) -> dict: payload = self._request("GET", "/markets/clock") or {} clock = payload.get("clock") or {} return { "is_open": str(clock.get("state", "")).lower() == "open", "timestamp": clock.get("timestamp"), "next_change": clock.get("next_change"), "state": clock.get("state"), "description": clock.get("description"), } def _positions_fresh(self) -> list[dict]: payload = self._acct("GET", "/positions") or {} raw = _one_or_many(payload.get("positions"), "position") if not raw: return [] # Tradier positions carry no mark. Price them off the Alpaca data # client's REAL-TIME trades when available — the exit manager's # breakeven/trail logic reads current_price, and Tradier sandbox # quotes are 15-min delayed (stale marks = lagging trails). Tradier # /markets/quotes is the fallback (real-time on funded live accounts). symbols = sorted({str(p["symbol"]) for p in raw if p.get("symbol")}) lasts: dict[str, float] = {} if self.data_client is not None: try: for sym, t in (self.data_client.get_latest_trades(symbols) or {}).items(): px = t.get("p") if px is not None: lasts[sym] = float(px) except Exception: # noqa: BLE001 — fall through to Tradier quotes logger.warning("alpaca latest-trades mark failed; falling back to tradier quotes") missing = [s for s in symbols if s not in lasts] if missing: try: qp = self._request("GET", "/markets/quotes", params={"symbols": ",".join(missing)}) or {} for q in _one_or_many(qp.get("quotes"), "quote"): last = q.get("last") if q.get("symbol") and last is not None: lasts[q["symbol"]] = float(last) except Exception: # noqa: BLE001 — quotes down: positions still report, unmarked logger.warning("tradier quotes lookup failed; positions unmarked") return [translate_position(p, lasts.get(str(p.get("symbol") or ""))) for p in raw] def get_positions(self, fresh: bool = False) -> list[dict]: if fresh: val = self._positions_fresh() with self._cache_lock: self._cache["positions"] = (val, time.monotonic()) return val return self._cached("positions", self._positions_fresh) def _orders_raw(self) -> list[dict]: """All orders for the current session, paged, tags included.""" out: list[dict] = [] seen: set = set() for page in range(1, 21): payload = self._acct("GET", "/orders", params={"includeTags": "true", "page": page}) or {} batch = _one_or_many(payload.get("orders"), "order") fresh = [o for o in batch if o.get("id") not in seen] for o in fresh: seen.add(o.get("id")) out.extend(fresh) if not batch or not fresh: break return out def get_orders( self, status: str = "all", limit: int = 500, nested: bool = True, symbols: list[str] | None = None, after: str | None = None, ) -> list[dict]: # Tradier has no server-side status/symbol filters — one paged fetch, # filter here. Legs always come inline (the `nested` arg is accepted # for signature parity and ignored). orders = [translate_order(o) for o in self._orders_raw()] if symbols: want = {s.upper() for s in symbols} def _matches(o: dict) -> bool: if (o.get("symbol") or "").upper() in want: return True # bracket parents/legs: match if ANY leg is on a wanted symbol return any((leg.get("symbol") or "").upper() in want for leg in (o.get("legs") or [])) orders = [o for o in orders if _matches(o)] if status == "open": orders = [o for o in orders if o.get("status") in _OPEN_STATUSES] elif status == "closed": orders = [o for o in orders if o.get("status") not in _OPEN_STATUSES] if after: orders = [o for o in orders if (o.get("submitted_at") or "") > after] return orders[:limit] def get_order_by_client_order_id(self, client_order_id: str) -> dict | None: for o in self._orders_raw(): if (o.get("tag") or "") == client_order_id: out = translate_order(o) # bracket container: surface the ENTRY LEG's fill on the parent # so the cancel-vs-fill reconcile sees executed shares (the # general translation deliberately leaves fills on legs only) legs = out.get("legs") or [] if legs and float(out.get("filled_qty") or 0) == 0: entry_leg = legs[0] if float(entry_leg.get("filled_qty") or 0) > 0: out = {**out, "filled_qty": entry_leg["filled_qty"], "filled_avg_price": entry_leg.get("filled_avg_price"), "status": entry_leg.get("status") or out.get("status")} return out return None def submit_order( self, symbol: str, qty: int, side: str = "buy", order_type: str = "market", time_in_force: str = "day", limit_price: float | None = None, stop_price: float | None = None, stop_loss_price: float | None = None, take_profit_price: float | None = None, client_order_id: str | None = None, ) -> dict: self._invalidate_cache() form = build_order_form(symbol, qty, side, order_type, time_in_force, limit_price, stop_price, stop_loss_price, take_profit_price, client_order_id) payload = self._acct("POST", "/orders", form=form) or {} order = payload.get("order") or {} broker_id = order.get("id") # POST returns only {id, status:"ok"} — echo request fields so the # executor's event payloads are complete without a second GET return { "id": _s(broker_id), "client_order_id": client_order_id or "", "symbol": symbol, "qty": str(qty), "filled_qty": "0", "filled_avg_price": None, "side": side, "type": order_type, "order_type": order_type, "status": _STATUS_MAP.get(str(order.get("status", "")).lower(), "accepted" if broker_id else "rejected"), "limit_price": _s(limit_price), "stop_price": _s(stop_price), "time_in_force": time_in_force, "legs": None, "_tradier": order, } def cancel_order(self, order_id: str) -> None: self._invalidate_cache() self._acct("DELETE", f"/orders/{order_id}") def replace_order( self, order_id: str, stop_price: float | None = None, limit_price: float | None = None, qty: int | None = None, ) -> dict: """PUT /orders/{id} — moves a working order's stop/limit price without cancelling it (keeps OCO protection intact, same contract as Alpaca's PATCH). Tradier cannot change quantity on a working order.""" if qty is not None: raise ValueError("Tradier cannot change qty on a working order; " "cancel and resubmit instead") self._invalidate_cache() form: dict = {} if stop_price is not None: form["stop"] = str(round(stop_price, 2)) if limit_price is not None: form["price"] = str(round(limit_price, 2)) if not form: return {} payload = self._acct("PUT", f"/orders/{order_id}", form=form) or {} order = payload.get("order") or {} return {"id": _s(order.get("id") or order_id), "status": _STATUS_MAP.get(str(order.get("status", "")).lower(), "accepted")} def close_position(self, symbol: str, qty: int | None = None, percentage: float | None = None) -> dict: """Market-sell held shares (Tradier has no close-position endpoint). Cancels the symbol's resting SELL orders (bracket stop/target legs) first — Tradier reserves held shares against open sells, so selling without releasing them rejects with 'more shares than your current long position' (2026-07-28 sandbox incident).""" self._invalidate_cache() held = 0 for p in self.get_positions(fresh=True): if (p.get("symbol") or "").upper() == symbol.upper(): held = int(float(p.get("qty") or 0)) break if held <= 0: raise TradierApiError(404, f"no long position in {symbol}") for o in self.get_orders(status="open", symbols=[symbol]): for node in [o, *(o.get("legs") or [])]: if (node.get("side") == "sell" and node.get("id") and node.get("status") in _OPEN_STATUSES and (node.get("symbol") or "").upper() == symbol.upper()): try: self.cancel_order(node["id"]) except TradierApiError as exc: # already finalized etc. logger.warning("leg cancel before close %s: %s", symbol, exc) sell = held if qty is not None: sell = min(int(qty), held) elif percentage is not None: sell = max(1, int(held * percentage / 100.0)) return self.submit_order(symbol=symbol, qty=sell, side="sell", order_type="market") # -- market data: delegate to Alpaca ------------------------------------ # Data does not require a live brokerage account, and every research / # ingest / discovery path is calibrated to Alpaca feeds — broker failover # must not change the data plane. def _data(self): if self.data_client is None: raise RuntimeError( "TradierClient has no Alpaca data_client; market data stays on " "Alpaca — construct TradierClient(data_client=AlpacaPaperClient())") return self.data_client def get_minute_bars(self, *a, **kw): return self._data().get_minute_bars(*a, **kw) def get_latest_trades(self, *a, **kw): return self._data().get_latest_trades(*a, **kw) def get_latest_quotes(self, *a, **kw): return self._data().get_latest_quotes(*a, **kw) def get_most_actives(self, *a, **kw): return self._data().get_most_actives(*a, **kw) def get_movers(self, *a, **kw): return self._data().get_movers(*a, **kw) def get_news(self, *a, **kw): return self._data().get_news(*a, **kw) def get_daily_bars(self, *a, **kw): return self._data().get_daily_bars(*a, **kw) def get_assets(self, *a, **kw): return self._data().get_assets(*a, **kw)