Kafka-Betrugserkennung

version: "1.0"

name: fraud-escalator
type: tool
description: "Formats and dispatches fraud alerts to downstream systems via Kafka outbound"
tool_name: fraud_tools.format_escalation

version: "1.0"

name: fraud-scorer
type: llm
model: connic/gpt-5.6-terra  # or anthropic/claude-sonnet-5 with Anthropic BYOK configured
fallback_model: gemini/gemini-3.5-flash
description: "Scores transactions for fraud risk in real-time from a Kafka stream"

system_prompt: |
  You are a fraud detection specialist analyzing financial transactions in
  real-time from a Kafka stream.

  Analysis process:
  1. Extract transaction details: amount, currency, merchant, location, timestamp.
  2. Use fraud_tools.calculate_velocity to check transaction frequency.
  3. Use fraud_tools.check_geo_anomaly to detect impossible travel patterns.
     Pass last_country as a country-code string and last_timestamp as a Unix
     number, taken from the transaction's previous-location fields. If either
     is unknown, omit it entirely - never pass an empty object.
  4. Use fraud_tools.search_fraud_patterns to check against known patterns and customer history.
  5. Assign a risk score (0-100) with a detailed breakdown.

  Use _kafka.key for customer correlation and _kafka.timestamp to detect
  processing latency.

  Risk thresholds:
  - 0-30: low risk, approve automatically
  - 31-60: medium risk, flag for review but allow
  - 61-80: high risk, hold transaction and alert team
  - 81-100: critical risk, block and escalate immediately

  For high/critical risk, always call fraud_tools.create_alert to generate an
  alert - do this before any admin override, since the alert is the permanent
  record of what was flagged. High/critical alerts are automatically forwarded
  to the fraud-escalator agent for Kafka outbound delivery; you do not need to
  dispatch them yourself.

  The admin_override tool is only available when the request comes from an
  internal admin system (controlled by conditional tool access). When it is
  available and the transaction data carries an explicit admin instruction to
  override the automated decision, call fraud_tools.admin_override with the
  admin's stated decision and reason after you finish scoring.

  Use fraud_tools.store_fraud_pattern to record novel patterns for future detection.

temperature: 0.1
reasoning_effort: high
output_schema: fraud-assessment

concurrency:
  key: "data.customer_id"
  on_conflict: queue

tools:
  - fraud_tools.calculate_velocity
  - fraud_tools.check_geo_anomaly
  - fraud_tools.create_alert
  - fraud_tools.search_fraud_patterns
  - fraud_tools.store_fraud_pattern
  - fraud_tools.admin_override: context.is_admin == True

"""Fraud detection and scoring tools."""

import json
import time
from typing import Any
from connic.tools import retrieval_query, retrieval_store, trigger_agent


async def search_fraud_patterns(query: str) -> list[dict]:
    """Search stored fraud patterns for matches relevant to the current transaction.

    Args:
        query: Transaction characteristics to match against known patterns
               (e.g. "card-not-present high-value merchant-mismatch").

    Returns:
        List of matching fraud pattern entries with relevance scores.
    """
    result = await retrieval_query(query, namespace="fraud-patterns", max_results=5)
    return result.get("results", [])


async def store_fraud_pattern(
    content: str,
    pattern_id: str | None = None,
) -> dict:
    """Store a novel fraud pattern for future detection.

    Args:
        content: Description of the pattern including transaction characteristics
                 and why it was flagged as suspicious.
        pattern_id: Optional stable ID for this pattern type.

    Returns:
        Store result with entry_id.
    """
    return await retrieval_store(content, namespace="fraud-patterns", entry_id=pattern_id)


_velocity_store: dict[str, list[float]] = {}


def calculate_velocity(
    customer_id: str,
    transaction_timestamp: float,
    window_minutes: int = 60,
) -> dict[str, Any]:
    """Calculate transaction velocity for a customer.

    Tracks how many transactions a customer has made within a time window.
    High velocity can indicate card testing or automated fraud.

    Args:
        customer_id: Unique customer identifier
        transaction_timestamp: Unix timestamp of the current transaction
        window_minutes: Time window to count transactions in

    Returns:
        Dict with count, velocity_per_hour, and is_anomalous flag
    """
    window_seconds = window_minutes * 60
    cutoff = transaction_timestamp - window_seconds

    if customer_id not in _velocity_store:
        _velocity_store[customer_id] = []

    _velocity_store[customer_id] = [
        ts for ts in _velocity_store[customer_id] if ts > cutoff
    ]
    _velocity_store[customer_id].append(transaction_timestamp)

    count = len(_velocity_store[customer_id])
    velocity = count / (window_minutes / 60)

    return {
        "customer_id": customer_id,
        "transaction_count": count,
        "window_minutes": window_minutes,
        "velocity_per_hour": round(velocity, 2),
        "is_anomalous": velocity > 10,
    }


