Decision Boundary Scenario: Dynamic Fraud Thresholds Without Deploys

Risk team wants to change the fraud threshold weekly without a deploy. Where does that rule live — and who's accountable when it's wrong?

The Setup

You're running a real-time fraud detection pipeline for a payment platform. The current logic is hardcoded:

# ❌ THE PROBLEM: Hardcoded fraud threshold
FRAUD_THRESHOLD = 50000  # Flag transactions above ₹50,000

def detect_fraud(transaction):
    if transaction.amount > FRAUD_THRESHOLD:
        return "FLAGGED"
    return "LEGITIMATE"

The Risk team files a request every week: "Change the threshold from ₹50,000 to ₹45,000" or "Add a new rule: flag all international transactions above ₹25,000".

Each change currently requires:

1. A code change in the Spark job
2. A PR review + approval
3. A CI/CD pipeline run
4. A rolling restart of the streaming job

Total turnaround: 2-4 hours minimum, often 1-2 days.

The Risk team needs this in minutes, not days.

The Architecture: Externalizing Decision Boundaries

The core principle is separation of concerns: the pipeline logic (how data flows) lives in code, but the business rules (what constitutes fraud) live in a dynamic configuration store that non-engineers can update.

graph LR
    A["Risk Team<br/>(Business Users)"] -->|"Update rules<br/>via UI/API"| B["Config Store<br/>(Delta Table / Redis)"]
    B -->|"Broadcast<br/>every N seconds"| C["Spark Streaming Job"]
    D["Kafka<br/>(Transaction Events)"] -->|"Main stream"| C
    C -->|"Legitimate"| E["Main Database"]
    C -->|"Flagged"| F["Fraud Review Queue"]
    B -->|"Every change logged"| G["Audit Trail<br/>(Who, What, When)"]

Where Does the Rule Live?

There are multiple options, each with different trade-offs:

Option 1: Delta Table (Recommended for Spark Ecosystems)

Store the rules in a Delta Lake table that the streaming job reads periodically:

-- Delta table: fraud_rules
CREATE TABLE fraud_rules (
    rule_id         STRING,
    rule_name       STRING,
    condition_field STRING,
    operator        STRING,
    threshold_value DOUBLE,
    currency        STRING,
    is_active       BOOLEAN,
    priority        INT,
    created_by      STRING,
    created_at      TIMESTAMP,
    approved_by     STRING,
    approved_at     TIMESTAMP,
    version         INT
);

Sample data:

+----------+-------------------------+----------------+----------+---------+-------+---------+----------+
| rule_id  |       rule_name         |condition_field | operator |threshold| curr  | active  | priority |
+----------+-------------------------+----------------+----------+---------+-------+---------+----------+
| FR-001   | High Value Transaction  | amount         | >        | 50000   | INR   | true    | 1        |
| FR-002   | International High Val  | amount         | >        | 25000   | USD   | true    | 2        |
| FR-003   | Rapid Fire (>5 in 1min) | txn_count_1min | >        | 5       | *     | true    | 3        |
| FR-004   | Night Owl (1AM-5AM)     | txn_hour       | BETWEEN  | 1,5     | INR   | false   | 4        |
+----------+-------------------------+----------------+----------+---------+-------+---------+----------+

Why Delta Table?

• Version History: Delta's time travel gives you a full audit trail for free: SELECT * FROM fraud_rules VERSION AS OF 5
• ACID Transactions: Rules are updated atomically — no partial reads.
• Native to Spark: No external system needed. Spark reads Delta natively.

Option 2: Redis / AWS AppConfig (Low-Latency Use Cases)

For sub-second rule updates (e.g., turning off a rule during an active attack), use a key-value store:

import redis
import json

r = redis.Redis(host='redis-cluster.internal', port=6379, db=0)

# Risk team updates a rule via API
rule_update = {
    "rule_id": "FR-001",
    "threshold_value": 45000,
    "updated_by": "risk_analyst_priya",
    "updated_at": "2026-05-30T18:30:00Z",
    "reason": "Increased fraud activity detected in last 24h"
}

r.hset("fraud_rules", "FR-001", json.dumps(rule_update))
r.publish("fraud_rules_channel", json.dumps(rule_update))  # Notify consumers

Implementation: Dynamic Rules in Spark Structured Streaming

Step 1: Load Rules as a Broadcast Variable (Refreshed Periodically)

from pyspark.sql import SparkSession, functions as F
from pyspark.sql.types import *
import time
import threading

spark = SparkSession.builder \
    .appName("FraudDetectionWithDynamicRules") \
    .getOrCreate()

