My program tracks betting odds for the given list of events and sends notifications when odds reach specified value.

The odds are in the database collected by another program. The required odds are in the same database. Each N seconds the required odds are retrieved from the database, compared with the actual odds and if the latter are good enough, the notification is being sent and required odds get deleted from the "wishlist".

The example of required odds:

[1168358979, 'totals', 'under', 10.5, 2.0]
Interpretation: we are looking for total under 10.5 in event 1168358979 with required odds >= 2.0

Besides general review of my code I'm highly interested in how to add a feature, which allows to specify what should happen with the "wishlist" when odds are good enough: at the moment the required odds are just deleted, however I would like to have an option to "mute" them for a specific period of time, or raise their value by some magnitude.

The program is split into 3 files:

  • odds_tracker.py is an entry point
  • database.py for making database queries
  • telegram.py for sending notifications via telegram


A tool for tracking betting odds for the selected events and sending notifications
when odds reach the value that we are looking for.
from datetime import date
import time
from typing import NamedTuple, Tuple
import database
import telegram


class DesiredOdds(NamedTuple):
    """Represents desired odds."""
    event_id: int
    bet_type: str
    side: str
    points: float
    price: float

def are_odds_good(desired_odds: DesiredOdds, actual_odds: Tuple[float, float]) -> bool:
    Returns True if actual odds are greater than or equal to desired odds.
    Returns False otherwise.
    if desired_odds.side in ['home', 'over']:
        return actual_odds[0] >= desired_odds.price
    elif desired_odds.side in ['away', 'under']:
        return actual_odds[1] >= desired_odds.price
        raise ValueError(f'Side should be home, away, over or under, {desired_odds.side} given.')

