# 六月订单净收入：修复结果与发布决定

**决定：暂缓发布完整六月净收入；可发布经核实子集，但必须标注不完整。** 已核实 CNY 子集净额为 **¥270.00**（已确认订单 O1、O2，扣除可归属退款 R1、R2）；另有订单 O6 的同版本金额冲突，以及退款 R3 对应的缺失订单 O9，二者均未计入该子集。USD 子集为 **$50.00**（O5，无退款记录）。币种不合并。由于 O9 的资格、月份和订单总额未知，**完整指标无法从现有记录确定，也不能以已知快照冲突给完整指标设上下界**。

## 指标契约与口径

- **决策/粒度：** 管理层是否可发布六月订单净收入；经济键为 `order_id`，不是导出行。按订单快照定状态、月份、客户；退款按唯一 `refund_id` 计一次并归入原订单。
- **月份/时点：** Asia/Shanghai 的 2027-06，即 UTC 半开区间 `[2027-05-31T16:00:00Z, 2027-06-30T16:00:00Z)`。指标截至六月结束；先排除晚于截止时点的快照，再按订单选最高版本。完全相同的导出行只留一份；相同最高版本/时间但金额不同，不自行选胜者。输入时间为 UTC `Z`，计算显式转换时区。
- **措施与人口：** 只计选定快照状态 `paid`；金额按币种、分精度 Decimal；净额 = 已纳入订单毛额 − 可映射到这些订单的唯一退款。`cancelled` 不计。客户维表仅提供订单时点的可选分群，不决定人口；无匹配客户保留并标为未知。
- **新鲜度/限制：** 仅有题给静态记录，没有到达时间、退款发生时间或修订时点；因此可复算所给快照，不能证明源数据完整或新鲜。退款表仅含 ID、订单、金额、币种；未提供退款时间，故仅按题定“六月订单”政策映射，不另造退款月份筛选。

## 复现与已执行结果

下方完整 Python 3 程序仅使用标准库，内嵌题给全部记录，既复现原查询，也执行修复口径、完整性检查、阶段对账和 as-of 客户维表 enrich。运行：`python3 audit.py`（将本文件代码块原样保存为 `audit.py`）。程序对结果作断言。

