Steam Move Detection Python API for Sports Betting
Steam move detection separates profitable line followers from noise traders. When sharp money hits a sportsbook, lines move fast and hard—that's your signal. The key is building a steam move detection Python API system that catches these movements in real-time, not after the juice settles.
I've built several steam-chasing systems over the years. The profitable ones share three characteristics: sub-second line monitoring, volume-weighted movement analysis, and automatic position sizing based on movement velocity. The losers focus on price alone without considering market context.
Understanding Steam Moves vs. Market Noise
Steam moves aren't just line movements—they're coordinated market responses to new information. Sharp bettors with proprietary models or insider knowledge place large wagers, forcing sportsbooks to adjust odds rapidly to limit exposure.
Steam Move Characteristics
Real steam moves exhibit specific patterns:
- Velocity: Lines move 3+ cents in under 60 seconds
- Direction: Movement is unidirectional across multiple books
- Volume: Betting volume spikes 300%+ from baseline
- Persistence: Lines don't bounce back within 5 minutes
Compare this to market noise: random 1-2 cent fluctuations, bidirectional movement, normal volume, and quick reversals.
Sharp Money vs. Public Money
Sharp money creates steam. Public money creates drift. Sharp money is concentrated, informed, and moves markets instantly. Public money is distributed, emotional, and moves markets slowly.
The MoneyLine API's edge detection endpoints track both patterns, but steam systems focus exclusively on sharp signatures.
Real-Time Line Movement Analysis
Building a steam detector requires continuous odds monitoring across multiple sportsbooks. You need baseline movement patterns to identify anomalies.
import asyncio
import aiohttp
import json
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import numpy as np
class SteamMoveDetector:
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://mlapi.bet"
self.baseline_movements = {}
self.active_games = {}
async def monitor_lines(self, sport: str = "nfl"):
"""Monitor line movements for steam detection"""
headers = {"Authorization": f"Bearer {self.api_key}"}
async with aiohttp.ClientSession() as session:
while True:
try:
# Get current odds
async with session.get(
f"{self.base_url}/v1/odds",
headers=headers,
params={"sport": sport, "live": True}
) as response:
data = await response.json()
# Process each game
for game in data.get('games', []):
await self.analyze_movement(game)
await asyncio.sleep(2) # 2-second intervals
except Exception as e:
print(f"Monitoring error: {e}")
await asyncio.sleep(5)
async def analyze_movement(self, game_data: Dict):
"""Analyze line movement for steam signals"""
game_id = game_data['id']
current_time = datetime.now()
if game_id not in self.active_games:
self.active_games[game_id] = {
'history': [],
'baseline_volume': 0,
'last_steam': None
}
game_state = self.active_games[game_id]
# Extract odds and volume data
for market in game_data.get('markets', []):
if market['type'] == 'spread':
movement_data = {
'timestamp': current_time,
'home_spread': market['home_price'],
'away_spread': market['away_price'],
'home_line': market['home_line'],
'volume': market.get('volume', 0)
}
game_state['history'].append(movement_data)
# Keep only last 10 minutes of data
cutoff = current_time - timedelta(minutes=10)
game_state['history'] = [
h for h in game_state['history']
if h['timestamp'] > cutoff
]
# Detect steam
steam_signal = self.detect_steam(game_state['history'])
if steam_signal:
await self.handle_steam_move(game_id, steam_signal)
def detect_steam(self, history: List[Dict]) -> Optional[Dict]:
"""Core steam detection logic"""
if len(history) < 5:
return None
recent = history[-5:] # Last 10 seconds
baseline = history[:-5] if len(history) > 5 else []
# Calculate movement velocity
price_changes = []
time_changes = []
for i in range(1, len(recent)):
prev = recent[i-1]
curr = recent[i]
price_delta = abs(curr['home_spread'] - prev['home_spread'])
time_delta = (curr['timestamp'] - prev['timestamp']).total_seconds()
if time_delta > 0:
velocity = price_delta / time_delta
price_changes.append(price_delta)
time_changes.append(velocity)
if not time_changes:
return None
avg_velocity = np.mean(time_changes)
total_movement = sum(price_changes)
# Volume analysis
recent_volume = np.mean([r['volume'] for r in recent])
baseline_volume = np.mean([b['volume'] for b in baseline]) if baseline else recent_volume
volume_spike = recent_volume / baseline_volume if baseline_volume > 0 else 1
# Steam criteria
is_steam = (
avg_velocity > 0.05 and # 5 cents per second
total_movement > 3 and # Total 3+ cent move
volume_spike > 3.0 # 300%+ volume increase
)
if is_steam:
return {
'velocity': avg_velocity,
'total_movement': total_movement,
'volume_spike': volume_spike,
'direction': 'up' if recent[-1]['home_spread'] > recent[0]['home_spread'] else 'down',
'confidence': min(avg_velocity * volume_spike / 10, 1.0)
}
return None
async def handle_steam_move(self, game_id: str, steam_signal: Dict):
"""Handle detected steam move"""
print(f"STEAM DETECTED - Game {game_id}")
print(f" Velocity: {steam_signal['velocity']:.3f} cents/second")
print(f" Movement: {steam_signal['total_movement']:.1f} cents")
print(f" Volume Spike: {steam_signal['volume_spike']:.1f}x")
print(f" Direction: {steam_signal['direction']}")
print(f" Confidence: {steam_signal['confidence']:.2f}")
# Mark last steam time to avoid duplicate alerts
self.active_games[game_id]['last_steam'] = datetime.now()
Volume-Weighted Movement Algorithms
Price movement without volume context generates false signals. A 5-cent line move on $100 volume differs fundamentally from the same move on $10,000 volume.
Volume Surge Detection
Sharp money creates volume spikes that precede line movements. Monitor volume patterns to predict steam before lines fully adjust:
def calculate_volume_momentum(self, volume_history: List[float], window: int = 30) -> float:
"""Calculate volume momentum indicator"""
if len(volume_history) < window * 2:
return 0.0
recent = volume_history[-window:]
baseline = volume_history[-window*2:-window]
recent_avg = np.mean(recent)
baseline_avg = np.mean(baseline)
# Weighted recent volume (more weight to latest data)
weights = np.linspace(0.5, 1.0, len(recent))
weighted_recent = np.average(recent, weights=weights)
momentum = (weighted_recent - baseline_avg) / baseline_avg if baseline_avg > 0 else 0
return momentum
def detect_pre_steam_volume(self, game_state: Dict) -> bool:
"""Detect volume patterns that precede steam moves"""
history = game_state['history']
if len(history) < 60: # Need 2 minutes of data
return False
volumes = [h['volume'] for h in history]
momentum = self.calculate_volume_momentum(volumes)
# Volume spike threshold
return momentum > 2.0 # 200%+ momentum increase
Market Depth Analysis
Steam moves often exhaust order book depth, creating temporary inefficiencies. Track ask/bid spread widening as another steam indicator.
Building Real-Time Alert Systems
Steam detection is worthless without immediate notification. Build alert systems that trigger within seconds of detection, not minutes.
Webhook Integration
async def send_steam_alert(self, game_id: str, steam_data: Dict):
"""Send real-time steam alerts via webhook"""
alert_payload = {
'type': 'steam_detected',
'game_id': game_id,
'timestamp': datetime.now().isoformat(),
'steam_data': steam_data,
'action': 'immediate' if steam_data['confidence'] > 0.8 else 'monitor'
}
webhook_urls = [
'https://your-discord-webhook.com/webhook',
'https://your-telegram-bot.com/send',
'https://your-trading-system.com/steam-alert'
]
async with aiohttp.ClientSession() as session:
tasks = []
for url in webhook_urls:
task = session.post(url, json=alert_payload)
tasks.append(task)
await asyncio.gather(*tasks, return_exceptions=True)
Mobile Push Notifications
For mobile alerts, integrate with services like Pushover or native push notification services. Steam moves require immediate action—email alerts arrive too late.
Historical Steam Move Backtesting
Before deploying any steam detection system, backtest against historical data. Not all steam moves are profitable. Some represent market corrections, others represent temporary inefficiencies.
Performance Metrics
Track these key metrics for your steam detection system:
- Detection Latency: Time from line movement to alert
- False Positive Rate: Percentage of alerts that don't represent tradeable steam
- Profit Factor: Average profitable steam / Average losing steam
- ROI per Alert: Return on investment per steam alert
class SteamBacktester:
def __init__(self, historical_data: List[Dict]):
self.data = historical_data
self.results = []
def backtest_period(self, start_date: str, end_date: str) -> Dict:
"""Backtest steam detection over specific period"""
period_data = [
d for d in self.data
if start_date <= d['timestamp'] <= end_date
]
detected_steams = []
for game_data in period_data:
steam_signal = self.detect_steam(game_data['line_history'])
if steam_signal:
detected_steams.append({
'game_id': game_data['game_id'],
'detection_time': game_data['timestamp'],
'signal': steam_signal,
'outcome': self.calculate_outcome(game_data)
})
return self.analyze_results(detected_steams)
def analyze_results(self, steams: List[Dict]) -> Dict:
"""Analyze backtest results"""
if not steams:
return {'error': 'No steams detected'}
profits = [s['outcome']['profit'] for s in steams]
wins = len([p for p in profits if p > 0])
return {
'total_steams': len(steams),
'win_rate': wins / len(steams),
'avg_profit': np.mean(profits),
'total_roi': sum(profits),
'sharpe_ratio': np.mean(profits) / np.std(profits) if np.std(profits) > 0 else 0
}
Integration with Sportsbook APIs
Steam detection requires real-time access to multiple sportsbooks. Different books move at different speeds—Pinnacle leads, others follow.
API Rate Limiting
Most sportsbooks limit API requests. Optimize your polling strategy:
- Poll sharp books (Pinnacle, Circa) every 1-2 seconds
- Poll soft books every 3-5 seconds
- Use websockets when available
- Implement exponential backoff for rate limit errors
The MoneyLine API aggregates odds from 100+ sportsbooks, providing unified access without managing individual API limits.
Error Handling and Redundancy
Build redundant data streams. When one API fails, others continue providing critical steam detection data:
async def redundant_odds_fetch(self, game_id: str) -> Optional[Dict]:
"""Fetch odds with redundancy across multiple sources"""
sources = [
('primary', f"{self.base_url}/v1/odds"),
('backup1', f"{self.backup_url}/odds"),
('backup2', f"{self.secondary_url}/lines")
]
for source_name, url in sources:
try:
async with aiohttp.ClientSession() as session:
async with session.get(url, params={'game_id': game_id}) as response:
if response.status == 200:
return await response.json()
except Exception as e:
print(f"Source {source_name} failed: {e}")
continue
return None
Frequently Asked Questions
How fast do steam moves typically occur?
Steam moves happen in 30-180 seconds from initiation to completion. Sharp money hits first, creating immediate line movement. The fastest steam moves complete within 30 seconds—these often represent the highest-quality information. Slower steam moves (3-5 minutes) may indicate public money following sharp action.
What's the minimum API polling frequency for steam detection?
Poll every 1-2 seconds for effective steam detection. Faster polling (sub-second) provides marginal improvement but increases API costs significantly. Slower polling (5+ seconds) misses fast-moving opportunities. The sweet spot is 1-2 seconds for sharp books, 3-5 seconds for recreational books.
Which sportsbooks lead steam moves?
Pinnacle, Circa Sports, and offshore books typically lead steam moves. These books accept larger limits from sharp bettors, making them first movers. DraftKings, FanDuel, and other recreational books follow 30-120 seconds later. Build your detection system around leading books, not followers.
How do you distinguish steam from line corrections?
Steam moves exhibit sustained volume and unidirectional price movement. Line corrections show immediate reversal after initial movement. True steam persists for 5+ minutes without significant retracement. Volume analysis is critical—corrections have normal volume, steam has spiked volume.
What's a profitable steam-chasing strategy?
Follow confirmed steam within 60 seconds of detection, bet only when multiple sharp books move in the same direction, and limit position size to 1-2% of bankroll per steam. Avoid chasing steam on recreational books—focus on following sharp money at soft books offering better prices. The key is speed and selectivity, not frequency.