# ============================================================
# DYNAMIC RULE LOADER (refreshes every 60 seconds)
# ============================================================
class DynamicRuleManager:
    """Manages fraud detection rules that can be updated without redeploying."""

    def __init__(self, spark, rules_table_path):
        self.spark = spark
        self.rules_table_path = rules_table_path
        self.broadcast_rules = None
        self.rule_version = 0
        self._refresh_rules()
        self._start_refresh_thread()

    def _refresh_rules(self):
        """Read the latest rules from the Delta table and broadcast them."""
        rules_df = self.spark.read.format("delta").load(self.rules_table_path) \
            .filter(F.col("is_active") == True) \
            .orderBy("priority")

        # Collect rules to driver (small dataset, typically < 100 rules)
        rules_list = rules_df.collect()

        # Broadcast to all executors
        self.broadcast_rules = self.spark.sparkContext.broadcast(rules_list)
        self.rule_version += 1
        print(f"✅ Rules refreshed (version {self.rule_version}): {len(rules_list)} active rules loaded")

    def _start_refresh_thread(self):
        """Background thread that refreshes rules every 60 seconds."""
        def refresh_loop():
            while True:
                time.sleep(60)  # Refresh interval
                try:
                    # Destroy old broadcast and create new one
                    if self.broadcast_rules:
                        self.broadcast_rules.unpersist()
                    self._refresh_rules()
                except Exception as e:
                    print(f"⚠️ Rule refresh failed (keeping old rules): {e}")

        thread = threading.Thread(target=refresh_loop, daemon=True)
        thread.start()

    def get_rules(self):
        return self.broadcast_rules

# Initialize the rule manager
rule_manager = DynamicRuleManager(spark, "s3://config/fraud_rules")

Step 2: Apply Rules in the Streaming Pipeline

# ============================================================
# RULE EVALUATION UDF
# ============================================================
def evaluate_fraud_rules(amount, currency, txn_hour, txn_count_1min):
    """Evaluate a transaction against all active fraud rules."""
    rules = rule_manager.get_rules().value

    triggered_rules = []

    for rule in rules:
        condition_field = rule["condition_field"]
        operator = rule["operator"]
        threshold = rule["threshold_value"]
        rule_currency = rule["currency"]

        # Currency filter
        if rule_currency != "*" and currency != rule_currency:
            continue

        # Get the field value
        field_map = {
            "amount": amount,
            "txn_hour": txn_hour,
            "txn_count_1min": txn_count_1min
        }
        field_value = field_map.get(condition_field)

        if field_value is None:
            continue

        # Evaluate the condition
        if operator == ">" and field_value > threshold:
            triggered_rules.append(rule["rule_id"])
        elif operator == "<" and field_value < threshold:
            triggered_rules.append(rule["rule_id"])
        elif operator == "BETWEEN":
            low, high = threshold  # threshold is a tuple for BETWEEN
            if low <= field_value <= high:
                triggered_rules.append(rule["rule_id"])

    return triggered_rules if triggered_rules else None

evaluate_udf = F.udf(evaluate_fraud_rules, ArrayType(StringType()))

Step 3: Process the Stream

# Read transaction events from Kafka
transactions_stream = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "broker1:9092") \
    .option("subscribe", "payment_transactions") \
    .option("maxOffsetsPerTrigger", 500000) \
    .load()

# Parse the transaction payload
parsed = transactions_stream.select(
    F.from_json(F.col("value").cast("string"), txn_schema).alias("txn")
).select("txn.*")

# Apply dynamic fraud rules
flagged = parsed.withColumn(
    "triggered_rules",
    evaluate_udf(
        F.col("amount"), 
        F.col("currency"), 
        F.hour("event_timestamp"),
        F.col("txn_count_1min")
    )
).withColumn(
    "is_fraud",
    F.col("triggered_rules").isNotNull()
).withColumn(
    "rule_version",
    F.lit(rule_manager.rule_version)  # Track which rule version was applied
)

# Route legitimate and flagged transactions
legitimate = flagged.filter(~F.col("is_fraud"))
fraud_flagged = flagged.filter(F.col("is_fraud"))

# Write legitimate transactions to main database
legitimate.writeStream \
    .format("delta") \
    .option("checkpointLocation", "/checkpoints/legitimate") \
    .start("s3://datalake/transactions/legitimate/")

# Write flagged transactions to fraud review queue
fraud_flagged.writeStream \
    .format("delta") \
    .option("checkpointLocation", "/checkpoints/fraud") \
    .start("s3://datalake/transactions/fraud_review/")

