Real-Time Risk Scoring with Streaming SQL: Account Takeover, Marketplaces, Credit, and Insurance Claims

Real-time risk scoring with streaming SQL means the behavioral signals a risk decision depends on, login velocity, device novelty, cancellation history, credit utilization, claim frequency, counterparty exposure, are maintained continuously as materialized views and read with a point lookup when the decision has to be made. This guide builds that layer on RisingWave, a PostgreSQL-compatible streaming database, for five risk domains that share one architecture: account takeover, two-sided marketplaces, consumer credit, insurance claims, and intraday trading exposure.

Why Risk Scoring Fails on Batch Data

Risk decisions happen at a moment: a login, a booking request, a credit application, a claim submission, a trade. The signals that matter most are the ones from the last few minutes, and a nightly pipeline cannot see them.

• Account takeover attackers exploit stolen credentials within hours of acquiring them, and the most damaging actions, changing the recovery email and draining the wallet, happen in the first ten minutes of access. A batch job runs after the account is gone.
• Marketplace fraud is relational. A seller who attracts a cluster of new accounts in one hour, or a buyer-seller pair that transacts repeatedly with glowing reviews and poor completion, only shows up when buyer, seller, and pair state are joined at decision time.
• Credit scores from a bureau reflect last month's statement cycle. A borrower who paid down 80 percent of revolving debt yesterday still looks maxed out; one who missed three payments in two weeks still looks current.
• Insurance claims fraud appears across claims, not within one: five claims in 30 days, the same procedure billed on many patients in an hour, a claim 20 days after policy inception.
• Trading risk computed at end of day leaves an institution blind to intraday exposure buildup, which is exactly when rogue positions and counterparty defaults do their damage. Regulators under FRTB and CCAR now expect intraday monitoring.

RisingWave maintains every signal below as an incrementally updated materialized view. When an event arrives, only the affected rows change, and the risk service reads precomputed features over the PostgreSQL wire protocol in single-digit milliseconds. RisingWave is open source under Apache 2.0, written in Rust, and checkpoints state to S3, so view state survives restarts and can be rebuilt by replaying Kafka. NOW() inside a WHERE or FILTER clause acts as a temporal filter, so rolling windows of several lengths can live in one view.

Shared Architecture

Auth / session / device events  ─┐
Bookings, reviews, payments      ├──►  Kafka  ──►  RisingWave materialized views  ──►  Risk service: point lookup per entity, score, decide
Open banking transactions        │                        ▲                              │
Claims intake                    │                        │                              └──►  Kafka sinks: step-up auth, review queues, alerts
Positions and market data        ─┘      Reference data via CDC: policies, bureau snapshots, limits

Each domain follows the same three steps: define sources, define per-entity signal views with the right windows, and define a composite view or a decision-time query that joins them. Composite weights are illustrative starting points; replace them with values calibrated on labeled outcomes.

Account Takeover: Identity-Continuity Signals

Payment fraud leans on transaction artifacts the attacker must forge. Account takeover introduces none of them, because the attacker presents valid credentials. The signals are about identity continuity instead: a device never seen for this account, a login from a country the user has never been in, 47 attempts across 12 IPs in an hour, and a recovery email change three minutes after login.

The attack lifecycle shapes which signals matter when. Credential stuffing (rotating IPs, 0.1 to 2 percent success at scale) is visible as login velocity before any success. Validation and account mutation (recovery email, phone, password) are visible as post-login action velocity. Cashout follows within minutes. Signals at the first stage let you block before access is granted; signals at the third are the last chance before the point of no return.

CREATE SOURCE auth_events (
    user_id             VARCHAR,
    event_type          VARCHAR,     -- 'login_success', 'login_failure', 'mfa_bypass'
    ip_address          VARCHAR,
    ip_country          VARCHAR,
    ip_city             VARCHAR,
    device_fingerprint  VARCHAR,
    event_time          TIMESTAMPTZ
) WITH (
    connector = 'kafka',
    topic = 'auth.events',
    properties.bootstrap.server = 'kafka:9092',
    scan.startup.mode = 'latest'
) FORMAT PLAIN ENCODE JSON;

