Sources

Read match statistics and odds from the feeds that publish them.

BaseOddsSchema

Bases: _BaseSchema

Base schema for odds snapshots.

Examples:

>>> from sportsbet.sources import BaseOddsSchema, optional_col, required_col
>>>
>>> class MyOddsSchema(BaseOddsSchema):
...     'The odds of a feed of your own.'
...
...     home_team: str = required_col()
...     away_team: str = required_col()
...     provider: str = required_col()
...     home_win: float = optional_col(include=['preplay', 'inplay'], fixed=False)
...     away_win: float = optional_col(include=['preplay', 'inplay'], fixed=False)

check_postplay_missing_odds(df) classmethod

Check that post-match snapshots carry no odds.

Source code in src/sportsbet/sources/_schema.py

@pa.dataframe_check
@classmethod
def check_postplay_missing_odds(cls, df: pd.DataFrame) -> pd.Series:
    """Check that post-match snapshots carry no odds."""
    odds_cols = cls.list_odds_cols()
    if not odds_cols:
        return pd.Series(True, index=df.index)
    is_post = df['event_status'].eq('postplay')
    ok_post = df.loc[is_post, odds_cols].isna().all(axis=1)
    out = pd.Series(True, index=df.index)
    out.loc[is_post] = ok_post
    return out

list_odds_cols() classmethod

Return the odds (market) columns.

Source code in src/sportsbet/sources/_schema.py

@classmethod
def list_odds_cols(cls) -> list[str]:
    """Return the odds (market) columns."""
    schema_cols = list(cls.to_schema().columns.keys())
    return [col for col in schema_cols if col not in cls.list_snapshot_cols() and col != 'provider']

BaseOddsSource

Bases: BaseSource

The abstract base class for odds sources.

Examples:

>>> import io
>>> import pandas as pd
>>> from sportsbet.sources import BaseOddsSource, RawItem, RawPayload
>>>
>>> class MyOdds(BaseOddsSource):
...     '''Odds from a feed of your own.'''
...
...     name = 'my_odds'
...     sport = 'soccer'
...
...     def list_index_items(self, selection=None):
...         return [RawItem(source=self.name, key='seasons', url='https://example.com/seasons.json')]
...
...     def read_catalogue(self, payloads):
...         return [{'league': 'Ruritania', 'division': 1, 'year': 2025}]
...
...     def list_required_items(self, params, schedule=None):
...         return [RawItem(source=self.name, key=f'odds_{param["year"]}',
...                         url=f'https://example.com/odds/{param["year"]}.csv')
...                 for param in params]
...
...     def to_snapshots(self, payloads):
...         odds = pd.read_csv(io.BytesIO(payloads[0].content))
...         odds['date'] = pd.to_datetime(odds['date'], utc=True)
...         return odds.assign(event_status='preplay', event_time=0)
>>>
>>> source = MyOdds()
>>> source.kind
'odds'
>>> csv = b'date,league,division,year,home_team,away_team,provider,home_win,draw,away_win\n'
>>> csv += b'2025-08-16,Ruritania,1,2025,A,B,acme,1.8,3.4,4.2\n'
>>> item = source.list_required_items([{'year': 2025}])[0]
>>> snapshots = source.to_snapshots([RawPayload(item=item, content=csv)])
>>> # The markets are the columns, and the provider is a column too.
>>> snapshots[['provider', 'home_win', 'event_status']].to_dict('records')
[{'provider': 'acme', 'home_win': 1.8, 'event_status': 'preplay'}]

BaseSource

Bases: ABC

The abstract base class for data sources.

A source declares the raw items a selection of parameters needs. It turns the returned payloads into long snapshots. sport names the one sport it carries. sport is None for a vendor that carries several sports and takes the sport of the source it is paired with. The list_available_params method returns what the source publishes.

Examples:

>>> from sportsbet.sources import FootballDataStats, OddsApi
>>> stats = FootballDataStats()
>>> stats.name, stats.kind, stats.sport
('football_data', 'stats', 'soccer')
>>> # A vendor selling every sport carries none of its own, and takes the sport it is paired with.
>>> OddsApi(key_env='ODDS_API_KEY', markets=['h2h']).sport is None
True
>>> # Asking what a source publishes declares the items to read.
>>> items = stats.list_index_items()
>>> items[0].source
'football_data'

list_available_params()

Return the league, division and season combinations the source publishes.

What a source publishes depends on how it is configured.

Returns:

Name	Type	Description
params	list[dict]	The available league, division and year combinations.

Source code in src/sportsbet/sources/_base.py

def list_available_params(self: Self) -> list[dict]:
    """Return the league, division and season combinations the source publishes.

    What a source publishes depends on how it is configured.

    Returns:
        params:
            The available `league`, `division` and `year` combinations.
    """
    return self.read_catalogue(fetch_payloads(self.list_index_items(), self.request_url))

list_fixtures_items(params, schedule=None)

Return the raw items the upcoming matches need.

By default it returns the same items training needs. That suits a source whose season file already carries the matches still to be played. It also suits an odds source that prices whatever the schedule lists. A source whose upcoming matches live elsewhere, in a separate fixtures file or the season in progress, overrides this method.

Parameters:

Name	Type	Description	Default
params	list[dict]	The selected parameter combinations.	required
schedule	DataFrame | None	The upcoming matches, for an odds source that prices by instant.	None

Returns:

Name	Type	Description
items	list[RawItem]	The items whose payloads yield the upcoming matches.

Source code in src/sportsbet/sources/_base.py

def list_fixtures_items(self: Self, params: list[dict], schedule: pd.DataFrame | None = None) -> list[RawItem]:
    """Return the raw items the upcoming matches need.

    By default it returns the same items training needs. That suits a source whose season file already carries the
    matches still to be played. It also suits an odds source that prices whatever the schedule lists. A source
    whose upcoming matches live elsewhere, in a separate fixtures file or the season in progress, overrides this
    method.

    Args:
        params:
            The selected parameter combinations.

        schedule:
            The upcoming matches, for an odds source that prices by instant.

    Returns:
        items:
            The items whose payloads yield the upcoming matches.
    """
    return self.list_required_items(params, schedule)

list_index_items(selection=None) abstractmethod

Return the items needed to discover what the source publishes.

Parameters:

Name	Type	Description	Default
selection	ParamGrid | None	What is being looked for. None asks for everything.	None

