""" Stake.com sportsbook scraper. Stake uses GraphQL at https://stake.com/_api/graphql, protected by Cloudflare. All requests are made via page.evaluate() from within the Playwright browser context so that CF cookies are sent automatically (same pattern as Betpawa). API flow per refresh: 1. Batch-fetch ALL category tournament lists in one GQL call (10 aliases) 2. Batch-fetch fixture lists for all tournaments in groups of 20 3. Collect active (prematch) match slugs 4. Batch-fetch odds via slugFixture in groups of 50 GraphQL endpoint: https://stake.com/_api/graphql Key queries: sport(sportId: uuid) { categoryList { id name slug } } slugCategory(category: "england", sport: "soccer") { tournamentList(limit:50) { id name slug } } slugTournament(tournament: "premier-league", category: "england", sport: "soccer") { fixtureList(limit: 50) { id name slug startTime status } } slugFixture(fixture: "46377219-west-ham-united-wolverhampton") { name startTime status tournament { name } groups { name templates(includeEmpty: false) { extId name markets { status specifiers outcomes { name extId odds } } } } } Fixture status values: "active" → prematch (upcoming) "live" → in-play "ended" → finished Football market template extIds (Sportradar standard): extId=1 → 1X2 outcome extIds: '1'=Home '2'=Draw '3'=Away extId=18 → Asian Total outcome extIds: '12'=Over '13'=Under filter specifiers=total=2.5 extId=29 → Both Teams to Score outcome extIds: '74'=Yes '76'=No Tennis market template extIds: extId=186 → Winner (2-way match result, outcome names = player names) Sport slugs used with slugSport(sport: ...): soccer / basketball / tennis (basketball is absent from sportList but accessible via slugSport) Note: Stake.com accepts crypto only — no Nigerian fiat. Relevant for global arb opportunities since the platform has different odds from Nigerian bookmakers. """ import json import logging import queue import threading import time from datetime import datetime from typing import List, Optional, Dict, Tuple import psutil from scrapers.base import BaseScraper, normalize_cs_score from core.models import Event, Outcome logger = logging.getLogger(__name__) BASE_URL = 'https://stake.com' GQL_URL = f'{BASE_URL}/_api/graphql' SPORT_SLUGS = { 'football': 'soccer', 'basketball': 'basketball', 'tennis': 'tennis', } # template extId → (market_name, {outcome_extId: label}) # None outcome map = use outcome names directly (tennis) FOOTBALL_MARKETS: Dict[str, Tuple[str, Optional[Dict[str, str]]]] = { '1': ('1X2', {'1': 'Home', '2': 'Draw', '3': 'Away'}), '5': ('HT 1X2', {'1': 'Home', '2': 'Draw', '3': 'Away'}), '10': ('Double Chance', {'8': '1X', '9': 'X2', '10': '12'}), '11': ('Asian Handicap', {'714': 'Home', '715': 'Away'}), # line from specifiers '18': ('Over/Under', {'12': 'Over', '13': 'Under'}), # line detected from specifiers '26': ('Draw No Bet', {'60': 'Home', '61': 'Away'}), '29': ('BTTS', {'74': 'Yes', '76': 'No'}), '47': ('Correct Score', None), # outcome names = score strings '68': ('HT Over/Under', {'12': 'Over', '13': 'Under'}), # 1st Half O/U, line from specifiers '90': ('2H Over/Under', {'12': 'Over', '13': 'Under'}), # 2nd Half O/U, line from specifiers } # O/U lines accepted for extId=18 _OU_LINES = ('0.5', '1.5', '2.5', '3.5', '4.5') _HT_OU_LINES = ('0.5', '1.5') _2H_OU_LINES = ('0.5', '1.5', '2.5') BASKETBALL_MARKETS: Dict[str, Tuple[str, Optional[Dict[str, str]]]] = { '219': ('Home/Away', {'4': 'Home', '5': 'Away'}), # Winner incl. OT '18': ('Over/Under', {'12': 'Over', '13': 'Under'}), # Total points (lines extracted dynamically) } TENNIS_MARKETS: Dict[str, Tuple[str, Optional[Dict[str, str]]]] = { '186': ('Home/Away', None), # outcome names = player names; first = home } MARKETS_CONFIG = { 'football': FOOTBALL_MARKETS, 'basketball': BASKETBALL_MARKETS, 'tennis': TENNIS_MARKETS, } # How many tournament fixture lists per batch GQL call TOURNAMENT_BATCH = 20 # How many match slugFixture per batch GQL call FIXTURE_BATCH = 50 # JS template executed from browser context via page.evaluate() _GQL_JS = """ async ({url, body}) => { const r = await fetch(url, { method: 'POST', credentials: 'include', headers: { 'Content-Type': 'application/json', 'Accept': 'application/json', }, body: JSON.stringify(body), }); if (!r.ok) throw new Error('HTTP ' + r.status); return await r.json(); } """ # Fires multiple GQL requests in parallel via Promise.all — one worker queue entry # instead of N sequential ones. Each request is {url, body}. _GQL_PARALLEL_JS = """ async (requests) => { const results = await Promise.all(requests.map(async ({url, body}) => { try { const r = await fetch(url, { method: 'POST', credentials: 'include', headers: {'Content-Type': 'application/json', 'Accept': 'application/json'}, body: JSON.stringify(body), }); if (!r.ok) return {error: 'HTTP ' + r.status}; return await r.json(); } catch(e) { return {error: e.message}; } })); return results; } """ _CAT_MAP_TTL = 3600 # rebuild category/tournament map every 60 min _FIXTURE_TTL = 600 # rebuild fixture slug list every 10 min class _PlaywrightWorker(threading.Thread): """ Dedicated daemon thread that owns the Playwright browser session. Playwright's sync API raises an error when started inside a running asyncio event loop (which APScheduler threads may have). Running Playwright entirely in a plain threading.Thread avoids this — plain threads have no asyncio loop. Callers send work via call(): a callable that receives the page object and returns a value. Results (or exceptions) are returned synchronously. """ def __init__(self): super().__init__(daemon=True, name='stake-playwright') self._q: queue.Queue = queue.Queue() self._ready = threading.Event() self._failed: Optional[Exception] = None self._pw = None self.start() # Wait for browser init before returning (up to 120s) if not self._ready.wait(timeout=120): raise RuntimeError('[Stake] browser worker timed out during init') if self._failed: raise self._failed def run(self): """Browser lifecycle lives entirely in this thread.""" try: from playwright.sync_api import sync_playwright pw = sync_playwright().start() self._pw = pw browser = pw.chromium.launch( headless=True, executable_path='/usr/bin/google-chrome-stable', args=[ '--no-sandbox', '--disable-blink-features=AutomationControlled', '--disable-gpu', '--disable-dev-shm-usage', '--disable-extensions', '--disable-background-networking', '--no-first-run', ], ) ctx = browser.new_context( locale='en-US', user_agent=( 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) ' 'AppleWebKit/537.36 (KHTML, like Gecko) ' 'Chrome/120.0.0.0 Safari/537.36' ), ) page = ctx.new_page() page.goto(f'{BASE_URL}/sports/soccer', wait_until='domcontentloaded', timeout=90_000) time.sleep(8) logger.info('[Stake] browser worker ready') except Exception as ex: self._failed = ex self._ready.set() return self._ready.set() # Process work items until sentinel (None) is received while True: item = self._q.get() if item is None: break fn, result_q = item try: result_q.put(('ok', fn(page))) except Exception as ex: result_q.put(('err', ex)) try: browser.close() except Exception: pass try: pw.stop() except Exception: pass def call(self, fn, timeout: float = 120): """Run fn(page) in the worker thread and return its result (or raise).""" result_q: queue.Queue = queue.Queue() self._q.put((fn, result_q)) try: status, val = result_q.get(timeout=timeout) except queue.Empty: raise TimeoutError('[Stake] worker call timed out') if status == 'err': raise val return val def _kill_processes(self): """Force-kill the Playwright node driver and all its Chrome children.""" pw, self._pw = self._pw, None if pw is None: return try: node_pid = pw._impl_obj._connection._transport._proc.pid proc = psutil.Process(node_pid) for child in proc.children(recursive=True): try: child.kill() except psutil.NoSuchProcess: pass proc.kill() except Exception: pass def stop(self): self._kill_processes() while True: try: self._q.get_nowait() except queue.Empty: break self._q.put(None) class StakeScraper(BaseScraper): def __init__(self): super().__init__('Stake') self._worker: Optional[_PlaywrightWorker] = None self._worker_lock = threading.Lock() # Per-sport category/tournament map — rebuilt every _CAT_MAP_TTL seconds self._cat_maps: Dict[str, Dict[str, str]] = {} # sport → {t_slug: c_slug} self._cat_maps_ts: Dict[str, float] = {} # sport → build timestamp # Per-sport fixture slug cache — rebuilt every _FIXTURE_TTL seconds self._fixture_cache: Dict[str, Tuple[float, List]] = {} # "sport:status" → (ts, slugs) # ── Lifecycle ───────────────────────────────────────────────────────────── def _ensure_browser(self) -> bool: with self._worker_lock: if self._worker is not None and self._worker.is_alive(): return True try: self._worker = _PlaywrightWorker() return True except Exception as ex: logger.error(f'[Stake] browser init failed: {ex}') self._worker = None return False def _close_browser(self): with self._worker_lock: if self._worker: try: self._worker.stop() except Exception: pass self._worker = None # Keep caches — category/tournament structure doesn't change when browser restarts # ── Public ──────────────────────────────────────────────────────────────── def get_all_events(self) -> List[Event]: if not self._ensure_browser(): return [] all_events: List[Event] = [] for sport in ('football', 'basketball', 'tennis'): try: events = self._fetch_sport(sport) all_events.extend(events) logger.info(f'[Stake] {sport}: {len(events)} events') except Exception as ex: logger.error(f'[Stake] {sport} error: {ex}') return all_events def get_events(self, sport: str) -> List[Event]: if not self._ensure_browser(): return [] try: return self._fetch_sport(sport) except Exception as ex: logger.error(f'[Stake] {sport} error: {ex}') self._close_browser() return [] def get_all_live_events(self) -> List[Event]: if not self._ensure_browser(): return [] all_events: List[Event] = [] for sport in ('football', 'basketball', 'tennis'): try: events = self._fetch_sport(sport, fixture_status='live') all_events.extend(events) if events: logger.info(f'[Stake] live {sport}: {len(events)} events') except Exception as ex: logger.error(f'[Stake] live {sport} error: {ex}') return all_events def get_live_events(self, sport: str) -> List[Event]: if not self._ensure_browser(): return [] try: return self._fetch_sport(sport, fixture_status='live') except Exception as ex: logger.error(f'[Stake] live {sport} error: {ex}') self._close_browser() return [] # ── Fetch pipeline ──────────────────────────────────────────────────────── def _fetch_sport(self, sport: str, fixture_status: str = 'active') -> List[Event]: sport_slug = SPORT_SLUGS.get(sport) if not sport_slug: return [] self._build_cat_map(sport, sport_slug) # Get fixture slugs with the requested status match_slugs = self._get_all_fixture_slugs(sport, sport_slug, fixture_status=fixture_status) if not match_slugs: logger.warning(f'[Stake] no {fixture_status} fixtures found for {sport}') return [] logger.info(f'[Stake] {sport}: {len(match_slugs)} {fixture_status} fixtures') # Batch-fetch odds return self._fetch_odds_batched(match_slugs, sport) def _build_cat_map(self, sport: str, sport_slug: str): """Fetch all categories and their tournaments for this sport. TTL-cached.""" now = time.time() if now - self._cat_maps_ts.get(sport, 0) < _CAT_MAP_TTL: return # still fresh # Step 1: get categories for this sport st, data = self._gql('CatList', '''query CatList($sport: String!) { slugSport(sport: $sport) { categoryList { id name slug } } }''', {'sport': sport_slug}) if 'errors' in data or 'data' not in data: logger.error(f'[Stake] categoryList error for {sport}: {data.get("errors")}') return categories = (data['data'].get('slugSport') or {}).get('categoryList', []) cat_slugs = [c['slug'] for c in categories] if not cat_slugs: return # Step 2: batch-fetch all tournament lists per category in one GQL call parts = [f'query BatchCats({", ".join(f"$c{i}: String!" for i in range(len(cat_slugs)))}, $sport: String!) {{'] for i, cs in enumerate(cat_slugs): parts.append(f' cat{i}: slugCategory(category: $c{i}, sport: $sport) {{ tournamentList(limit: 100) {{ slug }} }}') parts.append('}') batch_q = '\n'.join(parts) vs = {f'c{i}': cs for i, cs in enumerate(cat_slugs)} vs['sport'] = sport_slug st2, data2 = self._gql('BatchCats', batch_q, vs) cat_map: Dict[str, str] = {} if 'data' in data2: for i, cs in enumerate(cat_slugs): cat_data = data2['data'].get(f'cat{i}') or {} for t in cat_data.get('tournamentList', []): cat_map[t['slug']] = cs self._cat_maps[sport] = cat_map self._cat_maps_ts[sport] = now logger.debug(f'[Stake] cat_map[{sport}] built: {len(cat_map)} tournaments') def _get_all_fixture_slugs(self, sport: str, sport_slug: str, fixture_status: str = 'active') -> List[dict]: """ Collect fixture slugs matching fixture_status for this sport. Results are TTL-cached (_FIXTURE_TTL seconds) — the list of upcoming matches changes slowly; odds are always re-fetched separately. """ cache_key = f'{sport}:{fixture_status}' now = time.time() if cache_key in self._fixture_cache: ts, cached = self._fixture_cache[cache_key] if now - ts < _FIXTURE_TTL: return cached tournaments = list(self._cat_maps.get(sport, {}).items()) all_slugs: List[dict] = [] for batch_start in range(0, len(tournaments), TOURNAMENT_BATCH): batch = tournaments[batch_start:batch_start + TOURNAMENT_BATCH] vars_decl = ', '.join(f'$t{i}: String!, $c{i}: String!' for i in range(len(batch))) vars_decl += ', $sport: String!' parts = [f'query BatchTours({vars_decl}) {{'] for i, (t_slug, c_slug) in enumerate(batch): parts.append( f' t{i}: slugTournament(tournament: $t{i}, category: $c{i}, sport: $sport) {{' f' name fixtureList(limit: 50) {{ name slug startTime status }}' f' }}' ) parts.append('}') batch_q = '\n'.join(parts) vs = {'sport': sport_slug} for i, (t_slug, c_slug) in enumerate(batch): vs[f't{i}'] = t_slug vs[f'c{i}'] = c_slug st, data = self._gql('BatchTours', batch_q, vs) if 'data' not in data: logger.debug(f'[Stake] tournament batch {batch_start} no data') continue for i, (t_slug, c_slug) in enumerate(batch): td = data['data'].get(f't{i}') or {} t_name = td.get('name', t_slug) for f in td.get('fixtureList', []): if f.get('status') != fixture_status: continue all_slugs.append({ 'slug': f['slug'], 'name': f.get('name', ''), 'startTime': f.get('startTime', ''), 'tournament': t_name, 't_slug': t_slug, 'c_slug': c_slug, }) self._fixture_cache[cache_key] = (now, all_slugs) return all_slugs def _fetch_odds_batched(self, fixtures: List[dict], sport: str) -> List[Event]: """Batch slugFixture queries and parse results. All batches fire in parallel.""" if not fixtures: return [] # Build all batch GQL bodies upfront batches = [] for batch_start in range(0, len(fixtures), FIXTURE_BATCH): batch = fixtures[batch_start:batch_start + FIXTURE_BATCH] vars_decl = ', '.join(f'$f{i}: String!' for i in range(len(batch))) parts = [f'query BatchFix({vars_decl}) {{'] for i in range(len(batch)): parts.append(f''' f{i}: slugFixture(fixture: $f{i}) {{ name startTime status tournament {{ name }} groups {{ name templates(includeEmpty: false) {{ extId name markets {{ status specifiers outcomes {{ name extId odds }} }} }} }} }}''') parts.append('}') batches.append({ 'meta': batch, 'body': { 'operationName': 'BatchFix', 'query': '\n'.join(parts), 'variables': {f'f{i}': batch[i]['slug'] for i in range(len(batch))}, }, }) # Fire all batches in parallel via Promise.all (one worker queue entry) responses = self._gql_parallel([b['body'] for b in batches]) result: List[Event] = [] for resp, item in zip(responses, batches): if not isinstance(resp, dict) or 'data' not in resp: continue for i, meta in enumerate(item['meta']): sf = resp['data'].get(f'f{i}') if not sf: continue try: result.extend(self._parse_fixture(sf, sport, meta)) except Exception as ex: logger.debug(f'[Stake] parse error for {meta["slug"]}: {ex}') return result # ── Parser ──────────────────────────────────────────────────────────────── def _parse_fixture(self, sf: dict, sport: str, meta: dict) -> List[Event]: # Extract team names from fixture name "Home Team - Away Team" name = sf.get('name', '') if ' - ' not in name: return [] home, _, away = name.partition(' - ') home = home.strip() away = away.strip() if not home or not away: return [] starts_at = _parse_time(sf.get('startTime')) league = (sf.get('tournament') or {}).get('name', '') fix_slug = meta['slug'] sport_slug = SPORT_SLUGS.get(sport, sport) c_slug = meta.get('c_slug', '') t_slug = meta.get('t_slug', '') event_url = f'{BASE_URL}/sports/{sport_slug}/{c_slug}/{t_slug}/{fix_slug}' # Collect markets from main group only markets_config = MARKETS_CONFIG.get(sport, {}) # Index markets by template extId mkt_by_ext: Dict[str, list] = {} # extId → list of market dicts for grp in (sf.get('groups') or []): grp_name = grp.get('name', '') for tmpl in (grp.get('templates') or []): ext_id = str(tmpl.get('extId', '')) if ext_id not in markets_config: continue # CS (extId=47) can appear in any group; other markets restricted to main/popular if grp_name not in ('main', 'popular') and ext_id != '47': continue for mkt in (tmpl.get('markets') or []): if mkt.get('status') != 'active': continue mkt_by_ext.setdefault(ext_id, []).append({ 'specifiers': mkt.get('specifiers', ''), 'outcomes': mkt.get('outcomes', []), }) result: List[Event] = [] event_id_base = sf.get('id') or name for ext_id, (market_name, outcome_map) in markets_config.items(): markets = mkt_by_ext.get(ext_id, []) if not markets: continue # For AH: emit a separate Event per handicap line if ext_id == '11': for mkt in markets: spec = mkt.get('specifiers', '') ah_line = _parse_ah_specifier(spec) if ah_line is None: continue outcomes = self._parse_outcomes(mkt['outcomes'], outcome_map, home, away, event_url) if len(outcomes) == 2: result.append(Event( event_id = f'stk_{event_id_base}_11_{ah_line}', bookmaker = 'Stake', sport = sport, home_team = home, away_team = away, market = f'Asian Handicap {ah_line}', outcomes = outcomes, starts_at = starts_at, league = league, )) continue # For O/U: emit a separate Event per line. # Football: hardcoded lines (1.5, 2.5, 3.5). # Basketball: all available lines extracted dynamically from specifiers # (totals are typically 150–250, not 1.5/2.5/3.5). if ext_id == '18': if sport == 'football': lines_to_check = _OU_LINES else: # Extract all distinct "total=X" lines present in this fixture's markets lines_to_check = sorted( {m.get('specifiers', '').split('total=')[-1] for m in markets if 'total=' in m.get('specifiers', '')}, key=lambda x: float(x) if x.replace('.', '', 1).isdigit() else 0, ) for lv in lines_to_check: lv_mkts = [m for m in markets if f'total={lv}' in m.get('specifiers', '')] for mkt in lv_mkts: outcomes = self._parse_outcomes(mkt['outcomes'], outcome_map, home, away, event_url) if len(outcomes) == 2: result.append(Event( event_id = f'stk_{event_id_base}_18_{lv}', bookmaker = 'Stake', sport = sport, home_team = home, away_team = away, market = f'Over/Under {lv}', outcomes = outcomes, starts_at = starts_at, league = league, )) continue if ext_id == '68' and sport == 'football': for lv in _HT_OU_LINES: lv_mkts = [m for m in markets if f'total={lv}' in m.get('specifiers', '')] for mkt in lv_mkts: outcomes = self._parse_outcomes(mkt['outcomes'], outcome_map, home, away, event_url) if len(outcomes) == 2: result.append(Event( event_id = f'stk_{event_id_base}_68_{lv}', bookmaker = 'Stake', sport = sport, home_team = home, away_team = away, market = f'HT Over/Under {lv}', outcomes = outcomes, starts_at = starts_at, league = league, )) continue if ext_id == '90' and sport == 'football': for lv in _2H_OU_LINES: lv_mkts = [m for m in markets if f'total={lv}' in m.get('specifiers', '')] for mkt in lv_mkts: outcomes = self._parse_outcomes(mkt['outcomes'], outcome_map, home, away, event_url) if len(outcomes) == 2: result.append(Event( event_id = f'stk_{event_id_base}_90_{lv}', bookmaker = 'Stake', sport = sport, home_team = home, away_team = away, market = f'2H Over/Under {lv}', outcomes = outcomes, starts_at = starts_at, league = league, )) continue # Correct Score — outcome names are the score strings directly if ext_id == '47': cs_outcomes = [] for mkt in markets: for o in mkt.get('outcomes', []): try: odds = float(o.get('odds', 0)) except (TypeError, ValueError): continue if odds <= 1.0: continue name = normalize_cs_score(o.get('name', '')) if not name: continue cs_outcomes.append(Outcome(name=name, odds=odds, bookmaker='Stake', event_url=event_url)) # Deduplicate: keep best odds per score label best_cs: Dict[str, Outcome] = {} for o in cs_outcomes: if o.name not in best_cs or o.odds > best_cs[o.name].odds: best_cs[o.name] = o if len(best_cs) >= 2: result.append(Event( event_id = f'stk_{event_id_base}_47', bookmaker = 'Stake', sport = sport, home_team = home, away_team = away, market = 'Correct Score', outcomes = list(best_cs.values()), starts_at = starts_at, league = league, )) continue for mkt in markets: outcomes = self._parse_outcomes( mkt['outcomes'], outcome_map, home, away, event_url ) if not outcomes: continue # Expect 3 outcomes for 1X2/HT 1X2/DC, 2 for everything else expected = 3 if ext_id in ('1', '5', '10') else 2 if len(outcomes) != expected: continue result.append(Event( event_id = f'stk_{event_id_base}_{ext_id}', bookmaker = 'Stake', sport = sport, home_team = home, away_team = away, market = market_name, outcomes = outcomes, starts_at = starts_at, league = league, )) return result def _parse_outcomes( self, raw_outcomes: list, outcome_map: Optional[Dict[str, str]], home: str, away: str, event_url: Optional[str], ) -> List[Outcome]: outcomes = [] for o in raw_outcomes: try: odds = float(o.get('odds', 0)) except (TypeError, ValueError): continue if odds <= 1.0: continue if outcome_map is not None: # Map by extId (for football: '1'→Home, '2'→Draw, '3'→Away, etc.) ext = str(o.get('extId', '')) label = outcome_map.get(ext) if label is None: continue else: # Tennis winner: outcome name is player name; use fixture order for label o_name = o.get('name', '').strip() if o_name == home: label = 'Home' elif o_name == away: label = 'Away' else: continue outcomes.append(Outcome(name=label, odds=odds, bookmaker='Stake', event_url=event_url)) return outcomes # ── GQL helper ──────────────────────────────────────────────────────────── def _gql(self, op: str, query: str, variables: Optional[dict] = None): body = {'operationName': op, 'query': query, 'variables': variables or {}} try: data = self._worker.call( lambda page: page.evaluate(_GQL_JS, {'url': GQL_URL, 'body': body}) ) return 200, data except Exception as ex: logger.debug(f'[Stake] GQL {op} error: {ex}') return 0, {} def _gql_parallel(self, bodies: list, timeout: float = 90) -> list: """Fire multiple GQL request bodies in parallel via Promise.all. Returns a list of response dicts (same length as bodies).""" requests = [{'url': GQL_URL, 'body': b} for b in bodies] try: return self._worker.call( lambda page: page.evaluate(_GQL_PARALLEL_JS, requests), timeout=timeout, ) except Exception as ex: logger.debug(f'[Stake] parallel GQL error: {ex}') return [{} for _ in bodies] # ── Helpers ─────────────────────────────────────────────────────────────────── def _parse_ah_specifier(specifier: str): """Extract normalised AH line from 'hcp=N' or 'handicap=N' specifier string. Returns '+0.5', '-1', '0' etc. or None for quarter-ball lines or missing specifier.""" for key in ('hcp=', 'handicap='): if key in specifier: try: raw = specifier.split(key)[1].split('&')[0].strip() val = float(raw) if abs(val * 4) % 2 != 0: # reject quarter-ball return None return f'+{val:g}' if val > 0 else f'{val:g}' except (ValueError, IndexError): return None return None def _parse_time(s: Optional[str]) -> Optional[datetime]: """Parse Stake's date string: 'Fri, 10 Apr 2026 18:00:00 GMT' → naive UTC datetime.""" if not s: return None try: return datetime.strptime(s, '%a, %d %b %Y %H:%M:%S GMT') except Exception: return None def _parse_ts(s: Optional[str]) -> Optional[float]: """Return unix timestamp from Stake date string, or None.""" dt = _parse_time(s) if dt is None: return None return dt.timestamp()