Who's Accountable When It's Wrong?

This is the hardest question — and it's NOT a technical problem, it's an organizational one.

The Accountability Framework

┌─────────────────────────────────────────────────────────────────────┐
│                    ACCOUNTABILITY MATRIX                            │
├─────────────────────┬───────────────────────┬───────────────────────┤
│        ROLE         │    RESPONSIBLE FOR    │    ACCOUNTABLE FOR    │
├─────────────────────┼───────────────────────┼───────────────────────┤
│ Risk Analyst        │ Writing the rule      │ Business correctness  │
│ (e.g., Priya)       │ Setting the threshold │ of the threshold      │
│                     │                       │ value                 │
├─────────────────────┼───────────────────────┼───────────────────────┤
│ Risk Manager        │ APPROVING the rule    │ Ensuring the rule     │
│ (e.g., VP Risk)     │ change before it goes │ aligns with business  │
│                     │ live                  │ policy & risk appetite│
├─────────────────────┼───────────────────────┼───────────────────────┤
│ Data Engineer       │ Building the pipeline │ System reliability    │
│ (e.g., Mukesh)      │ and rule engine       │ and rule application  │
│                     │                       │ correctness           │
├─────────────────────┼───────────────────────┼───────────────────────┤
│ Platform / SRE      │ Monitoring & alerting │ System uptime and     │
│                     │                       │ rule propagation      │
│                     │                       │ latency               │
└─────────────────────┴───────────────────────┴───────────────────────┘

The Approval Workflow

Rules must go through an approval chain before going live — just like code PRs:

Step 1: Risk Analyst creates/modifies a rule
        ↓
Step 2: System validates the rule (syntax check, range check)
        ↓
Step 3: Risk Manager reviews and APPROVES
        ↓
Step 4: Rule enters "STAGED" state (active on shadow pipeline only)
        ↓
Step 5: Shadow pipeline runs for 24h — compare flagged vs not flagged
        ↓
Step 6: If shadow results acceptable → Risk Manager promotes to "ACTIVE"
        ↓
Step 7: Rule goes live. Audit log records: WHO, WHAT, WHEN, WHY.

The Audit Trail (Non-Negotiable)

Every rule change MUST be logged with complete provenance:

# Audit log entry for every rule change
audit_entry = {
    "change_id": "CHG-2026-05-30-001",
    "rule_id": "FR-001",
    "field_changed": "threshold_value",
    "old_value": 50000,
    "new_value": 45000,
    "changed_by": "risk_analyst_priya@company.com",
    "approved_by": "vp_risk_sharma@company.com",
    "changed_at": "2026-05-30T18:30:00Z",
    "reason": "Increased fraud pattern detected in electronics category",
    "ticket_id": "RISK-4521",
    "shadow_test_result": "2.3% increase in flagged txns, 0.1% false positive increase",
    "rollback_plan": "Revert to version 12 of fraud_rules Delta table"
}

Rollback Capability

If a bad rule causes false positives (blocking legitimate customers), rollback must be instant:

# INSTANT ROLLBACK using Delta Lake Time Travel
# Revert fraud rules to the previous version
spark.sql("""
    RESTORE TABLE fraud_rules TO VERSION AS OF 12
""")

# OR: Deactivate a specific rule immediately
spark.sql("""
    UPDATE fraud_rules 
    SET is_active = false, 
        deactivated_by = 'vp_risk_sharma',
        deactivated_at = current_timestamp(),
        deactivation_reason = 'False positive rate exceeded 5%'
    WHERE rule_id = 'FR-001'
""")

Testing Dynamic Rules: The Shadow Pipeline

Before any rule goes live, it runs in shadow mode — processing real data but NOT affecting real outcomes:

# Shadow pipeline: apply new rules alongside current rules
def shadow_test(batch_df, batch_id):
    # Apply CURRENT (production) rules
    current_result = apply_rules(batch_df, production_rules)

    # Apply NEW (candidate) rules
    candidate_result = apply_rules(batch_df, candidate_rules)

    # Compare the two
    comparison = current_result.join(
        candidate_result.select("transaction_id", 
            F.col("is_fraud").alias("candidate_is_fraud")),
        on="transaction_id"
    ).withColumn(
        "decision_changed",
        F.col("is_fraud") != F.col("candidate_is_fraud")
    )

    # Report the impact
    changed_count = comparison.filter(F.col("decision_changed")).count()
    total_count = comparison.count()

    print(f"""
    ╔══════════════════════════════════════╗
    ║   SHADOW TEST RESULTS (Batch {batch_id})    ║
    ╠══════════════════════════════════════╣
    ║ Total transactions:    {total_count:>12} ║
    ║ Decisions changed:     {changed_count:>12} ║
    ║ Change rate:           {changed_count/total_count*100:>10.2f}% ║
    ╚══════════════════════════════════════╝
    """)

