304 lines
13 KiB
Python
304 lines
13 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"
|
|
# 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)
|
|
|
|
|
|
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)
|
|
|
|
# Group transfers under the account type they carry (Amazon tags transfers with one).
|
|
acct_key = acct_raw or _UNSPEC
|
|
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)"
|
|
|
|
|
|
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,
|
|
) -> Classification:
|
|
"""
|
|
Mark each transfer received/in-transit (auto clearing-lag, with optional overrides
|
|
keyed by (marketplace, account_type, settlement_id)), derive the paid boundary, and
|
|
classify every settlement paid/receivable.
|
|
|
|
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 = (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
|