Returns:

Name	Type	Description
items	list[RawItem]	The items whose payloads describe the catalogue.

Source code in src/sportsbet/sources/_base.py

@abstractmethod
def list_index_items(self: Self, selection: ParamGrid | None = None) -> list[RawItem]:
    """Return the items needed to discover what the source publishes.

    Args:
        selection:
            What is being looked for. `None` asks for everything.

    Returns:
        items:
            The items whose payloads describe the catalogue.
    """

list_required_items(params, schedule=None) abstractmethod

Return the raw items the selected parameters need.

Parameters:

Name	Type	Description	Default
params	list[dict]	The selected parameter combinations.	required
schedule	DataFrame | None	The matches of the selected parameters, with their kick-off instants. An odds source that addresses its prices by timestamp needs it. It is None for a source that carries its own schedule.	None

Returns:

Name	Type	Description
items	list[RawItem]	The items to read. Deterministic for the same parameters.

Source code in src/sportsbet/sources/_base.py

@abstractmethod
def list_required_items(self: Self, params: list[dict], schedule: pd.DataFrame | None = None) -> list[RawItem]:
    """Return the raw items the selected parameters need.

    Args:
        params:
            The selected parameter combinations.

        schedule:
            The matches of the selected parameters, with their kick-off instants. An odds source that addresses its
            prices by timestamp needs it. It is `None` for a source that carries its own schedule.

    Returns:
        items:
            The items to read. Deterministic for the same parameters.
    """

needs_schedule()

Return whether the source has to be told when the matches are.

A source whose events and odds arrive in the same file carries its own schedule and returns False.

Returns:

Name	Type	Description
needed	bool	Whether list_required_items needs a schedule.

Source code in src/sportsbet/sources/_base.py

def needs_schedule(self: Self) -> bool:
    """Return whether the source has to be told when the matches are.

    A source whose events and odds arrive in the same file carries its own schedule and returns `False`.

    Returns:
        needed:
            Whether `list_required_items` needs a schedule.
    """
    return False

read_catalogue(payloads) abstractmethod

Return the parameter combinations the index payloads describe.

Parameters:

Name	Type	Description	Default
payloads	list[RawPayload]	The payloads of the index items.	required

Returns:

Name	Type	Description
params	list[dict]	The available league, division and year combinations.

Source code in src/sportsbet/sources/_base.py

@abstractmethod
def read_catalogue(self: Self, payloads: list[RawPayload]) -> list[dict]:
    """Return the parameter combinations the index payloads describe.

    Args:
        payloads:
            The payloads of the index items.

    Returns:
        params:
            The available `league`, `division` and `year` combinations.
    """

request_url(item)

Return the URL to fetch an item from, with the credential added at the moment of the request.

Parameters:

Name	Type	Description	Default
item	RawItem	The item to fetch.	required

Returns:

Name	Type	Description
url	str	Where to fetch it from.

Source code in src/sportsbet/sources/_base.py

def request_url(self: Self, item: RawItem) -> str:
    """Return the URL to fetch an item from, with the credential added at the moment of the request.

    Args:
        item:
            The item to fetch.

    Returns:
        url:
            Where to fetch it from.
    """
    return item.url

to_snapshots(payloads) abstractmethod

Transform the raw payloads into a long snapshot table.

Parameters:

Name	Type	Description	Default
payloads	list[RawPayload]	The payloads of the required items.	required

Returns:

Name	Type	Description
snapshots	DataFrame	The long snapshots.

Source code in src/sportsbet/sources/_base.py

@abstractmethod
def to_snapshots(self: Self, payloads: list[RawPayload]) -> pd.DataFrame:
    """Transform the raw payloads into a long snapshot table.

    Args:
        payloads:
            The payloads of the required items.

    Returns:
        snapshots:
            The long snapshots.
    """

BaseStatsSchema

Bases: _BaseSchema

Base schema for statistics snapshots.

Examples:

>>> from sportsbet.sources import BaseStatsSchema, optional_col, required_col
>>>
>>> class MyStatsSchema(BaseStatsSchema):
...     'The statistics of a feed of your own.'
...
...     home_team: str = required_col()
...     away_team: str = required_col()
...     home_goals: float = optional_col(include=['inplay', 'postplay'], fixed=False)

BaseStatsSource

Bases: BaseSource

The abstract base class for statistics sources.

Examples:

>>> import io
>>> import pandas as pd
>>> from sportsbet.sources import BaseStatsSource, RawItem, RawPayload, derive_market_outcomes
>>> IDENTITY = ['date', 'league', 'division', 'year', 'home_team', 'away_team']
>>>
>>> class MyStats(BaseStatsSource):
...     '''Statistics from a feed of your own.'''
...
...     name = 'my_stats'
...     sport = 'soccer'
...
...     def list_index_items(self, selection=None):
...         return [RawItem(source=self.name, key='seasons', url='https://example.com/seasons.json')]
...
...     def read_catalogue(self, payloads):
...         return [{'league': 'Ruritania', 'division': 1, 'year': 2025}]
...
...     def list_required_items(self, params, schedule=None):
...         return [RawItem(source=self.name, key=f'Ruritania_1_{param["year"]}',
...                         url=f'https://example.com/{param["year"]}.csv')
...                 for param in params]
...
...     def to_snapshots(self, payloads):
...         games = pd.read_csv(io.BytesIO(payloads[0].content))
...         games['date'] = pd.to_datetime(games['date'], utc=True)
...         preplay = games[IDENTITY].assign(event_status='preplay', event_time=0,
...                                          home_form=games['home_form'])
...         postplay = games[IDENTITY].assign(event_status='postplay', event_time=0)
...         outcomes = derive_market_outcomes(games['home_goals'], games['away_goals'], ['home_win', 'draw',
...                                                                               'away_win'])
...         postplay = pd.concat([postplay, outcomes], axis=1)
...         return pd.concat([preplay, postplay], ignore_index=True)
>>>
>>> source = MyStats()
>>> # It says what it needs, and the dataloader reads it.
>>> source.list_required_items([{'year': 2025}])[0].url
'https://example.com/2025.csv'
>>> csv = b'date,league,division,year,home_team,away_team,home_form,home_goals,away_goals\n'
>>> csv += b'2025-08-16,Ruritania,1,2025,A,B,0.5,2,1\n'
>>> item = source.list_required_items([{'year': 2025}])[0]
>>> snapshots = source.to_snapshots([RawPayload(item=item, content=csv)])
>>> sorted(snapshots['event_status'].unique())
['postplay', 'preplay']
>>> snapshots.loc[snapshots['event_status'].eq('postplay'), 'home_win'].item()
1.0

