A current price tells us what odds are available now. To see how that price moved over the last hour, we need to record what we observed and when.
In this tutorial, we use Python’s standard library and SQLite to sample live soccer odds. We store a row the first time we see a selection and whenever its price or activity flags change in a later sample.
PulseScore supplies the current odds through its REST API. We build and maintain the SQLite database ourselves, so the history starts with our first successful request. Changes that happen between requests can go unrecorded.
Disclosure: We build PulseScore. The API details for this example were checked on October 3, 2026.
Decide what each row represents
Each row identifies a selection within a specific bookmaker, event, and market. It records three things:
- The odds returned by the API.
- The market and selection activity flags, when available.
- The UTC time when our script received and parsed the response.
We compare the price and the activity flags together. If a selection becomes inactive while its price stays the same, we still record the change.
The observation time belongs to our collector. If the source provides a timestamp, we can store it separately and document what it means. A timestamp for when the source last checked a price, for example, doesn’t tell us when the bookmaker changed it.
Build a bounded REST sampler
Use Python 3.10 or later and set PULSESCORE_API_KEY in your environment. We’ll use the Bet365 live-events endpoint from the PulseScore REST documentation.
We keep two tables: latest holds the last observed state of each selection, and observations keeps the history of changes we detected. When a state changes, we update both tables in the same transaction.
Save this as record_odds.py:
import json
import os
import sqlite3
import time
from datetime import datetime, timezone
from urllib.error import HTTPError
from urllib.request import Request, urlopen
SOURCE = "bet365"
URL = "https://api.pulsescore.net/api/v3/bet365/live-events?sport=soccer&page=1&limit=5"
def initialize(db):
db.executescript("""
CREATE TABLE IF NOT EXISTS latest (
selection_key TEXT PRIMARY KEY,
state TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS observations (
id INTEGER PRIMARY KEY,
selection_key TEXT NOT NULL,
observed_at TEXT NOT NULL,
odds REAL,
state TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS observations_key_time
ON observations(selection_key, observed_at);
""")
def record(db, events, observed_at):
inserted = 0
with db:
for event in events:
event_id = event.get("eventId")
if event_id is None:
continue
for market in event.get("markets", []):
market_key = market.get("marketId") or [
market.get("canonicalMarket"),
market.get("period"),
market.get("line"),
market.get("rawName"),
]
for selection in market.get("selections", []):
selection_key = selection.get("selectionId") or [
selection.get("canonicalOutcome"),
selection.get("rawName"),
selection.get("line"),
]
key = json.dumps(
[SOURCE, str(event_id), market_key, selection_key]
)
odds = selection.get("odds")
state = json.dumps(
{
"odds": odds,
"market_active": market.get("isActive"),
"selection_active": selection.get("isActive"),
},
sort_keys=True,
)
previous = db.execute(
"SELECT state FROM latest WHERE selection_key = ?",
(key,),
).fetchone()
if previous and previous[0] == state:
continue
db.execute(
"INSERT INTO observations "
"(selection_key, observed_at, odds, state) "
"VALUES (?, ?, ?, ?)",
(key, observed_at, odds, state),
)
db.execute(
"INSERT INTO latest (selection_key, state) "
"VALUES (?, ?) "
"ON CONFLICT(selection_key) DO UPDATE "
"SET state = excluded.state",
(key, state),
)
inserted += 1
return inserted
def main():
api_key = os.environ["PULSESCORE_API_KEY"]
with sqlite3.connect("odds.db") as db:
initialize(db)
for attempt in range(20):
request = Request(
URL,
headers={
"X-Secret": api_key,
"Accept": "application/json",
},
)
try:
with urlopen(request, timeout=15) as response:
payload = json.load(response)
except HTTPError as error:
print(
f"Stopped on HTTP {error.code}; "
"check API access and rate limits."
)
break
observed_at = datetime.now(timezone.utc).isoformat()
count = record(db, payload.get("events", []), observed_at)
print(f"{observed_at}: stored {count} new or changed selections")
if attempt < 19:
time.sleep(60)
if __name__ == "__main__":
main()
Run the sampler:
python record_odds.py
The script creates odds.db in the current directory and makes up to 20 requests, waiting 60 seconds between samples. It uses only the Python standard library.
Query the observed history
Once the sampler has recorded data, we can pick the first selection it stored and print its ten most recent observations.
import json
import sqlite3
with sqlite3.connect("odds.db") as db:
first = db.execute(
"SELECT selection_key FROM observations ORDER BY id LIMIT 1"
).fetchone()
if first is None:
print("No observations yet.")
else:
rows = db.execute(
"SELECT observed_at, odds, state FROM observations "
"WHERE selection_key = ? ORDER BY id DESC LIMIT 10",
first,
)
for observed_at, odds, state_json in rows:
state = json.loads(state_json)
print(
observed_at,
f"odds={odds}",
f"market_active={state['market_active']}",
f"selection_active={state['selection_active']}",
)
The first stored row is our baseline: it tells us what we saw when we started observing that selection. Later rows show a change in the sampled price or activity flags. Including those flags in the query helps explain why two rows can show the same price.
We checked the insert and update logic locally with synthetic events covering unchanged prices, price movements, and activity changes. Live sampling requires an API key and available events.
Know what the sampler can tell you
The script reads only the first page of up to five live events. An event can drop off that page because the list order changed. Treat its absence as a gap in this sample until you have evidence that the event ended or the market closed.
Sampling also limits what we can say about timing. If we observe 1.90 at 14:01 and 1.85 at 14:02, we know what each response contained. We don’t know exactly when the price moved or whether it passed through another price in between.
For a larger collector, log successful requests and coverage gaps so you can distinguish a quiet market from a period with missing data. Keep the bookmaker in each selection key, and validate the fallback keys against the markets you collect before relying on them for analysis.
If your application needs changes delivered continuously, consider a Stream API collector. The same distinction still matters: when the source changed a value, when your app received it, and when you saved it can be three different times.
What would you query first from your own odds history: the first price you observed, the largest move you captured, or the last recorded price before a market went inactive?
Top comments (0)