Debugging in a vendor payout senario - not many cases, but hard 😬
ranked by score ↓Source
Paste as source: in your trap.yaml
git+https://github.com/trapstreet/trapstreet-tasks@e4084a9c3b892ccd855ca15b6ed4e4cc5473a7cf#subdirectory=tasks/debug_vendor_payout_pipelineShare
debug-vendor-payout-pipeline
An open-source evaluation task for cross-file consistency debugging — when a ticket asks for a change to a data pipeline, does the agent identify ALL the places that need updating so that TWO reports (with DIFFERENT lookup paths) both come out correct?
4 cases
Each case feeds files from inputs/<id>/ to the solution, expects files in expected/<id>/, and is scored by judge.py then aggregated by grader.py.
cases (4)
▸case_01Re-book misattributed sales to correct vendor (all-time); must update lookup path + baked fields + baked payout
input
README.md
# Task
A colleague filed a ticket asking for a fix to the vendor-payout pipeline.
**Your goal:** ensure both reports (`vendor_statement.py` and `itemised_statement.py`) come out CORRECT after the fix is applied.
You will not be graded on your intermediate reasoning or on the structure of your edits. You will be graded on **whether the reports the two scripts produce match the reports they would produce after a correct fix**. Judge applies your edits + runs both scripts + compares stdout to the gold reports.
## Business context
This supermarket runs a consignment model: vendors stock the shelves, and every period the store owes each vendor a share of what sold. These two reports are the source of truth for that monthly payout run:
- `vendor_statement` — the per-vendor summary Finance uses to cut monthly payout checks. The `total_payout_usd` column IS the amount each vendor gets paid.
- `itemised_statement` — the per-transaction breakdown vendors audit against their sales.
Getting `vendor_payout_usd` right on every affected line is the whole point. An inconsistency here is a real over/under-payment to a vendor. So when a ticket changes something upstream — attribution, pricing, terms — think through: does the change move payout money around, or change how much payout is owed? Both flow through these reports and both must land right.
## How to approach this
In a real production system:
- The data tables are LARGE — you cannot solve this by scanning every row.
- No one documents the business rules explicitly — you have to reverse-engineer them by reading the pipeline code (`vendor_statement.py` and `itemised_statement.py`) and understanding what each script does with the data.
To decide what edits are needed, read the scripts carefully:
- Where does each report get its values from? Which table? Which column? Lookup or baked?
- If you change X in the source data, which report changes and how?
- Multiple valid fix approaches may exist — pick the one that produces reports consistent with the business intent of the ticket.
## Files in this directory
- `ticket.md` — the change request
- `catalog.csv`, `suppliers.csv` — product/vendor master
- `transactions.csv` — retail (checkout) sales
- `b2b_details.csv` — wholesale/bulk orders (high volume: many units per order, e.g. restaurant supply)
- `promotions.csv`, `product_discounts.csv`, `currency_rates.csv` — auxiliary tables
- `vendor_statement.py`, `itemised_statement.py` — the two report scripts (Python + SQL)
## Output format
Emit a JSON array of edits to stdout. Nothing else — no explanation, no markdown fences.
Each edit is one of:
```
{"file": "<name>.csv", "op": "update", "match": {"<col>": "<value>"}, "set": {"<col>": <value>, ...}}
{"file": "<name>.csv", "op": "insert", "row": {"<col>": <value>, ...}}
```
For updates: `match` specifies which rows to change; `set` specifies which columns to update to which values.
For inserts: `row` is the new row to add (include all columns for that table).
Use `null` (JSON literal) for empty values.
Output ONLY the JSON array.
b2b_details.csv
b2b_txn_id,product_id,sku,supplier_name,sale_date,unit_count,unit_price_usd,revenue_gross_usd,vendor_share_pct,vendor_share_fixed_usd,status
B2B-311C,PID-311,SKU-311-AD,Anchor Distributors,2024-06-10,40,10.0,400.0,0.8,,COMPLETE
B2B-311D,PID-311,SKU-311-AD,Anchor Distributors,2024-07-20,40,13.75,550.0,0.8,,COMPLETE
B2B-201A,PID-201,SKU-201-BL,BayLine Distributors,2024-06-01,40,7.5,300.0,0.65,,COMPLETE
B2B-311E,PID-311,SKU-311-AD,Anchor Distributors,2024-04-15,50,12.0,600.0,0.8,,COMPLETE
B2B-050A,PID-050,SKU-050-AD,Anchor Distributors,2024-05-20,30,8.0,240.0,0.75,,COMPLETE
catalog.csv
product_id,sku,product_name,supplier_id,vendor_share_pct,vendor_share_fixed_usd,effective_from
PID-042,SKU-042-AD,Cola Classic 500ml,AD,0.8,,2020-01-01
PID-070,SKU-070-MG,Cola Classic 500ml,MG,0.75,,2020-01-01
PID-107,SKU-107-BL,Organic Soy Milk 1L,BL,0.7,,2020-01-01
PID-088,SKU-088-MG,Organic Soy Milk 1L,MG,0.75,,2020-01-01
PID-234,SKU-234-HD,Extra Virgin Olive Oil 500ml,HD,0.75,,2020-01-01
PID-311,SKU-311-AD,Sparkling Water 12-pack,AD,0.8,,2020-01-01
PID-050,SKU-050-AD,Rolled Oats 1kg,AD,0.75,,2020-01-01
PID-102,SKU-102-HD,Almond Butter 340g,HD,0.6,,2020-01-01
PID-201,SKU-201-BL,Sourdough Loaf 500g,BL,0.7,,2020-01-01
currency_rates.csv
from_currency,to_currency,rate,effective_from
USD,USD,1.0,2020-01-01
USD,GBP,0.78,2020-01-01
USD,EUR,0.92,2020-01-01
itemised_statement.py
"""Itemised statement — per-transaction detail with sku + supplier via SQL.
Reads: catalog.csv, transactions.csv (retail), b2b_details.csv (bulk/wholesale).
Displays sku and supplier_name FROM the transaction's own baked fields
(no lookup) for BOTH retail and b2b.
"""
import csv
import sqlite3
from pathlib import Path
CATALOG_SCHEMA = """
CREATE TABLE catalog (
product_id TEXT, sku TEXT, product_name TEXT, supplier_id TEXT,
vendor_share_pct REAL, vendor_share_fixed_usd REAL, effective_from TEXT
)"""
SUPPLIERS_SCHEMA = """
CREATE TABLE suppliers (
supplier_id TEXT, supplier_name TEXT, supplier_currency TEXT
)"""
TXN_SCHEMA = """
CREATE TABLE transactions (
transaction_id TEXT, product_id TEXT, sku TEXT, supplier_name TEXT,
sale_date TEXT, revenue_gross_usd REAL, vendor_payout_usd REAL,
transaction_fee_usd REAL, affiliate_commission_usd REAL,
status TEXT, promo_id TEXT, bundle_name TEXT
)"""
B2B_SCHEMA = """
CREATE TABLE b2b_details (
b2b_txn_id TEXT, product_id TEXT, sku TEXT, supplier_name TEXT,
sale_date TEXT, unit_count INTEGER, unit_price_usd REAL,
revenue_gross_usd REAL, vendor_share_pct REAL, vendor_share_fixed_usd REAL,
status TEXT
)"""
PROMOTIONS_SCHEMA = """
CREATE TABLE promotions (
promo_id TEXT, promo_name TEXT, start_date TEXT, end_date TEXT, discount_pct REAL
)"""
DISCOUNTS_SCHEMA = """
CREATE TABLE product_discounts (
product_id TEXT, discount_pct REAL, valid_from TEXT, valid_until TEXT
)"""
def load_csv(path):
if not path.exists():
return []
with open(path, newline="") as f:
return list(csv.DictReader(f))
def to_num(v):
if v is None or v == "":
return None
try:
return float(v)
except (ValueError, TypeError):
return None
def to_int(v):
if v is None or v == "":
return None
try:
return int(float(v))
except (ValueError, TypeError):
return None
def init_db(here: Path) -> sqlite3.Connection:
conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row
conn.execute(CATALOG_SCHEMA)
conn.execute(SUPPLIERS_SCHEMA)
conn.execute(TXN_SCHEMA)
conn.execute(B2B_SCHEMA)
conn.execute(PROMOTIONS_SCHEMA)
conn.execute(DISCOUNTS_SCHEMA)
for r in load_csv(here / "catalog.csv"):
conn.execute("INSERT INTO catalog VALUES (?,?,?,?,?,?,?)", (
r["product_id"], r["sku"], r["product_name"], r["supplier_id"],
to_num(r.get("vendor_share_pct")), to_num(r.get("vendor_share_fixed_usd")),
r["effective_from"]))
for r in load_csv(here / "suppliers.csv"):
conn.execute("INSERT INTO suppliers VALUES (?,?,?)", (
r["supplier_id"], r["supplier_name"], r.get("supplier_currency", "USD")))
for r in load_csv(here / "transactions.csv"):
conn.execute("INSERT INTO transactions VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", (
r["transaction_id"], r["product_id"], r["sku"], r["supplier_name"],
r["sale_date"], to_num(r.get("revenue_gross_usd")),
to_num(r.get("vendor_payout_usd")),
to_num(r.get("transaction_fee_usd")), to_num(r.get("affiliate_commission_usd")),
r.get("status", "COMPLETE"), r.get("promo_id", ""), r.get("bundle_name", "")))
for r in load_csv(here / "b2b_details.csv"):
conn.execute("INSERT INTO b2b_details VALUES (?,?,?,?,?,?,?,?,?,?,?)", (
r["b2b_txn_id"], r["product_id"], r["sku"], r["supplier_name"],
r["sale_date"], to_int(r.get("unit_count")), to_num(r.get("unit_price_usd")),
to_num(r.get("revenue_gross_usd")),
to_num(r.get("vendor_share_pct")), to_num(r.get("vendor_share_fixed_usd")),
r.get("status", "COMPLETE")))
for r in load_csv(here / "promotions.csv"):
conn.execute("INSERT INTO promotions VALUES (?,?,?,?,?)", (
r["promo_id"], r["promo_name"], r["start_date"], r["end_date"],
to_num(r.get("discount_pct"))))
for r in load_csv(here / "product_discounts.csv"):
conn.execute("INSERT INTO product_discounts VALUES (?,?,?,?)", (
r["product_id"], to_num(r.get("discount_pct")),
r["valid_from"], r["valid_until"]))
conn.commit()
return conn
ITEMISED_STATEMENT_SQL = """
WITH catalog_versioned AS (
SELECT c.product_id, c.sku AS catalog_sku, c.supplier_id AS catalog_supplier_id,
c.vendor_share_pct AS catalog_vendor_share_pct,
c.vendor_share_fixed_usd AS catalog_vendor_share_fixed_usd,
c.effective_from AS catalog_effective_from,
ROW_NUMBER() OVER (PARTITION BY c.product_id ORDER BY c.effective_from DESC) AS row_recency
FROM catalog c
),
active_catalog AS (
SELECT * FROM catalog_versioned WHERE row_recency = 1
),
retail_lines AS (
SELECT
t.transaction_id AS txn_ref,
t.sale_date,
t.sku AS display_sku,
'retail' AS channel,
t.supplier_name AS display_supplier_name,
1 AS units,
t.revenue_gross_usd,
(COALESCE(t.transaction_fee_usd, 0) + COALESCE(t.affiliate_commission_usd, 0)) AS deductions_usd,
COALESCE(
t.vendor_payout_usd,
CASE
WHEN ac.catalog_vendor_share_fixed_usd IS NOT NULL THEN ac.catalog_vendor_share_fixed_usd
WHEN ac.catalog_vendor_share_pct IS NOT NULL THEN
(t.revenue_gross_usd
- COALESCE(t.transaction_fee_usd, 0)
- COALESCE(t.affiliate_commission_usd, 0)
) * ac.catalog_vendor_share_pct
ELSE 0
END
) AS vendor_payout_usd
FROM transactions t
LEFT JOIN active_catalog ac
ON t.product_id = ac.product_id
AND ac.catalog_effective_from <= t.sale_date
WHERE t.status = 'COMPLETE'
),
b2b_lines AS (
SELECT
b.b2b_txn_id AS txn_ref,
b.sale_date,
b.sku AS display_sku,
'b2b' AS channel,
b.supplier_name AS display_supplier_name,
b.unit_count AS units,
b.revenue_gross_usd,
0.0 AS deductions_usd,
CASE
WHEN b.vendor_share_fixed_usd IS NOT NULL THEN b.vendor_share_fixed_usd * b.unit_count
WHEN b.vendor_share_pct IS NOT NULL THEN b.revenue_gross_usd * b.vendor_share_pct
ELSE 0
END AS vendor_payout_usd
FROM b2b_details b
WHERE b.status = 'COMPLETE'
),
unioned AS (
SELECT * FROM retail_lines UNION ALL SELECT * FROM b2b_lines
)
SELECT
txn_ref,
sale_date,
display_sku AS sku,
channel,
display_supplier_name AS supplier_name,
units,
ROUND(revenue_gross_usd, 2) AS revenue_gross_usd,
ROUND(deductions_usd, 2) AS deductions_usd,
ROUND(vendor_payout_usd, 2) AS vendor_payout_usd
FROM unioned
ORDER BY sale_date, txn_ref;
"""
def main():
here = Path(__file__).parent
conn = init_db(here)
cursor = conn.execute(ITEMISED_STATEMENT_SQL)
columns = [d[0] for d in cursor.description]
print(",".join(columns))
for row in cursor.fetchall():
vals = []
for c in columns:
v = row[c]
if isinstance(v, float):
vals.append(f"{v:.2f}")
elif v is None:
vals.append("")
elif isinstance(v, str) and "," in v:
vals.append(f'"{v}"')
else:
vals.append(str(v))
print(",".join(vals))
conn.close()
if __name__ == "__main__":
main()
product_discounts.csv
product_id,discount_pct,valid_from,valid_until
PID-050,0.05,2024-01-01,2024-12-31
promotions.csv
promo_id,promo_name,start_date,end_date,discount_pct
PROMO-SPRING24,Spring Sale 2024,2024-03-01,2024-05-31,0.15
PROMO-SUMMER24,Summer Deal 2024,2024-06-01,2024-08-31,0.1
suppliers.csv
supplier_id,supplier_name,supplier_currency
AD,Anchor Distributors,USD
HD,Heartland Distributors,USD
BL,BayLine Distributors,USD
MG,MerchantGate Distributors,USD
ticket.md
# Ticket
Cola Classic 500ml sales are being misattributed to Anchor Distributors — they should be attributed to MerchantGate Distributors under MG's vendor payout terms. Please re-book.
transactions.csv
transaction_id,product_id,sku,supplier_name,sale_date,revenue_gross_usd,vendor_payout_usd,transaction_fee_usd,affiliate_commission_usd,status,promo_id,bundle_name
TXN-001,PID-042,SKU-042-AD,Anchor Distributors,2024-03-15,100.0,74.4,5.0,2.0,COMPLETE,,
TXN-002,PID-042,SKU-042-AD,Anchor Distributors,2024-04-20,120.0,91.2,6.0,0.0,COMPLETE,,
TXN-003,PID-042,SKU-042-AD,Anchor Distributors,2024-08-10,150.0,114.0,7.5,0.0,COMPLETE,,
TXN-100,PID-107,SKU-107-BL,BayLine Distributors,2024-03-15,90.0,58.59,4.5,1.8,COMPLETE,,
TXN-101,PID-107,SKU-107-BL,BayLine Distributors,2024-05-30,110.0,73.15,5.5,0.0,COMPLETE,,
TXN-102,PID-107,SKU-107-BL,BayLine Distributors,2024-06-15,130.0,86.45,6.5,0.0,COMPLETE,,
TXN-103,PID-107,SKU-107-BL,BayLine Distributors,2024-08-01,100.0,66.5,5.0,0.0,COMPLETE,,
TXN-234A,PID-234,SKU-234-HD,Heartland Distributors,2024-03-01,50.0,34.875,2.5,1.0,COMPLETE,,
TXN-234B,PID-234,SKU-234-HD,Heartland Distributors,2024-05-01,60.0,42.75,3.0,0.0,COMPLETE,,
TXN-234C,PID-234,SKU-234-HD,Heartland Distributors,2024-07-01,55.0,39.1875,2.75,0.0,COMPLETE,,
TXN-311A,PID-311,SKU-311-AD,Anchor Distributors,2024-03-01,40.0,29.76,2.0,0.8,COMPLETE,,
TXN-311B,PID-311,SKU-311-AD,Anchor Distributors,2024-05-15,45.0,34.2,2.25,0.0,COMPLETE,,
TXN-050A,PID-050,SKU-050-AD,Anchor Distributors,2024-05-01,80.0,57.0,4.0,0.0,COMPLETE,,
TXN-102A,PID-102,SKU-102-HD,Heartland Distributors,2024-06-01,200.0,111.6,10.0,4.0,COMPLETE,,
vendor_statement.py
"""Vendor statement — aggregate vendor payout by supplier via SQL over multiple tables.
Reads: catalog.csv, suppliers.csv, transactions.csv (retail only),
b2b_details.csv (bulk/wholesale orders in a separate table),
promotions.csv, product_discounts.csv, currency_rates.csv
Runs a multi-CTE SQL query in in-memory sqlite. Retail and B2B are UNIONed
before aggregation.
"""
import csv
import sqlite3
from pathlib import Path
CATALOG_SCHEMA = """
CREATE TABLE catalog (
product_id TEXT, sku TEXT, product_name TEXT, supplier_id TEXT,
vendor_share_pct REAL, vendor_share_fixed_usd REAL, effective_from TEXT
)"""
SUPPLIERS_SCHEMA = """
CREATE TABLE suppliers (
supplier_id TEXT, supplier_name TEXT, supplier_currency TEXT
)"""
TXN_SCHEMA = """
CREATE TABLE transactions (
transaction_id TEXT, product_id TEXT, sku TEXT, supplier_name TEXT,
sale_date TEXT, revenue_gross_usd REAL, vendor_payout_usd REAL,
transaction_fee_usd REAL, affiliate_commission_usd REAL,
status TEXT, promo_id TEXT, bundle_name TEXT
)"""
B2B_SCHEMA = """
CREATE TABLE b2b_details (
b2b_txn_id TEXT, product_id TEXT, sku TEXT, supplier_name TEXT,
sale_date TEXT, unit_count INTEGER, unit_price_usd REAL,
revenue_gross_usd REAL, vendor_share_pct REAL, vendor_share_fixed_usd REAL,
status TEXT
)"""
PROMOTIONS_SCHEMA = """
CREATE TABLE promotions (
promo_id TEXT, promo_name TEXT, start_date TEXT, end_date TEXT, discount_pct REAL
)"""
DISCOUNTS_SCHEMA = """
CREATE TABLE product_discounts (
product_id TEXT, discount_pct REAL, valid_from TEXT, valid_until TEXT
)"""
CURRENCY_SCHEMA = """
CREATE TABLE currency_rates (
from_currency TEXT, to_currency TEXT, rate REAL, effective_from TEXT
)"""
def load_csv(path):
if not path.exists():
return []
with open(path, newline="") as f:
return list(csv.DictReader(f))
def to_num(v):
if v is None or v == "":
return None
try:
return float(v)
except (ValueError, TypeError):
return None
def to_int(v):
if v is None or v == "":
return None
try:
return int(float(v))
except (ValueError, TypeError):
return None
def init_db(here: Path) -> sqlite3.Connection:
conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row
conn.execute(CATALOG_SCHEMA)
conn.execute(SUPPLIERS_SCHEMA)
conn.execute(TXN_SCHEMA)
conn.execute(B2B_SCHEMA)
conn.execute(PROMOTIONS_SCHEMA)
conn.execute(DISCOUNTS_SCHEMA)
conn.execute(CURRENCY_SCHEMA)
for r in load_csv(here / "catalog.csv"):
conn.execute("INSERT INTO catalog VALUES (?,?,?,?,?,?,?)", (
r["product_id"], r["sku"], r["product_name"], r["supplier_id"],
to_num(r.get("vendor_share_pct")), to_num(r.get("vendor_share_fixed_usd")),
r["effective_from"]))
for r in load_csv(here / "suppliers.csv"):
conn.execute("INSERT INTO suppliers VALUES (?,?,?)", (
r["supplier_id"], r["supplier_name"], r.get("supplier_currency", "USD")))
for r in load_csv(here / "transactions.csv"):
conn.execute("INSERT INTO transactions VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", (
r["transaction_id"], r["product_id"], r["sku"], r["supplier_name"],
r["sale_date"], to_num(r.get("revenue_gross_usd")),
to_num(r.get("vendor_payout_usd")),
to_num(r.get("transaction_fee_usd")), to_num(r.get("affiliate_commission_usd")),
r.get("status", "COMPLETE"), r.get("promo_id", ""), r.get("bundle_name", "")))
for r in load_csv(here / "b2b_details.csv"):
conn.execute("INSERT INTO b2b_details VALUES (?,?,?,?,?,?,?,?,?,?,?)", (
r["b2b_txn_id"], r["product_id"], r["sku"], r["supplier_name"],
r["sale_date"], to_int(r.get("unit_count")), to_num(r.get("unit_price_usd")),
to_num(r.get("revenue_gross_usd")),
to_num(r.get("vendor_share_pct")), to_num(r.get("vendor_share_fixed_usd")),
r.get("status", "COMPLETE")))
for r in load_csv(here / "promotions.csv"):
conn.execute("INSERT INTO promotions VALUES (?,?,?,?,?)", (
r["promo_id"], r["promo_name"], r["start_date"], r["end_date"],
to_num(r.get("discount_pct"))))
for r in load_csv(here / "product_discounts.csv"):
conn.execute("INSERT INTO product_discounts VALUES (?,?,?,?)", (
r["product_id"], to_num(r.get("discount_pct")),
r["valid_from"], r["valid_until"]))
for r in load_csv(here / "currency_rates.csv"):
conn.execute("INSERT INTO currency_rates VALUES (?,?,?,?)", (
r["from_currency"], r["to_currency"], to_num(r.get("rate")),
r["effective_from"]))
conn.commit()
return conn
VENDOR_STATEMENT_SQL = """
WITH catalog_versioned AS (
SELECT c.product_id, c.sku AS catalog_sku, c.product_name,
c.supplier_id AS catalog_supplier_id, s.supplier_name AS resolved_supplier_name,
c.vendor_share_pct AS catalog_vendor_share_pct,
c.vendor_share_fixed_usd AS catalog_vendor_share_fixed_usd,
c.effective_from AS catalog_effective_from,
ROW_NUMBER() OVER (PARTITION BY c.product_id ORDER BY c.effective_from DESC) AS row_recency
FROM catalog c
JOIN suppliers s ON c.supplier_id = s.supplier_id
),
active_catalog AS (
SELECT * FROM catalog_versioned WHERE row_recency = 1
),
retail_payout AS (
SELECT
t.transaction_id AS txn_ref,
ac.resolved_supplier_name AS supplier_name,
t.revenue_gross_usd,
(COALESCE(t.transaction_fee_usd, 0) + COALESCE(t.affiliate_commission_usd, 0)) AS deductions_usd,
COALESCE(
t.vendor_payout_usd,
CASE
WHEN ac.catalog_vendor_share_fixed_usd IS NOT NULL THEN ac.catalog_vendor_share_fixed_usd
WHEN ac.catalog_vendor_share_pct IS NOT NULL THEN
(t.revenue_gross_usd
- COALESCE(t.transaction_fee_usd, 0)
- COALESCE(t.affiliate_commission_usd, 0)
) * ac.catalog_vendor_share_pct
ELSE 0
END
) AS payout_amount
FROM transactions t
LEFT JOIN active_catalog ac
ON t.product_id = ac.product_id
AND ac.catalog_effective_from <= t.sale_date
WHERE t.status = 'COMPLETE'
),
b2b_payout AS (
SELECT
b.b2b_txn_id AS txn_ref,
ac.resolved_supplier_name AS supplier_name,
b.revenue_gross_usd,
0.0 AS deductions_usd,
CASE
WHEN b.vendor_share_fixed_usd IS NOT NULL THEN b.vendor_share_fixed_usd * b.unit_count
WHEN b.vendor_share_pct IS NOT NULL THEN b.revenue_gross_usd * b.vendor_share_pct
ELSE 0
END AS payout_amount
FROM b2b_details b
LEFT JOIN active_catalog ac
ON b.product_id = ac.product_id
AND ac.catalog_effective_from <= b.sale_date
WHERE b.status = 'COMPLETE'
),
unioned AS (
SELECT * FROM retail_payout UNION ALL SELECT * FROM b2b_payout
)
SELECT
supplier_name,
COUNT(*) AS transaction_count,
ROUND(SUM(revenue_gross_usd), 2) AS total_revenue_usd,
ROUND(SUM(deductions_usd), 2) AS total_deductions_usd,
ROUND(SUM(revenue_gross_usd - deductions_usd), 2) AS total_net_usd,
ROUND(SUM(payout_amount), 2) AS total_payout_usd
FROM unioned
GROUP BY supplier_name
ORDER BY supplier_name;
"""
def main():
here = Path(__file__).parent
conn = init_db(here)
cursor = conn.execute(VENDOR_STATEMENT_SQL)
columns = [d[0] for d in cursor.description]
print(",".join(columns))
for row in cursor.fetchall():
vals = []
for c in columns:
v = row[c]
if isinstance(v, float):
vals.append(f"{v:.2f}")
elif v is None:
vals.append("")
elif isinstance(v, str) and "," in v:
vals.append(f'"{v}"')
else:
vals.append(str(v))
print(",".join(vals))
conn.close()
if __name__ == "__main__":
main()
expected output
answer.json
{
"id": "case_01",
"category": "supplier_change_alltime",
"gold_edits": [
{
"file": "transactions.csv",
"op": "update",
"match": {
"transaction_id": "TXN-001"
},
"set": {
"product_id": "PID-070",
"supplier_name": "MerchantGate Distributors",
"sku": "SKU-070-MG",
"vendor_payout_usd": 69.75
}
},
{
"file": "transactions.csv",
"op": "update",
"match": {
"transaction_id": "TXN-002"
},
"set": {
"product_id": "PID-070",
"supplier_name": "MerchantGate Distributors",
"sku": "SKU-070-MG",
"vendor_payout_usd": 85.5
}
},
{
"file": "transactions.csv",
"op": "update",
"match": {
"transaction_id": "TXN-003"
},
"set": {
"product_id": "PID-070",
"supplier_name": "MerchantGate Distributors",
"sku": "SKU-070-MG",
"vendor_payout_usd": 106.875
}
}
]
}Scored by judge.py — see Scoring logic below for the full rule.
▸case_02Re-book misattributed sales to correct vendor from a specific date onwards (partial re-book)
input
README.md
# Task
A colleague filed a ticket asking for a fix to the vendor-payout pipeline.
**Your goal:** ensure both reports (`vendor_statement.py` and `itemised_statement.py`) come out CORRECT after the fix is applied.
You will not be graded on your intermediate reasoning or on the structure of your edits. You will be graded on **whether the reports the two scripts produce match the reports they would produce after a correct fix**. Judge applies your edits + runs both scripts + compares stdout to the gold reports.
## Business context
This supermarket runs a consignment model: vendors stock the shelves, and every period the store owes each vendor a share of what sold. These two reports are the source of truth for that monthly payout run:
- `vendor_statement` — the per-vendor summary Finance uses to cut monthly payout checks. The `total_payout_usd` column IS the amount each vendor gets paid.
- `itemised_statement` — the per-transaction breakdown vendors audit against their sales.
Getting `vendor_payout_usd` right on every affected line is the whole point. An inconsistency here is a real over/under-payment to a vendor. So when a ticket changes something upstream — attribution, pricing, terms — think through: does the change move payout money around, or change how much payout is owed? Both flow through these reports and both must land right.
## How to approach this
In a real production system:
- The data tables are LARGE — you cannot solve this by scanning every row.
- No one documents the business rules explicitly — you have to reverse-engineer them by reading the pipeline code (`vendor_statement.py` and `itemised_statement.py`) and understanding what each script does with the data.
To decide what edits are needed, read the scripts carefully:
- Where does each report get its values from? Which table? Which column? Lookup or baked?
- If you change X in the source data, which report changes and how?
- Multiple valid fix approaches may exist — pick the one that produces reports consistent with the business intent of the ticket.
## Files in this directory
- `ticket.md` — the change request
- `catalog.csv`, `suppliers.csv` — product/vendor master
- `transactions.csv` — retail (checkout) sales
- `b2b_details.csv` — wholesale/bulk orders (high volume: many units per order, e.g. restaurant supply)
- `promotions.csv`, `product_discounts.csv`, `currency_rates.csv` — auxiliary tables
- `vendor_statement.py`, `itemised_statement.py` — the two report scripts (Python + SQL)
## Output format
Emit a JSON array of edits to stdout. Nothing else — no explanation, no markdown fences.
Each edit is one of:
```
{"file": "<name>.csv", "op": "update", "match": {"<col>": "<value>"}, "set": {"<col>": <value>, ...}}
{"file": "<name>.csv", "op": "insert", "row": {"<col>": <value>, ...}}
```
For updates: `match` specifies which rows to change; `set` specifies which columns to update to which values.
For inserts: `row` is the new row to add (include all columns for that table).
Use `null` (JSON literal) for empty values.
Output ONLY the JSON array.
b2b_details.csv
b2b_txn_id,product_id,sku,supplier_name,sale_date,unit_count,unit_price_usd,revenue_gross_usd,vendor_share_pct,vendor_share_fixed_usd,status
B2B-311C,PID-311,SKU-311-AD,Anchor Distributors,2024-06-10,40,10.0,400.0,0.8,,COMPLETE
B2B-311D,PID-311,SKU-311-AD,Anchor Distributors,2024-07-20,40,13.75,550.0,0.8,,COMPLETE
B2B-201A,PID-201,SKU-201-BL,BayLine Distributors,2024-06-01,40,7.5,300.0,0.65,,COMPLETE
B2B-311E,PID-311,SKU-311-AD,Anchor Distributors,2024-04-15,50,12.0,600.0,0.8,,COMPLETE
B2B-050A,PID-050,SKU-050-AD,Anchor Distributors,2024-05-20,30,8.0,240.0,0.75,,COMPLETE
catalog.csv
product_id,sku,product_name,supplier_id,vendor_share_pct,vendor_share_fixed_usd,effective_from
PID-042,SKU-042-AD,Cola Classic 500ml,AD,0.8,,2020-01-01
PID-070,SKU-070-MG,Cola Classic 500ml,MG,0.75,,2020-01-01
PID-107,SKU-107-BL,Organic Soy Milk 1L,BL,0.7,,2020-01-01
PID-088,SKU-088-MG,Organic Soy Milk 1L,MG,0.75,,2020-01-01
PID-234,SKU-234-HD,Extra Virgin Olive Oil 500ml,HD,0.75,,2020-01-01
PID-311,SKU-311-AD,Sparkling Water 12-pack,AD,0.8,,2020-01-01
PID-050,SKU-050-AD,Rolled Oats 1kg,AD,0.75,,2020-01-01
PID-102,SKU-102-HD,Almond Butter 340g,HD,0.6,,2020-01-01
PID-201,SKU-201-BL,Sourdough Loaf 500g,BL,0.7,,2020-01-01
currency_rates.csv
from_currency,to_currency,rate,effective_from
USD,USD,1.0,2020-01-01
USD,GBP,0.78,2020-01-01
USD,EUR,0.92,2020-01-01
itemised_statement.py
"""Itemised statement — per-transaction detail with sku + supplier via SQL.
Reads: catalog.csv, transactions.csv (retail), b2b_details.csv (bulk/wholesale).
Displays sku and supplier_name FROM the transaction's own baked fields
(no lookup) for BOTH retail and b2b.
"""
import csv
import sqlite3
from pathlib import Path
CATALOG_SCHEMA = """
CREATE TABLE catalog (
product_id TEXT, sku TEXT, product_name TEXT, supplier_id TEXT,
vendor_share_pct REAL, vendor_share_fixed_usd REAL, effective_from TEXT
)"""
SUPPLIERS_SCHEMA = """
CREATE TABLE suppliers (
supplier_id TEXT, supplier_name TEXT, supplier_currency TEXT
)"""
TXN_SCHEMA = """
CREATE TABLE transactions (
transaction_id TEXT, product_id TEXT, sku TEXT, supplier_name TEXT,
sale_date TEXT, revenue_gross_usd REAL, vendor_payout_usd REAL,
transaction_fee_usd REAL, affiliate_commission_usd REAL,
status TEXT, promo_id TEXT, bundle_name TEXT
)"""
B2B_SCHEMA = """
CREATE TABLE b2b_details (
b2b_txn_id TEXT, product_id TEXT, sku TEXT, supplier_name TEXT,
sale_date TEXT, unit_count INTEGER, unit_price_usd REAL,
revenue_gross_usd REAL, vendor_share_pct REAL, vendor_share_fixed_usd REAL,
status TEXT
)"""
PROMOTIONS_SCHEMA = """
CREATE TABLE promotions (
promo_id TEXT, promo_name TEXT, start_date TEXT, end_date TEXT, discount_pct REAL
)"""
DISCOUNTS_SCHEMA = """
CREATE TABLE product_discounts (
product_id TEXT, discount_pct REAL, valid_from TEXT, valid_until TEXT
)"""
def load_csv(path):
if not path.exists():
return []
with open(path, newline="") as f:
return list(csv.DictReader(f))
def to_num(v):
if v is None or v == "":
return None
try:
return float(v)
except (ValueError, TypeError):
return None
def to_int(v):
if v is None or v == "":
return None
try:
return int(float(v))
except (ValueError, TypeError):
return None
def init_db(here: Path) -> sqlite3.Connection:
conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row
conn.execute(CATALOG_SCHEMA)
conn.execute(SUPPLIERS_SCHEMA)
conn.execute(TXN_SCHEMA)
conn.execute(B2B_SCHEMA)
conn.execute(PROMOTIONS_SCHEMA)
conn.execute(DISCOUNTS_SCHEMA)
for r in load_csv(here / "catalog.csv"):
conn.execute("INSERT INTO catalog VALUES (?,?,?,?,?,?,?)", (
r["product_id"], r["sku"], r["product_name"], r["supplier_id"],
to_num(r.get("vendor_share_pct")), to_num(r.get("vendor_share_fixed_usd")),
r["effective_from"]))
for r in load_csv(here / "suppliers.csv"):
conn.execute("INSERT INTO suppliers VALUES (?,?,?)", (
r["supplier_id"], r["supplier_name"], r.get("supplier_currency", "USD")))
for r in load_csv(here / "transactions.csv"):
conn.execute("INSERT INTO transactions VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", (
r["transaction_id"], r["product_id"], r["sku"], r["supplier_name"],
r["sale_date"], to_num(r.get("revenue_gross_usd")),
to_num(r.get("vendor_payout_usd")),
to_num(r.get("transaction_fee_usd")), to_num(r.get("affiliate_commission_usd")),
r.get("status", "COMPLETE"), r.get("promo_id", ""), r.get("bundle_name", "")))
for r in load_csv(here / "b2b_details.csv"):
conn.execute("INSERT INTO b2b_details VALUES (?,?,?,?,?,?,?,?,?,?,?)", (
r["b2b_txn_id"], r["product_id"], r["sku"], r["supplier_name"],
r["sale_date"], to_int(r.get("unit_count")), to_num(r.get("unit_price_usd")),
to_num(r.get("revenue_gross_usd")),
to_num(r.get("vendor_share_pct")), to_num(r.get("vendor_share_fixed_usd")),
r.get("status", "COMPLETE")))
for r in load_csv(here / "promotions.csv"):
conn.execute("INSERT INTO promotions VALUES (?,?,?,?,?)", (
r["promo_id"], r["promo_name"], r["start_date"], r["end_date"],
to_num(r.get("discount_pct"))))
for r in load_csv(here / "product_discounts.csv"):
conn.execute("INSERT INTO product_discounts VALUES (?,?,?,?)", (
r["product_id"], to_num(r.get("discount_pct")),
r["valid_from"], r["valid_until"]))
conn.commit()
return conn
ITEMISED_STATEMENT_SQL = """
WITH catalog_versioned AS (
SELECT c.product_id, c.sku AS catalog_sku, c.supplier_id AS catalog_supplier_id,
c.vendor_share_pct AS catalog_vendor_share_pct,
c.vendor_share_fixed_usd AS catalog_vendor_share_fixed_usd,
c.effective_from AS catalog_effective_from,
ROW_NUMBER() OVER (PARTITION BY c.product_id ORDER BY c.effective_from DESC) AS row_recency
FROM catalog c
),
active_catalog AS (
SELECT * FROM catalog_versioned WHERE row_recency = 1
),
retail_lines AS (
SELECT
t.transaction_id AS txn_ref,
t.sale_date,
t.sku AS display_sku,
'retail' AS channel,
t.supplier_name AS display_supplier_name,
1 AS units,
t.revenue_gross_usd,
(COALESCE(t.transaction_fee_usd, 0) + COALESCE(t.affiliate_commission_usd, 0)) AS deductions_usd,
COALESCE(
t.vendor_payout_usd,
CASE
WHEN ac.catalog_vendor_share_fixed_usd IS NOT NULL THEN ac.catalog_vendor_share_fixed_usd
WHEN ac.catalog_vendor_share_pct IS NOT NULL THEN
(t.revenue_gross_usd
- COALESCE(t.transaction_fee_usd, 0)
- COALESCE(t.affiliate_commission_usd, 0)
) * ac.catalog_vendor_share_pct
ELSE 0
END
) AS vendor_payout_usd
FROM transactions t
LEFT JOIN active_catalog ac
ON t.product_id = ac.product_id
AND ac.catalog_effective_from <= t.sale_date
WHERE t.status = 'COMPLETE'
),
b2b_lines AS (
SELECT
b.b2b_txn_id AS txn_ref,
b.sale_date,
b.sku AS display_sku,
'b2b' AS channel,
b.supplier_name AS display_supplier_name,
b.unit_count AS units,
b.revenue_gross_usd,
0.0 AS deductions_usd,
CASE
WHEN b.vendor_share_fixed_usd IS NOT NULL THEN b.vendor_share_fixed_usd * b.unit_count
WHEN b.vendor_share_pct IS NOT NULL THEN b.revenue_gross_usd * b.vendor_share_pct
ELSE 0
END AS vendor_payout_usd
FROM b2b_details b
WHERE b.status = 'COMPLETE'
),
unioned AS (
SELECT * FROM retail_lines UNION ALL SELECT * FROM b2b_lines
)
SELECT
txn_ref,
sale_date,
display_sku AS sku,
channel,
display_supplier_name AS supplier_name,
units,
ROUND(revenue_gross_usd, 2) AS revenue_gross_usd,
ROUND(deductions_usd, 2) AS deductions_usd,
ROUND(vendor_payout_usd, 2) AS vendor_payout_usd
FROM unioned
ORDER BY sale_date, txn_ref;
"""
def main():
here = Path(__file__).parent
conn = init_db(here)
cursor = conn.execute(ITEMISED_STATEMENT_SQL)
columns = [d[0] for d in cursor.description]
print(",".join(columns))
for row in cursor.fetchall():
vals = []
for c in columns:
v = row[c]
if isinstance(v, float):
vals.append(f"{v:.2f}")
elif v is None:
vals.append("")
elif isinstance(v, str) and "," in v:
vals.append(f'"{v}"')
else:
vals.append(str(v))
print(",".join(vals))
conn.close()
if __name__ == "__main__":
main()
product_discounts.csv
product_id,discount_pct,valid_from,valid_until
PID-050,0.05,2024-01-01,2024-12-31
promotions.csv
promo_id,promo_name,start_date,end_date,discount_pct
PROMO-SPRING24,Spring Sale 2024,2024-03-01,2024-05-31,0.15
PROMO-SUMMER24,Summer Deal 2024,2024-06-01,2024-08-31,0.1
suppliers.csv
supplier_id,supplier_name,supplier_currency
AD,Anchor Distributors,USD
HD,Heartland Distributors,USD
BL,BayLine Distributors,USD
MG,MerchantGate Distributors,USD
ticket.md
# Ticket
Organic Soy Milk 1L sales from 2024-06-01 onwards are being misattributed to BayLine Distributors — from that date on, they should be attributed to MerchantGate Distributors under MG's vendor payout terms. Please re-book (sales before 2024-06-01 stay as-is).
transactions.csv
transaction_id,product_id,sku,supplier_name,sale_date,revenue_gross_usd,vendor_payout_usd,transaction_fee_usd,affiliate_commission_usd,status,promo_id,bundle_name
TXN-001,PID-042,SKU-042-AD,Anchor Distributors,2024-03-15,100.0,74.4,5.0,2.0,COMPLETE,,
TXN-002,PID-042,SKU-042-AD,Anchor Distributors,2024-04-20,120.0,91.2,6.0,0.0,COMPLETE,,
TXN-003,PID-042,SKU-042-AD,Anchor Distributors,2024-08-10,150.0,114.0,7.5,0.0,COMPLETE,,
TXN-100,PID-107,SKU-107-BL,BayLine Distributors,2024-03-15,90.0,58.59,4.5,1.8,COMPLETE,,
TXN-101,PID-107,SKU-107-BL,BayLine Distributors,2024-05-30,110.0,73.15,5.5,0.0,COMPLETE,,
TXN-102,PID-107,SKU-107-BL,BayLine Distributors,2024-06-15,130.0,86.45,6.5,0.0,COMPLETE,,
TXN-103,PID-107,SKU-107-BL,BayLine Distributors,2024-08-01,100.0,66.5,5.0,0.0,COMPLETE,,
TXN-234A,PID-234,SKU-234-HD,Heartland Distributors,2024-03-01,50.0,34.875,2.5,1.0,COMPLETE,,
TXN-234B,PID-234,SKU-234-HD,Heartland Distributors,2024-05-01,60.0,42.75,3.0,0.0,COMPLETE,,
TXN-234C,PID-234,SKU-234-HD,Heartland Distributors,2024-07-01,55.0,39.1875,2.75,0.0,COMPLETE,,
TXN-311A,PID-311,SKU-311-AD,Anchor Distributors,2024-03-01,40.0,29.76,2.0,0.8,COMPLETE,,
TXN-311B,PID-311,SKU-311-AD,Anchor Distributors,2024-05-15,45.0,34.2,2.25,0.0,COMPLETE,,
TXN-050A,PID-050,SKU-050-AD,Anchor Distributors,2024-05-01,80.0,57.0,4.0,0.0,COMPLETE,,
TXN-102A,PID-102,SKU-102-HD,Heartland Distributors,2024-06-01,200.0,111.6,10.0,4.0,COMPLETE,,
vendor_statement.py
"""Vendor statement — aggregate vendor payout by supplier via SQL over multiple tables.
Reads: catalog.csv, suppliers.csv, transactions.csv (retail only),
b2b_details.csv (bulk/wholesale orders in a separate table),
promotions.csv, product_discounts.csv, currency_rates.csv
Runs a multi-CTE SQL query in in-memory sqlite. Retail and B2B are UNIONed
before aggregation.
"""
import csv
import sqlite3
from pathlib import Path
CATALOG_SCHEMA = """
CREATE TABLE catalog (
product_id TEXT, sku TEXT, product_name TEXT, supplier_id TEXT,
vendor_share_pct REAL, vendor_share_fixed_usd REAL, effective_from TEXT
)"""
SUPPLIERS_SCHEMA = """
CREATE TABLE suppliers (
supplier_id TEXT, supplier_name TEXT, supplier_currency TEXT
)"""
TXN_SCHEMA = """
CREATE TABLE transactions (
transaction_id TEXT, product_id TEXT, sku TEXT, supplier_name TEXT,
sale_date TEXT, revenue_gross_usd REAL, vendor_payout_usd REAL,
transaction_fee_usd REAL, affiliate_commission_usd REAL,
status TEXT, promo_id TEXT, bundle_name TEXT
)"""
B2B_SCHEMA = """
CREATE TABLE b2b_details (
b2b_txn_id TEXT, product_id TEXT, sku TEXT, supplier_name TEXT,
sale_date TEXT, unit_count INTEGER, unit_price_usd REAL,
revenue_gross_usd REAL, vendor_share_pct REAL, vendor_share_fixed_usd REAL,
status TEXT
)"""
PROMOTIONS_SCHEMA = """
CREATE TABLE promotions (
promo_id TEXT, promo_name TEXT, start_date TEXT, end_date TEXT, discount_pct REAL
)"""
DISCOUNTS_SCHEMA = """
CREATE TABLE product_discounts (
product_id TEXT, discount_pct REAL, valid_from TEXT, valid_until TEXT
)"""
CURRENCY_SCHEMA = """
CREATE TABLE currency_rates (
from_currency TEXT, to_currency TEXT, rate REAL, effective_from TEXT
)"""
def load_csv(path):
if not path.exists():
return []
with open(path, newline="") as f:
return list(csv.DictReader(f))
def to_num(v):
if v is None or v == "":
return None
try:
return float(v)
except (ValueError, TypeError):
return None
def to_int(v):
if v is None or v == "":
return None
try:
return int(float(v))
except (ValueError, TypeError):
return None
def init_db(here: Path) -> sqlite3.Connection:
conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row
conn.execute(CATALOG_SCHEMA)
conn.execute(SUPPLIERS_SCHEMA)
conn.execute(TXN_SCHEMA)
conn.execute(B2B_SCHEMA)
conn.execute(PROMOTIONS_SCHEMA)
conn.execute(DISCOUNTS_SCHEMA)
conn.execute(CURRENCY_SCHEMA)
for r in load_csv(here / "catalog.csv"):
conn.execute("INSERT INTO catalog VALUES (?,?,?,?,?,?,?)", (
r["product_id"], r["sku"], r["product_name"], r["supplier_id"],
to_num(r.get("vendor_share_pct")), to_num(r.get("vendor_share_fixed_usd")),
r["effective_from"]))
for r in load_csv(here / "suppliers.csv"):
conn.execute("INSERT INTO suppliers VALUES (?,?,?)", (
r["supplier_id"], r["supplier_name"], r.get("supplier_currency", "USD")))
for r in load_csv(here / "transactions.csv"):
conn.execute("INSERT INTO transactions VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", (
r["transaction_id"], r["product_id"], r["sku"], r["supplier_name"],
r["sale_date"], to_num(r.get("revenue_gross_usd")),
to_num(r.get("vendor_payout_usd")),
to_num(r.get("transaction_fee_usd")), to_num(r.get("affiliate_commission_usd")),
r.get("status", "COMPLETE"), r.get("promo_id", ""), r.get("bundle_name", "")))
for r in load_csv(here / "b2b_details.csv"):
conn.execute("INSERT INTO b2b_details VALUES (?,?,?,?,?,?,?,?,?,?,?)", (
r["b2b_txn_id"], r["product_id"], r["sku"], r["supplier_name"],
r["sale_date"], to_int(r.get("unit_count")), to_num(r.get("unit_price_usd")),
to_num(r.get("revenue_gross_usd")),
to_num(r.get("vendor_share_pct")), to_num(r.get("vendor_share_fixed_usd")),
r.get("status", "COMPLETE")))
for r in load_csv(here / "promotions.csv"):
conn.execute("INSERT INTO promotions VALUES (?,?,?,?,?)", (
r["promo_id"], r["promo_name"], r["start_date"], r["end_date"],
to_num(r.get("discount_pct"))))
for r in load_csv(here / "product_discounts.csv"):
conn.execute("INSERT INTO product_discounts VALUES (?,?,?,?)", (
r["product_id"], to_num(r.get("discount_pct")),
r["valid_from"], r["valid_until"]))
for r in load_csv(here / "currency_rates.csv"):
conn.execute("INSERT INTO currency_rates VALUES (?,?,?,?)", (
r["from_currency"], r["to_currency"], to_num(r.get("rate")),
r["effective_from"]))
conn.commit()
return conn
VENDOR_STATEMENT_SQL = """
WITH catalog_versioned AS (
SELECT c.product_id, c.sku AS catalog_sku, c.product_name,
c.supplier_id AS catalog_supplier_id, s.supplier_name AS resolved_supplier_name,
c.vendor_share_pct AS catalog_vendor_share_pct,
c.vendor_share_fixed_usd AS catalog_vendor_share_fixed_usd,
c.effective_from AS catalog_effective_from,
ROW_NUMBER() OVER (PARTITION BY c.product_id ORDER BY c.effective_from DESC) AS row_recency
FROM catalog c
JOIN suppliers s ON c.supplier_id = s.supplier_id
),
active_catalog AS (
SELECT * FROM catalog_versioned WHERE row_recency = 1
),
retail_payout AS (
SELECT
t.transaction_id AS txn_ref,
ac.resolved_supplier_name AS supplier_name,
t.revenue_gross_usd,
(COALESCE(t.transaction_fee_usd, 0) + COALESCE(t.affiliate_commission_usd, 0)) AS deductions_usd,
COALESCE(
t.vendor_payout_usd,
CASE
WHEN ac.catalog_vendor_share_fixed_usd IS NOT NULL THEN ac.catalog_vendor_share_fixed_usd
WHEN ac.catalog_vendor_share_pct IS NOT NULL THEN
(t.revenue_gross_usd
- COALESCE(t.transaction_fee_usd, 0)
- COALESCE(t.affiliate_commission_usd, 0)
) * ac.catalog_vendor_share_pct
ELSE 0
END
) AS payout_amount
FROM transactions t
LEFT JOIN active_catalog ac
ON t.product_id = ac.product_id
AND ac.catalog_effective_from <= t.sale_date
WHERE t.status = 'COMPLETE'
),
b2b_payout AS (
SELECT
b.b2b_txn_id AS txn_ref,
ac.resolved_supplier_name AS supplier_name,
b.revenue_gross_usd,
0.0 AS deductions_usd,
CASE
WHEN b.vendor_share_fixed_usd IS NOT NULL THEN b.vendor_share_fixed_usd * b.unit_count
WHEN b.vendor_share_pct IS NOT NULL THEN b.revenue_gross_usd * b.vendor_share_pct
ELSE 0
END AS payout_amount
FROM b2b_details b
LEFT JOIN active_catalog ac
ON b.product_id = ac.product_id
AND ac.catalog_effective_from <= b.sale_date
WHERE b.status = 'COMPLETE'
),
unioned AS (
SELECT * FROM retail_payout UNION ALL SELECT * FROM b2b_payout
)
SELECT
supplier_name,
COUNT(*) AS transaction_count,
ROUND(SUM(revenue_gross_usd), 2) AS total_revenue_usd,
ROUND(SUM(deductions_usd), 2) AS total_deductions_usd,
ROUND(SUM(revenue_gross_usd - deductions_usd), 2) AS total_net_usd,
ROUND(SUM(payout_amount), 2) AS total_payout_usd
FROM unioned
GROUP BY supplier_name
ORDER BY supplier_name;
"""
def main():
here = Path(__file__).parent
conn = init_db(here)
cursor = conn.execute(VENDOR_STATEMENT_SQL)
columns = [d[0] for d in cursor.description]
print(",".join(columns))
for row in cursor.fetchall():
vals = []
for c in columns:
v = row[c]
if isinstance(v, float):
vals.append(f"{v:.2f}")
elif v is None:
vals.append("")
elif isinstance(v, str) and "," in v:
vals.append(f'"{v}"')
else:
vals.append(str(v))
print(",".join(vals))
conn.close()
if __name__ == "__main__":
main()
expected output
answer.json
{
"id": "case_02",
"category": "supplier_change_dated",
"gold_edits": [
{
"file": "transactions.csv",
"op": "update",
"match": {
"transaction_id": "TXN-102"
},
"set": {
"product_id": "PID-088",
"supplier_name": "MerchantGate Distributors",
"sku": "SKU-088-MG",
"vendor_payout_usd": 92.625
}
},
{
"file": "transactions.csv",
"op": "update",
"match": {
"transaction_id": "TXN-103"
},
"set": {
"product_id": "PID-088",
"supplier_name": "MerchantGate Distributors",
"sku": "SKU-088-MG",
"vendor_payout_usd": 71.25
}
}
]
}Scored by judge.py — see Scoring logic below for the full rule.
▸case_03SRP renegotiated to 2x current price; must recompute baked vendor payout to reflect new pricing
input
README.md
# Task
A colleague filed a ticket asking for a fix to the vendor-payout pipeline.
**Your goal:** ensure both reports (`vendor_statement.py` and `itemised_statement.py`) come out CORRECT after the fix is applied.
You will not be graded on your intermediate reasoning or on the structure of your edits. You will be graded on **whether the reports the two scripts produce match the reports they would produce after a correct fix**. Judge applies your edits + runs both scripts + compares stdout to the gold reports.
## Business context
This supermarket runs a consignment model: vendors stock the shelves, and every period the store owes each vendor a share of what sold. These two reports are the source of truth for that monthly payout run:
- `vendor_statement` — the per-vendor summary Finance uses to cut monthly payout checks. The `total_payout_usd` column IS the amount each vendor gets paid.
- `itemised_statement` — the per-transaction breakdown vendors audit against their sales.
Getting `vendor_payout_usd` right on every affected line is the whole point. An inconsistency here is a real over/under-payment to a vendor. So when a ticket changes something upstream — attribution, pricing, terms — think through: does the change move payout money around, or change how much payout is owed? Both flow through these reports and both must land right.
## How to approach this
In a real production system:
- The data tables are LARGE — you cannot solve this by scanning every row.
- No one documents the business rules explicitly — you have to reverse-engineer them by reading the pipeline code (`vendor_statement.py` and `itemised_statement.py`) and understanding what each script does with the data.
To decide what edits are needed, read the scripts carefully:
- Where does each report get its values from? Which table? Which column? Lookup or baked?
- If you change X in the source data, which report changes and how?
- Multiple valid fix approaches may exist — pick the one that produces reports consistent with the business intent of the ticket.
## Files in this directory
- `ticket.md` — the change request
- `catalog.csv`, `suppliers.csv` — product/vendor master
- `transactions.csv` — retail (checkout) sales
- `b2b_details.csv` — wholesale/bulk orders (high volume: many units per order, e.g. restaurant supply)
- `promotions.csv`, `product_discounts.csv`, `currency_rates.csv` — auxiliary tables
- `vendor_statement.py`, `itemised_statement.py` — the two report scripts (Python + SQL)
## Output format
Emit a JSON array of edits to stdout. Nothing else — no explanation, no markdown fences.
Each edit is one of:
```
{"file": "<name>.csv", "op": "update", "match": {"<col>": "<value>"}, "set": {"<col>": <value>, ...}}
{"file": "<name>.csv", "op": "insert", "row": {"<col>": <value>, ...}}
```
For updates: `match` specifies which rows to change; `set` specifies which columns to update to which values.
For inserts: `row` is the new row to add (include all columns for that table).
Use `null` (JSON literal) for empty values.
Output ONLY the JSON array.
b2b_details.csv
b2b_txn_id,product_id,sku,supplier_name,sale_date,unit_count,unit_price_usd,revenue_gross_usd,vendor_share_pct,vendor_share_fixed_usd,status
B2B-311C,PID-311,SKU-311-AD,Anchor Distributors,2024-06-10,40,10.0,400.0,0.8,,COMPLETE
B2B-311D,PID-311,SKU-311-AD,Anchor Distributors,2024-07-20,40,13.75,550.0,0.8,,COMPLETE
B2B-201A,PID-201,SKU-201-BL,BayLine Distributors,2024-06-01,40,7.5,300.0,0.65,,COMPLETE
B2B-311E,PID-311,SKU-311-AD,Anchor Distributors,2024-04-15,50,12.0,600.0,0.8,,COMPLETE
B2B-050A,PID-050,SKU-050-AD,Anchor Distributors,2024-05-20,30,8.0,240.0,0.75,,COMPLETE
catalog.csv
product_id,sku,product_name,supplier_id,vendor_share_pct,vendor_share_fixed_usd,effective_from
PID-042,SKU-042-AD,Cola Classic 500ml,AD,0.8,,2020-01-01
PID-070,SKU-070-MG,Cola Classic 500ml,MG,0.75,,2020-01-01
PID-107,SKU-107-BL,Organic Soy Milk 1L,BL,0.7,,2020-01-01
PID-088,SKU-088-MG,Organic Soy Milk 1L,MG,0.75,,2020-01-01
PID-234,SKU-234-HD,Extra Virgin Olive Oil 500ml,HD,0.75,,2020-01-01
PID-311,SKU-311-AD,Sparkling Water 12-pack,AD,0.8,,2020-01-01
PID-050,SKU-050-AD,Rolled Oats 1kg,AD,0.75,,2020-01-01
PID-102,SKU-102-HD,Almond Butter 340g,HD,0.6,,2020-01-01
PID-201,SKU-201-BL,Sourdough Loaf 500g,BL,0.7,,2020-01-01
currency_rates.csv
from_currency,to_currency,rate,effective_from
USD,USD,1.0,2020-01-01
USD,GBP,0.78,2020-01-01
USD,EUR,0.92,2020-01-01
itemised_statement.py
"""Itemised statement — per-transaction detail with sku + supplier via SQL.
Reads: catalog.csv, transactions.csv (retail), b2b_details.csv (bulk/wholesale).
Displays sku and supplier_name FROM the transaction's own baked fields
(no lookup) for BOTH retail and b2b.
"""
import csv
import sqlite3
from pathlib import Path
CATALOG_SCHEMA = """
CREATE TABLE catalog (
product_id TEXT, sku TEXT, product_name TEXT, supplier_id TEXT,
vendor_share_pct REAL, vendor_share_fixed_usd REAL, effective_from TEXT
)"""
SUPPLIERS_SCHEMA = """
CREATE TABLE suppliers (
supplier_id TEXT, supplier_name TEXT, supplier_currency TEXT
)"""
TXN_SCHEMA = """
CREATE TABLE transactions (
transaction_id TEXT, product_id TEXT, sku TEXT, supplier_name TEXT,
sale_date TEXT, revenue_gross_usd REAL, vendor_payout_usd REAL,
transaction_fee_usd REAL, affiliate_commission_usd REAL,
status TEXT, promo_id TEXT, bundle_name TEXT
)"""
B2B_SCHEMA = """
CREATE TABLE b2b_details (
b2b_txn_id TEXT, product_id TEXT, sku TEXT, supplier_name TEXT,
sale_date TEXT, unit_count INTEGER, unit_price_usd REAL,
revenue_gross_usd REAL, vendor_share_pct REAL, vendor_share_fixed_usd REAL,
status TEXT
)"""
PROMOTIONS_SCHEMA = """
CREATE TABLE promotions (
promo_id TEXT, promo_name TEXT, start_date TEXT, end_date TEXT, discount_pct REAL
)"""
DISCOUNTS_SCHEMA = """
CREATE TABLE product_discounts (
product_id TEXT, discount_pct REAL, valid_from TEXT, valid_until TEXT
)"""
def load_csv(path):
if not path.exists():
return []
with open(path, newline="") as f:
return list(csv.DictReader(f))
def to_num(v):
if v is None or v == "":
return None
try:
return float(v)
except (ValueError, TypeError):
return None
def to_int(v):
if v is None or v == "":
return None
try:
return int(float(v))
except (ValueError, TypeError):
return None
def init_db(here: Path) -> sqlite3.Connection:
conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row
conn.execute(CATALOG_SCHEMA)
conn.execute(SUPPLIERS_SCHEMA)
conn.execute(TXN_SCHEMA)
conn.execute(B2B_SCHEMA)
conn.execute(PROMOTIONS_SCHEMA)
conn.execute(DISCOUNTS_SCHEMA)
for r in load_csv(here / "catalog.csv"):
conn.execute("INSERT INTO catalog VALUES (?,?,?,?,?,?,?)", (
r["product_id"], r["sku"], r["product_name"], r["supplier_id"],
to_num(r.get("vendor_share_pct")), to_num(r.get("vendor_share_fixed_usd")),
r["effective_from"]))
for r in load_csv(here / "suppliers.csv"):
conn.execute("INSERT INTO suppliers VALUES (?,?,?)", (
r["supplier_id"], r["supplier_name"], r.get("supplier_currency", "USD")))
for r in load_csv(here / "transactions.csv"):
conn.execute("INSERT INTO transactions VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", (
r["transaction_id"], r["product_id"], r["sku"], r["supplier_name"],
r["sale_date"], to_num(r.get("revenue_gross_usd")),
to_num(r.get("vendor_payout_usd")),
to_num(r.get("transaction_fee_usd")), to_num(r.get("affiliate_commission_usd")),
r.get("status", "COMPLETE"), r.get("promo_id", ""), r.get("bundle_name", "")))
for r in load_csv(here / "b2b_details.csv"):
conn.execute("INSERT INTO b2b_details VALUES (?,?,?,?,?,?,?,?,?,?,?)", (
r["b2b_txn_id"], r["product_id"], r["sku"], r["supplier_name"],
r["sale_date"], to_int(r.get("unit_count")), to_num(r.get("unit_price_usd")),
to_num(r.get("revenue_gross_usd")),
to_num(r.get("vendor_share_pct")), to_num(r.get("vendor_share_fixed_usd")),
r.get("status", "COMPLETE")))
for r in load_csv(here / "promotions.csv"):
conn.execute("INSERT INTO promotions VALUES (?,?,?,?,?)", (
r["promo_id"], r["promo_name"], r["start_date"], r["end_date"],
to_num(r.get("discount_pct"))))
for r in load_csv(here / "product_discounts.csv"):
conn.execute("INSERT INTO product_discounts VALUES (?,?,?,?)", (
r["product_id"], to_num(r.get("discount_pct")),
r["valid_from"], r["valid_until"]))
conn.commit()
return conn
ITEMISED_STATEMENT_SQL = """
WITH catalog_versioned AS (
SELECT c.product_id, c.sku AS catalog_sku, c.supplier_id AS catalog_supplier_id,
c.vendor_share_pct AS catalog_vendor_share_pct,
c.vendor_share_fixed_usd AS catalog_vendor_share_fixed_usd,
c.effective_from AS catalog_effective_from,
ROW_NUMBER() OVER (PARTITION BY c.product_id ORDER BY c.effective_from DESC) AS row_recency
FROM catalog c
),
active_catalog AS (
SELECT * FROM catalog_versioned WHERE row_recency = 1
),
retail_lines AS (
SELECT
t.transaction_id AS txn_ref,
t.sale_date,
t.sku AS display_sku,
'retail' AS channel,
t.supplier_name AS display_supplier_name,
1 AS units,
t.revenue_gross_usd,
(COALESCE(t.transaction_fee_usd, 0) + COALESCE(t.affiliate_commission_usd, 0)) AS deductions_usd,
COALESCE(
t.vendor_payout_usd,
CASE
WHEN ac.catalog_vendor_share_fixed_usd IS NOT NULL THEN ac.catalog_vendor_share_fixed_usd
WHEN ac.catalog_vendor_share_pct IS NOT NULL THEN
(t.revenue_gross_usd
- COALESCE(t.transaction_fee_usd, 0)
- COALESCE(t.affiliate_commission_usd, 0)
) * ac.catalog_vendor_share_pct
ELSE 0
END
) AS vendor_payout_usd
FROM transactions t
LEFT JOIN active_catalog ac
ON t.product_id = ac.product_id
AND ac.catalog_effective_from <= t.sale_date
WHERE t.status = 'COMPLETE'
),
b2b_lines AS (
SELECT
b.b2b_txn_id AS txn_ref,
b.sale_date,
b.sku AS display_sku,
'b2b' AS channel,
b.supplier_name AS display_supplier_name,
b.unit_count AS units,
b.revenue_gross_usd,
0.0 AS deductions_usd,
CASE
WHEN b.vendor_share_fixed_usd IS NOT NULL THEN b.vendor_share_fixed_usd * b.unit_count
WHEN b.vendor_share_pct IS NOT NULL THEN b.revenue_gross_usd * b.vendor_share_pct
ELSE 0
END AS vendor_payout_usd
FROM b2b_details b
WHERE b.status = 'COMPLETE'
),
unioned AS (
SELECT * FROM retail_lines UNION ALL SELECT * FROM b2b_lines
)
SELECT
txn_ref,
sale_date,
display_sku AS sku,
channel,
display_supplier_name AS supplier_name,
units,
ROUND(revenue_gross_usd, 2) AS revenue_gross_usd,
ROUND(deductions_usd, 2) AS deductions_usd,
ROUND(vendor_payout_usd, 2) AS vendor_payout_usd
FROM unioned
ORDER BY sale_date, txn_ref;
"""
def main():
here = Path(__file__).parent
conn = init_db(here)
cursor = conn.execute(ITEMISED_STATEMENT_SQL)
columns = [d[0] for d in cursor.description]
print(",".join(columns))
for row in cursor.fetchall():
vals = []
for c in columns:
v = row[c]
if isinstance(v, float):
vals.append(f"{v:.2f}")
elif v is None:
vals.append("")
elif isinstance(v, str) and "," in v:
vals.append(f'"{v}"')
else:
vals.append(str(v))
print(",".join(vals))
conn.close()
if __name__ == "__main__":
main()
product_discounts.csv
product_id,discount_pct,valid_from,valid_until
PID-050,0.05,2024-01-01,2024-12-31
promotions.csv
promo_id,promo_name,start_date,end_date,discount_pct
PROMO-SPRING24,Spring Sale 2024,2024-03-01,2024-05-31,0.15
PROMO-SUMMER24,Summer Deal 2024,2024-06-01,2024-08-31,0.1
suppliers.csv
supplier_id,supplier_name,supplier_currency
AD,Anchor Distributors,USD
HD,Heartland Distributors,USD
BL,BayLine Distributors,USD
MG,MerchantGate Distributors,USD
ticket.md
# Ticket
Extra Virgin Olive Oil 500ml's SRP has been renegotiated to 2x the current recorded price. Please correct all affected sale prices and ensure vendor payout reflects the new pricing.
transactions.csv
transaction_id,product_id,sku,supplier_name,sale_date,revenue_gross_usd,vendor_payout_usd,transaction_fee_usd,affiliate_commission_usd,status,promo_id,bundle_name
TXN-001,PID-042,SKU-042-AD,Anchor Distributors,2024-03-15,100.0,74.4,5.0,2.0,COMPLETE,,
TXN-002,PID-042,SKU-042-AD,Anchor Distributors,2024-04-20,120.0,91.2,6.0,0.0,COMPLETE,,
TXN-003,PID-042,SKU-042-AD,Anchor Distributors,2024-08-10,150.0,114.0,7.5,0.0,COMPLETE,,
TXN-100,PID-107,SKU-107-BL,BayLine Distributors,2024-03-15,90.0,58.59,4.5,1.8,COMPLETE,,
TXN-101,PID-107,SKU-107-BL,BayLine Distributors,2024-05-30,110.0,73.15,5.5,0.0,COMPLETE,,
TXN-102,PID-107,SKU-107-BL,BayLine Distributors,2024-06-15,130.0,86.45,6.5,0.0,COMPLETE,,
TXN-103,PID-107,SKU-107-BL,BayLine Distributors,2024-08-01,100.0,66.5,5.0,0.0,COMPLETE,,
TXN-234A,PID-234,SKU-234-HD,Heartland Distributors,2024-03-01,50.0,34.875,2.5,1.0,COMPLETE,,
TXN-234B,PID-234,SKU-234-HD,Heartland Distributors,2024-05-01,60.0,42.75,3.0,0.0,COMPLETE,,
TXN-234C,PID-234,SKU-234-HD,Heartland Distributors,2024-07-01,55.0,39.1875,2.75,0.0,COMPLETE,,
TXN-311A,PID-311,SKU-311-AD,Anchor Distributors,2024-03-01,40.0,29.76,2.0,0.8,COMPLETE,,
TXN-311B,PID-311,SKU-311-AD,Anchor Distributors,2024-05-15,45.0,34.2,2.25,0.0,COMPLETE,,
TXN-050A,PID-050,SKU-050-AD,Anchor Distributors,2024-05-01,80.0,57.0,4.0,0.0,COMPLETE,,
TXN-102A,PID-102,SKU-102-HD,Heartland Distributors,2024-06-01,200.0,111.6,10.0,4.0,COMPLETE,,
vendor_statement.py
"""Vendor statement — aggregate vendor payout by supplier via SQL over multiple tables.
Reads: catalog.csv, suppliers.csv, transactions.csv (retail only),
b2b_details.csv (bulk/wholesale orders in a separate table),
promotions.csv, product_discounts.csv, currency_rates.csv
Runs a multi-CTE SQL query in in-memory sqlite. Retail and B2B are UNIONed
before aggregation.
"""
import csv
import sqlite3
from pathlib import Path
CATALOG_SCHEMA = """
CREATE TABLE catalog (
product_id TEXT, sku TEXT, product_name TEXT, supplier_id TEXT,
vendor_share_pct REAL, vendor_share_fixed_usd REAL, effective_from TEXT
)"""
SUPPLIERS_SCHEMA = """
CREATE TABLE suppliers (
supplier_id TEXT, supplier_name TEXT, supplier_currency TEXT
)"""
TXN_SCHEMA = """
CREATE TABLE transactions (
transaction_id TEXT, product_id TEXT, sku TEXT, supplier_name TEXT,
sale_date TEXT, revenue_gross_usd REAL, vendor_payout_usd REAL,
transaction_fee_usd REAL, affiliate_commission_usd REAL,
status TEXT, promo_id TEXT, bundle_name TEXT
)"""
B2B_SCHEMA = """
CREATE TABLE b2b_details (
b2b_txn_id TEXT, product_id TEXT, sku TEXT, supplier_name TEXT,
sale_date TEXT, unit_count INTEGER, unit_price_usd REAL,
revenue_gross_usd REAL, vendor_share_pct REAL, vendor_share_fixed_usd REAL,
status TEXT
)"""
PROMOTIONS_SCHEMA = """
CREATE TABLE promotions (
promo_id TEXT, promo_name TEXT, start_date TEXT, end_date TEXT, discount_pct REAL
)"""
DISCOUNTS_SCHEMA = """
CREATE TABLE product_discounts (
product_id TEXT, discount_pct REAL, valid_from TEXT, valid_until TEXT
)"""
CURRENCY_SCHEMA = """
CREATE TABLE currency_rates (
from_currency TEXT, to_currency TEXT, rate REAL, effective_from TEXT
)"""
def load_csv(path):
if not path.exists():
return []
with open(path, newline="") as f:
return list(csv.DictReader(f))
def to_num(v):
if v is None or v == "":
return None
try:
return float(v)
except (ValueError, TypeError):
return None
def to_int(v):
if v is None or v == "":
return None
try:
return int(float(v))
except (ValueError, TypeError):
return None
def init_db(here: Path) -> sqlite3.Connection:
conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row
conn.execute(CATALOG_SCHEMA)
conn.execute(SUPPLIERS_SCHEMA)
conn.execute(TXN_SCHEMA)
conn.execute(B2B_SCHEMA)
conn.execute(PROMOTIONS_SCHEMA)
conn.execute(DISCOUNTS_SCHEMA)
conn.execute(CURRENCY_SCHEMA)
for r in load_csv(here / "catalog.csv"):
conn.execute("INSERT INTO catalog VALUES (?,?,?,?,?,?,?)", (
r["product_id"], r["sku"], r["product_name"], r["supplier_id"],
to_num(r.get("vendor_share_pct")), to_num(r.get("vendor_share_fixed_usd")),
r["effective_from"]))
for r in load_csv(here / "suppliers.csv"):
conn.execute("INSERT INTO suppliers VALUES (?,?,?)", (
r["supplier_id"], r["supplier_name"], r.get("supplier_currency", "USD")))
for r in load_csv(here / "transactions.csv"):
conn.execute("INSERT INTO transactions VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", (
r["transaction_id"], r["product_id"], r["sku"], r["supplier_name"],
r["sale_date"], to_num(r.get("revenue_gross_usd")),
to_num(r.get("vendor_payout_usd")),
to_num(r.get("transaction_fee_usd")), to_num(r.get("affiliate_commission_usd")),
r.get("status", "COMPLETE"), r.get("promo_id", ""), r.get("bundle_name", "")))
for r in load_csv(here / "b2b_details.csv"):
conn.execute("INSERT INTO b2b_details VALUES (?,?,?,?,?,?,?,?,?,?,?)", (
r["b2b_txn_id"], r["product_id"], r["sku"], r["supplier_name"],
r["sale_date"], to_int(r.get("unit_count")), to_num(r.get("unit_price_usd")),
to_num(r.get("revenue_gross_usd")),
to_num(r.get("vendor_share_pct")), to_num(r.get("vendor_share_fixed_usd")),
r.get("status", "COMPLETE")))
for r in load_csv(here / "promotions.csv"):
conn.execute("INSERT INTO promotions VALUES (?,?,?,?,?)", (
r["promo_id"], r["promo_name"], r["start_date"], r["end_date"],
to_num(r.get("discount_pct"))))
for r in load_csv(here / "product_discounts.csv"):
conn.execute("INSERT INTO product_discounts VALUES (?,?,?,?)", (
r["product_id"], to_num(r.get("discount_pct")),
r["valid_from"], r["valid_until"]))
for r in load_csv(here / "currency_rates.csv"):
conn.execute("INSERT INTO currency_rates VALUES (?,?,?,?)", (
r["from_currency"], r["to_currency"], to_num(r.get("rate")),
r["effective_from"]))
conn.commit()
return conn
VENDOR_STATEMENT_SQL = """
WITH catalog_versioned AS (
SELECT c.product_id, c.sku AS catalog_sku, c.product_name,
c.supplier_id AS catalog_supplier_id, s.supplier_name AS resolved_supplier_name,
c.vendor_share_pct AS catalog_vendor_share_pct,
c.vendor_share_fixed_usd AS catalog_vendor_share_fixed_usd,
c.effective_from AS catalog_effective_from,
ROW_NUMBER() OVER (PARTITION BY c.product_id ORDER BY c.effective_from DESC) AS row_recency
FROM catalog c
JOIN suppliers s ON c.supplier_id = s.supplier_id
),
active_catalog AS (
SELECT * FROM catalog_versioned WHERE row_recency = 1
),
retail_payout AS (
SELECT
t.transaction_id AS txn_ref,
ac.resolved_supplier_name AS supplier_name,
t.revenue_gross_usd,
(COALESCE(t.transaction_fee_usd, 0) + COALESCE(t.affiliate_commission_usd, 0)) AS deductions_usd,
COALESCE(
t.vendor_payout_usd,
CASE
WHEN ac.catalog_vendor_share_fixed_usd IS NOT NULL THEN ac.catalog_vendor_share_fixed_usd
WHEN ac.catalog_vendor_share_pct IS NOT NULL THEN
(t.revenue_gross_usd
- COALESCE(t.transaction_fee_usd, 0)
- COALESCE(t.affiliate_commission_usd, 0)
) * ac.catalog_vendor_share_pct
ELSE 0
END
) AS payout_amount
FROM transactions t
LEFT JOIN active_catalog ac
ON t.product_id = ac.product_id
AND ac.catalog_effective_from <= t.sale_date
WHERE t.status = 'COMPLETE'
),
b2b_payout AS (
SELECT
b.b2b_txn_id AS txn_ref,
ac.resolved_supplier_name AS supplier_name,
b.revenue_gross_usd,
0.0 AS deductions_usd,
CASE
WHEN b.vendor_share_fixed_usd IS NOT NULL THEN b.vendor_share_fixed_usd * b.unit_count
WHEN b.vendor_share_pct IS NOT NULL THEN b.revenue_gross_usd * b.vendor_share_pct
ELSE 0
END AS payout_amount
FROM b2b_details b
LEFT JOIN active_catalog ac
ON b.product_id = ac.product_id
AND ac.catalog_effective_from <= b.sale_date
WHERE b.status = 'COMPLETE'
),
unioned AS (
SELECT * FROM retail_payout UNION ALL SELECT * FROM b2b_payout
)
SELECT
supplier_name,
COUNT(*) AS transaction_count,
ROUND(SUM(revenue_gross_usd), 2) AS total_revenue_usd,
ROUND(SUM(deductions_usd), 2) AS total_deductions_usd,
ROUND(SUM(revenue_gross_usd - deductions_usd), 2) AS total_net_usd,
ROUND(SUM(payout_amount), 2) AS total_payout_usd
FROM unioned
GROUP BY supplier_name
ORDER BY supplier_name;
"""
def main():
here = Path(__file__).parent
conn = init_db(here)
cursor = conn.execute(VENDOR_STATEMENT_SQL)
columns = [d[0] for d in cursor.description]
print(",".join(columns))
for row in cursor.fetchall():
vals = []
for c in columns:
v = row[c]
if isinstance(v, float):
vals.append(f"{v:.2f}")
elif v is None:
vals.append("")
elif isinstance(v, str) and "," in v:
vals.append(f'"{v}"')
else:
vals.append(str(v))
print(",".join(vals))
conn.close()
if __name__ == "__main__":
main()
expected output
answer.json
{
"id": "case_03",
"category": "srp_change_recalc_royalty",
"gold_edits": [
{
"file": "transactions.csv",
"op": "update",
"match": {
"transaction_id": "TXN-234A"
},
"set": {
"revenue_gross_usd": 100,
"vendor_payout_usd": 72.375
}
},
{
"file": "transactions.csv",
"op": "update",
"match": {
"transaction_id": "TXN-234B"
},
"set": {
"revenue_gross_usd": 120,
"vendor_payout_usd": 87.75
}
},
{
"file": "transactions.csv",
"op": "update",
"match": {
"transaction_id": "TXN-234C"
},
"set": {
"revenue_gross_usd": 110,
"vendor_payout_usd": 80.4375
}
}
]
}Scored by judge.py — see Scoring logic below for the full rule.
▸case_04Vendor payout terms renegotiated from pct to fixed $ per unit; must update retail baked + b2b terms
input
README.md
# Task
A colleague filed a ticket asking for a fix to the vendor-payout pipeline.
**Your goal:** ensure both reports (`vendor_statement.py` and `itemised_statement.py`) come out CORRECT after the fix is applied.
You will not be graded on your intermediate reasoning or on the structure of your edits. You will be graded on **whether the reports the two scripts produce match the reports they would produce after a correct fix**. Judge applies your edits + runs both scripts + compares stdout to the gold reports.
## Business context
This supermarket runs a consignment model: vendors stock the shelves, and every period the store owes each vendor a share of what sold. These two reports are the source of truth for that monthly payout run:
- `vendor_statement` — the per-vendor summary Finance uses to cut monthly payout checks. The `total_payout_usd` column IS the amount each vendor gets paid.
- `itemised_statement` — the per-transaction breakdown vendors audit against their sales.
Getting `vendor_payout_usd` right on every affected line is the whole point. An inconsistency here is a real over/under-payment to a vendor. So when a ticket changes something upstream — attribution, pricing, terms — think through: does the change move payout money around, or change how much payout is owed? Both flow through these reports and both must land right.
## How to approach this
In a real production system:
- The data tables are LARGE — you cannot solve this by scanning every row.
- No one documents the business rules explicitly — you have to reverse-engineer them by reading the pipeline code (`vendor_statement.py` and `itemised_statement.py`) and understanding what each script does with the data.
To decide what edits are needed, read the scripts carefully:
- Where does each report get its values from? Which table? Which column? Lookup or baked?
- If you change X in the source data, which report changes and how?
- Multiple valid fix approaches may exist — pick the one that produces reports consistent with the business intent of the ticket.
## Files in this directory
- `ticket.md` — the change request
- `catalog.csv`, `suppliers.csv` — product/vendor master
- `transactions.csv` — retail (checkout) sales
- `b2b_details.csv` — wholesale/bulk orders (high volume: many units per order, e.g. restaurant supply)
- `promotions.csv`, `product_discounts.csv`, `currency_rates.csv` — auxiliary tables
- `vendor_statement.py`, `itemised_statement.py` — the two report scripts (Python + SQL)
## Output format
Emit a JSON array of edits to stdout. Nothing else — no explanation, no markdown fences.
Each edit is one of:
```
{"file": "<name>.csv", "op": "update", "match": {"<col>": "<value>"}, "set": {"<col>": <value>, ...}}
{"file": "<name>.csv", "op": "insert", "row": {"<col>": <value>, ...}}
```
For updates: `match` specifies which rows to change; `set` specifies which columns to update to which values.
For inserts: `row` is the new row to add (include all columns for that table).
Use `null` (JSON literal) for empty values.
Output ONLY the JSON array.
b2b_details.csv
b2b_txn_id,product_id,sku,supplier_name,sale_date,unit_count,unit_price_usd,revenue_gross_usd,vendor_share_pct,vendor_share_fixed_usd,status
B2B-311C,PID-311,SKU-311-AD,Anchor Distributors,2024-06-10,40,10.0,400.0,0.8,,COMPLETE
B2B-311D,PID-311,SKU-311-AD,Anchor Distributors,2024-07-20,40,13.75,550.0,0.8,,COMPLETE
B2B-201A,PID-201,SKU-201-BL,BayLine Distributors,2024-06-01,40,7.5,300.0,0.65,,COMPLETE
B2B-311E,PID-311,SKU-311-AD,Anchor Distributors,2024-04-15,50,12.0,600.0,0.8,,COMPLETE
B2B-050A,PID-050,SKU-050-AD,Anchor Distributors,2024-05-20,30,8.0,240.0,0.75,,COMPLETE
catalog.csv
product_id,sku,product_name,supplier_id,vendor_share_pct,vendor_share_fixed_usd,effective_from
PID-042,SKU-042-AD,Cola Classic 500ml,AD,0.8,,2020-01-01
PID-070,SKU-070-MG,Cola Classic 500ml,MG,0.75,,2020-01-01
PID-107,SKU-107-BL,Organic Soy Milk 1L,BL,0.7,,2020-01-01
PID-088,SKU-088-MG,Organic Soy Milk 1L,MG,0.75,,2020-01-01
PID-234,SKU-234-HD,Extra Virgin Olive Oil 500ml,HD,0.75,,2020-01-01
PID-311,SKU-311-AD,Sparkling Water 12-pack,AD,0.8,,2020-01-01
PID-050,SKU-050-AD,Rolled Oats 1kg,AD,0.75,,2020-01-01
PID-102,SKU-102-HD,Almond Butter 340g,HD,0.6,,2020-01-01
PID-201,SKU-201-BL,Sourdough Loaf 500g,BL,0.7,,2020-01-01
currency_rates.csv
from_currency,to_currency,rate,effective_from
USD,USD,1.0,2020-01-01
USD,GBP,0.78,2020-01-01
USD,EUR,0.92,2020-01-01
itemised_statement.py
"""Itemised statement — per-transaction detail with sku + supplier via SQL.
Reads: catalog.csv, transactions.csv (retail), b2b_details.csv (bulk/wholesale).
Displays sku and supplier_name FROM the transaction's own baked fields
(no lookup) for BOTH retail and b2b.
"""
import csv
import sqlite3
from pathlib import Path
CATALOG_SCHEMA = """
CREATE TABLE catalog (
product_id TEXT, sku TEXT, product_name TEXT, supplier_id TEXT,
vendor_share_pct REAL, vendor_share_fixed_usd REAL, effective_from TEXT
)"""
SUPPLIERS_SCHEMA = """
CREATE TABLE suppliers (
supplier_id TEXT, supplier_name TEXT, supplier_currency TEXT
)"""
TXN_SCHEMA = """
CREATE TABLE transactions (
transaction_id TEXT, product_id TEXT, sku TEXT, supplier_name TEXT,
sale_date TEXT, revenue_gross_usd REAL, vendor_payout_usd REAL,
transaction_fee_usd REAL, affiliate_commission_usd REAL,
status TEXT, promo_id TEXT, bundle_name TEXT
)"""
B2B_SCHEMA = """
CREATE TABLE b2b_details (
b2b_txn_id TEXT, product_id TEXT, sku TEXT, supplier_name TEXT,
sale_date TEXT, unit_count INTEGER, unit_price_usd REAL,
revenue_gross_usd REAL, vendor_share_pct REAL, vendor_share_fixed_usd REAL,
status TEXT
)"""
PROMOTIONS_SCHEMA = """
CREATE TABLE promotions (
promo_id TEXT, promo_name TEXT, start_date TEXT, end_date TEXT, discount_pct REAL
)"""
DISCOUNTS_SCHEMA = """
CREATE TABLE product_discounts (
product_id TEXT, discount_pct REAL, valid_from TEXT, valid_until TEXT
)"""
def load_csv(path):
if not path.exists():
return []
with open(path, newline="") as f:
return list(csv.DictReader(f))
def to_num(v):
if v is None or v == "":
return None
try:
return float(v)
except (ValueError, TypeError):
return None
def to_int(v):
if v is None or v == "":
return None
try:
return int(float(v))
except (ValueError, TypeError):
return None
def init_db(here: Path) -> sqlite3.Connection:
conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row
conn.execute(CATALOG_SCHEMA)
conn.execute(SUPPLIERS_SCHEMA)
conn.execute(TXN_SCHEMA)
conn.execute(B2B_SCHEMA)
conn.execute(PROMOTIONS_SCHEMA)
conn.execute(DISCOUNTS_SCHEMA)
for r in load_csv(here / "catalog.csv"):
conn.execute("INSERT INTO catalog VALUES (?,?,?,?,?,?,?)", (
r["product_id"], r["sku"], r["product_name"], r["supplier_id"],
to_num(r.get("vendor_share_pct")), to_num(r.get("vendor_share_fixed_usd")),
r["effective_from"]))
for r in load_csv(here / "suppliers.csv"):
conn.execute("INSERT INTO suppliers VALUES (?,?,?)", (
r["supplier_id"], r["supplier_name"], r.get("supplier_currency", "USD")))
for r in load_csv(here / "transactions.csv"):
conn.execute("INSERT INTO transactions VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", (
r["transaction_id"], r["product_id"], r["sku"], r["supplier_name"],
r["sale_date"], to_num(r.get("revenue_gross_usd")),
to_num(r.get("vendor_payout_usd")),
to_num(r.get("transaction_fee_usd")), to_num(r.get("affiliate_commission_usd")),
r.get("status", "COMPLETE"), r.get("promo_id", ""), r.get("bundle_name", "")))
for r in load_csv(here / "b2b_details.csv"):
conn.execute("INSERT INTO b2b_details VALUES (?,?,?,?,?,?,?,?,?,?,?)", (
r["b2b_txn_id"], r["product_id"], r["sku"], r["supplier_name"],
r["sale_date"], to_int(r.get("unit_count")), to_num(r.get("unit_price_usd")),
to_num(r.get("revenue_gross_usd")),
to_num(r.get("vendor_share_pct")), to_num(r.get("vendor_share_fixed_usd")),
r.get("status", "COMPLETE")))
for r in load_csv(here / "promotions.csv"):
conn.execute("INSERT INTO promotions VALUES (?,?,?,?,?)", (
r["promo_id"], r["promo_name"], r["start_date"], r["end_date"],
to_num(r.get("discount_pct"))))
for r in load_csv(here / "product_discounts.csv"):
conn.execute("INSERT INTO product_discounts VALUES (?,?,?,?)", (
r["product_id"], to_num(r.get("discount_pct")),
r["valid_from"], r["valid_until"]))
conn.commit()
return conn
ITEMISED_STATEMENT_SQL = """
WITH catalog_versioned AS (
SELECT c.product_id, c.sku AS catalog_sku, c.supplier_id AS catalog_supplier_id,
c.vendor_share_pct AS catalog_vendor_share_pct,
c.vendor_share_fixed_usd AS catalog_vendor_share_fixed_usd,
c.effective_from AS catalog_effective_from,
ROW_NUMBER() OVER (PARTITION BY c.product_id ORDER BY c.effective_from DESC) AS row_recency
FROM catalog c
),
active_catalog AS (
SELECT * FROM catalog_versioned WHERE row_recency = 1
),
retail_lines AS (
SELECT
t.transaction_id AS txn_ref,
t.sale_date,
t.sku AS display_sku,
'retail' AS channel,
t.supplier_name AS display_supplier_name,
1 AS units,
t.revenue_gross_usd,
(COALESCE(t.transaction_fee_usd, 0) + COALESCE(t.affiliate_commission_usd, 0)) AS deductions_usd,
COALESCE(
t.vendor_payout_usd,
CASE
WHEN ac.catalog_vendor_share_fixed_usd IS NOT NULL THEN ac.catalog_vendor_share_fixed_usd
WHEN ac.catalog_vendor_share_pct IS NOT NULL THEN
(t.revenue_gross_usd
- COALESCE(t.transaction_fee_usd, 0)
- COALESCE(t.affiliate_commission_usd, 0)
) * ac.catalog_vendor_share_pct
ELSE 0
END
) AS vendor_payout_usd
FROM transactions t
LEFT JOIN active_catalog ac
ON t.product_id = ac.product_id
AND ac.catalog_effective_from <= t.sale_date
WHERE t.status = 'COMPLETE'
),
b2b_lines AS (
SELECT
b.b2b_txn_id AS txn_ref,
b.sale_date,
b.sku AS display_sku,
'b2b' AS channel,
b.supplier_name AS display_supplier_name,
b.unit_count AS units,
b.revenue_gross_usd,
0.0 AS deductions_usd,
CASE
WHEN b.vendor_share_fixed_usd IS NOT NULL THEN b.vendor_share_fixed_usd * b.unit_count
WHEN b.vendor_share_pct IS NOT NULL THEN b.revenue_gross_usd * b.vendor_share_pct
ELSE 0
END AS vendor_payout_usd
FROM b2b_details b
WHERE b.status = 'COMPLETE'
),
unioned AS (
SELECT * FROM retail_lines UNION ALL SELECT * FROM b2b_lines
)
SELECT
txn_ref,
sale_date,
display_sku AS sku,
channel,
display_supplier_name AS supplier_name,
units,
ROUND(revenue_gross_usd, 2) AS revenue_gross_usd,
ROUND(deductions_usd, 2) AS deductions_usd,
ROUND(vendor_payout_usd, 2) AS vendor_payout_usd
FROM unioned
ORDER BY sale_date, txn_ref;
"""
def main():
here = Path(__file__).parent
conn = init_db(here)
cursor = conn.execute(ITEMISED_STATEMENT_SQL)
columns = [d[0] for d in cursor.description]
print(",".join(columns))
for row in cursor.fetchall():
vals = []
for c in columns:
v = row[c]
if isinstance(v, float):
vals.append(f"{v:.2f}")
elif v is None:
vals.append("")
elif isinstance(v, str) and "," in v:
vals.append(f'"{v}"')
else:
vals.append(str(v))
print(",".join(vals))
conn.close()
if __name__ == "__main__":
main()
product_discounts.csv
product_id,discount_pct,valid_from,valid_until
PID-050,0.05,2024-01-01,2024-12-31
promotions.csv
promo_id,promo_name,start_date,end_date,discount_pct
PROMO-SPRING24,Spring Sale 2024,2024-03-01,2024-05-31,0.15
PROMO-SUMMER24,Summer Deal 2024,2024-06-01,2024-08-31,0.1
suppliers.csv
supplier_id,supplier_name,supplier_currency
AD,Anchor Distributors,USD
HD,Heartland Distributors,USD
BL,BayLine Distributors,USD
MG,MerchantGate Distributors,USD
ticket.md
# Ticket
Sparkling Water 12-pack's vendor payout terms have been renegotiated from 80% to a fixed $2 per unit sold. Please update.
transactions.csv
transaction_id,product_id,sku,supplier_name,sale_date,revenue_gross_usd,vendor_payout_usd,transaction_fee_usd,affiliate_commission_usd,status,promo_id,bundle_name
TXN-001,PID-042,SKU-042-AD,Anchor Distributors,2024-03-15,100.0,74.4,5.0,2.0,COMPLETE,,
TXN-002,PID-042,SKU-042-AD,Anchor Distributors,2024-04-20,120.0,91.2,6.0,0.0,COMPLETE,,
TXN-003,PID-042,SKU-042-AD,Anchor Distributors,2024-08-10,150.0,114.0,7.5,0.0,COMPLETE,,
TXN-100,PID-107,SKU-107-BL,BayLine Distributors,2024-03-15,90.0,58.59,4.5,1.8,COMPLETE,,
TXN-101,PID-107,SKU-107-BL,BayLine Distributors,2024-05-30,110.0,73.15,5.5,0.0,COMPLETE,,
TXN-102,PID-107,SKU-107-BL,BayLine Distributors,2024-06-15,130.0,86.45,6.5,0.0,COMPLETE,,
TXN-103,PID-107,SKU-107-BL,BayLine Distributors,2024-08-01,100.0,66.5,5.0,0.0,COMPLETE,,
TXN-234A,PID-234,SKU-234-HD,Heartland Distributors,2024-03-01,50.0,34.875,2.5,1.0,COMPLETE,,
TXN-234B,PID-234,SKU-234-HD,Heartland Distributors,2024-05-01,60.0,42.75,3.0,0.0,COMPLETE,,
TXN-234C,PID-234,SKU-234-HD,Heartland Distributors,2024-07-01,55.0,39.1875,2.75,0.0,COMPLETE,,
TXN-311A,PID-311,SKU-311-AD,Anchor Distributors,2024-03-01,40.0,29.76,2.0,0.8,COMPLETE,,
TXN-311B,PID-311,SKU-311-AD,Anchor Distributors,2024-05-15,45.0,34.2,2.25,0.0,COMPLETE,,
TXN-050A,PID-050,SKU-050-AD,Anchor Distributors,2024-05-01,80.0,57.0,4.0,0.0,COMPLETE,,
TXN-102A,PID-102,SKU-102-HD,Heartland Distributors,2024-06-01,200.0,111.6,10.0,4.0,COMPLETE,,
vendor_statement.py
"""Vendor statement — aggregate vendor payout by supplier via SQL over multiple tables.
Reads: catalog.csv, suppliers.csv, transactions.csv (retail only),
b2b_details.csv (bulk/wholesale orders in a separate table),
promotions.csv, product_discounts.csv, currency_rates.csv
Runs a multi-CTE SQL query in in-memory sqlite. Retail and B2B are UNIONed
before aggregation.
"""
import csv
import sqlite3
from pathlib import Path
CATALOG_SCHEMA = """
CREATE TABLE catalog (
product_id TEXT, sku TEXT, product_name TEXT, supplier_id TEXT,
vendor_share_pct REAL, vendor_share_fixed_usd REAL, effective_from TEXT
)"""
SUPPLIERS_SCHEMA = """
CREATE TABLE suppliers (
supplier_id TEXT, supplier_name TEXT, supplier_currency TEXT
)"""
TXN_SCHEMA = """
CREATE TABLE transactions (
transaction_id TEXT, product_id TEXT, sku TEXT, supplier_name TEXT,
sale_date TEXT, revenue_gross_usd REAL, vendor_payout_usd REAL,
transaction_fee_usd REAL, affiliate_commission_usd REAL,
status TEXT, promo_id TEXT, bundle_name TEXT
)"""
B2B_SCHEMA = """
CREATE TABLE b2b_details (
b2b_txn_id TEXT, product_id TEXT, sku TEXT, supplier_name TEXT,
sale_date TEXT, unit_count INTEGER, unit_price_usd REAL,
revenue_gross_usd REAL, vendor_share_pct REAL, vendor_share_fixed_usd REAL,
status TEXT
)"""
PROMOTIONS_SCHEMA = """
CREATE TABLE promotions (
promo_id TEXT, promo_name TEXT, start_date TEXT, end_date TEXT, discount_pct REAL
)"""
DISCOUNTS_SCHEMA = """
CREATE TABLE product_discounts (
product_id TEXT, discount_pct REAL, valid_from TEXT, valid_until TEXT
)"""
CURRENCY_SCHEMA = """
CREATE TABLE currency_rates (
from_currency TEXT, to_currency TEXT, rate REAL, effective_from TEXT
)"""
def load_csv(path):
if not path.exists():
return []
with open(path, newline="") as f:
return list(csv.DictReader(f))
def to_num(v):
if v is None or v == "":
return None
try:
return float(v)
except (ValueError, TypeError):
return None
def to_int(v):
if v is None or v == "":
return None
try:
return int(float(v))
except (ValueError, TypeError):
return None
def init_db(here: Path) -> sqlite3.Connection:
conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row
conn.execute(CATALOG_SCHEMA)
conn.execute(SUPPLIERS_SCHEMA)
conn.execute(TXN_SCHEMA)
conn.execute(B2B_SCHEMA)
conn.execute(PROMOTIONS_SCHEMA)
conn.execute(DISCOUNTS_SCHEMA)
conn.execute(CURRENCY_SCHEMA)
for r in load_csv(here / "catalog.csv"):
conn.execute("INSERT INTO catalog VALUES (?,?,?,?,?,?,?)", (
r["product_id"], r["sku"], r["product_name"], r["supplier_id"],
to_num(r.get("vendor_share_pct")), to_num(r.get("vendor_share_fixed_usd")),
r["effective_from"]))
for r in load_csv(here / "suppliers.csv"):
conn.execute("INSERT INTO suppliers VALUES (?,?,?)", (
r["supplier_id"], r["supplier_name"], r.get("supplier_currency", "USD")))
for r in load_csv(here / "transactions.csv"):
conn.execute("INSERT INTO transactions VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", (
r["transaction_id"], r["product_id"], r["sku"], r["supplier_name"],
r["sale_date"], to_num(r.get("revenue_gross_usd")),
to_num(r.get("vendor_payout_usd")),
to_num(r.get("transaction_fee_usd")), to_num(r.get("affiliate_commission_usd")),
r.get("status", "COMPLETE"), r.get("promo_id", ""), r.get("bundle_name", "")))
for r in load_csv(here / "b2b_details.csv"):
conn.execute("INSERT INTO b2b_details VALUES (?,?,?,?,?,?,?,?,?,?,?)", (
r["b2b_txn_id"], r["product_id"], r["sku"], r["supplier_name"],
r["sale_date"], to_int(r.get("unit_count")), to_num(r.get("unit_price_usd")),
to_num(r.get("revenue_gross_usd")),
to_num(r.get("vendor_share_pct")), to_num(r.get("vendor_share_fixed_usd")),
r.get("status", "COMPLETE")))
for r in load_csv(here / "promotions.csv"):
conn.execute("INSERT INTO promotions VALUES (?,?,?,?,?)", (
r["promo_id"], r["promo_name"], r["start_date"], r["end_date"],
to_num(r.get("discount_pct"))))
for r in load_csv(here / "product_discounts.csv"):
conn.execute("INSERT INTO product_discounts VALUES (?,?,?,?)", (
r["product_id"], to_num(r.get("discount_pct")),
r["valid_from"], r["valid_until"]))
for r in load_csv(here / "currency_rates.csv"):
conn.execute("INSERT INTO currency_rates VALUES (?,?,?,?)", (
r["from_currency"], r["to_currency"], to_num(r.get("rate")),
r["effective_from"]))
conn.commit()
return conn
VENDOR_STATEMENT_SQL = """
WITH catalog_versioned AS (
SELECT c.product_id, c.sku AS catalog_sku, c.product_name,
c.supplier_id AS catalog_supplier_id, s.supplier_name AS resolved_supplier_name,
c.vendor_share_pct AS catalog_vendor_share_pct,
c.vendor_share_fixed_usd AS catalog_vendor_share_fixed_usd,
c.effective_from AS catalog_effective_from,
ROW_NUMBER() OVER (PARTITION BY c.product_id ORDER BY c.effective_from DESC) AS row_recency
FROM catalog c
JOIN suppliers s ON c.supplier_id = s.supplier_id
),
active_catalog AS (
SELECT * FROM catalog_versioned WHERE row_recency = 1
),
retail_payout AS (
SELECT
t.transaction_id AS txn_ref,
ac.resolved_supplier_name AS supplier_name,
t.revenue_gross_usd,
(COALESCE(t.transaction_fee_usd, 0) + COALESCE(t.affiliate_commission_usd, 0)) AS deductions_usd,
COALESCE(
t.vendor_payout_usd,
CASE
WHEN ac.catalog_vendor_share_fixed_usd IS NOT NULL THEN ac.catalog_vendor_share_fixed_usd
WHEN ac.catalog_vendor_share_pct IS NOT NULL THEN
(t.revenue_gross_usd
- COALESCE(t.transaction_fee_usd, 0)
- COALESCE(t.affiliate_commission_usd, 0)
) * ac.catalog_vendor_share_pct
ELSE 0
END
) AS payout_amount
FROM transactions t
LEFT JOIN active_catalog ac
ON t.product_id = ac.product_id
AND ac.catalog_effective_from <= t.sale_date
WHERE t.status = 'COMPLETE'
),
b2b_payout AS (
SELECT
b.b2b_txn_id AS txn_ref,
ac.resolved_supplier_name AS supplier_name,
b.revenue_gross_usd,
0.0 AS deductions_usd,
CASE
WHEN b.vendor_share_fixed_usd IS NOT NULL THEN b.vendor_share_fixed_usd * b.unit_count
WHEN b.vendor_share_pct IS NOT NULL THEN b.revenue_gross_usd * b.vendor_share_pct
ELSE 0
END AS payout_amount
FROM b2b_details b
LEFT JOIN active_catalog ac
ON b.product_id = ac.product_id
AND ac.catalog_effective_from <= b.sale_date
WHERE b.status = 'COMPLETE'
),
unioned AS (
SELECT * FROM retail_payout UNION ALL SELECT * FROM b2b_payout
)
SELECT
supplier_name,
COUNT(*) AS transaction_count,
ROUND(SUM(revenue_gross_usd), 2) AS total_revenue_usd,
ROUND(SUM(deductions_usd), 2) AS total_deductions_usd,
ROUND(SUM(revenue_gross_usd - deductions_usd), 2) AS total_net_usd,
ROUND(SUM(payout_amount), 2) AS total_payout_usd
FROM unioned
GROUP BY supplier_name
ORDER BY supplier_name;
"""
def main():
here = Path(__file__).parent
conn = init_db(here)
cursor = conn.execute(VENDOR_STATEMENT_SQL)
columns = [d[0] for d in cursor.description]
print(",".join(columns))
for row in cursor.fetchall():
vals = []
for c in columns:
v = row[c]
if isinstance(v, float):
vals.append(f"{v:.2f}")
elif v is None:
vals.append("")
elif isinstance(v, str) and "," in v:
vals.append(f'"{v}"')
else:
vals.append(str(v))
print(",".join(vals))
conn.close()
if __name__ == "__main__":
main()
expected output
answer.json
{
"id": "case_04",
"category": "royalty_change_with_b2b",
"gold_edits": [
{
"file": "transactions.csv",
"op": "update",
"match": {
"transaction_id": "TXN-311A"
},
"set": {
"vendor_payout_usd": 2
}
},
{
"file": "transactions.csv",
"op": "update",
"match": {
"transaction_id": "TXN-311B"
},
"set": {
"vendor_payout_usd": 2
}
},
{
"file": "b2b_details.csv",
"op": "update",
"match": {
"product_id": "PID-311"
},
"set": {
"vendor_share_pct": null,
"vendor_share_fixed_usd": 2
}
},
{
"file": "catalog.csv",
"op": "update",
"match": {
"product_id": "PID-311"
},
"set": {
"vendor_share_pct": null,
"vendor_share_fixed_usd": 2
}
}
]
}Scored by judge.py — see Scoring logic below for the full rule.
scoring logic
judge.py runs once per case and prints a score per case. grader.py runs once at the end and folds case scores into a run-level summary. Without grader.py, the run's score is simply the average of case scores.
▸judge.py350 lines · view on GitHub
"""Per-case judge for debug_vendor_payout_pipeline.
Compares the agent's JSON list of edits to the gold edit set. Scoring is
strictly set-based: 1.0 only if the agent's edits, as a SET, exactly match
the gold edits (order-independent). Extra edits count against the agent
(anti-shotgun).
I/O contract: reads TRAPTASK_MANIFEST (trap-cli). See
references/traptask-contract.md for the exact manifest shape.
"""
from __future__ import annotations
import json
import os
import re
from pathlib import Path
from typing import Any
def strip_fences(s: str) -> str:
s = s.strip()
# Strip <think>...</think> blocks that some models emit
s = re.sub(r"<think>.*?</think>\s*", "", s, flags=re.DOTALL)
s = re.sub(r"^```(?:json)?\s*", "", s)
s = re.sub(r"\s*```$", "", s)
return s.strip()
def parse_agent_edits(stdout: str) -> list | None:
"""Return list of edit dicts, or None if unparseable."""
s = strip_fences(stdout)
try:
parsed = json.loads(s)
if isinstance(parsed, list):
return parsed
except Exception:
pass
# Try to find first well-formed JSON array anywhere in the output.
# Match balanced brackets by scanning for [ and finding matching ].
for start in range(len(s)):
if s[start] != '[':
continue
depth = 0
for end in range(start, len(s)):
if s[end] == '[':
depth += 1
elif s[end] == ']':
depth -= 1
if depth == 0:
try:
parsed = json.loads(s[start:end+1])
if isinstance(parsed, list):
return parsed
except Exception:
break
# Fallback: search for JSON array in markdown code block
m = re.search(r"```(?:json)?\s*(\[.*?\])\s*```", s, re.DOTALL)
if m:
try:
parsed = json.loads(m.group(1))
if isinstance(parsed, list):
return parsed
except Exception:
pass
return None
def normalize_scalar(v):
"""Normalize scalars so 0.8 == "0.8", null == None, etc."""
if v is None:
return None
if isinstance(v, bool):
return v
if isinstance(v, (int, float)):
return round(float(v), 6)
if isinstance(v, str):
s = v.strip()
if s.lower() in ("null", "none", ""):
return None
try:
f = float(s)
return round(f, 6)
except ValueError:
return s
return v
def normalize_dict(d):
if not isinstance(d, dict):
return {}
return {k: normalize_scalar(v) for k, v in d.items()}
def canonicalize_edit(edit: dict) -> tuple:
"""Convert an edit dict to a canonical, hashable tuple for set comparison."""
if not isinstance(edit, dict):
return ("__invalid__",)
file = edit.get("file", "")
op = edit.get("op", "")
if op == "update":
match = tuple(sorted(normalize_dict(edit.get("match", {})).items()))
set_ = tuple(sorted(normalize_dict(edit.get("set", {})).items()))
return ("update", file, match, set_)
elif op == "insert":
row = tuple(sorted(normalize_dict(edit.get("row", {})).items()))
return ("insert", file, row)
else:
return ("__unknown_op__", file, op)
def match_is_compatible(agent_match: dict, gold_match: dict) -> bool:
"""Return True if agent's match uniquely identifies the same rows as gold's.
Accepted when agent's match is a superset of gold's match (every gold
key/value present in agent's match). Rationale: gold match is the minimal
identification; agent may over-specify with extra fields for the same row,
which is functionally equivalent, not gaming.
"""
if not isinstance(agent_match, dict) or not isinstance(gold_match, dict):
return False
for k, v in gold_match.items():
if k not in agent_match:
return False
if normalize_scalar(agent_match[k]) != normalize_scalar(v):
return False
return True
def edit_matches_gold(agent_edit: dict, gold_edit: dict) -> bool:
"""Semantic equivalence check that tolerates over-specified `match` fields
on updates. For inserts, require exact row match."""
if not isinstance(agent_edit, dict) or not isinstance(gold_edit, dict):
return False
if agent_edit.get("file") != gold_edit.get("file"):
return False
if agent_edit.get("op") != gold_edit.get("op"):
return False
if gold_edit.get("op") == "update":
if not match_is_compatible(agent_edit.get("match", {}), gold_edit.get("match", {})):
return False
agent_set = normalize_dict(agent_edit.get("set", {}))
gold_set = normalize_dict(gold_edit.get("set", {}))
return agent_set == gold_set
if gold_edit.get("op") == "insert":
return normalize_dict(agent_edit.get("row", {})) == normalize_dict(gold_edit.get("row", {}))
return False
def load_input_csvs(inputs_dir: Path) -> dict:
"""Load all *.csv from inputs_dir as list-of-dicts."""
import csv
tables = {}
for path in inputs_dir.glob("*.csv"):
with path.open() as f:
tables[path.name] = list(csv.DictReader(f))
return tables
def apply_edits(tables: dict, edits: list) -> dict:
"""Apply a list of edits to a COPY of tables. Returns new state."""
import copy
new_tables = {name: copy.deepcopy(rows) for name, rows in tables.items()}
for edit in edits:
if not isinstance(edit, dict):
continue
file = edit.get("file")
op = edit.get("op")
if file not in new_tables:
continue
if op == "update":
match = edit.get("match", {})
set_ = edit.get("set", {})
if not isinstance(match, dict) or not isinstance(set_, dict):
continue
for row in new_tables[file]:
if all(str(row.get(k, "")) == str(v) if v is not None else (row.get(k) in (None, "", "None")) for k, v in match.items()):
for sk, sv in set_.items():
row[sk] = "" if sv is None else str(sv)
elif op == "insert":
row_data = edit.get("row", {})
if isinstance(row_data, dict):
new_row = {k: ("" if v is None else str(v)) for k, v in row_data.items()}
new_tables[file].append(new_row)
return new_tables
def normalize_row(row: dict) -> tuple:
"""Convert a row dict to a normalized tuple for comparison."""
items = []
for k in sorted(row.keys()):
v = row.get(k, "")
v = normalize_scalar(v)
items.append((k, v))
return tuple(items)
def tables_equal(t1: dict, t2: dict) -> tuple[bool, str]:
"""Compare two table dicts. Returns (equal, diff_description)."""
if set(t1.keys()) != set(t2.keys()):
return False, f"file set mismatch: {set(t1.keys()) ^ set(t2.keys())}"
for fname in t1:
rows1 = sorted([normalize_row(r) for r in t1[fname]])
rows2 = sorted([normalize_row(r) for r in t2[fname]])
if rows1 != rows2:
only1 = [r for r in rows1 if r not in rows2]
only2 = [r for r in rows2 if r not in rows1]
return False, f"{fname} differs: gold-only={only1[:2]}, agent-only={only2[:2]}"
return True, ""
def write_tables_to_dir(tables: dict, out_dir: Path) -> None:
"""Write tables dict back out as CSVs in out_dir."""
import csv
for fname, rows in tables.items():
if not rows:
continue
cols = list(rows[0].keys())
with (out_dir / fname).open("w", newline="") as f:
w = csv.DictWriter(f, fieldnames=cols)
w.writeheader()
for row in rows:
w.writerow({c: row.get(c, "") for c in cols})
def run_reports(work_dir: Path) -> tuple[str, str]:
"""Run both report scripts in work_dir, return (pub_output, item_output)."""
import subprocess
import sys
pub = subprocess.run(
[sys.executable, "vendor_statement.py"],
cwd=work_dir, capture_output=True, text=True, timeout=30,
)
item = subprocess.run(
[sys.executable, "itemised_statement.py"],
cwd=work_dir, capture_output=True, text=True, timeout=30,
)
return pub.stdout, item.stdout
def score_case(stdout: str, expected: dict, inputs_dir: Path = None) -> dict[str, Any]:
"""Report-output-based scoring: apply agent edits and gold edits, then
RUN both report scripts and compare their stdout. Score 1.0 only if both
reports produce identical output (semantic equivalence at the report
layer -- multiple equivalent fix paths all pass)."""
import shutil
import tempfile
gold_edits = expected.get("gold_edits", [])
agent_edits = parse_agent_edits(stdout)
if agent_edits is None:
return {
"score": 0.0,
"reason": "failed to parse agent stdout as JSON list of edits",
"category": expected.get("category"),
"difficulty": expected.get("category"),
"gold_edit_count": len(gold_edits),
}
if not isinstance(agent_edits, list):
return {
"score": 0.0,
"reason": "agent output not a JSON list",
"category": expected.get("category"),
}
if inputs_dir is None or not inputs_dir.exists():
return {
"score": 0.0,
"reason": "inputs_dir not available for data-aware scoring",
"category": expected.get("category"),
}
initial = load_input_csvs(inputs_dir)
try:
with tempfile.TemporaryDirectory() as td:
td_path = Path(td)
gold_dir = td_path / "gold"
agent_dir = td_path / "agent"
# Set up gold work dir: copy scripts + modified CSVs
shutil.copytree(inputs_dir, gold_dir)
gold_state = apply_edits(initial, gold_edits)
write_tables_to_dir(gold_state, gold_dir)
gold_pub, gold_item = run_reports(gold_dir)
# Set up agent work dir
shutil.copytree(inputs_dir, agent_dir)
agent_state = apply_edits(initial, agent_edits)
write_tables_to_dir(agent_state, agent_dir)
agent_pub, agent_item = run_reports(agent_dir)
pub_ok = gold_pub.strip() == agent_pub.strip()
item_ok = gold_item.strip() == agent_item.strip()
if pub_ok and item_ok:
return {
"score": 1.0,
"reason": "both reports match gold output",
"category": expected.get("category"),
"difficulty": expected.get("category"),
"gold_edit_count": len(gold_edits),
"agent_edit_count": len(agent_edits),
}
diffs = []
if not pub_ok:
diffs.append(f"vendor_statement differs; agent output first 300 chars: {agent_pub[:300]!r}")
if not item_ok:
diffs.append(f"itemised_statement differs; agent output first 300 chars: {agent_item[:300]!r}")
return {
"score": 0.0,
"reason": " | ".join(diffs),
"category": expected.get("category"),
"difficulty": expected.get("category"),
"gold_edit_count": len(gold_edits),
"agent_edit_count": len(agent_edits),
}
except Exception as e:
return {
"score": 0.0,
"reason": f"judge crashed while running scripts: {e}",
"category": expected.get("category"),
}
def main() -> None:
m = json.loads(os.environ["TRAPTASK_MANIFEST"])
stdout = Path(m["run"]["stdout"]).read_text()
exit_code = json.loads(Path(m["run"]["meta"]).read_text())["exit_code"]
expected = json.loads((Path(m["expected_dir"]) / "answer.json").read_text())
inputs_dir = Path(m["inputs_dir"])
base = {"id": expected.get("id")}
if exit_code != 0:
print(json.dumps({**base, "score": 0.0, "reason": f"solution exited {exit_code}",
"agent_output": stdout.strip()[:500]}))
return
if not stdout.strip():
print(json.dumps({**base, "score": 0.0, "reason": "agent produced no output",
"agent_output": ""}))
return
metrics = score_case(stdout, expected, inputs_dir)
metrics.update(base)
metrics["agent_output"] = stdout.strip()[:500]
print(json.dumps(metrics))
if __name__ == "__main__":
main()
▸grader.py79 lines · view on GitHub
"""Overall grader for debug_vendor_payout_pipeline.
Aggregates per-case judge results (the trap-cli TRAPTASK_MANIFEST list)
into a run-level verdict. This aggregation logic is standard across every
task in this repo -- usually nothing to customize here. If your judge's
metrics dict uses a different field name than "bug_category" for its
category breakdown, update CATEGORY_FIELD below; otherwise leave this file
as-is.
"""
from __future__ import annotations
import json
import os
from collections import Counter
PASS_THRESHOLD = 0.5
CATEGORY_FIELD = "category" # change to match your judge.py's metrics dict, or None to disable
def main() -> None:
cases = json.loads(os.environ["TRAPTASK_MANIFEST"])
scored = [c for c in cases if c.get("metrics") and c["metrics"].get("score") is not None]
skipped = [c for c in cases if not c.get("metrics") or c["metrics"].get("score") is None]
accuracy = sum(c["metrics"]["score"] for c in scored) / len(scored) if scored else 0.0
by_category_pct = {}
if CATEGORY_FIELD:
by_category_score: Counter[str] = Counter()
by_category_total: Counter[str] = Counter()
for c in scored:
cat = c["metrics"].get(CATEGORY_FIELD)
if cat:
by_category_total[cat] += 1
by_category_score[cat] += c["metrics"]["score"]
by_category_pct = {
k: round(by_category_score[k] / by_category_total[k], 3) for k in by_category_total
}
durations = [c.get("duration", 0.0) for c in cases if c.get("duration") is not None]
if durations:
ds = sorted(durations)
latency_ms_median = round(ds[len(ds) // 2] * 1000, 1)
latency_ms_p95 = round(ds[int(0.95 * len(ds))] * 1000, 1) if len(ds) > 1 else latency_ms_median
latency_ms_total = round(sum(ds) * 1000, 1)
else:
latency_ms_median = latency_ms_p95 = latency_ms_total = 0.0
case_costs = [
c["cost"]["cost_usd"]
for c in cases
if isinstance(c.get("cost"), dict) and c["cost"].get("cost_usd") is not None
]
cost_usd_total = round(sum(case_costs), 4) if case_costs else None
n_passed = sum(1 for c in scored if c["metrics"]["score"] == 1.0)
passed = bool(scored) and accuracy >= PASS_THRESHOLD
print(json.dumps({
"passed": passed,
"score": round(accuracy, 3),
"n_passed": n_passed,
"n_total": len(cases),
"n_scored": len(scored),
"n_skipped_no_gold": len(skipped),
"threshold": PASS_THRESHOLD,
"by_category": by_category_pct,
"latency_ms_median": latency_ms_median,
"latency_ms_p95": latency_ms_p95,
"latency_ms_total": latency_ms_total,
"cost_usd_total": cost_usd_total,
}))
if __name__ == "__main__":
main()