ক্যাপস্টোন — একটি সম্পূর্ণ AI এথিক্স ও সেফটি অডিট পরিচালনা করা

# ========== ধাপ ১: স্টেকহোল্ডার ইমপ্যাক্ট ম্যাট্রিক্স -- সম্পূর্ণ সিন্থেটিক/ইলাস্ট্রেটিভ ==========
stakeholders = [
    {"name": "ঋণ আবেদনকারী",         "power": 2, "impact": 9},
    {"name": "ঋণদান প্রতিষ্ঠান",       "power": 9, "impact": 6},
    {"name": "নিয়ন্ত্রক সংস্থা",        "power": 8, "impact": 4},
    {"name": "ঋণ কর্মকর্তা/কর্মীরা",    "power": 5, "impact": 4},
    {"name": "প্রতিযোগী ঋণদাতা",       "power": 6, "impact": 2},
    {"name": "স্থানীয় অর্থনীতি/সমাজ",  "power": 1, "impact": 7},
]

FLAG_GAP = 4  # impact - power >= এই মান হলে "উচ্চ-ইমপ্যাক্ট, নিম্ন-পাওয়ার" হিসেবে ফ্ল্যাগ করা হবে

for s in stakeholders:
    s["priority_gap"] = s["impact"] - s["power"]
    s["flagged"] = s["priority_gap"] >= FLAG_GAP

print(f"{'স্টেকহোল্ডার':22s} {'পাওয়ার':>7s} {'ইমপ্যাক্ট':>9s} {'গ্যাপ':>6s}  ফ্ল্যাগ")
for s in stakeholders:
    flag_mark = "বিশেষ মনোযোগ" if s["flagged"] else ""
    print(f"{s['name']:22s} {s['power']:7d} {s['impact']:9d} {s['priority_gap']:6d}  {flag_mark}")

flagged_stakeholders = [s["name"] for s in stakeholders if s["flagged"]]
print(f"\nউচ্চ-ইমপ্যাক্ট, নিম্ন-পাওয়ার স্টেকহোল্ডার (বিশেষ মনোযোগ প্রয়োজন): {flagged_stakeholders}")

# ========== ধাপ ২: ফ্রেমওয়ার্ক বিশ্লেষণ -- অনুমোদন থ্রেশহোল্ড নির্বাচন ==========
options = ["কঠোর থ্রেশহোল্ড", "মধ্যম থ্রেশহোল্ড", "শিথিল থ্রেশহোল্ড"]

# --- ইউটিলিটারিয়ান: প্রতিটি বিকল্পের জন্য stakeholder-ভিত্তিক ইলাস্ট্রেটিভ প্রত্যাশিত-কল্যাণ স্কোর ---
welfare = {
    "কঠোর থ্রেশহোল্ড":   {"applicants_access": -4, "institution_risk": 6, "society_stability": 1},
    "মধ্যম থ্রেশহোল্ড":  {"applicants_access": 3, "institution_risk": 3, "society_access": 3},
    "শিথিল থ্রেশহোল্ড":  {"applicants_access": 6, "institution_risk": -5, "society_stability": -1},
}
totals = {opt: sum(v.values()) for opt, v in welfare.items()}
utilitarian_best = max(totals, key=totals.get)

# --- ডিওন্টোলজিক্যাল রুল-চেক: "যাচাইযোগ্য-শোধযোগ্যতাসম্পন্ন আবেদনকারীদের সংখ্যাগরিষ্ঠাংশ প্রত্যাখ্যান করা যাবে না" ---
reject_rate_among_verified = {"কঠোর থ্রেশহোল্ড": 62, "মধ্যম থ্রেশহোল্ড": 22, "শিথিল থ্রেশহোল্ড": 8}
RULE_LIMIT = 50
violations = {opt: rate > RULE_LIMIT for opt, rate in reject_rate_among_verified.items()}

# --- ম্যাক্সিমিন (ভেইল-অফ-ইগনোরেন্স): প্রতিটি ভূমিকার সবচেয়ে খারাপ ফলাফল ---
role_outcomes = {
    "কঠোর থ্রেশহোল্ড":   {"creditworthy_applicant": -5, "risky_applicant": 2, "institution": 6, "economy": 1},
    "মধ্যম থ্রেশহোল্ড":  {"creditworthy_applicant": 4, "risky_applicant": -1, "institution": 3, "economy": 3},
    "শিথিল থ্রেশহোল্ড":  {"creditworthy_applicant": 6, "risky_applicant": -6, "institution": -5, "economy": -1},
}
worst_case = {opt: min(v.values()) for opt, v in role_outcomes.items()}
maximin_choice = max(worst_case, key=worst_case.get)

