py_alpaca_api.cache.cache_manager

Cache manager for py-alpaca-api.

Attributes

logger

Classes

CacheConfig	Configuration for cache system.
CacheType	Types of cache backends supported.
LRUCache	Least Recently Used (LRU) cache implementation.
RedisCache	Redis cache implementation.
CacheManager	Manages caching for py-alpaca-api.

Module Contents

class py_alpaca_api.cache.cache_manager.CacheConfig

Configuration for cache system.

cache_type

Type of cache backend to use

max_size

Maximum number of items in memory cache

default_ttl

Default time-to-live in seconds

data_ttls

TTL overrides per data type

redis_host

Redis host (if using Redis)

redis_port

Redis port (if using Redis)

redis_db

Redis database number (if using Redis)

redis_password

Redis password (if using Redis)

enabled

Whether caching is enabled

cache_type: CacheType

max_size: int = 1000

default_ttl: int = 300

data_ttls: dict[str, int]

redis_host: str = 'localhost'

redis_port: int = 6379

redis_db: int = 0

redis_password: str | None = None

enabled: bool = True

get_ttl(data_type: str) → int

Get TTL for a specific data type.

Parameters:

data_type – Type of data to get TTL for

Returns:

TTL in seconds

class py_alpaca_api.cache.cache_manager.CacheType(*args, **kwds)

Bases: enum.Enum

Types of cache backends supported.

MEMORY = 'memory'

REDIS = 'redis'

DISABLED = 'disabled'

py_alpaca_api.cache.cache_manager.logger

class py_alpaca_api.cache.cache_manager.LRUCache(max_size: int = 1000)

Least Recently Used (LRU) cache implementation.

get(key: str) → Any | None

Get item from cache.

Parameters:

key – Cache key

Returns:

Cached value or None if not found/expired

set(key: str, value: Any, ttl: int) → None

Set item in cache.

Parameters:

• key – Cache key

• value – Value to cache

• ttl – Time-to-live in seconds

delete(key: str) → bool

Delete item from cache.

Parameters:

key – Cache key

Returns:

True if deleted, False if not found

clear() → None

Clear all items from cache.

size() → int

Get current cache size.

Returns:

Number of items in cache

cleanup_expired() → int

Remove expired items from cache.

Returns:

Number of items removed

class py_alpaca_api.cache.cache_manager.RedisCache(config: py_alpaca_api.cache.cache_config.CacheConfig)

Redis cache implementation.

get(key: str) → Any | None

Get item from cache.

Parameters:

key – Cache key

Returns:

Cached value or None if not found

set(key: str, value: Any, ttl: int) → None

Set item in cache.

Parameters:

• key – Cache key

• value – Value to cache

• ttl – Time-to-live in seconds

delete(key: str) → bool

Delete item from cache.

Parameters:

key – Cache key

Returns:

True if deleted, False if not found

clear() → None

Clear all items from cache.

size() → int

Get current cache size.

Returns:

Number of items in cache

class py_alpaca_api.cache.cache_manager.CacheManager(config: py_alpaca_api.cache.cache_config.CacheConfig | None = None)

Manages caching for py-alpaca-api.

generate_key(prefix: str, **kwargs) → str

Generate cache key from prefix and parameters.

Parameters:

• prefix – Key prefix (e.g., “bars”, “quotes”)

• **kwargs – Parameters to include in key

Returns:

Cache key

get(key: str, data_type: str | None = None) → Any | None

Get item from cache.

Parameters:

• key – Cache key

• data_type – Optional data type for metrics

Returns:

Cached value or None if not found

set(key: str, value: Any, data_type: str, ttl: int | None = None) → None

Set item in cache.

Parameters:

• key – Cache key

• value – Value to cache

• data_type – Type of data (for TTL lookup)

• ttl – Optional TTL override in seconds

delete(key: str) → bool

Delete item from cache.

Parameters:

key – Cache key

Returns:

True if deleted, False if not found

clear(prefix: str | None = None) → int

Clear cache items.

Parameters:

prefix – Optional prefix to clear only specific items

Returns:

Number of items cleared

invalidate_pattern(pattern: str) → int

Invalidate cache items matching a pattern.

Parameters:

pattern – Pattern to match (e.g., “bars:AAPL”)

Returns:

Number of items invalidated

get_stats() → dict[str, Any]

Get cache statistics.

Returns:

Dictionary with cache stats

reset_stats() → None

Reset cache statistics.

cached(data_type: str, ttl: int | None = None) → collections.abc.Callable

Decorator for caching function results.

Parameters:

• data_type – Type of data being cached

• ttl – Optional TTL override

Returns:

Decorator function