Execution

Place the value bets a bettor found at a venue.

BaseVenue

Bases: ABC

A place where a user holds an account and can back a selection.

A venue that implements this contract places a bet once and only once for each identity.

authenticate() abstractmethod async

Authenticate, reading the secret from the variable the venue names.

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

@abc.abstractmethod
async def authenticate(self: BaseVenue) -> None:
    """Authenticate, reading the secret from the variable the venue names."""

cancel(identity) abstractmethod async

Cancel a bet, raising CancellationUnsupportedError where the venue cannot.

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

@abc.abstractmethod
async def cancel(self: BaseVenue, identity: BetIdentity) -> PlacementReceipt:
    """Cancel a bet, raising `CancellationUnsupportedError` where the venue cannot."""

list_markets(matches) abstractmethod async

Return the markets on offer for the given matches, with their current prices.

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

@abc.abstractmethod
async def list_markets(self: BaseVenue, matches: list[str]) -> pd.DataFrame:
    """Return the markets on offer for the given matches, with their current prices."""

place(intent) abstractmethod async

Place one bet, once and only once for its identity.

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

@abc.abstractmethod
async def place(self: BaseVenue, intent: PlacementIntent) -> PlacementReceipt:
    """Place one bet, once and only once for its identity."""

read_balance() abstractmethod async

Return the balance and the exposure currently open.

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

@abc.abstractmethod
async def read_balance(self: BaseVenue) -> tuple[float, float]:
    """Return the balance and the exposure currently open."""

read_status(identities) abstractmethod async

Return what the venue holds for these identities.

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

@abc.abstractmethod
async def read_status(self: BaseVenue, identities: list[BetIdentity]) -> pd.DataFrame:
    """Return what the venue holds for these identities."""

BetIdentity(venue, match, market, selection) dataclass

What makes two bets the same bet.

Two bets with the same venue, match, market and selection are the same bet.

Parameters:

Name	Type	Description	Default
venue	str	The venue the bet is placed at.	required
match	str	The match it is on.	required
market	str	The market within the match.	required
selection	str	The side backed.	required

Examples:

>>> from sportsbet.execution import BetIdentity
>>> identity = BetIdentity('exchange', 'Arsenal vs Chelsea', 'home_win', 'Arsenal')
>>> identity.ref_
'2f3152b1a4f9f6333904f3624320d921'
>>> BetIdentity('exchange', 'Arsenal vs Chelsea', 'home_win', 'Arsenal').ref_ == identity.ref_
True

ref_ property

The reference a venue carries for this bet.

BrowserSession(key, url, *, notes=None, credential_env=None, user_data_dir=None, min_interval=1.0, timeout=30000.0, headless=True)

A bookmaker's website, driven in a real browser on your own account.

You provide everything the session knows about the site.

Parameters:

Name	Type	Description	Default
key	str	What to call this venue.	required
url	str	The bookmaker's site.	required
notes	str | None	What you know about the site.	None
credential_env	tuple[str, str] | None	The names of the variables holding the username and the password.	None
user_data_dir	str | Path | None	Where the browser keeps its profile.	None
min_interval	float	The seconds to leave between actions.	1.0
timeout	float	The milliseconds to wait for an element before giving up on it.	30000.0
headless	bool	Whether to hide the browser window. False opens it.	True

Examples:

>>> from sportsbet.execution import BrowserSession
>>> session = BrowserSession(
...     key='example',
...     url='https://example.invalid/betting',
...     notes='The slip opens on the right after clicking a price.',
... )
>>> session.key
'example'
>>> session.notes
'The slip opens on the right after clicking a price.'
>>> hasattr(session, 'place')
False

Source code in src/sportsbet/execution/_browser.py

def __init__(
    self: BrowserSession,
    key: str,
    url: str,
    *,
    notes: str | None = None,
    credential_env: tuple[str, str] | None = None,
    user_data_dir: str | Path | None = None,
    min_interval: float = 1.0,
    timeout: float = 30000.0,
    headless: bool = True,
) -> None:
    """Keep what the session was configured with."""
    self.key = key
    self.url = url
    self.notes = notes
    self.credential_env = credential_env
    self.user_data_dir = user_data_dir
    self.min_interval = min_interval
    self.timeout = timeout
    self.headless = headless
    self.context_: BrowserContext | None = None
    self.fixed_: FixedSession | None = None
    self._playwright: object | None = None

