import os from dotenv import load_dotenv load_dotenv() import requests import pandas as pd from datetime import datetime, timedelta, timezone # ========================================================= # CONFIG # ========================================================= API_KEY = os.environ.get("ODDS_API_KEY") BASE_URL = "https://api.the-odds-api.com/v4" REGIONS = "uk" MARKETS = "h2h" ODDS_FORMAT = "decimal" DATE_FORMAT = "iso" FREE_BET_STAKE = 1.0 HOURS_AHEAD = 168 TOP_PRINT = 20 SHORTLIST_SIZE = 10 OUTPUT_DIR = r"C:\Users .... INSERT YOUR OUTPUT DIRECTORY HERE" OUTPUT_FILE = os.path.join( OUTPUT_DIR, f"uk_top_freebet_hedges_fresh_next_{HOURS_AHEAD}h.csv" ) # Fixed sports to check when currently offered by The Odds API. # Tennis is added dynamically below because the API uses separate # competition keys such as tennis_atp_* and tennis_wta_*. SPORT_KEYS = [ "baseball_mlb", "baseball_ncaa", "basketball_nba", "basketball_wnba", "basketball_ncaab", "basketball_wncaab", "basketball_euroleague", "icehockey_nhl", "mma_mixed_martial_arts", "cricket_ipl", "aussierules_afl", ] # Exchanges / marketplaces removed ALLOWED_BOOKMAKERS = { "sport888", "betfair_sb_uk", "betvictor", "betway", "boylesports", "casumo", "coral", "grosvenor", "ladbrokes_uk", "leovegas", "livescorebet", "paddypower", "skybet", "unibet_uk", "virginbet", "williamhill", } # ========================================================= # HELPERS # ========================================================= def iso_z(dt: datetime) -> str: return dt.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") def parse_time(ts: str | None) -> datetime | None: if not ts: return None return datetime.fromisoformat(ts.replace("Z", "+00:00")) def update_ago_text(last_update: str | None, now_utc: datetime) -> str: dt = parse_time(last_update) if dt is None: return "" seconds = max(0, int((now_utc - dt).total_seconds())) if seconds < 60: return f"{seconds} seconds ago" minutes = seconds // 60 if minutes < 60: return f"{minutes} minutes ago" hours = minutes // 60 return f"{hours} hours ago" def get_available_sports(api_key: str) -> list[dict]: url = f"{BASE_URL}/sports/" params = {"apiKey": api_key, "all": "true"} r = requests.get(url, params=params, timeout=30) r.raise_for_status() return r.json() def fetch_odds_for_sport( api_key: str, sport_key: str, commence_from: datetime, commence_to: datetime, ) -> tuple[list[dict], dict]: url = f"{BASE_URL}/sports/{sport_key}/odds/" params = { "apiKey": api_key, "regions": REGIONS, "markets": MARKETS, "oddsFormat": ODDS_FORMAT, "dateFormat": DATE_FORMAT, "commenceTimeFrom": iso_z(commence_from), "commenceTimeTo": iso_z(commence_to), } r = requests.get(url, params=params, timeout=30) r.raise_for_status() headers = { "x-requests-last": r.headers.get("x-requests-last"), "x-requests-used": r.headers.get("x-requests-used"), "x-requests-remaining": r.headers.get("x-requests-remaining"), } return r.json(), headers def fetch_fresh_event_odds(api_key: str, sport_key: str, event_id: str) -> tuple[dict, dict]: url = f"{BASE_URL}/sports/{sport_key}/events/{event_id}/odds" params = { "apiKey": api_key, "regions": REGIONS, "markets": MARKETS, "oddsFormat": ODDS_FORMAT, "dateFormat": DATE_FORMAT, } r = requests.get(url, params=params, timeout=30) r.raise_for_status() headers = { "x-requests-last": r.headers.get("x-requests-last"), "x-requests-used": r.headers.get("x-requests-used"), "x-requests-remaining": r.headers.get("x-requests-remaining"), } return r.json(), headers def dedupe_events(events: list[dict]) -> list[dict]: seen = set() out = [] for ev in events: ev_id = ev.get("id") if ev_id and ev_id not in seen: out.append(ev) seen.add(ev_id) return out def build_sport_keys_to_query(available_keys: set[str]) -> list[str]: fixed_keys = [k for k in SPORT_KEYS if k in available_keys] tennis_keys = sorted( k for k in available_keys if k.startswith("tennis_") ) return fixed_keys + tennis_keys # ========================================================= # PRICE EXTRACTION # ========================================================= def extract_allowed_prices(event: dict) -> dict: """ Return all allowed bookmaker prices for each outcome, but only from true 2-outcome h2h markets. Any bookmaker quote with 3 outcomes (for example, one including Draw) is ignored. The last-update timestamp is retained so the final recommendations show how fresh the quoted prices are. """ prices = {} for bookmaker in event.get("bookmakers", []): key = bookmaker.get("key") if key not in ALLOWED_BOOKMAKERS: continue title = bookmaker.get("title", key) for market in bookmaker.get("markets", []): if market.get("key") != "h2h": continue outcomes = market.get("outcomes", []) # Ignore 3-way markets such as regulation-time markets with Draw. if len(outcomes) != 2: continue market_last_update = market.get("last_update") or bookmaker.get("last_update") for outcome in outcomes: name = outcome.get("name") price = outcome.get("price") if name is None or price is None: continue try: price = float(price) except (TypeError, ValueError): continue if price <= 1.0: continue prices.setdefault(name, []).append({ "price": price, "bookmaker": title, "bookmaker_key": key, "last_update": market_last_update, }) return prices def get_best_price(price_list: list[dict]) -> dict | None: if not price_list: return None return max(price_list, key=lambda x: x["price"]) def get_best_other_book_price(price_list: list[dict], excluded_bookmaker_key: str) -> dict | None: candidates = [ row for row in price_list if row["bookmaker_key"] != excluded_bookmaker_key ] if not candidates: return None return max(candidates, key=lambda x: x["price"]) # ========================================================= # FREE BET HEDGE # ========================================================= def free_bet_hedge( free_bet_odds: float, hedge_odds: float, stake: float = 1.0, ) -> tuple[float, float]: """Stake-not-returned (SNR) free-bet hedge.""" hedge_stake = stake * (free_bet_odds - 1.0) / hedge_odds guaranteed_profit = stake * (free_bet_odds - 1.0) - hedge_stake return hedge_stake, guaranteed_profit # ========================================================= # BUILD OPPORTUNITIES # ========================================================= def rows_from_event(event: dict, as_of_utc: datetime) -> list[dict]: rows = [] prices = extract_allowed_prices(event) # Need exactly two outcomes after filtering to allowed books # and dropping 3-way markets. if len(prices) != 2: return rows outcome_names = list(prices.keys()) a_name, b_name = outcome_names[0], outcome_names[1] a_prices = prices[a_name] b_prices = prices[b_name] if not a_prices or not b_prices: return rows commence = parse_time(event.get("commence_time")) # Case 1: free bet on A, hedge on B at a DIFFERENT bookmaker. best_free_a = get_best_price(a_prices) if best_free_a is not None: best_hedge_b = get_best_other_book_price( b_prices, excluded_bookmaker_key=best_free_a["bookmaker_key"], ) if best_hedge_b is not None: hedge_stake, guaranteed_profit = free_bet_hedge( free_bet_odds=best_free_a["price"], hedge_odds=best_hedge_b["price"], stake=FREE_BET_STAKE, ) rows.append({ "sport_key": event.get("sport_key"), "sport": event.get("sport_title"), "event_id": event.get("id"), "match": f"{event.get('home_team')} vs {event.get('away_team')}", "time": commence.isoformat() if commence else event.get("commence_time"), "free_bet_on": a_name, "free_bet_odds": best_free_a["price"], "free_bet_bookmaker": best_free_a["bookmaker"], "free_bet_last_update": best_free_a["last_update"], "free_bet_update_ago": update_ago_text(best_free_a["last_update"], as_of_utc), "hedge_on": b_name, "hedge_odds": best_hedge_b["price"], "hedge_bookmaker": best_hedge_b["bookmaker"], "hedge_last_update": best_hedge_b["last_update"], "hedge_update_ago": update_ago_text(best_hedge_b["last_update"], as_of_utc), "free_bet_stake": FREE_BET_STAKE, "hedge_stake": hedge_stake, "guaranteed_profit": guaranteed_profit, "conversion_rate": guaranteed_profit / FREE_BET_STAKE, "recommendation_refreshed_at_utc": as_of_utc.isoformat(), }) # Case 2: free bet on B, hedge on A at a DIFFERENT bookmaker. best_free_b = get_best_price(b_prices) if best_free_b is not None: best_hedge_a = get_best_other_book_price( a_prices, excluded_bookmaker_key=best_free_b["bookmaker_key"], ) if best_hedge_a is not None: hedge_stake, guaranteed_profit = free_bet_hedge( free_bet_odds=best_free_b["price"], hedge_odds=best_hedge_a["price"], stake=FREE_BET_STAKE, ) rows.append({ "sport_key": event.get("sport_key"), "sport": event.get("sport_title"), "event_id": event.get("id"), "match": f"{event.get('home_team')} vs {event.get('away_team')}", "time": commence.isoformat() if commence else event.get("commence_time"), "free_bet_on": b_name, "free_bet_odds": best_free_b["price"], "free_bet_bookmaker": best_free_b["bookmaker"], "free_bet_last_update": best_free_b["last_update"], "free_bet_update_ago": update_ago_text(best_free_b["last_update"], as_of_utc), "hedge_on": a_name, "hedge_odds": best_hedge_a["price"], "hedge_bookmaker": best_hedge_a["bookmaker"], "hedge_last_update": best_hedge_a["last_update"], "hedge_update_ago": update_ago_text(best_hedge_a["last_update"], as_of_utc), "free_bet_stake": FREE_BET_STAKE, "hedge_stake": hedge_stake, "guaranteed_profit": guaranteed_profit, "conversion_rate": guaranteed_profit / FREE_BET_STAKE, "recommendation_refreshed_at_utc": as_of_utc.isoformat(), }) return rows def build_opportunity_rows(events: list[dict], as_of_utc: datetime) -> list[dict]: rows = [] for event in events: rows.extend(rows_from_event(event, as_of_utc)) return rows # ========================================================= # MAIN # ========================================================= def main() -> None: if not API_KEY: raise ValueError("ODDS_API_KEY not found in environment.") now_utc = datetime.now(timezone.utc) end_utc = now_utc + timedelta(hours=HOURS_AHEAD) available_sports = get_available_sports(API_KEY) available_keys = {s["key"] for s in available_sports} sport_keys_to_query = build_sport_keys_to_query(available_keys) if not sport_keys_to_query: print("None of the selected sport keys are available from the API.") return print("Using UK bookmakers:") print(", ".join(sorted(ALLOWED_BOOKMAKERS))) print() print("Scanning these sports:\n") for sport_key in sport_keys_to_query: print(f" - {sport_key}") print() all_events = [] coverage_rows = [] total_credits_used_this_run = 0 total_events_returned = 0 last_seen_used_total = None last_seen_remaining = None for sport_key in sport_keys_to_query: try: events, header_info = fetch_odds_for_sport( api_key=API_KEY, sport_key=sport_key, commence_from=now_utc, commence_to=end_utc, ) except requests.HTTPError as e: print(f"Skipping {sport_key} due to HTTP error: {e}") continue except requests.RequestException as e: print(f"Skipping {sport_key} due to request error: {e}") continue credits_last = int(header_info.get("x-requests-last", "0") or 0) total_credits_used_this_run += credits_last last_seen_used_total = header_info.get("x-requests-used") last_seen_remaining = header_info.get("x-requests-remaining") all_events.extend(events) total_events_returned += len(events) coverage_rows.append({ "sport_key": sport_key, "events_returned": len(events), }) print( f"{sport_key}: returned {len(events)} events, " f"credits this call = {credits_last}" ) if coverage_rows: coverage_df = pd.DataFrame(coverage_rows).sort_values( by=["events_returned", "sport_key"], ascending=[False, True], ) print("\nCoverage summary:\n") print(coverage_df.to_string(index=False)) print() if not all_events: print(f"No events found in the next {HOURS_AHEAD} hours.") return all_events = dedupe_events(all_events) total_unique_events = len(all_events) initial_as_of = datetime.now(timezone.utc) initial_rows = build_opportunity_rows(all_events, initial_as_of) if not initial_rows: print("Events were found, but none produced usable two-bookmaker 2-outcome hedges.") return initial_df = pd.DataFrame(initial_rows).sort_values( by=["guaranteed_profit", "free_bet_odds"], ascending=[False, False], ).reset_index(drop=True) shortlist_events = ( initial_df[["sport_key", "event_id"]] .drop_duplicates() .head(SHORTLIST_SIZE) .to_dict("records") ) print() print(f"Rechecking the top {len(shortlist_events)} candidate events using single-event odds...") print() fresh_events = [] for item in shortlist_events: try: fresh_event, header_info = fetch_fresh_event_odds( api_key=API_KEY, sport_key=item["sport_key"], event_id=item["event_id"], ) except requests.HTTPError as e: print(f"Skipping refreshed event {item['event_id']} due to HTTP error: {e}") continue except requests.RequestException as e: print(f"Skipping refreshed event {item['event_id']} due to request error: {e}") continue credits_last = int(header_info.get("x-requests-last", "0") or 0) total_credits_used_this_run += credits_last last_seen_used_total = header_info.get("x-requests-used") last_seen_remaining = header_info.get("x-requests-remaining") fresh_events.append(fresh_event) fresh_as_of = datetime.now(timezone.utc) fresh_rows = build_opportunity_rows(fresh_events, fresh_as_of) if not fresh_rows: print("The refreshed candidate events produced no usable two-bookmaker hedges.") return df = pd.DataFrame(fresh_rows).sort_values( by=["guaranteed_profit", "free_bet_odds"], ascending=[False, False], ).reset_index(drop=True) top = df.head(TOP_PRINT).copy() pd.set_option("display.max_columns", None) pd.set_option("display.width", 280) pd.set_option("display.max_colwidth", 60) print(f"Top {TOP_PRINT} freshly rechecked guaranteed returns from unit SNR free bets:\n") print( top[ [ "sport", "match", "time", "free_bet_on", "free_bet_odds", "free_bet_bookmaker", "hedge_on", "hedge_odds", "hedge_bookmaker", "free_bet_stake", "hedge_stake", "guaranteed_profit", "conversion_rate", ] ].to_string(index=False) ) print("\n----------------------------------------") print(f"Total credits used in this run: {total_credits_used_this_run}") if last_seen_used_total is not None: print(f"x-requests-used after last call: {last_seen_used_total}") if last_seen_remaining is not None: print(f"x-requests-remaining after last call: {last_seen_remaining}") print("----------------------------------------") os.makedirs(OUTPUT_DIR, exist_ok=True) df.to_csv(OUTPUT_FILE, index=False, encoding="utf-8-sig") print(f"\nSaved full refreshed results to: {OUTPUT_FILE}") print("\n===================================") print("FINAL SUMMARY") print("===================================") print(f"Markets checked: {', '.join(sport_keys_to_query)}") print(f"Events returned by API: {total_events_returned}") print(f"Unique events after dedupe: {total_unique_events}") print(f"Fresh candidate events checked: {len(shortlist_events)}") print(f"Fresh opportunities generated: {len(df)}") print(f"API credits used: {total_credits_used_this_run}") print("===================================") if __name__ == "__main__": main()