EuroLeagueStats

Bases: BaseStatsSource

The EuroLeague schedule and final scores from its official API, with home and away markets.

Read more in the user guide.

Examples:

>>> from sportsbet.dataloaders import DataLoader
>>> from sportsbet.sources import EuroLeagueStats, OddsApi
>>> source = EuroLeagueStats()
>>> source.name, source.kind, source.sport
('euroleague', 'stats', 'basketball')
>>> # A whole season arrives in one request, and asking what it publishes costs one more.
>>> len(source.list_index_items())
1
>>> # Pair the statistics with an odds source.
>>> dataloader = DataLoader(
...     param_grid={'league': ['Euroleague'], 'division': [1], 'year': [2025]},
...     stats=source,
...     odds=OddsApi(key_env='ODDS_API_KEY', markets=['h2h']),
... )
>>> dataloader.sport_
'basketball'

list_index_items(selection=None)

Return the seasons the competition publishes.

Source code in src/sportsbet/sources/_stats/_euroleague.py

def list_index_items(self: Self, selection: ParamGrid | None = None) -> list[RawItem]:
    """Return the seasons the competition publishes."""
    return [RawItem(source=self.name, key=SEASONS_KEY, url=SEASONS_URL)]

list_required_items(params, schedule=None)

Return one item per selected season.

Source code in src/sportsbet/sources/_stats/_euroleague.py

def list_required_items(self: Self, params: list[dict], schedule: pd.DataFrame | None = None) -> list[RawItem]:
    """Return one item per selected season."""
    return [
        RawItem(
            source=self.name,
            key=f'{LEAGUE}_{param["division"]}_{param["year"]}',
            url=GAMES_URL.format(season=param['year'] - 1),
        )
        for param in params
        if param['league'] == LEAGUE
    ]

read_catalogue(payloads)

Return the seasons the competition publishes, each named by the year it ends in.

Source code in src/sportsbet/sources/_stats/_euroleague.py

def read_catalogue(self: Self, payloads: list[RawPayload]) -> list[dict]:
    """Return the seasons the competition publishes, each named by the year it ends in."""
    if not payloads:
        return []
    seasons = json.loads(payloads[0].content).get('data', [])
    return sorted(
        ({'league': LEAGUE, 'division': DIVISION, 'year': int(season['year']) + 1} for season in seasons),
        key=lambda params: params['year'],
    )

to_snapshots(payloads)

Transform the seasons into the long statistics snapshots.

Source code in src/sportsbet/sources/_stats/_euroleague.py

def to_snapshots(self: Self, payloads: list[RawPayload]) -> pd.DataFrame:
    """Transform the seasons into the long statistics snapshots."""
    frames = []
    for payload in payloads:
        year = int(payload.item.key.rsplit('_', 1)[-1])
        games = _games(payload.content, year)
        if not games.empty:
            frames.append(_snapshots(games))
    if not frames:
        return pd.DataFrame()
    return pd.concat(frames, ignore_index=True).replace({np.nan: None}).infer_objects()

FootballDataOdds

Bases: _FootballDataSource, BaseOddsSource

The pre-match closing odds of the market average and market maximum from the football-data.co.uk feed.

Read more in the user guide.

Examples:

>>> from sportsbet.sources import FootballDataOdds
>>> source = FootballDataOdds()
>>> source.name, source.kind, source.sport
('football_data', 'odds', 'soccer')
>>> # It reads the same upstream files as the statistics.
>>> from sportsbet.sources import FootballDataStats
>>> stats_items = FootballDataStats().list_index_items({'league': ['Italy']})
>>> source.list_index_items({'league': ['Italy']}) == stats_items
True

FootballDataStats

Bases: _FootballDataSource, BaseStatsSource

The soccer schedule, results and match statistics from the football-data.co.uk feed.

Read more in the user guide.

Examples:

>>> from sportsbet.dataloaders import DataLoader
>>> from sportsbet.sources import FootballDataOdds, FootballDataStats
>>> source = FootballDataStats()
>>> source.name, source.kind, source.sport
('football_data', 'stats', 'soccer')
>>> # It declares what it would read to learn what it publishes, and reads nothing.
>>> [item.key for item in source.list_index_items({'league': ['Italy']})]
['index_Italy']
>>> # Hand it to a dataloader, together with wherever the odds come from.
>>> dataloader = DataLoader(
...     param_grid={'league': ['Italy'], 'division': [1], 'year': [2024]},
...     stats=source,
...     odds=FootballDataOdds(),
... )
>>> dataloader.sport_
'soccer'

NBAStats

Bases: BaseStatsSource

The NBA schedule and final scores from ESPN, updated through a season, with home and away markets.

Read more in the user guide.

Examples:

>>> from sportsbet.dataloaders import DataLoader
>>> from sportsbet.sources import NBAStats, OddsApi
>>> source = NBAStats()
>>> source.name, source.kind, source.sport
('nba', 'stats', 'basketball')
>>> # A season is named by the year it ends in, so 2026 is the 2025-26 season.
>>> dataloader = DataLoader(
...     param_grid={'league': ['NBA'], 'year': [2026]},
...     stats=source,
...     odds=OddsApi(key_env='ODDS_API_KEY', markets=['h2h']),
... )
>>> dataloader.sport_
'basketball'
>>> # A league is a source. The same sport is the same dataloader.
>>> from sportsbet.sources import EuroLeagueStats
>>> NBAStats().sport == EuroLeagueStats().sport
True

list_index_items(selection=None)

Return the seasons the competition publishes.

Source code in src/sportsbet/sources/_stats/_nba.py

def list_index_items(self: Self, selection: ParamGrid | None = None) -> list[RawItem]:
    """Return the seasons the competition publishes."""
    return [RawItem(source=self.name, key=SEASONS_KEY, url=SEASONS_URL)]

list_required_items(params, schedule=None)

Return one item per month of each selected season.

Source code in src/sportsbet/sources/_stats/_nba.py