authenticate() async

Open the browser at the site, starting from the saved profile.

Source code in src/sportsbet/execution/_browser.py

async def authenticate(self: BrowserSession) -> None:
    """Open the browser at the site, starting from the saved profile."""
    await self.navigate(self.url)

click(ref) async

Click an element and return the page it produced.

Parameters:

Name	Type	Description	Default
ref	str	The ref of the element, from a snapshot.	required

Returns:

Name	Type	Description
snapshot	PageSnapshot	The page after the click.

Source code in src/sportsbet/execution/_browser.py

async def click(self: BrowserSession, ref: str) -> PageSnapshot:
    """Click an element and return the page it produced.

    Args:
        ref:
            The ref of the element, from a snapshot.

    Returns:
        snapshot:
            The page after the click.
    """
    await self._paced()
    await self._read_page().locator(f'aria-ref={ref}').click(timeout=self.timeout)
    return await self._capture()

fix(match, locators)

Pin the locators exploring found for a match.

Parameters:

Name	Type	Description	Default
match	str	The match this session is pinned to.	required
locators	dict[str, str]	What was found, by name, as in {'stake': 'textbox[name="Stake"]'}.	required

Returns:

Name	Type	Description
fixed	FixedSession	The pinned session.

Raises:

Type	Description
ExecutionError	If a locator is a ref or looks like a price.

Source code in src/sportsbet/execution/_browser.py

def fix(self: BrowserSession, match: str, locators: dict[str, str]) -> FixedSession:
    """Pin the locators exploring found for a match.

    Args:
        match:
            The match this session is pinned to.
        locators:
            What was found, by name, as in `{'stake': 'textbox[name="Stake"]'}`.

    Returns:
        fixed:
            The pinned session.

    Raises:
        ExecutionError: If a locator is a ref or looks like a price.
    """
    for name, locator in locators.items():
        if REF_PATTERN.match(locator) or locator.startswith('aria-ref='):
            msg = (
                f'`{name}` is pinned to the ref `{locator}`, which belongs to one state of the page. '
                f'Pin a role and an accessible name instead, as in `textbox[name="Stake"]`.'
            )
            raise ExecutionError(msg)
        if 'price' in name or 'odds' in name:
            msg = f'`{name}` looks like a price. A price is read when the bet is placed rather than pinned.'
            raise ExecutionError(msg)
    self.fixed_ = FixedSession(match=match, url=self._read_page().url, locators=dict(locators))
    return self.fixed_

navigate(url) async

Go to a page and return it.

Parameters:

Name	Type	Description	Default
url	str	Where to go.	required

Returns:

Name	Type	Description
snapshot	PageSnapshot	The page, with a ref for every element that can be acted on.

Raises:

Type	Description
VenueBlockedError	If the site blocks automated access.

Source code in src/sportsbet/execution/_browser.py

async def navigate(self: BrowserSession, url: str) -> PageSnapshot:
    """Go to a page and return it.

    Args:
        url:
            Where to go.

    Returns:
        snapshot:
            The page, with a ref for every element that can be acted on.

    Raises:
        VenueBlockedError: If the site blocks automated access.
    """
    if self.context_ is None:
        await self.start()
    assert self.context_ is not None
    page = self.context_.pages[0] if self.context_.pages else await self.context_.new_page()
    await self._paced()
    response = await page.goto(url)
    if response is not None and response.status in BLOCKED_STATUS:
        msg = f'`{self.key}` refused automated access with status {response.status}.'
        raise VenueBlockedError(msg)
    return await self._capture()

read_snapshot(selector=None, depth=None) async

Return the page, or a part of it.

Parameters:

Name	Type	Description	Default
selector	str | None	The part to read.	None
depth	int | None	How far down to read.	None

Returns:

Name	Type	Description
snapshot	PageSnapshot	The page, with a ref for every element that can be acted on.

Source code in src/sportsbet/execution/_browser.py