print(f"{'বিকল্প':16s} {'ইউটিলিটারিয়ান':>15s} {'ডিওন্টো. লঙ্ঘন':>15s} {'ওয়ার্স্ট-কেস':>13s}")
for opt in options:
    print(f"{opt:16s} {totals[opt]:15d} {str(violations[opt]):>15s} {worst_case[opt]:13d}")

print(f"\nইউটিলিটারিয়ান-প্রেফার্ড বিকল্প: {utilitarian_best}")
print(f"ডিওন্টোলজিক্যাল নিয়ম লঙ্ঘন করে এমন বিকল্প: {[o for o in options if violations[o]]}")
print(f"ম্যাক্সিমিন-প্রেফার্ড বিকল্প (worst-case সবচেয়ে ভালো): {maximin_choice}")

allowed = [o for o in options if not violations[o]]
if utilitarian_best in allowed and maximin_choice in allowed and utilitarian_best == maximin_choice:
    framework_verdict = utilitarian_best
else:
    framework_verdict = maximin_choice if maximin_choice in allowed else "দ্বন্দ্ব -- আরও আলোচনা প্রয়োজন"

print(f"\nসামগ্রিক ফ্রেমওয়ার্ক-ভার্ডিক্ট: {framework_verdict}")

# ========== ধাপ ৩: ফেয়ারনেস অডিট -- সিন্থেটিক ঋণ-সিদ্ধান্ত ডেটাসেট ==========
import random
random.seed(11)

def gen_group(n, mean_repay_prob):
    rows = []
    for _ in range(n):
        # score = আবেদনকারীর প্রকৃত শোধ-করার-সম্ভাবনা (সিন্থেটিক গ্রাউন্ড-ট্রুথ probability)
        score = min(0.98, max(0.02, random.gauss(mean_repay_prob, 0.16)))
        repaid = 1 if random.random() < score else 0
        rows.append({"score": score, "repaid": repaid})
    return rows

N = 1500
# Group A ও Group B: দুটি সম্পূর্ণ সিন্থেটিক আবেদনকারী-গ্রুপ, ঐতিহাসিক কারণে ভিন্ন গড় বেস-রেট
group_a = gen_group(N, 0.72)
group_b = gen_group(N, 0.52)

FAIR_THRESHOLD = 0.60
for g in (group_a, group_b):
    for r in g:
        r["approved"] = r["score"] >= FAIR_THRESHOLD

def confusion(rows):
    tp = sum(1 for r in rows if r["approved"] and r["repaid"])
    fp = sum(1 for r in rows if r["approved"] and not r["repaid"])
    fn = sum(1 for r in rows if not r["approved"] and r["repaid"])
    tn = sum(1 for r in rows if not r["approved"] and not r["repaid"])
    return tp, fp, fn, tn

def report_group(rows, label):
    tp, fp, fn, tn = confusion(rows)
    n = len(rows)
    approve_rate = (tp + fp) / n * 100
    tpr = tp / (tp + fn) * 100
    fpr = fp / (fp + tn) * 100
    print(f"--- {label} (n={n}) ---")
    print(f"  ডেমোগ্রাফিক প্যারিটি (approve rate): {approve_rate:.1f}%")
    print(f"  True Positive Rate (TPR):            {tpr:.1f}%")
    print(f"  False Positive Rate (FPR):            {fpr:.1f}%")
    return {"approve_rate": approve_rate, "tpr": tpr, "fpr": fpr}

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

dp_gap = abs(metrics_a["approve_rate"] - metrics_b["approve_rate"])
tpr_gap = abs(metrics_a["tpr"] - metrics_b["tpr"])
fpr_gap = abs(metrics_a["fpr"] - metrics_b["fpr"])

# ক্যালিব্রেশন: score-bucket-ভিত্তিক প্রকৃত repay-rate গ্রুপভেদে কতটা মেলে (কমপক্ষে ২০ স্যাম্পলযুক্ত bucket-এ)
def calibration(rows, edges):
    out = []
    for lo, hi in zip(edges[:-1], edges[1:]):
        bucket = [r for r in rows if lo <= r["score"] < hi]
        if len(bucket) < 20:
            continue
        actual = sum(r["repaid"] for r in bucket) / len(bucket) * 100
        out.append((lo, hi, len(bucket), actual))
    return out

edges = [0.0, 0.2, 0.4, 0.6, 0.8, 1.0]
cal_a = calibration(group_a, edges)
cal_b = calibration(group_b, edges)