CREATE SOURCE session_events (
    session_id    VARCHAR,
    user_id       VARCHAR,
    action_type   VARCHAR,     -- 'password_change', 'email_change', 'phone_change', 'payment_method_add', 'withdrawal'
    is_forced_reset BOOLEAN,
    event_time    TIMESTAMPTZ
) WITH (
    connector = 'kafka',
    topic = 'session.events',
    properties.bootstrap.server = 'kafka:9092',
    scan.startup.mode = 'latest'
) FORMAT PLAIN ENCODE JSON;

A 90-day device history is long enough to include a secondary laptop and short enough to expire abandoned devices. A LEFT JOIN against it at login time with login_count IS NULL identifies first-seen devices:

CREATE MATERIALIZED VIEW user_device_history AS
SELECT
    user_id,
    device_fingerprint,
    COUNT(*)                                            AS login_count,
    MIN(event_time)                                     AS first_seen,
    MAX(event_time)                                     AS last_seen,
    ARRAY_AGG(DISTINCT ip_country ORDER BY ip_country)  AS countries_used
FROM auth_events
WHERE event_type = 'login_success'
  AND event_time >= NOW() - INTERVAL '90 DAYS'
GROUP BY user_id, device_fingerprint;

CREATE MATERIALIZED VIEW login_velocity AS
SELECT
    user_id,
    COUNT(*)                                                 AS login_attempts_1h,
    COUNT(DISTINCT ip_address)                               AS unique_ips_1h,
    COUNT(DISTINCT device_fingerprint)                       AS unique_devices_1h,
    COUNT(*) FILTER (WHERE event_type = 'login_failure')     AS failures_1h
FROM auth_events
WHERE event_time >= NOW() - INTERVAL '1 HOUR'
GROUP BY user_id;

CREATE MATERIALIZED VIEW post_login_actions AS
SELECT
    session_id,
    user_id,
    COUNT(*)                                                                                   AS actions_5min,
    COUNT(*) FILTER (WHERE action_type IN ('password_change', 'email_change', 'phone_change')
                       AND NOT is_forced_reset)                                                AS sensitive_changes_5min,
    COUNT(*) FILTER (WHERE action_type = 'withdrawal')                                         AS withdrawals_5min
FROM session_events
WHERE event_time >= NOW() - INTERVAL '5 MINUTES'
GROUP BY session_id, user_id;

A user with more than five unique IPs and more than ten failures in an hour before any success is a near-certain stuffing target, and that is actionable before the attacker gets in: a FIDO2 passkey challenge or temporary rate limiting. Post-login action velocity is the high-precision signal; legitimate users do not change email, phone, and password within three minutes of logging in. The is_forced_reset flag suppresses the signal when the platform itself required a password change.

Impossible travel compares each successful login with the previous one using LAG. Country-level comparison with a one-hour threshold is the simple version; production systems use stored coordinates and an implied speed above roughly 800 to 1,000 km/h as the cutoff, with VPN probability scoring below it:

CREATE MATERIALIZED VIEW travel_anomaly AS
SELECT user_id, event_time, ip_country, prev_country,
       EXTRACT(EPOCH FROM (event_time - prev_event_time)) / 3600.0 AS hours_since_last_login,
       ip_country <> prev_country
         AND EXTRACT(EPOCH FROM (event_time - prev_event_time)) < 3600 AS impossible_travel_flag
FROM (
    SELECT user_id, event_time, ip_country,
           LAG(event_time) OVER (PARTITION BY user_id ORDER BY event_time) AS prev_event_time,
           LAG(ip_country) OVER (PARTITION BY user_id ORDER BY event_time) AS prev_country
    FROM auth_events
    WHERE event_type = 'login_success'
) ordered
WHERE prev_country IS NOT NULL;

