bot_tg/games/betting.py

589 lines
22 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

import logging
import time as _time
import hashlib
import random
import urllib.request
import json
import gzip
from datetime import datetime
from zoneinfo import ZoneInfo
from bs4 import BeautifulSoup
import psycopg2
import psycopg2.extras
import aiohttp
import config
from db import get_conn
from games.casino import get_user_length, update_user_length
logger = logging.getLogger(__name__)
_matches_cache: dict[str, tuple[float, list]] = {}
_sports_cache: tuple[float, list] | None = None
_last_viewed_sport_by_user: dict[int, str] = {}
_DRAW_NAMES = {"draw", "tie", "ничья"}
BET_HISTORY_DAYS = 30
_LIQUIPEDIA_CACHE_TTL = 300 # 5 minutes cache to respect Liquipedia rate limits
def _init_bets_db() -> None:
with get_conn() as conn:
cur = conn.cursor()
cur.execute('''
CREATE TABLE IF NOT EXISTS bets (
id SERIAL PRIMARY KEY,
user_id BIGINT NOT NULL,
match_id TEXT NOT NULL,
sport TEXT NOT NULL,
home_team TEXT NOT NULL,
away_team TEXT NOT NULL,
chosen_team TEXT NOT NULL,
amount REAL NOT NULL,
odds REAL NOT NULL,
status TEXT DEFAULT 'pending',
payout REAL DEFAULT 0,
created_ts BIGINT,
resolved_ts BIGINT
)
''')
cur.execute("CREATE INDEX IF NOT EXISTS idx_bets_user ON bets(user_id, status)")
cur.execute("CREATE INDEX IF NOT EXISTS idx_bets_match ON bets(match_id, status)")
cur.execute("CREATE INDEX IF NOT EXISTS idx_bets_created ON bets(created_ts)")
cur.execute("ALTER TABLE bets ADD COLUMN IF NOT EXISTS payout REAL DEFAULT 0")
cur.execute("ALTER TABLE bets ADD COLUMN IF NOT EXISTS resolved_ts BIGINT")
try:
_init_bets_db()
except Exception:
logger.warning("Could not init bets DB at import time (will retry on first use)")
def remember_user_sport_context(user_id: int, sport_alias: str) -> None:
if sport_alias in config.ODDS_SPORTS:
_last_viewed_sport_by_user[user_id] = sport_alias
def clear_user_sport_context(user_id: int) -> None:
_last_viewed_sport_by_user.pop(user_id, None)
def get_user_sport_context(user_id: int) -> str | None:
return _last_viewed_sport_by_user.get(user_id)
async def _fetch_available_sports() -> list[dict]:
return [{"key": "dota2", "title": "🎮 Dota 2", "active": True}]
async def fetch_liquipedia_matches() -> list[dict]:
global _matches_cache
now = _time.time()
cached = _matches_cache.get("liquipedia_all")
if cached and (now - cached[0]) < _LIQUIPEDIA_CACHE_TTL:
return cached[1]
url = "https://liquipedia.net/dota2/api.php?action=parse&page=Liquipedia:Matches&format=json"
headers = {
"User-Agent": "Dota2BettingTelegramBot/1.0 (contact: danil@example.com)",
"Accept-Encoding": "gzip"
}
try:
async with aiohttp.ClientSession() as session:
async with session.get(url, headers=headers, timeout=aiohttp.ClientTimeout(total=15)) as resp:
if resp.status != 200:
logger.warning(f"Liquipedia API returned {resp.status}")
return cached[1] if cached else []
data = await resp.json()
html_content = data['parse']['text']['*']
soup = BeautifulSoup(html_content, "lxml")
match_divs = soup.find_all("div", class_="match-info")
matches = []
for idx, div in enumerate(match_divs):
# Tournament
tournament_div = div.find("div", class_="match-info-tournament")
tournament = tournament_div.get_text(strip=True) if tournament_div else "Unknown Tournament"
# Teams
opponents = div.find_all("div", class_="match-info-header-opponent")
team1 = "TBD"
team2 = "TBD"
winner = None
completed = False
if len(opponents) >= 2:
o1, o2 = opponents[0], opponents[1]
team1 = o1.get_text(strip=True)
team2 = o2.get_text(strip=True)
c1 = o1.get("class", [])
c2 = o2.get("class", [])
has_win_loss = any("winner" in c or "loser" in c for c in c1 + c2)
if has_win_loss:
completed = True
if any("winner" in c for c in c1):
winner = team1
elif any("winner" in c for c in c2):
winner = team2
else:
teams = div.find_all("span", class_="block-team")
if len(teams) >= 2:
team1 = teams[0].get_text(strip=True)
team2 = teams[1].get_text(strip=True)
# Start time
timer_span = div.find("span", class_="timer-object")
timestamp = ""
if timer_span:
timestamp = timer_span.get("data-timestamp", "")
if not team1 or not team2 or team1 == "TBD" or team2 == "TBD":
continue
# ID generation
match_id = hashlib.md5(f"lp_{team1}_{team2}_{timestamp}".encode('utf-8')).hexdigest()
# Deterministic Odds
seed = int(hashlib.md5(match_id.encode('utf-8')).hexdigest(), 16) % 10**8
r = random.Random(seed)
odds_home = round(r.uniform(1.3, 3.2), 2)
margin = 1.08
prob_home = 1.0 / odds_home
prob_away = margin - prob_home
if prob_away < 0.1:
prob_away = 0.1
odds_away = round(1.0 / prob_away, 2)
matches.append({
"id": match_id,
"home": team1,
"away": team2,
"commence": datetime.fromtimestamp(int(timestamp)).isoformat() + "Z" if timestamp else "",
"timestamp": timestamp,
"completed": completed,
"winner": winner,
"odds": {
team1: odds_home,
team2: odds_away
},
"tournament": tournament
})
_matches_cache["liquipedia_all"] = (now, matches)
logger.info(f"Scraped and cached {len(matches)} matches from Liquipedia")
return matches
except Exception:
logger.exception("Failed to fetch matches from Liquipedia")
return cached[1] if cached else []
async def fetch_matches(sport_key: str = "esports_dota2") -> list[dict]:
all_matches = await fetch_liquipedia_matches()
# Return only active (not completed) matches for betting
return [m for m in all_matches if not m["completed"]]
def _get_match_outcomes(match: dict) -> list[dict[str, str | float]]:
odds = match.get("odds", {})
outcomes: list[dict[str, str | float]] = []
added: set[str] = set()
def append_outcome(code: str, name: str) -> None:
if name in odds and name not in added:
outcomes.append({"code": code, "name": name, "odds": odds[name]})
added.add(name)
home = match.get("home", "")
away = match.get("away", "")
append_outcome("1", home)
append_outcome("2", away)
return outcomes
def describe_match_outcomes(match: dict) -> str:
return ", ".join(
f"{item['code']}={item['name']} x{item['odds']:.2f}"
for item in _get_match_outcomes(match)
)
def resolve_match_outcome(match: dict, raw_choice: str) -> str | None:
choice = raw_choice.strip()
if not choice:
return None
normalized = choice.casefold()
outcomes = _get_match_outcomes(match)
for item in outcomes:
if normalized == str(item["code"]).casefold():
return str(item["name"])
for item in outcomes:
if normalized == str(item["name"]).casefold():
return str(item["name"])
try:
choice_odds = round(float(choice.replace(",", ".")), 2)
except ValueError:
return None
for item in outcomes:
if round(float(item["odds"]), 2) == choice_odds:
return str(item["name"])
return None
def format_matches(matches: list[dict], sport_alias: str, sport_label: str) -> str:
if not matches:
return f"🏟 {sport_label} [{sport_alias}] — матчей не найдено."
lines = [f"🏟 {sport_label} [{sport_alias}] — ближайшие матчи:\n"]
for i, m in enumerate(matches[:10], 1):
outcomes = _get_match_outcomes(m)
odds_str = "\n".join(
f" {item['code']}) {item['name']}: {item['odds']:.2f}"
for item in outcomes
)
start_str = ""
if m["commence"]:
try:
dt = datetime.fromisoformat(m["commence"].replace("Z", "+00:00"))
msk = dt.astimezone(ZoneInfo(config.DEFAULT_TZ))
start_str = f"\n 🕐 {msk.strftime('%d.%m %H:%M')} МСК"
except Exception:
pass
lines.append(f"{i}️⃣ {m['home']} vs {m['away']}{start_str}\n{odds_str}\n")
lines.append("Ставка: /bet [номер] [исход] [ставка]")
lines.append("Исход: 1/2. Можно писать ещё название команды или коэффициент.")
return "\n".join(lines)
def _get_sport_label(key: str) -> str:
return "🎮 Dota 2"
async def get_formatted_matches(sport_alias: str | None) -> str:
matches = await fetch_matches()
return format_matches(matches, "dota2", "🎮 Dota 2")
async def get_match_by_index(sport_alias: str, index: int) -> dict | None:
matches = await fetch_matches()
if 0 < index <= len(matches):
return matches[index - 1]
return None
def place_bet(user_id: int, match: dict, sport: str, chosen_team: str, amount: float) -> tuple[bool, str]:
current = get_user_length(user_id)
if current is None:
return False, "Сначала заведи счёт через /penis, братуха."
if current <= 0:
return False, "🚫 С кредитом ставки не принимаем."
if amount <= 0:
return False, "Ставка должна быть больше нуля."
if amount > config.BET_MAX_AMOUNT:
return False, f"Максимальная ставка — {config.BET_MAX_AMOUNT} см."
if round(amount, 2) > round(current, 2):
return False, f"У тебя {current:.1f} см, а ставишь {amount:.1f}. Не хватает, фраер."
team_odds = match["odds"].get(chosen_team)
if team_odds is None:
return False, f"Исход не найден. Доступные: {describe_match_outcomes(match)}"
new_length = update_user_length(user_id, -amount)
if new_length is None:
return False, "Ошибка БД."
now_ts = int(_time.time())
try:
with get_conn() as conn:
cur = conn.cursor()
cur.execute('''
INSERT INTO bets
(user_id, match_id, sport, home_team, away_team,
chosen_team, amount, odds, status, payout, created_ts)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, 'pending', 0, %s)
''', (
user_id, match["id"], sport, match["home"], match["away"],
chosen_team, amount, team_odds, now_ts,
))
except psycopg2.Error:
logger.exception("Failed to place bet")
update_user_length(user_id, amount)
return False, "Ошибка БД."
potential = round(amount * team_odds, 1)
return True, (
f"✅ Ставка принята!\n"
f"📌 {chosen_team} (x{team_odds:.2f})\n"
f"💰 Ставка: {amount:.1f} см\n"
f"🎯 Возможный выигрыш: {potential:.1f} см\n"
f"📏 Остаток: {new_length:.1f} см"
)
def get_user_bets(user_id: int) -> str:
try:
with get_conn() as conn:
cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
cur.execute('''
SELECT * FROM bets
WHERE user_id = %s AND status = 'pending'
ORDER BY created_ts DESC LIMIT 10
''', (user_id,))
rows = cur.fetchall()
except psycopg2.Error:
return "Ошибка БД."
if not rows:
return "У тебя нет активных ставок."
lines = ["📋 Твои активные ставки:\n"]
for r in rows:
potential = round(r["amount"] * r["odds"], 1)
created = ""
if r["created_ts"]:
dt = datetime.fromtimestamp(r["created_ts"], tz=ZoneInfo(config.DEFAULT_TZ))
created = f" 📅 {dt.strftime('%d.%m %H:%M')}\n"
lines.append(
f"{r['home_team']} vs {r['away_team']}\n"
f" {r['chosen_team']} (x{r['odds']:.2f}), {r['amount']:.1f} см → {potential:.1f} см\n"
f"{created}"
)
return "\n".join(lines)
def get_user_bet_history(user_id: int) -> str:
cutoff_ts = int(_time.time() - BET_HISTORY_DAYS * 86400)
try:
with get_conn() as conn:
cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
cur.execute('''
SELECT * FROM bets
WHERE user_id = %s AND created_ts >= %s
ORDER BY created_ts DESC LIMIT 50
''', (user_id, cutoff_ts))
rows = cur.fetchall()
except psycopg2.Error:
logger.exception("Failed to get bet history")
return "Ошибка БД."
if not rows:
return "📋 За последний месяц ставок не было."
total_bet = total_won = total_lost = 0.0
wins = losses = pending = 0
lines = [f"📋 История ставок за {BET_HISTORY_DAYS} дней:\n"]
for r in rows:
status = r["status"]
amount = float(r["amount"])
odds_val = float(r["odds"])
payout_raw = float(r["payout"] or 0)
total_bet += amount
if status == "won":
wins += 1
actual_payout = payout_raw if payout_raw > 0 else round(amount * odds_val, 1)
total_won += actual_payout
status_icon = ""
result_text = f"+{actual_payout:.1f} см"
elif status == "lost":
losses += 1
total_lost += amount
status_icon = ""
result_text = f"-{amount:.1f} см"
else:
pending += 1
status_icon = ""
result_text = f"ожидание → {round(amount * odds_val, 1):.1f} см"
created = ""
if r["created_ts"]:
dt = datetime.fromtimestamp(r["created_ts"], tz=ZoneInfo(config.DEFAULT_TZ))
created = dt.strftime('%d.%m %H:%M')
lines.append(
f"{status_icon} {r['home_team']} vs {r['away_team']}\n"
f" {r['chosen_team']} (x{odds_val:.2f}) | {amount:.1f} см | {result_text}\n"
f" 📅 {created}"
)
net = total_won - total_lost
net_sign = "+" if net >= 0 else ""
lines.append("")
lines.append("━━━ 📊 СТАТИСТИКА ━━━")
lines.append(f"🎰 Всего ставок: {len(rows)}")
lines.append(f"✅ Побед: {wins}")
lines.append(f"❌ Поражений: {losses}")
if pending:
lines.append(f"В ожидании: {pending}")
lines.append(f"💰 Поставлено: {total_bet:.1f} см")
lines.append(f"🏆 Выиграно: +{total_won:.1f} см")
lines.append(f"💸 Проиграно: -{total_lost:.1f} см")
lines.append(f"📈 Итого: {net_sign}{net:.1f} см")
if wins + losses > 0:
lines.append(f"📊 Винрейт: {wins / (wins + losses) * 100:.0f}%")
return "\n".join(lines)
def cleanup_old_bets() -> int:
cutoff_ts = int(_time.time() - BET_HISTORY_DAYS * 86400)
try:
with get_conn() as conn:
cur = conn.cursor()
cur.execute(
"DELETE FROM bets WHERE status != 'pending' AND created_ts < %s",
(cutoff_ts,),
)
deleted = cur.rowcount
if deleted > 0:
logger.info(f"Cleaned up {deleted} old bets")
return deleted
except psycopg2.Error:
logger.exception("Failed to cleanup old bets")
return 0
_REFUND_AFTER_SECONDS = 7 * 86400 # авторефанд ставок старше 7 дней
def _refund_expired_bets() -> list[tuple[int, str]]:
"""Возвращает деньги за ставки, которые висят более 7 дней без результата."""
cutoff = int(_time.time()) - _REFUND_AFTER_SECONDS
notifications: list[tuple[int, str]] = []
try:
with get_conn() as conn:
cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
cur.execute(
"SELECT * FROM bets WHERE status = 'pending' AND created_ts < %s",
(cutoff,),
)
expired = cur.fetchall()
if not expired:
return []
wcur = conn.cursor()
now_ts = int(_time.time())
for bet in expired:
update_user_length(bet["user_id"], bet["amount"])
wcur.execute(
"UPDATE bets SET status = 'refunded', resolved_ts = %s WHERE id = %s",
(now_ts, bet["id"]),
)
notifications.append((
bet["user_id"],
f"↩️ Ставка возвращена\n"
f"🏟 {bet['home_team']} vs {bet['away_team']}\n"
f"📌 {bet['chosen_team']} — результат так и не пришёл\n"
f"💰 Возврат: {bet['amount']:.1f} см",
))
logger.info("Refunded expired bet id=%s user=%s amount=%s", bet["id"], bet["user_id"], bet["amount"])
except psycopg2.Error:
logger.exception("Failed to refund expired bets")
return notifications
async def settle_bets() -> list[tuple[int, str]]:
"""Возвращает список (user_id, текст_уведомления) для отправки в Telegram."""
notifications: list[tuple[int, str]] = []
# Сначала авторефанд совсем старых ставок
notifications.extend(_refund_expired_bets())
try:
with get_conn() as conn:
cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
cur.execute("SELECT DISTINCT match_id FROM bets WHERE status = 'pending'")
pending = cur.fetchall()
except psycopg2.Error:
return notifications
if not pending:
return notifications
all_matches = await fetch_liquipedia_matches()
match_map = {m["id"]: m for m in all_matches}
now_ts = int(_time.time())
for row in pending:
match_id = row["match_id"]
event = match_map.get(match_id)
if not event or not event.get("completed"):
continue
winner = event.get("winner")
if not winner:
continue
is_draw = (winner.casefold() in _DRAW_NAMES or winner == "Draw")
try:
with get_conn() as conn:
cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
cur.execute(
"SELECT * FROM bets WHERE match_id = %s AND status = 'pending'",
(match_id,),
)
bets = cur.fetchall()
wcur = conn.cursor()
for bet in bets:
chosen = bet["chosen_team"]
if is_draw:
bet_won = chosen.casefold() in _DRAW_NAMES or chosen == "Draw"
else:
bet_won = chosen == winner
if bet_won:
winnings = round(bet["amount"] * bet["odds"], 1)
update_user_length(bet["user_id"], winnings)
wcur.execute(
"UPDATE bets SET status = 'won', payout = %s, resolved_ts = %s WHERE id = %s",
(winnings, now_ts, bet["id"]),
)
notifications.append((
bet["user_id"],
f"🎉 Ставка сыграла!\n"
f"🏟 {bet['home_team']} vs {bet['away_team']}\n"
f"📌 {bet['chosen_team']} (x{bet['odds']:.2f})\n"
f"💰 Выигрыш: +{winnings:.1f} см",
))
logger.info("Bet won: user=%s match=%s winnings=%s", bet["user_id"], match_id, winnings)
else:
wcur.execute(
"UPDATE bets SET status = 'lost', payout = 0, resolved_ts = %s WHERE id = %s",
(now_ts, bet["id"]),
)
notifications.append((
bet["user_id"],
f"❌ Ставка не сыграла\n"
f"🏟 {bet['home_team']} vs {bet['away_team']}\n"
f"📌 {bet['chosen_team']} (x{bet['odds']:.2f})\n"
f"💸 Потеряно: {bet['amount']:.1f} см",
))
logger.info("Bet lost: user=%s match=%s amount=%s", bet["user_id"], match_id, bet["amount"])
except psycopg2.Error:
logger.exception("Failed to settle bets")
cleanup_old_bets()
return notifications
async def debug_sports() -> str:
return "🎮 Dota 2 — Scraper Liquipedia активен. Odds API отключен."