```python
import csv, io
from collections import Counter, defaultdict
from datetime import datetime, timezone
from decimal import Decimal
from zoneinfo import ZoneInfo

ORDERS = '''id,version,utc,status,currency,amount,customer
O1,1,2027-05-31T16:30:00Z,paid,CNY,100,C1
O1,2,2027-05-31T16:30:00Z,paid,CNY,120,C1
O1,2,2027-05-31T16:30:00Z,paid,CNY,120,C1
O2,1,2027-06-12T00:00:00Z,paid,CNY,200,C2
O3,1,2027-06-15T00:00:00Z,cancelled,CNY,80,C1
O4,1,2027-06-30T16:30:00Z,paid,CNY,300,C1
O5,1,2027-06-20T00:00:00Z,paid,USD,50,C1
O6,2,2027-06-21T00:00:00Z,paid,CNY,40,C1
O6,2,2027-06-21T00:00:00Z,paid,CNY,60,C1'''
REFUNDS = [
    {'refund_id':'R1','order_id':'O1','currency':'CNY','amount':Decimal('20')},
    {'refund_id':'R1','order_id':'O1','currency':'CNY','amount':Decimal('20')},
    {'refund_id':'R2','order_id':'O2','currency':'CNY','amount':Decimal('30')},
    {'refund_id':'R3','order_id':'O9','currency':'CNY','amount':Decimal('10')},
]
DIMS = [
    {'customer':'C1','effective':'2027-05','segment':'SMB'},
    {'customer':'C1','effective':'2027-07','segment':'Enterprise'},
]
TZ = ZoneInfo('Asia/Shanghai')
START = datetime(2027,6,1,tzinfo=TZ)
END = datetime(2027,7,1,tzinfo=TZ)
CUTOFF = END.astimezone(timezone.utc)
def parse(s): return datetime.fromisoformat(s.replace('Z','+00:00'))
def rows(text): return list(csv.DictReader(io.StringIO(text)))
def money(x): return Decimal(x)
def fmt(d): return ', '.join(f'{k}={v:.2f}' for k,v in sorted(d.items())) or '(none)'
def sums(rs):
    out=defaultdict(Decimal)
    for r in rs: out[r['currency']] += r['amount']
    return dict(out)
raw=rows(ORDERS)
for i,r in enumerate(raw,1):
    r['source_row']=f'orders.csv:{i+1}'
    r['version']=int(r['version']); r['instant']=parse(r['utc']); r['amount']=money(r['amount'])

# Exact export repeats versus same-precedence disagreements.
seen=set(); dedup=[]; dup=Counter()
for r in raw:
    signature=tuple(r[k] for k in ('id','version','utc','status','currency','amount','customer'))
    if signature in seen: dup[r['id']]+=1
    else: seen.add(signature); dedup.append(r)
print('EXACT_DUPLICATE_EXTRA_ROWS',dict(dup))

# Literal legacy SQL behavior: UTC text prefix, no status/refund filter, inner join all dimension rows.
legacy=[r for r in raw if r['utc'].startswith('2027-06')]
joined=[]
for r in legacy:
    matches=[d for d in DIMS if d['customer']==r['customer']]
    for d in matches: joined.append((r,d))
legacy_sums=defaultdict(Decimal)
for r,d in joined: legacy_sums[r['currency']]+=r['amount']
print('LEGACY_STAGE_utc_prefix rows=%d keys=%d gross=%s' %
      (len(legacy),len({r['id'] for r in legacy}),fmt(sums(legacy))))
print('LEGACY_STAGE_inner_customer_join rows=%d keys=%d gross=%s' %
      (len(joined),len({r['id'] for r,d in joined}),fmt(legacy_sums)))

# Anti-join checks are scoped to populations and identify actual missing keys.
known_customers={d['customer'] for d in DIMS}
print('CUSTOMER_ANTI_JOIN_legacy_population',sorted({r['customer'] for r in legacy if r['customer'] not in known_customers}))

# Contract: cutoff first, then choose max version; inspect all tied rows, never break value conflicts.
within_cutoff=[r for r in dedup if r['instant']<=CUTOFF]
by_order=defaultdict(list)
for r in within_cutoff: by_order[r['id']].append(r)
selected=[]; conflicts=[]
for oid,group in sorted(by_order.items()):
    maxrank=max((r['version'],r['instant']) for r in group)
    versions=[r for r in group if (r['version'],r['instant'])==maxrank]
    # Version is precedence; tied rows must agree on all economic/status/timestamp fields.
    signatures={(r['instant'],r['status'],r['currency'],r['amount'],r['customer']) for r in versions}
    if len(signatures)>1:
        conflicts.append((oid,versions)); continue
    selected.append(versions[0])
month=[r for r in selected if START<=r['instant'].astimezone(TZ)<END]
eligible=[r for r in month if r['status']=='paid']
conflict_month=[(oid,grp) for oid,grp in conflicts if START<=grp[0]['instant'].astimezone(TZ)<END]
print('CONTRACT cutoff=%s dedup_in_cutoff=%d unambiguous_selected=%d unambiguous_month=%d confirmed_paid=%d unresolved_month_keys=%d' %
      (CUTOFF.isoformat(),len(within_cutoff),len(selected),len(month),len(eligible),len(conflict_month)))
print('TIED_PRECEDENCE_CONFLICTS',[(oid,[(r['source_row'],str(r['amount'])) for r in grp]) for oid,grp in conflicts])
print('ELIGIBLE_ORDER_KEYS',[(r['id'],r['source_row'],r['currency'],str(r['amount'])) for r in eligible])

# Correctly enrich at order instant: latest customer dimension effective on/before order local month.
enriched=[]
for r in eligible:
    ym=r['instant'].astimezone(TZ).strftime('%Y-%m')
    matches=[d for d in DIMS if d['customer']==r['customer'] and d['effective']<=ym]
    segment=max(matches,key=lambda d:d['effective'])['segment'] if matches else 'UNKNOWN'
    enriched.append((r,segment))
print('ASOF_ENRICHED',[(r['id'],segment) for r,segment in enriched])

# Deduplicate refunds by refund_id; conflicting reuse is unresolved, not silently chosen.
refund_by_id=defaultdict(list)
for r in REFUNDS: refund_by_id[r['refund_id']].append(r)
refunds=[]; refund_conflicts=[]
for rid,grp in sorted(refund_by_id.items()):
    sig={(r['order_id'],r['currency'],r['amount']) for r in grp}
    if len(sig)>1: refund_conflicts.append(rid)
    else: refunds.append(grp[0])
paid_keys={r['id']:r for r in eligible}
orphans=[r for r in refunds if r['order_id'] not in paid_keys]
linked=[r for r in refunds if r['order_id'] in paid_keys and paid_keys[r['order_id']]['currency']==r['currency']]
refund_totals=sums(linked)
gross=sums(eligible)
net={c:gross.get(c,Decimal(0))-refund_totals.get(c,Decimal(0)) for c in set(gross)|set(refund_totals)}
print('GROSS_verified_paid',fmt(gross))
print('LINKED_REFUNDS',[(r['refund_id'],r['order_id'],str(r['amount'])) for r in linked],fmt(refund_totals))
print('NET_verified_subset',fmt(net))
print('ORPHAN_REFUNDS',[(r['refund_id'],r['order_id'],r['currency'],str(r['amount'])) for r in orphans])

assert dict(legacy_sums)=={'CNY':Decimal('960'),'USD':Decimal('100')}
assert net=={'CNY':Decimal('270'),'USD':Decimal('50')}
assert [r['id'] for r in eligible]==['O1','O2','O5']
assert [(oid,[r['amount'] for r in grp]) for oid,grp in conflicts]==[('O6',[Decimal('40'),Decimal('60')])]
assert [(r['refund_id'],r['order_id']) for r in orphans]==[('R3','O9')]
```