def list_required_items(self: Self, params: list[dict], schedule: pd.DataFrame | None = None) -> list[RawItem]:
    """Return one item per month of each selected season."""
    items = []
    for param in params:
        if param['league'] != LEAGUE:
            continue
        year = param['year']
        for offset, month in MONTHS:
            start = pd.Timestamp(year=year + offset, month=month, day=1)
            last = calendar.monthrange(start.year, month)[1]
            items.append(
                RawItem(
                    source=self.name,
                    key=f'{LEAGUE}_{param["division"]}_{year}_{start.year}{month:02d}',
                    url=GAMES_URL.format(start=f'{start.year}{month:02d}01', end=f'{start.year}{month:02d}{last}'),
                ),
            )
    return items

read_catalogue(payloads)

Return the seasons the competition publishes, each named by the year it ends in.

Source code in src/sportsbet/sources/_stats/_nba.py

def read_catalogue(self: Self, payloads: list[RawPayload]) -> list[dict]:
    """Return the seasons the competition publishes, each named by the year it ends in."""
    if not payloads:
        return []
    seasons = json.loads(payloads[0].content).get('items', [])
    years = {int(season['$ref'].rsplit('/', 1)[-1].split('?')[0]) for season in seasons if '$ref' in season}
    return sorted(
        ({'league': LEAGUE, 'division': DIVISION, 'year': year} for year in years),
        key=lambda params: params['year'],
    )

to_snapshots(payloads)

Transform the months into the long statistics snapshots.

Source code in src/sportsbet/sources/_stats/_nba.py

def to_snapshots(self: Self, payloads: list[RawPayload]) -> pd.DataFrame:
    """Transform the months into the long statistics snapshots."""
    seasons: dict[int, list[pd.DataFrame]] = {}
    for payload in payloads:
        year = int(payload.item.key.split('_')[2])
        games = _games(payload.content, year)
        if not games.empty:
            seasons.setdefault(year, []).append(games)
    frames = []
    for year in sorted(seasons):
        games = pd.concat(seasons[year], ignore_index=True).sort_values('date').reset_index(drop=True)
        frames.append(_snapshots(games))
    if not frames:
        return pd.DataFrame()
    return pd.concat(frames, ignore_index=True).replace({np.nan: None}).infer_objects()

OddsApi(key_env, markets=None, regions=None, moments=None)

Bases: BaseOddsSource

The time-stamped odds of The Odds API, quoting a match at each moment it reaches.

The key is read from the environment variable named by key_env.

Read more in the user guide.

Parameters:

Name	Type	Description	Default
key_env	str	The name of the environment variable holding your API key.	required
markets	list[str] | None	The markets to price, e.g. ['h2h', 'totals']. The default None uses both.	None
regions	list[str] | None	The bookmaker regions, e.g. ['eu', 'uk']. The default None uses ['eu'].	None
moments	list[tuple[str, int]] | None	The moments of a match to price, as (event_status, minutes) pairs. The default None prices the moments the statistics carry.	None

Examples:

>>> import os
>>> from sportsbet.sources import OddsApi, RawItem
>>> os.environ['ODDS_API_KEY'] = 'secret'
>>> source = OddsApi(key_env='ODDS_API_KEY', markets=['h2h'], regions=['eu'])
>>> source.name, source.kind
('odds_api', 'odds')
>>> # It carries no sport of its own and takes the sport it is paired with.
>>> source.sport is None
True
>>> # The key is read from the environment when the request is made.
>>> item = RawItem(source='odds_api', key='snapshot', url='https://api.the-odds-api.com/v4/sports?all=true')
>>> 'secret' in item.url
False
>>> source.request_url(item)
'https://api.the-odds-api.com/v4/sports?all=true&apiKey=secret'

Source code in src/sportsbet/sources/_odds/_odds_api.py

def __init__(
    self: Self,
    key_env: str,
    markets: list[str] | None = None,
    regions: list[str] | None = None,
    moments: list[tuple[str, int]] | None = None,
) -> None:
    self.key_env = key_env
    self.markets = markets
    self.regions = regions
    self.moments = moments

list_index_items(selection=None)

Return the catalogue of the vendor.

Source code in src/sportsbet/sources/_odds/_odds_api.py

def list_index_items(self: Self, selection: ParamGrid | None = None) -> list[RawItem]:
    """Return the catalogue of the vendor."""
    return [RawItem(source=self.name, key=SPORTS_KEY, url=f'{SPORTS_URL}?all=true')]

list_required_items(params, schedule=None)

Return one item per snapshot the selected matches need, matches kicking off together sharing one.

Source code in src/sportsbet/sources/_odds/_odds_api.py

def list_required_items(self: Self, params: list[dict], schedule: pd.DataFrame | None = None) -> list[RawItem]:
    """Return one item per snapshot the selected matches need, matches kicking off together sharing one."""
    if schedule is None or schedule.empty:
        return []
    _, _, moments = self._settings()
    if self.moments is None:
        moments = self._moments(schedule)
    sports = {(league, division): sport for sport, (league, division) in LEAGUES_MAPPING.items()}
    now = _now()

    items: list[RawItem] = []
    for (league, division, year), matches in schedule.groupby(['league', 'division', 'year']):
        sport = sports.get((league, division))
        if sport is None:
            continue
        for event_status, minutes in moments:
            offset = CLOSING_OFFSET if event_status == 'preplay' else pd.Timedelta(minutes=minutes)
            snapshots = (matches['date'] + offset).drop_duplicates()
            snapshots = snapshots[(snapshots >= HISTORICAL_START) & (snapshots < now)]
            for snapshot in sorted(snapshots):
                query = urlencode({**self._query(), 'date': _timestamp(snapshot)})
                url = f'{HISTORICAL_URL.format(sport=sport)}?{query}'
                key = DELIMITER.join([sport, str(year), _key_timestamp(snapshot), event_status, str(minutes)])
                items.append(RawItem(source=self.name, key=key, url=url))
        live = f'{LIVE_URL.format(sport=sport)}?{urlencode(self._query())}'
        live_key = DELIMITER.join([sport, str(year), LIVE_KEY])
        items.append(RawItem(source=self.name, key=live_key, url=live))
    return items

needs_schedule()

Return True.

Source code in src/sportsbet/sources/_odds/_odds_api.py

