src/idsc/Prophet.py

import requests
import json


class Prophet():
    def __init__(self, apikey):
        self.apikey = apikey

    def parse_res(self, res):
        if str(res.status_code)[0] != '2':
            raise ValueError('API Call Failed!\n', res.text)
        else:
            parsed_res = res.json()
            # parsed_res['prediction_result'] = json.loads(parsed_res['prediction_result'])
            return parsed_res

    def continuous(self, ts: str, n_predict: int, inter_order_cpi=None):
        '''
        ts: JSON string that contains "SKU", "date" and "target"
        '''
        endpoint = 'https://idsc.com.sg/foretell/prediction/time-series/continuous/prophet'
        payloads = {
            'time_series_table': ts,
            'num_predict': n_predict}
        
        # print('[Prophet] - continuous', ts)

        if inter_order_cpi is not None:
            payloads['inter_order_cpi'] = inter_order_cpi

        headers = {'api-key': self.apikey}

        res = requests.post(endpoint, json=payloads, headers=headers)
        return self.parse_res(res)

    def intermittent(self, ts: str, n_predict: int, inter_order_cpi=None):
        endpoint = 'https://idsc.com.sg/foretell/prediction/time-series/intermittent/prophet'
        payloads = {
            'time_series_table': ts,
            'num_predict': n_predict}

        if inter_order_cpi is not None:
            payloads['inter_order_cpi'] = inter_order_cpi

        headers = {'api-key': self.apikey}

        res = requests.post(endpoint, json=payloads, headers=headers)
        return self.parse_res(res)