AURA

def rf_cash_structuring_below_threshold(df, account_id, window_days=30):
    """Count cash deposits in $8,000-$9,999 range within window."""
    mask = (
        (df['txn_type'] == 'CR') &
        (df['is_cash'] == True) &
        (df['txn_amt'] >= 8000) &
        (df['txn_amt'] < 10000)
    )
    acct = df[df['account_id'] == account_id]
    acct_filtered = acct[mask]
    acct_filtered = acct_filtered.sort_values('txn_date')
    count = acct_filtered.rolling(
        f'{window_days}D', on='txn_date'
    )['txn_amt'].count().max()
    return count if pd.notna(count) else 0

def rf_multi_branch_rapid(df, account_id, window_days=7):
    """Count distinct branches used within rolling window."""
    acct = df[df['account_id'] == account_id].sort_values('txn_date')
    if len(acct) == 0:
        return 0
    branches_per_window = acct.set_index('txn_date').rolling(
        f'{window_days}D'
    )['branch_id'].apply(lambda x: x.nunique())
    return branches_per_window.max()

def rf_rapid_outflow_high_risk(df, account_id, hours=48):
    """Detect outbound to FATF high-risk within 48h of cash deposit."""
    HIGH_RISK = {'IR', 'KP', 'MM', 'PK', 'SY', 'YE'}
    acct = df[df['account_id'] == account_id].sort_values('txn_date')
    deposits = acct[(acct['is_cash']) & (acct['txn_type'] == 'CR')]
    outbound = acct[acct['country_code'].isin(HIGH_RISK)]
    count = 0
    for _, dep in deposits.iterrows():
        window = outbound[
            (outbound['txn_date'] > dep['txn_date']) &
            (outbound['txn_date'] <= dep['txn_date'] + pd.Timedelta(hours=hours))
        ]
        count += len(window)
    return count

def rf_cash_volume_anomaly(df, account_id, window_days=90):
    """Z-score of cash volume vs peer group mean."""
    cash = df[df['is_cash'] == True]
    peer_vol = cash.groupby('account_id')['txn_amt'].sum()
    acct_vol = peer_vol.get(account_id, 0)
    if peer_vol.std() == 0:
        return 0
    z = (acct_vol - peer_vol.mean()) / peer_vol.std()
    return max(z, 0)