The composite score weights velocity, device novelty, post-login mutation, and travel. Above 70 interrupt the session with step-up authentication; 40 to 70 flag silently and escalate monitoring:

CREATE MATERIALIZED VIEW ato_risk_score AS
SELECT
    v.user_id,
    LEAST(1.0, v.unique_ips_1h / 5.0) * 30                                             AS velocity_score,
    CASE WHEN dh.login_count IS NULL THEN 30 WHEN dh.login_count < 3 THEN 15 ELSE 0 END AS device_novelty_score,
    LEAST(1.0, COALESCE(pa.sensitive_changes_5min, 0) / 2.0) * 30                     AS mutation_score,
    CASE WHEN ta.impossible_travel_flag THEN 10 ELSE 0 END                             AS travel_score,
    LEAST(100,
          LEAST(1.0, v.unique_ips_1h / 5.0) * 30
        + CASE WHEN dh.login_count IS NULL THEN 30 WHEN dh.login_count < 3 THEN 15 ELSE 0 END
        + LEAST(1.0, COALESCE(pa.sensitive_changes_5min, 0) / 2.0) * 30
        + CASE WHEN ta.impossible_travel_flag THEN 10 ELSE 0 END)                        AS total_risk_score
FROM login_velocity v
LEFT JOIN user_device_history dh ON v.user_id = dh.user_id
LEFT JOIN post_login_actions pa  ON v.user_id = pa.user_id
LEFT JOIN travel_anomaly ta      ON v.user_id = ta.user_id AND ta.event_time >= NOW() - INTERVAL '1 HOUR';

False positives come from predictable places. Business travelers trip impossible travel, so relax the threshold for users with high countries_used cardinality. New phones trip device novelty, so never act on it alone; a new device in the same country at a normal hour with no mutations scores low in the composite. Analyst labels on false positives should feed back into weight calibration.

Two-Sided Marketplaces: Buyer, Seller, and the Edge Between Them

Marketplace fraud lives in the relationship. A rental listing priced below market that collects deposits and never delivers, shill bidding that needs coordinated buyer and seller accounts, a gig contractor who seeds a clean rating and then spikes high-value fake completions. The buyer's history, the seller's history, and the pair's shared history each carry part of the signal.

CREATE SOURCE marketplace_events (
    event_id            VARCHAR,
    buyer_id            VARCHAR,
    seller_id           VARCHAR,
    listing_id          VARCHAR,
    event_type          VARCHAR,     -- 'booking_requested', 'booking_accepted', 'booking_completed', 'booking_cancelled', 'review_submitted', 'payment_disputed'
    outcome             VARCHAR,     -- 'completed', 'cancelled_by_buyer', 'cancelled_by_seller'
    transaction_value   NUMERIC,
    response_time_minutes INT,
    review_score        INT,
    dispute_filed       BOOLEAN,
    event_time          TIMESTAMPTZ
) WITH (
    connector = 'kafka',
    topic = 'marketplace.events',
    properties.bootstrap.server = 'kafka:9092',
    scan.startup.mode = 'latest'
) FORMAT PLAIN ENCODE JSON;

CREATE MATERIALIZED VIEW seller_reliability AS
SELECT
    seller_id,
    COUNT(*)                                                    AS bookings_30d,
    COUNT(*) FILTER (WHERE outcome = 'completed')               AS completed_30d,
    COUNT(*) FILTER (WHERE outcome = 'cancelled_by_seller')     AS seller_cancels_30d,
    AVG(response_time_minutes)                                  AS avg_response_time,
    AVG(review_score) FILTER (WHERE review_score IS NOT NULL)   AS avg_rating
FROM marketplace_events
WHERE event_time >= NOW() - INTERVAL '30 DAYS'
GROUP BY seller_id;

