| New to the series? Start with the Series Index for setup, scope, and context. |
Vol. 1 was about writing a call that runs. Vol. 2 was about writing a call that means what you think it means. Vol. 3 is about writing a call that keeps working when the numbers get large.
A request that works perfectly for 10 instruments can time out at 500. A script that runs fine interactively can fall apart when scheduled. The patterns in this volume are what you reach for when the data grows and your workflow needs to grow with it - cleanly, reliably, and without rewriting from scratch.
Three tools. Each one independent, and more powerful in combination.
| Table of Contents |
|
Chunking
| 💡 Split large requests into reliable batches |
A single get_data() call has a server-side timeout and practical limits on the number of instruments and fields it can handle at once. Beyond those limits, requests time out or return incomplete results - often without a clear error.
The solution is chunking: split your request into smaller batches, run each one, then combine the results. Each batch is fast, predictable, and well inside the platform's limits.
To know about the limits, see The Data Library for Python - Maximum Usage Reference Guide | Devportal
There are two dimensions to chunk along - universe (instruments) and fields - and you can combine them when both are large.
Setup:
import pandas as pd
import time
# --- Helper: split any list into chunks of given size ---
def make_chunks(lst, size):
for i in range(0, len(lst), size):
yield lst[i : i + size]
DELAY = 1.0 # seconds between API calls — increase if hitting rate limits
Universe Chunking
Split a large instrument list into batches of 50, request each one, then stack rows.
instruments = ['AAPL.O', 'MSFT.O', 'GOOGL.O'] #...hundreds of RICs
fields = ['TR.Revenue', 'TR.NetIncome']
# --- Fetch: split RICs into batches of 50 ---
results = []
for i, batch in enumerate(make_chunks(instruments, 50)):
print(f"Fetching RIC batch {i+1} | {batch[0]} ... {batch[-1]}")
try:
df = ld.get_data(universe=batch, fields=fields)
results.append(df)
except Exception as e:
print(f" ⚠ RIC batch {i+1} failed: {e} — skipping")
time.sleep(DELAY)
# --- Combine: stack all RIC batches as rows ---
if results:
combined = pd.concat(results, ignore_index=True)
print(f"\nDone. Shape: {combined.shape}")
else:
print("No data retrieved — all batches failed.")
combined = pd.DataFrame()
Each call stays fast and predictable. pd.concat stitches the rows back into a single DataFrame.
Field Chunking
The same logic applies when you need more fields than a single call comfortably handles. Batch the field list instead of the instruments, then merge on the Instrument column.
instruments = ['AAPL.O', 'MSFT.O', 'GOOGL.O']
fields = ['TR.Revenue', 'TR.NetIncome',
'TR.TotalAssets', 'TR.TotalDebt', 'TR.EBITDA'] # many fields
# --- Fetch: split fields into batches of 20 ---
results = []
for i, batch in enumerate(make_chunks(fields, 20)):
print(f"Fetching field batch {i+1}")
try:
df = ld.get_data(universe=instruments, fields=batch)
results.append(df)
except Exception as e:
print(f" ⚠ Field batch {i+1} failed: {e} — skipping")
time.sleep(DELAY)
# --- Combine: merge field batches side-by-side on Instrument ---
if results:
combined = results[0]
for df in results[1:]:
combined = combined.merge(df, on="Instrument", how="outer")
print(f"\nDone. Shape: {combined.shape}")
else:
print("No data retrieved — all batches failed.")
combined = pd.DataFrame()
You get all fields in one wide DataFrame, built from narrow, reliable calls.
Combining Both
When both your universe and field list are large, nest the two loops:
instruments = ['AAPL.O', 'MSFT.O', 'GOOGL.O'] #...hundreds of RICs
fields = ['TR.Revenue', 'TR.NetIncome',
'TR.TotalAssets', 'TR.TotalDebt', 'TR.EBITDA'] # many fields
# --- Fetch: split both RICs and fields ---
results = []
for i, instrument_batch in enumerate(make_chunks(instruments, 50)):
# Inner collector — keeps field chunks grouped per RIC batch
field_results = []
for j, field_batch in enumerate(make_chunks(fields, 20)):
print(f"Fetching RIC batch {i+1}, Field batch {j+1}")
try:
df = ld.get_data(universe=instrument_batch, fields=field_batch)
field_results.append(df)
except Exception as e:
print(f" ⚠ RIC {i+1} / Field {j+1} failed: {e} — skipping")
time.sleep(DELAY)
# Merge field chunks for this RIC group (Case 2 combine)
if field_results:
merged = field_results[0]
for df in field_results[1:]:
merged = merged.merge(df, on="Instrument", how="outer")
results.append(merged)
# --- Combine: stack all RIC groups as rows (Case 1 combine) ---
if results:
combined = pd.concat(results, ignore_index=True)
print(f"\nDone. Shape: {combined.shape}")
else:
print("No data retrieved — all batches failed.")
combined = pd.DataFrame()
A note on pacing: when running many batches in a loop, add a small time.sleep(0.25) between calls. The platform allows around 5 requests per second (refer the docs to stay updated) - a brief pause keeps you well inside that without meaningfully slowing the overall run.
Chunking solves the size problem. The next section solves the time problem.
Sequential:
[── r1 ──][── r2 ──][── r3 ──] ≈ 6 seconds (considering 2 sec for each)
The processing is trivial - the waiting is the cost. Async eliminates it.
The Right Tool:
get_data_async() on Definition Objects
Before reaching for async, it's worth being precise about what kind works here.
Python's asyncio is designed for natively async operations - functions that yield control back to the event loop while they wait. ld.get_data() is not one of them. It's a blocking HTTP call. Wrapping it in async def doesn't make it concurrent; it just blocks the event loop instead of a thread.
The correct entry point is get_data_async(), available on Definition objects in the library's content layer. This is a native asyncio coroutine, built to run concurrently via asyncio.gather().
import asyncio
import lseg.data as ld
import pandas as pd
ld.open_session()
req1 = ld.content.fundamental_and_reference.Definition(
universe = ["AAPL.O", "MSFT.O"],
fields = ["TR.Revenue", "TR.NetIncome"]
)
req2 = ld.content.fundamental_and_reference.Definition(
universe = ["7203.T", "005930.KS"],
fields = ["TR.Revenue", "TR.NetIncome"]
)
req3 = ld.content.fundamental_and_reference.Definition(
universe = ["HSBA.L", "BNP.PA"],
fields = ["TR.Revenue", "TR.NetIncome"]
)
async def fetch_all():
tasks = await asyncio.gather(
req1.get_data_async(),
req2.get_data_async(),
req3.get_data_async(),
)
dfs = [t.data.df for t in tasks if t.data.df is not None]
return pd.concat(dfs, ignore_index=True)
combined = asyncio.run(fetch_all()) # use this in a .py script
All three requests go out at the same time. You wait only as long as the slowest one.
Concurrent:
[── r1 ──]
[── r2 ──] ≈ 2 seconds
[── r3 ──]
The time saving scales linearly with the number of requests.
On the content layer: ld.content.fundamental_and_reference is one of several typed templates in the content layer - others cover ESG, Bonds, Pricing, and more. It's more explicit than ld.get_data(), but that structure is what makes get_data_async() available. If you prefer to stay with ld.get_data(), there's a thread-based workaround - full details here.
Pairing Async with Chunking
If you're already splitting a large universe into chunks, async is the natural next step. Instead of processing chunks one by one, dispatch all of them at once:
universe = ["AAPL.O", "MSFT.O", "GOOGL.O", ...] # full list
fields = ["TR.Revenue", "TR.NetIncome", "TR.EPS"]
async def fetch_all():
definitions = [
ld.content.fundamental_and_reference.Definition(
universe = list(chunk),
fields = fields
)
for chunk in make_chunks(universe, 50)
]
tasks = await asyncio.gather(*[d.get_data_async() for d in definitions])
dfs = [t.data.df for t in tasks if t.data.df is not None]
return pd.concat(dfs, ignore_index=True)
combined = asyncio.run(fetch_all())
A 500-instrument universe split into 10 chunks now takes roughly the time of one chunk - not ten.
There's one thing to keep in mind before you fire all 10 simultaneously, though - which is what the next section covers.
Rate Limit Management
| 💡 Stay inside platform limits |
Concurrent doesn't mean unlimited. The platform allows around 5 requests per second. Fire 50 chunks simultaneously and you'll hit 429 Too Many Requests. The fix is asyncio.Semaphore - a valve that caps how many requests are in-flight at any given moment.
async def fetch_all():
semaphore = asyncio.Semaphore(4) # max 4 concurrent requests
async def fetch_one(definition):
async with semaphore:
try:
result = await definition.get_data_async()
return result.data.df
except Exception as e:
print(f" ⚠ Chunk failed: {e} — skipping")
return None
definitions = [
ld.content.fundamental_and_reference.Definition(
universe = list(chunk),
fields = fields
)
for chunk in make_chunks(universe, 50)
]
dfs = await asyncio.gather(*[fetch_one(d) for d in definitions])
valid = [df for df in dfs if df is not None]
return pd.concat(valid, ignore_index=True) if valid else pd.DataFrame()
combined = asyncio.run(fetch_all())
💡 Resilience: fetch_one catches exceptions and returns None instead of crashing. The valid filter then drops any failed chunks before combining. A single bad batch never takes down the whole run.
The semaphore acts as a sliding window: as soon as one request completes, the next queued request is released. You stay concurrent without exceeding the platform ceiling.
Check the current rate limits in the API documentation before tuning the semaphore value - and leave a small margin below the stated ceiling to account for other traffic.
Running in Jupyter
asyncio.run() assumes no event loop is currently running - which is true in a .py script. In Jupyter, an event loop is already active, and calling asyncio.run() raises a RuntimeError.
In a notebook cell, await directly instead:
# In a Jupyter cell - skip asyncio.run()
combined = await fetch_all()
Everything else - the Definition objects, the semaphore, the gather() call - stays exactly the same.
At a Glance
| Technique | Problem it solves | Key parameter to tune |
|---|---|---|
| Universe chunking | Request too large for one call | chunk_size (recommended: 50) |
| Field chunking | Too many fields for one call | chunk_size (recommended: 20) |
| Async (gather) | Sequential waiting adds up | Number of concurrent Definition objects |
| Semaphore | Async exceeds rate limit | asyncio.Semaphore(n) - keep n ≤ 4 |
| time.sleep | Synchronous loop hits rate limit | sleep(0.25) ≈ 4 req/s |
That's scale.
These three patterns work independently, but they're designed to be combined. Chunk to stay within limits. Run chunks async to reclaim the waiting time. Wrap with a semaphore to stay within rate limits while doing both. Each layer makes the next one safer to use - and together they take a workflow from "works on my laptop" to "runs reliably in production."
Up Next:
| ← Vol. 2 Asking What You Mean (Patterns for asking the library exactly what you meant) |
Vol. 4 Finding What to Fetch→ (Patterns for finding the data before retrieving it) |