def needs_schedule(self: Self) -> bool:
    """Return `True`."""
    return True

read_catalogue(payloads)

Return the combinations the vendor covers, the years being its historical coverage.

Source code in src/sportsbet/sources/_odds/_odds_api.py

def read_catalogue(self: Self, payloads: list[RawPayload]) -> list[dict]:
    """Return the combinations the vendor covers, the years being its historical coverage."""
    if not payloads:
        return []
    sports = json.loads(payloads[0].content)
    years = range(FIRST_YEAR, _now().year + 2)
    return [
        {'league': league, 'division': division, 'year': year}
        for sport in sports
        if (mapped := LEAGUES_MAPPING.get(sport['key'])) is not None
        for league, division in [mapped]
        for year in years
    ]

request_url(item)

Return the URL to fetch an item from, with the key read from the environment and added.

Parameters:

Name	Type	Description	Default
item	RawItem	The item to fetch.	required

Returns:

Name	Type	Description
url	str	Where to fetch it from.

Source code in src/sportsbet/sources/_odds/_odds_api.py

def request_url(self: Self, item: RawItem) -> str:
    """Return the URL to fetch an item from, with the key read from the environment and added.

    Args:
        item:
            The item to fetch.

    Returns:
        url:
            Where to fetch it from.
    """
    separator = '&' if '?' in item.url else '?'
    return f'{item.url}{separator}apiKey={os.environ[self.key_env]}'

to_snapshots(payloads)

Transform the vendor's responses into long odds snapshots, keeping the matches each item asked for.

Source code in src/sportsbet/sources/_odds/_odds_api.py

def to_snapshots(self: Self, payloads: list[RawPayload]) -> pd.DataFrame:
    """Transform the vendor's responses into long odds snapshots, keeping the matches each item asked for."""
    records: list[dict] = []
    for payload in payloads:
        events, endpoint = _events(payload)
        if endpoint == 'historical':
            records.extend(self._historical_records(payload, events))
        else:
            records.extend(self._live_records(payload, events))
    return pd.DataFrame(records)

RawItem(source, key, url) dataclass

A raw item to fetch, identified within its source by a key, at a URL or file:// path.

Two items with the same source and key are equal.

Parameters:

Name	Type	Description	Default
source	str	The source that declared it.	required
key	str	The item's identity within the source.	required
url	str	The URL, or a file:// path for a bundled file.	required

Examples:

>>> from sportsbet.sources import RawItem
>>> item = RawItem(source='my_stats', key='England_1_2025', url='https://example.com/2025.csv')
>>> item.key
'England_1_2025'
>>> # The same source and key make the same item.
>>> item == RawItem(source='my_stats', key='England_1_2025', url='https://example.com/2025.csv')
True

RawPayload(item, content) dataclass

A payload a source returned, pairing the fetched item with its raw bytes.

Parameters:

Name	Type	Description	Default
item	RawItem	The item that was fetched.	required
content	bytes	The bytes of the response.	required

Examples:

>>> from sportsbet.sources import RawItem, RawPayload
>>> item = RawItem(source='my_stats', key='England_1_2025', url='https://example.com/2025.csv')
>>> payload = RawPayload(item=item, content=b'date,home_team,away_team\n2025-08-16,A,B\n')
>>> payload.item.key
'England_1_2025'
>>> # The bytes are exactly what the feed returned.
>>> payload.content.splitlines()[0]
b'date,home_team,away_team'

SampleSoccerOdds

Bases: _SampleSource, BaseOddsSource

The market average and market maximum pre-match odds of the bundled soccer sample season.

Examples:

>>> from sportsbet.dataloaders import DataLoader
>>> from sportsbet.sources import SampleSoccerOdds, SampleSoccerStats
>>> source = SampleSoccerOdds()
>>> source.name, source.kind, source.sport
('sample_soccer', 'odds', 'soccer')
>>> dataloader = DataLoader(stats=SampleSoccerStats(), odds=source)
>>> X, Y, O = dataloader.extract_train_data(odds_type='market_maximum')
>>> # The providers and the markets are read from the data.
>>> dataloader.get_odds_types()
['market_average', 'market_maximum']
>>> list(Y.columns)
['home_win__postplay__0min', 'draw__postplay__0min', 'away_win__postplay__0min', 'over_2.5__postplay__0min', 'under_2.5__postplay__0min']

SampleSoccerStats

Bases: _SampleSource, BaseStatsSource

A frozen real season of the English and Spanish soccer first divisions, bundled with the library.

It carries the identity of every match, each team's form before it, the half-time score and the result. The season is finished.

Examples:

>>> from sportsbet.dataloaders import DataLoader
>>> from sportsbet.sources import SampleSoccerOdds, SampleSoccerStats
>>> source = SampleSoccerStats()
>>> source.name, source.kind, source.sport
('sample_soccer', 'stats', 'soccer')
>>> # The bundled data is known up front.
>>> source.list_available_params()
[{'division': 1, 'league': 'England', 'year': 2024}, {'division': 1, 'league': 'Spain', 'year': 2024}]
>>> dataloader = DataLoader(
...     param_grid={'league': ['England']},
...     stats=source,
...     odds=SampleSoccerOdds(),
... )
>>> X, Y, O = dataloader.extract_train_data(odds_type='market_average')
>>> # A whole season of the Premier League.
>>> len(X)
380

build_roster(data)

Return the clubs of a frame, keyed by their normalized name and valued by how they are written.

Parameters:

Name	Type	Description	Default
data	DataFrame	The snapshots whose team columns hold the clubs.	required

Returns:

Type	Description
dict[str, str]	The clubs, keyed by normalized name and valued by their original spelling.

Source code in src/sportsbet/sources/_resolver.py

def build_roster(data: pd.DataFrame) -> dict[str, str]:
    """Return the clubs of a frame, keyed by their normalized name and valued by how they are written.

    Args:
        data:
            The snapshots whose team columns hold the clubs.

    Returns:
        The clubs, keyed by normalized name and valued by their original spelling.
    """
    return {normalize_team_name(name): name for col in TEAMS_COLS for name in data[col]}

count_common_prefix(one, other)

Return how many characters two strings begin with in common.

Parameters:

Name	Type	Description	Default
one	str	The first string.	required
other	str	The second string.	required

Returns:

Type	Description
int	The number of leading characters the two share.

