"""Protocol-preserving retry wrappers for read-only adapters.
**These classes implement the read Protocols and nothing else, on purpose.**
`RetryingBroker` satisfies `BrokerAdapter` but not `OrderPlacingBroker`, so it
cannot be handed to `OrderExecutor` and cannot be made to retry a submission.
That matters more than it might look: a timeout on `submit_limit_buy` is not
evidence the order failed to reach Alpaca, so a retry can place a second one.
This project has already paid for a duplicate order once, on a live paper
account, and a comment asking future code not to retry writes is weaker than an
object that has no write to retry.
There is also no `__getattr__` passthrough. A wrapper that forwarded unknown
attributes would re-expose the wrapped broker's `submit_limit_buy` and undo the
whole guarantee, silently.
**What is not covered here.** Broker reads made *inside* a cycle
(`run_once` reads positions, orders, and the account) keep the raw
order-placing broker, because a read-only wrapper cannot be passed where an
`OrderPlacingBroker` is required. Those are covered by the daemon's cycle-level
catch-all and its backoff instead — one failed read costs one cycle, not the
process. Market data is the flakiest dependency and *is* wrapped, so a yfinance
blip is absorbed inside the cycle rather than discarding it.
"""
import logging
import time
from collections.abc import Callable
from datetime import date
from functools import partial
from trader.brokers.base import BrokerAdapter
from trader.domain import Account, Bar, MarketClock, Position, Quote, TradableAsset
from trader.errors import ConfigError, NonRetryableMarketDataError
from trader.marketdata.base import MarketDataProvider
__all__ = ["RetryingBroker", "RetryingMarketData"]
_logger = logging.getLogger("trader.resilience")
# Waiting does not fix these, and retrying one delays the fatal exit while
# logging the same permanent problem three times. `ConfigError` covers
# `UnsafeConfigError` and `MissingCredentialsError`, which is the safety gate
# refusing — never something to paper over with a sleep.
# `NonRetryableMarketDataError` (issue #99) is content, not transport: a
# malformed bar comes back identical on attempt 2 and 3, and even 17 minutes
# later, so the 2s+4s backoff only delays the identical failure it already
# knows about. Plain `MarketDataError` is deliberately still retried here --
# many of its causes (a timeout, a rate limit) genuinely can succeed on a
# second attempt, so this only exempts the narrower, proven-deterministic
# subclass, never the whole class.
_NEVER_RETRIED = (ConfigError, NonRetryableMarketDataError)
class _Retrier:
"""Shared retry mechanics. Not exported; both wrappers compose it."""
def __init__(
self,
*,
attempts: int,
base_seconds: int,
sleep: Callable[[float], None] = time.sleep,
) -> None:
if attempts < 1:
raise ValueError(f"attempts must be at least 1, got {attempts}.")
self._attempts = attempts
self._base_seconds = base_seconds
self._sleep = sleep
def call(self, name: str, operation: Callable[[], object]) -> object:
"""Run `operation`, retrying transient failures with a growing delay."""
last: Exception
for attempt in range(1, self._attempts + 1):
try:
return operation()
except _NEVER_RETRIED:
raise
except Exception as exc: # noqa: BLE001 - adapters raise many types
last = exc
if attempt == self._attempts:
break
# No sleep after the final attempt: the caller's own error
# handling should not be delayed by a wait that buys nothing.
delay = self._base_seconds * (2 ** (attempt - 1))
_logger.warning(
"%s failed (attempt %d of %d): %s; retrying in %ss",
name,
attempt,
self._attempts,
exc,
delay,
)
self._sleep(delay)
_logger.error("%s failed after %d attempts: %s", name, self._attempts, last)
raise last
[docs]
class RetryingBroker:
"""`BrokerAdapter` with bounded retry. Holds no order-placing method."""
def __init__(
self,
broker: BrokerAdapter,
*,
attempts: int,
base_seconds: int,
sleep: Callable[[float], None] = time.sleep,
) -> None:
self._broker = broker
self._retrier = _Retrier(
attempts=attempts, base_seconds=base_seconds, sleep=sleep
)
[docs]
def get_account(self) -> Account:
"""The account summary, retried on transient failure."""
return self._retrier.call("get_account", self._broker.get_account) # type: ignore[return-value]
[docs]
def get_positions(self) -> list[Position]:
"""Open positions, retried on transient failure."""
return self._retrier.call("get_positions", self._broker.get_positions) # type: ignore[return-value]
[docs]
def get_clock(self) -> MarketClock:
"""The market clock, retried on transient failure."""
return self._retrier.call("get_clock", self._broker.get_clock) # type: ignore[return-value]
[docs]
def get_asset(self, symbol: str) -> TradableAsset | None:
"""Asset metadata, retried on transient failure."""
return self._retrier.call( # type: ignore[return-value]
"get_asset", partial(self._broker.get_asset, symbol)
)
[docs]
class RetryingMarketData:
"""`MarketDataProvider` with bounded retry.
Wrapped where the broker's in-cycle reads are not, because this one is
genuinely flaky: yfinance is an unofficial scrape with no availability
guarantee, and losing a cycle to one failed bar fetch is avoidable.
"""
def __init__(
self,
provider: MarketDataProvider,
*,
attempts: int,
base_seconds: int,
sleep: Callable[[float], None] = time.sleep,
) -> None:
self._provider = provider
self._retrier = _Retrier(
attempts=attempts, base_seconds=base_seconds, sleep=sleep
)
[docs]
def get_quote(self, symbol: str) -> Quote:
"""The latest quote, retried on transient failure."""
return self._retrier.call( # type: ignore[return-value]
"get_quote", partial(self._provider.get_quote, symbol)
)
[docs]
def get_history(
self, symbol: str, start: date, end: date, interval: str | None = None
) -> list[Bar]:
"""Historical bars, retried on transient failure.
`interval` defaults to `None` here rather than to `"1d"` so an omitted
argument reaches the provider omitted, and the provider's own default
stays the single definition of it. Duplicating `"1d"` here would mean
two places to change it, one of which nobody would remember.
"""
arguments = (
(symbol, start, end)
if interval is None
else (
symbol,
start,
end,
interval,
)
)
return self._retrier.call( # type: ignore[return-value]
"get_history", partial(self._provider.get_history, *arguments)
)