feat: Add WebSocket streaming support (v3.1.0-alpha)

[TexasCoding]

Summary

This PR implements WebSocket streaming support for real-time market data, marking the completion of v3.1.0-alpha milestone from the DEVELOPMENT_PLAN.md.

Key Features

• 📡 Real-time streaming: WebSocket client for quotes, trades, and bars
• 🔄 Auto-reconnection: Exponential backoff for stable connections
• 📊 Multiple handlers: Support for concurrent data handlers
• 🎯 Dynamic subscriptions: Subscribe/unsubscribe symbols on the fly

Implementation Details

Core Components

• StreamClient class with full WebSocket management
• Data models: QuoteData, TradeData, BarData
• Thread-safe message dispatching
• Context manager support for clean resource management
• Feed selection (IEX, SIP, OTC)

Testing

• ✅ 36 unit tests covering core functionality
• ✅ 11 integration tests with real market data
• ✅ Proper connection cleanup and error handling
• ✅ Authentication and subscription management

Documentation

• Comprehensive streaming guide with examples
• Updated README with v3.1.0-alpha features
• API integration examples

Usage Example

# Create streaming client
stream = api.create_stream_client(feed="iex")

# Define handlers
def on_quote(quote):
    print(f"{quote.symbol}: Bid ${quote.bid_price} Ask ${quote.ask_price}")

# Connect and subscribe
stream.connect()
stream.subscribe_quotes(["AAPL", "GOOGL"], on_quote)

Test Plan

• Unit tests pass (36/36)
• Integration tests work with live API
• Authentication and reconnection tested
• Multiple subscription types verified
• Manual testing during market hours
• Performance testing with high volume symbols

Breaking Changes

None - this is a new feature addition

Related Issues

• Implements WebSocket streaming milestone from DEVELOPMENT_PLAN.md
• Part of v3.1.0 roadmap

🤖 Generated with Claude Code

[claude]

🔍 Code Review for WebSocket Streaming Support

Thank you for implementing this comprehensive WebSocket streaming feature! This is a significant enhancement to the library. I've reviewed the implementation thoroughly and here's my feedback:

✅ Strengths

Architecture & Design

• Well-structured module organization: Clean separation between StreamClient, data models, and handlers
• Proper use of design patterns: Context manager support, thread-safe message dispatching, and factory pattern for model creation
• Type safety maintained: Full type annotations throughout, consistent with the project's strict mypy configuration
• Follows project conventions: Adheres to CLAUDE.md guidelines for code style and patterns

Functionality

• Robust connection management: Exponential backoff reconnection strategy with configurable limits
• Thread-safe implementation: Proper handling of concurrent handlers and message dispatching
• Clean API surface: Intuitive subscribe/unsubscribe methods with support for single symbols or lists
• Comprehensive data models: Well-defined dataclasses for QuoteData, TradeData, and BarData

Documentation

• Excellent user guide: The streaming_guide.md provides comprehensive examples and best practices
• Clear code documentation: Well-documented methods with proper docstrings

🚨 Areas for Improvement

1. Thread Safety Concerns

The handlers list manipulation could have race conditions. Consider adding thread locks for handler list modifications to prevent concurrent modification issues.

2. Resource Management

• The _run_forever method catches all exceptions, which might hide critical errors
• Thread joining timeout (5 seconds) in disconnect() might not be sufficient for slow networks
• Consider implementing a graceful shutdown mechanism

3. Error Handling Enhancement

Add message content to error logs (sanitized) for debugging, and consider implementing a dead letter queue for failed messages.

4. Performance Considerations

• Synchronous handler execution could block message processing for slow handlers
• Consider using a queue-based approach for handler execution
• Add metrics for message processing time

5. Security Enhancements

• API credentials are stored in plain text in memory
• Consider using secure string storage for credentials
• Add support for credential refresh/rotation

🐛 Potential Bugs

1. Timestamp Parsing

The datetime.fromisoformat() call will raise an exception if 't' is missing or empty. Consider adding a fallback.

2. Feed URL Construction

The paper parameter is stored but not used in URL construction. Per the comment, this is intentional for data streams, but it might confuse users.

📊 Test Coverage

Good test coverage with 36 unit tests and 11 integration tests. Consider adding:

• Tests for concurrent subscription/unsubscription
• Tests for handler exceptions not affecting other handlers
• Tests for maximum reconnection attempts
• Memory leak tests for long-running connections

✅ Overall Assessment

This is a high-quality implementation that significantly enhances the library's capabilities. The code is well-structured, follows project conventions, and provides a clean API for users. The suggested improvements are mostly about edge cases and optimizations rather than fundamental issues.

Recommendation: Approve with minor revisions. Address the thread safety concerns and error handling improvements before the final v3.1.0 release.

Great work on this feature! The real-time streaming capability will be valuable for users building trading applications.

[claude]

🔍 Code Review for PR #77: WebSocket Streaming Support

