chronulus.estimator.binary_predictor

BinaryPredictor

Bases: Estimator

A prediction agent that estimates the probability binary events / outcomes.

This class handles the creation, queuing, and retrieval of binary event predictions and explanatory notes.

Parameters:

Name	Type	Description	Default
session	Session	Active session instance for API communication.	required
input_type	Type[BaseModelSubclass]	Pydantic model class that defines the expected input data structure.	required

Attributes:

Name	Type	Description
estimator_name	str	Name identifier for the estimator. Set to "BinaryPredictor".
estimator_version	str	Version string for the estimator. Set to "1".
prediction_version	str	Version string for the prediction. Set to "1".
estimator_id	str or None	Unique identifier assigned by the API after creation.

Source code in src/chronulus/estimator/binary_predictor.py

class BinaryPredictor(Estimator):
    """
   A prediction agent that estimates the probability binary events / outcomes.

   This class handles the creation, queuing, and retrieval of binary event predictions and explanatory notes.

   Parameters
   ----------
   session : Session
       Active session instance for API communication.
   input_type : Type[BaseModelSubclass]
       Pydantic model class that defines the expected input data structure.

   Attributes
   ----------
   estimator_name : str
       Name identifier for the estimator. Set to "BinaryPredictor".
   estimator_version : str
       Version string for the estimator. Set to "1".
   prediction_version : str
       Version string for the prediction. Set to "1".
   estimator_id : str or None
       Unique identifier assigned by the API after creation.

   """

    estimator_name = "BinaryPredictor"
    estimator_version = "1"
    prediction_version = "1"

    def __init__(self, session: Session, input_type: Type[BaseModelSubclass], estimator_id: Optional[str] = None, verbose: bool = True):
        super().__init__(session, input_type)
        self.verbose = verbose
        self.estimator_id = estimator_id
        if self.estimator_id is None:
            self.create()

    def create(self):
        """
        Initialize the agent instance with the API.

        Creates a agent instance on the API side with the specified input schema.
        The schema is serialized before transmission.

        Raises
        ------
        ValueError
            If the API fails to create the estimator or returns an invalid response.
        """

        request_data = EstimatorCreationRequest(
            estimator_name=self.estimator_name,
            session_id=self.session.session_id,
            input_model_info=InputModelInfo(
                validation_schema=self.input_type.model_json_schema(mode="validation"),
                serialization_schema=self.input_type.model_json_schema(mode="serialization"),
            )
        )

        resp = requests.post(
            url=f"{self.session.env.API_URI}/estimators/{self.get_route_prefix()}/create",
            headers=self.session.headers,
            json=request_data.model_dump()
        )
        if resp.status_code == 200:
            response_json = resp.json()
            if 'estimator_id' in response_json:
                self.estimator_id = response_json['estimator_id']
                if self.verbose:
                    print(f"Estimator created with estimator_id: {response_json['estimator_id']}")
            else:
                if self.verbose:
                    print(resp.status_code)
                    print(resp.text)
                raise ValueError("There was an error creating the estimator. Please try again.")
        else:
            raise ConnectionError(f"There was an error creating the estimator. Status code: {resp.status_code}. Response: {resp.text}")

    def queue(
            self,
            item: BaseModelSubclass,
            num_experts: int = 2,
            note_length: Tuple[int, int] = (3, 5),
    ):
        """
        Queue a prediction request for processing.

        Parameters
        ----------
        item : BaseModelSubclass
            The input data conforming to the specified input_type schema.
        num_experts : int, optional
            Number of experts to consult for the prediction request. (minimum = 2, default=2)
        note_length : tuple[int, int], optional
            Desired length range (number of sentences) for explanatory notes (min, max), by default (3, 5).

        Returns
        -------
        QueuePredictionResponse
            Response object containing the request status and ID.

        Raises
        ------
        TypeError
            If the provided item doesn't match the expected input_type.
        """

        if not 2 <= num_experts <= 30:
            raise ValueError("num_experts must be between 2 and 30")

        if not (isinstance(item, self.input_type) or are_models_equivalent(item, self.input_type)):
            try:
                assert item.model_json_schema(mode='validation') == self.input_type.model_json_schema(mode='validation')
                assert item.model_json_schema(mode='serialization') == self.input_type.model_json_schema(mode='serialization')

            except Exception as e:
                raise TypeError(f"Expect item to be an instance of {self.input_type}, but item has type {type(item)}")

        data = dict(
            estimator_id=self.estimator_id,
            item_data=item.model_dump(),
            num_experts=num_experts,
            note_length=note_length,
        )

        max_size_mb = 35
        data_mb = get_object_size_mb(data)
        if 3.0 < data_mb < max_size_mb :

            get_url_resp = requests.get(
                url=f'{self.session.env.API_URI}/uploads/get-upload-url',
                headers=self.session.headers
            )

            get_url_resp_json = get_url_resp.json()

            # Compress the JSON string
            compressed_data = gzip.compress(json.dumps(data).encode('utf-8'))

            upload_headers = {'Content-Type': 'application/json', 'Content-Encoding': 'gzip'}

            upload_response = requests.put(
                get_url_resp_json.get('url'),
                data=compressed_data,
                headers=upload_headers
            )

            resp = requests.post(
                url=f"{self.session.env.API_URI}/estimators/{self.get_route_prefix()}/queue-predict",
                headers=self.session.headers,
                json=dict(upload_id=get_url_resp_json.get('upload_id','')),
            )

        elif data_mb >= max_size_mb:
            return QueuePredictionResponse(
                success=False,
                request_id='',
                message=f'Queuing failed. Input size ({data_mb:5.2f} MB) exceeds {max_size_mb} MB.',
            )
        else:

            resp = requests.post(
                url=f"{self.session.env.API_URI}/estimators/{self.get_route_prefix()}/queue-predict",
                headers=self.session.headers,
                json=data,
            )

        if resp.status_code == 200:
            queue_response = QueuePredictionResponse(**resp.json())
            if self.verbose:
                print(f"Prediction queued successfully with request_id: {queue_response.request_id}")
            return queue_response
        else:
            return QueuePredictionResponse(
                success=False,
                request_id='',
                message=f'Queuing failed with status code {resp.status_code}: {resp.text}',
            )

    def get_request_predictions(self, request_id: str, try_every: int = 3, max_tries: int = 20) -> Union[BinaryPredictionSet, dict, None]:
        """
        Retrieve predictions for a queued request.

        Parameters
        ----------
        request_id : str
            The ID of the queued prediction request.
        try_every : int, optional
            Seconds to wait between retry attempts, by default 3.
        max_tries : int, optional
            Maximum number of retry attempts, by default 20.

        Returns
        -------
        Union[BinaryPredictionSet, dict, None]
            A BinaryPredictionSet containing predictions and explanations from each expert

        Raises
        ------
        Exception
            If the maximum retry limit is exceeded or if an API error occurs.
        """
        return self.get_request_predictions_static(
            request_id=request_id,
            try_every=try_every,
            max_tries=max_tries,
            env=self.session.env,
            verbose=self.verbose)



    @staticmethod
    def get_request_predictions_static(
            request_id: str,
            try_every: int = 3,
            max_tries: int = 20,
            env: Optional[dict] = None,
            verbose: bool = True) -> Union[BinaryPredictionSet, dict, None]:
        """
        Retrieve predictions for a queued request.

        Parameters
        ----------
        request_id : str
            The ID of the queued prediction request.
        try_every : int, optional
            Seconds to wait between retry attempts, by default 3.
        max_tries : int, optional
            Maximum number of retry attempts, by default 20.
        env : dict, optional
            Environment configuration dictionary. If None, default environment will be used.
        verbose : bool, optional
            Print feedback to stdout if True. Default: True

        Returns
        -------
        Union[BinaryPredictionSet, dict, None]
            A BinaryPredictionSet containing predictions and explanations from each expert

        Raises
        ------
        Exception
            If the maximum retry limit is exceeded or if an API error occurs.
        """

        prediction_version = BinaryPredictor.prediction_version
        if isinstance(env, Env):
            base = BaseEnv(**env.model_dump())
        else:
            env = env if env and isinstance(env, dict) else {}
            base = BaseEnv(**env)

        retries = 0

        while retries < max_tries:

            resp = requests.post(
                url=f"{base.env.API_URI}/predictions/{prediction_version}/check-by-request-id",
                headers=base.headers,
                json=dict(request_id=request_id),
            )

            if resp.status_code != 200:
                if verbose:
                    print(resp)
                raise Exception(f"An error occurred. Status code: {resp.status_code}. Response: {resp.text}")

            else:
                response_json = resp.json()

                if response_json['status'] == 'ERROR':
                    return response_json

                if response_json['status'] == 'SUCCESS':
                    if verbose:
                        print(f'{response_json["status"]} - {response_json["message"]} - Fetching predictions.')
                    prediction_ids = response_json.get('prediction_ids', [])
                    return BinaryPredictor.get_predictions_static(prediction_ids, base.env)

                if response_json['status'] in ['PENDING', 'NOT_FOUND', 'QUEUING']:
                    if verbose:
                        print(f'{response_json["status"]} - {response_json["message"]} - Checking again in {try_every} seconds...')
                    time.sleep(try_every)

                retries += 1

        if retries >= max_tries:
            raise Exception(f"Retry limit exceeded max_tries of {max_tries}")

    @staticmethod
    def get_predictions_static(prediction_ids: List[str], env: Optional[dict] = None, verbose: bool = True) -> Optional[BinaryPredictionSet]:
        """
        Static method to retrieve a batch of predictions with prediction_ids.

        Parameters
        ----------
        prediction_ids : List[str]
            A list of prediction ids
        env : dict, optional
            Environment configuration dictionary. If None, default environment will be used.
        verbose : bool, optional
            Print feedback to stdout if True. Default: True

        Returns
        -------
        BinaryPredictionSet or None
            A BinaryPredictionSet containing predictions and explanations from each expert,
            None if the predictions couldn't be retrieved.
        """
        prediction_version = BinaryPredictor.prediction_version
        if isinstance(env, Env):
            base = BaseEnv(**env.model_dump())
        else:
            env = env if env and isinstance(env, dict) else {}
            base = BaseEnv(**env)

        resp = requests.post(
            url=f"{base.env.API_URI}/predictions/{prediction_version}/get-by-prediction-id-batch",
            headers=base.headers,
            json=dict(prediction_ids=prediction_ids),
        )

        if resp.status_code == 200:
            response_json = resp.json()
            pred_response = PredictionBatchGetByIdResponse(**response_json)
            if pred_response.success:
                estimator_responses = pred_response.responses

                prediction = BinaryPredictionSet(
                    predictions=[BinaryPrediction(_id=pid, opinion_set=BinaryPair(**r)) for pid, r in zip(prediction_ids, estimator_responses)],
                    beta_params=BetaParams(**pred_response.meta.get('beta_params')),
                )
                return prediction
            else:
                if verbose:
                    print(f"The prediction could not be retrieved for batch of predictions ids")
                    print(pred_response.message)

        else:
            if verbose:
                print(f"The prediction could not be retrieved for batch of predictions ids")
                print(resp.status_code)
                print(resp.text)

    @staticmethod
    def load_from_saved_estimator(estimator_id: str, env: Optional[dict] = None, verbose: bool = True):

        # get estimator from estimator id
        env = env if env and isinstance(env, dict) else {}
        base = BaseEnv(**env)

        resp = requests.post(
            url=f"{base.env.API_URI}/estimators/{BinaryPredictor.estimator_version}/from-estimator-id",
            headers=base.headers,
            json=dict(estimator_id=estimator_id)
        )

        if resp.status_code != 200:
            raise ValueError(f"Failed to load estimator with status code: {resp.status_code}, {resp.text}")

        try:
            estimator_response = EstimatorGetResponse(**resp.json())
            if estimator_response.success:
                session = Session(**estimator_response.session.model_dump(), env=base.env.model_dump(), verbose=verbose)
                input_type = create_model_from_schema(estimator_response.input_model_info.validation_schema)
                estimator = BinaryPredictor(
                    session=session,
                    input_type=input_type,
                    estimator_id=estimator_response.estimator_id,
                    verbose=verbose
                )

                if verbose:
                    print(f"Estimator retrieved with estimator_id: {estimator_response.estimator_id}")

                return estimator

            else:
                raise ValueError(estimator_response.message)

        except Exception as e:
            raise ValueError(f"Failed to parse estimator response: {resp.text}")

