কেস স্টাডি: একটি সম্পূর্ণ AI ডিপ্লয়মেন্ট এথিক্স রিভিউ

# ========== ধাপ ১: স্টেকহোল্ডার ইমপ্যাক্ট ম্যাট্রিক্স -- সম্পূর্ণ সিন্থেটিক/ইলাস্ট্রেটিভ ==========
stakeholders = [
    {"name": "চাকরি আবেদনকারী",        "weight": 0.50, "financial": -3, "autonomy": -4, "privacy": -3, "safety": 0},
    {"name": "ক্লায়েন্ট কোম্পানি",       "weight": 0.25, "financial":  4, "autonomy":  2, "privacy":  0, "safety": 0},
    {"name": "HR স্টাফ",                "weight": 0.15, "financial":  1, "autonomy": -1, "privacy":  0, "safety": 0},
    {"name": "বৃহত্তর শ্রমবাজার/সমাজ",   "weight": 0.10, "financial":  0, "autonomy": -1, "privacy": -1, "safety": 0},
]

DIMENSIONS = ["financial", "autonomy", "privacy", "safety"]

def weighted_impact(stakeholder):
    mean_score = sum(stakeholder[d] for d in DIMENSIONS) / len(DIMENSIONS)
    return mean_score, mean_score * stakeholder["weight"]

print(f"{'স্টেকহোল্ডার':22s} {'গড় ইমপ্যাক্ট':>14s} {'ওয়েট':>8s} {'ওয়েটেড স্কোর':>14s}")
results = []
for s in stakeholders:
    mean_score, weighted = weighted_impact(s)
    results.append((s["name"], mean_score, s["weight"], weighted))
    print(f"{s['name']:22s} {mean_score:14.3f} {s['weight']:8.2f} {weighted:14.4f}")

most_affected = min(results, key=lambda r: r[3])
print(f"\nসবচেয়ে (নেতিবাচকভাবে) বেশি প্রভাবিত গোষ্ঠী: {most_affected[0]}  (ওয়েটেড স্কোর {most_affected[3]:.4f})")

# ========== ধাপ ২: ফেয়ারনেস মেট্রিক্স -- দুটি সিন্থেটিক ডেমোগ্রাফিক গ্রুপে ==========
group_a = [
    {"id": "A1", "predicted_positive": True,  "actual_positive": True},
    {"id": "A2", "predicted_positive": True,  "actual_positive": True},
    {"id": "A3", "predicted_positive": True,  "actual_positive": False},
    {"id": "A4", "predicted_positive": True,  "actual_positive": True},
    {"id": "A5", "predicted_positive": True,  "actual_positive": False},
    {"id": "A6", "predicted_positive": True,  "actual_positive": True},
    {"id": "A7", "predicted_positive": False, "actual_positive": False},
    {"id": "A8", "predicted_positive": False, "actual_positive": False},
    {"id": "A9", "predicted_positive": False, "actual_positive": True},
    {"id": "A10","predicted_positive": False, "actual_positive": False},
    {"id": "A11","predicted_positive": False, "actual_positive": False},
    {"id": "A12","predicted_positive": False, "actual_positive": False},
]

group_b = [
    {"id": "B1", "predicted_positive": True,  "actual_positive": True},
    {"id": "B2", "predicted_positive": True,  "actual_positive": False},
    {"id": "B3", "predicted_positive": True,  "actual_positive": False},
    {"id": "B4", "predicted_positive": True,  "actual_positive": True},
    {"id": "B5", "predicted_positive": False, "actual_positive": False},
    {"id": "B6", "predicted_positive": False, "actual_positive": False},
    {"id": "B7", "predicted_positive": False, "actual_positive": True},
    {"id": "B8", "predicted_positive": False, "actual_positive": True},
    {"id": "B9", "predicted_positive": False, "actual_positive": False},
    {"id": "B10","predicted_positive": False, "actual_positive": False},
    {"id": "B11","predicted_positive": False, "actual_positive": False},
    {"id": "B12","predicted_positive": False, "actual_positive": False},
]

def fairness_metrics(records, label):
    n = len(records)
    positives_predicted = [r for r in records if r["predicted_positive"]]
    actual_pos = [r for r in records if r["actual_positive"]]
    actual_neg = [r for r in records if not r["actual_positive"]]

    dp_rate = len(positives_predicted) / n

    fp = [r for r in records if r["predicted_positive"] and not r["actual_positive"]]
    fpr = len(fp) / len(actual_neg) if actual_neg else 0.0

    fn = [r for r in records if not r["predicted_positive"] and r["actual_positive"]]
    fnr = len(fn) / len(actual_pos) if actual_pos else 0.0

    correct = [r for r in records if r["predicted_positive"] == r["actual_positive"]]
    accuracy = len(correct) / n

    calib_pos = (sum(1 for r in positives_predicted if r["actual_positive"]) / len(positives_predicted)
                 if positives_predicted else 0.0)

    print(f"--- {label} (n={n}) ---")
    print(f"  ডেমোগ্রাফিক প্যারিটি (positive prediction rate): {dp_rate:.3f}")
    print(f"  False Positive Rate:                             {fpr:.3f}")
    print(f"  False Negative Rate:                             {fnr:.3f}")
    print(f"  সামগ্রিক accuracy:                                {accuracy:.3f}")
    print(f"  ক্যালিব্রেশন (predicted-positive-দের মধ্যে actual positive rate): {calib_pos:.3f}")
    return {"dp_rate": dp_rate, "fpr": fpr, "fnr": fnr, "accuracy": accuracy, "calibration": calib_pos}