async def read_snapshot(
    self: BrowserSession,
    selector: str | None = None,
    depth: int | None = None,
) -> PageSnapshot:
    """Return the page, or a part of it.

    Args:
        selector:
            The part to read.
        depth:
            How far down to read.

    Returns:
        snapshot:
            The page, with a ref for every element that can be acted on.
    """
    return await self._capture(selector, depth)

resolve(name) async

Read what a pinned locator points at now.

Parameters:

Name	Type	Description	Default
name	str	The name it was pinned under.	required

Returns:

Name	Type	Description
snapshot	PageSnapshot	What is there now.

Raises:

Type	Description
ExecutionError	If nothing is pinned or the name is not pinned.

Source code in src/sportsbet/execution/_browser.py

async def resolve(self: BrowserSession, name: str) -> PageSnapshot:
    """Read what a pinned locator points at now.

    Args:
        name:
            The name it was pinned under.

    Returns:
        snapshot:
            What is there now.

    Raises:
        ExecutionError: If nothing is pinned or the name is not pinned.
    """
    if self.fixed_ is None:
        msg = 'Nothing is pinned. Call `fix` with what exploring found.'
        raise ExecutionError(msg)
    if name not in self.fixed_.locators:
        msg = f'`{name}` is not pinned. Pinned: {", ".join(sorted(self.fixed_.locators))}.'
        raise ExecutionError(msg)
    return await self._capture(self.fixed_.locators[name])

select(ref, value) async

Choose an option and return the page it produced.

Parameters:

Name	Type	Description	Default
ref	str	The ref of the element, from a snapshot.	required
value	str	The option to choose.	required

Returns:

Name	Type	Description
snapshot	PageSnapshot	The page after the option was chosen.

Source code in src/sportsbet/execution/_browser.py

async def select(self: BrowserSession, ref: str, value: str) -> PageSnapshot:
    """Choose an option and return the page it produced.

    Args:
        ref:
            The ref of the element, from a snapshot.
        value:
            The option to choose.

    Returns:
        snapshot:
            The page after the option was chosen.
    """
    await self._paced()
    await self._read_page().locator(f'aria-ref={ref}').select_option(value, timeout=self.timeout)
    return await self._capture()

start() async

Open the browser and keep it open.

Source code in src/sportsbet/execution/_browser.py

async def start(self: BrowserSession) -> None:
    """Open the browser and keep it open."""
    from playwright.async_api import async_playwright  # noqa: PLC0415  # defer the optional browser extra

    if self.context_ is not None:
        return
    driver = await async_playwright().start()
    self._playwright = driver
    profile = Path(self.user_data_dir) if self.user_data_dir else None
    if profile is None:
        browser = await driver.chromium.launch(headless=self.headless)
        self.context_ = await browser.new_context()
    else:
        self.context_ = await driver.chromium.launch_persistent_context(str(profile), headless=self.headless)

stop() async

Close the browser.

Source code in src/sportsbet/execution/_browser.py

async def stop(self: BrowserSession) -> None:
    """Close the browser."""
    if self.context_ is not None:
        await self.context_.close()
        self.context_ = None
    if self._playwright is not None:
        await self._playwright.stop()  # type: ignore[attr-defined]  # driver untyped to keep the extra optional
        self._playwright = None

type(ref, text) async

Fill an element and return the page it produced.

Parameters:

Name	Type	Description	Default
ref	str	The ref of the element, from a snapshot.	required
text	str	What to put in it.	required

Returns:

Name	Type	Description
snapshot	PageSnapshot	The page after the text went in.

Source code in src/sportsbet/execution/_browser.py

async def type(self: BrowserSession, ref: str, text: str) -> PageSnapshot:
    """Fill an element and return the page it produced.

    Args:
        ref:
            The ref of the element, from a snapshot.
        text:
            What to put in it.

    Returns:
        snapshot:
            The page after the text went in.
    """
    await self._paced()
    await self._read_page().locator(f'aria-ref={ref}').fill(text, timeout=self.timeout)
    return await self._capture()

CancellationUnsupportedError

Bases: ExecutionError

Raised when a venue is asked to cancel and cannot.

CredentialError

Bases: ExecutionError

Raised when a named variable holds nothing.

CredentialRef(var) dataclass

The name of a variable holding a secret.

Parameters:

Name	Type	Description	Default
var	str	The name of the variable.	required

Examples:

>>> from sportsbet.execution import CredentialRef
>>> CredentialRef('VENUE_API_KEY').var
'VENUE_API_KEY'

__str__()

Return the variable name.

Source code in src/sportsbet/execution/_credentials.py

def __str__(self: CredentialRef) -> str:
    """Return the variable name."""
    return self.var

ExecutionError

Bases: Exception

Raised when placing cannot go ahead.

ExposureLimits(max_stake_per_bet=0.0, max_total_exposure=0.0, killed=False) dataclass

The limits a placement must stay within, and the kill switch that stops it.

Parameters:

Name	Type	Description	Default
max_stake_per_bet	float	The most to stake on one bet.	0.0
max_total_exposure	float	The most to have staked at once.	0.0
killed	bool	Whether the kill switch is on.	False

FixedSession(match, url, locators=dict()) dataclass

The locators found during exploration, pinned to a match.

Parameters:

Name	Type	Description	Default
match	str	The match the session is pinned to.	required
url	str	The page URL it was pinned at.	required
locators	dict[str, str]	The pinned locators, by name.	dict()

PageSnapshot(yaml, url) dataclass

A page in a form an agent can reason about and act on.

Parameters:

Name	Type	Description	Default
yaml	str	The page as an accessibility tree.	required
url	str	The page URL.	required

PlacementIntent(identity, stake, min_price, value_bet='') dataclass

What the caller means to do.

Parameters:

Name	Type	Description	Default
identity	BetIdentity	The bet to place.	required
stake	float	The amount to stake.	required
min_price	float	The lowest price to accept.	required
value_bet	str	The value bet the intent came from.	''

PlacementQuote(intents, total_stake, total_exposure, quoted_at) dataclass

What is about to be staked, before any bet is placed.

Parameters:

Name	Type	Description	Default
intents	list[PlacementIntent]	The bets the quote covers.	required
total_stake	float	The total to be staked.	required
total_exposure	float	The total exposure after staking.	required
quoted_at	datetime	When the quote was taken.	required

PlacementReceipt(identity, status, stake=0.0, price=None, venue_bet_id=None, value_bet='', placed_at=None, detail='') dataclass

What happened to an intended bet.

Parameters:

Name	Type	Description	Default
identity	BetIdentity	The bet.	required
status	PlacementStatus	What became of it.	required
stake	float	The amount staked.	0.0
price	float | None	The price it was matched at.	None
venue_bet_id	str | None	The venue's identifier for the bet.	None
value_bet	str	The value bet it came from.	''
placed_at	datetime | None	When it was placed.	None
detail	str	A human-readable note.	''

PlacementReceiptSchema

Bases: DataFrameModel

The receipts a placement returns.

Config

Allow no column beyond the ones named, and coerce the dtypes.

PlacementStatus

Bases: StrEnum

What became of an intended bet.

VenueBlockedError

Bases: ExecutionError

Raised when a venue blocks automated access.

build_receipts_frame(receipts)

Return receipts as a validated frame.

Parameters:

Name	Type	Description	Default
receipts	list[PlacementReceipt]	The receipts to frame.	required

Returns:

Name	Type	Description
frame	DataFrame	The receipts as a validated frame.

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

def build_receipts_frame(receipts: list[PlacementReceipt]) -> pd.DataFrame:
    """Return receipts as a validated frame.

    Args:
        receipts: The receipts to frame.

    Returns:
        frame: The receipts as a validated frame.
    """
    records = [
        {
            'ref': receipt.identity.ref_,
            'venue': receipt.identity.venue,
            'match': receipt.identity.match,
            'market': receipt.identity.market,
            'selection': receipt.identity.selection,
            'status': receipt.status.value,
            'stake': receipt.stake,
            'price': receipt.price,
            'venue_bet_id': receipt.venue_bet_id,
            'value_bet': receipt.value_bet,
            'placed_at': receipt.placed_at,
            'detail': receipt.detail,
        }
        for receipt in receipts
    ]
    frame = pd.DataFrame.from_records(records, columns=list(PlacementReceiptSchema.to_schema().columns))
    frame['placed_at'] = pd.to_datetime(frame['placed_at'], utc=True)
    result: pd.DataFrame = PlacementReceiptSchema.validate(frame)
    return result

build_venue(venue)

