FEATURES_SQL = '''
WITH parameters AS (
SELECT MAX(transaction_date) AS analysis_end_date
FROM {transactions}
),
transaction_features AS (
SELECT
account_id,
COUNT(*) AS transaction_count,
COUNT(DISTINCT transaction_month) AS active_months,
SUM(amount) FILTER (WHERE transaction_direction = 'inflow') AS total_inflow,
AVG(amount) FILTER (WHERE transaction_direction = 'inflow') AS average_inflow,
SUM(amount) FILTER (WHERE transaction_direction = 'outflow') AS total_outflow,
AVG(amount) FILTER (WHERE transaction_direction = 'outflow') AS average_outflow,
AVG(balance) AS average_balance,
MEDIAN(balance) AS median_balance,
STDDEV_SAMP(balance) AS balance_std,
MIN(balance) AS minimum_balance,
MAX(balance) AS maximum_balance,
AVG((balance < 0)::INTEGER) AS negative_balance_share,
COUNT(DISTINCT operation) AS operation_diversity,
COUNT(DISTINCT transaction_purpose) FILTER (
WHERE transaction_purpose <> 'unspecified'
) AS purpose_diversity,
COUNT(DISTINCT COALESCE(operation, 'not_recorded') || ':' || transaction_purpose) AS transaction_diversity,
AVG((operation = 'cash_deposit')::INTEGER) AS cash_deposit_share,
AVG((operation IN ('cash_withdrawal', 'card_withdrawal'))::INTEGER) AS cash_withdrawal_share,
AVG((operation = 'incoming_transfer')::INTEGER) AS incoming_transfer_share,
MAX((transaction_purpose = 'pension')::INTEGER) > 0 AS receives_pension,
MAX((transaction_purpose = 'household_payment')::INTEGER) > 0 AS has_household_transaction,
MAX((transaction_purpose = 'insurance')::INTEGER) > 0 AS has_insurance_transaction
FROM {transactions}
GROUP BY account_id
),
latest_balance AS (
SELECT account_id, balance AS latest_balance
FROM {transactions}
QUALIFY ROW_NUMBER() OVER (
PARTITION BY account_id ORDER BY transaction_date DESC, trans_id DESC
) = 1
),
recurring_income_sources AS (
SELECT
account_id,
counterparty_bank,
counterparty_account,
COUNT(DISTINCT transaction_month) AS active_months,
AVG(amount) AS mean_amount,
STDDEV_POP(amount) / NULLIF(AVG(amount), 0) AS amount_cv
FROM {transactions}
WHERE transaction_direction = 'inflow'
AND operation = 'incoming_transfer'
AND transaction_purpose = 'unspecified'
AND counterparty_account IS NOT NULL
GROUP BY account_id, counterparty_bank, counterparty_account
),
salary_like_accounts AS (
SELECT DISTINCT account_id
FROM recurring_income_sources
WHERE active_months >= 3
AND mean_amount >= 10000
AND amount_cv <= 0.25
),
order_features AS (
SELECT
account_id,
MAX((TRIM(k_symbol) = 'SIPO')::INTEGER) > 0 AS has_household_order,
MAX((TRIM(k_symbol) = 'POJISTNE')::INTEGER) > 0 AS has_insurance_order,
MAX((TRIM(k_symbol) = 'LEASING')::INTEGER) > 0 AS makes_leasing_payments
FROM "order"
GROUP BY account_id
),
combined AS (
SELECT
base.account_id,
DATE_DIFF('month', DATE_TRUNC('month', base.account_open_date), DATE_TRUNC('month', p.analysis_end_date)) + 1 AS observed_months,
DATE_DIFF('day', base.account_open_date, p.analysis_end_date) / 30.4375 AS account_tenure_months,
tf.* EXCLUDE (
account_id,
receives_pension,
has_household_transaction,
has_insurance_transaction
),
lb.latest_balance,
base.card_count,
base.loan_count,
base.standing_order_count,
base.has_card,
base.has_loan,
base.has_standing_order,
salary.account_id IS NOT NULL AS receives_salary_like_deposit,
tf.receives_pension,
tf.has_household_transaction OR COALESCE(of.has_household_order, FALSE) AS pays_household_expenses,
tf.has_insurance_transaction OR COALESCE(of.has_insurance_order, FALSE) AS pays_insurance,
COALESCE(of.makes_leasing_payments, FALSE) AS makes_leasing_payments
FROM {account_base} AS base
CROSS JOIN parameters AS p
JOIN transaction_features AS tf USING (account_id)
JOIN latest_balance AS lb USING (account_id)
LEFT JOIN salary_like_accounts AS salary USING (account_id)
LEFT JOIN order_features AS of USING (account_id)
)
SELECT
account_id,
observed_months,
account_tenure_months,
transaction_count,
active_months,
active_months::DOUBLE / observed_months AS active_month_ratio,
transaction_count::DOUBLE / observed_months AS transactions_per_observed_month,
transaction_count::DOUBLE / active_months AS transactions_per_active_month,
total_inflow,
average_inflow,
total_outflow,
average_outflow,
total_inflow / NULLIF(total_outflow, 0) AS inflow_to_outflow_ratio,
average_balance,
median_balance,
balance_std,
minimum_balance,
maximum_balance,
latest_balance,
maximum_balance - minimum_balance AS balance_range,
negative_balance_share,
operation_diversity,
purpose_diversity,
transaction_diversity,
cash_deposit_share,
cash_withdrawal_share,
incoming_transfer_share,
card_count,
loan_count,
standing_order_count,
has_card,
has_loan,
has_standing_order,
receives_salary_like_deposit,
receives_pension,
pays_household_expenses,
pays_insurance,
makes_leasing_payments,
has_card::INTEGER
+ has_loan::INTEGER
+ has_standing_order::INTEGER
+ receives_salary_like_deposit::INTEGER
+ receives_pension::INTEGER
+ pays_household_expenses::INTEGER
+ pays_insurance::INTEGER
+ makes_leasing_payments::INTEGER AS service_diversity
FROM combined
ORDER BY account_id
'''.format(transactions=transactions_scan, account_base=account_base_scan)
customer_features = connection.execute(FEATURES_SQL).fetchdf()
display(customer_features.head())
