Download tick data from LSEG Tick History in Python
My LSEG Workspace post pulled daily data from Workspace. LSEG also has “Tick History”, which is a different product, with a different API, the interface that code uses to request data, and its own login. Tick History gives us every trade and every quote, timestamped to the microsecond.
LSEG’s documentation does not include a simple end-to-end script, so here is one. The script has a gap that Step 4 explains: it does not wait for an extraction that runs past LSEG’s time limit.
I use the London Stock Exchange as the example, so every RIC, LSEG’s instrument code, ends in .L. The script handles any exchange LSEG covers: change the .L filter at the end of Step 1, or pass your own list of RICs.
What you end up with
- Tick-level Time and Sales: every trade, quote, auction, and correction
- Prices, sizes, and exchange timestamps down to the microsecond
- One CSV, a plain-text table file, per stock, per 30-day window
- Any date range and any instrument your Tick History license covers
What you need
- A Tick History (DataScope Select) username and password
- For Step 1 only, if you start from ISINs: LSEG Workspace running on your computer, with the
lseg.dataPython package - Python with pandas and requests
Step 1. Turning instruments into RICs
Tick History identifies every instrument by its RIC, which is like a ticker and unique across markets.
If you already have RICs, skip the conversion, but still run import pandas as pd, because Step 4 uses pandas. Put your RICs in a list called rics, and go to Step 2. If you have ISINs (International Securities Identification Numbers), the Workspace symbol-conversion API maps each ISIN to a RIC and returns a table with a row for each match, indexed by ISIN, with the RIC in a column. The conversion service slows down or refuses large requests, so send the ISINs in batches.
import pandas as pd
import lseg.data as ld
from lseg.data.content import symbol_conversion
ld.open_session() # LSEG Workspace desktop must be running
def chunks(sequence, size):
"""Walk a list in slices, because the conversion service throttles large requests."""
for start in range(0, len(sequence), size):
yield sequence[start:start + size]
isins = [...] # your list of ISINs
ric_map = {} # ISIN -> RIC
for batch in chunks(isins, 250):
resp = symbol_conversion.Definition(
symbols=batch,
from_symbol_type=symbol_conversion.SymbolTypes.ISIN,
to_symbol_types=[symbol_conversion.SymbolTypes.RIC],
).get_data()
converted = resp.data.df
for isin in batch:
if isin not in converted.index:
continue # the service returned no RIC for this ISIN
ric = converted.loc[isin, "RIC"]
# An ISIN that maps to several RICs comes back as a Series rather than one value,
# so .loc returns every match.
if isinstance(ric, pd.Series):
ric = ric.iloc[0]
if pd.notna(ric):
ric_map[isin] = str(ric)
# Drop the delisting suffix after the caret, then keep the London names.
base_names = [ric.split("^")[0] for ric in ric_map.values()]
# A set drops the duplicates two ISINs can produce, and sorted makes the run repeatable
rics = sorted({name for name in base_names if name.endswith(".L")})
print(f"Converted {len(ric_map)} ISINs, kept {len(rics)} London RICs")One ISIN can map to more than one RIC, so I take the first. When a stock is delisted, LSEG adds ^ and a code for the month and year of delisting to its RIC, with A for January: TRIL.L^I09 is TRIL.L, delisted in September 2009. Splitting on ^ drops that suffix, so the code requests TRIL.L. I have not verified that this identifies the right historical instrument in every case.
Step 2. Authenticating against Tick History
Tick History runs on DataScope Select. DataScope Select has its own REST API and its own credentials, separate from your Workspace login. You send a username and password to the authentication endpoint and receive a token, and every later call includes that token.
Keep the credentials out of any file you share. The safest place is your shell’s environment variables. The block below sets them in Python instead, so never save or share that block with your real login in it.
import os
os.environ["RTH_USER"] = "YOUR_USERNAME" # your DataScope user id
os.environ["RTH_PASS"] = "YOUR_PASSWORD" # your DataScope passwordThe token expires, so fetch a fresh one at the start of each run.
import requests
BASE = "https://selectapi.datascope.lseg.com/RestApi/v1"
AUTH_URL = f"{BASE}/Authentication/RequestToken"
EXTRACT_URL = f"{BASE}/Extractions/ExtractRaw"
class RTHError(RuntimeError):
pass
def auth_token():
u, p = os.environ.get("RTH_USER"), os.environ.get("RTH_PASS")
if not u or not p:
raise RTHError("Set RTH_USER and RTH_PASS first.")
r = requests.post(AUTH_URL, json={"Credentials": {"Username": u, "Password": p}})
if r.status_code != 200:
raise RTHError(f"AUTH {r.status_code}: {r.text[:500]}")
token = r.json().get("value")
if not token:
raise RTHError("Authenticated, but no token came back.")
return tokenStep 3. Building the request
One request describes one instrument over one date range. We list the fields we want, name the RIC, and set the window.
I pull the full Time and Sales set: trades, quotes, auctions, and corrections. Add or remove fields as you need. Copy the names exactly as they appear in DataScope.
def payload_tas(ric, start, end, fields=None):
fields = fields or [
# Trade
"Trade - Price", "Trade - Volume", "Trade - Exchange Time",
"Trade - Qualifiers", "Trade - Exchange/Contributor ID",
"Trade - Buyer ID", "Trade - Seller ID",
"Trade - Trade Price Currency", "Trade - ISIN",
# Auction
"Auction - Price", "Auction - Volume", "Auction - Exchange Time",
"Auction - Bid Price", "Auction - Bid Size",
"Auction - Ask Price", "Auction - Ask Size", "Auction - Qualifiers",
# Quote
"Quote - Bid Price", "Quote - Bid Size",
"Quote - Ask Price", "Quote - Ask Size", "Quote - Volume",
"Quote - Exchange Time", "Quote - Qualifiers",
"Quote - Exchange/Contributor ID",
"Quote - Buyer ID", "Quote - Seller ID", "Quote - ISIN",
# Correction
"Correction - Bid Price", "Correction - Ask Price",
"Correction - Exchange Time", "Correction - Exchange/Contributor ID",
"Correction - Original Exchange Time", "Correction - ISIN",
]
return {
"ExtractionRequest": {
"@odata.type": "#DataScope.Select.Api.Extractions.ExtractionRequests.TickHistoryTimeAndSalesExtractionRequest",
"ContentFieldNames": fields,
"IdentifierList": {
"@odata.type": "#DataScope.Select.Api.Extractions.ExtractionRequests.InstrumentIdentifierList",
"InstrumentIdentifiers": [{"Identifier": ric, "IdentifierType": "Ric"}],
"ValidationOptions": {
"AllowHistoricalInstruments": True, # keep delisted names
"AllowInactiveInstruments": True,
"AllowOpenAccessInstruments": True,
},
"UseUserPreferencesForValidationOptions": False,
},
"Condition": {
"MessageTimeStampIn": "GmtUtc",
"ApplyCorrectionsAndCancellations": True,
"ReportDateRangeType": "Range",
"QueryStartDate": start,
"QueryEndDate": end,
"DisplaySourceRIC": True,
},
}
}The Condition block sets the window and asks for timestamps in UTC, with corrections and cancellations already applied to the trades. AllowHistoricalInstruments keeps delisted stocks in the extraction. Without that option, the extraction drops delisted stocks, so the sample is survivorship-biased: it keeps only the stocks that survived, the same problem as in the daily post. The option does not add stocks to your list, so your list of ISINs or RICs also has to include the stocks that traded during your dates and were delisted later.
Step 4. Downloading in windows, with retries
Tick data is large, so I cut the date range into 30-day windows for each RIC and run several at once.
When an extraction finishes, DataScope answers with a JobId, and the helpers below download the file for that JobId. The helpers also accept a list of URLs or rows in the answer, and read the file whether it arrives as a ZIP or a bare CSV.
The script sends each request and waits for the answer. If LSEG has not finished the extraction within its time limit, the answer contains no data, the loop below sends a new request, and after MAX_RETRIES attempts the window goes to failed_rics.csv. LSEG documents a way to request an extraction and check back until it finishes, which this script does not use.
import io, json, zipfile
def _read_zip_or_csv_bytes(b: bytes) -> pd.DataFrame:
if len(b) > 4 and b[:4] == b"PK\x03\x04": # a ZIP file starts with these four bytes
with zipfile.ZipFile(io.BytesIO(b)) as z:
names = [n for n in z.namelist() if n.lower().endswith(".csv")] or z.namelist()
if not names:
raise RTHError("ZIP had no files.")
with z.open(names[0]) as f:
return pd.read_csv(f)
return pd.read_csv(io.BytesIO(b)) # otherwise a plain CSV
def _collect_links(j):
urls = []
for key in ("Notes", "notes"):
for n in (j.get(key) or []):
if isinstance(n, dict) and "URL" in n:
urls.append(n["URL"])
for c in (j.get("Contents") or []):
if isinstance(c, dict) and "Location" in c:
urls.append(c["Location"])
return urls
def download_stream_by_jobid(token, job_id) -> pd.DataFrame:
headers = {"Authorization": f"Token {token}"}
url = f"{BASE}/Extractions/RawExtractionResults('{job_id}')/$value"
r = requests.get(url, headers=headers, stream=True)
if r.status_code >= 400:
raise RTHError(f"DOWNLOAD {r.status_code}: {r.text[:800]}")
return _read_zip_or_csv_bytes(r.content)
def extract_tas_df(ric, start, end, token) -> pd.DataFrame:
headers = {"Authorization": f"Token {token}", "Content-Type": "application/json"}
resp = requests.post(EXTRACT_URL, headers=headers, data=json.dumps(payload_tas(ric, start, end)))
if resp.status_code >= 400:
raise RTHError(f"EXTRACT {resp.status_code}: {resp.text[:1200]}")
j = resp.json()
links = _collect_links(j)
if links: # case 1: URLs to fetch
frames = []
for u in links:
r = requests.get(u, headers=headers, stream=True)
if r.status_code >= 400:
raise RTHError(f"DOWNLOAD {r.status_code}: {r.text[:800]}")
frames.append(_read_zip_or_csv_bytes(r.content))
return pd.concat(frames, ignore_index=True) if frames else pd.DataFrame()
job_id = j.get("JobId") or j.get("jobId")
if job_id: # case 2: a JobId to stream
return download_stream_by_jobid(token, job_id)
rows = j.get("Data") or j.get("Rows") # case 3: rows inline
if rows is None:
raise RTHError(f"No links, no JobId, no rows. Body: {str(j)[:1000]}")
return pd.DataFrame(rows)The loop writes one CSV per RIC per window, which the code calls a chunk, and skips any window whose file already exists, so a stopped run restarts where it left off. It catches any error and waits longer after each attempt. A completely empty file raises EmptyDataError, and the loop skips the window without saving, so a restart asks for it again. An empty table that still has column names is saved like any other window.
from pathlib import Path
from datetime import datetime, timedelta
from concurrent.futures import ThreadPoolExecutor, as_completed
from pandas.errors import EmptyDataError
import csv, time
RICS = rics # from Step 1, or your own list
START = "2025-05-01T07:00:00.000Z" # UTC, inclusive
END = "2025-05-31T23:59:59.000Z"
BASE_DIR = r"C:\path\to\tick_data" # output folder
CHUNK_DAYS = 30 # window width per request
MAX_WORKERS = 3 # parallel downloads
MAX_RETRIES = 10 # per-chunk retries
def daterange_chunks(start_date, end_date, chunk_days):
"""Split a date range into windows of chunk_days, the last one possibly shorter.
A whole date range in one request can be more than the service will return in one go,
so each window is asked for separately. CHUNK_DAYS sets the width.
"""
current = start_date
while current < end_date:
window_end = current + timedelta(days=chunk_days)
yield current, min(window_end, end_date) # min: never run past the end asked for
current = window_end
def process_chunk(ric, chunk_start, chunk_end, token):
"""Download one stock over one window, unless its file is already on disk."""
start_text = chunk_start.strftime("%Y-%m-%dT%H:%M:%S.000Z")
end_text = chunk_end.strftime("%Y-%m-%dT%H:%M:%S.000Z")
label = f"{chunk_start.date()}_{chunk_end.date()}"
out_csv = Path(BASE_DIR) / f"{ric}_{label}.csv"
if out_csv.exists():
return {
"ok": True,
"msg": f"skip {ric} [{label}]: already downloaded",
}
for attempt in range(1, MAX_RETRIES + 1):
try:
df = extract_tas_df(ric, start_text, end_text, token)
# Write to a .part file and rename it once the write finished. A run killed
# mid-write then leaves no half-file that the resume check would trust.
tmp_csv = out_csv.with_suffix(".csv.part")
df.to_csv(tmp_csv, index=False)
tmp_csv.replace(out_csv)
return {
"ok": True,
"msg": f"OK {ric} [{label}]: {len(df):,} rows -> {out_csv}",
}
except EmptyDataError:
return {
"ok": True,
"msg": f"empty {ric} [{label}]: no data, skipping",
}
except Exception as ex:
err = f"{type(ex).__name__}: {ex}"
if attempt < MAX_RETRIES:
time.sleep(2 * attempt)
print(
f"... retry {attempt}/{MAX_RETRIES} "
f"for {ric} [{label}]: {err}"
)
continue
return {
"ok": False,
"ric": ric,
"chunk": label,
"error": err,
"msg": f"FAIL {ric} [{label}]: {err}",
}
def main():
Path(BASE_DIR).mkdir(parents=True, exist_ok=True)
token = auth_token()
start_dt = datetime.strptime(START, "%Y-%m-%dT%H:%M:%S.000Z")
end_dt = datetime.strptime(END, "%Y-%m-%dT%H:%M:%S.000Z")
tasks = [(ric, cs, ce, token)
for ric in RICS
for cs, ce in daterange_chunks(start_dt, end_dt, CHUNK_DAYS)]
failed = []
with ThreadPoolExecutor(max_workers=MAX_WORKERS) as ex:
for fut in as_completed(ex.submit(process_chunk, *t) for t in tasks):
res = fut.result()
print(res["msg"])
if not res["ok"]:
failed.append(res)
if failed:
path = Path(BASE_DIR) / "failed_rics.csv"
with path.open("w", newline="", encoding="utf-8") as f:
w = csv.writer(f)
w.writerow(["RIC", "Chunk", "Error"])
for row in failed:
w.writerow([row["ric"], row["chunk"], row["error"]])
print(f"\nSaved {len(failed)} failed chunks to {path}")
else:
print("\nNo failed chunks.")
if __name__ == "__main__":
main()The folder now holds one CSV file per RIC per window, with the Time and Sales fields you asked for, ready to read back into pandas.
Reading those files back is Lesson 16 in Practical Coding, and Lesson 15 covers the DataFrame, pandas’ table, that read_csv returns.
One warning on scale. In my own runs, a single month of one active London stock produced millions of rows, and a month of all London stocks can take hours to download. The script fetches its token once, and LSEG tokens last 24 hours, so in a run longer than 24 hours, every window after the token expires fails.
If daily closes would answer your question instead, How do I get market data into Python? sets out the cheaper routes and when each is enough.
Conclusion
The takeaway is that tick data has to be requested in small pieces: one stock and 30 days at a time, each saved as it arrives, so a stopped run restarts where it left off, and a run longer than 24 hours outlives its token.