print("\nক্যালিব্রেশন (score-bucket অনুযায়ী প্রকৃত repay-rate):")
diffs = []
for lo, hi, n_a, actual_a in cal_a:
    match = [(n_b, actual_b) for (lo_b, hi_b, n_b, actual_b) in cal_b if lo_b == lo and hi_b == hi]
    if not match:
        continue
    n_b, actual_b = match[0]
    diffs.append(abs(actual_a - actual_b))
    print(f"  [{lo:.1f},{hi:.1f})  Group A: {actual_a:5.1f}% (n={n_a})   Group B: {actual_b:5.1f}% (n={n_b})")

calibration_gap = sum(diffs) / len(diffs)

DP_GAP_THRESHOLD = 10.0
EO_GAP_THRESHOLD = 10.0
CALIBRATION_GAP_THRESHOLD = 5.0

dp_status = "FAIL" if dp_gap > DP_GAP_THRESHOLD else "PASS"
eo_status = "FAIL" if (tpr_gap > EO_GAP_THRESHOLD or fpr_gap > EO_GAP_THRESHOLD) else "PASS"
calibration_status = "PASS" if calibration_gap <= CALIBRATION_GAP_THRESHOLD else "FAIL"

print(f"\nডেমোগ্রাফিক প্যারিটি গ্যাপ: {dp_gap:.1f} পয়েন্ট  (থ্রেশহোল্ড {DP_GAP_THRESHOLD:.0f})  -> {dp_status}")
print(f"ইকুয়ালাইজড অডস গ্যাপ:  TPR {tpr_gap:.1f} পয়েন্ট, FPR {fpr_gap:.1f} পয়েন্ট  (থ্রেশহোল্ড {EO_GAP_THRESHOLD:.0f})  -> {eo_status}")
print(f"ক্যালিব্রেশন গ্যাপ (গড়):  {calibration_gap:.2f} পয়েন্ট  (থ্রেশহোল্ড {CALIBRATION_GAP_THRESHOLD:.0f})  -> {calibration_status}")

# ========== ধাপ ৪: প্রাইভেসি চেক -- k-anonymity, সিন্থেটিক প্রশিক্ষণ-ডেটার quasi-identifier-এ ==========
from collections import defaultdict

training_records = [
    {"id": 1,  "zip3": "712", "age_bracket": "30s", "employment": "Salaried"},
    {"id": 2,  "zip3": "712", "age_bracket": "30s", "employment": "Salaried"},
    {"id": 3,  "zip3": "712", "age_bracket": "30s", "employment": "Salaried"},
    {"id": 4,  "zip3": "712", "age_bracket": "40s", "employment": "Salaried"},
    {"id": 5,  "zip3": "712", "age_bracket": "40s", "employment": "Salaried"},
    {"id": 6,  "zip3": "845", "age_bracket": "20s", "employment": "Self-employed"},
    {"id": 7,  "zip3": "845", "age_bracket": "30s", "employment": "Salaried"},
    {"id": 8,  "zip3": "845", "age_bracket": "30s", "employment": "Salaried"},
    {"id": 9,  "zip3": "845", "age_bracket": "30s", "employment": "Salaried"},
    {"id": 10, "zip3": "930", "age_bracket": "50s", "employment": "Retired"},
    {"id": 11, "zip3": "930", "age_bracket": "50s", "employment": "Retired"},
    {"id": 12, "zip3": "110", "age_bracket": "20s", "employment": "Student"},
    {"id": 13, "zip3": "712", "age_bracket": "40s", "employment": "Salaried"},
]

QUASI_IDENTIFIERS = ("zip3", "age_bracket", "employment")
K_REQUIRED = 3

classes = defaultdict(list)
for r in training_records:
    key = tuple(r[q] for q in QUASI_IDENTIFIERS)
    classes[key].append(r["id"])

print(f"{'quasi-identifier গ্রুপ':45s} {'সাইজ':>5s}  স্ট্যাটাস")
for key, ids in classes.items():
    status = "OK" if len(ids) >= K_REQUIRED else "VIOLATION"
    print(f"{str(key):45s} {len(ids):5d}  {status}  {ids}")

violating_classes = {k: v for k, v in classes.items() if len(v) < K_REQUIRED}
violating_ids = sorted(i for v in violating_classes.values() for i in v)
k_anon_status = "PASS" if not violating_classes else "FAIL"

print(f"\nমোট রেকর্ড: {len(training_records)},  মোট quasi-identifier গ্রুপ: {len(classes)}")
print(f"k={K_REQUIRED}-anonymity লঙ্ঘনকারী গ্রুপ সংখ্যা: {len(violating_classes)}")
print(f"লঙ্ঘনকারী record id: {violating_ids}")
print(f"k-anonymity স্ট্যাটাস: {k_anon_status}")