Build a venue from a reference to where it lives.

Parameters:

Name	Type	Description	Default
venue	str	Where the venue lives, as in venue.py:VENUE. A venue with an API is a BaseVenue you write, and a bookmaker's website is a BrowserSession you configure.	required

Returns:

Name	Type	Description
built	BaseVenue | BrowserSession	The venue, or the browser session.

Raises:

Type	Description
BuildError	If the reference is malformed, the execution extra is missing, or it names no venue.

Source code in src/sportsbet/execution/_factory.py

def build_venue(venue: str) -> BaseVenue | BrowserSession:
    """Build a venue from a reference to where it lives.

    Args:
        venue:
            Where the venue lives, as in `venue.py:VENUE`. A venue with an API is a `BaseVenue` you write, and a
            bookmaker's website is a `BrowserSession` you configure.

    Returns:
        built:
            The venue, or the browser session.

    Raises:
        BuildError: If the reference is malformed, the execution extra is missing, or it names no venue.
    """
    if ':' not in venue:
        msg = f'`{venue}` should name a venue in a Python file, as in `venue.py:VENUE`. The library ships none.'
        raise BuildError(msg)
    built = load_object(venue)
    try:
        from ._browser import BrowserSession  # noqa: PLC0415  # defer the optional execution extra
    except ImportError as missing:
        raise BuildError(EXECUTION_EXTRA) from missing
    if not isinstance(built, BaseVenue | BrowserSession):
        msg = f'`{venue}` is not a venue and is not a browser session.'
        raise BuildError(msg)
    return built

execute_event(event, bettor, dataloader, session, *, stake, urls, live=False, placer=None, poll=DEFAULT_POLL, clock=None, wait=None) async

Watch one event and place the model's bet at its moment, once.

The function runs a fixed sequence of steps. It explores the URLs to find the event. It makes sure the session is logged in. It monitors the event until its betting moment and logs it. It applies the fitted bettor at that moment. It places one bet when the model finds value and the run is armed. A run stakes nothing unless you set live.

Parameters:

Name	Type	Description	Default
event	str	The one match to act on, as 'Home vs Away'. Must be among the dataloader's fixtures.	required
bettor	BaseBettor	A fitted bettor. Decides whether to bet and the selection, never the stake.	required
dataloader	BaseDataLoader	Supplies the event's features and odds and carries the fitted moment.	required
session	BrowserSession	A headless browser session for the bookmaker.	required
stake	float	The fixed amount to place.	required
urls	list[str]	Candidate bookmaker URLs, explored and matched to the event during setup.	required
live	bool	Arms the run. False is a no-stakes dry run that places nothing.	False
placer	Placer | None	How to place on the site. Defaults to driving the pinned stake and confirm controls.	None
poll	Timedelta	The interval between source polls while monitoring.	DEFAULT_POLL
clock	Clock | None	What returns the current time. Defaults to the real UTC now.	None
wait	Wait | None	What waits for a number of seconds. Defaults to sleeping.	None

Returns:

Name	Type	Description
receipts	DataFrame	A receipts frame with one row when a bet was placed and no rows otherwise.

Raises:

Type	Description
ExecutionError	If the event is not among the fixtures.

Source code in src/sportsbet/execution/_event.py

