Service performance disparity review

{
  "cutoff": "2026-09-15T00:00:00Z",
  "late_target": 0.1,
  "min_sample": 15,
  "review_gap_pp": 5,
  "priority_gap_pp": 15,
  "standard_weight": 0.5,
  "express_weight": 0.5
}

Data dictionary and decision rules

Assignment brief

You join the cold-chain fulfilment operations team. The operations lead needs to decide which orders and hubs to review first, where export controls failed, and what the next operations workbench must deliver. The four hubs are Harbour, Eastgate, Westhill and Southbay. Standard service promises delivery within 24 hours and express within 8 hours. Use the actual due_at value in orders.csv for each order.

The review cutoff is 2026-09-15T00:00:00Z, defined in rules.json. Convert every business timestamp to UTC. Records can contain events after the cutoff so that you can test whether a historical review accidentally uses future information.

Orders

One row in orders.csv is one order. order_id is a unique text identifier; keep its leading zeros. hub is the operating hub, service is standard or express, and route identifies the route. created_at and due_at are timezone-aware ISO 8601 timestamps. Counting event rows is not a valid way to count orders.

Event feeds and canonical schema

Feed A maps event_id to event_id, order_ref to order_id, hub to hub, service to service, event to kind, and event_time to happened_at.

Feed B maps id to event_id, shipment.number to order_id, shipment.station to hub, shipment.product to service, and at_ms to happened_at. at_ms is an integer Unix timestamp in milliseconds. Map OPEN, COLLECT, DROP and VOID to created, picked_up, delivered and cancelled respectively.

The canonical event contains event_id, order_id, hub, service, kind and happened_at. happened_at is UTC ISO 8601 with a Z suffix. Validate required fields, action names and dimensions. Reject timezone-free timestamps and non-integer epochs. Do not change the raw inputs.

Identical canonical content under the same event_id is an exact replay: retain one event and list every removed replay. Different content under one immutable event_id is a conflict: quarantine the entire group. Never accept the first arrival and silently overwrite it later. Keep the source filename, 1-based line number and reason for every rejected record or replay. Reconcile raw records = canonical events + quarantined records + removed replays.

Historical fulfilment cohort

Group events by order_id and retain only events with happened_at <= cutoff. A visible cancellation takes precedence and is excluded from the mature denominator. A non-cancelled order with due_at > cutoff is not_mature. Otherwise, no visible delivery means overdue_open; the first visible delivery after due_at means delivered_late; delivery at or before due_at means on_time. Exactly at the due timestamp is mature, and delivery exactly at that timestamp is on time.

Eligible mature orders = on_time + delivered_late + overdue_open. Late orders = delivered_late + overdue_open. Never classify cancellation or an order not yet due as successful delivery. Keep delivered lateness minutes (delivery minus due_at) separate from open overdue minutes (cutoff minus due_at). The inputs do not contain actual refunds, damaged-goods values or costs, so do not calculate those amounts.

Service review

The fixed late-rate target is 10%, and the minimum mature sample is 15 per hub/service cell. gap_pp = (late_rate - 0.10) * 100, measured in percentage points. Apply sample size first: below 15 is More data needed. A gap above 15 points is Priority review; above 5 and at most 15 is Review needed; at most 5 is Within threshold. These are operational triage rules, not a statistical finding of unfair treatment.

For a comparable service mix, give standard and express equal 50% weights. Calculate a hub's standardized late rate only when both service cells meet the minimum sample. Otherwise retain a missing value and state which cell is insufficient. Do not impute an overall mean or rank a small sample as the best hub. Separate observations, possible explanations and causal claims.

Access logs and policy

One access-log.csv row is one request. request_id is unique; at is UTC. user, session and role are gateway request context. method and path identify the resource; status is the response code; rows and bytes are gateway response counts. These counts do not prove unique customers, file retention or subsequent use.

access-policy.json defines the baseline. An operator can read only the assigned hub and cannot export. A supervisor or approved service role can export. A session lasts 30 minutes after a successful login: exactly 30 minutes is valid, and later requests must be rejected. Failed logins do not renew sessions. A 401 is a denied attempt, not confirmed disclosure. svc-archive is approved, so a frequent large export is not automatically suspicious. The IP ranges are documentation addresses; the logs cannot establish geography, a natural person's identity or the actual access path. Endpoint and VPN evidence would be needed.

