7
\$\begingroup\$

My code provides the following functionality for interacting with Pinnacle Bets API:

  • retrieving betting history
  • retrieving fixtures (future events)
  • retrieving odds for the given leagues (competitions)
  • retrieving odds for the given events (games)
  • placing a bet

Should you need to get acquainted with the API, it provides documentation.

With that in mind I have some questions:

How to add an abstraction:

  • In every function that makes an HTTP request I handle exceptions the same way which duplicates code. Could it be extracted from there and put into a separate function?
  • The helper functions starting with _determine also have very similar structure. Is it possible to abstract over them?
  • Functions get_leagues_odds() and get_events_odds() do the same job for a bit different inputs. How to union them into a single function?

Logging:

  • Do I use logging in the convenient way? While this question might sound unspecific, I will be happy to get any comments of my logging.

Documentation

  • Are my type annotations appropriate? Do I use it correctly for type aliasing?
  • What information can I add to make docstrings more helpful?

Any feedback beyond my questions is highly appreciated!

pinnacle_api.py

"Library for interacting with Pinnacle API"
from datetime import date, timedelta
from enum import Enum
import json
import logging
import time
from typing import Any, Dict, List, NamedTuple, Union
import uuid
import requests
import auth

# Constants
SOCCER_SPORT_ID = 29
GET_FIXTURES_URL = "https://api.pinnacle.com/v1/fixtures?"
GET_ODDS_URL = "https://api.pinnacle.com/v1/odds?"
GET_SETTLED_BETS_URL = "https://api.pinnacle.com/v2/bets?betlist=SETTLED"
PLACE_BET_URL = "https://api.pinnacle.com/v2/bets/straight"
ODDS_FORMAT = "Decimal"
ACCEPT_BETTER_LINE = "True"
FILL_TYPE = "FILLANDKILL"
WIN_RISK_STAKE = "RISK"
DEFAULT_PERIOD_NUMBER = 0

# Data definitions
Events = List[int]
Fixtures = Dict[Any, Any]
Leagues = List[Any]
Odds = Dict[Any, Any]
RequestParams = Dict[str, Any]
SettledBet = Dict[Any, Any]
SettledBets = List[SettledBet]

class Bet(NamedTuple):
    "Represents a bet to be made on Pinnacle."
    event_id: int
    line_id: int
    bet_type: str
    side: str
    bet_amount: Union[int, float]


class BetSide(Enum):
    "Represents possible bet sides."
    over = 1
    under = 2
    home = 3
    away = 4


class BetType(Enum):
    "Represents possible bet types."
    spreads = 1
    totals = 2


# Function definitions
def get_bets(start: date, end: date) -> Union[SettledBets, None]:
    "Gets settled bets between given dates"
    bets: SettledBets = []

    while start < end:
        end_date = start + timedelta(days=30)
        from_date = start.strftime('%Y-%m-%d')
        to_date = end_date.strftime('%Y-%m-%d')
        logging.info("Getting settled bets from %s to %s", from_date, to_date)
        params = {
            'fromDate': from_date + 'T00:00:00Z',
            'toDate': to_date + 'T00:00:00Z',
        }
        try:
            response = requests.get(GET_SETTLED_BETS_URL, headers=auth.HEADERS, params=params)
        except requests.ConnectionError:
            logging.critical("No Internet connection")
            return None
        except requests.HTTPError:
            logging.warning("An HTTP error occured.")
            return None
        data = response.json()
        if 'straightBets' in data.keys():
            new_bets: SettledBets = [bet for bet in data['straightBets'] if not bet in bets
                                     and bet['betStatus'] != 'CANCELLED']
            bets.extend(new_bets)
        start += timedelta(days=30)
        time.sleep(1)

    return bets


def get_events_odds(events: Events) -> Union[Odds, None]:
    "Gets odds for the given events."
    params: RequestParams = {
        'sportId': SOCCER_SPORT_ID,
        'oddsFormat': ODDS_FORMAT,
        'eventIds': ','.join([str(event) for event in events])
        }
    try:
        response = requests.get(GET_ODDS_URL, headers=auth.HEADERS, params=params)
    except requests.ConnectionError:
        logging.critical("No Internet connection")
        return None
    except requests.HTTPError:
        logging.warning("An HTTP error occured.")
        return None
    logging.info("Events odds have been retrieved.")
    return response.json()