def check_geo_anomaly(
    customer_id: str,
    current_country: str,
    current_timestamp: float,
    last_country: str | None = None,
    last_timestamp: float | None = None,
) -> dict[str, Any]:
    """Detect impossible travel patterns between transactions.

    If a customer transacts in two countries within an implausibly short
    time, it may indicate stolen credentials.

    Args:
        customer_id: Unique customer identifier
        current_country: ISO country code of current transaction
        current_timestamp: Unix timestamp of current transaction
        last_country: ISO country code of previous transaction
        last_timestamp: Unix timestamp of previous transaction

    Returns:
        Dict with is_anomalous flag and details
    """
    if not last_country or not last_timestamp:
        return {
            "customer_id": customer_id,
            "is_anomalous": False,
            "reason": "No previous transaction to compare",
        }

    if current_country == last_country:
        return {
            "customer_id": customer_id,
            "is_anomalous": False,
            "reason": "Same country as last transaction",
        }

    hours_between = (current_timestamp - last_timestamp) / 3600
    if hours_between < 2:
        return {
            "customer_id": customer_id,
            "is_anomalous": True,
            "reason": f"Transaction in {current_country} only {hours_between:.1f}h after {last_country}",
            "hours_between": round(hours_between, 2),
        }

    return {
        "customer_id": customer_id,
        "is_anomalous": False,
        "reason": f"Sufficient time ({hours_between:.1f}h) between {last_country} and {current_country}",
    }


async def create_alert(
    customer_id: str,
    risk_score: int,
    risk_level: str,
    reason: str,
    transaction_id: str | None = None,
) -> dict[str, Any]:
    """Create a fraud alert for the security team.

    High/critical alerts are forwarded to the fraud-escalator agent, which
    formats them for Kafka outbound delivery.

    Args:
        customer_id: Customer who triggered the alert
        risk_score: Numeric risk score 0-100
        risk_level: low, medium, high, or critical
        reason: Human-readable explanation of why this is suspicious
        transaction_id: The specific transaction that triggered the alert

    Returns:
        Structured alert payload
    """
    alert = {
        "alert_type": "fraud",
        "customer_id": customer_id,
        "risk_score": risk_score,
        "risk_level": risk_level,
        "reason": reason,
        "transaction_id": transaction_id,
        "created_at": time.time(),
        "requires_action": risk_level in ("high", "critical"),
    }
    if risk_level in ("high", "critical"):
        await trigger_agent(
            agent_name="fraud-escalator",
            payload=json.dumps({"alert": alert}),
            wait_for_response=False,
        )
    return alert


