Pull US Insider-Transaction Data from LSEG in Python

A resumable pipeline for pulling insider transactions for every canonical US stock from LSEG Workspace, chunked by RIC with a retry loop and a field-name probe.

downloading US insider-trade data from LSEG in Python

LSEG Workspace carries insider-transaction data for US stocks, and pulling all of it in one pass needs a script that survives interruptions. The documentation does not include one, so I wrote it. It walks every canonical US RIC over a full window, chunks the request, retries transient failures, and flushes progress to disk so you can stop and resume without starting over.

This is the download step behind two later posts. The name field it pulls is what lets me classify each insider as routine or opportunistic, and the signed shares are what feed the insider net-buyer backtest.

The data sits behind an entitlement. The code below is complete, and running it needs an LSEG Workspace session with insider-transaction access.

What you end up with

• Insider transactions for ~10,000 canonical US RICs, active and delisted
• Five fields per trade: role, date, transaction type, signed shares, price
• History from 2000 to whatever date your price panel ends
• One CSV, pulled in a single consistent pass, resumable if it breaks

Step 1. Imports and config

Constants sit at the top of one cell. The only one you are likely to touch is START, if you ever want history earlier than the panel provides.

import os                 # filesystem existence checks
import time               # progress timing and backoff sleeps
import shutil             # safe rename of the existing raw file to .bak
import pandas as pd       # chunked CSV reads, DataFrame ops
from pathlib import Path

DATA_DIR     = Path(r'...\Betting against beta')
US_PANEL_CSV = DATA_DIR / 'US_stocks.csv'                 # master daily panel
RAW_CSV      = DATA_DIR / 'insider_transactions_raw.csv'  # the output, replaced each run
BAK_CSV      = DATA_DIR / 'insider_transactions_raw.bak'  # safety copy of the prior file
PROGRESS_CSV = DATA_DIR / 'insider_transactions_raw.pending.csv'  # resumable checkpoint

START        = '2000-01-01'   # earliest date the pull asks for
# END is auto-detected from the panel in Step 2.

CHUNK_SIZE      = 50    # RICs per get_data call
PULL_RETRIES    = 2     # retries before giving up on a chunk
BATCH_DELAY     = 0.5   # seconds between chunks, to throttle LSEG
SAVE_EVERY_N    = 10    # flush the pending file every N chunks

INSIDER_FIELDS = [
    'TR.InsiderRole',           # role string
    'TR.TransactionDate',       # trade date
    'TR.ShortTransactionType',  # Open Market Purchase / Sale / ...
    'TR.AsRepSharesTraded',     # as-reported shares, signed
    'TR.TransactionPrice',      # per-share price on the trade
]

A RIC is LSEG’s instrument code. Think of it as a globally unique ticker.

Step 2. Build the canonical US universe

The pull universe comes from one chunked pass over the master price panel. That pass does two jobs: it builds the list of RICs to pull, and it reads the panel’s last date so the insider window ends wherever the panel ends.

A canonical RIC is the primary listing of an issuer. Some tickers are cross-listings that point back to a primary: AAPL.N points back to AAPL.OQ. Pulling insider data for both would count the same filings twice, so I keep only the RIC whose primary quote is itself.

pq_first = {}          # {Instrument: first-seen Primary Quote RIC}
max_date = None        # track the panel's latest date

for ch in pd.read_csv(US_PANEL_CSV,
                      usecols=['Instrument', 'Primary Quote RIC', 'Date'],
                      parse_dates=['Date'],
                      chunksize=2_000_000, low_memory=False):
    # First non-null Primary Quote RIC per Instrument.
    m = ch.dropna(subset=['Primary Quote RIC']).drop_duplicates('Instrument')
    for inst, pq in zip(m['Instrument'], m['Primary Quote RIC']):
        pq_first.setdefault(inst, pq)
    # An Instrument seen without a Primary Quote RIC is assumed self-canonical.
    for inst in ch['Instrument'].dropna().unique():
        pq_first.setdefault(inst, inst)
    chunk_max = ch['Date'].max()
    if max_date is None or chunk_max > max_date:
        max_date = chunk_max

# Keep an Instrument only if its primary quote is itself.
canonical_rics = sorted(inst for inst, pq in pq_first.items()
                        if (not isinstance(pq, str)) or inst == pq)
END_DATE = max_date.strftime('%Y-%m-%d')

On my panel this leaves 10,078 canonical RICs and an auto-detected end date of 2026-02-27. Every issuer appears once, so no filing counts twice downstream.

Nothing downstream is specific to the US. The insider fields are the same across markets. To pull another country, point the universe at that country’s RICs, built from its price panel or your own RIC list, and Steps 3 through 6 run unchanged.

Step 3. Resume check

The pull is long, so it writes progress as it goes. Every 10 chunks it flushes what it has to insider_transactions_raw.pending.csv. If the kernel stops and you re-run the cells, this step reloads that file and pulls only the RICs it does not already hold.

if PROGRESS_CSV.exists():
    seed_df = pd.read_csv(PROGRESS_CSV, low_memory=False)
    covered = set(seed_df['Instrument'].dropna().unique())   # RICs already pulled
else:
    seed_df, covered = pd.DataFrame(), set()

missing = sorted(r for r in canonical_rics if r not in covered)

On a cold run missing is the whole universe, about 202 chunks of 50 RICs, roughly 2.5 hours end to end. After an interruption it is only the tail that did not finish.

