def lookup_features()

in src/lambda/InvokeFraudEndpointLambda/lambda_function.py [0:0]

• 37 lines of code
• 4 McCabe index (conditional complexity)

def lookup_features(credit_card_number, transaction_amount):
    combined_agg_features = {}

    # Initialize agg feature values
    cc_num = '1111111111111111'
    avg_amt_last_10m = 0
    num_trans_last_10m = 0
    avg_amt_last_1w = 0
    num_trans_last_1w = 0

    # Retrieve first set of aggregated features
    cc_agg_feature_group = CC_AGG_FEATURE_GROUP
    feature_names = ['cc_num', 'avg_amt_last_10m', 'num_trans_last_10m', 'trans_time']
    agg_features = retrieve_aggregated_features(cc_agg_feature_group, credit_card_number, feature_names)

    agg_evt_time = 0.0

    if agg_features is not None:
        cc_num = agg_features['cc_num']
        avg_amt_last_10m = float(agg_features['avg_amt_last_10m'])
        num_trans_last_10m = int(agg_features['num_trans_last_10m'])
        agg_evt_time = float(agg_features['trans_time'])
        # store agg features in dict
        combined_agg_features.update(agg_features)

    logging.debug(f'In> lookup_features: #1 Agg Features: \n {combined_agg_features}')

    # Check cutoff condition before calling second Feature Group
    cutoff_condition = eval_cutoff_window(agg_evt_time)

    # Retrieve second set of aggregated features
    batch_agg_features = None
    if not cutoff_condition:
        # Retrieve data from 'batch' agg feature group
        cc_agg_batch_feature_group = CC_AGG_BATCH_FEATURE_GROUP
        feature_names = ['cc_num', 'num_trans_last_1w', 'avg_amt_last_1w']
        batch_agg_features = retrieve_aggregated_features(cc_agg_batch_feature_group, credit_card_number, feature_names)

    if batch_agg_features is not None:
        num_trans_last_1w = int(batch_agg_features['num_trans_last_1w'])
        avg_amt_last_1w = float(batch_agg_features['avg_amt_last_1w'])
        # store batch agg features in dict
        combined_agg_features.update(batch_agg_features)

    logging.debug(f'In> lookup_features: #2 Agg Features: \n {combined_agg_features}')

    amt_ratio1, amt_ratio2, count_ratio = calc_ratios_for_inference(
        transaction_amount, avg_amt_last_10m, avg_amt_last_1w,
        num_trans_last_10m, num_trans_last_1w, cutoff_condition)

    # Append ratios into dict
    combined_agg_features['amt_ratio1'] = amt_ratio1
    combined_agg_features['amt_ratio2'] = amt_ratio2
    combined_agg_features['count_ratio'] = count_ratio

    logging.debug(f'In> lookup_features: #3 Agg Features: \n {combined_agg_features}')
    return combined_agg_features, cutoff_condition