#!/usr/bin/env python3 # -*- coding: utf-8 -*- """价格索引 · 逐年核对（2024/2025/2026） 核对项： 1. 逐年 条数/单据数/货号数/客户数/销售员数/金额 2. 重复入库检查：同 (单据编号+货号+含税单价[+数量]) 出现多次；跨年单据交集 3. 价差异常：**按单位分组**（同一货号存在 U / ml / 瓶 多种销售单位， 跨单位比价无意义，必须先分组再比） 4. 三年价格趋势：按 (货号 × 单位) 分组，比较各年加权均价 5. 客户维度价差（同样按单位分组） 6. 年度分布字段抽查 7. 新增 2024 源文件口径核对（原始行数 / 有效行数 / 各月齐备） 用法：/usr/bin/python3 脚本/audit_price_years.py """ import collections import json import os import sys ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) IDX = os.path.join(ROOT, '数据', 'price_index.json') YEARS = ('2024', '2025', '2026') def wavg(rs): """加权均价（价税合计 / 数量），量不足则退回算术均""" q = sum((r.get('qty') or 0) for r in rs) t = sum((r.get('total') or 0) for r in rs) if q: return t / q ps = [r['price'] for r in rs if r.get('price') is not None] return sum(ps) / len(ps) if ps else 0.0 def main(): idx = json.load(open(IDX, encoding='utf-8')) m = idx['meta'] recs = [r for o in idx['skus'].values() for r in (o.get('records') or [])] print('索引：%s year_range=%s 明细=%d' % (m['generated'], m['year_range'], m['record_count'])) print('meta.by_year = %s' % json.dumps(m.get('by_year'), ensure_ascii=False)) # ① 逐年规模 print('\n① 逐年规模（明细口径）') for y in YEARS: rs = [r for r in recs if str(r['year']) == y] print(' %s: 条数 %5d 单据 %4d 货号 %4d 客户 %4d 销售员 %2d 金额 %12.0f' % ( y, len(rs), len({r['order'] for r in rs}), len({r['sku'] for r in rs}), len({r['customer'] for r in rs}), len({r['sales'] for r in rs}), sum(r['total'] or 0 for r in rs))) # ② 重复入库 print('\n② 重复入库检查（同 单据编号+货号+含税单价）') for y in YEARS: rs = [r for r in recs if str(r['year']) == y] g = collections.Counter((r['order'], r['sku'], r['price']) for r in rs) gq = collections.Counter((r['order'], r['sku'], r['price'], r['qty']) for r in rs) gf = collections.Counter((r['order'], r['sku'], r['price'], r['qty'], r['total']) for r in rs) d = [k for k, v in g.items() if v > 1] dq = [k for k, v in gq.items() if v > 1] df = [k for k, v in gf.items() if v > 1] print(' %s: 单据+货号+价 重复 %d 组（%d 条）；再含数量 %d 组；再含金额 %d 组（多出 %d 条）' % ( y, len(d), sum(g[k] for k in d), len(dq), len(df), sum(gf[k] - 1 for k in df))) oy = {y: {r['order'] for r in recs if str(r['year']) == y and r['order']} for y in YEARS} for a, b in (('2024', '2025'), ('2025', '2026')): print(' 跨年单据交集 %s∩%s = %d' % (a, b, len(oy[a] & oy[b]))) # ③ 价差异常（按单位） print('\n③ 价差过大（成交次数>=5，按单位分组，max/min 倍数 TOP12）') rows = [] for sku, o in idx['skus'].items(): s = o.get('stats') or {} bu = s.get('by_unit') if bu: for u, a in bu.items(): if a['deal_count'] >= 5 and a['min'] > 0: rows.append((a['max'] / a['min'], sku, u, a['deal_count'], a['min'], a['max'])) elif s.get('deal_count', 0) >= 5 and s.get('min', 0) > 0: rows.append((s['max'] / s['min'], sku, (s.get('units') or ['-'])[0], s['deal_count'], s['min'], s['max'])) rows.sort(reverse=True) for r in rows[:12]: print(' 倍数 %7.2f %-18s [%s] n=%-3d %.6f ~ %.6f' % r) # ④ 三年趋势（按 货号 × 单位） print('\n④ 三年价格趋势（同 货号×单位 在 ≥2 年有成交，各年 ≥2 次，比加权均价）') cell = collections.defaultdict(list) for r in recs: cell[(r['sku'], r.get('unit') or '')].append(r) trend = [] for (sku, u), rs in cell.items(): byy = {} for y in YEARS: ys = [r for r in rs if str(r['year']) == y] if len(ys) >= 2: byy[y] = wavg(ys) if len(byy) < 2: continue ys = sorted(byy) a, b = byy[ys[0]], byy[ys[-1]] if a <= 0: continue trend.append(((b - a) / a * 100.0, sku, u, ys[0], a, ys[-1], b, (idx['skus'][sku].get('product') or {}).get('name') or '')) trend.sort(reverse=True) up = sum(1 for t in trend if t[0] > 3) dn = sum(1 for t in trend if t[0] < -3) print(' 可比组合 %d 个（货号×单位）：涨(>+3%%) %d / 跌(<-3%%) %d / 基本持平 %d' % ( len(trend), up, dn, len(trend) - up - dn)) print(' ── 涨幅 TOP8 ──') for t in trend[:8]: print(' %+8.1f%% %-18s [%s] %s→%s %.4f→%.4f %s' % ( t[0], t[1], t[2], t[3], t[5], t[4], t[6], t[7][:24])) print(' ── 跌幅 TOP8 ──') for t in trend[-8:]: print(' %+8.1f%% %-18s [%s] %s→%s %.4f→%.4f %s' % ( t[0], t[1], t[2], t[3], t[5], t[4], t[6], t[7][:24])) # ⑤ 客户维度价差（按 货号×单位×客户） print('\n⑤ 客户维度价差（同 货号×单位×客户，成交>=3次，倍数 TOP10）') cc = collections.defaultdict(list) for r in recs: cc[(r['sku'], r.get('unit') or '', r.get('customer') or '')].append(r) cs = [] for (sku, u, cust), rs in cc.items(): if len(rs) >= 3: ps = sorted(r['price'] for r in rs) if ps[0] > 0: cs.append((ps[-1] / ps[0], sku, u, cust, len(rs), ps[0], ps[-1])) cs.sort(reverse=True) for x in cs[:10]: print(' 倍数 %6.2f %-14s [%-7s] %-24s n=%-2d %.4f~%.4f' % ( x[0], x[1], x[2], x[3][:24], x[4], x[5], x[6])) print(' （注：同一货号单价跨度大，多因「样品价/大包装价」并存，非一定是异常，需结合数量看）') # ⑤b 极端低价抽查（单价 < 1 元且单位非 U） print('\n⑤b 单价异常低（<1 元非 U 单位）抽查 TOP8（按金额倒序，多为 ¥1 象征价测试单）') low = [r for r in recs if (r.get('price') or 0) < 1 and (r.get('unit') or '') not in ('U',) and (r.get('price') or 0) > 0] low.sort(key=lambda r: -(r.get('total') or 0)) for r in low[:8]: print(' %s %s %s [%s] 单价%.4f x %s = %.2f 客户%s' % ( r['date'], r['sku'], (idx['skus'][r['sku']].get('product') or {}).get('name') or ''[:18], r.get('unit'), r['price'], r['qty'], r.get('total'), r['customer'][:16])) print(' 共 %d 条单价<1元且非U单位的记录（占全部 %.2f%%）' % (len(low), len(low) * 100.0 / len(recs))) # ⑥ 年度分布抽查 print('\n⑥ 年度分布抽查（成交最多的 3 个货号 by_year）') for sku, o in sorted(idx['skus'].items(), key=lambda kv: -(kv[1].get('stats') or {}).get('deal_count', 0))[:3]: print(' %-16s %s' % (sku, json.dumps((o['stats'] or {}).get('by_year'), ensure_ascii=False))) for u, a in ((o['stats'] or {}).get('by_unit') or {}).items(): print(' [%s] %s' % (u, json.dumps(a.get('by_year'), ensure_ascii=False))) # ⑦ 2024 源文件核对 print('\n⑦ 数据源逐文件（meta.sources）') for s in m['sources']: print(' %s | sheet=%s | %s(%s) | 原始 %d 行 → 有效 %d 条' % ( os.path.basename(s['file']), s['sheet'], s['date_field'], s['date_granularity'], s['raw_rows'], s.get('valid_rows'))) return 0 if __name__ == '__main__': sys.exit(main())