Fix slow loop passes: cache account/positions + fewer broker calls
The single-threaded passes (and the trigger thread) ballooned to 60-120s. Cause: flaky WSL DNS makes ~1 in 5 Alpaca calls take ~5s to resolve, and each pass fired ~20 calls (breaker, sync, guard, exit manager all re-reading account/positions/orders). Fixes: - client: 2.5s TTL cache for get_account/get_positions (invalidated on any order mutation) — collapses the redundant per-pass reads to one. - exit manager: fetch the (heavy) status=all orders snapshot ONCE per pass and reuse it for every position's stop-leg + entry-time lookup (was 2 calls/pos). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
1 parent
45d5484dd7
commit
52b43245d5
2 files changed
+51
-16
No files matched your search
+31
-2
@@ -8,6 +8,8 @@ from __future__ import annotations
|
|||||||
|
|
||||||
import json
|
import json
|
||||||
import logging
|
import logging
|
||||||
|
import threading
|
||||||
|
import time
|
||||||
import urllib.error
|
import urllib.error
|
||||||
import urllib.parse
|
import urllib.parse
|
||||||
import urllib.request
|
import urllib.request
|
||||||
@@ -32,6 +34,29 @@ class AlpacaPaperClient:
|
|||||||
# retry transient network/DNS errors (not HTTP 4xx/5xx) a couple times;
|
# retry transient network/DNS errors (not HTTP 4xx/5xx) a couple times;
|
||||||
# DNS failures fail fast, so this smooths brief blips without much latency
|
# DNS failures fail fast, so this smooths brief blips without much latency
|
||||||
self.max_retries = max_retries
|
self.max_retries = max_retries
|
||||||
|
# tiny TTL cache for hot read-only endpoints (account/positions). The
|
||||||
|
# breaker, sync, guard and exit manager each read these every pass; on a
|
||||||
|
# flaky-DNS host every call risks a multi-second resolve, so collapsing
|
||||||
|
# the redundant reads within a pass is the difference between ~2s and
|
||||||
|
# ~60s loop passes. Short TTL keeps risk checks effectively current.
|
||||||
|
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() # network call OUTSIDE the lock
|
||||||
|
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 -------------------------------------------------------
|
# -- low level -------------------------------------------------------
|
||||||
|
|
||||||
@@ -87,10 +112,10 @@ class AlpacaPaperClient:
|
|||||||
# -- trading API -------------------------------------------------------
|
# -- trading API -------------------------------------------------------
|
||||||
|
|
||||||
def get_account(self) -> dict:
|
def get_account(self) -> dict:
|
||||||
return self._trading("GET", "/account")
|
return self._cached("account", lambda: self._trading("GET", "/account"))
|
||||||
|
|
||||||
def get_positions(self) -> list[dict]:
|
def get_positions(self) -> list[dict]:
|
||||||
return self._trading("GET", "/positions")
|
return self._cached("positions", lambda: self._trading("GET", "/positions"))
|
||||||
|
|
||||||
def get_orders(
|
def get_orders(
|
||||||
self, status: str = "all", limit: int = 100, nested: bool = True
|
self, status: str = "all", limit: int = 100, nested: bool = True
|
||||||
@@ -114,6 +139,7 @@ class AlpacaPaperClient:
|
|||||||
take_profit_price: float | None = None,
|
take_profit_price: float | None = None,
|
||||||
client_order_id: str | None = None,
|
client_order_id: str | None = None,
|
||||||
) -> dict:
|
) -> dict:
|
||||||
|
self._invalidate_cache()
|
||||||
body: dict = {
|
body: dict = {
|
||||||
"symbol": symbol,
|
"symbol": symbol,
|
||||||
"qty": str(qty),
|
"qty": str(qty),
|
||||||
@@ -136,6 +162,7 @@ class AlpacaPaperClient:
|
|||||||
return self._trading("POST", "/orders", body=body)
|
return self._trading("POST", "/orders", body=body)
|
||||||
|
|
||||||
def cancel_order(self, order_id: str) -> None:
|
def cancel_order(self, order_id: str) -> None:
|
||||||
|
self._invalidate_cache()
|
||||||
self._trading("DELETE", f"/orders/{order_id}")
|
self._trading("DELETE", f"/orders/{order_id}")
|
||||||
|
|
||||||
def replace_order(
|
def replace_order(
|
||||||
@@ -146,6 +173,7 @@ class AlpacaPaperClient:
|
|||||||
trail it — WITHOUT cancelling it. Cancelling a bracket leg tears down the
|
trail it — WITHOUT cancelling it. Cancelling a bracket leg tears down the
|
||||||
OCO and leaves the position naked (the bug we already fixed once); a
|
OCO and leaves the position naked (the bug we already fixed once); a
|
||||||
replace keeps the protection intact and just changes the price."""
|
replace keeps the protection intact and just changes the price."""
|
||||||
|
self._invalidate_cache()
|
||||||
body: dict = {}
|
body: dict = {}
|
||||||
if stop_price is not None:
|
if stop_price is not None:
|
||||||
body["stop_price"] = str(round(stop_price, 2))
|
body["stop_price"] = str(round(stop_price, 2))
|
||||||
@@ -159,6 +187,7 @@ class AlpacaPaperClient:
|
|||||||
|
|
||||||
def close_position(self, symbol: str, qty: int | None = None,
|
def close_position(self, symbol: str, qty: int | None = None,
|
||||||
percentage: float | None = None) -> dict:
|
percentage: float | None = None) -> dict:
|
||||||
|
self._invalidate_cache()
|
||||||
params = {}
|
params = {}
|
||||||
if qty is not None:
|
if qty is not None:
|
||||||
params["qty"] = str(int(qty))
|
params["qty"] = str(int(qty))
|
||||||
|
|||||||
+20
-14
@@ -56,16 +56,19 @@ class LiveExitManager:
|
|||||||
_ACTIVE = {"held", "new", "accepted", "pending_new",
|
_ACTIVE = {"held", "new", "accepted", "pending_new",
|
||||||
"accepted_for_bidding", "partially_filled"}
|
"accepted_for_bidding", "partially_filled"}
|
||||||
|
|
||||||
def _stop_leg(self, symbol: str):
|
def _all_orders(self):
|
||||||
|
try:
|
||||||
|
return self.client.get_orders(status="all", limit=100, nested=True)
|
||||||
|
except Exception: # noqa: BLE001
|
||||||
|
return None
|
||||||
|
|
||||||
|
def _stop_leg(self, symbol: str, orders):
|
||||||
"""(order_id, stop_price) of the live sell-stop protecting symbol.
|
"""(order_id, stop_price) of the live sell-stop protecting symbol.
|
||||||
|
|
||||||
Queries status="all" because the protective leg is a 'held' child of the
|
Uses pre-fetched ``orders`` (status="all" — the protective leg is a
|
||||||
filled entry order — status="open" wouldn't return the filled parent and
|
'held' child of the FILLED entry, which status="open" wouldn't surface).
|
||||||
so wouldn't surface the leg.
|
|
||||||
"""
|
"""
|
||||||
try:
|
if orders is None:
|
||||||
orders = self.client.get_orders(status="all", limit=100, nested=True)
|
|
||||||
except Exception: # noqa: BLE001
|
|
||||||
return (None, None)
|
return (None, None)
|
||||||
for o in orders:
|
for o in orders:
|
||||||
for c in [o, *(o.get("legs") or [])]:
|
for c in [o, *(o.get("legs") or [])]:
|
||||||
@@ -77,11 +80,9 @@ class LiveExitManager:
|
|||||||
return (c.get("id"), float(sp) if sp else None)
|
return (c.get("id"), float(sp) if sp else None)
|
||||||
return (None, None)
|
return (None, None)
|
||||||
|
|
||||||
def _entry_time(self, symbol: str):
|
def _entry_time(self, symbol: str, orders):
|
||||||
"""UTC-naive fill time of the most recent FILLED buy entry for symbol."""
|
"""UTC-naive fill time of the most recent FILLED buy entry for symbol."""
|
||||||
try:
|
if orders is None:
|
||||||
orders = self.client.get_orders(status="all", limit=100, nested=True)
|
|
||||||
except Exception: # noqa: BLE001
|
|
||||||
return None
|
return None
|
||||||
best = None
|
best = None
|
||||||
for o in orders:
|
for o in orders:
|
||||||
@@ -121,6 +122,11 @@ class LiveExitManager:
|
|||||||
for sym in list(self._tracked): # stop tracking closed names
|
for sym in list(self._tracked): # stop tracking closed names
|
||||||
if sym not in held:
|
if sym not in held:
|
||||||
self._tracked.pop(sym, None)
|
self._tracked.pop(sym, None)
|
||||||
|
if not held:
|
||||||
|
return []
|
||||||
|
# fetch the orders snapshot ONCE per pass (status="all" is the heavy call;
|
||||||
|
# reuse it for every position's stop-leg + entry-time lookups)
|
||||||
|
orders = self._all_orders()
|
||||||
|
|
||||||
actions: list[str] = []
|
actions: list[str] = []
|
||||||
for sym, p in held.items():
|
for sym, p in held.items():
|
||||||
@@ -134,12 +140,12 @@ class LiveExitManager:
|
|||||||
if entry <= 0:
|
if entry <= 0:
|
||||||
continue
|
continue
|
||||||
if sym not in self._tracked:
|
if sym not in self._tracked:
|
||||||
_leg, cur_stop = self._stop_leg(sym)
|
_leg, cur_stop = self._stop_leg(sym, orders)
|
||||||
if cur_stop is None or cur_stop >= entry:
|
if cur_stop is None or cur_stop >= entry:
|
||||||
continue # no usable protective stop yet (or above entry)
|
continue # no usable protective stop yet (or above entry)
|
||||||
# trail from the ACTUAL entry fill time (so high-water/R reflect
|
# trail from the ACTUAL entry fill time (so high-water/R reflect
|
||||||
# the move since entry, not since we first noticed the position)
|
# the move since entry, not since we first noticed the position)
|
||||||
entry_ts = self._entry_time(sym) or _last_ts(bars)
|
entry_ts = self._entry_time(sym, orders) or _last_ts(bars)
|
||||||
self._tracked[sym] = _Tracked(entry, cur_stop, entry_ts, cur_stop)
|
self._tracked[sym] = _Tracked(entry, cur_stop, entry_ts, cur_stop)
|
||||||
tr = self._tracked[sym]
|
tr = self._tracked[sym]
|
||||||
|
|
||||||
@@ -175,7 +181,7 @@ class LiveExitManager:
|
|||||||
logger.warning("scale-out failed %s: %s", sym, exc)
|
logger.warning("scale-out failed %s: %s", sym, exc)
|
||||||
|
|
||||||
# (3) ratchet the protective stop UP (breakeven / trail) via replace
|
# (3) ratchet the protective stop UP (breakeven / trail) via replace
|
||||||
leg_id, cur_stop = self._stop_leg(sym)
|
leg_id, cur_stop = self._stop_leg(sym, orders)
|
||||||
if cur_stop is not None and d.desired_stop > cur_stop + 0.01:
|
if cur_stop is not None and d.desired_stop > cur_stop + 0.01:
|
||||||
last_px = float(p.get("current_price") or 0.0)
|
last_px = float(p.get("current_price") or 0.0)
|
||||||
if last_px > 0 and d.desired_stop >= last_px:
|
if last_px > 0 and d.desired_stop >= last_px:
|
||||||
|
|||||||
Reference in new issue
Block a user