Download Tick Data from LSEG Tick History in Python

Download trade and quote data from LSEG Tick History using Python, with chunked requests, automatic retries and resumable CSV exports.

Illustration of bid and ask depth in a limit order book

My LSEG post pulled daily data from Workspace. LSEG also has “Tick History”, which is a different product, with a different API and its own login. It gives you every trade and every quote, timestamped to the microsecond.

There is documentation, but not a simple end-to-end script you can copy, paste, and run. So here is one.

I use the London Stock Exchange as the example, so every RIC ends in .L. Swap the RICs and it works for any exchange LSEG covers.

The pipeline is four steps: turn your instruments into RICs, trade your login for a token, build the request, then download in date-chunked batches with retries.

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 per stock, per monthly chunk
• Any date range and any instrument your Tick History license covers

Step 1. Turn instruments into RICs

Tick History wants RICs. RIC is LSEG’s instrument code. Think ticker, but globally unique.

If you already have RICs, skip to Step 2. If you have ISINs, the Workspace symbol-conversion API maps them across. It throttles 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(seq, n):
    for i in range(0, len(seq), n):
        yield seq[i : i + n]

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()
    conv = resp.data.df
    for isin in batch:
        if isin not in conv.index:
            continue
        ric = conv.loc[isin, "RIC"]
        if isinstance(ric, pd.Series):      # one ISIN, several RICs: take the first
            ric = ric.iloc[0]
        if pd.notna(ric):
            ric_map[isin] = str(ric)

# Strip any chain suffix after "^", then keep the London names.
bases = (r.split("^")[0] for r in ric_map.values())
rics  = sorted({b for b in bases if b.endswith(".L")})
print(f"Converted {len(ric_map)} ISINs, kept {len(rics)} London RICs")

One ISIN can map to more than one RIC, so we take the first. Splitting on ^ drops the chain code and leaves the base name, so VOD.L^A21 becomes VOD.L, which is the name Tick History expects.

Step 2. Authenticate against Tick History

Tick History runs on DataScope Select. It has its own REST API and its own credentials, separate from your Workspace login. You trade a username and password for a token, and every later call carries that token.

Keep the credentials out of the notebook. Set them in your shell, or fill the two lines below and do not commit them anywhere.

import os

os.environ["RTH_USER"] = "YOUR_USERNAME"   # your DataScope user id
os.environ["RTH_PASS"] = "YOUR_PASSWORD"   # your DataScope password

Now the token. It 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 token

Step 3. Build the request

One request describes one instrument over one date range. You list the fields you want, point it at a RIC, and set the window.

I pull the full Time and Sales set: trades, quotes, auctions, and corrections. Add or remove to taste. The names are exactly as they appear in DataScope, so a typo returns an empty extraction rather than an error.

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,
            },
        }
    }

AllowHistoricalInstruments is what keeps delisted names in. Leave it out and you get a survivorship-biased sample, same trap as the daily post.

Step 4. Download, chunk, and retry

Now the work. Tick data is large, so we cut each RIC into monthly chunks and run several at once.

DataScope answers in one of three ways. A small extraction comes straight back as a file. A larger one comes back as a set of URLs, or as a JobId you stream separately. The helpers below handle all three, and unzip the result whether it arrives as a ZIP or a bare CSV.

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":          # ZIP magic number
        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)

Now the loop. The knobs are at the top. One CSV per RIC per chunk lands in BASE_DIR, so a run that dies halfway can pick up where it left off. Transient errors retry with a growing back-off. An empty result is not an error: a stock with no ticks in a window just gets skipped. Everything that truly fails is written to failed_rics.csv.

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):
    current = start_date
    while current < end_date:
        nxt = current + timedelta(days=chunk_days)
        yield current, min(nxt, end_date)
        current = nxt

def process_chunk(ric, chunk_start, chunk_end, token):
    s = chunk_start.strftime("%Y-%m-%dT%H:%M:%S.000Z")
    e = 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, s, e, token)
            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()

That is it. Every trade and quote for your London names, on disk as CSV, ready for whatever comes next.

One warning on scale. A single month of active London stocks is millions of rows per name and can run for hours. The one-file-per-chunk layout and the retry loop are what make that survivable: kill the run, restart it, and it fills in the gaps rather than starting over.