本文分享自天翼雲開發者社區《警惕大數據處理中的“檢查者悖論》.作者：王****淋
什麼是檢查者悖論：
觀察的角度不同，得出的統計結論也不同。有時又稱為"候車悖論", "等待時間悖論"
為了形象説明，我們設計了一種模擬場景: 班級人數統計，來用實例説明這個問題
模擬場景: 班級人數統計
小明與小華要完成一個任務：統計學校中的 平均班級人數。但二人的實施方案不同：
1）小明找到了教務處老師，拿到了一份每班級人數統計名單。 於是他計算到了班級平均人數
, 其中N為班級數量。
2）小華則不同, 他選擇去街頭詢問。小華在校園中隨機詢問了M人, 得到了M個數字, 每個數字即為該被詢問的同學所在的班級人數。於是他計算到了班級平均人數
顯然，二者的答案是不同的：
假設此學校一共2個班級: 一個90人，另一個10人。則小明計算結果： C^{mean} = (10 + 90) / 2 = 50人。
假設小華抽了10人，在隨機抽樣的情況下，大約9人屬於班級1，1人屬於班級2，小華計算結果 X^{mean} = (10*1 + 90*9) / 10 = 82人
原因初探
很明顯, 結果出現了偏差。是什麼導致了這種情況？其實，這就是"檢查悖論"
每個班級人數不均衡情況下:
新問題提出 ================
這時，自然的，我們提出一個問題：如果只有小華的數據，如何得到真實的統計結果，即班級平均人數 ?
EM算法可以勝任這一問題：EM是一類算法， 在包含隱變量的情況下，可以估算模型參數。
下面以班級平均人數統計問題為例：
其中，Si是已知量，ki為參數，sigma為隱變量，ni是觀測變量。
上代碼：
import random
def print_dict(d:dict, title=""):
k = [ki for ki in d.keys()]
k.sort()
s = []
for ki in k:
s.append(str(ki) + ": " + str(d[ki]))
print(title, "{", "; ".join(s), "}")
### 1. 數據集生成：通過隨機數================================================
# 1.1. 首先生成每班人數
classes = 20
classes = classes // 2 * 2
class_members = [random.randint(40, 50) for i in range(classes//2)] + \
[random.randint(100, 150) for i in range(classes//2)] \
class_dict = {} # 統計量: 人數為Ni的班級>共有多少個
for ci in class_members:
if ci in class_dict.keys(): class_dict[ci] += 1
else:class_dict[ci] = 1
# 1.2. 在街上抽學生(依據每班人數)
cy = random.choices(
population=class_members,
weights=[ci/sum(class_members) for ci in class_members], # 依概率抽樣
k = 400 # 街頭抽樣人數：應儘可能保證所有班級都有抽到
)
asked_class_dict = {} # 統計量: 班級人數為Xi的人>共抽到了幾個人
for cyi in cy:
if cyi in asked_class_dict.keys(): asked_class_dict[cyi] += 1
else:asked_class_dict[cyi] = 1
### 2. EM =====================================================================
# EM \theta _0 init
theta = {k: 1 for k in asked_class_dict.keys()}
# run iter: ------------------
run_i = 20
while run_i > 0:
run_i -= 1
# 1. E-step: 依照{\theta}^t, 求P( Z | X, {\theta}^t )
sigma = 0
for k, v in theta.items():
sigma += k * v
sigma = len(cy) / sigma
# 2. M-step: theta=argmax()
for k, v in theta.items():
theta[k] = round(asked_class_dict[k] / (k * v * sigma) * theta[k])
if theta[k] < 1: theta[k] = 1
### 3. result showing ==============================================================
print_dict(theta, "預測：{班級人數Ni: <人數為Ni的班級>共有多少個} \n\t")
print_dict(class_dict, "真實：{班級人數Ni: <人數為Ni的班級>共有多少個} \n\t")
N_class = 0
N_students = 0
for k, v in theta.items():
N_students += k * v
N_class += v
print("預測平均 = ", N_students / N_class)
print("真實平均 = ", sum(class_members)/classes )
print("街頭抽樣平均 = ", sum(cy)/len(cy))