实际执行输出：

```text
EXACT_DUPLICATE_EXTRA_ROWS {'O1': 1}
LEGACY_STAGE_utc_prefix rows=6 keys=5 gross=CNY=680.00, USD=50.00
LEGACY_STAGE_inner_customer_join rows=10 keys=4 gross=CNY=960.00, USD=100.00
CUSTOMER_ANTI_JOIN_legacy_population ['C2']
CONTRACT cutoff=2027-06-30T16:00:00+00:00 dedup_in_cutoff=7 unambiguous_selected=4 unambiguous_month=4 confirmed_paid=3 unresolved_month_keys=1
TIED_PRECEDENCE_CONFLICTS [('O6', [('orders.csv:9', '40'), ('orders.csv:10', '60')])]
ELIGIBLE_ORDER_KEYS [('O1', 'orders.csv:3', 'CNY', '120'), ('O2', 'orders.csv:5', 'CNY', '200'), ('O5', 'orders.csv:8', 'USD', '50')]
ASOF_ENRICHED [('O1', 'SMB'), ('O2', 'UNKNOWN'), ('O5', 'SMB')]
GROSS_verified_paid CNY=320.00, USD=50.00
LINKED_REFUNDS [('R1', 'O1', '20'), ('R2', 'O2', '30')] CNY=50.00
NET_verified_subset CNY=270.00, USD=50.00
ORPHAN_REFUNDS [('R3', 'O9', 'CNY', '10')]
```

## 缺陷、金额影响与处置

| 检查/证据 | 已核实结果及影响 | 决定性处置 |
|---|---|---|
| 原查询复现（O2–O6；客户连接） | UTC 前缀选中 6 行/5 个订单键，连接后 10 行/4 个键。CNY 从连接前 ¥680 变为 ¥960（多 ¥280）；USD 从 $50 变为 $100（多 $50）。C1 两行维表使 O3、O4、O5、O6 各复制两次；O2/C2 被内连接丢弃。 | 禁止用原查询值发布。维表只做按订单时点的左连接。 |
| UTC 月界/状态与快照 | O1 `2027-05-31 16:30Z` 实为上海六月，最高版 v2 ¥120；旧版 v1 ¥100 不重复计。O4 `2027-06-30 16:30Z` 为上海七月，排除。O3 已取消，排除。 | 按本地月份、截止时点、版本、状态顺序过滤。 |
| 重复与版本冲突 | O1 完全重复导出 1 行，去重。O6 两条 v2、同时间快照分别 ¥40/¥60（`orders.csv:9–10`），不是完全重复；最高优先级仍冲突。未计入核实子集。若 O6 确认为六月已支付，影响毛额及净额各 +¥40 或 +¥60；选择哪一条的差额为 ¥20。 | 查源系统 O6 v2 的权威金额/修订记录后重算，不可任择。 |
| 退款映射 | R1 重复两份但唯一 `refund_id`，仅扣 O1 ¥20；R2 扣 O2 ¥30。核实 CNY 毛额 ¥320、可映射退款 ¥50、净额 ¥270。 | 按 refund_id 去重并确认订单/币种一致。 |
| 孤儿退款 R3 / O9 | R3 ¥10 CNY 映射至完全缺失的 O9（`refunds` 记录 R3）。不从已知订单扣除。O9 的订单毛额、状态、币种及月份皆未知；若属于六月已支付，完整 CNY 净额应在已知子集上另加未知毛额并扣 ¥10，不能只把 ¥10 当完整影响，也不能给完整总额设界。 | 保留为未分配退款；查 O9 原始订单及支付/月份，再决定是否纳入并映射退款。 |
| 未受影响控制 | O5 `orders.csv:8` 为六月已支付 USD 50，保留为 $50；C1 当时分群 SMB。C2 无维表记录仍保留 O2，分群 UNKNOWN。 | 金额按币种呈报；未知维度不剔除订单。 |

**修复优先级：** 1) 查 O9 的原始订单、支付状态/时间及退款 R3 归属；它决定指标完整性，且可能增加未知毛额。2) 查 O6 最高版本权威值。随后按同一契约重跑。当前只能接受“已核实子集”¥270 CNY、$50 USD；完整六月指标 **HOLD**。仅当 O9 完整性和 O6 冲突均消解，且退款映射完整性获来源确认，才可评估完整发布。维表分群不影响金额，不阻碍此决定。