Step 4. Probe the field names

LSEG silently drops a field name it does not recognise. There is no error, and the column does not appear in the response. So, the only reliable test of a candidate field is whether it comes back. I probe each candidate in its own get_data call against a handful of heavy-insider names, and I keep the ones that return with data.

import lseg.data as ld
ld.open_session()

PROBE_RICS = ['LCID.OQ', 'GME.N', 'TSLA.OQ', 'AMZN.OQ']   # names with dense insider activity
BASELINE   = ['TR.TransactionDate', 'TR.ShortTransactionType', 'TR.AsRepSharesTraded']

CANDIDATE_FIELDS = [
    'TR.InsiderName', 'TR.InsiderFullName', 'TR.InsiderPersonName',
    'TR.InsiderCIK', 'TR.InsiderFilerID', 'TR.InsiderPersonID',
    'TR.InsiderTitle', 'TR.InsiderJobTitle', 'TR.InsiderPosition',
]

for field in CANDIDATE_FIELDS:
    df = ld.get_data(universe=PROBE_RICS,
                     fields=BASELINE + [field],
                     parameters={'SDate': '2022-01-01', 'EDate': '2024-06-30'})
    # The field is valid only if a matching column comes back.
    short   = field.replace('TR.', '').lower()
    matched = [c for c in df.columns if c.replace(' ', '').lower() == short]
    status  = 'ACCEPTED' if matched else 'DROPPED'
    nn      = int(df[matched[0]].notna().sum()) if matched else 0
    print(f'{field:<28} {status:<10} {nn:>6,} rows')

Of the identity candidates, TR.InsiderName, TR.InsiderFullName, and TR.InsiderTitle came back with data. LSEG dropped the rest, including every CIK and person-identifier field I tried. So, my entitlement exposes the insider’s name and title. It does not expose a numeric person identifier, so any per-person classification later has to key on the name string.

Run the probe first, add whatever comes back to INSIDER_FIELDS in Step 1, then do the full pull once with the final field list.

Step 5. Pull, chunked with retries and flushing

The pull walks the missing RICs in chunks of 50. Each chunk gets a short retry loop so one transient failure does not sink the run, and every 10 successful chunks the loop flushes the running result to the pending file.

def chunked(lst, n):
    for i in range(0, len(lst), n):
        yield lst[i:i+n]

collected  = [seed_df] if len(seed_df) else []   # start from any recovered rows
since_save = []                                  # chunks held since the last flush
n_chunks   = (len(missing) + CHUNK_SIZE - 1) // CHUNK_SIZE

for i, chunk in enumerate(chunked(missing, CHUNK_SIZE)):
    df = None
    for attempt in range(PULL_RETRIES):          # retry on transient errors
        try:
            df = ld.get_data(universe=chunk,
                             fields=INSIDER_FIELDS,
                             parameters={'SDate': START, 'EDate': END_DATE})
            break
        except Exception:
            time.sleep(2 * (attempt + 1))        # simple backoff

    if df is None:
        print(f'  chunk {i+1}/{n_chunks}: FAILED, skipping')
        continue                                 # never flush a failed chunk
    if len(df) > 0:
        since_save.append(df)

    # Flush every N chunks, so an interruption loses at most N chunks of work.
    if (i + 1) % SAVE_EVERY_N == 0 and since_save:
        collected.extend(since_save)
        since_save = []
        pd.concat(collected, ignore_index=True).to_csv(PROGRESS_CSV, index=False)

    time.sleep(BATCH_DELAY)                       # throttle LSEG

if since_save:
    collected.extend(since_save)
combined = pd.concat(collected, ignore_index=True) if collected else seed_df

The loop skips a failed chunk and never flushes it, so the pending file never records a RIC as done when its rows did not arrive. That is what makes the resume in Step 3 safe: anything in the pending file is complete.

Step 6. Finalise

The last step protects the previous file before writing the new one. It renames the existing CSV to .bak, writes the fresh pull, then deletes the pending checkpoint.

# 1. Back up the previous cache, if there is one.
if RAW_CSV.exists():
    if BAK_CSV.exists():
        BAK_CSV.unlink()
    shutil.move(str(RAW_CSV), str(BAK_CSV))

# 2. Write the fresh pull as the new raw file.
combined.to_csv(RAW_CSV, index=False)

# 3. Remove the now-redundant pending file.
if PROGRESS_CSV.exists():
    PROGRESS_CSV.unlink()

If the pull dies before this step, the old file is untouched and the pending file holds the partial run. Re-running the cells picks up where it stopped.

What you need

• An LSEG Workspace session running locally, which is what ld.open_session() connects to, with insider-transaction entitlement.
• The lseg.data Python package, bundled with the Workspace license.
• A master US price panel to define the canonical universe and the end date. Swap in your own list of RICs if you already have one.

With the file saved, the routine-versus-opportunistic classification reads the name and date fields to label each insider, and the insider net-buyer backtest reads the signed shares to build the signal. Both start from this one pull.

Read next

• Following insider trades? The backtest this pull feeds: a net-insider-buying signal on US stocks.
• Classify Insiders as Informed Splitting insiders by how predictable their trades are, using the name field pulled here.

Disclaimer: this is a data-pipeline walkthrough for discussion purposes only. Not investment advice.

Data: US insider transactions from LSEG Workspace (licensed). The code requires an entitled Workspace session, and the raw pulled file is not reproduced here.