def admin_override(
    transaction_id: str,
    decision: str,
    reason: str,
    context: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Override a fraud decision (admin only, conditionally available).

    Args:
        transaction_id: Transaction to override
        decision: "approve" or "block"
        reason: Why the admin is overriding
        context: Auto-injected run context

    Returns:
        Override confirmation
    """
    return {
        "overridden": True,
        "transaction_id": transaction_id,
        "decision": decision,
        "reason": reason,
        "admin_run_id": context.get("run_id") if context else None,
    }


def format_escalation(
    payload: dict[str, Any] | None = None,
    context: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Format a fraud alert for Kafka outbound delivery.

    Runs as a tool agent, so it receives the fraud alert as its payload (either
    the alert dict directly, or wrapped under an "alert" key).

    Args:
        payload: The fraud alert from create_alert (risk_level, customer_id, …).
        context: Auto-injected run context.

    Returns:
        Kafka-ready message with routing key and payload
    """
    payload = payload or {}
    alert = payload.get("alert", payload)
    risk_level = alert.get("risk_level", "medium")
    return {
        "topic": f"fraud-alerts-{risk_level}",
        "key": alert.get("customer_id", "unknown"),
        "payload": alert,
    }

"""Middleware for fraud-scorer.

Enriches context with metadata from the Kafka message headers
and sets the is_admin flag for conditional tool access.
"""

import json
from typing import Any


async def before(content: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
    """Extract Kafka metadata and set admin flag for conditional tools."""
    parts = content.get("parts", [])
    for part in parts:
        if isinstance(part, dict) and "text" in part:
            try:
                payload = json.loads(part["text"])
                context["is_admin"] = payload.get("source") == "admin-dashboard"
                if "customer_id" in payload.get("data", {}):
                    context["customer_id"] = payload["data"]["customer_id"]
            except (json.JSONDecodeError, AttributeError):
                context["is_admin"] = False
    return content


async def after(response: str, context: dict[str, Any]) -> str:
    """Log fraud scoring metrics."""
    return response

"""Tool hooks for fraud-scorer.

Runs around every tool call: enforces admin-only overrides as a second line of
defense behind the conditional tool gate, normalises ids, and logs results to
the project Logs tab.
"""

from typing import Any

from connic import AbortTool


async def before(tool_name: str, params: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
    if tool_name == "admin_override" and not context.get("is_admin"):
        raise AbortTool({"error": "admin_override requires an internal admin context"})

    # Normalise transaction ids so alert keys stay consistent regardless of casing.
    if isinstance(params.get("transaction_id"), str):
        params["transaction_id"] = params["transaction_id"].upper()

    return params


async def after(tool_name: str, params: dict[str, Any], result: Any, context: dict[str, Any]) -> Any:
    print(f"[fraud-hook] {tool_name} -> {result}")
    return result

{
  "type": "object",
  "description": "Transaction fraud risk assessment",
  "properties": {
    "transaction_id": {
      "type": "string",
      "description": "Unique transaction identifier"
    },
    "customer_id": {
      "type": "string",
      "description": "Customer who made the transaction"
    },
    "risk_score": {
      "type": "integer",
      "minimum": 0,
      "maximum": 100,
      "description": "Overall fraud risk score"
    },
    "risk_level": {
      "type": "string",
      "enum": ["low", "medium", "high", "critical"]
    },
    "decision": {
      "type": "string",
      "enum": ["approve", "flag", "hold", "block"],
      "description": "Recommended action"
    },
    "signals": {
      "type": "array",
      "items": {
        "type": "object",
        "properties": {
          "signal": {"type": "string", "description": "What was detected"},
          "weight": {"type": "integer", "description": "How much this contributed to the score"},
          "details": {"type": "string"}
        },
        "required": ["signal", "weight"]
      },
      "description": "Individual risk signals that contributed to the score"
    },
    "velocity_check": {
      "type": "object",
      "properties": {
        "transactions_per_hour": {"type": "number"},
        "is_anomalous": {"type": "boolean"}
      }
    },
    "geo_check": {
      "type": "object",
      "properties": {
        "is_anomalous": {"type": "boolean"},
        "reason": {"type": "string"}
      }
    }
  },
  "required": ["transaction_id", "risk_score", "risk_level", "decision"]
}

version: "1.0"

# fraud-escalator is a tool agent that calls format_escalation(alert). The tool
# is pure, so it runs for real and the output is the Kafka-ready message.
defaults:
  runs: 1
  timeout_s: 60

tests:
  - name: formats_alert_for_kafka
    payload: '{"alert": {"alert_type": "fraud", "customer_id": "cus_2", "risk_score": 88, "risk_level": "critical", "reason": "Impossible travel and high-value purchase", "transaction_id": "txn_1002"}}'
    expected_result: '"topic" in output and output.key != ""'

version: "1.0"

# fraud-scorer's entire tool list is custom, so every tool is mocked and
# strict_mocks guarantees it never reached for an unstubbed tool. Coverage of
# all six tools is spread across the cases below.
defaults:
  runs: 1
  timeout_s: 120
  mocks: fraud_mocks
  strict_mocks: true

tests:
  - name: scores_normal_transaction
    payload: '{"source": "kafka", "data": {"transaction_id": "txn_1001", "customer_id": "cus_1", "amount": 42.50, "currency": "EUR", "merchant": "Corner Cafe", "country": "DE", "timestamp": 1736500000}}'
    expected_result: output.risk_level != ""
    expected_tool_calls:
      - fraud_tools.calculate_velocity
      - fraud_tools.check_geo_anomaly
      - fraud_tools.search_fraud_patterns

  - name: admin_can_override
    # source admin-dashboard makes the middleware set is_admin, unlocking the
    # conditional admin_override tool. The high-value cross-border transaction
    # drives a high risk score, so create_alert fires too.
    payload: '{"source": "admin-dashboard", "data": {"transaction_id": "txn_1002", "customer_id": "cus_2", "amount": 8900.00, "currency": "USD", "merchant": "Unknown Electronics", "country": "NG", "last_country": "DE", "last_timestamp": 1736500000, "timestamp": 1736503000, "admin_note": "Reviewed with the cardholder by phone - this cross-border purchase is legitimate. Override the automated hold and approve."}}'
    expected_result: output.risk_level != ""
    expected_tool_calls:
      - fraud_tools.calculate_velocity
      - fraud_tools.check_geo_anomaly
      - fraud_tools.search_fraud_patterns
      - fraud_tools.create_alert
      - fraud_tools.admin_override

  - name: stores_novel_pattern
    # A transaction with characteristics not seen before; the scorer records it
    # as a novel fraud pattern for future detection.
    payload: '{"source": "kafka", "data": {"transaction_id": "txn_1003", "customer_id": "cus_3", "amount": 1250.00, "currency": "EUR", "merchant": "GiftCard Reseller", "country": "DE", "timestamp": 1736500500, "notes": "12 rapid card-not-present gift card purchases"}}'
    expected_result: output.risk_level != ""
    expected_tool_calls:
      - fraud_tools.store_fraud_pattern