Data Lineage / Field-Source Mapping — Mirakl / PIM / Offers

Companion to docs/fullstack/README.md and index.md, which document pipeline architecture (what system talks to what). This document tracks data lineage: per data domain, which system is authoritative, which systems merge/gap-fill it, and what breaks if a source is missing.

See also: File Lineage (every file each task reads/writes, C4-context style — the field-level ground truth this document's claims are checked against), Open Questions (things we don't know, need the team to answer), Risks (things we do know, that are structurally fragile), and Opportunities (dormant data sources found in the codebase — NeuCo, Rheem, Hydros — that are real and working, but don't currently feed supplysync.com and are intentionally kept out of this document).

Grounded entirely in code read in this session from:

airflow-pim-dags/airflow/dags/bloomreach/normalization/{availabilityNormalizer,erpJoin,xrefNormalizer,attributeReduce,stockNormalizer,regionNormalizer,branchNormalizer,filterSales}.py
airflow-pim-dags/airflow/dags/bloomreach/xmlbuilder/{mirakl,all_products}.py
airflow-pim-dags/airflow/dags/jobs-inf-w1-full.py, jobs-alg-all.py, jobs-alg-cene.py, jobs-alg-gm.py, jobs-alg-bk.py, jobs-alg-hm.py, jobs-alg-pp.py, jobs-alg-ec.py
watsco1-etl-offers/dags/pfx_offer_sync.py, dags/etl_lib/transform.py

Sources

Both source repos are on Bitbucket. Citations below are pinned to the commit read at analysis time — re-verify against main if it's been a while, since neither pin is enforced/updated automatically.

Repo	Remote	Commit pin (at analysis time)
`airflow-pim-dags`	bitbucket.org/watscoenterprise/airflow-pim-dags	`58359a03` (2026-08-11)
`watsco1-etl-offers`	bitbucket.org/watscomagento/watsco1-etl-offers	`72fa008` (2026-06-02, unchanged since analysis)

airflow-pim-dags has ~150 generated DAGs (per-BU/locale variants); only the file list above was read for this analysis. Domains/claims not backed by one of these files are marked unconfirmed, needs verification with the team rather than inferred from repo structure alone.

Anything not backed by a specific function/file citation below is marked unconfirmed, needs verification with the team.

Coarse table

Data domain	PriceFX	BI/Snowflake	BU PIM (per-BU)	BU ERP (per-BU)	ALL/W1 PIM export	Conduit	Used by
Long description	○	○	●	○	○	○	A, S
Other attributes (structure, category, flex)	○	○	◐	○	●	○	A, S
PID ↔ ERP ID matching	○	○	●	○	○	○	S, M, A
Branch inventory / stock (batch)	○	◇	○	●	○	○	A, S
Live inventory (PLP/PDP/Cart/checkout)	○	○	○	●	○	◇	S
Pricing (offers/pricing pushed to Mirakl)	●	◆	○	○	○	○	S
Product-to-sell selection	●	◆	○	○	○	○	M, S, A
Historical sales	○	●	○	？	○	○	A

Legend: ● authoritative (source of record) · ◐ gap-fill (fills holes left by the authoritative source, doesn't override it) · ◇ proxy (stands in for a different real-time source) · ◆ confirmed but out-of-band (human/business process, not implemented in either repo's code) · ？ unconfirmed, needs verification with the team · ○ not sourced here (no code evidence this system supplies this domain) · small dot after an icon = click it for detail, full citation in the numbered section below. Used by: A Algolia, S Shopware, M Mirakl.

BU PIM vs. BU ERP — these are different systems, not two names for the same thing. BU PIM is each BU's own PIM/normalization output ({bu}_product.json, e.g. GM_product.json, CENE_product.json) — this is where the BU's own PIM team performs PID↔ERP ID matching, and it's what merge_bu_erp() reads for Section 2. Confirmed real and live: a Download-BU-Feeds task log from a production ecom_w1_delta run shows all 18 per-BU files (BK, CE, CECA, CECPD, CEFL, CEMA, CEMX, CENE, CEPR, CESC, CESE, CEST, CIAC, EC, GM, HM, PP, TW) plus all_products.json being downloaded together from s3://watsco-aux/w1_json_feeds/production/ in the same sync — this also confirms W1_S3_ARCHIVE resolves to the watsco-aux bucket. (An earlier pass this session incorrectly concluded these files had no live producer, based on checking only jobs-alg-all.py/jobs-alg-*-inr.py/jobs-shw-*.py; the live log evidence overrides that — the actual producer DAG/task wasn't re-identified, but the files' existence and consumption are directly confirmed, which is what matters for this document.) BU ERP is the raw transactional/availability feed in that BU's native format (SAP CSV, Mincron JSON, TE-specific CSV, etc.) — this is what feeds Section 3's stock/availability converters (convert_ce_data, convert_mincron_json, etc.) directly, and it is not what Section 2's ID-matching join reads. A BU's raw ERP export and that same BU's PIM output can, in principle, disagree — this doc doesn't currently trace whether/how that's reconciled.

1. Long description / attributes

Long description — authoritative per-BU, same mechanism as §2's ID matching, not a separate gap-fill layer. Merge-Descriptions (all_products.merge_bu_erp(input_dir_path, colname='description', input_file_path, output_file_path, env), all_products.py:242) is the same merge_bu_erp fan-in documented in §2, called a second time with colname='description' instead of erp_id. No separate base/default description field appears elsewhere in the W1 DAG's item CSV — description for a product comes entirely from whichever BU's {bu}_product.json feed matches it (missing='' if none do). Treat this the same way §2 treats erp_id: per-BU authoritative, no fallback, silent-empty on no match.

Other attributes (not description) — base from raw CSV exports, gap-filled per-BU. Handle-Attributes (all_products.merge_attrs(merged_attr_path, env), all_products.py:288) reads the raw attribute export (env.attr_file) and renames columns per env.attr_colmap. Handle-Structures/process_category/process_flex (all_products.py:415 and neighboring functions) handle structure/category/flex attribute data similarly — each reads its own CSV export directly, no normalization-layer indirection. Merge-BU-Attrs (all_products.merge_bu_attrs(input_dir_path, item_file_path, output_file_path, env, attr_map, drop=[]), all_products.py:297) left-joins each BU's own {bu}_product.json feed (bu_data.cut('objectID', 'attributes')) onto the base item data by product_id↔objectID — missing={} on the join, so a BU with no matching product simply contributes an empty attributes dict for that field, not a dropped row.

Attribute value normalization (Algolia-facing): supply_sync_algolia_process (bloomreach.xmlbuilder.supply_sync, called from the W1 DAG) imports bloomreach.normalization.attributeReduce (aliased attribute_reduce) and uses attribute_reduce._attrs_normalized/_normalize_alt to normalize attribute values for search — this is the one place attributeReduce.py genuinely is reachable from W1, via supply_sync.py's import, not via a direct DAG-level import. Scoped to the Algolia output specifically, not the base attribute merge.

Consequence if a source is missing: for description, no match in merge_bu_erp → empty string, same as §2's erp_id case. For other attributes, merge_bu_attrs's missing={} leftjoin means a BU with no matching {bu}_product.json row for a product simply contributes an empty attributes object — no hard failure, no row dropped, just an absent per-BU attribute contribution for that product. This is the same missing-data pattern documented elsewhere in this doc (§2, §3): silent, not a raised error.

2. PID ↔ ERP ID matching

Authoritative per-BU source, one-hop join (was two-hop before the refactor). all_products.merge_bu_erp(input_dir_path, colname, input_file_path, output_file_path, env, drop=[], drop_val=[], key=None) (bloomreach/xmlbuilder/all_products.py:242) left-joins the ALL/W1 item file directly against each BU's own {bu}_product.json feed (lkey='product_id', rkey='objectID', missing=''), producing a per-BU {bu}_{colname} field. Called via python_callable=all_etl.merge_bu_erp for 7 fields (erp_id, catalog_code, description, status, title, uom, url_key) at jobs-inf-w1-full.py lines 333, 347, 361, 375, 389, 403, 420.

This replaces the older all_products.merge_erp() for the W1 ID-matching fan-in — that function still exists (all_products.py:219) but is now scoped only to jobs-alg-all.py (5 calls) and jobs-alg-hydros.py (2 calls), a different domain. Its former two-hop join via each BU's erp_pim_data_{bu}.csv xref file (written by convert_te_data/convert_hy_data/convert_mincron_json/xrefNormalizer.norm_ext_attr/xrefNormalizer.join) is no longer in the ID-matching join's path — merge_bu_erp reads each BU's {bu}_product.json directly. Those convert_* functions are otherwise unchanged (0 commits) and their xref CSVs may still feed other domains (e.g. Section 1's attribute gap-fill, still on xrefNormalizer.join() — untouched by this refactor) — don't assume they're dead code.