Source code in src/sportsbet/sources/_resolver.py

def count_common_prefix(one: str, other: str) -> int:
    """Return how many characters two strings begin with in common.

    Args:
        one:
            The first string.

        other:
            The second string.

    Returns:
        The number of leading characters the two share.
    """
    common = 0
    for character, candidate in zip(one, other, strict=False):
        if character != candidate:
            break
        common += 1
    return common

derive_market_outcomes(home_points, away_points, markets)

Derive boolean outcomes for the given markets from home and away points.

Parameters:

Name	Type	Description	Default
home_points	Series | DataFrame	The home team points per snapshot.	required
away_points	Series | DataFrame	The away team points per snapshot.	required
markets	list[str]	The markets to derive (e.g. home_win, over_2.5).	required

Returns:

Type	Description
DataFrame	A dataframe with one integer 0/1 column per requested market.

Examples:

>>> import pandas as pd
>>> from sportsbet.sources import derive_market_outcomes
>>> home_points = pd.Series([2, 1, 0])
>>> away_points = pd.Series([1, 1, 3])
>>> derive_market_outcomes(home_points, away_points, ['home_win', 'draw', 'away_win', 'over_2.5'])
   home_win  draw  away_win  over_2.5
0         1     0         0         1
1         0     1         0         0
2         0     0         1         1

Source code in src/sportsbet/sources/_utils.py

def derive_market_outcomes(
    home_points: pd.Series | pd.DataFrame,
    away_points: pd.Series | pd.DataFrame,
    markets: list[str],
) -> pd.DataFrame:
    """Derive boolean outcomes for the given markets from home and away points.

    Args:
        home_points:
            The home team points per snapshot.
        away_points:
            The away team points per snapshot.
        markets:
            The markets to derive (e.g. `home_win`, `over_2.5`).

    Returns:
        A dataframe with one integer 0/1 column per requested market.

    Examples:
        >>> import pandas as pd
        >>> from sportsbet.sources import derive_market_outcomes
        >>> home_points = pd.Series([2, 1, 0])
        >>> away_points = pd.Series([1, 1, 3])
        >>> derive_market_outcomes(home_points, away_points, ['home_win', 'draw', 'away_win', 'over_2.5'])
           home_win  draw  away_win  over_2.5
        0         1     0         0         1
        1         0     1         0         0
        2         0     0         1         1
    """
    index = home_points.index if isinstance(home_points, pd.Series | pd.DataFrame) else None
    home_points = column_or_1d(home_points, dtype=int)
    away_points = column_or_1d(away_points, dtype=int)
    check_consistent_length(home_points, away_points)
    total = home_points + away_points
    outcomes = {}
    for market in markets:
        if market == 'home_win':
            outcomes[market] = home_points > away_points
        elif market == 'draw':
            outcomes[market] = home_points == away_points
        elif market == 'away_win':
            outcomes[market] = away_points > home_points
        elif market.startswith('over_'):
            outcomes[market] = total > float(market.removeprefix('over_'))
        elif market.startswith('under_'):
            outcomes[market] = total < float(market.removeprefix('under_'))
    return pd.DataFrame(outcomes, index=index).astype(int)

fetch_payloads(items, authorize)

Read each item at the URL authorize gives it and pair the bytes back with the item, in order.

Parameters:

Name	Type	Description	Default
items	list[RawItem]	The items to read.	required
authorize	Callable[[RawItem], str]	A callable turning an item into the URL to fetch it from, adding any credential.	required

Returns:

Type	Description
list[RawPayload]	The payloads, each pairing an item with its bytes, in the order given.

Source code in src/sportsbet/sources/_base.py

def fetch_payloads(items: list[RawItem], authorize: Callable[[RawItem], str]) -> list[RawPayload]:
    """Read each item at the URL `authorize` gives it and pair the bytes back with the item, in order.

    Args:
        items:
            The items to read.

        authorize:
            A callable turning an item into the URL to fetch it from, adding any credential.

    Returns:
        The payloads, each pairing an item with its bytes, in the order given.
    """
    contents = _read_urls_content([authorize(item) for item in items])
    return [RawPayload(item=item, content=content) for item, content in zip(items, contents, strict=True)]

measure_names_similarity(one, other)

Return how alike two names are, by the tokens they begin with in common.

Parameters:

Name	Type	Description	Default
one	str	The first name.	required
other	str	The second name.	required

Returns:

Type	Description
float	A score in [0.0, 1.0], higher the more the names share.

Source code in src/sportsbet/sources/_resolver.py

def measure_names_similarity(one: str, other: str) -> float:
    """Return how alike two names are, by the tokens they begin with in common.

    Args:
        one:
            The first name.

        other:
            The second name.

    Returns:
        A score in `[0.0, 1.0]`, higher the more the names share.
    """
    tokens, others = sorted([one.split(), other.split()], key=len)
    if not tokens or not others:
        return 0.0
    total = 0.0
    for token in tokens:
        scores = [
            count_common_prefix(token, candidate) / min(len(token), len(candidate))
            for candidate in others
            if count_common_prefix(token, candidate) >= MIN_PREFIX
        ]
        total += max(scores, default=0.0)
    return total / len(tokens)

normalize_identity(data, mapping=None)

Return the match columns with the team names normalized, and remapped when a mapping is given.

Parameters:

Name	Type	Description	Default
data	DataFrame	The snapshots whose match columns are read.	required
mapping	dict | None	The per-league, per-season odds-to-stats name mapping. None only normalizes.	None

Returns:

Type	Description
DataFrame	The match columns with normalized, and optionally remapped, team names.

Source code in src/sportsbet/sources/_resolver.py

def normalize_identity(data: pd.DataFrame, mapping: dict | None = None) -> pd.DataFrame:
    """Return the match columns with the team names normalized, and remapped when a mapping is given.

    Args:
        data:
            The snapshots whose match columns are read.

        mapping:
            The per-league, per-season odds-to-stats name mapping. `None` only normalizes.

    Returns:
        The match columns with normalized, and optionally remapped, team names.
    """
    identity = data[MATCH_COLS].copy()
    for col in TEAMS_COLS:
        identity[col] = identity[col].map(normalize_team_name)
    if mapping is None:
        return identity
    keys = list(zip(*[data[col] for col in GROUPS_COLS], strict=True))
    for col in TEAMS_COLS:
        identity[col] = [mapping.get(key, {}).get(name, name) for key, name in zip(keys, identity[col], strict=True)]
    return identity

