Python Workflow Examples
Live Update (HugeAPI)
python
import requests
import json
import time
headers = {
"x-portal-apikey": "__YOUR_KEY__",
}
base_url = "https://pinnacle-odds-api.hgapi.top"
s = requests.Session()
s.headers.update(headers)
since_items = {}
all_events = {}
while True:
for sport_id in [1]: # endpoint /kit/v1/sports
# is_have_odds = 0 or 1, event_type = prematch, live
params = {
'sport_id': sport_id,
'is_have_odds': True,
'since': since_items.get(sport_id)
}
print('[REQUEST] %s' % params)
response = s.get(base_url + '/kit/v1/markets', params=params)
if response.status_code != 200:
raise Exception(response.status_code, response.text)
result = json.loads(response.text)
since_items[sport_id] = result['last']
for event in result['events']:
all_events[str(event['event_id'])] = event
try:
print(' %s: %s — %s %s' % (
event['league_name'],
event['home'],
event['away'],
event['periods']['num_0']['money_line']
))
except KeyError:
pass
print('Sport: %s' % result['sport_name'])
print('Number of changes: %s' % len(result['events']))
print(' ')
time.sleep(3)Live Update (RapidAPI)
python
import requests
import json
import time
headers = {
"x-rapidapi-key": "__YOUR_KEY__",
"x-rapidapi-host": "pinbook-odds.p.rapidapi.com"
}
base_url = "https://pinbook-odds.p.rapidapi.com"
s = requests.Session()
s.headers.update(headers)
since_items = {}
all_events = {}
while True:
for sport_id in [1]:
params = {
'sport_id': sport_id,
'is_have_odds': True,
'since': since_items.get(sport_id)
}
print('[REQUEST] %s' % params)
response = s.get(base_url + '/kit/v1/markets', params=params)
if response.status_code != 200:
raise Exception(response.status_code, response.text)
result = json.loads(response.text)
since_items[sport_id] = result['last']
for event in result['events']:
all_events[str(event['event_id'])] = event
time.sleep(3)Workflow 1: Complete API Client Class
A reusable class for interacting with the PinBook Odds API.
python
import requests
import json
import time
from datetime import datetime
from typing import Optional, Dict, List, Any
class PinBookClient:
"""
PinBook Odds API Client
Supports both HugeAPI and RapidAPI providers.
"""
def __init__(self, api_key: str, provider: str = "hugeapi"):
"""
Initialize the client.
Args:
api_key: Your API key
provider: "hugeapi" or "rapidapi"
"""
self.session = requests.Session()
if provider == "hugeapi":
self.base_url = "https://pinnacle-odds-api.hgapi.top"
self.session.headers.update({
"x-portal-apikey": api_key
})
else: # rapidapi
self.base_url = "https://pinbook-odds.p.rapidapi.com"
self.session.headers.update({
"x-rapidapi-key": api_key,
"x-rapidapi-host": "pinbook-odds.p.rapidapi.com"
})
# Store since timestamps for each sport
self._since_cache: Dict[int, int] = {}
self._special_since_cache: Dict[int, int] = {}
def _request(self, endpoint: str, params: Optional[Dict] = None) -> Dict:
"""Make an API request."""
url = f"{self.base_url}{endpoint}"
response = self.session.get(url, params=params)
if response.status_code != 200:
raise Exception(f"API Error {response.status_code}: {response.text}")
return response.json()
def check_betting_status(self) -> Dict:
"""Check Pinnacle server betting status."""
return self._request("/kit/v1/betting-status")
def get_sports(self) -> List[Dict]:
"""Get list of all available sports."""
return self._request("/kit/v1/sports")
def get_periods(self, sport_id: int) -> Dict:
"""Get period definitions for a sport."""
return self._request("/kit/v1/meta-periods", {"sport_id": sport_id})
def get_leagues(self, sport_id: int) -> List[Dict]:
"""Get leagues for a sport."""
return self._request("/kit/v1/leagues", {"sport_id": sport_id})
def get_markets(
self,
sport_id: int,
since: Optional[int] = None,
league_ids: Optional[str] = None,
event_ids: Optional[str] = None,
event_type: Optional[str] = None,
is_have_odds: Optional[bool] = None
) -> Dict:
"""
Get markets/odds for a sport.
Args:
sport_id: Sport ID (required)
since: UTC timestamp for delta updates
league_ids: Comma-separated league IDs
event_ids: Comma-separated event IDs
event_type: "prematch" or "live"
is_have_odds: Filter events with odds
"""
params = {"sport_id": sport_id}
if since is not None:
params["since"] = since
if league_ids:
params["league_ids"] = league_ids
if event_ids:
params["event_ids"] = event_ids
if event_type:
params["event_type"] = event_type
if is_have_odds is not None:
params["is_have_odds"] = str(is_have_odds).lower()
result = self._request("/kit/v1/markets", params)
# Update since cache
if "last" in result:
self._since_cache[sport_id] = result["last"]
return result
def get_special_markets(
self,
sport_id: int,
since: Optional[int] = None,
event_type: Optional[str] = None,
is_have_lines: Optional[bool] = None
) -> Dict:
"""Get special markets for a sport."""
params = {"sport_id": sport_id}
if since is not None:
params["since"] = since
if event_type:
params["event_type"] = event_type
if is_have_lines is not None:
params["is_have_lines"] = str(is_have_lines).lower()
result = self._request("/kit/v1/special-markets", params)
if "last" in result:
self._special_since_cache[sport_id] = result["last"]
return result
def get_event_details(self, event_id: int) -> Dict:
"""Get detailed event info with historical odds."""
return self._request("/kit/v1/details", {"event_id": event_id})
def get_archive(
self,
sport_id: int,
page_num: int = 1,
league_ids: Optional[str] = None,
start_after: Optional[str] = None,
start_before: Optional[str] = None
) -> Dict:
"""Get archived/settled events."""
params = {"sport_id": sport_id, "page_num": page_num}
if league_ids:
params["league_ids"] = league_ids
if start_after:
params["start_after"] = start_after
if start_before:
params["start_before"] = start_before
return self._request("/kit/v1/archive", params)
def get_since(self, sport_id: int) -> Optional[int]:
"""Get cached since timestamp for a sport."""
return self._since_cache.get(sport_id)
def get_special_since(self, sport_id: int) -> Optional[int]:
"""Get cached special since timestamp for a sport."""
return self._special_since_cache.get(sport_id)
# Usage Example
if __name__ == "__main__":
client = PinBookClient("YOUR_API_KEY", provider="hugeapi")
# Get all sports
sports = client.get_sports()
print(f"Available sports: {len(sports)}")
# Get initial snapshot for Soccer (sport_id=1)
markets = client.get_markets(sport_id=1, is_have_odds=True)
print(f"Initial events: {len(markets.get('events', []))}")
# Get delta updates using cached since
time.sleep(5)
updates = client.get_markets(
sport_id=1,
since=client.get_since(1),
is_have_odds=True
)
print(f"Updated events: {len(updates.get('events', []))}")Workflow 2: Real-time Odds Monitor
Continuously monitor odds changes for specific leagues.
python
import requests
import json
import time
from datetime import datetime
from collections import defaultdict
class OddsMonitor:
"""
Real-time odds monitoring for specific sports/leagues.
Tracks odds changes and notifies on significant movements.
"""
def __init__(self, api_key: str, base_url: str):
self.session = requests.Session()
self.session.headers.update({"x-portal-apikey": api_key})
self.base_url = base_url
# Storage
self.since_cache = {}
self.events_store = {}
self.odds_history = defaultdict(list)
def _request(self, endpoint: str, params: dict) -> dict:
response = self.session.get(f"{self.base_url}{endpoint}", params=params)
response.raise_for_status()
return response.json()
def fetch_markets(self, sport_id: int, league_ids: list = None) -> dict:
"""Fetch markets with delta updates."""
params = {
"sport_id": sport_id,
"is_have_odds": True,
"since": self.since_cache.get(sport_id)
}
if league_ids:
params["league_ids"] = ",".join(map(str, league_ids))
result = self._request("/kit/v1/markets", params)
self.since_cache[sport_id] = result.get("last")
return result
def track_odds_changes(self, event: dict) -> list:
"""Track odds changes for an event."""
changes = []
event_id = event["event_id"]
try:
current_ml = event["periods"]["num_0"]["money_line"]
except KeyError:
return changes
if event_id in self.events_store:
old_event = self.events_store[event_id]
try:
old_ml = old_event["periods"]["num_0"]["money_line"]
for key in ["home", "draw", "away"]:
if key in current_ml and key in old_ml:
old_val = old_ml[key]
new_val = current_ml[key]
if old_val != new_val:
changes.append({
"field": f"money_line.{key}",
"old": old_val,
"new": new_val,
"diff": round(new_val - old_val, 4)
})
except KeyError:
pass
# Store current state
self.events_store[event_id] = event
# Store in history
if changes:
self.odds_history[event_id].append({
"timestamp": datetime.utcnow().isoformat(),
"changes": changes
})
return changes
def monitor(self, sport_id: int, league_ids: list = None, interval: int = 5):
"""
Start monitoring loop.
Args:
sport_id: Sport to monitor
league_ids: Optional list of league IDs to filter
interval: Polling interval in seconds
"""
print(f"Starting monitor for sport_id={sport_id}")
print(f"Leagues: {league_ids or 'All'}")
print("-" * 50)
while True:
try:
result = self.fetch_markets(sport_id, league_ids)
events = result.get("events", [])
print(f"\n[{datetime.utcnow().isoformat()}]")
print(f"Events with changes: {len(events)}")
for event in events:
changes = self.track_odds_changes(event)
if changes:
print(f"\n Event: {event['home']} vs {event['away']}")
print(f" League: {event['league_name']}")
for change in changes:
print(f" {change['field']}: {change['old']} -> {change['new']} ({change['diff']:+.4f})")
time.sleep(interval)
except requests.exceptions.RequestException as e:
print(f"Request error: {e}")
time.sleep(interval * 2)
except KeyboardInterrupt:
print("\nMonitoring stopped.")
break
def get_event_history(self, event_id: int) -> list:
"""Get odds change history for an event."""
return self.odds_history.get(event_id, [])
# Usage
if __name__ == "__main__":
monitor = OddsMonitor(
api_key="YOUR_API_KEY",
base_url="https://pinnacle-odds-api.hgapi.top"
)
# Monitor Soccer (sport_id=1) for specific leagues
# Bundesliga: 1842, Premier League: 1980
monitor.monitor(sport_id=1, league_ids=[1842, 1980], interval=5)Workflow 3: Value Bet Finder
Find potential value bets by comparing odds movements.
python
import requests
import json
import time
from datetime import datetime, timedelta
class ValueBetFinder:
"""
Find potential value bets by analyzing odds movements.
"""
def __init__(self, api_key: str, base_url: str):
self.session = requests.Session()
self.session.headers.update({"x-portal-apikey": api_key})
self.base_url = base_url
self.since_cache = {}
def _request(self, endpoint: str, params: dict) -> dict:
response = self.session.get(f"{self.base_url}{endpoint}", params=params)
response.raise_for_status()
return response.json()
def get_events_with_odds(self, sport_id: int, event_type: str = "prematch") -> list:
"""Get all events with odds for a sport."""
params = {
"sport_id": sport_id,
"is_have_odds": True,
"event_type": event_type
}
result = self._request("/kit/v1/markets", params)
self.since_cache[sport_id] = result.get("last")
return result.get("events", [])
def calculate_implied_probability(self, odds: float) -> float:
"""Convert decimal odds to implied probability."""
return 1 / odds if odds > 0 else 0
def calculate_margin(self, money_line: dict) -> float:
"""Calculate bookmaker margin from money line odds."""
total_prob = sum(
self.calculate_implied_probability(odds)
for odds in money_line.values()
)
return total_prob - 1 # Margin as percentage over 100%
def find_low_margin_events(
self,
sport_id: int,
max_margin: float = 0.04,
event_type: str = "prematch"
) -> list:
"""
Find events with low bookmaker margin (good value).
Args:
sport_id: Sport ID
max_margin: Maximum acceptable margin (e.g., 0.02 = 2%)
event_type: "prematch" or "live"
"""
events = self.get_events_with_odds(sport_id, event_type)
low_margin_events = []
for event in events:
try:
money_line = event["periods"]["num_0"]["money_line"]
if not money_line:
continue
margin = self.calculate_margin(money_line)
if margin <= max_margin:
low_margin_events.append({
"event_id": event["event_id"],
"home": event["home"],
"away": event["away"],
"league": event["league_name"],
"starts": event["starts"],
"money_line": money_line,
"margin": round(margin * 100, 2),
"implied_probs": {
k: round(self.calculate_implied_probability(v) * 100, 2)
for k, v in money_line.items()
}
})
except (KeyError, TypeError):
continue
return low_margin_events
def find_dropping_odds(
self,
sport_id: int,
threshold: float = 0.05,
event_type: str = "prematch"
) -> list:
"""
Find events with significantly dropping odds.
Requires comparing current odds with historical data.
"""
# First get current snapshot
events = self.get_events_with_odds(sport_id, event_type)
dropping_odds = []
for event in events:
event_id = event["event_id"]
# Get historical data
try:
details = self._request("/kit/v1/details", {"event_id": event_id})
events_details = details.get("events", [])
if events_details:
event_detail = events_details[0]
periods = event_detail.get("periods", [])
for period in periods:
history = period.get("history", {})
money_line_history = history.get("home", [])
if len(money_line_history) >= 2:
# history format: [timestamp, price, max_bet]
current = money_line_history[-1]
previous = money_line_history[0]
if len(current) >= 2 and len(previous) >= 2:
current_price = current[1]
previous_price = previous[1]
# Price drop means odds increased (more payout)
drop = (current_price - previous_price) / previous_price
if drop >= threshold:
dropping_odds.append({
"event_id": event_id,
"home": event["home"],
"away": event["away"],
"league": event["league_name"],
"previous_price": previous_price,
"current_price": current_price,
"drop_percent": round(drop * 100, 2)
})
except (requests.exceptions.RequestException, KeyError):
continue
return dropping_odds
def analyze_spreads(self, sport_id: int, event_type: str = "prematch") -> list:
"""Analyze spread markets for value opportunities."""
events = self.get_events_with_odds(sport_id, event_type)
spread_analysis = []
for event in events:
try:
spreads = event["periods"]["num_0"]["spreads"]
if not spreads:
continue
for hdp_str, spread_data in spreads.items():
handicap = spread_data.get("hdp")
home_odds = spread_data.get("home")
away_odds = spread_data.get("away")
max_bet = spread_data.get("max")
# Calculate fair odds (without margin)
home_prob = self.calculate_implied_probability(home_odds)
away_prob = self.calculate_implied_probability(away_odds)
total_prob = home_prob + away_prob
spread_analysis.append({
"event_id": event["event_id"],
"home": event["home"],
"away": event["away"],
"league": event["league_name"],
"handicap": handicap,
"home_odds": home_odds,
"away_odds": away_odds,
"max_bet": max_bet,
"margin": round((total_prob - 1) * 100, 2)
})
except (KeyError, TypeError):
continue
return spread_analysis
# Usage Example
if __name__ == "__main__":
finder = ValueBetFinder(
api_key="YOUR_API_KEY",
base_url="https://pinnacle-odds-api.hgapi.top"
)
# Find low margin events (Soccer)
print("Finding low margin events...")
low_margin = finder.find_low_margin_events(sport_id=1, max_margin=0.04)
print(f"\nFound {len(low_margin)} events with margin <= 4%:")
for event in low_margin[:10]:
print(f"\n {event['home']} vs {event['away']}")
print(f" League: {event['league']}")
print(f" Margin: {event['margin']}%")
print(f" Money Line: {event['money_line']}")
# Analyze spreads
print("\n" + "=" * 50)
print("Analyzing spreads...")
spreads = finder.analyze_spreads(sport_id=1)
# Sort by margin (lowest first)
spreads.sort(key=lambda x: x["margin"])
print(f"\nTop 5 lowest margin spreads:")
for spread in spreads[:5]:
print(f"\n {spread['home']} vs {spread['away']}")
print(f" Handicap: {spread['handicap']}")
print(f" Home: {spread['home_odds']} | Away: {spread['away_odds']}")
print(f" Margin: {spread['margin']}% | Max: ${spread['max_bet']}")Workflow 4: Event Scheduler & Reminder
Track upcoming events and get reminders.
python
import requests
import json
import time
from datetime import datetime, timedelta
from typing import List, Dict, Optional
class EventScheduler:
"""
Track upcoming events and provide scheduling functionality.
"""
def __init__(self, api_key: str, base_url: str):
self.session = requests.Session()
self.session.headers.update({"x-portal-apikey": api_key})
self.base_url = base_url
self.monitored_events = {}
def _request(self, endpoint: str, params: dict) -> dict:
response = self.session.get(f"{self.base_url}{endpoint}", params=params)
response.raise_for_status()
return response.json()
def get_upcoming_events(
self,
sport_id: int,
hours_ahead: int = 24,
league_ids: List[int] = None
) -> List[Dict]:
"""
Get events starting within specified hours.
Args:
sport_id: Sport ID
hours_ahead: How many hours ahead to look
league_ids: Optional list of league IDs to filter
"""
params = {
"sport_id": sport_id,
"is_have_odds": True,
"event_type": "prematch"
}
if league_ids:
params["league_ids"] = ",".join(map(str, league_ids))
result = self._request("/kit/v1/markets", params)
events = result.get("events", [])
now = datetime.utcnow()
cutoff = now + timedelta(hours=hours_ahead)
upcoming = []
for event in events:
try:
starts = datetime.fromisoformat(event["starts"].replace("Z", "+00:00"))
starts_naive = starts.replace(tzinfo=None)
if now <= starts_naive <= cutoff:
event["time_until"] = starts_naive - now
event["starts_local"] = starts_naive
upcoming.append(event)
except (KeyError, ValueError):
continue
# Sort by start time
upcoming.sort(key=lambda x: x["starts_local"])
return upcoming
def get_todays_events(self, sport_id: int, league_ids: List[int] = None) -> List[Dict]:
"""Get all events starting today."""
return self.get_upcoming_events(sport_id, hours_ahead=24, league_ids=league_ids)
def get_events_next_hour(self, sport_id: int) -> List[Dict]:
"""Get events starting in the next hour."""
return self.get_upcoming_events(sport_id, hours_ahead=1)
def add_to_watchlist(
self,
event_id: int,
reminder_minutes: int = 30,
notes: str = ""
) -> None:
"""Add an event to the watchlist."""
self.monitored_events[event_id] = {
"reminder_minutes": reminder_minutes,
"notes": notes,
"added_at": datetime.utcnow().isoformat()
}
def check_reminders(self) -> List[Dict]:
"""Check if any monitored events need reminders."""
reminders = []
now = datetime.utcnow()
for event_id, config in self.monitored_events.items():
try:
# Get event details
details = self._request("/kit/v1/details", {"event_id": event_id})
events = details.get("events", [])
if events:
event = events[0]
starts = datetime.fromisoformat(event["starts"].replace("Z", "+00:00"))
starts_naive = starts.replace(tzinfo=None)
time_until = starts_naive - now
reminder_delta = timedelta(minutes=config["reminder_minutes"])
if time_until <= reminder_delta and time_until > timedelta(0):
reminders.append({
"event_id": event_id,
"home": event["home"],
"away": event["away"],
"league": event["league_name"],
"starts": event["starts"],
"minutes_until": int(time_until.total_seconds() / 60),
"notes": config["notes"]
})
except (requests.exceptions.RequestException, KeyError):
continue
return reminders
def print_daily_schedule(self, sport_id: int, league_ids: List[int] = None) -> None:
"""Print a formatted daily schedule."""
events = self.get_todays_events(sport_id, league_ids)
print(f"\n{'=' * 60}")
print(f"DAILY SCHEDULE - {datetime.utcnow().strftime('%Y-%m-%d')}")
print(f"{'=' * 60}")
if not events:
print("No events scheduled for today.")
return
current_hour = None
for event in events:
hour = event["starts_local"].strftime("%H:00")
if hour != current_hour:
current_hour = hour
print(f"\n[{hour}]")
time_str = event["starts_local"].strftime("%H:%M")
time_until = str(event["time_until"]).split(".")[0]
print(f" {time_str} | {event['home']} vs {event['away']}")
print(f" | League: {event['league_name']}")
print(f" | In: {time_until}")
# Show money line if available
try:
ml = event["periods"]["num_0"]["money_line"]
print(f" | Odds: H:{ml['home']} D:{ml['draw']} A:{ml['away']}")
except Exception:
pass
print(f"\n{'=' * 60}")
print(f"Total events: {len(events)}")
def run_scheduler(self, sport_id: int, check_interval: int = 60) -> None:
"""Run the scheduler loop."""
print("Starting event scheduler...")
print("Press Ctrl+C to stop.\n")
try:
while True:
# Check reminders
reminders = self.check_reminders()
for reminder in reminders:
print(f"\n{'!' * 40}")
print(f"REMINDER: Event starting in {reminder['minutes_until']} minutes!")
print(f" {reminder['home']} vs {reminder['away']}")
print(f" League: {reminder['league']}")
if reminder['notes']:
print(f" Notes: {reminder['notes']}")
print(f"{'!' * 40}\n")
# Show events in next hour
next_hour = self.get_events_next_hour(sport_id)
if next_hour:
print(f"[{datetime.utcnow().strftime('%H:%M:%S')}] Events in next hour: {len(next_hour)}")
time.sleep(check_interval)
except KeyboardInterrupt:
print("\nScheduler stopped.")
# Usage Example
if __name__ == "__main__":
scheduler = EventScheduler(
api_key="YOUR_API_KEY",
base_url="https://pinnacle-odds-api.hgapi.top"
)
# Print today's schedule
scheduler.print_daily_schedule(sport_id=1)
# Add events to watchlist
upcoming = scheduler.get_events_next_hour(sport_id=1)
for event in upcoming[:3]:
scheduler.add_to_watchlist(
event_id=event["event_id"],
reminder_minutes=15,
notes=f"Important match: {event['home']} vs {event['away']}"
)
# Run scheduler
scheduler.run_scheduler(sport_id=1, check_interval=60)Workflow 5: Multi-Sport Dashboard
Build a dashboard showing data across multiple sports.
python
import requests
import json
import time
from datetime import datetime
from collections import defaultdict
from typing import Dict, List
class MultiSportDashboard:
"""
Dashboard for monitoring multiple sports simultaneously.
"""
# Common sport IDs
SPORTS = {
1: "Soccer",
2: "Tennis",
3: "Basketball",
4: "Ice Hockey",
5: "American Football",
6: "Baseball",
7: "MMA",
8: "Handball",
9: "Volleyball",
10: "Cricket"
}
def __init__(self, api_key: str, base_url: str):
self.session = requests.Session()
self.session.headers.update({"x-portal-apikey": api_key})
self.base_url = base_url
self.since_cache = {}
def _request(self, endpoint: str, params: dict = None) -> dict:
response = self.session.get(f"{self.base_url}{endpoint}", params=params)
response.raise_for_status()
return response.json()
def get_all_sports(self) -> List[Dict]:
"""Get list of all available sports."""
return self._request("/kit/v1/sports")
def get_sport_summary(self, sport_id: int) -> Dict:
"""Get summary statistics for a sport."""
params = {
"sport_id": sport_id,
"is_have_odds": True
}
result = self._request("/kit/v1/markets", params)
self.since_cache[sport_id] = result.get("last")
events = result.get("events", [])
summary = {
"sport_id": sport_id,
"sport_name": result.get("sport_name", "Unknown"),
"total_events": len(events),
"prematch_events": 0,
"live_events": 0,
"open_markets": 0,
"leagues": set(),
"next_event": None
}
now = datetime.utcnow()
for event in events:
# Count by type
if event.get("event_type") == "prematch":
summary["prematch_events"] += 1
elif event.get("event_type") == "live":
summary["live_events"] += 1
# Count open markets
if event.get("is_have_open_markets"):
summary["open_markets"] += 1
# Track leagues
summary["leagues"].add(event.get("league_name", ""))
# Find next event
try:
starts = datetime.fromisoformat(event["starts"].replace("Z", "+00:00"))
starts_naive = starts.replace(tzinfo=None)
if starts_naive > now:
if summary["next_event"] is None or starts_naive < summary["next_event"]["starts"]:
summary["next_event"] = {
"home": event["home"],
"away": event["away"],
"league": event["league_name"],
"starts": starts_naive
}
except (KeyError, ValueError):
pass
summary["leagues"] = len(summary["leagues"])
return summary
def get_all_sports_summary(self) -> List[Dict]:
"""Get summary for all active sports."""
summaries = []
for sport_id in self.SPORTS.keys():
try:
summary = self.get_sport_summary(sport_id)
if summary["total_events"] > 0:
summaries.append(summary)
time.sleep(0.3) # Rate limiting
except requests.exceptions.RequestException:
continue
return summaries
def print_dashboard(self) -> None:
"""Print a formatted dashboard."""
print("\n" + "=" * 70)
print(f"PINBOOK ODDS DASHBOARD - {datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')} UTC")
print("=" * 70)
summaries = self.get_all_sports_summary()
# Header
print(f"\n{'Sport':<20} {'Events':>8} {'Prematch':>10} {'Live':>6} {'Leagues':>10}")
print("-" * 70)
total_events = 0
total_prematch = 0
total_live = 0
for summary in summaries:
print(f"{summary['sport_name']:<20} "
f"{summary['total_events']:>8} "
f"{summary['prematch_events']:>10} "
f"{summary['live_events']:>6} "
f"{summary['leagues']:>10}")
total_events += summary["total_events"]
total_prematch += summary["prematch_events"]
total_live += summary["live_events"]
print("-" * 70)
print(f"{'TOTAL':<20} {total_events:>8} {total_prematch:>10} {total_live:>6}")
# Next events
print("\n" + "-" * 70)
print("UPCOMING EVENTS:")
print("-" * 70)
upcoming = []
for summary in summaries:
if summary["next_event"]:
upcoming.append({
"sport": summary["sport_name"],
**summary["next_event"]
})
upcoming.sort(key=lambda x: x["starts"])
for event in upcoming[:5]:
time_until = event["starts"] - datetime.utcnow()
minutes = int(time_until.total_seconds() / 60)
print(f"\n [{event['sport']}] {event['home']} vs {event['away']}")
print(f" League: {event['league']}")
print(f" Starts in: {minutes} minutes")
print("\n" + "=" * 70)
def run_live_dashboard(self, refresh_interval: int = 60) -> None:
"""Run live updating dashboard."""
try:
while True:
# Clear screen (optional)
# os.system('cls' if os.name == 'nt' else 'clear')
self.print_dashboard()
print(f"\nRefreshing in {refresh_interval} seconds... (Ctrl+C to stop)")
time.sleep(refresh_interval)
except KeyboardInterrupt:
print("\n\nDashboard stopped.")
def get_live_events(self) -> List[Dict]:
"""Get all currently live events across sports."""
live_events = []
for sport_id in self.SPORTS.keys():
try:
params = {
"sport_id": sport_id,
"event_type": "live",
"is_have_odds": True
}
result = self._request("/kit/v1/markets", params)
events = result.get("events", [])
for event in events:
live_events.append({
"sport": result.get("sport_name"),
"home": event["home"],
"away": event["away"],
"league": event["league_name"],
"periods": event.get("periods", {})
})
time.sleep(0.3)
except requests.exceptions.RequestException:
continue
return live_events
# Usage Example
if __name__ == "__main__":
dashboard = MultiSportDashboard(
api_key="YOUR_API_KEY",
base_url="https://pinnacle-odds-api.hgapi.top"
)
# Print one-time dashboard
dashboard.print_dashboard()
# Or run live dashboard
# dashboard.run_live_dashboard(refresh_interval=60)