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.pyis an entry pointdatabase.pyfor making database queriestelegram.pyfor sending notifications via telegram
odds_tracker.py
"""
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
REQUESTS_DELAY = 5
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
else:
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,
desired_odds.bet_type,
desired_odds.points)
if are_odds_good(desired_odds, actual_odds):
send_notification(desired_odds, actual_odds)
database.delete_event(event[0])
time.sleep(REQUESTS_DELAY)
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]
else:
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)
telegram.send_message(message)
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__':
track_odds()
database.py
"""
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'
PASSWORD = ''
DATABASE = 'bets'
Odds = Tuple[float, float]
TrackedEvent = Tuple[int, str, str, float, float]
TrackedEvents = Tuple[TrackedEvent]
@contextmanager
def get_connection():
"""
Creates database connection.
"""
connection = pymysql.connect(host=SERVER, user=USER, password=PASSWORD, db=DATABASE)
try:
yield connection
finally:
connection.close()
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"
)
cursor.execute(sql)
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))
con.commit()
telegram.py
from typing import Any, Dict
import requests
TELEGRAM_TOKEN = ''
TELEGRAM_ID = ''
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()