CREATE MATERIALIZED VIEW buyer_behavior AS
SELECT
    buyer_id,
    COUNT(*) FILTER (WHERE outcome = 'cancelled_by_buyer' AND event_time >= NOW() - INTERVAL '30 DAYS') AS buyer_cancels_30d,
    COUNT(*) FILTER (WHERE dispute_filed)                       AS disputes_90d,
    COUNT(DISTINCT seller_id)                                   AS unique_sellers_90d,
    SUM(transaction_value)                                      AS total_spend_90d
FROM marketplace_events
WHERE event_time >= NOW() - INTERVAL '90 DAYS'
GROUP BY buyer_id;

Mixed windows are deliberate: 30 days for cancellations, 90 for disputes because disputes lag the transaction. The pair view and the seller's aggregated buyer-risk view are what make marketplace scoring different from single-party fraud:

CREATE MATERIALIZED VIEW buyer_seller_history AS
SELECT
    buyer_id,
    seller_id,
    COUNT(*)                                                    AS total_interactions,
    COUNT(*) FILTER (WHERE outcome = 'completed')               AS completed_together,
    AVG(review_score) FILTER (WHERE review_score IS NOT NULL)   AS avg_review_given,
    MAX(event_time)                                             AS last_interaction_time
FROM marketplace_events
WHERE event_time >= NOW() - INTERVAL '90 DAYS'
GROUP BY buyer_id, seller_id;

CREATE MATERIALIZED VIEW seller_buyer_risk_profile AS
SELECT
    me.seller_id,
    AVG(bb.buyer_cancels_30d)                                   AS avg_buyer_cancel_rate,
    AVG(bb.disputes_90d)                                        AS avg_buyer_dispute_rate,
    COUNT(DISTINCT me.buyer_id)                                 AS distinct_buyers_30d,
    COUNT(*) FILTER (WHERE me.event_time >= NOW() - INTERVAL '7 DAYS') AS interactions_7d
FROM marketplace_events me
JOIN buyer_behavior bb ON me.buyer_id = bb.buyer_id
WHERE me.event_time >= NOW() - INTERVAL '30 DAYS'
GROUP BY me.seller_id;

CREATE MATERIALIZED VIEW seller_velocity_1h AS
SELECT seller_id, COUNT(*) AS bookings_last_1h, SUM(transaction_value) AS value_last_1h, COUNT(DISTINCT buyer_id) AS unique_buyers_last_1h
FROM marketplace_events
WHERE event_type = 'booking_requested' AND event_time >= NOW() - INTERVAL '1 HOUR'
GROUP BY seller_id;

A pair with ten transactions, every one followed by a five-star review, and a completion rate under 50 percent is either a loyal customer or a shill ring; the seller's buyer-risk profile and the hourly velocity disambiguate. At decision time the risk service runs four point lookups (2 to 10 ms), scores them with a rules function or a trained model (1 to 5 ms), and returns inside 20 ms. A starting rule set might add 0.3 for more than three seller cancellations, 0.4 for more than 20 bookings in an hour, 0.35 for more than two buyer disputes, and 0.3 for a repeated pair with poor completion.

Cold start is handled with signals that exist at creation time: listing price far below category market, incomplete profile, account age under seven days. The highest-risk case is both parties being new; route it to manual review regardless of score.

Consumer Credit: Behavioral Signals Between Bureau Pulls

A bureau score computed on the first of the month cannot see this week's behavior. Open banking transaction feeds can. The behavioral layer does not replace the bureau; it sits beside it and catches the movement between pulls.

CREATE SOURCE account_transactions (
    transaction_id    VARCHAR,
    customer_id       VARCHAR,
    account_id        VARCHAR,
    account_type      VARCHAR,    -- 'checking', 'savings', 'credit', 'loan'
    txn_category      VARCHAR,    -- 'purchase', 'payment', 'cash_advance', 'fee', 'transfer'
    amount            NUMERIC,
    running_balance   NUMERIC,
    credit_limit      NUMERIC,
    txn_time          TIMESTAMPTZ
) WITH (
    connector = 'kafka',
    topic = 'open.banking.transactions',
    properties.bootstrap.server = 'broker:9092',
    scan.startup.mode = 'latest'
) FORMAT PLAIN ENCODE JSON;

