graph/duplicate_detection.py

from __future__ import annotations

from typing import Any, Dict, List

from shield.graph.graph_store import GraphStore
from shield.models.evidence import Evidence


_WEIGHTS = {
    "email": 100,
    "phone": 95,
    "device": 95,
    "ip": 20,
    "username": 15,
}


def detect_duplicates(
    entities: Dict[str, str],
    *,
    account_hash: str,
    graph: GraphStore | None = None,
) -> Dict[str, Any]:
    graph = graph or GraphStore()
    matches = graph.find_linked_accounts(entities)
    linked_accounts: Dict[str, Dict[str, Any]] = {}
    reasons: List[str] = []
    score = 0

    for entity_type, accounts in matches.items():
        other_accounts = [
            account for account in accounts
            if account.get("account_hash") != account_hash
        ]
        if not other_accounts:
            continue
        for account in other_accounts:
            linked_accounts[account["account_hash"]] = account
        weight = _WEIGHTS.get(entity_type, 0)
        score = max(score, weight)
        if entity_type == "ip":
            reasons.append("IP is shared with existing accounts")
        else:
            reasons.append(f"{entity_type.replace('_', ' ').title()} matches existing accounts")

    matched_types = len(reasons)
    if matched_types >= 2:
        score = min(100, score + 10)

    confidence = 0.0
    if score:
        confidence = min(0.99, 0.65 + (matched_types * 0.1))

    evidence: List[Evidence] = []
    if score:
        evidence.append(Evidence(
            source="relationship_graph",
            category="multi_account_abuse",
            risk_score=score,
            weight=0.85,
            confidence=confidence,
            explanation=reasons[0] if len(reasons) == 1 else "Multiple identity signals match existing accounts",
            metadata={"matched_signal_types": list(matches.keys())},
        ))

    return {
        "duplicate_score": score,
        "confidence": round(confidence, 2),
        "linked_accounts": list(linked_accounts.keys())[:50],
        "reasons": reasons,
        "evidence": evidence,
    }