def get_fixtures(leagues: Leagues) -> Union[Fixtures, None]:
    "Gets fixtures list for the given leagues."
    params: RequestParams = {
        'sportId': SOCCER_SPORT_ID,
        'leagueIds': leagues,
        }
    try:
        response = requests.get(GET_FIXTURES_URL, headers=auth.HEADERS, params=params)
    except requests.ConnectionError:
        logging.critical("No Internet connection")
        return None
    except requests.HTTPError:
        logging.warning("An HTTP error occured.")
        return None
    if response.text == '':
        logging.info("There are no fixtures available")
        return None
    logging.info("Events have been retrieved.")

    return response.json()


def get_leagues_odds(leagues: Leagues) -> Union[Odds, None]:
    "Gets odds for the given leagues"
    params: RequestParams = {
        'sportId': SOCCER_SPORT_ID,
        'oddsFormat': ODDS_FORMAT,
        'leagueIds': ','.join([str(league) for league in leagues]),
        }
    try:
        response = requests.get(GET_ODDS_URL, headers=auth.HEADERS, params=params)
    except requests.ConnectionError:
        logging.critical("No Internet connection")
        return None
    except requests.HTTPError:
        logging.warning("An HTTP error occured.")
        return None
    logging.info("Leagues odds have been retrieved.")
    return response.json()


def get_line_id(odds: Odds) -> int:
    "Gets a line_id for bets on fulltime for the given odds."
    return odds['leagues'][0]['events'][0]['periods'][0]['lineId']


def place_bet(bet: Bet) -> Union[str, None]:
    "Sends a POST request to place a bet."
    bet_type = _determine_bet_type(bet.bet_type)
    team_or_side = _determine_team_or_side(bet.side)
    side = _determine_side(bet.side)
    data = json.dumps({
        "uniqueRequestId": str(uuid.uuid4().hex),
        "acceptBetterLine": ACCEPT_BETTER_LINE,
        "stake": bet.bet_amount,
        "winRiskStake": WIN_RISK_STAKE,
        "fillType": FILL_TYPE,
        "sportId": SOCCER_SPORT_ID,
        "eventId": bet.event_id,
        "lineId": bet.line_id,
        "periodNumber": DEFAULT_PERIOD_NUMBER,
        "betType": bet_type,
        team_or_side: side,
        "oddsFormat": ODDS_FORMAT.capitalize(),
    })
    try:
        response = requests.post(PLACE_BET_URL, data=data, headers=auth.HEADERS)
    except requests.ConnectionError:
        logging.critical("No Internet connection")
        return None
    except requests.HTTPError:
        logging.warning("An HTTP error occured.")
        return None
    return response.text


def _determine_bet_type(bet_type):
    "Determines appropriate bet type for making a bet."
    bet_type = bet_type.lower()
    if bet_type == "spreads":
        return "SPREAD"
    if bet_type == "totals":
        return "TOTAL_POINTS"
    raise ValueError(f'bet type should be one of {BetType._member_names_}, {bet_type} given.')


def _determine_team_or_side(side):
    "Determines whether the bet is on totals or on spreads."
    side = side.lower()
    if side in ['over', 'under']:
        return 'side'
    if side in ['home', 'away']:
        return 'team'
    raise ValueError(f'side must be one of [BetSide._member_names_], {side} given.')


def _determine_side(side):
    "Determines appropriate side for making a bet."
    side = side.lower()
    if side in ["over", "under"]:
        return side.capitalize()
    if side == "home":
        return "TEAM1"
    if side == "away":
        return "TEAM2"
    raise ValueError(f'side must be one of [over, under, home, away], {side} given.')


if __name__ == '__main__':
    logging.basicConfig(format='%(levelname)s: %(asctime)s %(message)s',
                        filename='pinnacle_api.log', level=logging.DEBUG)

auth.py

"Creates signature and headers for interacting with Pinnacle API"

import base64
from typing import Dict

USERNAME = ''
PASSWORD = ''

def create_signature(username: str, password: str) -> bytes:
    "Given username and password creates base64 encoded signature username:password"

    return base64.b64encode(f'{username}:{password}'.encode('utf-8'))


