name: OpenDemi-DataAnalysis(得觅·数据分析) version: 5.2.0 description: | 「OpenDemi · 得觅」审计四环之数据分析引擎。 面向企业通用审计场景的深度数据分析技能。三层解耦架构(L1通用引擎→L2业务路由→L3场景模板)。 支持8种分析模式路由、7类数据源接入、统计检验、安全规范、降级策略、结论撰写。 触发场景:数据分析、审计分析、Excel核查、合规检测、趋势预警、钩稽验证、差异分解、内控测试。 agent_created: true brand: OpenDemi tags: [OpenDemi, data-analysis, audit, excel, csv, sql, pandas, python, visualization, statistics, reconciliation, sox, business-intelligence]
面向企业通用审计场景的通用深度数据分析技能。
当用户提供业务数据(Excel/CSV/数据库表),需要进行以下任意操作时,启用本技能:
| 部门 | 典型需求 | 报告风格 |
|---|---|---|
| 审计监察部 | 风险排查、钩稽核查、合规检测 | 风险导航式(→见阶段5.1) |
| 财务部(BP/管会) | 预算差异、成本分析、投入产出比 | 偏差归因式(→见阶段5.1) |
| 生产车间 | 合格率/损耗率/效率趋势、异常定位 | 运营监控式(→见阶段5.1) |
| 采购中心 | 供应商合规、价格异常、合同执行 | 合规审计式(→见阶段5.1) |
| 品管部 | 质量指标趋势、批次追溯、标准偏离 | 质量报告式(→见阶段5.1) |
数据分析不是罗列数字,是回答"然后呢"。先判问题类型,再选分析方法;高风险的深挖,低风险的简述。 每个分析模块都应回答:这对决策/行动有什么用?
┌──────────────────────────────────────────────────────────────┐
│ L1 通用分析引擎(方法层,业务无关) │
│ ├ 1.1 数据接入(7类数据源 + 大数据量分档) │
│ ├ 1.2 数据画像(audit_profile + 质量评分) │
│ ├ 1.3 差异分解方法论(价量/率混合/瀑布图归因) │
│ ├ 1.4 对账钩稽体系(3类差异+老化分析+升级阈值) │
│ ├ 1.5 重要性阈值框架(分层阈值+调查优先级) │
│ ├ 1.6 安全规范(3域12条 + 脱敏函数) │
│ ├ 1.7 分析模式路由(8种问题→方法映射 + 统计检验工具箱) │
│ ├ 1.8 降级策略(4级降级链 + 6类错误映射) │
│ ├ 1.9 分析结论撰写规范(四条铁律 + 发现-建议关联模板) │
│ └ 1.10 统计增强工具箱(描述性统计+假设检验+陷阱警告+验证清单)│
├──────────────────────────────────────────────────────────────┤
│ L2 业务场景映射(路由层,按场景调度L1能力) │
│ ├ 产业链场景路由(养殖/加工/制造/流通/副产/综合) │
│ ├ 部门需求路由(审计/财务BP/生产/采购/品管) │
│ ├ 内控测试路由(CEAVOP认定+样本选择+缺陷分级) │
│ ├ 场景→分析方法→图表→报告风格 映射表 │
│ └ 个性化需求适配入口 │
├──────────────────────────────────────────────────────────────┤
│ L3 场景模板(示例层,按需加载,不污染通用逻辑) │
│ ├ 养殖:投产钩稽/原料溯源/合格率趋势 │
│ ├ 加工:原料出入库/配方成本/损耗分析 │
│ ├ 制造:产出效率/生产损耗/副产回收 │
│ ├ 流通:原料分级/出品率/加工损耗 │
│ ├ 财务:预算差异/成本结构/投入产出/结账流程 │
│ └ 综合:跨事业部效率对比/集团趋势预警 │
└──────────────────────────────────────────────────────────────┘
L1模块逻辑顺序说明:编号1.1-1.10为历史迭代顺序(v3.0先建1.1/1.2/1.7/1.8/1.9,v5.0插入1.3-1.6,v5.2新增1.10)。逻辑上的阅读顺序为:输入(1.1→1.2) → 方法(1.3→1.4→1.5) → 路由(1.7) → 统计增强(1.10) → 约束(1.6→1.8) → 输出(1.9)。下次大版本迭代时将统一重编号。
| 被调用技能 | 调用场景 | 调用时机 |
|---|---|---|
weknora-connector(WK检索) |
检索相关制度条款、合同约定作为分析依据 | 数据分析前按需加载 |
obsidian-connector(OB检索) |
读取星觅智库成品仓历史分析底稿/风险判断 | 需要参考历史结论时加载 |
| 数据源 | 接入方式 | 典型场景 |
|---|---|---|
| Excel (.xlsx/.xls) | pd.read_excel() |
业务台账、生产报表、盘点表 |
| CSV (.csv/.tsv) | pd.read_csv(encoding=...) |
系统导出、日志数据 |
| SQLite | sqlite3.connect() + pd.read_sql() |
本地业务数据库、系统备份 |
| MySQL/MariaDB | sqlalchemy + pd.read_sql() |
ERP/财务系统直连 |
| PostgreSQL | sqlalchemy + pd.read_sql() |
集团数据中心 |
| SQL Server | pyodbc + pd.read_sql() |
用友/金蝶等国产ERP |
| DuckDB | duckdb.connect() + SQL查询 |
大文件OLAP分析(百万行级) |
from sqlalchemy import create_engine
import pandas as pd
# MySQL/MariaDB
engine = create_engine('mysql+pymysql://user:pass@host:3306/dbname?charset=utf8mb4')
# PostgreSQL
engine = create_engine('postgresql://user:pass@host:5432/dbname')
# SQLite(零配置)
import sqlite3
conn = sqlite3.connect('data.db')
# DuckDB(百万行级文件分析利器)
import duckdb
conn = duckdb.connect()
df = conn.execute("SELECT * FROM read_csv_auto('big_file.csv')").df()
# 通用安全查询模板
import re
def safe_query(engine_or_conn, sql, max_rows=10000):
"""安全查询:仅允许单条SELECT + 自动LIMIT + 防注入
安全措施:
1. 禁止多语句(分号分隔)——防 DROP/DELETE 注入
2. 仅允许SELECT开头的查询
3. 检测最外层是否有限制行数的子句(LIMIT/OFFSET/FETCH),无则追加
"""
sql = sql.strip().rstrip(';') # 去尾部分号
# 防多语句注入
if ';' in sql:
raise ValueError('安全约束:禁止多语句查询(含分号)')
# 仅允许SELECT
if not sql.upper().startswith('SELECT'):
raise ValueError('安全约束:仅允许SELECT查询')
# 检测最外层是否已有行数限制(只看SQL尾部,避免子查询中的LIMIT干扰)
has_row_limit = bool(re.search(
r'\b(LIMIT\s+\d|OFFSET\s+\d|FETCH\s+FIRST\s+\d+\s+ROWS?)\b',
sql[-80:].upper()
))
if not has_row_limit:
sql += f' LIMIT {max_rows}'
return pd.read_sql(sql, engine_or_conn)
| 数据量 | 策略 | 场景举例 |
|---|---|---|
| \< 10万行 | 全量加载到内存 | 常规业务台账 |
| 10万\~100万行 | 分块读取 chunksize=50000 或采样 |
年度生产明细 |
| 100万\~1000万行 | 数据库端聚合,仅拉取结果 | 集团全量交易流水 |
| > 1000万行 | DuckDB/SQL聚合 + 采样分析 | 全集团历史数据 |
# 大文件分块处理
chunks = pd.read_csv('huge_file.csv', chunksize=50000, encoding='utf-8')
result = []
for chunk in chunks:
agg = chunk.groupby(chunk.iloc[:, 2]).agg({chunk.columns[10]: 'sum'})
result.append(agg)
final = pd.concat(result).groupby(level=0).sum()
| 维度 | 统计项 | 分析价值 |
|---|---|---|
| 基本信息 | 行数、列数、重复行数、内存占用 | 评估数据规模,决定处理策略 |
| 数据类型 | 数值型/分类型/日期型/布尔型列数 | 快速识别可用分析维度 |
| 缺失值 | 每列缺失率、缺失模式(随机/系统性) | 系统性缺失=控制缺陷/流程盲区信号 |
| 唯一值 | 基数(cardinality)、重复率 | 高基数=可能需要分组聚合 |
| 数值分布 | 均值、中位数、偏度、峰度、IQR异常值数 | 偏度异常=可能有数据操纵或异常业务 |
| 分类分布 | Top-N类别、长尾比例 | 集中度异常=垄断/利益输送风险 |
| 相关性 | 数值列间Pearson/Spearman相关矩阵 | 强相关=钩稽校验候选对 |
| 质量评分 | 完整性×唯一性综合评分(0-1) | 跨表/跨期质量对比基准 |
import pandas as pd
import numpy as np
def audit_profile(df, name='数据集'):
"""通用数据画像:输出结构化质量报告"""
profile = {
'name': name,
'overview': {
'rows': len(df), 'columns': len(df.columns),
'duplicated_rows': int(df.duplicated().sum()),
'total_missing': int(df.isnull().sum().sum()),
'missing_pct': round(df.isnull().sum().sum() / (len(df) * len(df.columns)) * 100, 2),
},
'dtypes': {
'numeric': len(df.select_dtypes(include=[np.number]).columns),
'categorical': len(df.select_dtypes(include=['object', 'category']).columns),
'datetime': len(df.select_dtypes(include=['datetime64']).columns),
},
'columns': [],
}
for col_idx, col in enumerate(df.columns):
ci = {
'idx': col_idx, 'name': col, 'dtype': str(df[col].dtype),
'missing_pct': round(df[col].isnull().sum() / len(df) * 100, 2),
'unique_count': int(df[col].nunique()),
}
if pd.api.types.is_numeric_dtype(df[col]):
s = df[col].dropna()
if len(s) > 0:
ci['stats'] = {
'mean': round(float(s.mean()), 2), 'median': float(s.median()),
'std': round(float(s.std()), 2), 'skewness': round(float(s.skew()), 2),
}
Q1, Q3 = s.quantile(0.25), s.quantile(0.75)
IQR = Q3 - Q1
outliers = ((s < Q1 - 1.5*IQR) | (s > Q3 + 1.5*IQR)).sum()
ci['outliers'] = int(outliers)
ci['outliers_pct'] = round(outliers / len(df) * 100, 2)
elif pd.api.types.is_object_dtype(df[col]):
vc = df[col].value_counts()
ci['top5'] = {str(k): int(v) for k, v in vc.head(5).items()}
profile['columns'].append(ci)
# 高相关性对(钩稽候选)——标注高缺失列的不可靠性
numeric_cols = df.select_dtypes(include=[np.number]).columns
if len(numeric_cols) >= 2:
corr = df[numeric_cols].corr()
high_corr = []
for i in range(len(corr.columns)):
for j in range(i+1, len(corr.columns)):
val = corr.iloc[i, j]
if abs(val) > 0.7:
# 检查参与列的缺失率——高缺失列的相关性不可靠
col_a_name, col_b_name = corr.columns[i], corr.columns[j]
miss_a = round(df[col_a_name].isnull().sum() / len(df) * 100, 1)
miss_b = round(df[col_b_name].isnull().sum() / len(df) * 100, 1)
reliability = '可靠'
if miss_a > 30 or miss_b > 30:
reliability = '不可靠(缺失率>{:.0f}%)'.format(max(miss_a, miss_b))
elif miss_a > 15 or miss_b > 15:
reliability = '需谨慎(缺失率>{:.0f}%)'.format(max(miss_a, miss_b))
high_corr.append((col_a_name, col_b_name, round(float(val), 3), reliability))
profile['high_correlations'] = high_corr
# 质量评分
completeness = 1 - profile['overview']['missing_pct'] / 100
uniqueness = 1 - profile['overview']['duplicated_rows'] / max(len(df), 1)
profile['quality_score'] = round((completeness + uniqueness) / 2, 3)
return profile
def profile_to_flags(profile, context='通用'):
"""根据画像生成关注点,context可为'审计'/'生产'/'财务'/'采购'/'品管'"""
flags = []
# 通用关注点——所有context都适用
if profile['overview']['missing_pct'] > 10:
flags.append('数据完整性风险:整体缺失率{:.1f}%,系统性缺失可能掩盖问题'.format(profile['overview']['missing_pct']))
if profile['overview']['duplicated_rows'] > 0:
flags.append('重复行:{}条重复记录,可能存在重复记账或录入错误'.format(profile['overview']['duplicated_rows']))
for ci in profile['columns']:
if ci.get('missing_pct', 0) > 20:
flags.append('字段[{}]缺失率{:.1f}%,高于20%警戒线'.format(ci['name'], ci['missing_pct']))
if ci.get('outliers_pct', 0) > 5:
flags.append('字段[{}]异常值占比{:.1f}%,需核查原因'.format(ci['name'], ci['outliers_pct']))
for a, b, r, reliability in profile.get('high_correlations', []):
flags.append('钩稽候选:{}与{}强相关(r={}),可用于交叉验证 [{}]'.format(a, b, r, reliability))
# context差异化关注点
if context == '审计':
# 审计最怕:系统性缺失=控制缺陷、高重复率=重复记账、金额字段异常=操纵风险
for ci in profile['columns']:
if ci.get('missing_pct', 0) > 5 and ci.get('missing_pct', 0) <= 20:
flags.append('审计关注:字段[{}]缺失率{:.1f}%,虽未超20%但可能暗示控制缺陷'.format(ci['name'], ci['missing_pct']))
if ci.get('outliers_pct', 0) > 3 and any(kw in ci['name'] for kw in ['金额','价格','成本','费用']):
flags.append('审计关注:金额字段[{}]异常值占比{:.1f}%,存在操纵或舞弊风险'.format(ci['name'], ci['outliers_pct']))
elif context == '生产':
# 生产最怕:日期列缺失=排产混乱、合格率字段偏斜=数据不实
for ci in profile['columns']:
if 'date' in ci.get('dtype', '') or any(kw in ci['name'] for kw in ['日期','时间']):
if ci.get('missing_pct', 0) > 5:
flags.append('生产关注:日期字段[{}]缺失率{:.1f}%,排产与追溯可能不完整'.format(ci['name'], ci['missing_pct']))
if any(kw in ci['name'] for kw in ['合格率','损耗率','成活率']):
if ci.get('stats', {}).get('skewness', 0) > 1:
flags.append('生产关注:率指标[{}]偏度{:.1f},可能存在选择性记录'.format(ci['name'], ci['stats']['skewness']))
elif context in ('财务', 'BP', '管会'):
# 财务最怕:金额字段缺失=入账不全、高相关异常=调节空间
for ci in profile['columns']:
if any(kw in ci['name'] for kw in ['金额','价格','成本','费用','收入']):
if ci.get('missing_pct', 0) > 2:
flags.append('财务关注:金额字段[{}]缺失率{:.1f}%,可能存在未入账交易'.format(ci['name'], ci['missing_pct']))
for a, b, r, reliability in profile.get('high_correlations', []):
if abs(r) > 0.95:
flags.append('财务关注:{}与{}近乎完全相关(r={}),需核实是否同一数据源或调节关系 [{}]'.format(a, b, r, reliability))
elif context == '采购':
# 采购最怕:供应商/合同字段缺失=绕过管控、价格字段异常=利益输送
for ci in profile['columns']:
if any(kw in ci['name'] for kw in ['供应商','合同','订单']):
if ci.get('missing_pct', 0) > 3:
flags.append('采购关注:[{}]缺失率{:.1f}%,可能存在绕过管控的线下交易'.format(ci['name'], ci['missing_pct']))
elif context == '品管':
# 品管最怕:批次/检验字段缺失=追溯断裂、标准偏离字段异常
for ci in profile['columns']:
if any(kw in ci['name'] for kw in ['批次','检验','检测','标准']):
if ci.get('missing_pct', 0) > 3:
flags.append('品管关注:[{}]缺失率{:.1f}%,追溯链可能断裂'.format(ci['name'], ci['missing_pct']))
return flags
借鉴来源: finance插件 variance-analysis 的价量分解、率混合分解、瀑布图归因方法论
v4.0的"对比分析"模式只能告诉你"A和B差异显著",但不能回答"差异由什么驱动,各贡献多少"。差异分解就是把这个"黑箱"拆开。
适用:任何可表达为 单价 × 数量 的指标(收入、成本、采购额等)
总差异 = 实际值 - 基准值(预算/上期/标准)
数量效应 = (实际数量 - 基准数量) × 基准单价
价格效应 = (实际单价 - 基准单价) × 实际数量
验证:数量效应 + 价格效应 = 总差异
扩展:三因素分解(分离组合效应)
当指标可表达为 单价 × 数量 × 结构(多品类/多区域场景)时,双因素分解会产生残差——这就是组合效应。
结构 = 各品类数量占总数量之比(如A品类占60%、B品类占40%)
数量效应 = (实际数量 - 基准数量) × 基准单价 × 基准结构
价格效应 = (实际单价 - 基准单价) × 基准数量 × 实际结构
组合效应 = 基准单价 × 基准数量 × (实际结构 - 基准结构)
验证:数量效应 + 价格效应 + 组合效应 = 总差异
何时用双因素 vs 三因素:
注:三因素分解的完整函数实现与
rate_mix_decompose()共享同一方法论。对于"单价×数量"型指标的三因素分解,建议先按品类分别做价量分解,再用率混合分解归因组合效应。
def price_volume_decompose(actual_qty, actual_price, base_qty, base_price):
"""价量分解:返回数量效应、价格效应、总差异"""
volume_effect = (actual_qty - base_qty) * base_price
price_effect = (actual_price - base_price) * actual_qty
total_var = actual_qty * actual_price - base_qty * base_price
return {
'total_variance': round(total_var, 2),
'volume_effect': round(volume_effect, 2),
'price_effect': round(price_effect, 2),
'verification': round(volume_effect + price_effect, 2) == round(total_var, 2),
'volume_pct': round(volume_effect / total_var * 100, 1) if total_var != 0 else 0,
'price_pct': round(price_effect / total_var * 100, 1) if total_var != 0 else 0,
}
适用:分析加权平均指标(如毛利率、平均单价、投入产出比)的变化原因——是各分项本身的比率变了,还是分项之间的比例变了?
比率效应 = Σ(实际数量_i × (实际比率_i - 基准比率_i))
组合效应 = Σ(基准比率_i × (实际数量_i - 基准数量下按基准结构应分配的数量_i))
典型场景:加工综合毛利率下降了2个百分点——是各品类毛利都降了,还是低毛利品类占比增加了?
def rate_mix_decompose(df_actual, df_base, value_col, weight_col, category_col):
"""率混合分解:比率效应 vs 组合效应(按品类分组计算)
原理:
加权平均比率 = Σ(比率_i × 权重_i) / Σ(权重_i)
比率效应 = 各品类自身比率变化对加权平均的影响
组合效应 = 各品类权重(占比)变化对加权平均的影响
验证:比率效应 + 组合效应 = 实际加权平均 - 基准加权平均
"""
import pandas as pd
# 按品类聚合
actual = df_actual.groupby(category_col).agg({value_col: 'sum', weight_col: 'sum'})
base = df_base.groupby(category_col).agg({value_col: 'sum', weight_col: 'sum'})
# 对齐品类索引(处理某期有而另一期没有的品类)
all_cats = actual.index.union(base.index)
actual = actual.reindex(all_cats, fill_value=0)
base = base.reindex(all_cats, fill_value=0)
# 计算各品类比率(避免除零)
actual_rate = (actual[value_col] / actual[weight_col]).replace([float('inf'), float('-inf')], 0).fillna(0)
base_rate = (base[value_col] / base[weight_col]).replace([float('inf'), float('-inf')], 0).fillna(0)
# 计算权重占比
total_actual_weight = actual[weight_col].sum()
total_base_weight = base[weight_col].sum()
actual_mix = actual[weight_col] / total_actual_weight if total_actual_weight != 0 else actual[weight_col] * 0
base_mix = base[weight_col] / total_base_weight if total_base_weight != 0 else base[weight_col] * 0
# 加权平均比率(正确计算方式)
weighted_actual_rate = (actual_rate * actual_mix).sum()
weighted_base_rate = (base_rate * base_mix).sum()
total_change = weighted_actual_rate - weighted_base_rate
# 比率效应:各品类比率变化 × 实际权重
rate_effect = ((actual_rate - base_rate) * actual_mix).sum()
# 组合效应:基准比率 × 权重变化
mix_effect = (base_rate * (actual_mix - base_mix)).sum()
# 验证
verification = abs(rate_effect + mix_effect - total_change) < 1e-6
return {
'weighted_actual_rate': round(float(weighted_actual_rate), 4),
'weighted_base_rate': round(float(weighted_base_rate), 4),
'total_rate_change': round(float(total_change), 4),
'rate_effect': round(float(rate_effect), 4),
'mix_effect': round(float(mix_effect), 4),
'verification_passed': verification,
'interpretation': '比率效应为主→分项本身变化; 组合效应为主→结构变化',
# 各品类明细
'by_category': pd.DataFrame({
'actual_rate': actual_rate, 'base_rate': base_rate,
'actual_mix': actual_mix, 'base_mix': base_mix,
'rate_contribution': (actual_rate - base_rate) * actual_mix,
'mix_contribution': base_rate * (actual_mix - base_mix),
}).round(4).to_dict('index'),
}
将总差异拆解为各驱动因素的正负贡献,形成"瀑布"——从基准值出发,各因素依次增减,最终达到实际值。
def waterfall_attribution(drivers):
"""瀑布图归因:drivers = [(名称, 金额), ...] 金额正值=增,负值=减"""
sorted_drivers = sorted(drivers, key=lambda x: abs(x[1]), reverse=True)
total = sum(d[1] for d in sorted_drivers)
cumulative = 0
rows = []
for name, amount in sorted_drivers:
cumulative += amount
rows.append({
'driver': name,
'amount': amount,
'pct_of_total': round(amount / total * 100, 1) if total != 0 else 0,
'cumulative': round(cumulative, 2),
'direction': '有利' if amount > 0 else '不利',
})
# 合并小项(<5%贡献的归入"其他")
major = [r for r in rows if abs(r['pct_of_total']) >= 5]
minor = [r for r in rows if abs(r['pct_of_total']) < 5]
if minor:
other_amount = sum(r['amount'] for r in minor)
# 计算合并后的cumulative:major最后一项的cumulative + other_amount
last_cumulative = major[-1]['cumulative'] if major else 0
major.append({'driver': '其他(合计)', 'amount': round(other_amount, 2),
'pct_of_total': round(other_amount / total * 100, 1) if total != 0 else 0,
'cumulative': round(last_cumulative + other_amount, 2),
'direction': '有利' if other_amount > 0 else '不利'})
return {'total_variance': round(total, 2), 'drivers': major, 'driver_count': len(sorted_drivers)}
文本瀑布格式(无图表工具时使用):
瀑布归因:XX成本 — 本期 vs 预算
预算成本 ¥1,000,000
|
|--[+] 原料价格上涨 +¥80,000
|--[+] 产量增加(多产200吨) +¥50,000
|--[-] 采购议价降本 -¥30,000
|--[-] 工艺优化节耗 -¥15,000
|--[+] 人民币贬值增加进口成本 +¥5,000
|
实际成本 ¥1,090,000
净差异:+¥90,000 (+9.0% 不利)
每个重大差异的叙述必须包含:
**[项目名]**:[有利/不利]差异 ¥[金额] ([百分比]%)
vs [比较基准] for [期间]
驱动因素:[主要驱动因素描述]
[2-3句量化解释,每个驱动因素的具体金额贡献]
趋势判断:[一次性 / 预计持续 / 改善中 / 恶化中]
行动建议:[无需 / 持续观察 / 深入调查 / 更新预算]
叙述反模式(必须避免):
借鉴来源: finance插件 reconciliation 的3类差异分类 + 老化分析 + 升级阈值
v4.0的"钩稽比对"模式能找到差异,但没有对差异进行系统分类和生命周期管理。对账钩稽体系就是把"对不上的数"分为三类、跟踪老化、设置升级阈值。
| 类别 | 含义 | 典型例子 | 处理方式 |
|---|---|---|---|
| 时点差异 | 正常处理时差导致,后续期间自动消除 | 在途物资、未达账项、跨期入账 | 无需调整,跟踪至消除 |
| 需调整差异 | 记录错误或遗漏,需要做调整分录 | 金额录错、重复记账、漏记交易 | 编制调整分录 |
| 待查差异 | 无法立即解释,需深入调查 | 不明原因差额、争议金额 | 调查根因,文档化,必要时升级 |
跟踪未解决差异的"年龄",识别过期项:
| 老化区间 | 状态 | 行动 |
|---|---|---|
| 0-30天 | 正常 | 跟踪——在正常处理周期内 |
| 31-60天 | 老化 | 调查——跟进为何未消除 |
| 61-90天 | 过期 | 升级——通知主管,记录调查过程 |
| 90天+ | 呆滞 | 管理层升级——可能需要核销或强制调整 |
def aging_analysis(reconciling_items, as_of_date):
"""对账差异老化分析"""
aged = []
for item in reconciling_items:
age_days = (as_of_date - item['origin_date']).days
if age_days <= 30:
bucket, status, action = '0-30天', '正常', '跟踪'
elif age_days <= 60:
bucket, status, action = '31-60天', '老化', '调查'
elif age_days <= 90:
bucket, status, action = '61-90天', '过期', '升级主管'
else:
bucket, status, action = '90天+', '呆滞', '管理层升级'
aged.append({
**item,
'age_days': age_days, 'bucket': bucket,
'status': status, 'action': action,
})
return aged
| 触发条件 | 示例阈值 | 升级对象 |
|---|---|---|
| 单项差异金额 | > ¥50,000 | 主管审核 |
| 单项差异金额 | > ¥200,000 | 部门负责人审核 |
| 差异总额 | > ¥500,000 | 部门负责人审核 |
| 差异年龄 | > 60天 | 主管跟进 |
| 差异年龄 | > 90天 | 管理层审核 |
| 未解释差异 | 任何金额 | 不能结账——必须解决或文档化 |
| 连续增长 | 3期以上 | 流程改进调查 |
钩稽结果:[表A名] × [表B名] — [期间]
A表总量: ¥XX,XXX
B表总量: ¥XX,XXX
--------
初始差异: ¥X,XXX
加: 时点差异项
[项目描述] ¥X,XXX
[项目描述] ¥X,XXX
--------
小计: ¥X,XXX
减: 需调整差异项
[项目描述] (¥X,XXX)
[项目描述] (¥X,XXX)
--------
小计: (¥X,XXX)
调整后差异: ¥X,XXX
待查差异项:
[项目描述] ¥X,XXX (老化XX天, [状态])
[项目描述] ¥X,XXX (老化XX天, [状态])
最终未解释差异: ¥X,XXX
借鉴来源: finance插件 variance-analysis 的分层阈值 + audit-support 的质量阈值 + sox-testing 的样本量策略
v4.0缺少统一的"什么才算重要"的判断标准。不同规模的数字、不同类型的分析,重要性门槛应该不同。
原则:金额越大,百分比阈值越低;波动性越高,阈值可适当放宽。
| 项目规模 | 金额阈值 | 比例阈值 | 触发条件 |
|---|---|---|---|
| > ¥1,000万 | ¥50万 | 5% | 任一超过即触发 |
| ¥100万\~1,000万 | ¥10万 | 10% | 任一超过即触发 |
| \< ¥100万 | ¥5万 | 15% | 任一超过即触发 |
比较类型的差异化阈值:
| 比较类型 | 比例阈值 | 说明 |
|---|---|---|
| 实际 vs 预算 | 10% | 预算差异通常更受关注 |
| 实际 vs 上期 | 15% | 环比波动较大属正常 |
| 实际 vs 预测 | 5% | 预测应更准确,偏差更敏感 |
| 环比(MoM) | 20% | 月度波动天然较大 |
| 同比(YoY) | 10% | 消除季节性后的变化更值得关注 |
当多个差异超过阈值时,按以下优先级排列:
def prioritize_variances(variances, threshold_config=None):
"""差异优先级排序——5级综合排序
输入字典需含字段:
variance_amount: 差异金额
variance_pct: 差异比例
base_amount: 基准金额(用于分层阈值)
可选字段(缺失时该维度不参与排序):
is_unexpected: 是否方向反预期(bool)
is_new: 是否新出现的差异(bool)
is_growing: 是否连续增长(bool)
"""
config = threshold_config or {
'amount_tiers': [(1e7, 5e5, 0.05), (1e6, 1e5, 0.10), (0, 5e4, 0.15)],
}
flagged = []
for v in variances:
base = abs(v.get('base_amount', 0))
for tier_base, amt_thr, pct_thr in config['amount_tiers']:
if base >= tier_base:
if abs(v['variance_amount']) >= amt_thr or abs(v['variance_pct']) >= pct_thr:
v['flagged'] = True
v['threshold_used'] = '金额≥¥{:,.0f} 或 比例≥{:.0%}'.format(amt_thr, pct_thr)
break
if v.get('flagged'):
flagged.append(v)
# 5级综合排序:每级给分,总分降序
for v in flagged:
score = 0
# 1. 绝对金额(归一化到0-40分,最大金额=40分)
max_amt = max(abs(x['variance_amount']) for x in flagged) if flagged else 1
score += abs(v['variance_amount']) / max_amt * 40 if max_amt != 0 else 0
# 2. 比例差异(归一化到0-25分)
max_pct = max(abs(x['variance_pct']) for x in flagged) if flagged else 1
score += abs(v['variance_pct']) / max_pct * 25 if max_pct != 0 else 0
# 3. 方向反预期(+15)
score += 15 if v.get('is_unexpected') else 0
# 4. 新出现的差异(+10)
score += 10 if v.get('is_new') else 0
# 5. 累计扩大趋势(+10)
score += 10 if v.get('is_growing') else 0
v['priority_score'] = round(score, 1)
flagged.sort(key=lambda x: x.get('priority_score', 0), reverse=True)
return flagged
| 指标 | 预算 | 预测 | 实际 | 预算差异(¥) | 预算差异(%) | 预测差异(¥) | 预测差异(%) |
|---|---|---|---|---|---|---|---|
| [科目1] | ¥X | ¥X | ¥X | ¥X | X% | ¥X | X% |
三向比较的用途:
| 规则 | 说明 |
|---|---|
| 只读原则 | 数据库连接仅执行SELECT,严禁INSERT/UPDATE/DELETE/DROP |
| LIMIT保护 | 无LIMIT查询自动添加 LIMIT 10000 |
| 注入防护 | SQL参数化查询,不拼接用户输入 |
| 代码沙盒 | 禁止 os.system, subprocess, exec, eval |
| 权限边界 | 仅访问用户授权的数据库和表,不探测未授权对象 |
| 超时控制 | 查询设置 max_execution_time,Python脚本设60秒超时 |
| 数据类型 | 识别规则 | 脱敏策略 |
|---|---|---|
| 身份证号 | 15/18位数字+校验位 | 保留前3后4:340***1234 |
| 手机号 | 11位,1开头 | 保留前3后4:138****5678 |
| 银行卡号 | 16-19位数字 | 保留前6后4:622848****1234 |
| 金额字段 | 含"金额""价格""成本"等关键词 | 报告中仅展示汇总统计,明细用区间 |
| 供应商名称 | 含"公司""有限"等 | 正式报告可全称,外发材料用代号 |
def mask_field(value, data_type):
"""敏感字段脱敏"""
v = str(value)
if data_type == 'id_card':
return v[:3] + '***' + v[-4:] if len(v) >= 7 else '***'
elif data_type == 'phone':
return v[:3] + '****' + v[-4:] if len(v) >= 7 else '***'
elif data_type == 'bank_card':
return v[:6] + '****' + v[-4:] if len(v) >= 10 else '***'
return v
| 问题类型 | 分析模式 | 推荐方法 | 输出形式 |
|---|---|---|---|
| "两个数对不对得上?" | 钩稽比对 | 外关联merge + 差异率计算 | 差异明细表 + 差异率柱状图 |
| "这个数正不正常?" | 异常检测 | IQR/Z-score + 阈值筛选 | 异常记录清单 + 分布图 |
| "趋势有没有问题?" | 趋势分析 | 同比环比 + 变点检测(CUSUM) | 趋势折线图 + 预警标注 |
| "两组有没有差异?" | 对比分析 | t检验/卡方检验/ANOVA | 检验结论 + 对比柱状图 |
| "合不合规?" | 合规检测 | 规则词典匹配 + 制度条款映射 | 违规清单 + 制度对照表 |
| "数据质量怎么样?" | 数据画像 | audit_profile() 一键画像 |
质量评分 + 关注点清单 |
| "成本结构如何?" | 构成分析 | 分组占比 + 帕累托分析 | 构成饼图 + 累计曲线 |
| "投入产出怎样?" | 效率分析 | 比率计算 + 标杆对比 | 效率指标表 + 标杆对比图 |
完整方法论见 1.10 统计增强工具箱——含假设检验框架、效应量/置信区间/样本量考量、统计陷阱警告。
| 对比类型 | 数据条件 | 推荐方法 | Python实现 |
|---|---|---|---|
| 两组均值对比 | 正态+等方差 | 独立样本t检验 | scipy.stats.ttest_ind |
| 两组均值对比 | 非正态 | Mann-Whitney U | scipy.stats.mannwhitneyu |
| 多组均值对比 | 正态+等方差 | 单因素ANOVA | scipy.stats.f_oneway |
| 比例对比 | 计数数据 | 卡方检验 | scipy.stats.chi2_contingency |
| 前后对比 | 配对数据 | 配对t检验 | scipy.stats.ttest_rel |
| 分布对比 | 任意分布 | KS检验 | scipy.stats.ks_2samp |
from scipy import stats
def auto_compare(group_a, group_b, alpha=0.05):
"""自动选择检验方法并返回结论
正态性检验策略:n<=2000用Shapiro-Wilk,n>2000用D'Agostino-Pearson(大样本更稳健)。
抽样策略:超过2000时随机抽样,避免取前N条的时间/排序偏差。
"""
import numpy as np
# 正态性检验
for grp, label in [(group_a, 'A'), (group_b, 'B')]:
if len(grp) < 3:
return {'method': '样本不足', 'statistic': None, 'p_value': None,
'conclusion': '样本量<3,无法进行统计检验'}
def _test_normality(data, alpha=0.05):
"""正态性检验:n<=2000用Shapiro-Wilk,n>2000用D'Agostino"""
if len(data) <= 2000:
_, p = stats.shapiro(data)
else:
# 随机抽样2000条,避免排序偏差
sample = np.random.choice(data, size=2000, replace=False)
_, p = stats.normaltest(sample)
return p > alpha
is_normal_a = _test_normality(group_a, alpha)
is_normal_b = _test_normality(group_b, alpha)
if is_normal_a and is_normal_b:
stat, p_value = stats.ttest_ind(group_a, group_b)
method = '独立样本t检验'
else:
stat, p_value = stats.mannwhitneyu(group_a, group_b, alternative='two-sided')
method = 'Mann-Whitney U检验'
significant = p_value < alpha
conclusion = '差异显著' if significant else '差异不显著'
return {'method': method, 'statistic': round(float(stat), 4),
'p_value': round(float(p_value), 6), 'conclusion': conclusion}
当首选方案不可用时,逐级降级而非直接失败:
全量分析 → 采样分析 → 聚合统计 → 提示用户数据过大需手动处理
精确计算 → 近似计算 → 估算 → 告知精度限制
实时查询 → 缓存数据 → 历史快照 → 告知数据时效
L1精确匹配 → L2桥接匹配 → L3编码匹配 → 标记"需人工核实"
| 错误类型 | 首选方案 | 降级方案 |
|---|---|---|
| MemoryError | 全量加载 | chunksize 分块 + 逐块聚合 |
| 查询超时 | 全表关联 | 先WHERE过滤再JOIN,或子查询分步 |
| 编码错误 | UTF-8 | 尝试GBK → Latin-1 → 忽略错误字符 |
| 连接超时 | 直连数据库 | 重试3次(1s/3s/5s递增) → 建议导出CSV本地分析 |
| 列名不匹配 | 精确列名merge | iloc位置索引 + 列名映射字典 |
| 数据为空 | 全量分析 | 给出友好提示,建议检查过滤条件 |
### 发现N:{标题}
**数据锚点**:{量化结论 + 置信度/显著性}
**制度/标准红线**:{违反的条款编号及内容(审计场景)或 偏离的预算/标准(财务场景)}
**等级**:🔴高 / 🟡中 / 🟢低
**→行动指引**:{审计→现场怎么查;财务→需调整什么;生产→需改进什么}
**→建议**:{优先级 + 预期效果 + 实施难度}
借鉴来源: Scene #8 "数据分析及可视化" 插件
statistical-analysis+data-validation精华
v5.1的1.7提供了统计检验方法选择指南,但缺少:
本模块补全这些能力。
| 数据特征 | 使用 | 原因 | 审计示例 |
|---|---|---|---|
| 对称分布,无异常值 | 均值 | 最有效估计量 | 标准化产品的单位成本 |
| 偏态分布 | 中位数 | 抗异常值 | 供应商单笔金额(少数大单拉高均值) |
| 分类/定序数据 | 众数 | 仅非数值选项 | 违规类型分布 |
| 高度偏态+极端值 | 中位数+均值 | 差距揭示偏斜程度 | 采购单价(均>中是供应商集中信号) |
铁律:审计场景中的金额/成本/价格指标,均值和中位数同时报告。两者差距越大,数据越偏斜,单看均值越误导。
每遇到一个数值分布,必须回答五问:
| 维度 | 问题 | 方法 |
|---|---|---|
| 形态 | 正态/右偏/左偏/双峰/均匀/重尾? | 偏度+直方图+核密度估计 |
| 中心 | 均值和中位数各是多少?差距大吗? | mean() + median() |
| 离散 | 标准差还是IQR更合适? | 正态→标准差;偏态→IQR |
| 异常值 | 有多少?多极端?集中在什么区间? | Z-score/IQR/百分位法 |
| 边界 | 是否有自然下限(0)或上限(100%)? | 业务逻辑检验 |
报告关键百分位以讲出比均值更丰富的故事:
p1: 底部1%(最低典型值)
p5: 正常范围低端
p25: 第一四分位
p50: 中位数(典型值)
p75: 第三四分位
p90: 前10%(高产/高耗区间)
p95: 正常范围高端
p99: 前1%(极端值)
审计叙事示例:
"供应商单笔金额中位数为¥8.5万,但前10%的供应商单笔金额超过¥35万,将均值拉至¥14.2万。这一偏斜提示需关注前10%供应商是否存在异常交易。"
| 审计场景 | 检验方法 | Python | 注意事项 |
|---|---|---|---|
| 两供应商单价是否有显著差异 | 独立t检验 | scipy.stats.ttest_ind |
正态+等方差 |
| 同上(非正态) | Mann-Whitney U | scipy.stats.mannwhitneyu |
报告中位数差而非均值差 |
| 某车间合格率下降是否真实 | z检验(比例) | statsmodels.stats.proportion.proportions_ztest |
需样本量和事件数 |
| 多事业部费用率是否随机差异 | 单因素ANOVA | scipy.stats.f_oneway |
正态+等方差 |
| 整改前后指标是否真的改善 | 配对t检验 | scipy.stats.ttest_rel |
同一对象前后对比 |
| 违规事件与班次是否相关 | 卡方检验 | scipy.stats.chi2_contingency |
期望频数≥5 |
| 两批数据分布是否相同 | KS检验 | scipy.stats.ks_2samp |
任意分布均可用 |
统计显著:差异不太可能是随机所致。
实际重要:差异大到影响业务决策。
大样本下,极小的差异也可能统计显著但毫无实际意义。审计报告中必须同时报告:
| 必须报告 | 示例 | 为什么 |
|---|---|---|
| 效应量 | "B组合格率比A组高2.3个百分点" | 差异有多大 |
| 置信区间 | "95%置信区间[1.1%, 3.5%]" | 真实差异的可能范围 |
| 业务影响 | "相当于每万枚原料多出230枚合格蛋" | 翻译为业务语言 |
反模式:只报p值("p\<0.01,差异显著")而不报差异大小——这是统计学,不是审计结论。
| 规则 | 审计应用 |
|---|---|
| 小样本结果不可靠,哪怕p值显著 | 单月\<30笔的交易类型,不做统计推断 |
| 比例检验每组至少30个事件 | 违规率\<1%的部门,需要千级样本才有检测力 |
| 检测小效应需要大样本 | 要检测1%的差异,可能需要每组上千样本 |
| 样本不足时如实说明 | "本月仅12笔此类交易,不足以做统计检验,以下为描述性分析" |
当检验多个假设时,必有部分"显著"纯属碰巧。
Bonferroni校正:α_adjusted = 0.05 / N(N为检验次数)
审计场景:同时对比30个供应商的单价 → α调整为0.05/30≈0.0017
或者直接放弃校正,坦诚报告:"我们检验了30项,以下X项差异显著(未校正),需结合业务判断跟进优先级。"
你只能分析"存活"到数据集中的实体。
| 审计场景 | 幸存者偏差陷阱 |
|---|---|
| 仅分析在册供应商 | 已被淘汰/拉黑的供应商不在数据中,均价被人为压低 |
| 仅分析当月在产车间 | 已关停车间的问题被自动排除 |
| 仅分析有交易的客户 | 零交易客户(可能因纠纷)被忽略 |
审计铁律:每次分析前问——"谁不在这份数据里?他们的缺席是否改变结论?"
对已有平均值再求平均,忽略各组样本量不同 → 结果错误。
案例:
(95%+70%)/2 = 82.5%(简单平均)(100×95%+10×70%)/110 = 92.7%(加权平均)预防:永远从原始数据聚合,不对已有聚合结果二次求平均。
群体层面的结论不适用于个体。
v5.1提供了Z-score/IQR/百分位三种检测方法,v5.2补充处理原则。
禁止自动删除离群值。改为四步:
必须报告处理方式:"排除47笔单笔>¥50万的交易(0.3%),这些为批量采购,单独分析。"
| 类型 | 特征 | 审计含义 |
|---|---|---|
| 点异常 | 单点偏离 | 可能是录入错误或单次异常交易 |
| 变点 | 持续偏移 | 可能是流程变化、政策调整或系统性舞弊 |
检测方法:
| 方法 | 公式 | 适用场景 |
|---|---|---|
| 简单增长率 | (本期-上期)/上期 |
单期对比 |
| 年复合增长率(CAGR) | (期末/期初)^(1/年数)-1 |
多年趋势 |
| 对数增长率 | ln(本期/上期) |
波动大的序列 |
审计应用:养殖产蛋率/加工消耗量/生产量的季节性属于正常波动,异常检测不能把季节性当异常警报。
借鉴
data-validation的结构化QA框架。分析报告交付前逐项检查。
[ ] 数据源已确认正确(表/文件/日期)
[ ] 时间范围完整,无意外断层
[ ] 关键列缺失率已检查,空值处理方式已记录
小葱技能站7w4.net发现了升级插件。
[ ] 无因JOIN类型错误导致的重复计数
[ ] 过滤条件已验证,无意外的排除
[ ] GROUP BY已包含所有非聚合列
[ ] 比率/百分比的分子分母正确,分母非零
[ ] JOIN类型适当(INNER vs LEFT),多对多JOIN未膨胀行数
[ ] 指标定义与利益相关方一致
[ ] 数值在合理范围内(金额非负、百分比在0-100%)
[ ] 时间序列无未解释的跳变
[ ] 关键数字与已知基准交叉验证
[ ] 边界情况已考虑(空分组、零活动期、新增实体)
清单中的"红旗信号"在审计场景下尤为关键——完美的一致性和精确的整数往往指向数据操纵而非自然业务结果。
| 产业链环节 | 核心业务 | 典型分析维度 | 高频问题 | 推荐分析模式 |
|---|---|---|---|---|
| 祖代产品 | 引种→育成→产蛋 | 引种成本、育成率、产蛋性能 | "引种投入产出合理吗?" | 效率分析+构成分析 |
| 父母代养殖 | 孵化→养殖→原料生产 | 合格率、孵化率、来源追溯 | "投产对不对得上?""来源可追溯吗?" | 钩稽比对+合规检测 |
| 肉产品养殖 | 雏产品→育肥→产出 | 成活率、投入产出比、产出均重 | "损耗正不正常?" | 异常检测+趋势分析 |
| 加工加工 | 原料采购→配方→生产 | 原料损耗、配方成本、产出率 | "原料去哪了?""成本结构如何?" | 钩稽比对+构成分析 |
| 生产初加工 | 活产品→白条→分割 | 出肉率、副产回收、损耗率 | "出肉率达标吗?" | 对比分析+效率分析 |
| 流通初加工 | 原料→水洗→分拣 | 出品率、质量标准、加工损耗 | "出品率波动原因?" | 趋势分析+异常检测 |
| 副产加工 | 产品肠/产品血/产品毛等 | 回收率、加工产出、废弃率 | "副产回收充分吗?" | 效率分析+构成分析 |
| 场景 | 首选分析 | 辅助分析 | 核心图表 | 报告风格 |
|---|---|---|---|---|
| 原料挑选→上孵钩稽 | 钩稽比对 | 合规检测 | 差异率柱状图 | 风险导航式 |
| 加工原料出入库 | 钩稽比对 | 构成分析 | 差异明细表+成本构成饼图 | 对比归因式 |
| 肉产品产出效率 | 效率分析 | 趋势预警 | 投入产出比趋势图+标杆线 | 运营监控式 |
| 流通出品率 | 趋势分析 | 异常检测 | 出品率折线图+异常标注 | 运营监控式 |
| 供应商合规 | 合规检测 | 对比分析 | 违规清单+对照表 | 合规审计式 |
| 预算执行差异 | 对比分析 | 构成分析 | 预算vs实际对比柱状图 | 偏差归因式 |
| 成本结构分析 | 构成分析 | 趋势分析 | 帕累托图+成本趋势线 | 决策支持式 |
| 投入产出分析 | 效率分析 | 对比分析 | 投入产出散点图+标杆 | 效率评估式 |
核心需求:风险排查、钩稽核查、合规检测、历史对照
报告风格:风险导航式(TOP3风险前置→数据全景→分层支撑→制度对照,→见阶段5.1)
分析侧重:
核心需求:预算差异分析、成本结构拆解、投入产出比、费用趋势
报告风格:偏差归因式(差异概览→价量分解→趋势→调整建议,→见阶段5.1)
分析侧重:
核心需求:合格率/损耗率趋势、异常批次定位、效率标杆对比
报告风格:运营监控式(KPI卡片→趋势图→异常标注→改进措施,→见阶段5.1)
分析侧重:
核心需求:供应商合规、价格异常、合同执行率
报告风格:合规审计式(合规总览→违规清单→对照表→处置建议,→见阶段5.1)
分析侧重:
核心需求:质量指标趋势、批次追溯、标准偏离
报告风格:质量报告式(质量评分→指标趋势→批次追溯→整改要求,→见阶段5.1)
分析侧重:
借鉴来源: finance插件 audit-support + sox-testing 的CEAVOP认定+样本选择+缺陷分级方法论
当审计需求涉及内控测试时,将分析目标映射到6类认定:
| 认定 | 英文 | 含义 | 典型审计场景 |
|---|---|---|---|
| 完整性 | Completeness | 所有交易均已记录 | 入库单vs系统记录、费用是否全部入账 |
| 存在/发生 | Existence/Occurrence | 交易真实发生 | 销售是否虚构、存货是否真实存在 |
| 准确性 | Accuracy | 金额记录正确 | 单价×数量=金额验证、计算复核 |
| 计价 | Valuation | 资产/负债计价合理 | 存货跌价准备、坏账计提充分性 |
| 权利/义务 | Rights/Obligations | 权属清晰 | 固定资产归属、或有负债识别 |
| 列报 | Presentation | 分类披露恰当 | 关联方交易披露、会计政策一致性 |
| 控制频率 | 总体量(约) | 低风险样本 | 中风险样本 | 高风险样本 |
|---|---|---|---|---|
| 年度 | 1 | 1 | 1 | 1 |
| 季度 | 4 | 2 | 2 | 3 |
| 月度 | 12 | 2 | 3 | 4 |
| 每周 | 52 | 5 | 8 | 15 |
| 每日 | \~250 | 20 | 30 | 40 |
| 每笔(小量) | \<250 | 20 | 30 | 40 |
| 每笔(大量) | 250+ | 25 | 40 | 60 |
样本选择方法:
| 级别 | 定义 | 指示信号 |
|---|---|---|
| 缺陷 | 控制设计或运行不足以防止/发现错报 | 评估可能性+金额+是否有补偿控制 |
| 重要缺陷 | 严重程度低于重大缺陷但需治理层关注 | 可能导致超过非重要性但未达重要性的错报 |
| 重大缺陷 | 有合理可能性导致重大错报未被发现 | 管理层舞弊(任何金额)、重大错报重述、审计师发现公司内控未发现的重大错报 |
缺陷聚合:单项不重要的缺陷,同一流程/认定内合并后可能构成重要缺陷甚至重大缺陷。
当用户需求不在预设场景中时,按以下流程适配:
用户需求:"帮我看看流通车间最近三个月的加工损耗是不是有异常"
→ 识别本质:趋势分析("最近三个月")+ 异常检测("是不是有异常")
→ 匹配场景:流通初加工(L3模板)
→ 调整参数:时间范围=最近3个月,关注指标=加工损耗率,阈值=历史均值±2σ
→ 输出风格:运营监控式
使用规则:L3模板是示例代码,按需加载。不要在通用逻辑中硬编码特定场景的字段名或业务规则。
适用:原料挑选表 × 上孵表的数量匹配与差异核查
关键步骤:
来源分类示例(按实际数据调整模式):
import re
def classify_source(row, name_col_idx=0, supplier_col_idx=7, zone_col_idx=3):
"""三级降级来源分类——以养殖场景为例
注意:row 为 df.apply(func, axis=1) 传入的 Series,用 iloc[n] 一维索引
"""
farmer = str(row.iloc[name_col_idx]).strip()
supplier = str(row.iloc[supplier_col_idx]).strip()
zone_raw = str(row.iloc[zone_col_idx]).strip()
# 第一级:名称模式匹配
if re.search(r'[A-Z]{2}\d+', farmer):
return '基地来蛋'
if re.search(r'有限公司|农牧|牧业|养殖场', farmer):
return '社会外来蛋'
if '某区域' in zone_raw or '内蒙' in farmer:
return '社会外来蛋'
# 第二级:供应商列有值→外来
if supplier not in ('', 'nan', 'NaN'):
return '社会外来蛋'
# 第三级:默认基地
return '基地来蛋'
核心教训:
astype(str) 后 NaN 变成字符串 'nan',需同时判断 '', 'nan', 'NaN'适用:按日/周/月的合格率趋势监测与异常预警
关键步骤:
适用:原料采购入库 × 领料出库 × 库存盘点的三方匹配
关键步骤:
适用:标准配方成本 vs 实际生产成本的差异分析
关键步骤:
适用:肉产品产出成活率、投入产出比、产出均重的趋势与异常
关键步骤:
适用:白条出肉率、副产(产品肠/产品血/产品毛)回收率分析
关键步骤:
适用:原料毛→水洗→分拣各环节的出品率趋势与损耗
关键步骤:
适用:各部门/事业部的预算vs实际对比
关键步骤:
适用:产品/事业部/期间的成本构成与变动分析
关键步骤:
适用:各事业部/产品的投入产出效率评估
关键步骤:
借鉴来源: finance插件 close-management 的依赖图+瓶颈分析
适用:财务部月度结账效率分析、结账流程改进
关键步骤:
结账依赖图模板:
Level 1(无依赖,T+1立即启动):
├── 现金收支录入
├── 银行对账单获取
├── 固定资产折旧运行
├── 预付费用摊销
├── 应付计提准备
└── 内部交易过账
Level 2(依赖Level 1完成):
├── 银行对账(需:现金分录+银行对账单)
├── 收入确认(需:出库/交付数据最终确认)
├── 应收子账核对(需:所有收入/现金分录)
├── 应付子账核对(需:所有应付分录/计提)
└── 汇率重估(需:所有外币分录过账)
Level 3(依赖Level 2完成):
├── 所有资产负债表对账
├── 内部交易对账与抵消
├── 对账发现的调整分录
└── 试算平衡表初稿
Level 4(依赖Level 3完成):
├── 税务计提
├── 合并与抵消
├── 财务报表初稿
└── 差异分析
Level 5(依赖Level 4完成):
├── 管理层审阅
├── 最终调整
├── 封账 / 期间锁定
└── 报告发布
常见瓶颈与解决方案:
| 瓶颈 | 根因 | 解决方案 |
|---|---|---|
| 应付计提延迟 | 等待部门确认 | 推行持续计提估算;设置截止时间 |
| 手工日记账 | 每月手工编制重复分录 | ERP中自动化标准循环分录 |
| 对账缓慢 | 每月从零开始 | 推行持续/滚动对账 |
| 内部交易延迟 | 等待对方确认 | 自动化内部交易匹配;设更严截止时间 |
| 管理层审阅后大额调整 | 审阅中发现问题 | 改进前期审核流程;提前发现问题 |
适用:养殖/加工/制造/流通/副产各事业部的核心效率指标横向对比
关键步骤:
注意事项:
适用:集团级关键指标(收入/利润/成本/质量)的月度/季度趋势与异常预警
关键步骤:
注意事项:
在写任何分析代码前,先回答三个问题:
输出:策略映射表
| 核心问题 | 对应分析 | 可视化 | 行动指引 |
|---|---|---|---|
| 问题A | 分析模块X | 图表类型 | 指向什么行动 |
| 问题B | 分析模块Y | 图表类型 | 指向什么行动 |
在正式探索列名之前,对数据集执行 audit_profile() 一键画像:
profile = audit_profile(df, name='数据集名称')
flags = profile_to_flags(profile, context='审计') # context按部门选择
for f in flags:
print('⚠', f)
画像结果直接输入策略映射——缺失率高的字段不适合做主匹配键,异常值占比高的字段优先做风险分析。
import pandas as pd
df = pd.read_excel('xxx.xlsx', sheet_name='Sheet1', header=0)
print("shape:", df.shape)
print("columns:", list(df.columns))
print("head:\n", df.head(3))
print(df.iloc[:, 2].unique()[:10]) # 用位置索引,不用列名
核心原则:
iloc[:, n] 位置索引,不依赖中文列名(中文列名含空格/换行/合并单元格时极易出错)df.shape 确认行列数,防止遗漏 sheet 或误读 header# 过滤合计/汇总行
df = df[df.iloc[:, 0] != '合计'].copy()
df = df[df.iloc[:, 0].notna()].copy()
# 统一日期格式
df['_dt'] = pd.to_datetime(df.iloc[:, 2], errors='coerce')
# 数值转换
df['_qty'] = pd.to_numeric(df.iloc[:, 10], errors='coerce')
# 字符串标准化
df['_key'] = df.iloc[:, 5].astype(str).str.strip()
关键注意:
errors='coerce' 是数值/日期转换的标配,非法值变 NaN 而非报错.copy() 防止 SettingWithCopyWarningisna() | (str.strip() == '') 双重判断rows = []
for key, group in df.groupby(df.iloc[:, 1]):
total = pd.to_numeric(group.iloc[:, 10], errors='coerce').sum()
avg_rate = pd.to_numeric(group.iloc[:, 37], errors='coerce').mean()
rows.append({
'分组键': str(key),
'总量': int(total),
'平均值': round(float(avg_rate) * 100, 2) if pd.notna(avg_rate) else 0,
})
# pandas 2.x 中 agg(col=(Series, 'func')) 语法不再支持,优先用 for 循环
# Layer 1: 精确名称匹配
match_l1 = pd.merge(df_a, df_b, on=['date', 'zone', 'name'], how='inner')
# Layer 2: 桥接键匹配(当主键不匹配时,用辅助字段桥接)
# 构建 name ↔ supplier 映射字典作为第二匹配键
# Layer 3: 编码匹配(提取通用编码模式)
def extract_code(name, pattern=r'([A-Z]{2}\d+)'):
m = re.search(pattern, str(name))
return m.group(1) if m else None
匹配结果文档化:
match_report = {
'L1_exact_match': len(match_l1),
'L2_bridge_match': len(match_l2),
'L3_code_match': len(match_l3),
'unmatched_after_L3': len(unmatched),
'note': 'L2/L3匹配的记录需在报告中标注"间接匹配,需人工核实"'
}
关键教训:跨表名称不一致会导致大量记录"消失"。永远在匹配后检查 unmatched 数量级——如果达到异常量级,一定是匹配策略有问题。
# 规则词典方式:关键词/阈值 → 违规标记 + 条款映射
RULES = {
'禁用对象交易': {'keyword': '禁用|黑名单|停用', 'field_idx': 6, 'clause': '采购管理制度第5条'},
'低于标准阈值': {'field_idx': 37, 'op': '<', 'value': 0.85, 'clause': '管理办法第12条'},
}
# 对每条规则检测数据,在结果中附上条款编号
通过 weknora-connector 检索相关制度条款,将分析发现自动映射到制度依据。
import json
results = { ... } # 分析结果字典
with open('analysis_results.json', 'w', encoding='utf-8') as f:
json.dump(results, f, ensure_ascii=False, indent=2, default=str)
分析与报告解耦:先生成 JSON,再由独立脚本读取生成报告。修改报告样式无需重跑分析。
| 调用部门 | 报告风格 | 结构特征 |
|---|---|---|
| 审计监察 | 风险导航式 | TOP3风险前置→数据全景→分层支撑→制度对照 |
| 财务BP | 偏差归因式 | 差异概览→价量分解→趋势→调整建议 |
| 生产车间 | 运营监控式 | KPI卡片→趋势图→异常标注→改进措施 |
| 采购中心 | 合规审计式 | 合规总览→违规清单→对照表→处置建议 |
| 品管部 | 质量报告式 | 质量评分→指标趋势→批次追溯→整改要求 |
<script src="https://cdn.jsdelivr.net/npm/chart.js@4.4.0/dist/chart.umd.min.js"></script>
<div class="chart-container" style="max-width:700px;margin:20px auto">
<canvas id="chart_trend"></canvas>
</div>
<script>
const trendData = JSON.parse(document.getElementById('data_trend').textContent);
new Chart(document.getElementById('chart_trend'), {
type: 'line',
data: {
labels: trendData.labels,
datasets: trendData.datasets
},
options: {
responsive: true,
plugins: {
annotation: {
annotations: {
threshold: {
type: 'line', yMin: 85, yMax: 85,
borderColor: '#E67E22', borderWidth: 2, borderDash: [6,4],
label: { content: '警戒线', enabled: true }
}
}
}
}
}
});
</script>
数据注入方式:<script type="application/json" id="data_xxx"> 标签嵌入JSON。
from docx import Document
from docx.shared import Pt, RGBColor
doc = Document()
t = doc.add_table(rows=1 + len(data_rows), cols=len(headers))
t.style = 'Light Grid Accent 1'
for i, h in enumerate(headers):
t.rows[0].cells[i].text = str(h)
for ri, row in enumerate(data_rows):
for ci, val in enumerate(row):
t.rows[ri+1].cells[ci].text = str(val)
doc.save('分析报告.docx')
定位:报告主体用静态图表保证归档确定性,附录可附带交互式数据看板供深入查阅。 适用:领导要"自己翻翻数据",或需要多维度交叉筛查时。 不适合:正式归档件、对外报送件(交互元素打印后失效)。
| 位置 | 内容 | 交互性 | 归档性 |
|---|---|---|---|
| 报告主体 | 核心发现 + 静态证据图 | 无(线性阅读) | ✅ 可打印归档 |
| 附录看板 | 多维度筛选 + 联动图表 + 明细下钻 | ✅ 点击/筛选联动 | ❌ 仅电子版 |
技术实现:ECharts单文件看板(筛选器变化→全图重绘,图表点击联动)
看板输出规则:
xxx_dashboard_appendix.html,报告中仅放链接@media print 隐藏筛选器图表不是报告装饰,是决策辅助工具。 每张图表都应有明确的"看了能做什么"的落点。
| 图表类型 | 适用场景 | 典型落点 |
|---|---|---|
| 折线图(趋势) | 指标随时间变化 | 定位异常时间窗口,锁定排查日期 |
| 柱状图(对比) | 分组/期间对比 | 定位偏差最大的维度,决定关注方向 |
| 环形图(占比) | 构成分布 | 量化关键占比,评估结构合理性 |
| 散点图(关联) | 两变量关系 | 检测相关性,识别离群点 |
| 帕累托图 | 成本/问题集中度 | 聚焦关键少数(80/20法则) |
| 热力图 | 多维度交叉 | 快速定位异常交叉点 |
| # | 错误类型 | 根因 | 解决方案 |
|---|---|---|---|
| 1 | ValueError: Length mismatch |
手动赋列名数量与实际列数不符 | 不手动赋列名,全用 iloc 位置索引 |
| 2 | TypeError: unhashable type 'Series' |
pandas 2.x agg 语法变更 |
改用 for key, group in df.groupby(...) 循环 |
| 3 | SyntaxError: invalid syntax |
变量名含空格 | 变量名用下划线,先做语法检查 |
| 4 | JSON 序列化失败 | numpy.int64 不可序列化 |
json.dump(..., default=str) 或显式 int() |
| 5 | HTML 特殊字符异常 | < > & 未转义 |
先转义再拼接,不在f-string里二次转义 |
| 6 | 来源分类误判 | 仅依赖单字段+astype(str)后NaN判断混乱 |
三级降级分类+同时判断''/'nan'/'NaN' |
| 7 | 跨表匹配大量"消失" | 名称不一致导致merge失败 | 多层匹配+必须检查unmatched数量级 |
| 8 | 报告面面俱到无焦点 | 所有分析并列,无重点区分 | 前置策略映射,高价值深挖低价值简述 |
以下维度尚未内置为独立模块,但可基于现有L1/L2能力组合实现。如高频使用,可沉淀为L3模板。
| 内置能力 | 位置 | 说明 |
|---|---|---|
| 价量分解 | 1.3 | 成本差异拆解为价格变动+数量变动 |
| 率混合分解 | 1.3 | 加权平均指标变化的比率效应vs组合效应 |
| 瀑布图归因 | 1.3 | 总差异按驱动因素正负贡献可视化 |
| 对账老化 | 1.4 | 未解决差异的年龄跟踪与升级 |
| CEAVOP认定 | 2.3 | 内控测试目标的认定映射 |
| 样本选择 | 2.3 | 按控制频率和风险等级确定样本量 |
| 三表钩稽 | 3.2 | 入库×出库×库存端到端追踪 |
| 制度关联 | 2.2+3.4 | 从知识库拉取制度条款,自动标注违规依据 |
| 跨期对比 | 1.7 | 同比环比+变点检测 |
| 帕累托分析 | 1.7 | 成本/问题集中度,聚焦关键少数 |
| 趋势预警 | 1.7 | 滚动均值+阈值线+连续偏离触发 |
| ⚡均值vs中位数选择 | 1.10.1 | 按数据分布特征选择集中趋势度量 |
| ⚡假设检验完整框架 | 1.10.2 | 五步法+效应量+置信区间+样本量考量 |
| ⚡统计陷阱警告 | 1.10.3 | 多重比较/幸存者偏差/平均的平均/生态谬误 |
| ⚡异常检测处理原则 | 1.10.4 | 调查优先四步法+点异常vs变点 |
| ⚡季节性检测 | 1.10.6 | 目视→周均值→月均值→同期对比 |
| ⚡交付前验证清单 | 1.10.7 | 数据质量+计算逻辑+合理性+红旗信号 |
C:\Users\DESKTOP-9F3C\.workbuddy\binaries\python\envs\default\Scripts\pip install pandas openpyxl python-docx numpy scipy sqlalchemy pymysql duckdb
运行脚本时使用:
C:\Users\DESKTOP-9F3C\.workbuddy\binaries\python\envs\default\Scripts\python analyze_xxx.py
| 文件 | 用途 | 命名规范 |
|---|---|---|
analyze_xxx.py |
分析脚本 | 动词+对象+py |
analysis_results.json |
中间结果 | 固定名,便于报告脚本引用 |
generate_report.py |
HTML 报告生成脚本 | generate_前缀 |
generate_docx.py |
Word 报告生成脚本 | generate_前缀 |
xxx分析报告.html |
HTML 报告 | 业务名称+分析报告 |
xxx分析报告.docx |
Word 报告 | 业务名称+分析报告 |
xxx_dashboard_appendix.html |
附录交互看板 | 业务名称+dashboard_appendix |
| 版本 | 变更 |
|---|---|
| v5.2 | 新增L1.10统计增强工具箱:描述性统计方法论(均值vs中位数选择/分布五问/百分位叙事)、假设检验完整框架(五步法+审计检验速查+统计显著≠实际重要+样本量考量)、统计陷阱警告体系(多重比较/幸存者偏差/平均的平均/生态谬误/锚定效应)、异常检测增强(离群值处理四步法+点异常vs变点区分)、增长率方法(简单/CAGR/对数)、季节性检测方法、交付前验证清单(4维度+红旗信号)。增强1.7路由表交叉引用。借鉴Scene#8数据分析及可视化 statistical-analysis + data-validation 精华 |
| v5.1 | 架构复盘审查修复15项:修复率混合分解代码错误(P4按品类分组+加权平均)、修复瀑布图归因cumulative缺失(P5)、补全差异优先级5级排序(P6)、profile_to_flags按context差异化(P7)、safe_query防注入增强(P11)、auto_compare大样本正态性检验改用D'Agostino+随机抽样(P10)、audit_profile相关性标注缺失率可靠性(P12)、classify_source iloc二维改一维(P15)、三因素分解补充定义与使用场景(P14)、补充L3.6综合场景模板(P13)、统一报告风格三处描述(P8)、清理可扩展维度冗余(P9)、修复七阶段→六阶段(P1)、架构图与正文对齐+L1编号顺序说明(P2/P3) |
| v5.0 | 差异分解方法论(价量/率混合/瀑布图归因+叙述规范)、对账钩稽体系(3类差异+老化分析+升级阈值)、重要性阈值框架(分层阈值+调查优先级+三向比较)、SOX内控测试路由(CEAVOP+样本选择+缺陷分级)、结账流程分析模板(依赖图+瓶颈定位)。借鉴finance插件variance-analysis/reconciliation/audit-support/sox-testing/close-management精华 |
| v4.0 | 三层解耦重构(L1通用引擎/L2场景映射/L3场景模板),从养殖专用扩展为全产业链通用,增加5个部门路由和8个产业链场景,新增构成分析和效率分析模式 |
| v3.0 | 多源数据接入、标准化数据画像、安全规范、分析模式路由、降级策略、结论撰写规范(借鉴ai-data-copilot精华) |
| v2.0 | 报告策略前置、多层匹配、Chart.js可视化、风险导航结构 |