Jevを使ったクレジットカード不正利用検知の可能性を探る

from typesafe_sdk import Choice, Noul, TypeSafeClient

# (テスト文, 期待値: 個人情報を含むか)
SAMPLES = [
    ("明日の会議は10時から第2会議室で行います。", False),
    ("担当の山田太郎（090-1234-5678）までご連絡ください。", True),
    ("東京都渋谷区神南1-2-3に住む佐藤花子さん宛に送付済みです。", True),
    ("今期の売上は前年比120%で着地しました。", False),
    ("ログイン用メールは taro.yamada@example.com です。", True),
    ("田中さんが昨日ランチでカレーを食べていました。", None),  # グレーゾーン
]

QUESTIONS = {
    "has_pii": Noul(
        instructions="この文章に個人を特定できる情報（個人情報）が含まれているか？",
        criteria={
            "true": "氏名・電話番号・住所・メールアドレスなど、特定の個人を識別できる情報が含まれる",
            "false": "特定の個人を識別できる情報は含まれない",
        },
    ),
    "pii_type": Choice(
        instructions="最も主要な個人情報の種類は？",
        criteria={
            "none": "個人情報なし",
            "name": "氏名のみ",
            "contact": "電話番号やメールアドレスなどの連絡先",
            "address": "住所",
        },
    ),
}

THRESHOLD = 0.5


def main() -> None:
    ok = total = 0
    with TypeSafeClient(model="jev-latest") as client:
        for text, expected in SAMPLES:
            res = client.system_one(state=text, questions=QUESTIONS)
            p = res.nouls["has_pii"].noul
            kind = res.choices["pii_type"].choice
            pred = p >= THRESHOLD
            if expected is None:
                mark = "  -"
            else:
                total += 1
                ok += pred == expected
                mark = " OK" if pred == expected else " NG"
            print(f"[{mark}] P(PII)={p:.2f} 種類={kind:<8} {text}")
    print(f"\n正解率: {ok}/{total}")


