""" SportPesa odds scraper. CONFIRMED real endpoints (found via DevTools Network tab, Aug 2026): 1. Live events list - returns all currently live matches: GET https://www.ke.sportpesa.com/api/live/events?limit=15&offset=0 Relevant fields per event: id -> use as eventId in the markets request below marketCount -> 0 means no odds available yet (not started / no live market yet); only fetch markets for events where this is > 0 kickoffTimeUTC -> ISO timestamp competitors -> [{"name": ""}, {"name": ""}] status -> "Started" / "NotStarted" / etc. NOTE: with limit=15, you only get 15 events per page ("loadMore": true means there are more). Increase `limit` or add pagination if you want to scan every live match rather than just the first page. 2. Markets/odds for one event: GET https://www.ke.sportpesa.com/api/live/event/markets ?eventId=&type=194%230%230,147%230%230,105%232.50%230,138%230%230&sportId=1 Confirmed response shape: { "markets": [ { "eventId": 1303139, "markets": [ {"id": 194, "name": "1x2", "selections": [ {"name": "", "odds": "1.85"}, {"name": "draw", "odds": "3.00"}, {"name": "", "odds": "4.50"} ]}, ... "Total" (over/under), "Both teams to score", "Double chance" ... ] } ] } Auth: neither endpoint appeared to require auth headers in what you copied, but if you start getting 403s or empty results, you'll need to add your session Cookie to HEADERS below (keep it in a local .env, never commit it). """ import requests from typing import List, Dict from .base import BaseScraper, OddsEntry EVENTS_URL = "https://www.ke.sportpesa.com/api/live/events" MARKETS_URL = "https://www.ke.sportpesa.com/api/live/event/markets" HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", # TODO: add Cookie/Authorization here if requests start failing - see note above } # Market type codes used in the `type=` query param on the markets request. MARKET_TYPES = "194%230%230,147%230%230,105%232.50%230,138%230%230" EVENTS_PAGE_LIMIT = 50 # how many live events to pull per request class SportPesaScraper(BaseScraper): name = "sportpesa" def fetch_odds(self, sport: str = "football") -> List[OddsEntry]: live_events = self._fetch_live_events() entries: List[OddsEntry] = [] for event in live_events: if event["market_count"] <= 0: continue # no odds available yet for this match try: entries.extend(self._fetch_event_markets(event, sport)) except Exception as e: print(f"[sportpesa] failed to fetch markets for event {event['id']}: {e}") if not entries: # nothing live right now (or endpoints unreachable) - fall back # to mock data so the rest of the pipeline still has something # to work with while you're testing return self._mock_odds(sport) return entries def _fetch_live_events(self) -> List[Dict]: params = {"limit": EVENTS_PAGE_LIMIT, "offset": 0} resp = requests.get(EVENTS_URL, headers=HEADERS, params=params, timeout=10) resp.raise_for_status() data = resp.json() events = [] for e in data.get("events", []): competitors = e.get("competitors", []) if len(competitors) != 2: continue events.append({ "id": e["id"], "home": competitors[0]["name"], "away": competitors[1]["name"], "kickoff_utc": e.get("kickoffTimeUTC", ""), "market_count": e.get("marketCount", 0), }) return events def _fetch_event_markets(self, event: Dict, sport: str) -> List[OddsEntry]: params = { "eventId": event["id"], "type": MARKET_TYPES, "sportId": 1, } resp = requests.get(MARKETS_URL, headers=HEADERS, params=params, timeout=10) resp.raise_for_status() data = resp.json() return self._parse_markets_response(data, event, sport) def _parse_markets_response(self, data: dict, event: Dict, sport: str) -> List[OddsEntry]: entries: List[OddsEntry] = [] home_team = event["home"] away_team = event["away"] kickoff = event["kickoff_utc"] for event_block in data.get("markets", []): for market in event_block.get("markets", []): market_name = market.get("name", "").lower() if market_name == "1x2": outcome_map = {home_team.lower(): "home", "draw": "draw", away_team.lower(): "away"} for sel in market.get("selections", []): sel_name = sel.get("name", "").lower() outcome = outcome_map.get(sel_name, sel_name) entries.append(self._make_entry( sport, home_team, away_team, kickoff, "match_winner", outcome, sel.get("odds") )) elif market_name == "total": line = market.get("specialValue", "") for sel in market.get("selections", []): outcome = sel.get("name", "").lower() # "over 2.5" / "under 2.5" entries.append(self._make_entry( sport, home_team, away_team, kickoff, f"total_{line}", outcome, sel.get("odds") )) elif market_name == "both teams to score": for sel in market.get("selections", []): outcome = sel.get("name", "").lower() # "yes" / "no" entries.append(self._make_entry( sport, home_team, away_team, kickoff, "btts", outcome, sel.get("odds") )) # double chance skipped for now - add later if useful for arbs return entries def _make_entry(self, sport, home_team, away_team, kickoff, market, outcome, odds_str) -> OddsEntry: return OddsEntry( bookmaker=self.name, sport=sport, home_team=home_team, away_team=away_team, kickoff_utc=kickoff, market=market, outcome=outcome, odds=float(odds_str), ) def _mock_odds(self, sport: str) -> List[OddsEntry]: """Fallback so the pipeline still has data if the live endpoints return nothing right now.""" return [ OddsEntry(self.name, sport, "Gor Mahia", "AFC Leopards", "2026-08-23T15:00:00Z", "match_winner", "home", 2.30), OddsEntry(self.name, sport, "Gor Mahia", "AFC Leopards", "2026-08-23T15:00:00Z", "match_winner", "draw", 3.10), OddsEntry(self.name, sport, "Gor Mahia", "AFC Leopards", "2026-08-23T15:00:00Z", "match_winner", "away", 3.40), ]