def create_headers(signature: bytes) -> Dict[str, str]:
    "Given a signature creates required headers for interacting with Pinnacle API"

    return  {
        'Content-length' : '0',
        'Content-type'   : 'application/json',
        'Authorization'  : 'Basic ' + signature.decode('utf-8')
        }

SIGNATURE = create_signature(USERNAME, PASSWORD)
HEADERS = create_headers(SIGNATURE)
\$\endgroup\$
3
+50
\$\begingroup\$

Extracting the error handling for the requests seems to be very easy:

def get(url, **params):
    """Get a URL with authentication, error handling and optional parameters."""
    try:
        response = requests.get(url, headers=auth.HEADERS, params=params)
        response.raise_for_status()
    except requests.ConnectionError:
        logging.critical("No Internet connection")
        return None
    except requests.HTTPError:
        logging.warning("An HTTP error occured.")
        return None
    return response

Which you can then use like this:

def get_events_odds(events: Events) -> Optional[Odds]:
    """Gets odds for the given events."""
    params: RequestParams = {
        'sportId': SOCCER_SPORT_ID,
        'oddsFormat': ODDS_FORMAT,
        'eventIds': ','.join([str(event) for event in events])
        }
    response = get(GET_ODDS_URL, **params)
    if response is not None:
        logging.info("Events odds have been retrieved.")
        return response.json()

A couple of additional notes:

  • Use response.raise_for_status() to also catch also HTTP errors such as 500: Internal Server Error, or 404: Page not Found. You will probably want to add logging for that as well (which I was too lazy to do in the code above).

  • Docstrings are usually denoted with triple quotes """docstring""", even if only one line long.

  • Union[Odds, None] is the same as Optional[Odds], which is slightly more descriptive, IMO.

  • You should use a requests.Session to make consecutive requests to the same server faster.

  • The authentication method is not very secure, although at least the password is not transmitted unencrypted due to the API using https.

  • if 'straightBets' in data.keys() is the same as if 'straightBets' in data.

  • If you have to check if an item is in a list before adding it, you probably want a set instead. However, since your bets are just dictionaries this is not possible. If one of the keys of the dictionaries is a unique identifier, though, you could just use a dictionary:

     bets.update({bet['id']: bet for bet in data['straightBets']
                  if bet['betStatus'] != 'CANCELLED'})
    
  • The requests module can directly work with a dictionary for the data. No need to json.dumps it into a string first.

  • You have a BetType enum, but don't use it when determining bet types. The same is true of the side. I think it would be easier if you used a dictionary:

     SIDES = {"over": "Over", "under": "Under", "home": "TEAM1", "away": "TEAM2"}
    

    Which you can then use like this:

     def _determine_team_or_side(side):
         """Determines whether the bet is on totals or on spreads."""
         side = side.lower()
         try:
             return SIDES[side]
         except KeyError:
             raise ValueError(f'side must be one of {SIDES.keys()}, {side} given.')
    
| improve this answer | |
\$\endgroup\$
  • 1
    \$\begingroup\$ @KonstantinKostanzhoglo Yeah, I figured as much. Just wanted to make sure you are aware that your credentials are there for everyone to see. I'll try to add some example with the enums later. \$\endgroup\$ – Graipher Jul 29 at 13:53
  • 2
    \$\begingroup\$ One advantage of typing.NamedTuple classes (and the new dataclasses.dataclass if you set them up right) is that they're hashable. \$\endgroup\$ – ShapeOfMatter Aug 3 at 16:13
  • 1
    \$\begingroup\$ @shapeofmatter Could you show an example what you mean by "setting them up right"? \$\endgroup\$ – Konstantin Kostanzhoglo Aug 3 at 18:18
  • 2
    \$\begingroup\$ the username and password are transmitted unencrypted No they are not, all requests are through HTTPS, which is encrypted (only the client, the server, and certificate authorities can know it). If you want to hide it from CAs (and the state actors that may control CAs), then I guess you have a much stricter security requirement. \$\endgroup\$ – danuker Aug 6 at 10:11
  • 2
    \$\begingroup\$ In get_events_odds(), response may be None and return response.json() will throw an error: "AttributeError: 'NoneType' object has no attribute 'json'" \$\endgroup\$ – RootTwo Aug 7 at 6:06

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.