Spaces:
Runtime error
Runtime error
File size: 3,459 Bytes
7cee5a0 cf739bf 7cee5a0 cf739bf 7cee5a0 cf739bf 7cee5a0 cf739bf 7cee5a0 cf739bf 3a44431 cf739bf 3a44431 cf739bf 7cee5a0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 | """
Structuring Detector - Fixed
- Scans both source and target
- Rolling window via pandas
- Configurable thresholds from config.yaml
- No Streamlit
"""
import uuid
import pandas as pd
from datetime import datetime
from src.config_loader import get_config
def detect_structuring(df: pd.DataFrame) -> list[dict]:
cfg = get_config()['detectors']
THRESHOLD = cfg['structuring_threshold']
LOWER_PCT = cfg['structuring_lower_pct']
WINDOW_DAYS = cfg['structuring_window_days']
MIN_TX_COUNT = cfg['structuring_min_tx_count']
CHECK_TARGET = cfg.get('structuring_check_target', True)
lower = THRESHOLD * LOWER_PCT
upper = THRESHOLD
window = f"{WINDOW_DAYS}D"
fraud_accounts = (
set(df[df['is_laundering'] == 1]['source'].unique()) |
set(df[df['is_laundering'] == 1]['target'].unique())
)
df = df.copy()
df['timestamp'] = pd.to_datetime(df['timestamp'])
# Filter to transactions in the structuring band
band_df = df[df['amount'].between(lower, upper, inclusive='left')].copy()
if band_df.empty:
return []
alerts = []
seen = set()
def _scan_column(col: str, band: pd.DataFrame) -> None:
for account in band[col].unique():
if account in seen:
continue
acct = band[band[col] == account].set_index('timestamp').sort_index()
if acct.empty:
continue
# Rolling window count + sum
rolled_count = acct['amount'].rolling(window).count()
rolled_sum = acct['amount'].rolling(window).sum()
triggered = (rolled_sum >= THRESHOLD) & (rolled_count >= MIN_TX_COUNT)
if not triggered.any():
continue
seen.add(account)
best_idx = triggered[triggered].index[-1]
# Use .loc then .iloc[-1] to handle duplicate timestamps safely
total_val = rolled_sum.loc[best_idx]
total = float(total_val.iloc[-1]) if hasattr(total_val, 'iloc') else float(total_val)
count_val = rolled_count.loc[best_idx]
count = int(count_val.iloc[-1]) if hasattr(count_val, 'iloc') else int(count_val)
risk_score = min(
int((total / THRESHOLD) * 40 + (count / MIN_TX_COUNT) * 30 + 20),
99
)
# Velocity acceleration: tx in last 24h vs full window
last_24h = acct['amount'].rolling('1D').count()
last_24h_val = last_24h.iloc[-1]
accel = float(last_24h_val) / max(count, 1)
risk_score = min(int(risk_score + accel * 10), 99)
explanation = (
f"{count} transactions totalling {total:,.0f} detected within "
f"{WINDOW_DAYS} days, clustering just below the "
f"reporting threshold of {THRESHOLD:,}"
)
alerts.append({
'alert_id': uuid.uuid4().hex[:8],
'account': account,
'typology': 'Structuring',
'risk_score': risk_score,
'amount_involved': round(total, 2),
'tx_count': count,
'explanation': explanation,
'timestamp_detected': datetime.now().isoformat(),
'confirmed_fraud': account in fraud_accounts,
})
_scan_column('source', band_df)
if CHECK_TARGET:
_scan_column('target', band_df)
return alerts
|