#### Fill in Definition
- **Source Tables** * - one or more tables linked to the entity. Each chosen table becomes a variable in the editor: a table aliased `account` is referenced as `account` in code. - **Source Columns** * - the specific columns from those tables that your aggregation reads. Each one is exposed inside the corresponding table variable. - **Input** - any reusable global function registered on the platform that the aggregation should call. - **Input Type** - pick the language used for the aggregation (Python, Pandas, or Spark), then write the logic in the editor below.
```python
import datetime
# sum charge_off_amount for records within 1 year of `account_application_date`
application_date = account['account_application_date'][0]
total_charge_off_amount = 0
for date, co_amt in zip(
account_performance['record_date'],
account_performance['charge_off_amount'],
):
if (
date >= application_date
and date - application_date < datetime.timedelta(days=365)
and co_amt
):
total_charge_off_amount += co_amt
# default flag if total_charge_off_amount > 0
if not total_charge_off_amount:
return 0
elif total_charge_off_amount > 0:
return 1
else:
return 0
```
**Example 3: Total deposit amount in the 90 days before application**
*Inputs:*
- **Source Tables**: `application_table`, `application_cashflow_transactions`
- **Source Columns**: `application_table.loan_applied_date`, `application_cashflow_transactions.transaction_date`, `.amount`, `.transaction_type`
The Entity here is Application, so both dictionaries are sliced to one application's rows. The platform hands your code the data like this:
```python
application_table = {
'loan_applied_date': ['2025-04-01'],
}
application_cashflow_transactions = {
'transaction_date': ['2025-01-15', '2025-02-20', '2025-03-10'],
'amount': [200, 500, 300],
'transaction_type': ['CREDIT', 'DEBIT', 'CREDIT'],
}
```
Then your aggregation runs on top of it:
```python
import datetime
loan_applied_date = application_table["loan_applied_date"][0]
window_start = loan_applied_date - datetime.timedelta(days=90)
total_deposits = 0
for txn_date, amt, txn_type in zip(
application_cashflow_transactions["transaction_date"],
application_cashflow_transactions["amount"],
application_cashflow_transactions["transaction_type"],
):
if (
txn_type == "CREDIT"
and amt is not None
and window_start <= txn_date.date() < loan_applied_date
):
total_deposits += amt
return total_deposits
```
```python
import datetime
# get the account records with `record_date` within 1 year of `account_application_date`
application_date = account['account_application_date'].iat[0]
account_perf_within_1_year = account_performance[
(application_date < account_performance['record_date']) &
(account_performance['record_date'] - application_date < datetime.timedelta(days=365))
]
# drop the records that don't have a charge-off amount
account_perf_with_co_amt = account_perf_within_1_year.dropna(subset=['charge_off_amount'])
if account_perf_with_co_amt.empty:
return 0
if account_perf_with_co_amt['charge_off_amount'].sum() > 0:
return 1
else:
return 0
```
**Example 3: Total deposit amount in the 90 days before application**
*Inputs:*
- **Source Tables**: `application_table`, `application_cashflow_transactions`
- **Source Columns**: `application_table.loan_applied_date`, `application_cashflow_transactions.transaction_date`, `.amount`, `.transaction_type`
Entity is Application, so both DataFrames are sliced to one application's rows. The platform hands your code DataFrames that look like this:
```text
# `application_table`:
# loan_applied_date
# 0 2025-04-01
# `application_cashflow_transactions`:
# transaction_date amount transaction_type
# 0 2025-01-15 200 credit
# 1 2025-02-20 500 debit
# 2 2025-03-10 300 credit
```
Then your aggregation runs on top of it:
```python
import pandas as pd
loan_applied_date = pd.Timestamp(application_table['loan_applied_date'].iat[0])
window_start = loan_applied_date - pd.Timedelta(days=90)
txns = application_cashflow_transactions
deposits_in_window = txns[
(txns['transaction_type'] == 'credit')
& (txns['transaction_date'] >= window_start)
& (txns['transaction_date'] < loan_applied_date)
]
if deposits_in_window.empty:
return 0
return deposits_in_window['amount'].sum()
```
```python
import pyspark.sql.functions as F
from pyspark.sql.functions import udf
from pyspark.sql.types import IntegerType
acc_perf = account_performance.select(
'account_id', 'record_date', 'COAMT'
).join(
account.select('account_id', 'application_date'),
on='account_id', how='left',
)
CO1yr_df = acc_perf.filter(
acc_perf.record_date < F.date_add(acc_perf['application_date'], 365)
)
COAMT_df = (
CO1yr_df.groupBy('account_id')
.agg(F.sum('COAMT'))
.toDF('account_id', 'COAMT')
)
def charge_off(x):
return 1 if x > 0 else 0
charge_off_flag_udf = udf(lambda s: charge_off(s), IntegerType())
COAMT_df = COAMT_df.withColumn(
'charge_off_flag', charge_off_flag_udf('COAMT')
)
return COAMT_df.select('account_id', 'charge_off_flag')
```
**Example 3: Total deposit amount in the 90 days before application**
*Inputs:*
- **Source Tables**: `application_table`, `application_cashflow_transactions`
- **Source Columns**: `application_table.application_id`, `application_table.loan_applied_date`, `application_cashflow_transactions.application_id`, `.transaction_date`, `.amount`, `.transaction_type`
Full unfiltered tables across all applications. The platform hands your code DataFrames that look like this:
```text
# `application_table`:
+--------------+-----------------+
|application_id|loan_applied_date|
+--------------+-----------------+
| A001| 2025-04-01|
| A002| 2025-04-05|
+--------------+-----------------+
# `application_cashflow_transactions`:
+--------------+----------------+------+----------------+
|application_id|transaction_date|amount|transaction_type|
+--------------+----------------+------+----------------+
| A001| 2025-01-15| 200| CREDIT|
| A001| 2025-02-20| 500| DEBIT|
| A001| 2025-03-10| 300| CREDIT|
| A002| 2025-02-25| 400| CREDIT|
+--------------+----------------+------+----------------+
```
Then your aggregation runs on top of it. Join on `application_id`, filter, then aggregate back to one row per application:
```python
import pyspark.sql.functions as F
txns = application_cashflow_transactions.join(
application_table.select('application_id', 'loan_applied_date'),
on='application_id', how='left',
)
deposits_in_window = txns.filter(
(F.col('transaction_type') == 'CREDIT')
& (F.to_date('transaction_date') >= F.date_sub(F.col('loan_applied_date'), 90))
& (F.to_date('transaction_date') < F.col('loan_applied_date'))
)
return deposits_in_window.groupBy('application_id').agg(
F.sum('amount').alias('total_deposits_last_90d')
)
```
:::note[Set the Output Dataframe Key]
Spark adds an extra **Output Dataframe Key** field below the editor. Enter the name of the entity key column in your returned DataFrame (for example, `customer_id`, `account_id`, or `application_id`). The platform uses this to join the aggregation result back to the parent entity.
:::