create()

Initialize the agent instance with the API.

Creates a agent instance on the API side with the specified input schema. The schema is serialized before transmission.

Raises:

Type	Description
ValueError	If the API fails to create the estimator or returns an invalid response.

Source code in src/chronulus/estimator/binary_predictor.py

def create(self):
    """
    Initialize the agent instance with the API.

    Creates a agent instance on the API side with the specified input schema.
    The schema is serialized before transmission.

    Raises
    ------
    ValueError
        If the API fails to create the estimator or returns an invalid response.
    """

    request_data = EstimatorCreationRequest(
        estimator_name=self.estimator_name,
        session_id=self.session.session_id,
        input_model_info=InputModelInfo(
            validation_schema=self.input_type.model_json_schema(mode="validation"),
            serialization_schema=self.input_type.model_json_schema(mode="serialization"),
        )
    )

    resp = requests.post(
        url=f"{self.session.env.API_URI}/estimators/{self.get_route_prefix()}/create",
        headers=self.session.headers,
        json=request_data.model_dump()
    )
    if resp.status_code == 200:
        response_json = resp.json()
        if 'estimator_id' in response_json:
            self.estimator_id = response_json['estimator_id']
            if self.verbose:
                print(f"Estimator created with estimator_id: {response_json['estimator_id']}")
        else:
            if self.verbose:
                print(resp.status_code)
                print(resp.text)
            raise ValueError("There was an error creating the estimator. Please try again.")
    else:
        raise ConnectionError(f"There was an error creating the estimator. Status code: {resp.status_code}. Response: {resp.text}")

