#!/usr/bin/env python3 # -*- coding: utf-8 -*- """clean_eval_blocks.py — تقييم كامل بكتل 300 مرشح على كل العناصر (لا عيّنة 300): - تقسيم كل مجموعة (20K) لكتل 300 بعشوائية بذرة 41 (66 كتلة كاملة لكل مجموعة؛ تُهمَل البقية) - R@1 لكل عنصر (WTA داخل الكتلة) عبر المُرمِّز الدفعي v3 - فصل: عناصر مموّهة (تداخل تدريبي) مقابل نظيفة ⇒ اختبار فجوة التسرب (leak-gap) - نطاق 95% bootstrap + اختبار نسبتين + اختبار McNemar مقترن لعناصر متطابقة السؤال (غير ممكن — مجموعتان منفصلتان) المخرجات: results/clean_eval_blocks.json + results/block_hits.npz (إصابات لكل عنصر)""" import os, sys, json, math, time import numpy as np sys.path.insert(0, "/home/user/lhc/code"); sys.path.insert(0, "/home/user/lhc/recon") from phase_w1_data import download_corpus from phase_c_encoder import TrainableEncoder from phase_p2_slot import hashed_features_slot from features_v3 import encode_batch_v3, clear_caches T0=time.time() def log(*a): print(f"[{time.time()-T0:6.1f}s]", *a, flush=True) R="/home/user/lhc/results" M=json.load(open(f"{R}/masks_v2.json")); masks={cn:set(M["masks_strong"][cn]) for cn in M["masks_strong"]} DD="/home/user/lhc/data/opus" OPUS=[("Tanzil","https://object.pouta.csc.fi/OPUS-Tanzil/v1/moses/ar-en.txt.zip",20000), ("Bible","https://object.pouta.csc.fi/OPUS-bible-uedin/v1/moses/ar-en.txt.zip",20000), ("NeuLab-TED","https://object.pouta.csc.fi/OPUS-NeuLab-TedTalks/v1/moses/ar-en.txt.zip",20000)] pairs={} for name,url,lim in OPUS: pairs[name]=download_corpus(name,url,lim,DD) DEV="/home/user/lhc/data/flores101_dataset/devtest" ld=lambda p:[l.strip() for l in open(p,encoding="utf-8") if l.strip()] pairs["FLORES"]=[{"ar":a,"en":e} for a,e in zip(ld(f"{DEV}/ara.devtest"),ld(f"{DEV}/eng.devtest"))] BS=300 def blocks(n, seed=41): rng=np.random.RandomState(seed); idx=rng.permutation(n) return [idx[i:i+BS] for i in range(0, n-BS+1, BS)] # كتل كاملة فقط def wilson_ci(k,n,z=1.96): if n==0: return [None,None] p=k/n; d=1+z*z/n c=(p+z*z/(2*n))/d h=z*math.sqrt(p*(1-p)/n+z*z/(4*n*n))/d return [round(max(0,c-h),4), round(min(1,c+h),4)] def boot_diff_ci(hit_a, hit_b, n_boot=10000, seed=7): rng=np.random.RandomState(seed) a=hit_a.astype(np.float64); b=hit_b.astype(np.float64) diffs=np.empty(n_boot) for i in range(n_boot): ia=rng.randint(0,len(a),len(a)); ib=rng.randint(0,len(b),len(b)) diffs[i]=a[ia].mean()-b[ib].mean() return [round(float(np.percentile(diffs,2.5)),4), round(float(np.percentile(diffs,97.5)),4)] def twoprop_p(k1,n1,k2,n2): if n1==0 or n2==0: return None p1,p2=k1/n1,k2/n2; p=(k1+k2)/(n1+n2) se=math.sqrt(p*(1-p)*(1/n1+1/n2)) if se==0: return 1.0 z=(p1-p2)/se # احتساب ثنائي الجانب عبر دالة الخطأ return round(float(math.erfc(abs(z)/math.sqrt(2))),6) models={"W4":"/home/user/lhc/weights/phase_w1_W.npz","W4.1":"/home/user/lhc/weights/phase_w1_W_robust.npz"} encs={} for tag,wp in models.items(): z=np.load(wp); encs[tag]=TrainableEncoder(F=int(z["F"]),d=int(z["d"]),tau=float(z["tau"]),W=z["W"],features_fn=hashed_features_slot) out={"block_size":BS, "sets":{}, "notes":"كتل كاملة فقط؛ R@1 داخل الكتلة (WTA). المُرمِّز الدفعي v3 (متكافئ بايت-بايت)."} hits_store={} for cn in ("Tanzil","Bible","NeuLab-TED","FLORES"): P=pairs[cn]; n=len(P); bl=blocks(n) used=int(sum(len(b) for b in bl)) mask=masks[cn] log(f"{cn}: n={n:,} → {len(bl)} كتلة (تُستخدم {used:,} عنصرًا)") for tag in models: hit_ae=np.zeros(used, bool); hit_ea=np.zeros(used, bool) inmask=np.zeros(used, bool) pos=0 for b in bl: ar=[P[i]["ar"] for i in b]; en=[P[i]["en"] for i in b] clear_caches() Ea=encode_batch_v3(encs[tag], ar); Ee=encode_batch_v3(encs[tag], en) sim=Ea@Ee.T pred=sim.argmax(axis=1) # بروتوكول twin_retrieval النصّي: نجاح = نص الهدف المُرجَع يطابق الحقيقي hit_ae[pos:pos+len(b)]=[P[b[i]]["en"]==en[pred[i]] for i in range(len(b))] pred2=sim.argmax(axis=0) hit_ea[pos:pos+len(b)]=[P[b[j]]["ar"]==ar[pred2[j]] for j in range(len(b))] inmask[pos:pos+len(b)]=[b[i] in mask for i in range(len(b))] pos+=len(b) k_ae=int(hit_ae.sum()); k_ea=int(hit_ea.sum()) m=inmask; c=~inmask res={ "used":used, "r1_ar_en":round(k_ae/used,4), "r1_en_ar":round(k_ea/used,4), "ci95_ar_en":wilson_ci(k_ae,used), "ci95_en_ar":wilson_ci(k_ea,used), "n_masked":int(m.sum()), "n_clean":int(c.sum()), "masked": {"r1_ar_en":round(float(hit_ae[m].mean()),4) if m.any() else None, "r1_en_ar":round(float(hit_ea[m].mean()),4) if m.any() else None}, "clean": {"r1_ar_en":round(float(hit_ae[c].mean()),4) if c.any() else None, "r1_en_ar":round(float(hit_ea[c].mean()),4) if c.any() else None}, } if m.any() and c.any(): res["leak_gap_ar_en"]={"diff":round(float(hit_ae[m].mean()-hit_ae[c].mean()),4), "ci95":boot_diff_ci(hit_ae[m],hit_ae[c]), "p_twoprop":twoprop_p(int(hit_ae[m].sum()),int(m.sum()),int(hit_ae[c].sum()),int(c.sum())), "significant": boot_diff_ci(hit_ae[m],hit_ae[c])[0]>0 or boot_diff_ci(hit_ae[m],hit_ae[c])[1]<0} res["leak_gap_en_ar"]={"diff":round(float(hit_ea[m].mean()-hit_ea[c].mean()),4), "ci95":boot_diff_ci(hit_ea[m],hit_ea[c]), "p_twoprop":twoprop_p(int(hit_ea[m].sum()),int(m.sum()),int(hit_ea[c].sum()),int(c.sum()))} out["sets"][f"{cn}|{tag}"]=res hits_store[f"{cn}|{tag}|ar_en"]=hit_ae; hits_store[f"{cn}|{tag}|en_ar"]=hit_ea hits_store[f"{cn}|{tag}|mask"]=inmask log(f" [{tag}] كامل: {res['r1_ar_en']:.4f}/{res['r1_en_ar']:.4f} | مموّه(n={res['n_masked']}): {res['masked']['r1_ar_en']}/{res['masked']['r1_en_ar']} | نظيف(n={res['n_clean']}): {res['clean']['r1_ar_en']}/{res['clean']['r1_en_ar']}") # تجميع ALL: كتل مختلطة؟ نستخدم اتحاد كتل المجموعات (تقييم داخلي لكل مجموعة) ونعرض المتوسط المرجّح for tag in models: tot=sum(out["sets"][f"{cn}|{tag}"]["used"] for cn in ("Tanzil","Bible","NeuLab-TED")) k=sum(out["sets"][f"{cn}|{tag}"]["r1_ar_en"]*out["sets"][f"{cn}|{tag}"]["used"] for cn in ("Tanzil","Bible","NeuLab-TED")) kc=sum(out["sets"][f"{cn}|{tag}"]["clean"]["r1_ar_en"]*(out["sets"][f"{cn}|{tag}"]["used"]-out["sets"][f"{cn}|{tag}"]["n_masked"]) for cn in ("Tanzil","Bible","NeuLab-TED")) nc=sum(out["sets"][f"{cn}|{tag}"]["used"]-out["sets"][f"{cn}|{tag}"]["n_masked"] for cn in ("Tanzil","Bible","NeuLab-TED")) out.setdefault("ALL_union",{})[tag]={"pooled_r1_ar_en":round(k/tot,4), "clean_only_r1_ar_en":round(kc/nc,4), "clean_n":nc} np.savez_compressed(f"{R}/block_hits.npz", **hits_store) json.dump(out, open(f"{R}/clean_eval_blocks.json","w"), ensure_ascii=False, indent=1) log("SAVED clean_eval_blocks.json + block_hits.npz")