Fraud Detection

# Create environment
python3 -m venv venv && source venv/bin/activate

# Install packages
pip install praisonaiagents praisonai

# Set API key
export OPENAI_API_KEY="your-key"

# Create sample transactions
cat > transactions.json << 'EOF'
[
  {"id": "TXN001", "amount": 15000, "location": "Nigeria", "type": "wire", "user_id": "U123"},
  {"id": "TXN002", "amount": 50, "location": "USA", "type": "card", "user_id": "U456"},
  {"id": "TXN003", "amount": 9999, "location": "Russia", "type": "crypto", "user_id": "U789"},
  {"id": "TXN004", "amount": 200, "location": "USA", "type": "card", "user_id": "U123"}
]
EOF

from praisonaiagents import Agent, Agents, Task, tool
from pydantic import BaseModel
from typing import List
import json

# Structured output
class FraudAlert(BaseModel):
    transaction_id: str
    risk_level: str  # high, medium, low
    risk_score: int  # 0-100
    patterns_detected: List[str]
    action: str  # block, review, allow

# Database persistence is configured via memory={} parameter

# Tools
@tool
def analyze_transaction(transaction: str) -> str:
    """Analyze a transaction for fraud indicators."""
    txn = json.loads(transaction)
    risk_factors = []
    score = 0
    
    if txn["amount"] > 10000:
        risk_factors.append("large_amount")
        score += 30
    if txn["location"] in ["Nigeria", "Russia", "China"]:
        risk_factors.append("high_risk_country")
        score += 40
    if txn["type"] == "crypto":
        risk_factors.append("crypto_transaction")
        score += 20
    if txn["type"] == "wire" and txn["amount"] > 5000:
        risk_factors.append("large_wire_transfer")
        score += 25
        
    return json.dumps({"transaction_id": txn["id"], "risk_factors": risk_factors, "score": min(score, 100)})

@tool
def check_user_history(user_id: str) -> str:
    """Check user's transaction history for anomalies."""
    # Simulated history check
    histories = {
        "U123": {"avg_transaction": 500, "fraud_flags": 0, "account_age_days": 365},
        "U456": {"avg_transaction": 100, "fraud_flags": 0, "account_age_days": 730},
        "U789": {"avg_transaction": 50, "fraud_flags": 2, "account_age_days": 30}
    }
    return json.dumps(histories.get(user_id, {"avg_transaction": 0, "fraud_flags": 0, "account_age_days": 0}))

# Agents
analyzer = Agent(
    name="TransactionAnalyzer",
    instructions="Analyze transactions for fraud indicators. Use analyze_transaction tool.",
    tools=[analyze_transaction],
    memory={
        "db": "sqlite:///fraud_alerts.db",
        "session_id": "fraud-detection"
    }
)

verifier = Agent(
    name="HistoryVerifier",
    instructions="Check user history for anomalies. Flag new accounts with large transactions.",
    tools=[check_user_history]
)

alerter = Agent(
    name="AlertGenerator",
    instructions="""Generate fraud alert with:
    - risk_level: high (score>70), medium (40-70), low (<40)
    - action: block (high), review (medium), allow (low)
    Return structured JSON."""
)

# Tasks
analyze_task = Task(
    description="Analyze this transaction: {transaction}",
    agent=analyzer,
    expected_output="Risk factors and score"
)

verify_task = Task(
    description="Check user history for user_id from previous analysis",
    agent=verifier,
    expected_output="User history analysis"
)

alert_task = Task(
    description="Generate fraud alert based on analysis and history",
    agent=alerter,
    expected_output="Structured fraud alert",
    output_pydantic=FraudAlert
)

# Run
with open("transactions.json") as f:
    transactions = json.load(f)

for txn in transactions:
    agents = Agents(agents=[analyzer, verifier, alerter], tasks=[analyze_task, verify_task, alert_task])
    result = agents.start(transaction=json.dumps(txn))
    print(f"Transaction {txn['id']}: {result}")

# Single transaction check
praisonai "Analyze this transaction for fraud: amount=15000, location=Nigeria, type=wire" --verbose

# With memory for pattern learning
praisonai "Check transaction: amount=9999, location=Russia, type=crypto" --memory --user-id fraud_team

# Batch analysis
praisonai "Analyze transactions.json for fraud patterns" --file transactions.json

framework: praisonai
topic: "fraud detection system"
roles:
  analyzer:
    role: Transaction Analyst
    goal: Identify fraud indicators in transactions
    backstory: Expert at detecting financial fraud patterns
    tasks:
      analyze:
        description: |
          Analyze transaction for:
          - Amount anomalies (>$10k = high risk)
          - Geographic risk (certain countries)
          - Transaction type risk
        expected_output: Risk score 0-100 with factors
        
  verifier:
    role: Identity Verifier
    goal: Verify user legitimacy
    backstory: Expert at user behavior analysis
    tasks:
      verify:
        description: |
          Check user history:
          - Account age
          - Previous fraud flags
          - Transaction patterns
        expected_output: Verification status
        
  alerter:
    role: Alert Manager
    goal: Generate appropriate fraud alerts
    backstory: Expert at risk-based decision making
    tasks:
      alert:
        description: |
          Generate alert:
          - HIGH: score>70, action=block
          - MEDIUM: score 40-70, action=review
          - LOW: score<40, action=allow
        expected_output: Structured alert with action

praisonai agents.yaml --verbose

# View alert history
praisonai --history 20 --user-id fraud_team

# Check metrics
praisonai --metrics

# Export alerts
praisonai --save fraud_alerts

from praisonaiagents import Agent, tool
import json

@tool
def quick_fraud_check(amount: float, location: str, txn_type: str) -> str:
    """Quick fraud risk assessment."""
    score = 0
    if amount > 10000: score += 30
    if location in ["Nigeria", "Russia"]: score += 40
    if txn_type == "crypto": score += 20
    
    risk = "high" if score > 70 else "medium" if score > 40 else "low"
    action = "block" if risk == "high" else "review" if risk == "medium" else "allow"
    return json.dumps({"risk": risk, "score": score, "action": action})

agent = Agent(
    name="FraudAPI",
    instructions="Assess fraud risk for transactions.",
    tools=[quick_fraud_check]
)

agent.launch(path="/fraud-check", port=8000)

curl -X POST http://localhost:8000/fraud-check \
  -H "Content-Type: application/json" \
  -d '{"message": "Check fraud risk: amount=15000, location=Nigeria, type=wire"}'

rm -f fraud_alerts.db transactions.json
deactivate