CREATE TABLE bureau_snapshots (
    customer_id        VARCHAR PRIMARY KEY,
    bureau_score       INTEGER,
    derogatory_marks   INTEGER,
    total_debt         NUMERIC,
    snapshot_date      DATE
);

CREATE MATERIALIZED VIEW credit_behavioral_signals AS
SELECT
    customer_id,
    window_start,
    window_end,
    AVG(running_balance / NULLIF(credit_limit, 0)) FILTER (WHERE account_type = 'credit')  AS avg_utilization_30d,
    MAX(running_balance / NULLIF(credit_limit, 0)) FILTER (WHERE account_type = 'credit')  AS peak_utilization_30d,
    COUNT(*)    FILTER (WHERE txn_category = 'payment' AND account_type = 'credit')        AS credit_payments_30d,
    SUM(amount) FILTER (WHERE txn_category = 'payment' AND account_type = 'credit')        AS total_credit_payments_30d,
    COUNT(*)    FILTER (WHERE txn_category = 'cash_advance')                               AS cash_advances_30d,
    COUNT(*)    FILTER (WHERE running_balance < 0 AND account_type = 'checking')           AS overdraft_events_30d,
    MIN(running_balance) FILTER (WHERE account_type = 'checking')                          AS min_checking_balance_30d,
    STDDEV(amount) FILTER (WHERE txn_category = 'purchase')                                AS purchase_volatility
FROM HOP(account_transactions, txn_time, INTERVAL '1 DAY', INTERVAL '30 DAYS')
GROUP BY customer_id, window_start, window_end;

A temporal join picks the bureau snapshot that was current at the start of each window, so a later bureau refresh does not rewrite history, and a rule layer turns the signals into a behavioral flag that the decision model consumes alongside the bureau score:

CREATE MATERIALIZED VIEW composite_credit_profile AS
SELECT
    s.customer_id,
    b.bureau_score,
    b.derogatory_marks,
    s.avg_utilization_30d,
    s.peak_utilization_30d,
    s.cash_advances_30d,
    s.overdraft_events_30d,
    s.credit_payments_30d,
    CASE
        WHEN s.cash_advances_30d >= 3 OR s.overdraft_events_30d >= 5 OR s.peak_utilization_30d > 0.95 THEN 'ELEVATED'
        WHEN s.avg_utilization_30d > 0.70 OR s.overdraft_events_30d >= 2                                THEN 'MODERATE'
        ELSE 'NORMAL'
    END AS behavioral_risk_flag,
    b.snapshot_date AS bureau_as_of,
    s.window_end    AS signals_as_of
FROM credit_behavioral_signals s
JOIN bureau_snapshots FOR SYSTEM_TIME AS OF s.window_start AS b ON s.customer_id = b.customer_id;

The two timestamps, bureau_as_of and signals_as_of, belong in every decision record. They make it possible to explain a decision later and to measure how much the behavioral layer moved outcomes relative to the bureau alone.

Insurance Claims: Patterns Across Claims

Insurance fraud costs the US industry more than $80 billion a year, much of it staged accidents, duplicate claims, and provider schemes that are detectable from patterns if you look quickly. The scoring views join the claims stream with policy data that arrives through CDC from the policy system.

CREATE SOURCE insurance_claims (
    claim_id            VARCHAR,
    policy_id           VARCHAR,
    claimant_id         VARCHAR,
    provider_id         VARCHAR,
    claim_type          VARCHAR,     -- 'AUTO', 'HEALTH', 'PROPERTY', 'LIFE'
    claimed_amount      DECIMAL(18,2),
    incident_date       DATE,
    submission_time     TIMESTAMPTZ,
    ip_address          VARCHAR,
    device_fingerprint  VARCHAR
) WITH (
    connector = 'kafka',
    topic = 'claims-intake',
    properties.bootstrap.server = 'kafka:9092',
    scan.startup.mode = 'latest'
) FORMAT PLAIN ENCODE JSON;