normalize_team_name(name)

Return a team name with the parts that carry no meaning removed.

Parameters:

Name	Type	Description	Default
name	str	The team name to normalize.	required

Returns:

Type	Description
str	The normalized name: lower case, without accents, punctuation or noise words.

Source code in src/sportsbet/sources/_resolver.py

def normalize_team_name(name: str) -> str:
    """Return a team name with the parts that carry no meaning removed.

    Args:
        name:
            The team name to normalize.

    Returns:
        The normalized name: lower case, without accents, punctuation or noise words.
    """
    text = unicodedata.normalize('NFKD', str(name))
    text = ''.join(character for character in text if not unicodedata.combining(character))
    text = re.sub(r'[^a-z0-9 ]', '', text.lower())
    tokens = [token for token in text.split() if token not in NOISE]
    return ' '.join(tokens)

optional_col(include, fixed, alias=None)

Define an optional feature or odds column.

Parameters:

Name	Type	Description	Default
include	list[str]	The event statuses at which the column is meaningful.	required
fixed	bool	Whether the column is time-invariant within a match.	required
alias	str | None	The column name to use when it differs from the field's Python identifier.	None

Returns:

Type	Description
Any	A pandera field marking the column as an optional feature or odds column.

Examples:

>>> from sportsbet.sources import BaseStatsSchema, optional_col, required_col
>>>
>>> class MyStatsSchema(BaseStatsSchema):
...     'The columns a statistics feed of your own may carry.'
...
...     home_team: str = required_col()
...     away_team: str = required_col()
...     home_goals: float = optional_col(include=['inplay', 'postplay'], fixed=False)
...     stadium_capacity: float = optional_col(include=['preplay'], fixed=True)
>>>
>>> # There is no score before the match starts.
>>> MyStatsSchema.to_schema().columns['home_goals'].metadata['include']
['inplay', 'postplay']
>>> # A stadium does not change size at half time.
>>> MyStatsSchema.to_schema().columns['stadium_capacity'].metadata['fixed']
True

Source code in src/sportsbet/sources/_schema.py

def optional_col(include: list[str], fixed: bool, alias: str | None = None) -> Any:  # noqa: ANN401  # varied defaults
    """Define an optional feature or odds column.

    Args:
        include:
            The event statuses at which the column is meaningful.
        fixed:
            Whether the column is time-invariant within a match.
        alias:
            The column name to use when it differs from the field's Python
            identifier.

    Returns:
        A pandera field marking the column as an optional feature or odds column.

    Examples:
        >>> from sportsbet.sources import BaseStatsSchema, optional_col, required_col
        >>>
        >>> class MyStatsSchema(BaseStatsSchema):
        ...     'The columns a statistics feed of your own may carry.'
        ...
        ...     home_team: str = required_col()
        ...     away_team: str = required_col()
        ...     home_goals: float = optional_col(include=['inplay', 'postplay'], fixed=False)
        ...     stadium_capacity: float = optional_col(include=['preplay'], fixed=True)
        >>>
        >>> # There is no score before the match starts.
        >>> MyStatsSchema.to_schema().columns['home_goals'].metadata['include']
        ['inplay', 'postplay']
        >>> # A stadium does not change size at half time.
        >>> MyStatsSchema.to_schema().columns['stadium_capacity'].metadata['fixed']
        True
    """
    return pa.Field(nullable=True, metadata={'include': include, 'fixed': fixed}, alias=alias)

pair_rosters(normalized_stats_names, normalized_odds_names)

Return the odds names paired to the stats names, and the names left unpaired on each side.

Parameters:

Name	Type	Description	Default
normalized_stats_names	set[str]	The normalized club names of the statistics source.	required
normalized_odds_names	set[str]	The normalized club names of the odds source.	required

Returns:

Type	Description
tuple[dict[str, str], set[str], set[str]]	The odds-to-stats name pairing, the odds names left unpaired, and the stats names left unpaired.

Source code in src/sportsbet/sources/_resolver.py

def pair_rosters(
    normalized_stats_names: set[str],
    normalized_odds_names: set[str],
) -> tuple[dict[str, str], set[str], set[str]]:
    """Return the odds names paired to the stats names, and the names left unpaired on each side.

    Args:
        normalized_stats_names:
            The normalized club names of the statistics source.

        normalized_odds_names:
            The normalized club names of the odds source.

    Returns:
        The odds-to-stats name pairing, the odds names left unpaired, and the stats names left unpaired.
    """
    matched = {name: name for name in normalized_odds_names & normalized_stats_names}
    unpaired_odds = normalized_odds_names - set(matched)
    unpaired_stats = normalized_stats_names - set(matched.values())
    candidates = []
    for name in unpaired_odds:
        scores = sorted(((measure_names_similarity(name, other), other) for other in unpaired_stats), reverse=True)
        if not scores:
            break
        best, runner_up = scores[0], (scores[1] if len(scores) > 1 else (0.0, ''))
        candidates.append((best[0], best[0] - runner_up[0], name, best[1]))
    for score, margin, name, other in sorted(candidates, reverse=True):
        if name not in unpaired_odds or other not in unpaired_stats:
            continue
        alone = len(unpaired_odds) == 1 and len(unpaired_stats) == 1
        if score >= MIN_SIMILARITY and (margin >= MIN_MARGIN or alone):
            matched[name] = other
            unpaired_odds.discard(name)
            unpaired_stats.discard(other)
    return matched, unpaired_odds, unpaired_stats

read_csv_content(content)

Return a data frame read from raw CSV content.

Parameters:

Name	Type	Description	Default
content	bytes	The raw CSV bytes.	required

Returns:

Type	Description
DataFrame	The parsed data frame.

Source code in src/sportsbet/sources/_base.py

def read_csv_content(content: bytes) -> pd.DataFrame:
    """Return a data frame read from raw CSV content.

    Args:
        content:
            The raw CSV bytes.

    Returns:
        The parsed data frame.
    """
    text = content.decode(ENCODING)
    names = pd.read_csv(io.StringIO(text), nrows=0, encoding=ENCODING).columns.to_list()
    return pd.read_csv(io.StringIO(text), names=names, skiprows=1, encoding=ENCODING, on_bad_lines='skip')