# ========== ধাপ ৫: স্পেসিফিকেশন-গেমিং / Goodhart চেক ==========
# অপ্টিমাইজেশন টার্গেট: "অনুমোদিত ঋণের মধ্যে শোধ-হওয়ার হার সর্বোচ্চ করা" (proxy metric)
# thin-file (সীমিত ক্রেডিট-হিস্ট্রি) সাবগ্রুপের ক্ষেত্রে বিভিন্ন নীতি-কৌশল তুলনা করা হচ্ছে

POP = 400                          # thin-file সাবগ্রুপের আবেদনকারী সংখ্যা
TRUE_CREDITWORTHY_RATE = 0.78      # এই সাবগ্রুপের প্রকৃত creditworthy অনুপাত (প্রকৃতপক্ষে বেশ ভালো)
ASSESS_TPR = 0.85                  # ব্যক্তি-ভিত্তিক মূল্যায়নে creditworthy-দের সঠিকভাবে অনুমোদনের হার
ASSESS_FPR = 0.15                  # ব্যক্তি-ভিত্তিক মূল্যায়নে non-creditworthy-দের ভুলভাবে অনুমোদনের হার

reject_fractions = [0.0, 0.25, 0.5, 0.75, 1.0]  # সাবগ্রুপের কত অংশ ব্যক্তি-মূল্যায়ন ছাড়াই categorically বাতিল

results = []
for r in reject_fractions:
    n_rejected = round(r * POP)
    n_assessed = POP - n_rejected
    n_credit = n_assessed * TRUE_CREDITWORTHY_RATE
    n_noncredit = n_assessed * (1 - TRUE_CREDITWORTHY_RATE)

    approved_credit = n_credit * ASSESS_TPR
    approved_noncredit = n_noncredit * ASSESS_FPR
    total_approved = approved_credit + approved_noncredit

    # proxy metric: অনুমোদিতদের মধ্যে শোধ-হওয়ার হার -- কাউকে অনুমোদন না করলে সংজ্ঞা অনুযায়ী ১০০%
    proxy_repayment_rate = (approved_credit / total_approved * 100) if total_approved > 0 else 100.0

    excluded_creditworthy = n_rejected * TRUE_CREDITWORTHY_RATE + n_credit * (1 - ASSESS_TPR)
    # true_goal: প্রকৃতপক্ষে সঠিকভাবে সেবা-পাওয়া creditworthy মানুষ, বিয়োগ ভুল-অনুমোদন ও ভুল-বর্জনের ক্ষতি
    true_goal_score = approved_credit - 0.5 * approved_noncredit - 1.0 * excluded_creditworthy

    results.append({
        "reject_fraction": r, "total_approved": total_approved,
        "proxy_repayment_rate": proxy_repayment_rate, "true_goal_score": true_goal_score,
    })

print(f"{'বাতিল-অংশ':>10s} {'মোট অনুমোদিত':>13s} {'proxy: repay-rate':>18s} {'true-goal স্কোর':>16s}")
for res in results:
    print(f"{res['reject_fraction']:10.2f} {res['total_approved']:13.1f} {res['proxy_repayment_rate']:18.1f} {res['true_goal_score']:16.1f}")

best_by_proxy = max(results, key=lambda x: x["proxy_repayment_rate"])
best_by_true_goal = max(results, key=lambda x: x["true_goal_score"])
goodhart_flagged = best_by_proxy["reject_fraction"] != best_by_true_goal["reject_fraction"]

print(f"\nproxy metric সর্বোচ্চ করে এমন কৌশল: বাতিল-অংশ={best_by_proxy['reject_fraction']:.2f}  (repay-rate {best_by_proxy['proxy_repayment_rate']:.1f}%)")
print(f"true-goal সর্বোচ্চ করে এমন কৌশল:   বাতিল-অংশ={best_by_true_goal['reject_fraction']:.2f}  (true-goal স্কোর {best_by_true_goal['true_goal_score']:.1f})")
print(f"Goodhart ঝুঁকি ফ্ল্যাগড: {goodhart_flagged}")

# ========== ধাপ ৬: রিস্ক-টিয়ার শ্রেণীবিভাগ (EU AI Act-ধাঁচের রুল-বেসড ক্লাসিফায়ার) ==========
def classify_risk_tier(system):
    if system.get("social_scoring") or system.get("subliminal_manipulation"):
        return "অগ্রহণযোগ্য ঝুঁকি (Unacceptable risk)"
    high_risk_domains = {"credit_scoring", "employment", "law_enforcement",
                          "healthcare_diagnosis", "critical_infrastructure", "education_admission"}
    if system.get("automated_decision") and system.get("domain") in high_risk_domains:
        return "উচ্চ ঝুঁকি (High risk)"
    if system.get("user_facing_chat") or system.get("generates_content"):
        return "সীমিত ঝুঁকি (Limited risk)"
    return "সর্বনিম্ন ঝুঁকি (Minimal risk)"