async def execute_event(
    event: str,
    bettor: BaseBettor,
    dataloader: BaseDataLoader,
    session: BrowserSession,
    *,
    stake: float,
    urls: list[str],
    live: bool = False,
    placer: Placer | None = None,
    poll: pd.Timedelta = DEFAULT_POLL,
    clock: Clock | None = None,
    wait: Wait | None = None,
) -> pd.DataFrame:
    """Watch one event and place the model's bet at its moment, once.

    The function runs a fixed sequence of steps. It explores the URLs to find the event. It makes sure the session is
    logged in. It monitors the event until its betting moment and logs it. It applies the fitted bettor at that
    moment. It places one bet when the model finds value and the run is armed. A run stakes nothing unless you set
    `live`.

    Args:
        event:
            The one match to act on, as `'Home vs Away'`. Must be among the dataloader's fixtures.
        bettor:
            A fitted bettor. Decides whether to bet and the selection, never the stake.
        dataloader:
            Supplies the event's features and odds and carries the fitted moment.
        session:
            A headless browser session for the bookmaker.
        stake:
            The fixed amount to place.
        urls:
            Candidate bookmaker URLs, explored and matched to the event during setup.
        live:
            Arms the run. `False` is a no-stakes dry run that places nothing.
        placer:
            How to place on the site. Defaults to driving the pinned `stake` and `confirm` controls.
        poll:
            The interval between source polls while monitoring.
        clock:
            What returns the current time. Defaults to the real UTC now.
        wait:
            What waits for a number of seconds. Defaults to sleeping.

    Returns:
        receipts:
            A receipts frame with one row when a bet was placed and no rows otherwise.

    Raises:
        ExecutionError: If the event is not among the fixtures.
    """
    read_now = clock or _now
    hold = wait or asyncio.sleep
    place = placer or _default_placer

    if not await _match_url(session, event, urls):
        logger.info('No candidate URL carried %s, so nothing was placed.', event)
        return build_receipts_frame([])

    try:
        await session.authenticate()
    except ExecutionError:
        logger.info('Authentication for %s failed, so nothing was placed.', session.key)
        return build_receipts_frame([])

    X_event, O_event, kickoff = _select_event(dataloader, event)
    moment = find_betting_moment(dataloader, kickoff)
    if read_now() > moment:
        logger.info('The betting moment for %s has passed, so nothing was placed.', event)
        return build_receipts_frame([])

    await _monitor(event, O_event, next(iter(bettor.betting_markets_)), moment, kickoff, poll, read_now, hold)

    intent = _build_intent(session.key, event, bettor, X_event, O_event, stake)
    if intent is None:
        logger.info('No value bet found for %s, so nothing was placed.', event)
        return build_receipts_frame([])

    logger.info(
        'About to stake %s on %s %s at %s, price %s.',
        stake,
        intent.identity.selection,
        intent.identity.market,
        session.key,
        intent.min_price,
    )
    if live:
        receipt = await place(intent, session)
        logger.info('%s: %s.', event, receipt.status.value)
    else:
        receipt = PlacementReceipt(
            identity=intent.identity,
            status=PlacementStatus.DRY_RUN,
            value_bet=intent.value_bet,
            detail='Dry run, so nothing was staked.',
        )
        logger.info('%s: dry run, nothing staked.', event)
    return build_receipts_frame([receipt])

find_betting_moment(dataloader, kickoff)

Return when the bet goes on for a match.

A live model bets at the kickoff plus the time into the match it was fitted for. Any other model bets at the kickoff.

Parameters:

Name	Type	Description	Default
dataloader	BaseDataLoader	The dataloader the model was fitted on.	required
kickoff	Timestamp	The kickoff of the match.	required

Returns:

Name	Type	Description
moment	Timestamp	When the bet goes on.

Source code in src/sportsbet/execution/_schedule.py

def find_betting_moment(dataloader: BaseDataLoader, kickoff: pd.Timestamp) -> pd.Timestamp:
    """Return when the bet goes on for a match.

    A live model bets at the kickoff plus the time into the match it was fitted for. Any other model bets at the
    kickoff.

    Args:
        dataloader:
            The dataloader the model was fitted on.
        kickoff:
            The kickoff of the match.

    Returns:
        moment:
            When the bet goes on.
    """
    if dataloader.target_event_status_ == 'inplay':
        return kickoff + dataloader.target_event_time_
    return kickoff

resolve(ref)

Return what a named variable holds, raising when it holds nothing.

Parameters:

Name	Type	Description	Default
ref	CredentialRef	The name of the variable.	required

Returns:

Name	Type	Description
secret	str	What the variable holds.

Raises:

Type	Description
CredentialError	If the variable is unset or empty.

Source code in src/sportsbet/execution/_credentials.py

def resolve(ref: CredentialRef) -> str:
    """Return what a named variable holds, raising when it holds nothing.

    Args:
        ref:
            The name of the variable.

    Returns:
        secret:
            What the variable holds.

    Raises:
        CredentialError: If the variable is unset or empty.
    """
    secret = os.environ.get(ref.var)
    if not secret:
        msg = f'`{ref.var}` is not set. Set it to the secret, or name another variable.'
        raise CredentialError(msg)
    return secret