Summary

The Principle: Treat business rules like data, not like code. Data can be updated, versioned, rolled back, and audited — without touching the deployment pipeline.

Follow-Up Questions & Answers

Q1: What if the rules engine becomes a bottleneck? Reading a Delta table every 60 seconds adds latency.

A: At scale, reading from Delta every 60 seconds introduces I/O overhead. The solution depends on your latency requirements:

For ultra-low-latency, use a Kafka compacted topic for rule distribution:

# Rules published to a Kafka compacted topic
# Each key = rule id, value = latest rule JSON
# Spark reads this as a streaming source and maintains an in-memory state store
rules_stream = spark.readStream \
    .format("kafka") \
    .option("subscribe", "fraud_rules_config") \
    .option("startingOffsets", "earliest") \
    .load()

Q2: How do you A/B test a new rule before rolling it out to 100% of traffic?

A: Use a percentage-based rollout strategy embedded in the rules themselves:

{
    "rule_id": "FR-005",
    "rule_name": "New ML Score Threshold",
    "condition": "ml_fraud_score > 0.85",
    "rollout_percentage": 10,
    "traffic_split_field": "user_id",
    "control_group_action": "LOG_ONLY",
    "treatment_group_action": "FLAG_AND_BLOCK"
}

import hashlib

def should_apply_rule(user_id, rollout_pct):
    """Deterministic percentage-based routing using hash."""
    hash_val = int(hashlib.md5(user_id.encode()).hexdigest(), 16) % 100
    return hash_val < rollout_pct  # 10% → users with hash 0-9

# Apply rule to treatment group, log-only for control group
df = df.withColumn(
    "in_treatment_group",
    F.udf(should_apply_rule)(F.col("user_id"), F.lit(10))
)

The key insight: Using a hash of user_id ensures the same user is always in the same group (deterministic assignment), preventing flip-flopping between treatment and control across requests.

Q3: What happens if two rules conflict? (e.g., Rule 1 says "flag" and Rule 2 says "pass")

A: You need a conflict resolution strategy defined in the rule schema:

Resolution Strategies:
├── PRIORITY: Higher priority rule wins (use the "priority" field)
├── ALL_MATCH: ALL rules must trigger for the event to be flagged
├── ANY_MATCH: If ANY rule triggers, the event is flagged
└── WEIGHTED: Each rule contributes a score; threshold on cumulative score

# Priority-based resolution
def resolve_rules(triggered_rules):
    if not triggered_rules:
        return "LEGITIMATE"

    # Rules are sorted by priority — first match wins
    highest_priority_rule = sorted(triggered_rules, key=lambda r: r["priority"])[0]
    return highest_priority_rule["action"]

# Weighted scoring resolution
def weighted_score(triggered_rules):
    total_score = sum(rule["weight"] for rule in triggered_rules)
    return "FLAGGED" if total_score >= 100 else "LEGITIMATE"

Q4: Who has access to change the rules? How do you prevent unauthorized changes?

A: Implement role-based access control (RBAC) on the config store:

Role Hierarchy:
├── Rule Viewer:   Can read rules and audit history (any analyst)
├── Rule Author:   Can create/modify draft rules (senior risk analysts)
├── Rule Approver:  Can promote drafts to active (risk managers only)
└── Rule Admin:    Can delete rules, manage schema (platform team only)

Implementation with Delta Lake + AWS Lake Formation:

-- Only risk_approver role can UPDATE the is_active field
GRANT SELECT ON TABLE fraud_rules TO role_rule_viewer;
GRANT INSERT, UPDATE ON TABLE fraud_rules TO role_rule_author;
-- Promotion (is_active = true) requires a separate approval table
-- with dual-signature requirement

For additional security, implement dual approval — two separate risk managers must both approve before a rule goes live.

Q5: How do you measure the impact of a rule change retroactively?

A: Since every rule version is tracked, you can replay historical transactions against old and new rules:

# Retroactive impact analysis: compare rule v12 vs v13
historical_data = spark.read.format("delta") \
    .option("versionAsOf", "latest") \
    .load("s3://datalake/transactions/")

