Aggregating Records from Multiple Data Sources (Primary Source Wins)

0) Problem Restatement

DE Shaw asked (LLD, with Python code): aggregate data about the same entities (say, securities or companies) coming from multiple data sources. The sources share common fields (name, price, currency) and each also has its own extra fields. One source is designated primary: when common fields disagree, the primary wins. Build clean, extensible code and explain the design patterns.

1) Example

Source A (primary):  {id: "AAPL", name: "Apple Inc", price: 190.1, currency: "USD", sector: "Tech"}
Source B:            {id: "AAPL", name: "APPLE INC.", price: 190.3, exchange: "NASDAQ"}
Source C:            {id: "AAPL", price: 190.2, esg_score: 71}

Merged:              {id: "AAPL", name: "Apple Inc", price: 190.1, currency: "USD",
                      sector: "Tech", exchange: "NASDAQ", esg_score: 71,
                      _sources: {name: "A", price: "A", exchange: "B", esg_score: "C", ...}}

Rules: common fields come from the primary if present (otherwise from the next source by priority), and extra fields are added from whichever source has them.

2) Design

classDiagram
    class SourceAdapter {
        <<interface>>
        +name() str
        +fetch() list
        +normalize(raw) dict
    }
    class MergeStrategy {
        <<interface>>
        +merge(records_by_source) dict
    }
    class PriorityMerge
    class Aggregator {
        +sources: list
        +strategy: MergeStrategy
        +run() dict
    }
    SourceAdapter <|.. CsvSource
    SourceAdapter <|.. ApiSource
    MergeStrategy <|.. PriorityMerge
    Aggregator --> SourceAdapter
    Aggregator --> MergeStrategy

• Adapter pattern: each source turns its raw format into a common dict shape (the same key names, units and ID).
• Strategy pattern: the merge rule is pluggable (priority-based, newest-wins, average for prices...).
• Aggregator: fetches from all sources, groups records by entity ID, and applies the strategy.

3) Code (Python)

from collections import defaultdict

COMMON = ("name", "price", "currency")

class PriorityMerge:
    def __init__(self, priority):              # e.g. ["A", "B", "C"], primary first
        self.rank = {s: i for i, s in enumerate(priority)}

    def merge(self, by_source):                # {"A": {...}, "B": {...}}
        merged, lineage = {}, {}
        for src in sorted(by_source, key=lambda s: self.rank.get(s, len(self.rank))):
            for field, value in by_source[src].items():
                if value is None:
                    continue                    # missing -> let a lower-priority source fill it
                if field not in merged:        # higher-priority source already set it? keep it
                    merged[field] = value
                    lineage[field] = src
        merged["_sources"] = lineage
        return merged

class Aggregator:
    def __init__(self, sources, strategy):
        self.sources, self.strategy = sources, strategy

    def run(self):
        grouped = defaultdict(dict)            # id -> {source_name: record}
        for s in self.sources:
            for raw in s.fetch():
                rec = s.normalize(raw)
                grouped[rec["id"]][s.name()] = rec
        return {eid: self.strategy.merge(recs) for eid, recs in grouped.items()}

class ListSource:                               # simple adapter for tests
    def __init__(self, name, rows): self._n, self.rows = name, rows
    def name(self): return self._n
    def fetch(self): return self.rows
    def normalize(self, raw): return {k.lower(): v for k, v in raw.items()}

Because the strategy walks sources in priority order and only sets a field the first time it sees a non-null value, the primary wins for common fields, and missing primary values are filled by other sources. Source-only fields are simply added. _sources records the lineage of every field.

4) Follow-ups

• Different IDs across sources (ticker vs ISIN): add a mapping step (a security master) before grouping.
• Timestamps: a strategy that prefers the primary unless its value is older than X minutes.
• Validation: per-field validators (a price must be > 0), with invalid values treated as missing.
• Streaming: keep the latest record per (entity, source) in a store, and re-merge an entity when any source updates it.

5) Wrap-Up

Normalize each source through an adapter, group records by entity ID, and merge with a pluggable priority strategy that walks sources primary-first, keeping the first non-null value for each field and adding source-specific fields, while recording lineage per field. The adapter + strategy design makes new sources and new merge rules easy to add, and ID mapping, freshness rules and validation extend it naturally.