get_predictions_static(prediction_ids, env=None, verbose=True) staticmethod

Static method to retrieve a batch of predictions with prediction_ids.

Parameters:

Name	Type	Description	Default
prediction_ids	List[str]	A list of prediction ids	required
env	dict	Environment configuration dictionary. If None, default environment will be used.	None
verbose	bool	Print feedback to stdout if True. Default: True	True

Returns:

Type	Description
BinaryPredictionSet or None	A BinaryPredictionSet containing predictions and explanations from each expert, None if the predictions couldn't be retrieved.

Source code in src/chronulus/estimator/binary_predictor.py

@staticmethod
def get_predictions_static(prediction_ids: List[str], env: Optional[dict] = None, verbose: bool = True) -> Optional[BinaryPredictionSet]:
    """
    Static method to retrieve a batch of predictions with prediction_ids.

    Parameters
    ----------
    prediction_ids : List[str]
        A list of prediction ids
    env : dict, optional
        Environment configuration dictionary. If None, default environment will be used.
    verbose : bool, optional
        Print feedback to stdout if True. Default: True

    Returns
    -------
    BinaryPredictionSet or None
        A BinaryPredictionSet containing predictions and explanations from each expert,
        None if the predictions couldn't be retrieved.
    """
    prediction_version = BinaryPredictor.prediction_version
    if isinstance(env, Env):
        base = BaseEnv(**env.model_dump())
    else:
        env = env if env and isinstance(env, dict) else {}
        base = BaseEnv(**env)

    resp = requests.post(
        url=f"{base.env.API_URI}/predictions/{prediction_version}/get-by-prediction-id-batch",
        headers=base.headers,
        json=dict(prediction_ids=prediction_ids),
    )

    if resp.status_code == 200:
        response_json = resp.json()
        pred_response = PredictionBatchGetByIdResponse(**response_json)
        if pred_response.success:
            estimator_responses = pred_response.responses

            prediction = BinaryPredictionSet(
                predictions=[BinaryPrediction(_id=pid, opinion_set=BinaryPair(**r)) for pid, r in zip(prediction_ids, estimator_responses)],
                beta_params=BetaParams(**pred_response.meta.get('beta_params')),
            )
            return prediction
        else:
            if verbose:
                print(f"The prediction could not be retrieved for batch of predictions ids")
                print(pred_response.message)

    else:
        if verbose:
            print(f"The prediction could not be retrieved for batch of predictions ids")
            print(resp.status_code)
            print(resp.text)

