commit 9fbc75ddca21dedc3a1bc256a9f87ac19d4ed221
parent a8c97f04c501eed04d2ffeb6497270c8d996d8eb
Author: Florian Dold <dold@taler.net>
Date: Mon, 7 Sep 2026 12:37:13 +0200
exchange TOPS: track deposit anomaly episodes without renewing KYC
Count gross batch increments and monitor open accounts immediately.
Require the last 28 days to exceed CHF 100,000 and the preceding
336 days, comparing amounts at full fractional precision.
Keep episode state independent of officer decisions so cleared cases
reopen only after the condition clears and is crossed again. Preserve
KYC expiration and MROS reporting states, and append the anomaly reason
only once. Update the undeployed customization in place.
Issue: https://bugs.taler.net/n/9639
Diffstat:
1 file changed, 161 insertions(+), 123 deletions(-)
diff --git a/src/exchangedb/sql-schema/tops-0001.sql b/src/exchangedb/sql-schema/tops-0001.sql
@@ -80,154 +80,192 @@ VALUES
)
ON CONFLICT DO NOTHING;
+-- One record per monitored account, independent of officer decisions. An
+-- officer clearing a case must not cause the same anomaly to reopen it.
+CREATE TABLE deposit_monitor_state
+ (h_payto BYTEA PRIMARY KEY REFERENCES exchange.kyc_targets(h_normalized_payto)
+ ON DELETE CASCADE
+ ,open_time INT8 NOT NULL
+ ,clears_at INT8 -- UTC seconds; NULL if the volume condition is false
+ );
+COMMENT ON TABLE deposit_monitor_state
+ IS 'tracks deposit anomaly episodes independently of KYC rule validity';
+
+-- When would the current anomaly first clear if no further deposits arrived?
+-- Only expiration from the 28-day window can clear it. Expiration from the
+-- 52-week window can make it true again, so remember the FIRST clearing time,
+-- even if that whole false interval falls between two deposits. Events and
+-- interval statistics have the same daily precision and inclusive lower bound.
+CREATE FUNCTION deposit_anomaly_clears_at(in_h_payto BYTEA, in_now INT8)
+RETURNS INT8
+LANGUAGE SQL
+VOLATILE
+SET search_path TO exchange, pg_temp
+AS $$
+ WITH events AS MATERIALIZED (
+ SELECT e.slot, (e.delta).val::NUMERIC * 100000000 + (e.delta).frac AS units
+ FROM exchange_statistic_amount_event e
+ JOIN exchange_statistic_interval_meta m USING (imeta_serial_id)
+ WHERE m.slug = 'deposit-volume'
+ AND m.stype = 'amount'
+ AND e.h_payto = in_h_payto
+ AND e.slot >= in_now - 31449600
+ ), boundaries AS (
+ SELECT slot + 2419200 + 1 AS at_time
+ FROM events
+ WHERE slot >= in_now - 2419200
+ )
+ SELECT MIN(b.at_time)
+ FROM boundaries b
+ CROSS JOIN LATERAL (
+ SELECT COALESCE(SUM(units) FILTER (WHERE slot >= b.at_time - 2419200), 0) AS month,
+ COALESCE(SUM(units) FILTER (WHERE slot >= b.at_time - 31449600), 0) AS year
+ FROM events
+ ) totals
+ WHERE totals.month <= 100000::NUMERIC * 100000000
+ OR totals.year >= 2 * totals.month;
+$$;
+
DROP FUNCTION IF EXISTS tops_deposit_statistics_trigger CASCADE;
CREATE FUNCTION tops_deposit_statistics_trigger()
RETURNS trigger
LANGUAGE plpgsql
+SET search_path TO exchange, pg_temp
AS $$
DECLARE
- my_h_payto BYTEA; -- normalized h_payto of target account
+ my_h_payto BYTEA;
+ my_delta taler_amount;
my_rec RECORD;
- my_last_year taler_amount; -- sum of deposits this year
- my_last_month taler_amount; -- sum of deposits this month
+ my_month NUMERIC := 0;
+ my_year NUMERIC := 0;
my_old_rules RECORD;
- my_properties TEXT;
- my_investigate_property JSONB;
- my_measure_name TEXT;
- my_rules TEXT;
+ my_properties JSONB;
+ my_reason TEXT;
+ my_state TEXT;
my_now INT8;
+ my_seconds INT8;
+ my_open_time INT8;
+ my_close_time INT8;
+ my_previous_open INT8;
+ my_previous_clear INT8;
+ my_anomaly BOOL;
+ my_new_episode BOOL;
BEGIN
- -- legitimization_outcomes.decision_time and .expiration_time are stored in
- -- microseconds (that is what GNUNET_PQ_query_param_timestamp() writes), so
- -- the durations below have to be scaled accordingly.
- my_now = ROUND(EXTRACT(epoch FROM CURRENT_TIMESTAMP(0)::TIMESTAMP))::INT8
- * 1000000;
- SELECT wt.h_normalized_payto
- INTO my_h_payto
- FROM wire_targets wt
- WHERE wire_target_h_payto = NEW.wire_target_h_payto;
+ IF TG_OP = 'UPDATE'
+ THEN
+ -- Retries and unrelated updates contribute nothing. Gross deposits only
+ -- grow; refunds are recorded separately and do not reduce this statistic.
+ IF NEW.total_amount <= OLD.total_amount
+ THEN
+ RETURN NEW;
+ END IF;
+ SELECT (diff).* INTO my_delta
+ FROM amount_left_minus_right(NEW.total_amount, OLD.total_amount);
+ ELSE
+ my_delta = NEW.total_amount;
+ END IF;
+
+ my_seconds = EXTRACT(epoch FROM exchange_now())::INT8;
+ my_now = my_seconds * 1000000;
+ SELECT h_normalized_payto INTO STRICT my_h_payto
+ FROM wire_targets WHERE wire_target_h_payto = NEW.wire_target_h_payto;
+
+ -- Serialize monitoring for this account, also with account open/close
+ -- updates. The exchange runs these operations in retryable transactions.
+ SELECT open_time, close_time INTO my_open_time, my_close_time
+ FROM kyc_targets WHERE h_normalized_payto = my_h_payto FOR UPDATE;
CALL exchange_do_bump_amount_stat
- ('deposit-volume'
- ,my_h_payto
- ,CURRENT_TIMESTAMP(0)::TIMESTAMP
- ,NEW.total_amount);
-
--- FIXME: this is just for testing, I want to also check
--- the 'counter'-based functions.
- CALL exchange_do_bump_number_stat
- ('deposit-transactions'
- ,my_h_payto
- ,CURRENT_TIMESTAMP(0)::TIMESTAMP
- ,1);
-
- -- Get historical deposit volumes and extract the yearly and monthly
- -- interval statistic values from the result for the AML trigger check.
+ ('deposit-volume', my_h_payto, exchange_now(), my_delta);
+ IF TG_OP = 'INSERT'
+ THEN
+ -- Count batches once, even when more coins arrive in a later request.
+ CALL exchange_do_bump_number_stat
+ ('deposit-transactions', my_h_payto, exchange_now(), 1);
+ END IF;
+
+ -- Keep statistics for all deposits, but monitor only open bank accounts.
+ IF my_open_time IS NULL OR my_close_time IS NOT NULL
+ THEN
+ DELETE FROM tops.deposit_monitor_state WHERE h_payto = my_h_payto;
+ RETURN NEW;
+ END IF;
+
FOR my_rec IN
- SELECT *
- FROM exchange_statistic_interval_amount_get(
- 'deposit-volume'
- ,my_h_payto
- )
+ SELECT * FROM exchange_statistic_interval_amount_get('deposit-volume', my_h_payto)
LOOP
- IF (my_rec.range = 60*60*24*7*52)
+ IF my_rec.range = 2419200
THEN
- my_last_year = my_rec.rvalue;
- END IF;
- IF (my_rec.range = 60*60*24*7*4)
+ my_month = (my_rec.rvalue).val::NUMERIC * 100000000 + (my_rec.rvalue).frac;
+ ELSIF my_rec.range = 31449600
THEN
- my_last_month = my_rec.rvalue;
+ my_year = (my_rec.rvalue).val::NUMERIC * 100000000 + (my_rec.rvalue).frac;
END IF;
END LOOP;
- -- Note: it is OK to ignore '.frac', as that cannot be significant.
- -- Also, we effectively exclude the current month's revenue from
- -- "last year" as otherwise the rule makes no sense.
- -- Finally, we define the "current month" always as the last 4 weeks,
- -- just like the "last year" is the last 52 weeks.
- IF (my_last_year.val < my_last_month.val * 2)
+ -- Latest 28 days exceed both CHF 100,000 and the preceding 336 days.
+ -- Amounts in the exchange DB have no currency tag; TOPS operates in CHF.
+ -- Compare all fractional units, including at the strict CHF 100,000 floor.
+ my_anomaly = my_month > 100000::NUMERIC * 100000000 AND my_year < 2 * my_month;
+
+ SELECT open_time, clears_at INTO my_previous_open, my_previous_clear
+ FROM tops.deposit_monitor_state WHERE h_payto = my_h_payto;
+ my_new_episode = NOT FOUND OR my_previous_open IS DISTINCT FROM my_open_time
+ OR my_previous_clear IS NULL OR my_previous_clear <= my_seconds;
+ INSERT INTO tops.deposit_monitor_state (h_payto, open_time, clears_at)
+ VALUES (my_h_payto, my_open_time,
+ CASE WHEN my_anomaly THEN tops.deposit_anomaly_clears_at(my_h_payto, my_seconds) END)
+ ON CONFLICT (h_payto) DO UPDATE SET
+ open_time = EXCLUDED.open_time, clears_at = EXCLUDED.clears_at;
+ IF NOT my_anomaly
THEN
- -- This is suspicious. => Flag account for AML review!
- --
- -- FIXME: we probably want to factor the code from
- -- this branch out into a generic
- -- function to trigger investigations at some point!
- --
- -- First, get existing rules and clear an 'is_active'
- -- flag, but ONLY if we are not _already_ investigating
- -- the account (as in the latter case, we'll do no INSERT).
- UPDATE legitimization_outcomes
- SET is_active=NOT to_investigate
- WHERE h_payto = my_h_payto
- AND is_active
- RETURNING jproperties
- ,new_measure_name
- ,jnew_rules
- ,to_investigate
- INTO my_old_rules;
-
- -- Note that if we have no active legitimization_outcome
- -- that means we are on default rules and the account
- -- did not cross KYC thresholds and thus we have no
- -- established business relationship. In this case, we
- -- do not care as the overall volume is insignificant.
- -- This also takes care of the case where a customer
- -- is new (and obviously the first few months are
- -- basically always above the inherently zero or near-zero
- -- transactions from the previous year).
- -- Thus, we only proceed IF FOUND.
- IF FOUND
- THEN
- my_properties = my_old_rules.jproperties;
- my_measure_name = my_old_rules.new_measure_name;
- my_rules = my_old_rules.jnew_rules;
- my_investigate_property = json_object(ARRAY['AML_INVESTIGATION_STATE',
- 'AML_INVESTIGATION_TRIGGER'],
- ARRAY['INVESTIATION_PENDING',
- 'DEPOSIT_ANOMALY']);
- IF my_properties IS NULL
- THEN
- my_properties = my_investigate_property::TEXT;
- ELSE
- my_properties = (my_properties::JSONB || my_investigate_property)::TEXT;
- END IF;
-
- -- Note: here we could in theory manipulate my_properties,
- -- say to set a note as to why the investigation was started.
- IF NOT my_old_rules.to_investigate
- THEN
- -- Only insert if 'to_investigate' was not already set.
- INSERT INTO legitimization_outcomes (
- h_payto
- ,decision_time
- ,expiration_time
- ,jproperties
- ,new_measure_name
- ,to_investigate
- ,is_active
- ,jnew_rules
- ) VALUES (
- my_h_payto
- ,my_now
- ,my_now + 366*24*60*60*1000000::INT8
- ,my_properties::JSONB
- ,my_measure_name
- ,TRUE
- ,TRUE
- ,my_rules::JSONB);
- END IF;
- END IF;
+ RETURN NEW;
+ END IF;
+
+ -- An open account can have expired rules. Carry their deadline unchanged:
+ -- monitoring must neither renew KYC nor shorten a permanent approval.
+ SELECT * INTO my_old_rules
+ FROM legitimization_outcomes WHERE h_payto = my_h_payto AND is_active
+ ORDER BY outcome_serial_id DESC LIMIT 1 FOR UPDATE;
+ IF NOT my_new_episode AND NOT COALESCE(my_old_rules.to_investigate, FALSE)
+ THEN
+ RETURN NEW;
+ END IF;
+ my_properties = COALESCE(my_old_rules.jproperties::JSONB, '{}'::JSONB);
+ my_reason = my_properties ->> 'INVESTIGATION_TRIGGER';
+ IF NOT ('DEPOSIT_ANOMALY' = ANY(regexp_split_to_array(COALESCE(my_reason, ''), ';[[:space:]]*')))
+ THEN
+ my_properties = my_properties || jsonb_build_object('INVESTIGATION_TRIGGER',
+ CASE WHEN COALESCE(my_reason, '') = '' THEN 'DEPOSIT_ANOMALY'
+ ELSE my_reason || '; DEPOSIT_ANOMALY' END);
+ END IF;
+ my_state = my_properties ->> 'INVESTIGATION_STATE';
+ IF my_state IS NULL OR
+ (NOT COALESCE(my_old_rules.to_investigate, FALSE) AND
+ my_state NOT IN ('REPORTED_SUSPICION_SIMPLE', 'REPORTED_SUSPICION_SUBSTANTIATED'))
+ THEN
+ my_properties = my_properties || jsonb_build_object('INVESTIGATION_STATE', 'INVESTIGATION_PENDING');
+ END IF;
+ IF COALESCE(my_old_rules.to_investigate, FALSE) AND
+ my_properties = my_old_rules.jproperties::JSONB
+ THEN
+ RETURN NEW;
END IF;
+
+ UPDATE legitimization_outcomes SET is_active = FALSE
+ WHERE h_payto = my_h_payto AND is_active;
+ INSERT INTO legitimization_outcomes
+ (h_payto, decision_time, expiration_time, jproperties, new_measure_name,
+ to_investigate, is_active, jnew_rules)
+ VALUES
+ (my_h_payto, my_now, COALESCE(my_old_rules.expiration_time, 9223372036854775807),
+ my_properties, my_old_rules.new_measure_name, TRUE, TRUE, my_old_rules.jnew_rules);
RETURN NEW;
END $$;
COMMENT ON FUNCTION tops_deposit_statistics_trigger
- IS 'creates deposit statistics';
+ IS 'counts deposit increments and flags new TOPS deposit anomaly episodes';
--- Whenever a deposit is made, call our trigger to bump statistics
CREATE TRIGGER tops_batch_deposits_on_insert
- AFTER INSERT
- ON batch_deposits
+ AFTER INSERT OR UPDATE OF total_amount ON batch_deposits
FOR EACH ROW EXECUTE FUNCTION tops_deposit_statistics_trigger();
-
-
COMMIT;
