def create_predictor_handler()

in cdk/cdk/afc_lambdas/create_predictor.py [0:0]

• 45 lines of code
• 1 McCabe index (conditional complexity)

def create_predictor_handler(event, context):
    """
    """

    payload = event["input"]["Payload"]
    prefix = payload["prefix"]

    update_status_json(payload, "IN_PROGRESS:create_predictor",
        payload["StatusJsonS3Path"])

    AFC_DATASET_GROUP_ARN = payload["DatasetGroupArn"]
    AFC_FORECAST_HORIZON = payload["horiz"] + 1
    AFC_FORECAST_FREQUENCY = payload["freq"]
    #AFC_ALGORITHM_NAME = "NPTS"
    #AFC_ALGORITHM_ARN = "arn:aws:forecast:::algorithm/NPTS"
    AFC_PREDICTOR_NAME = f"{prefix}_AutoML"

    create_predictor_resp = afc.create_predictor(
        PredictorName=AFC_PREDICTOR_NAME,
        ForecastHorizon=AFC_FORECAST_HORIZON,
        #AlgorithmArn=AFC_ALGORITHM_ARN, # TODO: delete this when ready
        PerformAutoML=True, # TODO: Uncomment this when ready
        #PerformHPO=False,
        EvaluationParameters={
            "NumberOfBacktestWindows": 5
        },
        InputDataConfig={
            "DatasetGroupArn": AFC_DATASET_GROUP_ARN
        },
        FeaturizationConfig={
            "ForecastFrequency": AFC_FORECAST_FREQUENCY,
            "Featurizations": [
                {
                    "AttributeName": "demand",
                    "FeaturizationPipeline": [
                        {
                            "FeaturizationMethodName": "filling",
                            "FeaturizationMethodParameters": {
                                "aggregation": "sum",
                                "frontfill": "none",
                                "middlefill": "zero",
                                "backfill": "zero"
                            }
                        }
                    ]
                }
            ]
        }
    )

    resp = payload
    resp["PredictorArn"] = create_predictor_resp["PredictorArn"]
    resp["PredictorName"] = AFC_PREDICTOR_NAME

    update_status_json(resp, "DONE:create_predictor",
        payload["StatusJsonS3Path"])

    return resp