fez5587/momentum-system
1
0 Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases Wiki Activity
Files
df85201c457a855dbb16bf85f7d672202f28518a
momentum-system/tradier_broker/client.py
T
momentum devandClaude Fable 5 1cefb51c63 feat: Tradier broker mirror — config-switch redundancy for Alpaca execution 
tradier_broker/ mirrors every trading call the live loop makes on
AlpacaPaperClient (account, clock, positions, orders, submit incl.
bracket->OTOCO, cancel, replace, close) with identical signatures and
Alpaca-shaped returns, so trading_execution / exit_manager / flatten /
execution / sync run unchanged. Market data always delegates to Alpaca.

- BROKER=tradier (+ TRADIER_* env) swaps execution client in run_live_paper;
  default path untouched
- client_order_id -> tag preserves the lost-submit reconcile
- TradierApiError subclasses AlpacaApiError so the executor's
  rejection-vs-transport fork works as-is
- sandbox-verified: balances/clock/positions/orders, OTOCO preview
  (result:true), place -> tag lookup -> cancel round trip
- 18 unit tests (translation + client paths); executor integration checked
- docs/TRADIER-FALLBACK.md switch runbook

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-26 11:01:50 -05:00

529 lines
21 KiB
Python
Raw Blame History

"""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
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),
"last_equity": _s(b.get("total_equity") or 0),
"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 /markets/quotes
symbols = sorted({str(p["symbol"]) for p in raw if p.get("symbol")})
lasts: dict[str, float] = {}
try:
qp = self._request("GET", "/markets/quotes",
params={"symbols": ",".join(symbols)}) 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}
orders = [o for o in orders if (o.get("symbol") or "").upper() in want]
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:
return translate_order(o)
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)."""
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}")
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)
Reference in new issue View Git Blame Copy Permalink