Tracker delle quote delle scommesse
Il mio programma tiene traccia delle quote di scommessa per l'elenco di eventi fornito e invia notifiche quando le quote raggiungono il valore specificato.
Le quote sono nel database raccolte da un altro programma. Le quote richieste si trovano nello stesso database. Ogni Nsecondo le quote richieste vengono recuperate dal database, confrontate con le quote effettive e se queste ultime sono abbastanza buone, la notifica viene inviata e le quote richieste vengono cancellate dalla "lista dei desideri".
L'esempio delle quote richieste:
[1168358979, 'totals', 'under', 10.5, 2.0]
Interpretazione : stiamo cercando un totale inferiore a 10.5 nell'evento 1168358979 con quota richiesta> = 2.0
Oltre alla revisione generale del mio codice, sono molto interessato a come aggiungere una funzionalità, che consente di specificare cosa dovrebbe accadere con la "lista dei desideri" quando le probabilità sono abbastanza buone: al momento le quote richieste sono solo cancellate, comunque lo vorrei avere un'opzione per "disattivarli" per un periodo di tempo specifico o aumentare il loro valore di una certa grandezza.
Il programma è suddiviso in 3 file:
odds_tracker.pyè un punto di ingressodatabase.pyper eseguire query sul databasetelegram.pyper l'invio di notifiche tramite telegramma
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()
Risposte
Sei partito davvero bene. È evidente che questo codice è stato eseguito con cura e nulla qui sembra irragionevole. Ho alcuni suggerimenti sulla gestione degli errori, sul DRY-up del codice e sul test / testabilità del codice.
Le richieste HTTP possono fallire . Sono sicuro che lo sai già, ma dovresti gestire questa possibilità send_message()con a try-except- direttamente nella funzione o a un livello superiore nel programma.
Potresti aver bisogno di più ambienti DB prima di quanto pensi. Non conosco il contesto più ampio per la tua applicazione, ma non è raro che un progetto richieda immediatamente (o in ultima analisi) la capacità di connettersi a database in ambienti diversi. Come minimo, potresti voler scrivere test automatizzati per questo codice e quindi vorrai avere sia un DB reale / di produzione che un DB di test. Tutto ciò significa che avrai bisogno di credenziali e parametri di connessione diversi per ogni ambiente. Ci sono molti modi ragionevoli per affrontarlo, ma un approccio a bassa tecnologia consiste nel definire una semplice funzione che restituisca il pacchetto corretto di parametri di connessione (come un dict, namedtuple, qualunque cosa) basato su un argomento (ad esempio, 'test' o 'produzione') e / o una variabile di ambiente e / o un argomento della riga di comando. Ci sono molte possibilità, me ne rendo conto, ma ci sononon c'è una sola risposta qui. Il punto principale è usare il tuo giudizio ed essere ragionevole (non cercare di ingegnerizzarlo eccessivamente) mentre prepari il tuo codice per la necessità di diversi ambienti DB.
DRY up quelle funzioni di query del database . Non ho studiato ogni dettaglio, ma le funzioni DB sembrano ragionevoli da sole. Ma visto da lontano, nota lo schema ripetitivo che sta emergendo. Questo è un indicatore di un problema futuro: se l'elenco delle query DB del tuo programma continua a crescere, ti ritroverai con una montagna di blocchi ripetitivi, quasi ma non uguali di codice noioso. Ecco uno schizzo approssimativo di come ASCIUGARE le cose (non l'ho eseguito, quindi potrebbero esserci errori di battitura). Ci sono anche altri approcci che funzionerebbero bene. Ma l'idea generale è di visualizzare questo problema sullo schermo del radar, perché questo tipo di codice DB ripetitivo può diventare un vero grattacapo se il progetto diventa grande.
# 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()
else:
raise ...
def get_latest_odds(event_id: int, bet_type: str, points: float) -> Odds:
return run_db_query(
SqlQueries.get_latest_odds,
(event_id, bet_type, points),
QueryModes.ONE,
)
# Same idea for the other DB functions.
...
Considera l'idea di uscire dal mondo della scrittura del tuo SQL . Detto questo, ci sono altre librerie che ridurranno gran parte di questo codice DB a quasi nulla: tutto da ORM in piena regola che non consiglierei a opzioni più leggere che semplificano semplicemente i meccanismi delle interazioni DB. Potresti voler esaminare queste opzioni, se non l'hai già fatto.
Le interazioni DB possono fallire . Stesso punto qui: hai bisogno di una gestione delle eccezioni qui. Ma nota quanto sarebbe più facile questa correzione se DRY il pugno del codice DB ( try-exceptin un posto piuttosto che in molti).
Stringhe magiche persistenti . Ci sono ancora alcuni ritardatari ( home, over, ecc). Definiscili come costanti.
Testa il tuo codice e il design di solito migliorerà . A proposito di test, ne hai? In caso contrario, inseriscilo nel tuo piano di progetto (consiglio pytest ma ci sono diverse opzioni ragionevoli). Quando provi a testare il tuo codice, probabilmente scoprirai la necessità di altri passaggi di refactoring. Se qualcosa è difficile da testare senza imbarazzanti derisioni e altri salti, usa quel dolore come segnale che la progettazione e la decomposizione del programma potrebbero richiedere ulteriori aggiustamenti.
La tua domanda sulle funzionalità . Non ho molto da suggerire, perché non ho abbastanza dettagli e contesto. In generale, ogni volta che hai bisogno di fare le cose "più tardi" significa che dovrai persistere quel fatto fuori dal programma (nel tuo caso, probabilmente nel DB). Ad esempio, si potrebbe avere una semplice tabella MutedDesiredOddscontenente l'ID della DesiredOddsvoce applicabile , alcuni metadati temporali e forse altri semplici parametri. All'interno del tuo track_odds()ciclo, puoi anche controllare il DB per eventuali azioni che sono state disattivate ma richiedono attenzione ora. Suggerimenti piuttosto vaghi, mi rendo conto, ma le specifiche potrebbero influenzare notevolmente l'approccio.