"""Build the annual drinkware business analysis from the single approved fact source.""" import json from pathlib import Path import numpy as np import pandas as pd from release_config import VERSION, EVIDENCE, prepare ROOT = Path(__file__).resolve().parents[1] DATA = ROOT / "research" / "private" PERIOD = "2025-09至2026-08" CHANNEL_LABELS = { "吉拉福": "吉拉福", "自营实物": "直营实物（优选生活、东福超市）", } FLOORS = {"吉拉福": 0.20, "自营实物": 0.30} METRICS = ["amount", "quantity", "gross"] SOURCE_FIELDS = { "年月-订单创建": "month", "频道": "channel", "管理品牌": "brand", "后台一级分类": "category1", "后台二级分类": "category2", "后台三级分类": "category3", "后台四级分类": "category4", "sku_code": "sku", "商品名称": "name", "产品系列-汇总-C端表 (4个)": "series", "卡大类-v2": "card_type", "兑换金额": "amount", "销量": "quantity", "兑换毛利额（含C端运费）": "gross", "兑换毛利率（含C端运费）": "source_margin", } def load_json(path): return json.loads(path.read_text(encoding="utf-8")) def records(frame): if frame.empty: return [] clean = frame.replace([np.inf, -np.inf], np.nan) return json.loads(clean.to_json(orient="records", force_ascii=False)) def facts(filename="P02_年度饮水用具明细.json"): frame = pd.DataFrame(load_json(DATA / filename)) frame = frame.rename(columns=SOURCE_FIELDS) text_fields = [ "month", "channel", "brand", "category1", "category2", "category3", "category4", "sku", "name", "series", "card_type", ] for field in text_fields: frame[field] = frame[field].fillna("").astype(str).str.strip() for field in METRICS + ["source_margin"]: frame[field] = pd.to_numeric(frame[field], errors="coerce").fillna(0) frame["month"] = frame["month"].where(frame["month"].str.match(r"^\d{4}-\d{2}$"), "") frame["unit_value"] = frame["amount"] / frame["quantity"].replace(0, np.nan) return frame def aggregate(frame): total = frame[METRICS].sum().to_dict() total.update( { "rows": int(len(frame)), "skus": int(frame["sku"].nunique()), "brands": int(frame["brand"].replace("", np.nan).nunique()), "margin": float(total["gross"] / total["amount"]) if total["amount"] else None, "unit_value": float(total["amount"] / total["quantity"]) if total["quantity"] else None, } ) return { key: (float(value) if isinstance(value, (np.floating, np.integer)) else value) for key, value in total.items() } def grouped(frame, keys, slots=None): if isinstance(keys, str): keys = [keys] result = frame.groupby(keys, dropna=False)[METRICS].sum().reset_index() result["margin"] = result["gross"] / result["amount"].replace(0, np.nan) result["unit_value"] = result["amount"] / result["quantity"].replace(0, np.nan) if slots is not None: result = result.merge(slots, on=keys, how="left") return result def labelled(frame, key): output = frame.copy() if key == "channel": output["channelLabel"] = output["channel"].map(CHANNEL_LABELS).fillna(output["channel"]) return output def load_listing(): listing = pd.DataFrame(load_json(DATA / "P01_上架清单9.18.json")) for field in ["sku_code", "spu_code", "商品名称", "管理品牌名称", "后台四级分类", "上架频道"]: listing[field] = listing[field].fillna("").astype(str).str.strip() listing["spu_code"] = listing["spu_code"].replace("", np.nan) return listing def safe_date(values): parsed = pd.to_datetime(values, format="mixed", errors="coerce") return parsed.min().strftime("%Y-%m-%d") if parsed.notna().any() else None def main(): from plan_analysis import enrich, export_tables from current_listing import load_current_listing prepare() sales = facts() kitchen = facts("P02_年度厨房用具明细.json") listing, snapshot = load_current_listing(sales) mapping = listing.drop_duplicates("sku_code").set_index("sku_code")["spu_code"].to_dict() sales["spu"] = sales["sku"].map(mapping) sales["channelLabel"] = sales["channel"].map(CHANNEL_LABELS).fillna(sales["channel"]) category_slots = ( listing.groupby("后台四级分类")["spu_code"] .nunique() .rename("currentSlots") .reset_index() .rename(columns={"后台四级分类": "category"}) ) category_slots["currentSlots"] = category_slots["currentSlots"].astype(int) channel_slots = ( listing.groupby("上架频道")["spu_code"] .nunique() .rename("spus") .reset_index() .rename(columns={"上架频道": "listingChannel"}) ) total = aggregate(sales) monthly = labelled(grouped(sales, ["month", "channel"]), "channel") monthly_total = grouped(sales, ["month"]) monthly_total["share"] = monthly_total["amount"] / total["amount"] channel_rows = labelled(grouped(sales, ["channel"]), "channel") category2_rows = grouped(sales, ["category2"]) category3_rows = grouped(sales, ["category3"]) category4_rows = grouped( sales, ["category4"], category_slots.rename(columns={"category": "category4"}), ) category4_rows["share"] = category4_rows["amount"] / total["amount"] brand_rows = grouped(sales, ["brand"]).sort_values("amount", ascending=False) series_rows = grouped(sales, ["series"]).sort_values("amount", ascending=False) card_rows = grouped(sales, ["card_type"]).sort_values("amount", ascending=False) price = sales[sales["amount"] > 0].copy() price["price_band"] = pd.cut( price["unit_value"], bins=[0, 50, 100, 200, 400, 800, np.inf], right=False, labels=["0-49.99", "50-99.99", "100-199.99", "200-399.99", "400-799.99", "800及以上"], ).astype(str) price_rows = labelled(grouped(price, ["price_band", "channel"]), "channel") margin_rows = labelled(grouped(kitchen, ["category2", "channel"]), "channel") margin_rows["floor"] = margin_rows["channel"].map(FLOORS) margin_rows["status"] = np.where( margin_rows["margin"] >= margin_rows["floor"], "达到参考线", "毛利修复" ) margin_rows["static_gap"] = margin_rows["floor"] * margin_rows["amount"] - margin_rows["gross"] margin_rows["grain"] = "后台二级分类×频道×年度" listing_sku_sets = listing.groupby("spu_code")["sku_code"].apply(set).to_dict() spu_rows = [] for spu, group in listing.groupby("spu_code", sort=False): sku_set = listing_sku_sets[spu] related = sales[sales["sku"].isin(sku_set)] by_channel = labelled(grouped(related, ["channel"]), "channel") spu_rows.append( { "spu": spu, "skus": sorted(sku_set), "skuCount": len(sku_set), "name": group["商品名称"].iloc[0], "brand": " / ".join(sorted(set(group["管理品牌名称"]) - {""})), "category": " / ".join(sorted(set(group["后台四级分类"]) - {""})), "listedChannels": sorted(set(group["上架频道"]) - {""}), "firstListed": safe_date(group["首次上架时间"]), "snapshotInventory": None, "newlyObserved": bool(group["快照新增SPU"].any()), "managementBrandBasis": sorted(set(group["管理品牌依据"])), "operatingTags": sorted(set(group["运营标签"])), "occupancyFlags": sorted(set(group["占坑标记"])), "listedSkus": sorted(sku_set), "currentVariants": [ {"sku": r["sku_code"], "spec": r["规格"], "tag": r["运营标签"], "occupancy": r["占坑标记"], "sourceRow": r["来源行"], "states": r["当前规格状态"]} for r in group.drop_duplicates("sku_code").to_dict("records") ], "annual": aggregate(related), "channels": records(by_channel), } ) spu_rows.sort(key=lambda row: (row["annual"]["amount"], row["spu"])) products = grouped(sales, ["sku", "name", "brand", "category4", "channel"]) products = products.sort_values("amount", ascending=False) products["spu"] = products["sku"].map(mapping) products = labelled(products, "channel") products = products[ ["sku", "spu", "name", "brand", "category4", "channel", "channelLabel", *METRICS, "margin", "unit_value"] ] category_channel_rows = labelled(grouped(sales, ["category4", "channel"]), "channel") category_channel_rows["category2"] = sales["category2"].mode().iat[0] if not sales.empty else "厨房用具" unmatched = sales[sales["spu"].isna()] unmapped_rows = grouped( unmatched, ["sku", "name", "brand", "category4"] ).sort_values("amount", ascending=False).head(100) mapping_stats = { "annualMatched": aggregate(sales[sales["spu"].notna()]), "annualUnmatched": aggregate(unmatched), "mappedSkuCount": int(sales.loc[sales["spu"].notna(), "sku"].nunique()), "unmappedSkuCount": int(unmatched["sku"].nunique()), "currentListingSkuCount": int(listing["sku_code"].nunique()), "currentListingSpuCount": int(listing["spu_code"].nunique()), } quality = { "rows": int(len(sales)), "months": sorted(sales.loc[sales["month"] != "", "month"].unique().tolist()), "expectedMonths": [ f"{year}-{month:02d}" for year in (2025, 2026) for month in range(1, 13) if (year == 2025 and month >= 9) or (year == 2026 and month <= 8) ], "blankBrandRows": int(sales["brand"].eq("").sum()), "missingSkuRows": int(sales["sku"].eq("").sum()), "negativeAmountRows": int(sales["amount"].lt(0).sum()), "zeroAmountRows": int(sales["amount"].eq(0).sum()), "fractionalQuantityRows": int( (sales["quantity"] - sales["quantity"].round()).abs().gt(0.001).sum() ), "sourceMarginMismatchRows": int( ( sales["source_margin"] - sales["gross"] / sales["amount"].replace(0, np.nan) ) .abs() .gt(1e-6) .fillna(False) .sum() ), } listing_summary = { "rows": int(len(listing)), "uniqueSkus": int(listing["sku_code"].nunique()), "uniqueSpus": int(listing["spu_code"].nunique()), "categoryCount": int(listing["后台四级分类"].nunique()), "coveredCategoryCount": int((category_slots["currentSlots"] > 0).sum()), "categories": records(category_slots.rename(columns={"category": "category4"})), "listingChannels": records(channel_slots), "spuCategoryConflicts": int( (listing.groupby("spu_code")["后台四级分类"].nunique() > 1).sum() ), "skuDuplicateRows": int(listing["sku_code"].duplicated().sum()), } result = { "version": VERSION, "snapshot": snapshot, "asOf": "2026-09-21", "factSource": "P02", "sourceFile": "新补充私有数据/兑换明细_业绩-2025.9.1-2026.8.31.xlsx", "period": PERIOD, "categoryScope": {"category2": "厨房用具", "category3": "饮水用具"}, "priority": ["兑换金额", "兑换毛利率（含C端运费）", "利润"], "marginRule": { "grain": "后台二级分类×频道×年度", "formula": "兑换毛利额（含C端运费）合计÷兑换金额合计", "floors": FLOORS, "channelDisplay": CHANNEL_LABELS, }, "overall": total, "monthly": records(monthly_total), "monthlyByChannel": records(monthly), "channels": records(channel_rows), "category2": records(category2_rows), "category3": records(category3_rows), "categories": records(category4_rows), "categoryChannels": records(category_channel_rows), "margins": records(margin_rows), "brands": records(brand_rows), "series": records(series_rows), "cardTypes": records(card_rows), "priceBands": records(price_rows), "products": records(products), "spus": spu_rows, "listing": listing_summary, "listingSource": "P06", "mapping": mapping_stats, "unmappedSkus": records(unmapped_rows), "quality": quality, "kitchenOverall": aggregate(kitchen), } enrich(result, sales, kitchen, listing) export_tables(result) (EVIDENCE / "analysis.json").write_text( json.dumps(result, ensure_ascii=False, indent=2, allow_nan=False), encoding="utf-8", newline="\n", ) print( json.dumps( { "factSource": result["factSource"], "period": result["period"], "overall": result["overall"], "listing": result["listing"], "margins": result["margins"], "quality": result["quality"], }, ensure_ascii=False, indent=2, ) ) if __name__ == "__main__": main()