required_col(alias=None)

Define a required snapshot-identity column.

Parameters:

Name	Type	Description	Default
alias	str | None	The column name to use when it differs from the field's Python identifier.	None

Returns:

Type	Description
Any	A pandera field marking the column as a required snapshot-identity column.

Examples:

>>> from sportsbet.sources import BaseStatsSchema, required_col
>>>
>>> class MyStatsSchema(BaseStatsSchema):
...     'The columns a statistics feed of your own must always carry.'
...
...     home_team: str = required_col()
...     away_team: str = required_col()
>>>
>>> # A required column may not be missing.
>>> MyStatsSchema.to_schema().columns['home_team'].nullable
False

Source code in src/sportsbet/sources/_schema.py

def required_col(alias: str | None = None) -> Any:  # noqa: ANN401  # varied defaults
    """Define a required snapshot-identity column.

    Args:
        alias:
            The column name to use when it differs from the field's Python
            identifier.

    Returns:
        A pandera field marking the column as a required snapshot-identity column.

    Examples:
        >>> from sportsbet.sources import BaseStatsSchema, required_col
        >>>
        >>> class MyStatsSchema(BaseStatsSchema):
        ...     'The columns a statistics feed of your own must always carry.'
        ...
        ...     home_team: str = required_col()
        ...     away_team: str = required_col()
        >>>
        >>> # A required column may not be missing.
        >>> MyStatsSchema.to_schema().columns['home_team'].nullable
        False
    """
    return pa.Field(nullable=False, metadata={'snapshot': True}, alias=alias)

resolve_odds(stats, odds, aliases=None)

Return the odds with the identity of the matches they belong to.

Parameters:

Name	Type	Description	Default
stats	DataFrame	The long statistics snapshots, which say which matches exist.	required
odds	DataFrame	The long odds snapshots.	required
aliases	dict[str, str] | None	The team names of the odds source, mapped to the names of the statistics source, for the clubs the pairing leaves over.	None

Returns:

Name	Type	Description
odds	DataFrame	The odds carrying the identity of the statistics.

Examples:

>>> import pandas as pd
>>> from sportsbet.sources import resolve_odds
>>> identity = {'league': 'England', 'division': 1, 'year': 2025}
>>> moment = {'event_status': 'preplay', 'event_time': pd.Timedelta(0)}
>>> stats = pd.DataFrame([
...     {'date': pd.Timestamp('2025-08-16', tz='UTC'), **identity, **moment,
...      'home_team': 'Man United', 'away_team': 'Arsenal'},
... ])
>>> odds = pd.DataFrame([
...     {'date': pd.Timestamp('2025-08-16', tz='UTC'), **identity, **moment,
...      'home_team': 'Manchester United', 'away_team': 'Arsenal',
...      'provider': 'acme', 'home_win': 1.8},
... ])
>>> paired = resolve_odds(stats, odds)
>>> # The odds carry the identity of the statistics, so `Manchester United` becomes `Man United`.
>>> paired[['home_team', 'away_team', 'home_win']].to_dict('records')
[{'home_team': 'Man United', 'away_team': 'Arsenal', 'home_win': 1.8}]
>>> # A club the pairing cannot place is named, and can be given as an alias.
>>> paired = resolve_odds(stats, odds.assign(home_team='Utd of Manchester'),
...                  aliases={'Utd of Manchester': 'Man United'})
>>> len(paired)
1

Source code in src/sportsbet/sources/_resolver.py

def resolve_odds(
    stats: pd.DataFrame,
    odds: pd.DataFrame,
    aliases: dict[str, str] | None = None,
) -> pd.DataFrame:
    """Return the odds with the identity of the matches they belong to.

    Args:
        stats:
            The long statistics snapshots, which say which matches exist.

        odds:
            The long odds snapshots.

        aliases:
            The team names of the odds source, mapped to the names of the statistics source, for the clubs the pairing
            leaves over.

    Returns:
        odds:
            The odds carrying the identity of the statistics.

    Examples:
        >>> import pandas as pd
        >>> from sportsbet.sources import resolve_odds
        >>> identity = {'league': 'England', 'division': 1, 'year': 2025}
        >>> moment = {'event_status': 'preplay', 'event_time': pd.Timedelta(0)}
        >>> stats = pd.DataFrame([
        ...     {'date': pd.Timestamp('2025-08-16', tz='UTC'), **identity, **moment,
        ...      'home_team': 'Man United', 'away_team': 'Arsenal'},
        ... ])
        >>> odds = pd.DataFrame([
        ...     {'date': pd.Timestamp('2025-08-16', tz='UTC'), **identity, **moment,
        ...      'home_team': 'Manchester United', 'away_team': 'Arsenal',
        ...      'provider': 'acme', 'home_win': 1.8},
        ... ])
        >>> paired = resolve_odds(stats, odds)
        >>> # The odds carry the identity of the statistics, so `Manchester United` becomes `Man United`.
        >>> paired[['home_team', 'away_team', 'home_win']].to_dict('records')
        [{'home_team': 'Man United', 'away_team': 'Arsenal', 'home_win': 1.8}]
        >>> # A club the pairing cannot place is named, and can be given as an alias.
        >>> paired = resolve_odds(stats, odds.assign(home_team='Utd of Manchester'),
        ...                  aliases={'Utd of Manchester': 'Man United'})
        >>> len(paired)
        1
    """
    mapping = _map_odds_names(stats, odds, {**ALIASES, **(aliases or {})})
    matches = stats[[*MATCH_COLS, 'date']].drop_duplicates(subset=MATCH_COLS)
    canonical = normalize_identity(matches).assign(
        date_=matches['date'].to_numpy(),
        home_team_=matches['home_team'].to_numpy(),
        away_team_=matches['away_team'].to_numpy(),
    )
    resolved = odds.drop(columns=['date']).assign(**{col: normalize_identity(odds, mapping)[col] for col in MATCH_COLS})
    resolved = resolved.merge(canonical, on=MATCH_COLS, how='left')

    found = resolved['date_'].notna()
    resolved = resolved[found.to_numpy()].copy()
    resolved['date'] = resolved.pop('date_')
    resolved['home_team'] = resolved.pop('home_team_')
    resolved['away_team'] = resolved.pop('away_team_')
    resolved = resolved[odds.columns]
    return resolved