def track_odds() -> None:
    Tracks odds for the given list of events, sends notification when odds are good.
    while True:
        tracked_events = database.get_tracked_events()
        for event in tracked_events:
            desired_odds = DesiredOdds(*event[1:])
            actual_odds = database.get_latest_odds(desired_odds.event_id,
            if are_odds_good(desired_odds, actual_odds):
                send_notification(desired_odds, actual_odds)

def send_notification(event: DesiredOdds, actual_odds: Tuple[float, float]) -> None:
    Sends notification about good odds being available.
    if event.side in ['home', 'over']:
        odds = actual_odds[0]
        odds = actual_odds[1]
    event_date, home_team, away_team = database.get_event_info(event.event_id)
    message = create_message(event_date, home_team, away_team, event.bet_type,
                             event.side, event.points, event.price, odds)

def create_message(event_date: date, home_team: str, away_team: str, bet_type: str,
                   side: str, points: float, desired_price: float, odds: float) -> str:
    Creates notification about good odds being available.
    message = f'{event_date} {home_team} - {away_team} {bet_type} {side} {points}\n'
    message += f'{desired_price} required, {odds} current odds. {odds - desired_price:.3f} diff.'
    return message

if __name__ == '__main__':


Functionality for interacting with the database.
from contextlib import contextmanager
from datetime import date
from typing import Optional, Tuple
import pymysql

SERVER = 'localhost'
USER = 'root'
DATABASE = 'bets'

Odds = Tuple[float, float]
TrackedEvent = Tuple[int, str, str, float, float]
TrackedEvents = Tuple[TrackedEvent]

def get_connection():
    Creates database connection.
    connection = pymysql.connect(host=SERVER, user=USER, password=PASSWORD, db=DATABASE)
        yield connection

def get_latest_odds(event_id: int, bet_type: str, points: float) -> Odds:
    Retrieves the latest odds for the given event with bet_type and points.
    with get_connection() as con:
        with con.cursor() as cursor:
            sql = (
                "SELECT left_price, right_price "
                "FROM odds "
                "WHERE event_id = %s "
                "AND bet_type = %s AND points = %s "
                "ORDER BY time_updated DESC "
                "LIMIT 1"
            cursor.execute(sql, (event_id, bet_type, points))
            result = cursor.fetchone()
            return result

def get_tracked_events() -> Optional[TrackedEvents]:
    Retrieves all the tracked events.
    with get_connection() as con:
        with con.cursor() as cursor:
            sql = (
                "SELECT * "
                "FROM tracked_events"
            result = cursor.fetchall()
            return result

def get_event_info(event_id: int) -> Tuple[date, str, str]:
    Retrieves date and teams for the given event.
    with get_connection() as con:
        with con.cursor() as cursor:
            sql = (
                "SELECT match_date, home_team, away_team "
                "FROM fixture "
                "WHERE event_id = %s"
            cursor.execute(sql, (event_id))
            result = cursor.fetchone()
            return result

def delete_event(_id: int) -> None:
    Deletes tracked event with given id.
    with get_connection() as con:
        with con.cursor() as cursor:
            sql = (
                "DELETE FROM tracked_events "
                "WHERE id = %s "
            cursor.execute(sql, (_id))


from typing import Any, Dict
import requests

BASE_URL = f'https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage?'
PARSE_MODE = 'Markdown'

def send_message(message: str) -> Any:
    params: Dict[str, Any] = {
        'chat_id': TELEGRAM_ID,
        'parse_mode': PARSE_MODE,
        'text': message,
    response = requests.get(BASE_URL, params=params)
    return response.json()

1 Answer 1


You are off to a good start indeed. It's evident that this code was carefully done and nothing here looks unreasonable. I do have a few suggestions about error handling, DRY-ing up the code, and code testing/testability.

HTTP requests can fail. I'm sure you know that already, but you should be handle that possibility in send_message() with a try-except -- either directly in the function or at a higher level in the program.

You might need multiple DB environments sooner than you think. I don't know the larger context for you application, but it's not uncommon for a project to immediately (or ultimately) require the ability to connection to databases in different environments. At a minimum, you might want to write automated tests for this code and therefore will want to have both a real/production DB and a test DB. All of which means that you'll need different credentials and connection parameters for each environment. There are many reasonable ways to address that, but a low-tech approach is to define a simple function that returns the correct bundle of connection parameters (as a dict, namedtuple, whatever) based on either an argument (eg, 'test' or 'production') and/or an environment variable and/or a command-line argument. That's a lot of and-or possibilities, I realize, but there's no single answer here. The main point is to use your judgment and be reasonable (don't try to over-engineer it) as you prepare your code for the need for different DB environments.

DRY up those database query functions. I did not study every detail, but the DB functions look reasonable in isolation. But viewed from afar, notice the repetitive pattern that is emerging. That's an indicator of a future problem: if your program's roster of DB queries keeps growing, you'll end up with a mountain of repetitive, almost-but-not-quite equal blocks of tedious code. Here's a rough sketch of how to DRY things up (I did not run it, so there might be typos). There are other approaches that would work well, too. But the general idea is to get this issue on your radar screen, because this type of repetitive DB code can become a real headache if the project gets big.

# This import is a tiny library I wrote. Or you can use enum.Enum for a
# similar approach (but not quite as convenient, IMHO).
from short_con import constants, cons

SqlQueries = cons('SqlQueries',
    get_latest_odds = (
        'SELECT left_price, right_price '
        'FROM odds '
        'WHERE event_id = %s '
        'AND bet_type = %s AND points = %s '
        'ORDER BY time_updated DESC '
        'LIMIT 1'
    get_tracked_events = ('SELECT ... etc'),
    get_event_info = ('SELECT ... etc'),
    delete_event = ('DELETE FROM ... etc'),

QueryModes = constants('QueryModes', 'ONE ALL DELETE')

# You might need to use typing.overload to set up the type checks
# for this general-purpose function, but it is solvable.
def run_db_query(query_key, query_params, mode):
    sql = SQL_QUERIES[query_key]
    with get_connection() as con:
        with con.cursor() as cursor:
            cursor.execute(sql, query_params)
            if mode == QueryModes.ONE:
                return cursor.fetchone()
            elif mode == QueryModes.ALL:
                return cursor.fetchall()
            elif mode == QueryModes.DELETE:
                return con.commit()
                raise ...

def get_latest_odds(event_id: int, bet_type: str, points: float) -> Odds:
    return run_db_query(
        (event_id, bet_type, points),

# Same idea for the other DB functions.

Consider getting out of the world of writing your own SQL. All that said, there are other libraries that will reduce much of this DB code to almost nothing -- everything from full blown ORMs that I would not recommend to more lightweight options that merely simplify the mechanics of DB interactions. You might want to look into those options, if you haven't done so already.

DB interactions can fail. Same point here: you need some exception handling here. But notice how much easier this fix would be if you DRY up the DB code fist (try-except in one place rather than many).

Lingering magic strings. There are still some stragglers (home, over, etc). Define those as constants.

Test your code and the design will usually improve. Speaking of tests, do you have any? If not, get that on your project plan (I recommend pytest but there are several reasonable options). When you try to test your code, you'll probably discover the need for other refactoring steps. If something is difficult to test without awkward mocking and other hoop-jumping, use that pain as a signal that you program design and decomposition might need more adjustments.

Your feature question. I don't have much to suggest, because I don't have enough of the details and context. In general, any time you need to do things "later" that means you'll need to persist that fact outside the program (in your case, probably in the DB). For example, one might have a simple table of MutedDesiredOdds holding the ID of the applicable DesiredOdds entry, some time metadata, and perhaps other simple parameters. Inside your track_odds() loop, you could also check the DB for any actions that were muted but require attention now. Pretty vague suggestions, I realize, but the specifics could influence the approach considerably.

  • \$\begingroup\$ Thanks for the review! As far as error handling is concerned, I'm aware of it and did that in another code I posted, please take a look codereview.stackexchange.com/questions/246150/… Talking about database query functions, your suggestions are very useful, how can I reference it in order to get acquainted with that? Also what are the lightweight options to avoid writing SQL-queries manually? \$\endgroup\$ Sep 13, 2020 at 23:37
  • \$\begingroup\$ I don't have any tests at the moment because this code represents a minimal working version so I would like to get some feedback before going any further. \$\endgroup\$ Sep 13, 2020 at 23:39
  • \$\begingroup\$ **For example, one might have a simple table of MutedDesiredOdds holding the ID of the applicable DesiredOdds entry, some time metadata, and perhaps other simple parameters. ** Could you provide a small example so that I can get an idea how it looks like? The details don't matter, only the skeleton of the solution. \$\endgroup\$ Sep 13, 2020 at 23:41
  • 1
    \$\begingroup\$ @KonstantinKostanzhoglo Imagine a different system where our program must complete "Tasks" on some basis (daily, hourly, etc). The DB would have a tasks table with fields like id, description, frequency, etc. To support the ability to defer tasks under certain conditions, we could have a table called deferred_tasks with fields such as: task_id, deferred_at (timestamp when it was deferred), and maybe n_deferrals (how many times we've already deferred the task). Then, in the main program loop, we check not only tasks but also deferred_tasks to see what actions to take. \$\endgroup\$
    – FMc
    Sep 14, 2020 at 0:49
  • 1
    \$\begingroup\$ @KonstantinKostanzhoglo Regarding ORMs and such, there are many options and even more opinions. On my last project involving a DB I selected sqlalchemy, but only the Core, not the full ORM. But don't necessary take my advice. Do some searching yourself. My anti-ORM concerns might not be as relevant in your specific situation. My project was large, complex, and long-term. On smaller projects, ORMs can work just fine – they are convenient, that's their top selling point. \$\endgroup\$
    – FMc
    Sep 14, 2020 at 1:00

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

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