CREATE TABLE policy_data (
    policy_id        VARCHAR PRIMARY KEY,
    claimant_id      VARCHAR,
    policy_type      VARCHAR,
    inception_date   DATE,
    coverage_limit   DECIMAL(18,2),
    is_active        BOOLEAN
) WITH (
    connector = 'postgres-cdc',
    hostname = 'policy-db.internal', port = '5432',
    username = 'rw_reader', password = 'secret',
    database.name = 'policies', schema.name = 'public', table.name = 'policies'
);

CREATE MATERIALIZED VIEW claimant_claim_profile AS
SELECT
    claimant_id,
    COUNT(*)                        AS claims_last_30d,
    SUM(claimed_amount)             AS total_claimed_30d,
    AVG(claimed_amount)             AS avg_claim_amount,
    COUNT(DISTINCT claim_type)      AS claim_type_diversity,
    COUNT(DISTINCT provider_id)     AS distinct_providers
FROM insurance_claims
WHERE submission_time >= NOW() - INTERVAL '30 DAYS'
GROUP BY claimant_id;

CREATE MATERIALIZED VIEW provider_billing_anomalies AS
SELECT window_start, window_end, provider_id, claim_type,
       COUNT(*) AS claims_in_window, COUNT(DISTINCT claimant_id) AS unique_claimants, SUM(claimed_amount) AS total_billed
FROM TUMBLE(insurance_claims, submission_time, INTERVAL '1 HOUR')
GROUP BY window_start, window_end, provider_id, claim_type
HAVING COUNT(*) > 10;

CREATE MATERIALIZED VIEW potential_duplicates AS
SELECT window_start, window_end, claimant_id, claim_type,
       COUNT(*) AS claim_count, MIN(claim_id) AS first_claim_id, MAX(claim_id) AS latest_claim_id
FROM TUMBLE(insurance_claims, submission_time, INTERVAL '7 DAYS')
GROUP BY window_start, window_end, claimant_id, claim_type
HAVING COUNT(*) > 1;

Per-claim flags sum into a score, and claims with two or more flags go to the adjuster review queue before they reach the payment queue:

CREATE MATERIALIZED VIEW claim_fraud_signals AS
SELECT
    c.claim_id, c.claimant_id, c.policy_id, c.claimed_amount, c.submission_time,
    (cp.claims_last_30d > 5)::INT                                        AS high_frequency_flag,
    (c.claimed_amount > cp.avg_claim_amount * 3)::INT                    AS high_amount_flag,
    (c.incident_date <= p.inception_date + INTERVAL '30 DAYS')::INT      AS new_policy_flag,
    (cp.distinct_providers > 4)::INT                                     AS multi_provider_flag,
    (NOT p.is_active)::INT                                               AS inactive_policy_flag
FROM insurance_claims c
LEFT JOIN claimant_claim_profile cp ON c.claimant_id = cp.claimant_id
LEFT JOIN policy_data p             ON c.policy_id = p.policy_id;

CREATE SINK fraud_review_queue AS
SELECT claim_id, claimant_id, policy_id, claimed_amount, submission_time,
       high_frequency_flag + high_amount_flag + new_policy_flag + multi_provider_flag + inactive_policy_flag AS fraud_score
FROM claim_fraud_signals
WHERE high_frequency_flag + high_amount_flag + new_policy_flag + multi_provider_flag + inactive_policy_flag >= 2
WITH (connector = 'kafka', properties.bootstrap.server = 'kafka:9092', topic = 'fraud-review-queue')
FORMAT PLAIN ENCODE JSON;

The device_fingerprint and ip_address columns on the claims source let you reuse the account takeover device-history pattern: many claimants submitting from one device fingerprint is a ring, not a coincidence.