"""Reference implementation for the Ape Intern cold-chain operations lab."""
import csv,json
from collections import defaultdict,Counter
from datetime import datetime,timezone
from pathlib import Path
from urllib.parse import urlparse,parse_qs
UTC=timezone.utc
KINDS={'OPEN':'created','COLLECT':'picked_up','DROP':'delivered','VOID':'cancelled'}
HUBS={'Harbour','Eastgate','Westhill','Southbay'}
def parse_time(value):
    t=datetime.fromisoformat(value.replace('Z','+00:00'))
    if t.tzinfo is None:raise ValueError('timezone_required')
    return t.astimezone(UTC)
def iso(t):return t.isoformat(timespec='seconds').replace('+00:00','Z')
def readcsv(p):
    with Path(p).open(encoding='utf-8-sig',newline='') as f:return list(csv.DictReader(f))
def writecsv(p,rows,fields=None):
    rows=list(rows)
    with Path(p).open('w',encoding='utf-8-sig',newline='') as f:
        w=csv.DictWriter(f,fieldnames=fields or list(rows[0]));w.writeheader();w.writerows(rows)
def dump(p,value):Path(p).write_text(json.dumps(value,ensure_ascii=False,indent=2)+'\n')
def _normalize(record,source):
    if source=='A':
        e=dict(event_id=record['event_id'],order_id=record['order_ref'],hub=record['hub'],service=record['service'],kind=record['event'],happened_at=iso(parse_time(record['event_time'])))
    else:
        ms=record['at_ms']
        if type(ms) is not int:raise ValueError('epoch_must_be_integer')
        e=dict(event_id=record['id'],order_id=record['shipment']['number'],hub=record['shipment']['station'],service=record['shipment']['product'],kind=KINDS[record['action']],happened_at=iso(datetime.fromtimestamp(ms/1000,UTC)))
    if e['kind'] not in KINDS.values():raise ValueError('unsupported_event')
    if e['hub'] not in HUBS or e['service'] not in {'standard','express'}:raise ValueError('unknown_dimension')
    if not e['event_id'] or not e['order_id']:raise ValueError('empty_key')
    return e

def normalize(record,source):
    try:return _normalize(record,source)
    except (KeyError,TypeError,OverflowError) as error:raise ValueError(str(error).strip("'")) from error

def ingest(data):
    grouped=defaultdict(list);rejected=[];duplicates=[]
    for source,name in [('A','dispatch-a.jsonl'),('B','courier-b.jsonl')]:
        for line,text in enumerate((data/name).read_text().splitlines(),1):
            loc={'source':name,'line':line}
            try:
                event=normalize(json.loads(text),source);grouped[event['event_id']].append((event,loc))
            except (ValueError,KeyError,TypeError,OverflowError) as ex:
                rejected.append(dict(loc,reason=str(ex).strip("'")))
    accepted=[]
    for eid,items in grouped.items():
        distinct={json.dumps(e,sort_keys=True,ensure_ascii=False) for e,loc in items}
        if len(distinct)>1:
            rejected.extend(dict(loc,reason='immutable_id_conflict') for e,loc in items);continue
        accepted.append(items[0][0])
        duplicates.extend(dict(loc,event_id=eid,reason='exact_replay') for e,loc in items[1:])
    accepted.sort(key=lambda e:(e['happened_at'],e['event_id']))
    return accepted,rejected,duplicates

def classify_order(order,events,cutoff):
    # Only facts available at cutoff can influence a retrospective result.
    visible=[e for e in events if parse_time(e['happened_at'])<=cutoff]
    visible.sort(key=lambda e:parse_time(e['happened_at']))
    delivered=[parse_time(e['happened_at']) for e in visible if e['kind']=='delivered']
    cancelled=any(e['kind']=='cancelled' for e in visible)
    due=parse_time(order['due_at'])
    if cancelled:state='cancelled'
    elif due>cutoff:state='not_mature'
    elif not delivered:state='overdue_open'
    elif min(delivered)>due:state='delivered_late'
    else:state='on_time'
    delay=round((min(delivered)-due).total_seconds()/60) if state=='delivered_late' else 0
    overdue=round((cutoff-due).total_seconds()/60) if state=='overdue_open' else 0
    return dict(order,cohort=state,late_minutes=delay,overdue_minutes=overdue)

def aggregate(rows,keys):
    groups=defaultdict(list)
    for row in rows:groups[tuple(row[k] for k in keys)].append(row)
    result=[]
    for dims,records in sorted(groups.items()):
        counts=Counter(r['cohort'] for r in records)
        eligible=sum(counts[x] for x in ['on_time','delivered_late','overdue_open'])
        late=counts['delivered_late']+counts['overdue_open']
        result.append(dict(zip(keys,dims),total=len(records),eligible=eligible,late=late,late_rate=round(late/eligible,6) if eligible else None,on_time=counts['on_time'],delivered_late=counts['delivered_late'],overdue_open=counts['overdue_open'],cancelled=counts['cancelled'],not_mature=counts['not_mature'],delivered_late_minutes=sum(r['late_minutes'] for r in records),open_overdue_minutes=sum(r['overdue_minutes'] for r in records)))
    return result