# Apply old rules (version 12)
old_rules = spark.read.format("delta") \
    .option("versionAsOf", 12) \
    .load("s3://config/fraud_rules")
old_result = apply_rules(historical_data, old_rules)

# Apply new rules (version 13)
new_rules = spark.read.format("delta") \
    .option("versionAsOf", 13) \
    .load("s3://config/fraud_rules")
new_result = apply_rules(historical_data, new_rules)

# Compare outcomes
impact = old_result.join(new_result.select("transaction_id", 
    F.col("is_fraud").alias("new_is_fraud")), on="transaction_id")

false_positives = impact.filter((~F.col("is_fraud")) & F.col("new_is_fraud")).count()
false_negatives = impact.filter(F.col("is_fraud") & (~F.col("new_is_fraud"))).count()

print(f"New rule impact: +{false_positives} new flags, -{false_negatives} removed flags")

Sub-Scenarios

Sub-Scenario A: Rule Change Causes a Fraud Surge Overnight

Situation: A risk analyst lowers the threshold from ₹50,000 to ₹5,000 (a typo — they meant ₹50,000 → ₹45,000). Overnight, 60% of all transactions are flagged as fraud. Customer support is overwhelmed with blocked payments.

What should have prevented this:

1. Range validation: The system should reject threshold values below ₹10,000 as obviously incorrect.
2. Impact simulation: Before activation, a shadow test should have shown: "This rule change will flag 60% of transactions (up from 2%)."
3. Gradual rollout: The new threshold should have been applied to 5% of traffic first, not 100%.
4. Auto-circuit-breaker: If fraud flag rate exceeds 10%, automatically revert to the previous rule version.

# Auto-revert circuit breaker
def monitor_fraud_rate(batch_df, batch_id):
    fraud_rate = batch_df.filter(F.col("is_fraud")).count() / batch_df.count()

    if fraud_rate > 0.10:  # > 10% flagged → something is wrong
        send_alert("🚨 FRAUD RATE ANOMALY: Auto-reverting to previous rule version!")
        spark.sql("RESTORE TABLE fraud_rules TO VERSION AS OF (SELECT MAX(version) - 1 FROM ...)")

Sub-Scenario B: Rules Need to Be Region-Specific

Situation: The ₹50,000 threshold makes sense for India, but the equivalent $600 threshold is too low for US transactions. Rules need to be region-aware.

Solution: Add a region dimension to the rules table:

INSERT INTO fraud_rules VALUES
    ('FR-001-IN', 'High Value INR', 'amount', '>', 50000, 'INR', 'IN', true, 1),
    ('FR-001-US', 'High Value USD', 'amount', '>', 5000, 'USD', 'US', true, 1),
    ('FR-001-EU', 'High Value EUR', 'amount', '>', 4000, 'EUR', 'EU', true, 1);

The rule evaluation UDF first filters rules by the transaction's region before applying them.

Sub-Scenario C: The Config Store (Delta Table) Goes Down

Situation: The S3 bucket hosting the fraud_rules Delta table becomes temporarily unavailable. Your streaming job can't refresh rules.

Safeguard: The DynamicRuleManager should cache the last known good rules and continue operating with stale rules rather than crashing:

def _refresh_rules(self):
    try:
        rules_df = self.spark.read.format("delta").load(self.rules_table_path)
        rules_list = rules_df.filter(F.col("is_active") == True).collect()
        self.broadcast_rules = self.spark.sparkContext.broadcast(rules_list)
        self._cached_rules = rules_list  # Cache for fallback
        self.rule_version += 1
    except Exception as e:
        print(f"⚠️ Config store unavailable. Using cached rules (version {self.rule_version})")
        # Continue with self._cached_rules — don't crash!
        send_alert(f"⚠️ Rule refresh failed: {e}. Operating with stale rules v{self.rule_version}")

Principle: A stale rule is better than no rule. The pipeline must remain operational even when the config store is temporarily down.

Sub-Scenario D: Multiple Teams Want to Manage Rules Independently

Situation: The Risk team manages fraud rules, the Compliance team manages regulatory rules, and the Marketing team manages promotional flags. They all want independent control.

Solution: Namespace isolation with a unified evaluation engine:

fraud_rules     (owned by Risk team)
compliance_rules (owned by Compliance team)
marketing_rules  (owned by Marketing team)

All three are evaluated against the same transaction stream.
Each team has RBAC access ONLY to their own namespace.
The pipeline applies ALL active rules from ALL namespaces.

Each team gets their own approval workflow, their own audit trail, and their own rollback capability — without interfering with other teams' rules.