Trading: Intraday Exposure, Limits, and Concentration

The same architecture serves institutional risk. Positions stream from the booking system, prices from market data, and limits sit in a reference table. End-of-day risk becomes continuous risk.

CREATE SOURCE trading_positions (
    position_id       VARCHAR,
    instrument_id     VARCHAR,
    counterparty_id   VARCHAR,
    desk_id           VARCHAR,
    long_quantity     DECIMAL(18,6),
    short_quantity    DECIMAL(18,6),
    avg_cost          DECIMAL(18,6),
    currency          VARCHAR(3),
    asset_class       VARCHAR,     -- 'EQUITY', 'FX', 'RATES', 'CREDIT', 'COMMODITY'
    position_time     TIMESTAMPTZ
) WITH (
    connector = 'kafka',
    topic = 'trading-positions',
    properties.bootstrap.server = 'kafka:9092',
    scan.startup.mode = 'latest'
) FORMAT PLAIN ENCODE JSON;

CREATE SOURCE risk_factor_updates (
    instrument_id  VARCHAR,
    factor_type    VARCHAR,       -- 'PRICE', 'VOLATILITY', 'CREDIT_SPREAD', 'FX_RATE'
    value          DECIMAL(18,8),
    update_time    TIMESTAMPTZ
) WITH (
    connector = 'kafka',
    topic = 'risk-factors',
    properties.bootstrap.server = 'kafka:9092',
    scan.startup.mode = 'latest'
) FORMAT PLAIN ENCODE JSON;

CREATE TABLE risk_limits (
    entity_id     VARCHAR,
    limit_type    VARCHAR,        -- 'NOTIONAL', 'PNL_STOP_LOSS', 'CONCENTRATION'
    limit_value   DECIMAL(18,2),
    is_active     BOOLEAN,
    PRIMARY KEY (entity_id, limit_type)
);

CREATE MATERIALIZED VIEW desk_notional_exposure AS
SELECT desk_id, asset_class, currency,
       SUM((long_quantity - short_quantity) * avg_cost)   AS net_notional,
       SUM(long_quantity * avg_cost)                      AS gross_long,
       SUM(short_quantity * avg_cost)                     AS gross_short,
       COUNT(DISTINCT counterparty_id)                    AS counterparty_count,
       MAX(position_time)                                 AS as_of_time
FROM trading_positions
GROUP BY desk_id, asset_class, currency;

CREATE MATERIALIZED VIEW counterparty_exposure AS
SELECT counterparty_id, asset_class,
       SUM(ABS(long_quantity - short_quantity) * avg_cost) AS gross_exposure,
       COUNT(DISTINCT desk_id)                             AS desk_count,
       MAX(position_time)                                  AS as_of_time
FROM trading_positions
GROUP BY counterparty_id, asset_class;

CREATE MATERIALIZED VIEW limit_breach_monitor AS
SELECT dne.desk_id, dne.asset_class, dne.net_notional, rl.limit_value AS notional_limit,
       dne.net_notional / NULLIF(rl.limit_value, 0)   AS utilization_pct,
       ABS(dne.net_notional) > rl.limit_value         AS is_breached,
       dne.as_of_time
FROM desk_notional_exposure dne
JOIN risk_limits rl ON dne.desk_id = rl.entity_id AND rl.limit_type = 'NOTIONAL' AND rl.is_active;

CREATE MATERIALIZED VIEW concentration_risk AS
SELECT ce.counterparty_id, ce.asset_class, ce.gross_exposure, te.total_gross,
       ce.gross_exposure / NULLIF(te.total_gross, 0)         AS concentration_pct,
       ce.gross_exposure / NULLIF(te.total_gross, 0) > 0.10  AS exceeds_10pct_threshold
FROM counterparty_exposure ce
JOIN (SELECT asset_class, SUM(gross_exposure) AS total_gross FROM counterparty_exposure GROUP BY asset_class) te
  ON ce.asset_class = te.asset_class;