Thank you for this comprehensive WebSocket streaming implementation! This is a valuable addition to py-alpaca-api. The feature is well-designed with good test coverage and documentation. However, I've identified several critical issues that should be addressed before merging to ensure production stability.

✅ Strengths

• Clean architecture with separation of concerns (models, client, tests)
• Comprehensive data models with proper type hints
• Good integration tests covering real-world scenarios
• Context manager support for resource management
• Reconnection logic with exponential backoff
• Well-documented with examples and guides

🚨 Critical Issues (Must Fix)

1. Thread Safety Violations

The implementation has race conditions due to unsynchronized access to shared state:

File: src/py_alpaca_api/streaming/stream_client.py

# Lines 55-56, 58-68 - Unprotected state modifications
self.is_connected = False  # Race condition
self.is_authenticated = False  # Race condition
self.subscriptions = {...}  # Concurrent modification risk

Fix Required:

def __init__(self, ...):
    self._state_lock = threading.RLock()
    self._handler_lock = threading.RLock()
    self._subscription_lock = threading.RLock()
    # ... rest of init

@property
def is_connected(self) -> bool:
    with self._state_lock:
        return self._is_connected

2. Improper Exception Handling

Using logger.error() instead of logger.exception() loses stack traces:

Locations:

• Line 119: _run_forever
• Lines 189, 191: _on_message
• Line 229: _handle_reconnect
• Line 269: _dispatch_message

Fix Required:

# Replace all instances of:
except Exception as e:
    logger.error(f"Error: {e}")

# With:
except Exception:
    logger.exception("Descriptive error message")

3. Message Corruption Vulnerability

No validation for empty/partial/corrupted messages:

File: src/py_alpaca_api/streaming/stream_client.py:137-192

Fix Required:

def _on_message(self, ws, message: str) -> None:
    if not message or not message.strip():
        logger.warning("Received empty message")
        return
        
    try:
        data = json.loads(message)
    except json.JSONDecodeError:
        logger.exception(f"Invalid JSON: {message[:100]}...")
        return  # Don't crash

4. Memory Leak Potential

No way to remove handlers or clean up on disconnect:

Fix Required:

def remove_handler(self, stream_type: StreamType, handler: Callable) -> bool:
    with self._handler_lock:
        if handler in self.handlers[stream_type]:
            self.handlers[stream_type].remove(handler)
            return True
    return False

def clear_handlers(self):
    with self._handler_lock:
        for stream_type in StreamType:
            self.handlers[stream_type].clear()

🔧 Important Issues

5. Non-Atomic State Checks

Connection state checks aren't atomic (lines 84-86, 245-246), leading to potential race conditions.

6. Resource Cleanup

Thread cleanup relies on timeout (line 421) without guaranteed termination.

7. Missing Test Coverage

Need tests for:

• Concurrent access scenarios
• Corrupted message handling
• Memory leak detection
• Handler error isolation

📝 Recommendations

Immediate Actions

1. Add thread synchronization for all shared state
2. Fix exception handling to preserve stack traces
3. Add message validation and corruption handling
4. Implement handler lifecycle management

Before Stable Release

1. Implement atomic state management
2. Add comprehensive thread safety tests
3. Add metrics/monitoring capabilities
4. Performance test with 100+ concurrent symbols

🏆 Performance Considerations

• Handler execution is synchronous - keep them lightweight
• Consider adding a queue-based dispatcher for heavy processing
• Monitor memory usage with large subscription sets
• Add backpressure handling for high-volume streams

🔒 Security Considerations

• ✅ Credentials properly handled in authentication
• ✅ No sensitive data logged
• ⚠️ Consider adding rate limiting for handler errors
• ⚠️ Add input validation for symbols

📊 Test Results Review

The test suite is comprehensive with 36 unit tests and 11 integration tests. However, critical edge cases are missing:

• Thread safety under concurrent load
• Partial message handling
• Memory leak detection
• Chaos testing scenarios

💡 Suggested Enhancements

1. Add health check method for monitoring
2. Implement metrics collection (messages/sec, errors, uptime)
3. Add debug mode with verbose logging
4. Consider weak references for handlers to prevent leaks
5. Add circuit breaker pattern for repeated failures

✅ Checklist for Approval

• Add thread synchronization (critical)
• Fix exception handling (critical)
• Add message validation (critical)
• Implement handler removal (critical)
• Add missing test coverage
• Update documentation with thread safety notes
• Performance test with high volume

📌 Summary

This is a great feature addition with solid architecture and good test coverage. The identified issues are common in concurrent systems and can be addressed systematically. Once the critical thread safety and exception handling issues are resolved, this will be a robust addition to the library.

The documentation (especially WEBSOCKET_IMPROVEMENTS.md) shows excellent planning for addressing these issues. I recommend implementing the P0 fixes from that document before merging.

Verdict: Approved with required changes. The foundation is solid, but thread safety and error handling must be addressed for production use.

Great work on this implementation! Looking forward to seeing the fixes and using this in production. 🚀

---

Review performed using Claude Code with focus on CLAUDE.md compliance and production readiness