Overview The WebSocket Updater (OnWebsocketUpdater) is PROPPR’s real-time data pipeline that maintains a persistent WebSocket connection to the Odds API and updates MongoDB collections with live odds changes.
The service handles updates for all bots (Team, Arb, EV, Player, Overtime) from a single WebSocket connection, as Odds API allows only ONE connection per API key.
Architecture Core Components UnifiedWebSocketService ├── WebSocket Connection Management ├── Multi-sport Rotation ├── Duplicate Update Filtering └── Bot-specific Callbacks ├── Team Bot Callback ├── Arb Bot Callback ├── EV Bot Callback ├── Player Bot Callback └── Overtime Bot Callback
Source: /WebsocketUpdater/services/websocket_service.pyCollections Updated Collection Purpose Bot team_oddsTeam market odds Team Bot player_oddsPlayer market odds Player Bot arbitrage_betsArbitrage opportunities Arb Bot all_value_betsEV betting opportunities EV Bot overtime_oddsOvertime market odds Overtime Bot live_inplay_oddsLive in-play odds Live Bot
Key Features 1. Real-Time Updates WebSocket Message Processing
def _process_odds_update ( self , data : Dict): """ Process odds update from WebSocket. Distributes update to relevant bot callbacks. """ event_id = data.get( 'id' ) bookmaker = data.get( 'bookie' ) markets = data.get( 'markets' , []) # Filter deprecated V2 IDs if str (event_id).startswith( '1' ) and len ( str (event_id)) == 10 : return # Distribute to all bot callbacks for callback in self .team_bot_callbacks: callback(event_id, bookmaker, normalized_data)
Key Implementation Details:
Filters deprecated V2 IDs (10-digit IDs starting with ‘1’)
Adds source: 'websocket' label to all documents
Caches sharp odds (M88/FB Sports) for EV calculations
Tracks bookmaker accumulation for arbitrage detection
2. Sport Rotation The WebSocket API requires a sport parameter. The service rotates through all major sports every 30 seconds to ensure comprehensive coverage.
sports = [ 'football' , 'basketball' , 'tennis' , 'ice-hockey' , 'baseball' , 'american-football' , 'rugby-league' , 'cricket' , 'golf' , 'boxing' , 'mma' , 'motorsports' , 'esports' ] rotation_interval = 30 # seconds
3. Automatic Reconnection def _connect_loop ( self ): """Main connection loop with exponential backoff""" while self .running: try : self ._connect() self .reconnect_attempts = 0 except Exception as e: self .reconnect_attempts += 1 if self .reconnect_attempts >= self .max_reconnect_attempts: self .running = False break # Exponential backoff (max 300s) wait_time = min ( 2 ** self .reconnect_attempts, 300 ) time.sleep(wait_time)
4. Duplicate Filtering WebSocket can send duplicate updates within seconds. The service maintains a 5-second cache to prevent duplicate processing.
processed_updates: Dict[ str , float ] = {} # event_id -> timestamp duplicate_cache_ttl = 5 # seconds update_key = f " { event_id } : { bookmaker } : { data.get( 'type' ) } " if update_key in self .processed_updates: last_processed = self .processed_updates[update_key] if current_time - last_processed < self .duplicate_cache_ttl: return # Skip duplicate
Bot Callbacks Team Bot Callback
def team_bot_callback ( event_id , bookmaker , data ): """ Creates/updates team_odds documents. Filters to team markets only (ML, Totals, Spread, Corners, etc.) """ markets = data.get( 'markets' , []) team_markets = _filter_team_markets(markets) if not team_markets: return # Update or create document team_odds_collection.update_one( { 'id' : event_id}, { '$set' : { f 'bookmakers. { bookmaker } ' : team_markets, 'last_updated' : datetime.now(), 'source' : 'websocket' }}, upsert = True )
Filtered Markets: ML, Totals, Spread, Corners, Shots, Cards, BTTSArbitrage Bot Callback
def arb_bot_callback ( event_id , bookmaker , data ): """ Accumulates odds from multiple bookmakers. Schedules delayed arbitrage check (10s) to allow multiple bookmakers to arrive. """ # Cache odds by bookmaker arb_odds_cache[event_id][bookmaker] = data # Schedule delayed check check_time = time.time() + 10 arb_pending_checks[event_id] = check_time
Why Delayed? WebSocket sends updates ONE BOOKMAKER AT A TIME. Immediate checks would miss arbitrages requiring 2+ bookmakers.EV Bot Callback
def ev_bot_callback ( event_id , bookmaker , data ): """ Updates odds_history for existing value bets. Creates new value bets when sharp odds shift. """ # Find all value bets for this event value_bets = all_value_bets_collection.find({ 'event_id' : event_id }) for bet in value_bets: # Update odds history all_value_bets_collection.update_one( { '_id' : bet[ '_id' ]}, { '$push' : { 'odds_history' : { 'timestamp' : datetime.now(), 'bookmaker_odds' : extract_odds(data), 'source' : 'websocket' } }} )
Rate Limiting
Global Tracker Uses SharedServices.tracking.global_request_tracker to coordinate with API Poller
API Quota 5000 requests/hour shared across all services
if RATE_TRACKER_AVAILABLE and global_request_tracker: if not global_request_tracker.can_make_request(): logger.warning( "Rate limit reached" ) return global_request_tracker.record_request()
Running the Service Manual Start python /opt/PROPPR/WebsocketUpdater/runners/run_websocket_updater.py
Systemd Service proppr-websocket-updater.service
[Unit] Description =PROPPR WebSocket Updater After =network.target mongod.service [Service] Type =simple User =proppr WorkingDirectory =/opt/PROPPR ExecStart =/usr/bin/python3 /opt/PROPPR/WebsocketUpdater/runners/run_websocket_updater.py Restart =always RestartSec =10 [Install] WantedBy =multi-user.target
sudo systemctl start proppr-websocket-updater sudo systemctl enable proppr-websocket-updater sudo systemctl status proppr-websocket-updater
Health Monitoring Statistics Tracking stats = { 'messages_received' : 0 , 'updates_processed' : 0 , 'team_bot_alerts' : 0 , 'arb_bot_alerts' : 0 , 'ev_bot_alerts' : 0 , 'player_bot_alerts' : 0 , 'overtime_bot_alerts' : 0 , 'errors' : 0 , 'reconnections' : 0 }
Health Check def is_healthy ( self ) -> bool : """Check if WebSocket is healthy""" if not self .running or not self .connected: return False # Check if messages received recently if self .last_message_time: time_since_last = (datetime.now() - self .last_message_time).total_seconds() if time_since_last > 60 : return False return True
Troubleshooting
Symptoms: last_message_time is stale (>60 seconds)Solutions:
Check WebSocket connection status
Verify API key is valid
Check network connectivity
Review reconnection logs
Duplicate documents created
Cause: V2 IDs (deprecated) not filteredSolution: Verify V2 ID filter is active:if str (event_id).startswith( '1' ) and len ( str (event_id)) == 10 : return # Skip V2 IDs
Missing arbitrage opportunities
Cause: Delayed check not runningSolution: Check arb_processor_thread is running:arb_processor_thread = threading.Thread( target = self ._process_delayed_arb_checks, daemon = True )
Configuration Environment Variables # MongoDB MONGO_CONNECTION_STRING = "mongodb://localhost:27017" MONGO_DATABASE = "proppr" # Odds API ODDS_API_KEY = "your_api_key_here" # WebSocket WS_ROTATION_INTERVAL = 30 # seconds WS_MAX_RECONNECT_ATTEMPTS = 5
Market Filters Team Markets
Player Markets
TEAM_MARKET_NAMES = { 'ML' , 'Totals' , 'Spread' , 'BTTS' , 'Corners Totals' , 'Total Cards' , 'Team Shots Home' , 'Team Shots Away' , 'Goalkeeper Saves Home' , 'Goalkeeper Saves Away' }
PLAYER_MARKET_NAMES = { 'Player Shots' , 'Player Shots on Target' , 'Player Goals' , 'Player Assists' , 'Player Tackles' , 'Player Passes' }
Metric Value Message Processing <100ms per update Callback Latency <50ms per callback Cache Cleanup Every 5 minutes Reconnection Time <10s (with backoff)
API Poller Polling-based odds updates
Team Bot Team market alerts
EV Bot Expected value betting