get_request_predictions(request_id, try_every=3, max_tries=20)

Retrieve predictions for a queued request.

Parameters:

Name	Type	Description	Default
request_id	str	The ID of the queued prediction request.	required
try_every	int	Seconds to wait between retry attempts, by default 3.	3
max_tries	int	Maximum number of retry attempts, by default 20.	20

Returns:

Type	Description
Union[BinaryPredictionSet, dict, None]	A BinaryPredictionSet containing predictions and explanations from each expert

Raises:

Type	Description
Exception	If the maximum retry limit is exceeded or if an API error occurs.

Source code in src/chronulus/estimator/binary_predictor.py

def get_request_predictions(self, request_id: str, try_every: int = 3, max_tries: int = 20) -> Union[BinaryPredictionSet, dict, None]:
    """
    Retrieve predictions for a queued request.

    Parameters
    ----------
    request_id : str
        The ID of the queued prediction request.
    try_every : int, optional
        Seconds to wait between retry attempts, by default 3.
    max_tries : int, optional
        Maximum number of retry attempts, by default 20.

    Returns
    -------
    Union[BinaryPredictionSet, dict, None]
        A BinaryPredictionSet containing predictions and explanations from each expert

    Raises
    ------
    Exception
        If the maximum retry limit is exceeded or if an API error occurs.
    """
    return self.get_request_predictions_static(
        request_id=request_id,
        try_every=try_every,
        max_tries=max_tries,
        env=self.session.env,
        verbose=self.verbose)