CREATE MATERIALIZED VIEW intraday_pnl_monitor AS
SELECT tp.desk_id,
       SUM((rf.value - tp.avg_cost) * (tp.long_quantity - tp.short_quantity))  AS unrealized_pnl,
       rl.limit_value                                                          AS stop_loss_limit,
       SUM((rf.value - tp.avg_cost) * (tp.long_quantity - tp.short_quantity)) < -ABS(rl.limit_value) AS stop_loss_breached
FROM trading_positions tp
JOIN risk_factor_updates rf ON tp.instrument_id = rf.instrument_id AND rf.factor_type = 'PRICE'
LEFT JOIN risk_limits rl ON tp.desk_id = rl.entity_id AND rl.limit_type = 'PNL_STOP_LOSS'
GROUP BY tp.desk_id, rl.limit_value;

CREATE SINK risk_alerts_feed AS
SELECT desk_id AS entity_id, 'DESK' AS entity_type, asset_class, net_notional AS current_exposure,
       notional_limit, utilization_pct, 'LIMIT_BREACH' AS alert_type, as_of_time
FROM limit_breach_monitor
WHERE is_breached
WITH (connector = 'kafka', properties.bootstrap.server = 'kafka:9092', topic = 'risk-alerts')
FORMAT PLAIN ENCODE JSON;

In production, risk_factor_updates should be reduced to the latest price per instrument before the P&L join, and the join should be a temporal join so each position is marked at the price current when it was booked or at the latest tick, depending on the desk's convention.

Serving Risk Decisions

Every domain ends the same way. The decision service runs a parameterized point lookup against the composite view for the entity in question, which returns in single-digit milliseconds because nothing is aggregated at query time, then applies a model or a rule set. Alerts and review items flow to Kafka sinks for step-up authentication, adjuster queues, and risk operations. For high query rates, a short-TTL cache in front of RisingWave handles the hot keys while the views remain the authoritative, always-fresh source.

-- Account takeover, at login
SELECT total_risk_score, velocity_score, device_novelty_score, mutation_score, travel_score
FROM ato_risk_score WHERE user_id = $1;

-- Marketplace, at booking request
SELECT s.seller_cancels_30d, s.avg_rating, v.bookings_last_1h, b.buyer_cancels_30d, b.disputes_90d,
       p.total_interactions, p.completed_together
FROM seller_reliability s
JOIN seller_velocity_1h v USING (seller_id)
CROSS JOIN buyer_behavior b
LEFT JOIN buyer_seller_history p ON p.seller_id = s.seller_id AND p.buyer_id = b.buyer_id
WHERE s.seller_id = $1 AND b.buyer_id = $2;

Batch vs Streaming Risk Scoring

Frequently Asked Questions

How do I calibrate the composite weights? Start with the illustrative weights, log every score alongside the eventual outcome, and fit a logistic regression or gradient boosted model on the labeled set. Keep the features as the materialized views; only the weights change.

Can one RisingWave cluster serve several risk domains? Yes. Each domain is a set of sources and views. Shared reference data such as device history can be reused across domains, which is how claims fraud can borrow the account takeover device pattern.

How are false positives fed back? Publish analyst decisions to a Kafka topic, ingest it as a source, and join it into the composite views as a suppression or down-weighting signal. The same topic is the training set for the next calibration.

What happens when a reference table changes, such as a new credit limit or a revised risk limit? CDC-fed tables propagate the change to every dependent view within seconds. Temporal joins with FOR SYSTEM_TIME AS OF preserve the historical value for past windows while new windows use the new value.

Is a one-hour window right for every velocity signal? No. Login velocity and seller bookings use one hour, post-login actions use five minutes, cancellations use 30 days, and disputes use 90. Choose the window from how fast the behavior unfolds, and expose several horizons as separate columns when the model benefits from both.