def service_band(rate,n,rules):
    if n<rules['min_sample']:return 'More data needed'
    gap=(rate-rules['late_target'])*100
    if gap>rules['priority_gap_pp']:return 'Priority review'
    if gap>rules['review_gap_pp']:return 'Review needed'
    return 'Within threshold'
def fairness(hub_tiers,rules):
    rows=[]
    for x in hub_tiers:
        rate=x['late_rate'];rows.append(dict(hub=x['hub'],service=x['service'],eligible=x['eligible'],late=x['late'],late_rate=rate,target=rules['late_target'],gap_pp=round((rate-rules['late_target'])*100,3) if rate is not None else None,review=service_band(rate,x['eligible'],rules)))
    standardized=[]
    for hub in sorted(HUBS):
        parts=[r for r in rows if r['hub']==hub]
        enough=len(parts)==2 and all(r['eligible']>=rules['min_sample'] for r in parts)
        value=sum(r['late_rate']*rules[r['service']+'_weight'] for r in parts) if enough else None
        standardized.append(dict(hub=hub,standardized_late_rate=round(value,6) if value is not None else None,status='Comparable' if enough else 'More data needed'))
    return rows,standardized

def investigate(logs,policy):
    logins={};findings=[]
    for row in sorted(logs,key=lambda r:(r['at'],r['request_id'])):
        t=parse_time(row['at']);status=int(row['status']);route=urlparse(row['path']);query=parse_qs(route.query)
        if route.path=='/login' and status==200:logins[(row['user'],row['session'])]=t;continue
        if status!=200:continue
        types=[]
        if route.path=='/export' and row['role'] not in policy['export_roles']:types.append('unauthorised_export')
        hub=query.get('hub',[''])[0]
        if row['role']=='operator' and hub and hub!=policy['operator_hub'].get(row['user']):types.append('hub_scope_violation')
        login=logins.get((row['user'],row['session']))
        if login is None or (t-login).total_seconds()>policy['session_ttl_minutes']*60:types.append('expired_or_missing_session')
        if types:findings.append(dict(request_id=row['request_id'],at=row['at'],user=row['user'],session=row['session'],finding=';'.join(types),status=status,rows=int(row['rows']),bytes=int(row['bytes'])))
    return findings

def solve(data,out):
    data=Path(data);out=Path(out);out.mkdir(parents=True,exist_ok=True)
    events,rejects,dups=ingest(data)
    (out/'canonical-events.jsonl').write_text(''.join(json.dumps(e,ensure_ascii=False)+'\n' for e in events))
    writecsv(out/'rejected.csv',rejects);writecsv(out/'duplicates.csv',dups)
    rules=json.loads((data/'rules.json').read_text());cutoff=parse_time(rules['cutoff'])
    by_order=defaultdict(list)
    for e in events:by_order[e['order_id']].append(e)
    orders=readcsv(data/'orders.csv')
    records=[classify_order(o,by_order[o['order_id']],cutoff) for o in orders]
    writecsv(out/'order-review.csv',records)
    hubs=aggregate(records,['hub']);tiers=aggregate(records,['hub','service']);routes=aggregate(records,['route'])
    writecsv(out/'hub-metrics.csv',hubs);writecsv(out/'route-metrics.csv',routes)
    bands,std=fairness(tiers,rules)
    writecsv(out/'service-review.csv',bands);writecsv(out/'standardized-comparison.csv',std)
    findings=investigate(readcsv(data/'access-log.csv'),json.loads((data/'access-policy.json').read_text()))
    writecsv(out/'security-findings.csv',findings)
    summary=dict(orders=len(orders),accepted_events=len(events),rejected=len(rejects),duplicates=len(dups),cohorts=dict(Counter(r['cohort'] for r in records)),hubs=hubs,service_review=bands,standardized=std,security=dict(flagged_requests=len(findings),potential_export_rows=sum(f['rows'] for f in findings if 'unauthorised_export' in f['finding']),potential_export_bytes=sum(f['bytes'] for f in findings if 'unauthorised_export' in f['finding']),findings=findings))
    dump(out/'results.json',summary)
    return summary
if __name__=='__main__':
    import sys
    print(json.dumps(solve(sys.argv[1],sys.argv[2]),ensure_ascii=False,indent=2))