# তুলনার জন্য কয়েকটি উদাহরণ সিস্টেম, ও অডিট করা লোন সিস্টেমটি নিজে
example_systems = [
    {"name": "স্প্যাম-ফিল্টার",                  "domain": "email", "automated_decision": True,
     "social_scoring": False, "user_facing_chat": False},
    {"name": "কাস্টমার-সাপোর্ট চ্যাটবট",           "domain": "customer_service", "automated_decision": False,
     "social_scoring": False, "user_facing_chat": True},
    {"name": "AI ঋণ-অনুমোদন সিস্টেম (এই অডিট)",   "domain": "credit_scoring", "automated_decision": True,
     "social_scoring": False, "user_facing_chat": False},
]

for sysinfo in example_systems:
    tier = classify_risk_tier(sysinfo)
    print(f"{sysinfo['name']:32s} -> {tier}")

loan_system = example_systems[2]
risk_tier = classify_risk_tier(loan_system)
print(f"\nঅডিট করা লোন সিস্টেমের রিস্ক-টিয়ার: {risk_tier}")

# ========== ধাপ ৭: চূড়ান্ত অডিট সারাংশ -- ধাপ ১-৬-এর প্রকৃত গণনা করা ফলাফল থেকে ==========
fairness_summary = {
    "demographic_parity": dp_status,
    "equalized_odds": eo_status,
    "calibration": calibration_status,
}
fairness_all_pass = all(v == "PASS" for v in fairness_summary.values())

print("=" * 78)
print("চূড়ান্ত অডিট রিপোর্ট -- হাইপোথেটিক্যাল AI ঋণ-অনুমোদন সিস্টেম")
print("=" * 78)
print(f"১. স্টেকহোল্ডার প্রায়োরিটি ফ্ল্যাগ (উচ্চ-ইমপ্যাক্ট, নিম্ন-পাওয়ার): {flagged_stakeholders}")
print(f"২. ফ্রেমওয়ার্ক ভার্ডিক্ট (থ্রেশহোল্ড নীতি): {framework_verdict}")
print(f"৩. ফেয়ারনেস মেট্রিক্স: ডেমোগ্রাফিক-প্যারিটি={fairness_summary['demographic_parity']}, "
      f"ইকুয়ালাইজড-অডস={fairness_summary['equalized_odds']}, ক্যালিব্রেশন={fairness_summary['calibration']}  "
      f"(সবগুলো pass: {fairness_all_pass})")
print(f"৪. k-anonymity (k={K_REQUIRED}): {k_anon_status}  (লঙ্ঘনকারী record id: {violating_ids})")
print(f"৫. Goodhart/স্পেসিফিকেশন-গেমিং ঝুঁকি: {'ফ্ল্যাগড' if goodhart_flagged else 'ক্লিয়ার'}")
print(f"৬. রিস্ক-টিয়ার (EU AI Act-ধাঁচ): {risk_tier}")
print()

issues = []
if not fairness_all_pass:
    issues.append("ফেয়ারনেস মেট্রিক্সে গ্যাপ থ্রেশহোল্ড ছাড়িয়েছে (demographic parity ও equalized odds)")
if k_anon_status == "FAIL":
    issues.append("প্রশিক্ষণ-ডেটা k-anonymity পূরণ করছে না")
if goodhart_flagged:
    issues.append("অপ্টিমাইজেশন টার্গেট একটি creditworthy সাবগ্রুপ বাদ দেওয়ার পক্ষে perverse incentive তৈরি করছে")

print(f"মোট {len(issues)}টি সমস্যা চিহ্নিত হয়েছে:")
for i in issues:
    print(f"  - {i}")

if len(issues) >= 2:
    final_recommendation = "বর্তমান অবস্থায় ডিপ্লয় করা উচিত নয় -- উপরের সমস্যাগুলো সমাধানের পর পুনরায় অডিট আবশ্যক"
elif len(issues) == 1:
    final_recommendation = "শর্তসাপেক্ষে ডিপ্লয় করা যেতে পারে -- চিহ্নিত সমস্যাটি সমাধান ও নিবিড় মনিটরিং সাপেক্ষে"
else:
    final_recommendation = "ডিপ্লয় করা যেতে পারে"

print(f"\nচূড়ান্ত সুপারিশ: {final_recommendation}")