if __name__ == "__main__":
    main()

            p = res.nouls["has_pii"].noul
            kind = res.choices["pii_type"].choice

    "pii_type": Choice(
        instructions="最も主要な個人情報の種類は？",
        criteria={
            "none": "個人情報なし",
            "name": "氏名のみ",
            "contact": "電話番号やメールアドレスなどの連絡先",
            "address": "住所",
        },

"""Jev（TypeSafe AI）でクレジット決済の不正検知を試すサンプル

実行:
    python jev_fraud_test.py                     # 内蔵サンプルで実行
    python jev_fraud_test.py transactions.csv    # 自前のCSVで実行
    python jev_fraud_test.py --template          # CSVのひな型 (transactions_template.csv) を出力

出力:
    画面に判定結果、fraud_results.csv に1件ずつの詳細、最後にしきい値ごとの検知漏れ/誤検知表

注意: カード番号は下4桁などマスク済みの値だけを渡してください（生のPANは送らない）。
"""
import csv
import sys
from datetime import datetime

from typesafe_sdk import Choice, Noul, Score, TypeSafeClient, TypeSafeError

FRAUD_TH = 0.7    # これ以上 → 拒否（DECLINE）
REVIEW_TH = 0.3   # これ以上 → 目視確認（REVIEW）
RESULT_CSV = "fraud_results.csv"

FIELDS = [
    "id", "amount", "currency", "merchant", "mcc", "time", "channel",
    "card_country", "ip_country", "home_area", "avg_amount", "tx_last_1h",
    "new_device", "shipping_match", "prev_location", "prev_hours_ago",
    "distance_km", "travel_notice", "note", "expected",
]

# expected: True=不正, False=正常, None=グレー（正解率の計算から除外）
SAMPLES = [
    dict(id="T001", amount=3480, currency="JPY", merchant="コンビニ（東京都新宿区）", mcc="5411",
         time="2026-09-28 12:15", channel="店頭IC", card_country="JP", ip_country="-",
         home_area="東京都", avg_amount=5000, tx_last_1h=1, new_device="no", shipping_match="-",
         prev_location="東京都新宿区", prev_hours_ago=20, distance_km=0, travel_notice="no",
         note="", expected=False),
    dict(id="T002", amount=298000, currency="JPY", merchant="家電ECサイト", mcc="5732",
         time="2026-09-28 03:42", channel="EC", card_country="JP", ip_country="NG",
         home_area="大阪府", avg_amount=8000, tx_last_1h=6, new_device="yes",
         shipping_match="no（転送業者の住所）", prev_location="-", prev_hours_ago="", distance_km="",
         travel_notice="no", note="直前に少額決済の失敗が5回", expected=True),
    dict(id="T003", amount=120, currency="JPY", merchant="海外デジタルコンテンツ", mcc="5815",
         time="2026-09-28 04:01", channel="EC", card_country="JP", ip_country="US",
         home_area="福岡県", avg_amount=6000, tx_last_1h=14, new_device="yes", shipping_match="-",
         prev_location="-", prev_hours_ago="", distance_km="", travel_notice="no",
         note="", expected=True),
    dict(id="T004", amount=45000, currency="JPY", merchant="ホテル（京都市）", mcc="7011",
         time="2026-09-27 15:30", channel="EC", card_country="JP", ip_country="JP",
         home_area="神奈川県", avg_amount=12000, tx_last_1h=1, new_device="no", shipping_match="-",
         prev_location="神奈川県横浜市", prev_hours_ago=48, distance_km=370, travel_notice="no",
         note="毎年この時期に旅行予約あり", expected=False),
    dict(id="T005", amount=86000, currency="JPY", merchant="ブランド品店（パリ）", mcc="5944",
         time="2026-09-28 18:20", channel="店頭IC", card_country="JP", ip_country="-",
         home_area="東京都", avg_amount=15000, tx_last_1h=1, new_device="no", shipping_match="-",
         prev_location="東京都港区（店頭IC）", prev_hours_ago=3, distance_km=9700, travel_notice="no",
         note="", expected=True),
    dict(id="T006", amount=65000, currency="JPY", merchant="家電ECサイト", mcc="5732",
         time="2026-09-28 21:10", channel="EC", card_country="JP", ip_country="JP",
         home_area="愛知県", avg_amount=9000, tx_last_1h=1, new_device="yes", shipping_match="yes",
         prev_location="愛知県名古屋市", prev_hours_ago=5, distance_km=5, travel_notice="no",
         note="スマホ機種変更直後", expected=None),
    dict(id="T007", amount=72000, currency="JPY", merchant="レストラン（パリ）", mcc="5812",
         time="2026-09-28 20:05", channel="店頭IC", card_country="JP", ip_country="-",
         home_area="東京都", avg_amount=15000, tx_last_1h=1, new_device="no", shipping_match="-",
         prev_location="成田空港（店頭IC）", prev_hours_ago=16, distance_km=9700, travel_notice="yes",
         note="", expected=False),
    dict(id="T008", amount=500, currency="JPY", merchant="寄付サイト", mcc="8398",
         time="2026-09-28 02:30", channel="EC", card_country="JP", ip_country="JP",
         home_area="北海道", avg_amount=7000, tx_last_1h=9, new_device="yes", shipping_match="-",
         prev_location="-", prev_hours_ago="", distance_km="", travel_notice="no",
         note="", expected=True),
    dict(id="T009", amount=150000, currency="JPY", merchant="家具ECサイト", mcc="5712",
         time="2026-09-27 11:00", channel="EC", card_country="JP", ip_country="JP",
         home_area="埼玉県", avg_amount=10000, tx_last_1h=1, new_device="no", shipping_match="yes",
         prev_location="埼玉県さいたま市", prev_hours_ago=30, distance_km=0, travel_notice="no",
         note="引っ越し直後", expected=False),
]

QUESTIONS = {
    "is_fraud": Noul(
        instructions="この取引は不正利用（本人以外による利用）か？",
        criteria={
            "true": "カード情報の盗用・なりすまし・テスト決済など、本人以外による不正な取引である",
            "false": "本人による正常な取引である（普段と違っても本人の行動として説明がつく）",
        },
    ),
    "risk_level": Score(
        instructions="この取引の不正リスクの高さは？",
        criteria=[
            "低：普段の利用傾向と一致し、不審な点がない",
            "中：普段と異なる点があるが、本人の利用として説明がつく",
            "高：複数の不審なシグナルが重なっており、不正の可能性が高い",
        ],
    ),
    "pattern": Choice(
        instructions="最も当てはまる不正パターンは？",
        criteria={
            "none": "不正パターンに該当しない",
            "card_testing": "短時間に少額決済が連続するなど、盗んだカードが使えるか試すテスト決済",
            "account_takeover": "新しい端末・国外IP・転送先配送など、なりすましによる購入",
            "impossible_travel": "前回決済からの経過時間では物理的に移動できない場所で決済されている",
            "unusual_amount": "他に不審な点はないが、普段の利用額から大きく外れた金額",
        },
    ),
}


def _num(v):
    try:
        return float(v)
    except (TypeError, ValueError):
        return None


def to_state(tx: dict) -> str:
    """取引データを Jev に渡すテキストに整形（計算できる特徴量はここで計算）"""
    amount, avg = _num(tx["amount"]), _num(tx["avg_amount"])
    ratio = f"平均の {amount / avg:.1f} 倍" if amount and avg else "不明"

    hours, dist = _num(tx.get("prev_hours_ago")), _num(tx.get("distance_km"))
    if hours and dist is not None:
        speed = dist / hours
        feasible = "移動可能" if speed <= 900 else "移動不可能（旅客機の速度 約900km/h を超える）"
        travel = (f"前回決済: {tx['prev_location']}、{hours:g}時間前、距離 約{dist:g}km\n"
                  f"必要な移動速度: 約{speed:,.0f}km/h → {feasible}")
    else:
        travel = "前回決済: 情報なし"

    try:
        hour = datetime.strptime(tx["time"], "%Y-%m-%d %H:%M").hour
        time_note = "（深夜帯）" if hour < 6 else ""
    except ValueError:
        time_note = ""

    lines = [
        f"金額: {tx['amount']} {tx['currency']}（この会員の平均利用額 {tx['avg_amount']}、{ratio}）",
        f"加盟店: {tx['merchant']}（MCC {tx['mcc']}）",
        f"日時: {tx['time']}{time_note}",
        f"チャネル: {tx['channel']}",
        f"カード発行国: {tx['card_country']} / 接続元IPの国: {tx['ip_country']}",
        f"会員の居住地: {tx['home_area']}",
        travel,
        f"海外渡航の事前登録: {tx.get('travel_notice', '不明')}",
        f"直近1時間の決済回数: {tx['tx_last_1h']}",
        f"新しい端末からの利用: {tx['new_device']}",
        f"配送先が登録住所と一致: {tx['shipping_match']}",
    ]
    if tx.get("note"):
        lines.append(f"補足: {tx['note']}")
    return "\n".join(lines)


def decide(p: float) -> str:
    if p >= FRAUD_TH:
        return "DECLINE"
    if p >= REVIEW_TH:
        return "REVIEW"
    return "APPROVE"


def load_csv(path: str) -> list[dict]:
    rows = []
    with open(path, encoding="utf-8-sig") as f:
        for r in csv.DictReader(f):
            exp = (r.get("expected") or "").strip().lower()
            r["expected"] = {"true": True, "1": True, "false": False, "0": False}.get(exp)
            rows.append(r)
    return rows


def write_template() -> None:
    with open("transactions_template.csv", "w", encoding="utf-8-sig", newline="") as f:
        w = csv.DictWriter(f, fieldnames=FIELDS)
        w.writeheader()
        for s in SAMPLES[:3]:
            w.writerow({**s, "expected": "" if s["expected"] is None else str(s["expected"]).lower()})
    print("transactions_template.csv を出力しました")


def threshold_table(results: list[dict]) -> None:
    labeled = [r for r in results if r["expected"] is not None]
    if not labeled:
        return
    print("\nしきい値ごとの精度（このしきい値以上を『検知』とした場合）")
    print(" しきい値  検知漏れ(FN)  誤検知(FP)  正解率")
    for th in (0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9):
        fn = sum(1 for r in labeled if r["expected"] and r["p_fraud"] < th)
        fp = sum(1 for r in labeled if not r["expected"] and r["p_fraud"] >= th)
        acc = (len(labeled) - fn - fp) / len(labeled)
        mark = " ← 現在の REVIEW" if th == REVIEW_TH else (" ← 現在の DECLINE" if th == FRAUD_TH else "")
        print(f"   {th:.1f}      {fn:>4}         {fp:>4}      {acc:5.0%}{mark}")


def main() -> None:
    if "--template" in sys.argv:
        write_template()
        return
    txs = load_csv(sys.argv[1]) if len(sys.argv) > 1 else SAMPLES
    results = []

    with TypeSafeClient(model="jev-latest") as client:
        for tx in txs:
            try:
                res = client.system_one(state=to_state(tx), questions=QUESTIONS)
            except TypeSafeError as e:
                print(f"[ERR] {tx['id']} APIエラーのためスキップ: {e}")
                continue
            p = res.nouls["is_fraud"].noul
            risk = res.scores["risk_level"].score
            pattern = res.choices["pattern"].choice if p >= REVIEW_TH else "-"
            action = decide(p)
            exp = tx.get("expected")

            if exp is None:
                mark = "  -"
            else:
                mark = " OK" if (p >= REVIEW_TH) == exp else " NG"
            print(f"[{mark}] {tx['id']} {action:<7} P(不正)={p:.2f} リスク={risk:.2f}/2 "
                  f"パターン={pattern:<17} {tx['merchant']} {tx['amount']}{tx['currency']}")
            results.append(dict(id=tx["id"], action=action, p_fraud=round(p, 4),
                                risk=round(risk, 3), pattern=pattern, expected=exp,
                                merchant=tx["merchant"], amount=tx["amount"]))

    if not results:
        return
    with open(RESULT_CSV, "w", encoding="utf-8-sig", newline="") as f:
        w = csv.DictWriter(f, fieldnames=list(results[0].keys()))
        w.writeheader()
        w.writerows(results)

    labeled = [r for r in results if r["expected"] is not None]
    if labeled:
        ok = sum(1 for r in labeled if (r["p_fraud"] >= REVIEW_TH) == r["expected"])
        print(f"\n正解率: {ok}/{len(labeled)}（REVIEW 以上を検知とみなす）")
    threshold_table(results)
    print(f"\n詳細は {RESULT_CSV} に保存しました")


if __name__ == "__main__":
    main()