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]
7w4.net小蔥技能站,你的AI助手技能庫。
面向企業通用審計場景的通用深度資料分析技能。
當用戶提供業務資料(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框架。分析報告交付前逐項檢查。
[ ] 資料來源已確認正確(表/檔案/日期)
[ ] 時間範圍完整,無意外斷層
[ ] 關鍵列缺失率已檢查,空值處理方式已記錄
[ ] 無因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視覺化、風險導航結構 |