Mirakl seller-record generation — signature and data source changed. mirakl.mirakl_seller_process(input_path, w1_feed_path, output_dir_path, output_file_pattern, bu, env, drop_ahri=False) (mirakl.py:310) now reads erp_id directly off the BU's own per-BU feed (input_path → cut to objectID, active, attributes, catalog_code, description, erp_id, image, mpn, title), then left-joins w1_feed_path only to filter to products present in the W1 feed (_w1.cut('objectID', 'cat_id').addfield('w1', True), select('w1', lambda p: p is True)) and attach category. Call sites moved from jobs-alg-{cene,gm,bk,hm,pp,ec}.py to jobs-shw-{cene,gm,bk,hm,pp,ec}.py (plus new -inr variants), per the "move Mirakl API operators from W1 into BU DAGs" / "move Mirakl files export from Algolia to Shopware BU DAGs" commits.

mirakl_seller_wso_process() — the function this section previously cited for the "no ERP match → no seller" consequence — no longer exists. It has been removed from mirakl.py along with the diff_list/pim=True filtering it implemented. There is no direct replacement doing the same "diff against current PIM run" filter; mirakl_seller_process now does its own W1-presence check inline via the w1_feed_path join described above.

mirakl_seller_process reads erp_id straight from the per-BU feed with no row-drop on a missing value — a merge_bu_erp join with no match produces erp_id='' (per missing=''), and that empty string still flows through to the generated seller CSV rather than excluding the product from the file.

