py_alpaca_api.http.feed_manager

Attributes

logger

Exceptions

APIRequestError	Raised when an API request fails.
ValidationError	Raised when input validation fails.

Classes

FeedType	Available data feed types.
SubscriptionLevel	User subscription levels.
FeedConfig	Configuration for feed management.
FeedManager	Manages data feed selection and fallback logic.

Module Contents

exception py_alpaca_api.http.feed_manager.APIRequestError(status_code: int | None = None, message: str = '')

Bases: PyAlpacaAPIError

Raised when an API request fails.

exception py_alpaca_api.http.feed_manager.ValidationError

Bases: PyAlpacaAPIError

Raised when input validation fails.

py_alpaca_api.http.feed_manager.logger

class py_alpaca_api.http.feed_manager.FeedType(*args, **kwds)

Bases: enum.Enum

Available data feed types.

SIP = 'sip'

IEX = 'iex'

OTC = 'otc'

classmethod from_string(value: str) → FeedType

Create FeedType from string value.

class py_alpaca_api.http.feed_manager.SubscriptionLevel(*args, **kwds)

Bases: enum.Enum

User subscription levels.

BASIC = 'basic'

UNLIMITED = 'unlimited'

BUSINESS = 'business'

classmethod from_error(error_message: str) → SubscriptionLevel | None

Detect subscription level from error message.

class py_alpaca_api.http.feed_manager.FeedConfig

Configuration for feed management.

preferred_feed: FeedType

fallback_feeds: list[FeedType]

auto_fallback: bool = True

subscription_level: SubscriptionLevel | None = None

endpoint_feeds: dict[str, FeedType]

get_feed_for_endpoint(endpoint: str) → FeedType

Get the configured feed for a specific endpoint.

class py_alpaca_api.http.feed_manager.FeedManager(config: FeedConfig | None = None)

Manages data feed selection and fallback logic.

FEED_SUPPORTED_ENDPOINTS: ClassVar[set[str]]

SUBSCRIPTION_FEEDS: ClassVar[dict[SubscriptionLevel, list[FeedType]]]

get_feed(endpoint: str, symbol: str | None = None) → str | None

Get the appropriate feed for an endpoint.

Parameters:

• endpoint – The API endpoint being called

• symbol – Optional symbol for endpoint-specific logic

Returns:

Feed parameter value or None if endpoint doesn’t support feeds

handle_feed_error(endpoint: str, feed: str, error: py_alpaca_api.exceptions.APIRequestError, symbol: str | None = None) → str | None

Handle feed-related errors and return alternative feed if available.

Parameters:

• endpoint – The API endpoint that failed

• feed – The feed that caused the error

• error – The API error

• symbol – Optional symbol for endpoint-specific tracking

Returns:

Alternative feed to try, or None if no alternatives available

detect_subscription_level(api_client: Any) → SubscriptionLevel

Detect user’s subscription level by testing API access.

Parameters:

api_client – API client instance to test with

Returns:

Detected subscription level

validate_feed(endpoint: str, feed: str) → bool

Validate if a feed is appropriate for an endpoint.

Parameters:

• endpoint – The API endpoint

• feed – The feed to validate

Returns:

True if feed is valid for endpoint

reset_failures(endpoint: str | None = None) → None

Reset tracked feed failures.

Parameters:

endpoint – Optional endpoint to reset. If None, resets all.

get_available_feeds() → list[FeedType]

Get list of available feeds based on subscription level.

Returns:

List of available feed types