metrics_a = fairness_metrics(group_a, "Group A")
print()
metrics_b = fairness_metrics(group_b, "Group B")

dp_gap = abs(metrics_a["dp_rate"] - metrics_b["dp_rate"])
fnr_gap = abs(metrics_a["fnr"] - metrics_b["fnr"])
print(f"\nডেমোগ্রাফিক প্যারিটি ফারাক: {dp_gap:.3f}  ({dp_gap*100:.1f} পার্সেন্টেজ পয়েন্ট)")
print(f"False Negative Rate ফারাক: {fnr_gap:.3f}  ({fnr_gap*100:.1f} পার্সেন্টেজ পয়েন্ট)")

# ========== ধাপ ৩: k-anonymity চেক -- ট্রেনিং ডেটার কোয়াসি-আইডেন্টিফায়ারে ==========
training_records = [
    {"age_bracket": "20-29", "zip5": "10001", "gender": "F"},
    {"age_bracket": "20-29", "zip5": "10001", "gender": "F"},
    {"age_bracket": "20-29", "zip5": "10002", "gender": "F"},
    {"age_bracket": "20-29", "zip5": "10003", "gender": "M"},
    {"age_bracket": "30-39", "zip5": "10001", "gender": "M"},
    {"age_bracket": "30-39", "zip5": "10001", "gender": "M"},
    {"age_bracket": "30-39", "zip5": "10002", "gender": "M"},
    {"age_bracket": "30-39", "zip5": "10002", "gender": "F"},
    {"age_bracket": "40-49", "zip5": "10001", "gender": "F"},
    {"age_bracket": "40-49", "zip5": "10001", "gender": "F"},
    {"age_bracket": "40-49", "zip5": "10003", "gender": "M"},
    {"age_bracket": "20-29", "zip5": "10002", "gender": "M"},
    {"age_bracket": "30-39", "zip5": "10003", "gender": "F"},
    {"age_bracket": "40-49", "zip5": "10002", "gender": "M"},
    {"age_bracket": "20-29", "zip5": "10001", "gender": "M"},
]

def group_sizes(records, quasi_id_fields):
    groups = {}
    for r in records:
        key = tuple(r[f] for f in quasi_id_fields)
        groups.setdefault(key, []).append(r)
    return {k: len(v) for k, v in groups.items()}

def check_k_anonymity(records, quasi_id_fields, k, label):
    sizes = group_sizes(records, quasi_id_fields)
    min_size = min(sizes.values())
    violating = sum(1 for size in sizes.values() if size < k)
    satisfied = min_size >= k
    print(f"--- {label} ---")
    print(f"  কোয়াসি-আইডেন্টিফায়ার: {quasi_id_fields}")
    print(f"  মোট গ্রুপ সংখ্যা: {len(sizes)},  সর্বনিম্ন গ্রুপ সাইজ: {min_size}")
    print(f"  k={k} লঙ্ঘনকারী গ্রুপ সংখ্যা: {violating}")
    print(f"  k={k}-anonymity satisfied: {satisfied}")
    return min_size, satisfied

K = 3
print("ধাপ A: মূল, জেনারালাইজ না করা কোয়াসি-আইডেন্টিফায়ার (বয়স-ব্র্যাকেট + সম্পূর্ণ zip + লিঙ্গ)")
min_a, ok_a = check_k_anonymity(training_records, ["age_bracket", "zip5", "gender"], K, "ধাপ A -- মূল ডেটা")

print("\nধাপ B: zip জেনারালাইজেশন (৫-সংখ্যা -> প্রথম ৩ সংখ্যা), লিঙ্গ এখনো রাখা হয়েছে")
generalized = [dict(r, zip3=r["zip5"][:3]) for r in training_records]
min_b, ok_b = check_k_anonymity(generalized, ["age_bracket", "zip3", "gender"], K, "ধাপ B -- zip3 + gender")

print("\nধাপ C: zip3 জেনারালাইজেশন + লিঙ্গ ফিল্ড সম্পূর্ণ বাদ দেওয়া")
min_c, ok_c = check_k_anonymity(generalized, ["age_bracket", "zip3"], K, "ধাপ C -- শুধু age_bracket + zip3")

print(f"\nসারসংক্ষেপ: সর্বনিম্ন গ্রুপ সাইজ ধাপ A-তে {min_a} -> ধাপ B-তে {min_b} -> ধাপ C-তে {min_c}")