Per the team: this doesn't require Mirakl to reject anything at the API layer — the failure is structural, not a validation rejection. A missing/empty erp_id means that BU's seller-channel entity doesn't get correctly matched to the product in Mirakl, so there's no seller record for that BU to attach an offer to. PriceFX can have a valid offer for that PID, but if the seller-record side has nothing to attach it to for that BU, the offer has no destination. If none of a product's BUs match, the product has no seller record anywhere in Mirakl, and PriceFX's offer for that PID — regardless of which BU it's for — has nothing to attach to at all. This is a cross-cutting failure mode, not specific to one BU: a PID-level ID-matching gap (across all BUs) breaks the offers pipeline even though the offers pipeline itself (PriceFX → watsco1-etl-offers) has no code-level dependency on merge_bu_erp.

3. Branch inventory / stock

Corrected/extended [2026-08-20]: this section previously stopped at the per-BU converter functions and the availabilityNormalizer.merge()/merge_pid() fan-in, without tracing how that merged data actually reaches a product record's availability field in the W1 pipeline's own output. It does — via all_products.py's _availability_dict() call, consumed unchanged by supply_sync_algolia_process. That construction is added below, confirmed directly against a production tar (W1_DELTA_.processed.en.2026-08-20_02-08-58.tar.gz) where supply_sync.json (139,088 records) carries a real, non-empty availability.branches on 59,746 records, e.g. {"bu": ["Baker Distributing"], "availability": {"bu": ["Baker Distributing"], "branches": {"bk": ["418"]}, "status": 1}}.

Authoritative construction of availability — confirmed by reading bloomreach/xmlbuilder/all_products.py::process() and bloomreach/util/pim_utils.py::_availability_dict() in full.

all_products.py::process() (the function that builds all_products.json — the exact file supply_sync_algolia_process reads as its all_feed_file argument, see below) does the following, in order:

Reads availability.csv (env.getImportFile('availability.csv')) — columns include bu, erp_id, branch (the branches column is ast.literal_eval'd from a string-encoded list). This is the merged per-BU stock output — the same file availabilityNormalizer.merge()/merge_pid() (already documented above) produces — confirming the converter table above and this construction are the same lineage, not two separate mechanisms.
Reads erp_data.csv (env.getImportFile('erp_data.csv')) — per-product, per-BU ERP IDs ({bu}_erp_id columns).
For each BU code b in _schema_prefixes (= OrderedDict(env.bu_availability_files), e.g. BK, CE, CENE, GM, HM, PP, EC, …):
- Filters availability.csv to that BU's rows (_availability.select('bu', lambda x: x == b)).
- Left-joins it against erp_data.csv's {b}_erp_id column on erp_id to resolve to the product's product_id.
- Left-joins the result onto the main item dataset, renaming the joined branches column to {b}_branches (.rename('branches', '{}_branches'.format(b))).
After all BUs are joined, calls _data.addfield('availability', lambda row: _availability_dict(row, _schema_prefixes, env.bu_labels)) (all_products.py:151).

_availability_dict(row, schema_prefixes, bu_labels) (bloomreach/util/pim_utils.py:9-24) is the exact function building the shape seen in the live data:

python

def _availability_dict(row, schema_prefixes, bu_labels):
    out = {}
    bu_availability = []
    for b in schema_prefixes:
        if len(row['{}_branches'.format(b)]) > 0:
            bu_availability.append(bu_labels[b])
    out['bu'] = bu_availability
    if len(bu_availability) > 0:
        out['status'] = 1
    else:
        out['status'] = 0
    out['branches'] = {}
    for b in schema_prefixes:
        if len(row['{}_branches'.format(b)]) > 0:
            out['branches'][b.lower()] = str_to_structure(row['{}_branches'.format(b)])
    return out

The resulting availability.json-embedded field is carried into all_products.json untouched, and supply_sync_algolia_process (bloomreach/xmlbuilder/supply_sync.py:20-21) reads all_feed_file (all_products.json) via etl.fromjson, .cut(...) including availability explicitly in the retained columns, and left-joins it onto _items by objectID with missing=None (supply_sync.py:35) — supply_sync.py does not construct or modify availability/branches at all; it is a pure pass-through of whatever all_products.py already computed. The only place supply_sync.py reads row['availability'] itself is the optional sunshine flag (if hasattr(env, 'sunshine')...), which inspects row['availability']['branches'] but does not write back to it.

bu-name vs. branches-key asymmetry — confirmed, not a bug, both come from the same loop but use different lookups. env.bu_labels (defined per-environment, e.g. bloomreach/environments/W1Environment.py:156 — {'BK': 'Baker Distributing', 'CE': 'Carrier Enterprise', 'CENE': 'CE Northeast', ...}) maps uppercase BU code → human-readable BU name. Inside _availability_dict:

The bu list is built as bu_labels[b] — i.e. it looks up the name for the code and stores that.
The branches dict is keyed as b.lower() — i.e. it lowercases the code itself, never touching bu_labels.

For the sample record (bu: ["Baker Distributing"], branches: {"bk": ["418"]}): b == 'BK', bu_labels['BK'] == 'Baker Distributing' → appended to bu; 'BK'.lower() == 'bk' → used as the branches key, with str_to_structure(row['BK_branches']) (the branch ID list, e.g. ["418"]) as its value. Two different derivations of the same loop variable b, one through a name lookup table, one through simple case-folding — this is why the two fields disagree in naming convention (full name vs. lowercase code) despite describing the same BU.

Why 59,746/139,088 records have non-empty branches and the rest don't. Per the construction above, a product's {b}_branches field is non-empty only if: (a) availability.csv has a row for that BU+erp_id with a non-empty branches list (i.e. the per-BU converter — Mincron/SAP/TE/Baker/etc., see the table below — actually found stock at ≥1 branch for that item), and (b) that erp_id successfully resolves to the product's product_id via the erp_data.csv join in step 3. If a product has no ERP stock record at all for any BU (never stocked, discontinued, ERP export gap), or if its erp_id doesn't resolve for any BU (an ID-matching gap — the same class of failure documented in §2, though via a different join: this one runs through erp_data.csv, not merge_bu_erp's {bu}_product.json join), every {b}_branches column is empty, bu_availability stays [], and _availability_dict returns {"bu": [], "branches": {}, "status": 0}. This is a confirmed, structurally normal outcome, not an error state — plenty of PIM-catalogued products (long-tail, superseded, or BU-exclusive items not carried by any stocking branch) legitimately have no branch stock to report. Unconfirmed: the exact split between "genuinely unstocked" vs. "ID-matching gap" within the 79,342 empty-branches records — distinguishing those would require joining against erp_data.csv/availability.csv directly, which wasn't done this session (only the final supply_sync.json output was inspected).

Connection to §2's PID↔ERP matching — real, but via a separate join, not the same code path. Both this section's availability construction and §2's Mirakl seller-record generation depend on ERP IDs being correctly matched per-BU, but they resolve IDs through different mechanisms: §2's mirakl_seller_process reads erp_id off each BU's own {bu}_product.json (BU PIM output); this section's _availability_dict resolves erp_id→product_id via erp_data.csv (a separate, directly-joined per-BU ERP-ID export read in all_products.py::process(), _data.leftjoin(_erp, key='product_id', ...) — not the merge_bu_erp() fan-in). Both are "per-BU ID matching," and a gap in either can independently zero out that BU's contribution to a product (no seller record in §2; no branch stock in §3) — but they are not the same join, and a product could conceivably match in one and not the other. This wasn't tested directly this session — flagged as unconfirmed whether the two ID-resolution paths are kept in sync or can diverge.

Data source for availability.csv/erp_data.csv — per-BU ERP feeds, not Mirakl. supply_sync_algolia_process's other file arguments (items_file, bu_attr_file, mirakl_path, descriptions_path, titles_path, item_taxonomy_path, taxonomy_paths_path, offers_file) were all read in the function body (supply_sync.py:19-107) and traced individually:

mirakl_path feeds a filter only — _items is left-joined to _mirakl and then filtered to rows where mirakl_product_id is not None (supply_sync.py:31-33), i.e. it restricts the output to products that have a Mirakl product ID at all. It plays no role in constructing availability/branches.
offers_file (read with |-delimiter, pid/offers columns) is joined onto _items and stored as its own offers field (supply_sync.py:87-91) — a separate field from availability, not a source for it. _fulfillment_adjust() (in pim_utils.py, used by xref.py, a different output builder, not supply_sync.py) does read row['availability']['branches'] to adjust fulfillment, but that's the reverse direction (availability informing fulfillment) and it's in xref.py's pipeline, not this one.
None of items_file, bu_attr_file, descriptions_path, titles_path, item_taxonomy_path, taxonomy_paths_path reference availability/branches anywhere in supply_sync.py.

Conclusion: availability/branches in supply_sync.json is derived entirely from all_products.json (built by all_products.py::process() from availability.csv + erp_data.csv, both raw per-BU ERP exports), not from Mirakl's offers_file or mirakl_path. Mirakl's read-back data plays no role in this field — the per-BU/per-branch structure comes from the raw ERP stock feed structure itself (BU-partitioned by construction, since each BU's ERP is a separate system), not from any Mirakl seller/offer partitioning. This directly explains why w1_products.json (the Mirakl-facing output, built by a different code path not traced this session) has no availability field at all: it's a supply_sync_algolia_process/all_products.py-specific field, never carried into whatever function builds w1_products.json — unconfirmed which function that is, since it wasn't part of this session's file list.

Cross-DAG input, not a W1-computed value — confirmed the same handoff pattern as §2's {bu}_product.json. all_products.py::process() — the function that reads availability.csv/erp_data.csv and calls _availability_dict() — is called from jobs-alg-all.py (python_callable=all_etl.process at jobs-alg-all.py:834), not from any jobs-inf-w1-*.py file. jobs-alg-all.py then uploads the resulting all_products.json to a shared S3 path via an Upload-W1-style task (jobs-alg-all.py:894, 'file': 'all_products.json') — the same cross-DAG S3 handoff pattern already confirmed in §2 for {bu}_product.json. W1's Supply-Sync-Algolia task downloads and reads this file as all_feed_file without modification. This means availability/branches is not computed by the W1 pipeline itself — it's produced by jobs-alg-all.py and consumed by W1 as an external input, exactly analogous to how merge_bu_erp in §2 consumes each BU's externally-produced {bu}_product.json. The per-BU converter table and availabilityNormalizer.merge()/merge_pid() fan-in documented below are jobs-alg-all.py's own internal mechanism for building availability.csv/erp_data.csv — real, and the direct upstream source of what W1 ultimately carries into supply_sync.json, but not W1's own code.

BU code(s)	Converter function	Source format	Call site
GM (Gemaire)	`availabilityNormalizer.convert_mincron_json`	Mincron JSON export (`product_headers.json`)	`jobs-alg-all.py:291` (`Convert-GM-Availability`)
AC, HC (split out of GM)	`availabilityNormalizer.split_gm_availability`	(post-Mincron-conversion split by branch list)	`jobs-alg-all.py:303` (`Split-GM-Availability`)
CE (Carrier Enterprise, and by extension all CE* regional variants sharing the feed)	`availabilityNormalizer.convert_ce_data`	SAP-style CSV (`BRANCH_NUMBER`, `SUB_BUSINESS_UNIT_NUMBER`, `INTERNAL_PRODUCT_NUMBER`, `NET_AVAILABLE`)	`jobs-alg-all.py:353` (`Convert-CE-Availability`)
TE	`availabilityNormalizer.convert_te_data`	TE-specific CSV (`BUName`, `BranchNumber`, `InternalProductNumber`, `AvailableQty`, `ProductId`)	`jobs-alg-all.py:397` (`Convert-TE-Availability`)
BK (Baker), FC	`availabilityNormalizer.merge_baker`	Baker + FC dual-source CSV, deduplicated by `(branch, erp_id)`, `qty` conflict resolved via `max()`	referenced via `Fix-Ext-Case`/`Remove-Extra-Delimiters` steps in `jobs-alg-all.py` (lines ~ near "AVAILABILITY_BK.CSV\|AVAILABILITY_FC.CSV")
HM (Homans)	`availabilityNormalizer.split_homans`	split out of a combined feed by `sub == '6'`	`bloomreach/normalization/availabilityNormalizer.py:split_homans` (call site not directly observed this session — unconfirmed exact DAG line)
any BU using region/division splits (e.g. CE sub-regions)	`availabilityNormalizer.split_ce_reg`	splits one input file into per-BU files by branch→region mapping (`env.bu_availability_files`)	`bloomreach/normalization/availabilityNormalizer.py:split_ce_reg`
generic SAP-fed BUs	`stockNormalizer.process_sap`	generic SAP stock-status CSV filtered by `Centre` regex	`bloomreach/normalization/stockNormalizer.py:process_sap`

Fan-in / merge: availabilityNormalizer.merge()/merge_pid() cat all per-BU availability files together, select only rows with qty > 0, and aggregate branches per erp_id. merge_pid() resolves to product_id (PIM PID) via each BU's own erp_pim_data_{bu} file rather than staying at erp_id — i.e. this is where the PID↔ERP-ID matching from Section 2 becomes load-bearing for stock: a stock row with no matching product_id in that BU's xref is dropped (.select('product_id', lambda p: p not in [None, ''])).

Region roll-up: stockNormalizer.create_stocks() / regionNormalizer.create_regions() further reduce branch-level stock into region-level stock+status flags for search-facing consumption (region_bu_active, stock_status Y/N, idx_hi/idx_md search-index hints) — these are derived views, not new sources.

Consequence if missing: a BU's stock file that's empty, absent, or fails its converter step causes that BU's branches to simply not appear in the merged/region-rolled-up files (no row to join) — no explicit code path forces a hard failure at this stage; it silently reads as "no stock at that branch."

Batch usage (this lineage). The merged/region-rolled-up stock has two confirmed batch consumers: Algolia search ranking (via the BI inventory-by-branch proxy signal — see the coarse table's BI/Snowflake column and the "Key relationships" note in README.md), and — per the team — the batch BU ERP feed's stock is separately matched to BU offers to drive pickup availability and delivery decisions at checkout in Shopware.

Live path — separate row in the coarse table ("Live inventory (PLP/PDP/Cart/checkout)"), not part of this batch lineage. Real-time per-branch inventory across the full ecom experience — product listing, product detail, cart, and checkout, not checkout alone — is served by a separate, live query path through Conduit, bypassing this entire batch pipeline: both Shopware and a GraphQL proxy layer in front of it call Conduit directly for live stock, and Conduit fronts the per-BU ERPs with per-BU auth (see README.md's "Shopware also queries live inventory via Conduit" note and the context diagram). For this live path, the BU ERP is the source of truth (●) and Conduit is the proxy (◇) that fronts it — the storefront never queries the ERP directly, it always goes through Conduit. This means a product's batch stock/availability value (documented above) and what a customer sees across the site can come from two different code paths reading the same underlying ERPs — treat the batch value as directional (search ranking, general availability signal), not as the live source of truth.

4. Pricing (offers/pricing pushed to Mirakl)

Authoritative: PriceFX, via watsco1-etl-offers — confirmed in dags/pfx_offer_sync.py: the DAG factory builds pricefx_mirakl_load_{stg,prd} with task chain prepare_offers >> process_offers >> prepare_pricing >> process_pricing >> report_stats. process_phase() (imported from etl_lib.pipeline) drives PriceFX batch pull → S3 raw JSON → validate → transform.records_to_csv() (renames PriceFX JSON fields to Mirakl CSV columns per cfg.offer_field_mapping/price_field_mapping) → mkl.import_offers/import_prices (Mirakl OF01/PRI01) → poll → error report. No code path in this DAG reads from the PIM pipeline for price values — this repo's pricing values originate solely from PriceFX's batch API responses.

"PriceFX pushes a priced-items list into PIM via S3": this claim from the existing docs (docs/fullstack/README.md) is not evidenced in code in either repo searched this session — no pricefx_pim_priced_items, priced_items, or equivalent string/path appears anywhere in airflow-pim-dags or watsco1-etl-offers. Flagging as unconfirmed, needs verification with the team — it may be true but implemented outside both repos (e.g. directly in Informatica Product360, or a script not in these two codebases), or it may be stale/aspirational documentation.

5. Product-to-sell selection

Confirmed out-of-band, not implemented in either repo's code: the PriceFX team selects which products to sell, informed by BI reports. This is a human/business process, not an automated data pipeline — there's no code in airflow-pim-dags or watsco1-etl-offers implementing a "BI tells PriceFX" mechanism because none exists; the decision is made by people reading BI reports, and the output of that decision enters the code lineage only downstream (e.g. wherever PriceFX's resulting item/price selection is fed into these pipelines, per the batch pull in pfx_offer_sync.py — see §4).

Downstream consumers. Per the team: this selection flows to Mirakl, Shopware, and Algolia. In Mirakl specifically, it's broken out into seller channels, one per BU — i.e. the product-to-sell decision doesn't just gate whether a product exists in Mirakl at all, it also determines which BU-specific seller channel(s) carry it. This is consistent with, and adds business context to, the per-BU seller-record generation traced in §2 (mirakl_seller_process), though the exact code path connecting "PriceFX's selection" to "which BU seller channels get a record" was not traced in this session — flagged as unconfirmed, needs verification with the team.

6. Historical sales

Confirmed, but narrower than the general "BI supplies historical sales" claim: bloomreach/normalization/filterSales.py has three functions:

process_ce(salesdata_file_path, env) — CE-specific, filters sales rows by region_bu_code (env.salesdata_filter) and rounds sales_quantity. Called from jobs-alg-cene.py:408 (filterSales.process_ce) — i.e. this runs inside the CE Algolia DAG, not the shared W1/PIM/Mirakl pipeline.
process(salesdata_file_path, branchregion_file_path, env) — generic branch→region sales aggregation (rolls branch-level sales up to region level using a branch-region map).
process_sap(...) — explicitly commented as a placeholder, not yet active: "This code is a place-holder for the future use, once and if SALES are available as a SAP export for Latam CEMX".

No call site for filterSales.process (the generic one) was found in the DAG files read this session — only process_ce's call site was confirmed. Per the team: historical sales come from BI, delivered via a nightly feed to the far end of the ETL pipeline through S3 — i.e. BI drops a file to S3 on a nightly cadence, and that's the salesdata_file_path process_ce reads. BI is the source of truth (●) this pipeline actually reads for historical sales — unlike the branch inventory/stock domain (§3), where the pipeline's BI signal is explicitly a proxy for a separately-identified authoritative BU ERP record, this domain has no such confirmed BU ERP counterpart to point to as "the real source" — see the open question below.

This confirms the ingestion mechanism (S3-delivered, BI-originated, nightly) that the code alone couldn't show. It does not settle a separate, narrower question the team also flagged: whether BI's own feed is itself derived from CE's ERP upstream of BI — i.e. whether BI is the true point of origin for the values, or just the last system to touch them before this pipeline reads them. Marked unconfirmed (？) in the BU ERP column, not resolved either way — this pipeline doesn't read the BU ERP directly for this domain regardless of how that question is eventually answered, so it doesn't change what's actually implemented, only where the values ultimately trace back to.

The existing docs' claim that "BI also supplies historical item sales data, feeding into the same normalization step as the inventory-by-branch signal" is confirmed by the above.

Data Lineage / Field-Source Mapping — Mirakl / PIM / Offers ​

Sources ​

Coarse table ​

1. Long description / attributes ​

2. PID ↔ ERP ID matching ​

3. Branch inventory / stock ​

4. Pricing (offers/pricing pushed to Mirakl) ​

5. Product-to-sell selection ​

6. Historical sales ​