Every number the fund trades on comes through here. Learn what each piece is responsible for, how it's configured, and what actually happens when it runs.
This week is orientation, not construction. By Friday you should be able to open data-ngin and know what you're looking at — which module does what, where the config lives, and what happens end to end when the pipeline runs.
You should be able to answer:
Two things deliberately left out. Scheduling — how the pipeline runs automatically every morning — is Week 7. What happens to the data downstream is Week 6. This week is just the pipeline itself.
data-ngin is the fund's data engine. It's a Python application, and its job is to get market data from a vendor into our database in a form strategies can use.
It's worth being precise about the scope, because it's easy to under-describe. data-ngin does all three stages:
Calls the vendor's API and pulls raw bars — at the frequency, coverage, and cost we're paying for.
Normalizes the vendor's format into ours and handles missing data.
Writes to Postgres/TimescaleDB — both the raw pull and the cleaned result.
data-ngin never decides what to trade. It produces bars. What happens to them is the trade engine's problem.
It's tempting to picture a vendor as the first stage of the pipeline — as though data flows out of it and into us. A vendor is a source, not a stage. data-ngin is the thing that does the calling, so the ingestion is ours: nothing in our stack runs before data-ngin does. Keep that straight and the rest of the architecture lines up.
data-ngin is deliberately small. Four modules do the work, and one controller wires them together. Each of the four is an abstract base class with one or more concrete implementations — which is what makes the vendor swappable.
Reads a CSV from contracts/ and returns a {symbol: asset_type} dictionary. That's the whole job.
The CSV is our universe definition — if a symbol isn't in the file, the pipeline never fetches it, so it never reaches a strategy. Each row carries the metadata that matters downstream: the symbol, the exchange, the contract multiplier, the currency, the instrument type.
assetDesc,underlying,dataSymbol,ibSymbol,exchange,multiplier,currency,dataSet,instrumentType
EURUSD currency,FX,6E,6E,CME,125000,USD,GLOBEX,FUTURE
Feeder cattle,Ags,GF,GF,CME,50000,USD,GLOBEX,FUTUREImplementation: CSVLoader. There's one universe file per pipeline, and they're the first place to look when a symbol you expected isn't in the database.
Takes a symbol and a date range, calls the vendor's API, returns a pandas DataFrame. This is the only part of the pipeline that touches the outside world — and therefore the only part that needs an API key.
There's one fetcher per vendor, plus variants for bulk historical pulls. Each vendor's quirks — auth, rate limits, pagination — stay inside its own fetcher and leak no further.
Takes the vendor's DataFrame and returns rows in our schema. Vendors disagree about column names, timestamp conventions, and price units; the cleaner is where those differences stop.
It also applies the missing-data policy — which is config, not code (see below).
One cleaner per vendor, because each vendor is wrong in its own way. After this stage, nothing downstream can tell which vendor the data came from — that's the point.
Connects to Postgres and writes rows. Uses INSERT ... ON CONFLICT DO NOTHING, which is what makes the pipeline safe to re-run: a second run over the same days inserts nothing and changes nothing.
Implementation: TimescaleDBInserter. Note it writes to tables — it doesn't create them. Schema setup happens outside this repo.
The only piece that isn't swappable. It builds the other four from config, loads the symbol list, works out the date range, and runs every symbol through fetch → clean → insert.
It's also where the concurrency lives: all symbols are processed at once via asyncio.gather, not one after another.
This is the part worth understanding properly, because it explains the shape of everything else.
The modules aren't chosen in code. They're chosen in YAML. Every one of the four is named by two strings — a class and a module path:
# src/config/<pipeline>.yaml
loader:
class: "CSVLoader"
module: "loader.csv_loader"
file_path: ".../contracts/<universe>.csv"
fetcher:
class: "<Vendor>Fetcher" # ← the vendor lives here...
module: "fetcher.<vendor>_fetcher"
cleaner:
class: "<Vendor>Cleaner" # ← ...and here. Nowhere else.
module: "cleaner.<vendor>_cleaner"
inserter:
class: "TimescaleDBInserter"
module: "inserter.timescaledb_inserter"At startup, get_instance() reads those two strings, imports the module, grabs the class, and constructs it:
# utils/dynamic_loader.py (abridged)
class_name = module_config.get(class_key) # e.g. "AcmeCleaner"
module_name = f"src.modules.{module_config.get('module')}" # e.g. "src.modules.cleaner.acme_cleaner"
cls = load_class(module_name, class_name) # importlib + getattr
return cls(config=config, **kwargs)So the Orchestrator's constructor is four lines and knows nothing about any vendor:
self.loader = get_instance(self.config, "loader", "class")
self.fetcher = get_instance(self.config, "fetcher", "class")
self.cleaner = get_instance(self.config, "cleaner", "class")
self.inserter = get_instance(self.config, "inserter", "class")Switching vendors is a config edit. Point fetcher.class and cleaner.class at a different vendor's implementations, and the same Orchestrator runs an entirely different pipeline. That's why the four abstract base classes exist — not architectural decoration, but the thing that makes a YAML file able to rewire the application.
The cost: a typo in module: is not a compile error. It's an ImportError at startup. The config is real code — it just fails later than code does.
The same file carries four more blocks — the vendor parameters, the destination, the date range, and the missing-data policy:
provider: # which vendor, which asset class, and its vendor-specific settings
database: # which database, schema, raw table, and clean table to write
time_range: # what window to fetch (blank = fill in whatever's missing)
missing_data: # what to do about gapsTwo of those deserve a note now, because they'll matter to you later:
provider block decides the roll. For futures we request a continuous contract from the vendor, and settings here choose how it's stitched — on volume, on open interest, or on the calendar — and which contract in the sequence. The roll is resolved on the vendor's side, not ours; everything downstream receives an already-stitched series. (W2 covers why futures need stitching at all.) It's worth sitting with how much rests on those two lines.missing_data is a policy, not a default. Fill gaps forward, fill them with zero, interpolate, or drop the row — it's a config switch, and different pipelines choose differently depending on what a gap means for that asset class. A missing bar in a thin contract is not the same event as a missing bar in a liquid one, and the config is where that judgement gets recorded.One symbol's journey through the pipeline, in order:
The code is almost exactly that shape:
# src/orchestrator.py (abridged)
raw_data = await self.fetcher.fetch_data(symbol=..., start_date=..., end_date=...)
self.inserter.connect()
# Insert raw data — BEFORE cleaning
self.inserter.insert_data(
data=raw_data.to_dict(orient="records"),
schema=self.config["database"]["target_schema"],
table=self.config["database"]["raw_table"], # ohlcv_1d_raw
)
# Clean data
cleaned_data = self.cleaner.clean(raw_data)
# Insert cleaned data
self.inserter.insert_data(
data=cleaned_data,
schema=self.config["database"]["target_schema"],
table=self.config["database"]["table"], # ohlcv_1d
)This is the design decision in the pipeline most worth internalizing, and it costs us a whole extra table per dataset.
Vendor data costs money and can't be re-pulled for free. Our vendors bill per request. If we fetched, cleaned, and stored only the cleaned result, then a bug in the cleaner would mean the original is gone — and getting it back means paying for it again, assuming the vendor still serves that window at all.
Writing raw first buys three things:
ohlcv_1d_raw. No vendor call, no cost.Store what you received before you store what you concluded. Raw data is expensive and irreplaceable; derived data is cheap and reproducible. Never let a bug in the cheap thing destroy the expensive thing.
asyncio.gather fires every symbol at once. For a small futures universe that's fine; for a universe of several hundred names it's enough to cause connection-timeout storms, so a fetcher facing a large universe caps its own concurrency. Concurrency that's free at one scale is an outage at another — the code didn't change, the universe grew.try/except — a failure is logged and the other symbols continue. The failed symbol just gets picked up on the next run, because the date range is derived from what's already in the table.Here's the thing new engineers most often get wrong: there isn't "the" pipeline. There are several.
We buy data from more than one vendor — Databento for futures and Tiingo for equities — and each asset class gets its own configuration. Same Orchestrator, same four roles, different everything else: its own universe file, its own fetcher and cleaner, its own destination database and schema, its own missing-data policy.
You don't need to memorize which is which; those details change, and the config is the authority. What matters is the habit:
Before you debug anything — a missing symbol, a stale table, a number that looks wrong — open the config and find out which database and schema you're actually talking about. More than one pipeline writes similar-looking tables to different places. Hours get lost querying an empty table that was never the one being filled.
Some universe files are hand-maintained. Others are generated by a script — for equities, by taking a published index's constituent list and merging it with a baseline of ETFs and a few curated names.
Generating it is the better engineering: it's reproducible, and it catches ticker changes a human would miss. Either way, the file is a real input with real consequences — the universe definition decides what the fund can and cannot see, before any strategy code runs. Week 8 looks at how you test that.
Postgres, with the TimescaleDB extension — a Postgres add-on built for time-series data, which is exactly what OHLCV bars are. It gives us time-partitioned tables that stay fast as they grow.
Two things to know at this stage:
<name>_raw alongside <name>. Strategies read the clean one. The raw one is the safety net.That database is also the entire interface to the rest of the fund. data-ngin doesn't call the trade engine; it doesn't push, notify, or hand off. It writes rows. The trade engine reads them. Neither repo mentions the other anywhere — the shared table schema is the contract. Week 6 picks up on the other side of that boundary.
config.yaml is the config. More than one pipeline writes similar-looking tables to different databases. Check db_name and target_schema before you go hunting for a table that's "missing."class: and module: strings get imported. A typo there is an application that doesn't start.