get_request_predictions_static(request_id, try_every=3, max_tries=20, env=None, verbose=True) staticmethod

Retrieve predictions for a queued request.

Parameters:

Name	Type	Description	Default
request_id	str	The ID of the queued prediction request.	required
try_every	int	Seconds to wait between retry attempts, by default 3.	3
max_tries	int	Maximum number of retry attempts, by default 20.	20
env	dict	Environment configuration dictionary. If None, default environment will be used.	None
verbose	bool	Print feedback to stdout if True. Default: True	True

Returns:

Type	Description
Union[BinaryPredictionSet, dict, None]	A BinaryPredictionSet containing predictions and explanations from each expert

Raises:

Type	Description
Exception	If the maximum retry limit is exceeded or if an API error occurs.

Source code in src/chronulus/estimator/binary_predictor.py

@staticmethod
def get_request_predictions_static(
        request_id: str,
        try_every: int = 3,
        max_tries: int = 20,
        env: Optional[dict] = None,
        verbose: bool = True) -> Union[BinaryPredictionSet, dict, None]:
    """
    Retrieve predictions for a queued request.

    Parameters
    ----------
    request_id : str
        The ID of the queued prediction request.
    try_every : int, optional
        Seconds to wait between retry attempts, by default 3.
    max_tries : int, optional
        Maximum number of retry attempts, by default 20.
    env : dict, optional
        Environment configuration dictionary. If None, default environment will be used.
    verbose : bool, optional
        Print feedback to stdout if True. Default: True

    Returns
    -------
    Union[BinaryPredictionSet, dict, None]
        A BinaryPredictionSet containing predictions and explanations from each expert

    Raises
    ------
    Exception
        If the maximum retry limit is exceeded or if an API error occurs.
    """

    prediction_version = BinaryPredictor.prediction_version
    if isinstance(env, Env):
        base = BaseEnv(**env.model_dump())
    else:
        env = env if env and isinstance(env, dict) else {}
        base = BaseEnv(**env)

    retries = 0

    while retries < max_tries:

        resp = requests.post(
            url=f"{base.env.API_URI}/predictions/{prediction_version}/check-by-request-id",
            headers=base.headers,
            json=dict(request_id=request_id),
        )

        if resp.status_code != 200:
            if verbose:
                print(resp)
            raise Exception(f"An error occurred. Status code: {resp.status_code}. Response: {resp.text}")

        else:
            response_json = resp.json()

            if response_json['status'] == 'ERROR':
                return response_json

            if response_json['status'] == 'SUCCESS':
                if verbose:
                    print(f'{response_json["status"]} - {response_json["message"]} - Fetching predictions.')
                prediction_ids = response_json.get('prediction_ids', [])
                return BinaryPredictor.get_predictions_static(prediction_ids, base.env)

            if response_json['status'] in ['PENDING', 'NOT_FOUND', 'QUEUING']:
                if verbose:
                    print(f'{response_json["status"]} - {response_json["message"]} - Checking again in {try_every} seconds...')
                time.sleep(try_every)

            retries += 1

    if retries >= max_tries:
        raise Exception(f"Retry limit exceeded max_tries of {max_tries}")

