356 lines
16 KiB
Python
356 lines
16 KiB
Python
"""
|
||
Settlement aggregation & classification.
|
||
|
||
Reverse-engineered rule (verified to the penny on Jan-2026):
|
||
* Settlement ids are monotonic (higher = newer).
|
||
* Amazon closes a settlement when the next opens and disburses it via a `Transfer`
|
||
row that is *tagged with the next/open settlement id* and pays out the *prior*
|
||
settlement(s).
|
||
* A settlement is RECEIVABLE until the payout that clears it is *received* in the
|
||
bank by month-end. A disbursement is "received" if its date is on/before
|
||
(month_end - clearing_lag_days); otherwise it is "in transit".
|
||
* Per (marketplace, account_type):
|
||
paid_boundary = max(tagged settlement_id over RECEIVED transfers)
|
||
receivable settlements = those with settlement_id >= paid_boundary
|
||
(equivalently: a settlement is paid iff its id < paid_boundary)
|
||
* Receivable amount = sum of `total` for NON-transfer rows whose account_type is a
|
||
real account (Standard/Invoiced Orders) in receivable settlements. Transfer rows
|
||
are always excluded from the receivable (they are receipts, not receivable).
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from collections import defaultdict
|
||
from dataclasses import dataclass, field
|
||
from datetime import date, timedelta
|
||
from typing import Iterable
|
||
|
||
TRANSFER_TYPE = "Transfer"
|
||
# Bucket label for rows carrying no account type (Amazon leaves it blank on transfers in every
|
||
# marketplace except the USA). Persisted rows must be labelled with this exact string, or the
|
||
# post-classification UPDATE — which matches on (settlement, marketplace, account_type) — silently
|
||
# misses every payout row and leaves its status NULL.
|
||
UNSPECIFIED_ACCOUNT = "(unspecified)"
|
||
# Real order account types that participate in the receivable SUMIFS. "All Orders" is the
|
||
# synthetic single stream for marketplaces whose report has no `account type` column
|
||
# (every marketplace except USA).
|
||
RECEIVABLE_ACCOUNT_TYPES = {"standard orders", "invoiced orders", "all orders"}
|
||
|
||
|
||
def _as_int(settlement_id: str | None) -> int:
|
||
if settlement_id is None:
|
||
return -1
|
||
try:
|
||
return int(str(settlement_id).strip())
|
||
except (ValueError, TypeError):
|
||
# Non-numeric ids: fall back to a stable-ish hash order (rare / never in real data).
|
||
return -1
|
||
|
||
|
||
@dataclass
|
||
class SettlementStat:
|
||
marketplace: str
|
||
account_type: str
|
||
settlement_id: str
|
||
order_total: float = 0.0 # sum of `total` over NON-transfer rows
|
||
row_count: int = 0
|
||
transfer_total: float = 0.0 # sum of `total` over Transfer rows in this settlement
|
||
transfer_count: int = 0
|
||
first_date: date | None = None
|
||
last_date: date | None = None
|
||
type_totals: dict[str, float] = field(default_factory=dict)
|
||
status: str = "receivable" # "paid" | "receivable" (filled by classify)
|
||
|
||
@property
|
||
def sid_int(self) -> int:
|
||
return _as_int(self.settlement_id)
|
||
|
||
|
||
@dataclass
|
||
class Transfer:
|
||
marketplace: str
|
||
account_type: str
|
||
settlement_id: str # the id the transfer row is tagged with
|
||
txn_date: date | None
|
||
amount: float
|
||
source_file: str = ""
|
||
source_row: int = 0
|
||
received: bool = True # filled by classify
|
||
|
||
@property
|
||
def sid_int(self) -> int:
|
||
return _as_int(self.settlement_id)
|
||
|
||
|
||
@dataclass
|
||
class AggregationResult:
|
||
# key = (marketplace, account_type, settlement_id)
|
||
settlements: dict[tuple[str, str, str], SettlementStat] = field(default_factory=dict)
|
||
transfers: list[Transfer] = field(default_factory=list)
|
||
total_rows: int = 0
|
||
uploaded_total: float = 0.0 # sum of `total` over every row (incl. transfers)
|
||
type_totals: dict[str, float] = field(default_factory=dict)
|
||
account_types_seen: set[str] = field(default_factory=set)
|
||
marketplaces_seen: set[str] = field(default_factory=set)
|
||
duplicate_count: int = 0
|
||
duplicate_samples: list[str] = field(default_factory=list)
|
||
# Storage-fee detection (req #8)
|
||
storage_total_by_mkt: dict[str, float] = field(default_factory=dict) # canonical storage rows
|
||
potential_storage_by_mkt: dict[str, float] = field(default_factory=dict) # storage-like text, other type
|
||
undated_count: int = 0
|
||
undated_total: float = 0.0
|
||
undated_samples: list[str] = field(default_factory=list)
|
||
potential_storage_count: int = 0
|
||
potential_storage_samples: list[str] = field(default_factory=list)
|
||
# --- Control C3 (bucket completeness) ------------------------------------
|
||
# Money-carrying order rows the engine cannot place. Both are silently EXCLUDED from the
|
||
# receivable downstream (an unrecognized account type fails the SUMIFS filter; a
|
||
# non-numeric settlement id sorts to -1 and is classified "paid"), so they must be
|
||
# counted here and block the close rather than quietly reduce the number.
|
||
unclassified_acct_count: int = 0
|
||
unclassified_acct_total: float = 0.0
|
||
unclassified_acct_samples: list[str] = field(default_factory=list)
|
||
unclassified_sid_count: int = 0
|
||
unclassified_sid_total: float = 0.0
|
||
unclassified_sid_samples: list[str] = field(default_factory=list)
|
||
|
||
|
||
def _dupe_key(rec: dict) -> int:
|
||
"""Duplicate identity: settlement + type + order + sku + timestamp + amount."""
|
||
return hash((
|
||
rec.get("settlement_id"), rec.get("txn_type"), rec.get("order_id"),
|
||
rec.get("sku"), rec.get("date_time"), round(float(rec.get("total") or 0.0), 4),
|
||
))
|
||
|
||
|
||
def aggregate(records: Iterable[dict], default_marketplace: str = "USA",
|
||
detect_duplicates: bool = True) -> AggregationResult:
|
||
"""Single streaming pass over normalized records -> settlement stats + transfers.
|
||
|
||
Detected duplicates are counted and surfaced (never silently dropped), so overlapping
|
||
date ranges or a file imported twice are reported rather than double-counted-and-hidden.
|
||
"""
|
||
res = AggregationResult()
|
||
settlements = res.settlements
|
||
seen: set[int] | None = set() if detect_duplicates else None
|
||
for rec in records:
|
||
if seen is not None:
|
||
k = _dupe_key(rec)
|
||
if k in seen:
|
||
res.duplicate_count += 1
|
||
if len(res.duplicate_samples) < 25:
|
||
res.duplicate_samples.append(
|
||
f"{rec.get('_source_file')}:{rec.get('_source_row')} "
|
||
f"settlement={rec.get('settlement_id')} order={rec.get('order_id')} "
|
||
f"total={rec.get('total')}"
|
||
)
|
||
else:
|
||
seen.add(k)
|
||
total = float(rec.get("total") or 0.0)
|
||
# Canonical English type when the pipeline resolved one (localized reports).
|
||
txn_type = (rec.get("_type_en") or rec.get("txn_type") or "").strip()
|
||
acct_raw = (rec.get("account_type") or "").strip()
|
||
marketplace = _marketplace_of(rec, default_marketplace)
|
||
sid = rec.get("settlement_id") or ""
|
||
d = rec.get("_date")
|
||
|
||
# Storage-fee tracking (req #8): canonical storage rows per market, plus rows whose
|
||
# description reads storage-like but were booked under another type.
|
||
if rec.get("_storage"):
|
||
if txn_type == "FBA Inventory Fee":
|
||
res.storage_total_by_mkt[marketplace] = (
|
||
res.storage_total_by_mkt.get(marketplace, 0.0) + total)
|
||
elif txn_type != TRANSFER_TYPE:
|
||
res.potential_storage_by_mkt[marketplace] = (
|
||
res.potential_storage_by_mkt.get(marketplace, 0.0) + total)
|
||
res.potential_storage_count += 1
|
||
if len(res.potential_storage_samples) < 25:
|
||
res.potential_storage_samples.append(
|
||
f"{rec.get('_source_file')}:{rec.get('_source_row')} "
|
||
f"[{marketplace}] type={rec.get('txn_type')} total={total}")
|
||
|
||
res.total_rows += 1
|
||
res.uploaded_total += total
|
||
if d is None:
|
||
res.undated_count += 1
|
||
res.undated_total += total
|
||
if len(res.undated_samples) < 10:
|
||
res.undated_samples.append(
|
||
f"{rec.get('_source_file')} row {rec.get('_source_row')} "
|
||
f"type={rec.get('txn_type') or '(blank)'} total={total:,.2f}")
|
||
res.type_totals[txn_type] = res.type_totals.get(txn_type, 0.0) + total
|
||
res.marketplaces_seen.add(marketplace)
|
||
if acct_raw:
|
||
res.account_types_seen.add(acct_raw)
|
||
|
||
# Control C3: an order row carrying money must be placeable in a receivable bucket.
|
||
if txn_type != TRANSFER_TYPE and total:
|
||
if acct_raw.lower() not in RECEIVABLE_ACCOUNT_TYPES:
|
||
res.unclassified_acct_count += 1
|
||
res.unclassified_acct_total += total
|
||
if len(res.unclassified_acct_samples) < 10:
|
||
res.unclassified_acct_samples.append(
|
||
f"{rec.get('_source_file')} row {rec.get('_source_row')} "
|
||
f"[{marketplace}] account type={acct_raw or '(blank)'!r} "
|
||
f"total={total:,.2f}")
|
||
if _as_int(sid) < 0:
|
||
res.unclassified_sid_count += 1
|
||
res.unclassified_sid_total += total
|
||
if len(res.unclassified_sid_samples) < 10:
|
||
res.unclassified_sid_samples.append(
|
||
f"{rec.get('_source_file')} row {rec.get('_source_row')} "
|
||
f"[{marketplace}] settlement id={sid or '(blank)'!r} "
|
||
f"total={total:,.2f}")
|
||
|
||
# Group transfers under the account type they carry (Amazon tags transfers with one).
|
||
acct_key = account_bucket(acct_raw)
|
||
key = (marketplace, acct_key, sid)
|
||
st = settlements.get(key)
|
||
if st is None:
|
||
st = SettlementStat(marketplace, acct_key, sid)
|
||
settlements[key] = st
|
||
|
||
if txn_type == TRANSFER_TYPE:
|
||
st.transfer_total += total
|
||
st.transfer_count += 1
|
||
res.transfers.append(Transfer(
|
||
marketplace=marketplace, account_type=acct_key, settlement_id=sid,
|
||
txn_date=d, amount=total,
|
||
source_file=rec.get("_source_file", ""), source_row=rec.get("_source_row", 0),
|
||
))
|
||
else:
|
||
st.order_total += total
|
||
st.type_totals[txn_type] = st.type_totals.get(txn_type, 0.0) + total
|
||
st.row_count += 1
|
||
if d is not None:
|
||
if st.first_date is None or d < st.first_date:
|
||
st.first_date = d
|
||
if st.last_date is None or d > st.last_date:
|
||
st.last_date = d
|
||
return res
|
||
|
||
|
||
_UNSPEC = UNSPECIFIED_ACCOUNT
|
||
|
||
|
||
def account_bucket(account_type: str | None) -> str:
|
||
"""The bucket label a row is aggregated under — the single definition of that mapping,
|
||
shared by the aggregator and by transaction persistence so the two cannot disagree."""
|
||
return (account_type or "").strip() or UNSPECIFIED_ACCOUNT
|
||
|
||
|
||
def _marketplace_of(rec: dict, default: str) -> str:
|
||
"""
|
||
The uploaded USA files carry marketplace='amazon.com'. We normalize to the
|
||
reporting marketplace label (e.g. 'USA'). Multi-marketplace uploads keep their
|
||
own label so the engine stays generic.
|
||
"""
|
||
return rec.get("_marketplace") or default
|
||
|
||
|
||
@dataclass
|
||
class Classification:
|
||
# (marketplace, account_type) -> paid_boundary settlement_id (int) or None
|
||
boundary: dict[tuple[str, str], int | None] = field(default_factory=dict)
|
||
# (marketplace, account_type) -> the received transfer that defines the boundary
|
||
boundary_transfer: dict[tuple[str, str], Transfer | None] = field(default_factory=dict)
|
||
receivable_settlements: set[tuple[str, str, str]] = field(default_factory=set)
|
||
paid_settlements: set[tuple[str, str, str]] = field(default_factory=set)
|
||
# settlement_id -> owning marketplace (majority of non-transfer rows)
|
||
settlement_owner: dict[str, str] = field(default_factory=dict)
|
||
# bucket keys whose rows belong to a settlement owned by ANOTHER marketplace
|
||
# (cross-market stragglers: excluded from the receivable and surfaced for review)
|
||
cross_market: set[tuple[str, str, str]] = field(default_factory=set)
|
||
|
||
|
||
def classify(
|
||
agg: AggregationResult,
|
||
month_end: date,
|
||
clearing_lag_days: int = 2,
|
||
received_overrides: dict[tuple[str, str, str], bool] | None = None,
|
||
manual_payouts: bool = False,
|
||
) -> Classification:
|
||
"""
|
||
Mark each transfer received/in-transit, derive the paid boundary, and classify every
|
||
settlement paid/receivable.
|
||
|
||
Received status, in priority order:
|
||
1. `received_overrides` — keyed (marketplace, account_type, settlement_id). Built from
|
||
Finance's bank-receipt entries: True iff the money reached the BANK by month-end.
|
||
Amazon's Transfer date is only when the payout was initiated; the bank credit lands
|
||
3-5 working days later, so a receipt is the ground truth and always wins.
|
||
2. No override, manual_payouts=False (auto): the clearing-lag heuristic —
|
||
received iff transfer date ≤ month_end − clearing_lag_days.
|
||
3. No override, manual_payouts=True: NOT received. Finance records every bank credit
|
||
by hand, so a payout without a receipt has, by definition, not been received.
|
||
|
||
Multi-marketplace mechanics (verified against the Jan-2026 workbook):
|
||
* A settlement belongs to the marketplace owning the majority of its non-transfer
|
||
rows. European settlements can span marketplaces (e.g. the Belgium chain carries
|
||
a few amazon.de rows and its payouts appear in the Germany report).
|
||
* Transfers attach to the OWNER of their tagged settlement — never to the file they
|
||
happen to appear in (Amazon repeats account-level transfers across country files).
|
||
* The boundary forms per (owner marketplace, account stream). USA keeps its
|
||
Standard/Invoiced streams; other markets have a single "All Orders" stream, so a
|
||
transfer with no account type bounds that stream.
|
||
* Rows sitting in a DIFFERENT marketplace than their settlement's owner are
|
||
cross-market stragglers: excluded from the receivable and surfaced for review
|
||
(matches the Finance workbook's per-marketplace tabs).
|
||
"""
|
||
cutoff = month_end - timedelta(days=clearing_lag_days)
|
||
overrides = received_overrides or {}
|
||
cls = Classification()
|
||
|
||
# 1) Settlement owner = marketplace with the most non-transfer rows for that id.
|
||
rows_by_sid: dict[str, dict[str, int]] = {}
|
||
for (mkt, _acct, sid), st in agg.settlements.items():
|
||
order_rows = st.row_count - st.transfer_count
|
||
if order_rows > 0:
|
||
rows_by_sid.setdefault(sid, {})[mkt] = (
|
||
rows_by_sid.get(sid, {}).get(mkt, 0) + order_rows)
|
||
for sid, per_mkt in rows_by_sid.items():
|
||
cls.settlement_owner[sid] = max(per_mkt.items(), key=lambda kv: kv[1])[0]
|
||
|
||
def _stream(acct: str) -> str:
|
||
"""Boundary stream for an account label: USA keeps its real streams; blank -> All Orders."""
|
||
return acct if acct.lower() in RECEIVABLE_ACCOUNT_TYPES else "All Orders"
|
||
|
||
# 2) Mark received status; boundary per (owner marketplace, stream).
|
||
received_max: dict[tuple[str, str], int] = {}
|
||
boundary_tx: dict[tuple[str, str], Transfer] = {}
|
||
for t in agg.transfers:
|
||
ok = overrides.get((t.marketplace, t.account_type, t.settlement_id))
|
||
if ok is None:
|
||
ok = False if manual_payouts else (
|
||
t.txn_date is not None and t.txn_date <= cutoff)
|
||
t.received = bool(ok)
|
||
if t.received:
|
||
owner = cls.settlement_owner.get(t.settlement_id, t.marketplace)
|
||
k = (owner, _stream(t.account_type))
|
||
if t.sid_int > received_max.get(k, -1):
|
||
received_max[k] = t.sid_int
|
||
boundary_tx[k] = t
|
||
|
||
for k, sid_int in received_max.items():
|
||
cls.boundary[k] = sid_int
|
||
cls.boundary_transfer[k] = boundary_tx.get(k)
|
||
|
||
# 3) Classify each settlement bucket against its OWNER's boundary.
|
||
for key, st in agg.settlements.items():
|
||
mkt, acct, sid = key
|
||
owner = cls.settlement_owner.get(sid, mkt)
|
||
bkey = (owner, _stream(acct))
|
||
cls.boundary.setdefault(bkey, None)
|
||
boundary = cls.boundary.get(bkey)
|
||
is_receivable = boundary is None or st.sid_int >= boundary
|
||
if owner != mkt and (st.row_count - st.transfer_count) > 0:
|
||
# Cross-market straggler rows: not part of this marketplace's receivable.
|
||
st.status = "cross_market"
|
||
cls.cross_market.add(key)
|
||
elif is_receivable:
|
||
st.status = "receivable"
|
||
cls.receivable_settlements.add(key)
|
||
else:
|
||
st.status = "paid"
|
||
cls.paid_settlements.add(key)
|
||
return cls
|