from pathlib import Path import json, math, hashlib, re from decimal import Decimal from datetime import datetime, timezone import pandas as pd import numpy as np import openpyxl ROOT = Path(__file__).resolve().parent SRC = Path(r'C:\Users\Administrator\Desktop\犀牛百货（北海店最终版本.xlsx') d = pd.read_excel(SRC, dtype=str).fillna('') d.insert(0, '源表行号', np.arange(2, len(d)+2)) numcols = ['商品原价','商品活动价','商品库存','最低起订量','重量'] for c in numcols: d[c+'_数值'] = pd.to_numeric(d[c], errors='coerce') d['月销数值'] = pd.to_numeric(d['商品月销'], errors='coerce') d['想买人数'] = pd.to_numeric(d['商品月销'].str.extract(r'^(\d+)人想买$')[0], errors='coerce') d['月销字段类型'] = np.select([d['月销数值'].notna(), d['想买人数'].notna()], ['数值销量','意向人数'], default='未识别') d['折扣率'] = 1-d['商品活动价_数值']/d['商品原价_数值'] d['价量可比候选'] = (d['最低起订量_数值']==1)&(d['商品活动价_数值']>0)&(d['商品活动价_数值']<=d['商品原价_数值']) d['库存标价规模'] = np.where(d['价量可比候选'], d['商品库存_数值']*d['商品活动价_数值'], np.nan) records=[] for spu,g in d.groupby('SPU',sort=False): consistent=g['商品月销'].nunique()==1 numeric=consistent and g['月销数值'].notna().all() intent=consistent and g['想买人数'].notna().all() sale=float(g['月销数值'].iloc[0]) if numeric else None want=float(g['想买人数'].iloc[0]) if intent else None unit_ok=bool(g['价量可比候选'].all()) stock=float(g['商品库存_数值'].sum()) single=len(g)==1 records.append({'SPU':spu,'商品名称':g['商品名称'].iloc[0], '一级类目':g['美团一级类目'].iloc[0], '二级类目':g['美团二级类目'].iloc[0], '三级类目':g['美团三级类目'].iloc[0], 'SKU数':len(g),'月销标签一致':consistent,'月销字段原文':' / '.join(g['商品月销'].unique()), '销量类型':'数值销量' if numeric else ('意向人数' if intent else '同SPU冲突'), '近30天销量_一次计数':sale,'想买人数_一次计数':want, '规格库存字段合计':stock,'有库存规格数':int((g['商品库存_数值']>0).sum()), '最低活动价':float(g['商品活动价_数值'].min()),'最高活动价':float(g['商品活动价_数值'].max()), '价量可比候选':unit_ok,'单SKU':single, '价格乘销量下界情景':sale*g['商品活动价_数值'].min() if numeric and unit_ok else None, '价格乘销量上界情景':sale*g['商品活动价_数值'].max() if numeric and unit_ok else None, '单SKU价格乘销量':sale*g['商品活动价_数值'].iloc[0] if numeric and single and unit_ok else None, '单SKU库存覆盖天数':30*stock/sale if numeric and single and sale>0 else None, '单SKU补至14天件数':max(math.ceil(sale*14/30)-int(stock),0) if numeric and single and sale>0 else None, '单SKU超90天库存件数':max(stock-sale*3,0) if numeric and single and sale>0 else None, '单SKU库存标价规模':stock*g['商品活动价_数值'].iloc[0] if single and unit_ok else None, '源表行号':','.join(g['源表行号'].astype(str)), '活动类型':' / '.join(g['活动类型'].unique())}) s = pd.DataFrame(records) known = s[s['销量类型']=='数值销量'].copy() positive=known[known['近30天销量_一次计数']>0].copy() single=known[known['单SKU']].copy() sp=positive[positive['单SKU']].copy() intent=s[s['销量类型']=='意向人数'].copy() zero=known[known['近30天销量_一次计数']==0].copy() conflict=s[s['销量类型']=='同SPU冲突'].copy() def clean(v): if isinstance(v, dict): return {str(k):clean(x) for k,x in v.items()} if isinstance(v, (list, tuple)): return [clean(x) for x in v] if isinstance(v, (np.integer,)): return int(v) if isinstance(v, (np.floating,)): return None if not np.isfinite(v) else float(v) if isinstance(v, float) and not math.isfinite(v): return None return v def rows(df): return clean(df.to_dict('records')) def write_json(name,obj): (ROOT/name).write_text(json.dumps(clean(obj),ensure_ascii=False,indent=2,allow_nan=False),encoding='utf-8') def csv(name,df): df.to_csv(ROOT/name,index=False,encoding='utf-8-sig') def pct(n,base): return n/base if base else None K=float(known['近30天销量_一次计数'].sum()) stats={ 'source':str(SRC),'sha256':hashlib.sha256(SRC.read_bytes()).hexdigest(), 'analysis_date':'2026-11-14','export_date':'未提供','sales_period':'近30天，用户确认；对应起止日期未提供', 'rows':len(d),'cols':18,'spu':len(s),'sku':d.SKU.nunique(),'cat1':d['美团一级类目'].nunique(), 'cat2':d['美团二级类目'].nunique(),'cat3':d['美团三级类目'].nunique(), 'numeric_sku_rows':int(d['月销数值'].notna().sum()),'intent_sku_rows':int(d['想买人数'].notna().sum()), 'numeric_spu':len(known),'positive_spu':len(positive),'zero_spu':len(zero),'intent_spu':len(intent),'conflict_spu':len(conflict), 'multi_spu':int((s['SKU数']>1).sum()),'multi_rows':int(s.loc[s['SKU数']>1,'SKU数'].sum()), 'monthly_qty_spu':K,'naive_monthly_qty_sku':float(d['月销数值'].sum()), 'same_label_naive_monthly_qty':float(d.loc[d.SPU.isin(known.SPU),'月销数值'].sum()), 'single_numeric_spu':len(single),'single_positive_spu':len(sp), 'single_numeric_qty':float(single['近30天销量_一次计数'].sum()), 'catalog_stock_fields_sum':float(d['商品库存_数值'].sum()),'stock_zero_sku':int((d['商品库存_数值']==0).sum()), 'stock_zero_spu':int((s['规格库存字段合计']==0).sum()), 'sku_original_value_stock':float((d['商品库存_数值']*d['商品原价_数值']).sum()), 'sku_activity_value_stock':float((d['商品库存_数值']*d['商品活动价_数值']).sum()), 'comparable_inventory_list_value':float(d['库存标价规模'].sum()), 'single_inventory_list_value':float(s['单SKU库存标价规模'].sum()), 'single_price_qty_proxy':float(single['单SKU价格乘销量'].sum()), 'price_qty_low_scenario':float(known['价格乘销量下界情景'].sum()), 'price_qty_high_scenario':float(known['价格乘销量上界情景'].sum()), 'price_qty_scenario_spu':int(known['价格乘销量下界情景'].notna().sum()), 'price_qty_scenario_qty':float(known.loc[known['价格乘销量下界情景'].notna(),'近30天销量_一次计数'].sum()), 'intent_spu_stock':float(intent['规格库存字段合计'].sum()), 'zero_spu_stock':float(zero['规格库存字段合计'].sum()), 'positive_spu_stock':float(positive['规格库存字段合计'].sum()), 'intent_stock_list_value':float(d.loc[d.SPU.isin(intent.SPU),'库存标价规模'].sum()), 'zero_stock_list_value':float(d.loc[d.SPU.isin(zero.SPU),'库存标价规模'].sum()), 'numeric_zero_with_stock_spu':int((zero['规格库存字段合计']>0).sum()), 'intent_with_stock_spu':int((intent['规格库存字段合计']>0).sum()), 'want_sum':float(intent['想买人数_一次计数'].sum()), 'nulls':{c:int((d[c]=='').sum()) for c in d.columns[:19]}, 'exact_duplicate_rows':int(d.iloc[:,1:19].duplicated().sum()), 'duplicate_sku':int(d.SKU.duplicated().sum()), 'price_inversion_sku':int((d['商品活动价_数值']>d['商品原价_数值']).sum()), 'price_inversion_moq_gt1':int(((d['商品活动价_数值']>d['商品原价_数值'])&(d['最低起订量_数值']>1)).sum()), 'moq_gt1':int((d['最低起订量_数值']>1).sum()), 'stock_lt_moq':int(((d['商品库存_数值']>0)&(d['商品库存_数值']100000).sum()), 'upc_nonstandard_len':int(((d.UPC!='')&(~d.UPC.str.fullmatch(r'\d{8}|\d{12}|\d{13}|\d{14}'))).sum()), 'nonpositive_price':int(((d['商品原价_数值']<=0)|(d['商品活动价_数值']<=0)).sum()), 'negative_stock_sales':int(((d['商品库存_数值']<0)|(d['月销数值']<0)).sum()), 'numeric_parse_fail':{c:int(((d[c]!='')&d[c+'_数值'].isna()).sum()) for c in numcols}, 'activity_counts':d['活动类型'].replace('','未标注').value_counts().to_dict(), 'price_quantiles':d['商品活动价_数值'].quantile([.1,.25,.5,.75,.9,.95,.99]).to_dict(), 'stock_quantiles':d['商品库存_数值'].quantile([.1,.5,.9,.95,.99]).to_dict(), } category=[] for cat,g in s.groupby('一级类目',sort=False): kg=g[g['销量类型']=='数值销量']; pg=kg[kg['近30天销量_一次计数']>0] dg=d[d.SPU.isin(g.SPU)] q=float(kg['近30天销量_一次计数'].sum());st=float(g['规格库存字段合计'].sum()) category.append({'一级类目':cat,'SPU数':len(g),'SKU数':int(g['SKU数'].sum()),'数值销量SPU数':len(kg), '正销量SPU数':len(pg),'零销量SPU数':int((kg['近30天销量_一次计数']==0).sum()), '意向SPU数':int((g['销量类型']=='意向人数').sum()),'冲突SPU数':int((g['销量类型']=='同SPU冲突').sum()), '数值销量覆盖率':pct(len(kg),len(g)),'近30天销量_一次计数':q,'销量份额':pct(q,K), 'SPU份额':len(g)/len(s),'库存字段合计':st,'库存份额':st/stats['catalog_stock_fields_sum'], '每个数值销量SPU月销':pct(q,len(kg)),'单SKU价格乘销量':float(g['单SKU价格乘销量'].sum()), '平均活动价':float(dg['商品活动价_数值'].mean()),'活动价中位数':float(dg['商品活动价_数值'].median()), '可比候选平均折扣率':float(dg.loc[dg['价量可比候选'],'折扣率'].mean()), '单SKU正销量SPU数':int(((g['单SKU'])&(g['近30天销量_一次计数']>0)).sum()), '库存标价规模':float(dg['库存标价规模'].sum())}) cat=pd.DataFrame(category).sort_values('近30天销量_一次计数',ascending=False) stats['category_sales_hhi']=float((cat['销量份额']**2).sum()) ranked=positive.sort_values(['近30天销量_一次计数','SPU'],ascending=[False,True]).copy() ranked['累计销量占比']=ranked['近30天销量_一次计数'].cumsum()/K ranked['排名']=np.arange(1,len(ranked)+1) ranked['ABC']=np.select([(ranked['近30天销量_一次计数'].cumsum()-ranked['近30天销量_一次计数'])/K<.8, (ranked['近30天销量_一次计数'].cumsum()-ranked['近30天销量_一次计数'])/K<.95],['A','B'],default='C') pareto=[] for fraction in [.01,.05,.1,.2,.5]: n=math.ceil(len(known)*fraction) q=float(ranked.head(n)['近30天销量_一次计数'].sum()) pareto.append({'数值销量SPU的比例':fraction,'SPU数':n,'近30天销量':q,'销量份额':q/K}) stats['pareto']=pareto stats['spu_for_80pct']=int((ranked['累计销量占比']<.8).sum()+1) stats['spu_for_95pct']=int((ranked['累计销量占比']<.95).sum()+1) stats['top1_spu_sales_share']=float(ranked.iloc[0]['近30天销量_一次计数']/K) stats['abc']=rows(ranked.groupby('ABC',sort=False).agg(SPU数=('SPU','size'),近30天销量=('近30天销量_一次计数','sum'),库存字段合计=('规格库存字段合计','sum')).reset_index()) cover_bins=[-1,0,7,14,30,90,np.inf] cover_labels=['库存为0','0至7天','7至14天','14至30天','30至90天','超过90天'] sp['覆盖分组']=pd.cut(sp['单SKU库存覆盖天数'],bins=cover_bins,labels=cover_labels,right=True) coverage=sp.groupby('覆盖分组',observed=False).agg(SPU数=('SPU','size'),近30天销量=('近30天销量_一次计数','sum'),库存字段合计=('规格库存字段合计','sum'),库存标价规模=('单SKU库存标价规模','sum')).reset_index() low=sp[sp['单SKU库存覆盖天数']<7].sort_values(['近30天销量_一次计数','单SKU库存覆盖天数'],ascending=[False,True]).copy() low['补至14天活动价预算情景']=np.where(low['价量可比候选'],low['单SKU补至14天件数']*low['最低活动价'],np.nan) over=sp[sp['单SKU库存覆盖天数']>90].sort_values('单SKU库存标价规模',ascending=False).copy() over['超90天库存标价规模']=np.where(over['价量可比候选'],over['单SKU超90天库存件数']*over['最低活动价'],np.nan) stats.update({'under7_single_spu':len(low),'under7_monthly_qty':float(low['近30天销量_一次计数'].sum()), 'under7_refill14_qty':float(low['单SKU补至14天件数'].sum()),'under7_refill14_list_budget':float(low['补至14天活动价预算情景'].sum()), 'over90_single_spu':len(over),'over90_excess_qty':float(over['单SKU超90天库存件数'].sum()), 'over90_excess_list_value':float(over['超90天库存标价规模'].sum()), 'positive_zero_stock_spu':int((positive['规格库存字段合计']==0).sum()), 'single_positive_zero_stock_spu':int((sp['规格库存字段合计']==0).sum())}) discounts=d[d['价量可比候选']].copy() discounts['折扣分组']=pd.cut(discounts['折扣率'],bins=[-.001,0,.1,.2,.3,.4,.5,1],labels=['未降价','0至10%','10至20%','20至30%','30至40%','40至50%','50%以上'],include_lowest=True) disc=discounts.groupby('折扣分组',observed=False).agg(SKU数=('SKU','size'),库存字段合计=('商品库存_数值','sum')).reset_index() stats['comparable_price_sku']=len(discounts) stats['discount_mean_sku']=float(discounts['折扣率'].mean()) stats['discount_median_sku']=float(discounts['折扣率'].median()) stats['discount_50plus_sku']=int((discounts['折扣率']>=.5).sum()) stats['discount_30plus_sku']=int((discounts['折扣率']>=.3).sum()) stats['discount_activities_sku']=int(d['活动类型'].str.contains('折扣|神价|爆品').sum()) stats['no_activity_price_different']=int(((d['活动类型']=='无活动')&(d['商品活动价_数值']!=d['商品原价_数值'])).sum()) stats['discount_mark_no_decrease']=int((d['活动类型'].str.contains('折扣')&(d['商品活动价_数值']>=d['商品原价_数值'])).sum()) # All correlations below are descriptive, never causal; only single-SKU numeric SPUs. tmp=d[d.SPU.isin(single.SPU)&d['价量可比候选']].copy() tmp['折扣分组']=pd.cut(tmp['折扣率'],bins=[-.001,.1,.2,.3,.4,.5,1],labels=['0至10%','10至20%','20至30%','30至40%','40至50%','50%以上']) disc_sales=tmp.groupby('折扣分组',observed=False).agg(单SKU商品数=('SPU','size'),近30天销量=('月销数值','sum'),月销中位数=('月销数值','median'),月销均值=('月销数值','mean')).reset_index() price_bands=d.copy() price_bands['活动价带']=pd.cut(price_bands['商品活动价_数值'],bins=[0,5,10,20,30,50,100,np.inf],labels=['5元及以下','5至10元','10至20元','20至30元','30至50元','50至100元','100元以上']) prices=price_bands.groupby('活动价带',observed=False).agg(SKU数=('SKU','size'),库存字段合计=('商品库存_数值','sum')).reset_index() intent_rank=intent.sort_values(['想买人数_一次计数','规格库存字段合计'],ascending=False) zeros_rank=zero.sort_values('单SKU库存标价规模',ascending=False) exceptions=d[(d['商品活动价_数值']>d['商品原价_数值'])|((d['商品库存_数值']>0)&(d['商品库存_数值']100000][['源表行号','SKU','商品名称','规格名称','重量']].head(20))} # Independent, integer/Decimal controls read directly from the original workbook. ws=openpyxl.load_workbook(SRC,read_only=True,data_only=True).active source=list(ws.iter_rows(min_row=2,values_only=True)) assert len(source)==len(d)==11132 assert len({r[1] for r in source})==len(d) assert sum(int(str(r[9])) for r in source)==stats['catalog_stock_fields_sum'] by={} for r in source: by.setdefault(str(r[0]),[]).append(r) ind_known={k:v for k,v in by.items() if len({str(x[10]) for x in v})==1 and re.fullmatch(r'\d+(\.\d+)?',str(v[0][10]))} ind_qty=sum(Decimal(str(v[0][10])) for v in ind_known.values()) assert float(ind_qty)==K assert float(cat['近30天销量_一次计数'].sum())==K assert int(cat['SPU数'].sum())==len(s) assert int(cat['SKU数'].sum())==len(d) assert float(cat['库存字段合计'].sum())==stats['catalog_stock_fields_sum'] assert len(known)+len(intent)+len(conflict)==len(s) assert int(coverage['SPU数'].sum())==len(sp) assert int(disc['SKU数'].sum())==len(discounts) assert int(prices['SKU数'].sum())==len(d) assert not d['月销字段类型'].eq('未识别').any() stats['validation']={'independent_original_workbook_recheck':'PASS','category_reconciliation':'PASS','population_partition':'PASS','coverage_partition':'PASS','unknown_sales_labels':0} csv('sku_evidence.csv',d);csv('spu_evidence.csv',s);csv('category_evidence.csv',cat);csv('subcategory_evidence.csv',subcats) csv('replenish_candidates.csv',low);csv('over90_candidates.csv',over);csv('zero_sales_candidates.csv',zeros_rank);csv('intent_candidates.csv',intent_rank);csv('price_exceptions.csv',exceptions) csv('abc_rankings.csv',ranked) pool_specs=[('雪花清爽500ml','25093399234',1),('雪花清爽500ml','25094341211',3),('雪花清爽500ml','25094870888',6),('雪花清爽500ml','25095329052',12),('雪花清爽500ml','25102815880',12),('雪花清爽500ml','26624543042',4),('可口可乐经典330ml','25093193845',1),('可口可乐经典330ml','25095604000',6),('可口可乐经典330ml','25102819830',6)] pool=[] for family,spu,factor in pool_specs: rr=d[d.SPU==spu].iloc[0] pool.append({'商品组候选':family,'SPU':spu,'商品名称':rr['商品名称'],'源表行号':int(rr['源表行号']), '每销售单位罐数':factor,'库存销售单位':float(rr['商品库存_数值']), '折算罐库存':float(rr['商品库存_数值'])*factor,'月销原文':rr['商品月销'], '数值月销折算罐':None if pd.isna(rr['月销数值']) else float(rr['月销数值'])*factor, '包装活动价':float(rr['商品活动价_数值']),'每罐活动价':float(rr['商品活动价_数值'])/factor}) tables['pack_pool_candidates']=pool stats['snow_pool_scenario']={'catalog_base_stock_sum':sum(r['折算罐库存'] for r in pool if r['商品组候选']=='雪花清爽500ml'), 'assumed_shared_pool_stock':1329,'known_sales_base_cans':sum((r['数值月销折算罐'] or 0) for r in pool if r['商品组候选']=='雪花清爽500ml'), 'coverage_days_if_shared':1329/1282*30,'stock_sum_ratio_if_shared':7952/1329} def gtin_check(v): if not re.fullmatch(r'\d{8}|\d{12}|\d{13}|\d{14}',v):return None digits=[int(x) for x in v] return (10-sum(x*(3 if i%2==0 else 1) for i,x in enumerate(digits[-2::-1]))%10)%10==digits[-1] checks=d.UPC.map(gtin_check) stats['upc_format_valid']=int(checks.notna().sum()) stats['upc_check_digit_pass']=int(checks.eq(True).sum()) stats['upc_check_digit_fail']=int(checks.eq(False).sum()) stats['sku_10yuan_or_less']=int((d['商品活动价_数值']<=10).sum()) stats['stock_10yuan_or_less']=float(d.loc[d['商品活动价_数值']<=10,'商品库存_数值'].sum()) stats['single_money_proxy_spu']=int(single['单SKU价格乘销量'].notna().sum()) stats['single_money_proxy_qty']=float(single.loc[single['单SKU价格乘销量'].notna(),'近30天销量_一次计数'].sum()) stats['single_known_vs_spu_known_qty_share']=stats['single_numeric_qty']/K stats['numeric_catalog_spu_coverage']=len(known)/len(s) stats['sum_sku_vs_spu_difference_pct']=stats['naive_monthly_qty_sku']/K-1 stats['avg_numeric_spu_sales']=K/len(known) stats['median_numeric_spu_sales']=float(known['近30天销量_一次计数'].median()) stats['monthly_qty_le2_spu']=int((known['近30天销量_一次计数']<=2).sum()) csv('pack_pool_candidates.csv',pd.DataFrame(pool)) write_json('metrics.json',stats);write_json('tables.json',tables) write_json('evidence_snapshot.json',{'metrics':stats,'tables':tables}) print(json.dumps(clean(stats),ensure_ascii=False,indent=2)) print('TOP_CATEGORIES',cat.head(12).to_json(orient='records',force_ascii=False)) print('COVERAGE',coverage.to_json(orient='records',force_ascii=False)) print('CONFLICTS',conflict.to_json(orient='records',force_ascii=False))