queue(item, num_experts=2, note_length=(3, 5))

Queue a prediction request for processing.

Parameters:

Name	Type	Description	Default
item	BaseModelSubclass	The input data conforming to the specified input_type schema.	required
num_experts	int	Number of experts to consult for the prediction request. (minimum = 2, default=2)	2
note_length	tuple[int, int]	Desired length range (number of sentences) for explanatory notes (min, max), by default (3, 5).	(3, 5)

Returns:

Type	Description
QueuePredictionResponse	Response object containing the request status and ID.

Raises:

Type	Description
TypeError	If the provided item doesn't match the expected input_type.

Source code in src/chronulus/estimator/binary_predictor.py

def queue(
        self,
        item: BaseModelSubclass,
        num_experts: int = 2,
        note_length: Tuple[int, int] = (3, 5),
):
    """
    Queue a prediction request for processing.

    Parameters
    ----------
    item : BaseModelSubclass
        The input data conforming to the specified input_type schema.
    num_experts : int, optional
        Number of experts to consult for the prediction request. (minimum = 2, default=2)
    note_length : tuple[int, int], optional
        Desired length range (number of sentences) for explanatory notes (min, max), by default (3, 5).

    Returns
    -------
    QueuePredictionResponse
        Response object containing the request status and ID.

    Raises
    ------
    TypeError
        If the provided item doesn't match the expected input_type.
    """

    if not 2 <= num_experts <= 30:
        raise ValueError("num_experts must be between 2 and 30")

    if not (isinstance(item, self.input_type) or are_models_equivalent(item, self.input_type)):
        try:
            assert item.model_json_schema(mode='validation') == self.input_type.model_json_schema(mode='validation')
            assert item.model_json_schema(mode='serialization') == self.input_type.model_json_schema(mode='serialization')

        except Exception as e:
            raise TypeError(f"Expect item to be an instance of {self.input_type}, but item has type {type(item)}")

    data = dict(
        estimator_id=self.estimator_id,
        item_data=item.model_dump(),
        num_experts=num_experts,
        note_length=note_length,
    )

    max_size_mb = 35
    data_mb = get_object_size_mb(data)
    if 3.0 < data_mb < max_size_mb :

        get_url_resp = requests.get(
            url=f'{self.session.env.API_URI}/uploads/get-upload-url',
            headers=self.session.headers
        )

        get_url_resp_json = get_url_resp.json()

        # Compress the JSON string
        compressed_data = gzip.compress(json.dumps(data).encode('utf-8'))

        upload_headers = {'Content-Type': 'application/json', 'Content-Encoding': 'gzip'}

        upload_response = requests.put(
            get_url_resp_json.get('url'),
            data=compressed_data,
            headers=upload_headers
        )

        resp = requests.post(
            url=f"{self.session.env.API_URI}/estimators/{self.get_route_prefix()}/queue-predict",
            headers=self.session.headers,
            json=dict(upload_id=get_url_resp_json.get('upload_id','')),
        )

    elif data_mb >= max_size_mb:
        return QueuePredictionResponse(
            success=False,
            request_id='',
            message=f'Queuing failed. Input size ({data_mb:5.2f} MB) exceeds {max_size_mb} MB.',
        )
    else:

        resp = requests.post(
            url=f"{self.session.env.API_URI}/estimators/{self.get_route_prefix()}/queue-predict",
            headers=self.session.headers,
            json=data,
        )

    if resp.status_code == 200:
        queue_response = QueuePredictionResponse(**resp.json())
        if self.verbose:
            print(f"Prediction queued successfully with request_id: {queue_response.request_id}")
        return queue_response
    else:
        return QueuePredictionResponse(
            success=False,
            request_id='',
            message=f'Queuing failed with status code {resp.status_code}: {resp.text}',
        )