Python API Client

This page provides auto-generated documentation from the client library's docstrings.

crowdcent_challenge.client.ChallengeClient

Client for interacting with a specific CrowdCent Challenge.

Handles authentication and provides methods for accessing challenge data, training datasets, inference data, and managing prediction submissions for a specific challenge identified by its slug.

Source code in src/crowdcent_challenge/client.py

class ChallengeClient:
    """
    Client for interacting with a specific CrowdCent Challenge.

    Handles authentication and provides methods for accessing challenge data,
    training datasets, inference data, and managing prediction submissions for
    a specific challenge identified by its slug.
    """

    DEFAULT_BASE_URL = "http://crowdcent.com/api"
    API_KEY_ENV_VAR = "CROWDCENT_API_KEY"

    def __init__(
        self,
        challenge_slug: str,
        api_key: Optional[str] = None,
        base_url: Optional[str] = None,
    ):
        """
        Initializes the ChallengeClient for a specific challenge.

        Args:
            challenge_slug: The unique identifier (slug) for the challenge.
            api_key: Your CrowdCent API key. If not provided, it will attempt
                     to load from the CROWDCENT_API_KEY environment variable
                     or a .env file.
            base_url: The base URL of the CrowdCent API. Defaults to
                      http://crowdcent.com/api.
        """
        load_dotenv()  # Load .env file if present
        self.api_key = api_key or os.getenv(self.API_KEY_ENV_VAR)
        if not self.api_key:
            raise AuthenticationError(
                f"API key not provided and not found in environment variable "
                f"'{self.API_KEY_ENV_VAR}' or .env file."
            )

        self.challenge_slug = challenge_slug
        self.base_url = (base_url or self.DEFAULT_BASE_URL).rstrip("/")
        self.session = requests.Session()
        self.session.headers.update({"Authorization": f"Api-Key {self.api_key}"})
        logger.info(
            f"ChallengeClient initialized for '{challenge_slug}' at URL: {self.base_url}"
        )

    def _request(
        self,
        method: str,
        endpoint: str,
        params: Optional[Dict] = None,
        json_data: Optional[Dict] = None,
        files: Optional[Dict[str, IO]] = None,
        stream: bool = False,
    ) -> requests.Response:
        """
        Internal helper method to make authenticated API requests.

        Args:
            method: HTTP method (e.g., 'GET', 'POST').
            endpoint: API endpoint path (e.g., '/challenges/').
            params: URL parameters.
            json_data: JSON data for the request body.
            files: Files to upload (for multipart/form-data).
            stream: Whether to stream the response (for downloads).

        Returns:
            The requests.Response object.

        Raises:
            AuthenticationError: If the API key is invalid (401).
            NotFoundError: If the resource is not found (404).
            ClientError: For other 4xx errors.
            ServerError: For 5xx errors.
            CrowdCentAPIError: For other request exceptions.
        """
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        logger.debug(
            f"Request: {method} {url} Params: {params} JSON: {json_data is not None} Files: {files is not None}"
        )

        try:
            response = self.session.request(
                method, url, params=params, json=json_data, files=files, stream=stream
            )
            response.raise_for_status()  # Raise HTTPError for bad responses (4xx or 5xx)
            logger.debug(f"Response: {response.status_code}")
            return response
        except requests.exceptions.HTTPError as e:
            status_code = e.response.status_code

            # Try to parse standardized error format: {"error": {"code": "ERROR_CODE", "message": "Description"}}
            try:
                error_data = e.response.json()
                if "error" in error_data and isinstance(error_data["error"], dict):
                    error_code = error_data["error"].get("code", "UNKNOWN_ERROR")
                    error_message = error_data["error"].get("message", e.response.text)
                else:
                    error_code = "API_ERROR"
                    error_message = e.response.text
            except requests.exceptions.JSONDecodeError:
                error_code = "API_ERROR"
                error_message = e.response.text

            logger.error(
                f"API Error ({status_code}): {error_code} - {error_message} for {method} {url}"
            )

            if status_code == 401:
                raise AuthenticationError(
                    f"Authentication failed (401): {error_message} [{error_code}]"
                ) from e
            elif status_code == 404:
                raise NotFoundError(
                    f"Resource not found (404): {error_message} [{error_code}]"
                ) from e
            elif 400 <= status_code < 500:
                raise ClientError(
                    f"Client error ({status_code}): {error_message} [{error_code}]"
                ) from e
            elif 500 <= status_code < 600:
                raise ServerError(
                    f"Server error ({status_code}): {error_message} [{error_code}]"
                ) from e
            else:
                raise CrowdCentAPIError(
                    f"HTTP error ({status_code}): {error_message} [{error_code}]"
                ) from e
        except requests.exceptions.RequestException as e:
            logger.error(f"Request failed: {e} for {method} {url}")
            raise CrowdCentAPIError(f"Request failed: {e}") from e

    # --- Class Method for Listing All Challenges ---

    @classmethod
    def list_all_challenges(
        cls, api_key: Optional[str] = None, base_url: Optional[str] = None
    ) -> List[Dict[str, Any]]:
        """Lists all active challenges.

        This is a class method that doesn't require a challenge_slug.
        Use this to discover available challenges before initializing a ChallengeClient.

        Args:
            api_key: Your CrowdCent API key. If not provided, it will attempt
                     to load from the CROWDCENT_API_KEY environment variable
                     or a .env file.
            base_url: The base URL of the CrowdCent API. Defaults to
                      http://crowdcent.com/api.

        Returns:
            A list of dictionaries, each representing an active challenge.
        """
        # Create a temporary session for this request
        load_dotenv()
        api_key = api_key or os.getenv(cls.API_KEY_ENV_VAR)
        if not api_key:
            raise AuthenticationError(
                f"API key not provided and not found in environment variable "
                f"'{cls.API_KEY_ENV_VAR}' or .env file."
            )

        base_url = (base_url or cls.DEFAULT_BASE_URL).rstrip("/")
        session = requests.Session()
        session.headers.update({"Authorization": f"Api-Key {api_key}"})

        url = f"{base_url}/challenges/"
        try:
            response = session.get(url)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.HTTPError as e:
            status_code = e.response.status_code
            if status_code == 401:
                raise AuthenticationError("Authentication failed (401)")
            elif status_code == 404:
                raise NotFoundError("Resource not found (404)")
            elif 400 <= status_code < 500:
                raise ClientError(f"Client error ({status_code})")
            elif 500 <= status_code < 600:
                raise ServerError(f"Server error ({status_code})")
            else:
                raise CrowdCentAPIError(f"HTTP error ({status_code})")
        except requests.exceptions.RequestException as e:
            raise CrowdCentAPIError(f"Request failed: {e}")

    # --- Challenge Methods ---

    def get_challenge(self) -> Dict[str, Any]:
        """Gets details for this challenge.

        Returns:
            A dictionary representing this challenge.

        Raises:
            NotFoundError: If the challenge with the given slug is not found.
        """
        response = self._request("GET", f"/challenges/{self.challenge_slug}/")
        return response.json()

    # --- Training Data Methods ---

    def list_training_datasets(self) -> List[Dict[str, Any]]:
        """Lists all training dataset versions for this challenge.

        Returns:
            A list of dictionaries, each representing a training dataset version.

        Raises:
            NotFoundError: If the challenge is not found.
        """
        response = self._request(
            "GET", f"/challenges/{self.challenge_slug}/training_data/"
        )
        return response.json()

    def get_latest_training_dataset(self) -> Dict[str, Any]:
        """Gets the latest training dataset for this challenge.

        Returns:
            A dictionary representing the latest training dataset.

        Raises:
            NotFoundError: If the challenge or its latest training dataset is not found.
        """
        response = self._request(
            "GET", f"/challenges/{self.challenge_slug}/training_data/latest/"
        )
        return response.json()

    def get_training_dataset(self, version: str) -> Dict[str, Any]:
        """Gets details for a specific training dataset version.

        Args:
            version: The version string of the training dataset (e.g., '1.0', '2.1').

        Returns:
            A dictionary representing the specified training dataset.

        Raises:
            NotFoundError: If the challenge or the specified training dataset is not found.
        """
        response = self._request(
            "GET", f"/challenges/{self.challenge_slug}/training_data/{version}/"
        )
        return response.json()

    def download_training_dataset(self, version: str, dest_path: str):
        """Downloads the training data file for a specific dataset version.

        Args:
            version: The version string of the training dataset (e.g., '1.0', '2.1')
                    or 'latest' to get the latest version.
            dest_path: The local file path to save the downloaded dataset.

        Raises:
            NotFoundError: If the challenge, dataset, or its file is not found.
        """
        if version == "latest":
            endpoint = (
                f"/challenges/{self.challenge_slug}/training_data/latest/download/"
            )
        else:
            endpoint = (
                f"/challenges/{self.challenge_slug}/training_data/{version}/download/"
            )

        logger.info(
            f"Downloading training data for challenge '{self.challenge_slug}' v{version} to {dest_path}"
        )
        response = self._request("GET", endpoint, stream=True)

        try:
            with open(dest_path, "wb") as f:
                for chunk in response.iter_content(chunk_size=8192):
                    f.write(chunk)
            logger.info(f"Successfully downloaded training data to {dest_path}")
        except IOError as e:
            logger.error(f"Failed to write dataset to {dest_path}: {e}")
            raise CrowdCentAPIError(f"Failed to write dataset file: {e}") from e

    # --- Inference Data Methods ---

    def list_inference_data(self) -> List[Dict[str, Any]]:
        """Lists all inference data periods for this challenge.

        Returns:
            A list of dictionaries, each representing an inference data period.

        Raises:
            NotFoundError: If the challenge is not found.
        """
        response = self._request(
            "GET", f"/challenges/{self.challenge_slug}/inference_data/"
        )
        return response.json()

    def get_current_inference_data(self) -> Dict[str, Any]:
        """Gets the current inference data period for this challenge.

        Returns:
            A dictionary representing the current inference data period.

        Raises:
            NotFoundError: If the challenge has no active inference period.
        """
        response = self._request(
            "GET", f"/challenges/{self.challenge_slug}/inference_data/current/"
        )
        return response.json()

    def get_inference_data(self, release_date: str) -> Dict[str, Any]:
        """Gets details for a specific inference data period by its release date.

        Args:
            release_date: The release date of the inference data in 'YYYY-MM-DD' format.

        Returns:
            A dictionary representing the specified inference data period.

        Raises:
            NotFoundError: If the challenge or the specified inference data is not found.
            ClientError: If the date format is invalid.
        """
        # Validate date format
        try:
            datetime.strptime(release_date, "%Y-%m-%d")
        except ValueError:
            raise ClientError(
                f"Invalid date format: {release_date}. Use 'YYYY-MM-DD' format."
            )

        response = self._request(
            "GET", f"/challenges/{self.challenge_slug}/inference_data/{release_date}/"
        )
        return response.json()

    def download_inference_data(self, release_date: str, dest_path: str):
        """Downloads the inference features file for a specific period.

        Args:
            release_date: The release date of the inference data in 'YYYY-MM-DD' format
                         or 'current' to get the current period's data.
            dest_path: The local file path to save the downloaded features file.

        Raises:
            NotFoundError: If the challenge, inference data, or its file is not found.
            ClientError: If the date format is invalid.
        """
        if release_date == "current":
            endpoint = (
                f"/challenges/{self.challenge_slug}/inference_data/current/download/"
            )
        else:
            # Validate date format
            try:
                datetime.strptime(release_date, "%Y-%m-%d")
            except ValueError:
                raise ClientError(
                    f"Invalid date format: {release_date}. Use 'YYYY-MM-DD' format."
                )

            endpoint = f"/challenges/{self.challenge_slug}/inference_data/{release_date}/download/"

        logger.info(
            f"Downloading inference data for challenge '{self.challenge_slug}' {release_date} to {dest_path}"
        )
        response = self._request("GET", endpoint, stream=True)

        try:
            with open(dest_path, "wb") as f:
                for chunk in response.iter_content(chunk_size=8192):
                    f.write(chunk)
            logger.info(f"Successfully downloaded inference data to {dest_path}")
        except IOError as e:
            logger.error(f"Failed to write inference data to {dest_path}: {e}")
            raise CrowdCentAPIError(f"Failed to write inference data file: {e}") from e

    # --- Submission Methods ---

    def list_submissions(self, period: Optional[str] = None) -> List[Dict[str, Any]]:
        """Lists the authenticated user's submissions for this challenge.

        Args:
            period: Optional filter for submissions by period:
                  - 'current': Only show submissions for the current active period
                  - 'YYYY-MM-DD': Only show submissions for a specific inference period date

        Returns:
            A list of dictionaries, each representing a submission.
        """
        params = {}
        if period:
            params["period"] = period

        response = self._request(
            "GET", f"/challenges/{self.challenge_slug}/submissions/", params=params
        )
        return response.json()

    def get_submission(self, submission_id: int) -> Dict[str, Any]:
        """Gets details for a specific submission by its ID.

        Args:
            submission_id: The ID of the submission to retrieve.

        Returns:
            A dictionary representing the specified submission.

        Raises:
            NotFoundError: If the submission with the given ID is not found
                           or doesn't belong to the user.
        """
        response = self._request(
            "GET", f"/challenges/{self.challenge_slug}/submissions/{submission_id}/"
        )
        return response.json()

    def submit_predictions(self, file_path: str) -> Dict[str, Any]:
        """Submits a prediction file for the current active inference period of this challenge.

        The file must be a Parquet file with the required prediction columns:
        id, pred_1M, pred_3M, pred_6M, pred_9M, pred_12M

        Args:
            file_path: The path to the prediction Parquet file.

        Returns:
            A dictionary representing the newly created submission.

        Raises:
            FileNotFoundError: If the specified file_path does not exist.
            ClientError: If the submission is invalid (e.g., wrong format,
                         outside submission window, already submitted, etc).
        """
        logger.info(
            f"Submitting predictions from {file_path} to challenge '{self.challenge_slug}'"
        )
        try:
            with open(file_path, "rb") as f:
                files = {
                    "prediction_file": (
                        os.path.basename(file_path),
                        f,
                        "application/octet-stream",
                    )
                }
                response = self._request(
                    "POST",
                    f"/challenges/{self.challenge_slug}/submissions/",
                    files=files,
                )
            logger.info(
                f"Successfully submitted predictions to challenge '{self.challenge_slug}'"
            )
            return response.json()
        except FileNotFoundError as e:
            logger.error(f"Prediction file not found at {file_path}")
            raise FileNotFoundError(f"Prediction file not found at {file_path}") from e
        except IOError as e:
            logger.error(f"Failed to read prediction file {file_path}: {e}")
            raise CrowdCentAPIError(f"Failed to read prediction file: {e}") from e

    # --- Challenge Switching ---

    def switch_challenge(self, new_challenge_slug: str):
        """Switch this client to interact with a different challenge.

        Args:
            new_challenge_slug: The slug identifier for the new challenge.

        Returns:
            None. The client is modified in-place.
        """
        self.challenge_slug = new_challenge_slug
        logger.info(f"Client switched to challenge '{new_challenge_slug}'")

__init__(challenge_slug, api_key=None, base_url=None)

Initializes the ChallengeClient for a specific challenge.

Parameters:

Name	Type	Description	Default
challenge_slug	str	The unique identifier (slug) for the challenge.	required
api_key	Optional[str]	Your CrowdCent API key. If not provided, it will attempt to load from the CROWDCENT_API_KEY environment variable or a .env file.	None
base_url	Optional[str]	The base URL of the CrowdCent API. Defaults to http://crowdcent.com/api.	None

Source code in src/crowdcent_challenge/client.py

def __init__(
    self,
    challenge_slug: str,
    api_key: Optional[str] = None,
    base_url: Optional[str] = None,
):
    """
    Initializes the ChallengeClient for a specific challenge.

    Args:
        challenge_slug: The unique identifier (slug) for the challenge.
        api_key: Your CrowdCent API key. If not provided, it will attempt
                 to load from the CROWDCENT_API_KEY environment variable
                 or a .env file.
        base_url: The base URL of the CrowdCent API. Defaults to
                  http://crowdcent.com/api.
    """
    load_dotenv()  # Load .env file if present
    self.api_key = api_key or os.getenv(self.API_KEY_ENV_VAR)
    if not self.api_key:
        raise AuthenticationError(
            f"API key not provided and not found in environment variable "
            f"'{self.API_KEY_ENV_VAR}' or .env file."
        )

    self.challenge_slug = challenge_slug
    self.base_url = (base_url or self.DEFAULT_BASE_URL).rstrip("/")
    self.session = requests.Session()
    self.session.headers.update({"Authorization": f"Api-Key {self.api_key}"})
    logger.info(
        f"ChallengeClient initialized for '{challenge_slug}' at URL: {self.base_url}"
    )

list_all_challenges(api_key=None, base_url=None) classmethod

Lists all active challenges.

This is a class method that doesn't require a challenge_slug. Use this to discover available challenges before initializing a ChallengeClient.

Parameters:

Name	Type	Description	Default
api_key	Optional[str]	Your CrowdCent API key. If not provided, it will attempt to load from the CROWDCENT_API_KEY environment variable or a .env file.	None
base_url	Optional[str]	The base URL of the CrowdCent API. Defaults to http://crowdcent.com/api.	None

Returns:

Type	Description
List[Dict[str, Any]]	A list of dictionaries, each representing an active challenge.

Source code in src/crowdcent_challenge/client.py

@classmethod
def list_all_challenges(
    cls, api_key: Optional[str] = None, base_url: Optional[str] = None
) -> List[Dict[str, Any]]:
    """Lists all active challenges.

    This is a class method that doesn't require a challenge_slug.
    Use this to discover available challenges before initializing a ChallengeClient.

    Args:
        api_key: Your CrowdCent API key. If not provided, it will attempt
                 to load from the CROWDCENT_API_KEY environment variable
                 or a .env file.
        base_url: The base URL of the CrowdCent API. Defaults to
                  http://crowdcent.com/api.

    Returns:
        A list of dictionaries, each representing an active challenge.
    """
    # Create a temporary session for this request
    load_dotenv()
    api_key = api_key or os.getenv(cls.API_KEY_ENV_VAR)
    if not api_key:
        raise AuthenticationError(
            f"API key not provided and not found in environment variable "
            f"'{cls.API_KEY_ENV_VAR}' or .env file."
        )

    base_url = (base_url or cls.DEFAULT_BASE_URL).rstrip("/")
    session = requests.Session()
    session.headers.update({"Authorization": f"Api-Key {api_key}"})

    url = f"{base_url}/challenges/"
    try:
        response = session.get(url)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.HTTPError as e:
        status_code = e.response.status_code
        if status_code == 401:
            raise AuthenticationError("Authentication failed (401)")
        elif status_code == 404:
            raise NotFoundError("Resource not found (404)")
        elif 400 <= status_code < 500:
            raise ClientError(f"Client error ({status_code})")
        elif 500 <= status_code < 600:
            raise ServerError(f"Server error ({status_code})")
        else:
            raise CrowdCentAPIError(f"HTTP error ({status_code})")
    except requests.exceptions.RequestException as e:
        raise CrowdCentAPIError(f"Request failed: {e}")

get_challenge()

Gets details for this challenge.

Returns:

Type	Description
Dict[str, Any]	A dictionary representing this challenge.

Raises:

Type	Description
NotFoundError	If the challenge with the given slug is not found.

Source code in src/crowdcent_challenge/client.py

def get_challenge(self) -> Dict[str, Any]:
    """Gets details for this challenge.

    Returns:
        A dictionary representing this challenge.

    Raises:
        NotFoundError: If the challenge with the given slug is not found.
    """
    response = self._request("GET", f"/challenges/{self.challenge_slug}/")
    return response.json()

list_training_datasets()

Lists all training dataset versions for this challenge.

Returns:

Type	Description
List[Dict[str, Any]]	A list of dictionaries, each representing a training dataset version.

Raises:

Type	Description
NotFoundError	If the challenge is not found.

Source code in src/crowdcent_challenge/client.py

def list_training_datasets(self) -> List[Dict[str, Any]]:
    """Lists all training dataset versions for this challenge.

    Returns:
        A list of dictionaries, each representing a training dataset version.

    Raises:
        NotFoundError: If the challenge is not found.
    """
    response = self._request(
        "GET", f"/challenges/{self.challenge_slug}/training_data/"
    )
    return response.json()

get_latest_training_dataset()

Gets the latest training dataset for this challenge.

Returns:

Type	Description
Dict[str, Any]	A dictionary representing the latest training dataset.

Raises:

Type	Description
NotFoundError	If the challenge or its latest training dataset is not found.

Source code in src/crowdcent_challenge/client.py

def get_latest_training_dataset(self) -> Dict[str, Any]:
    """Gets the latest training dataset for this challenge.

    Returns:
        A dictionary representing the latest training dataset.

    Raises:
        NotFoundError: If the challenge or its latest training dataset is not found.
    """
    response = self._request(
        "GET", f"/challenges/{self.challenge_slug}/training_data/latest/"
    )
    return response.json()

get_training_dataset(version)

Gets details for a specific training dataset version.

Parameters:

Name	Type	Description	Default
version	str	The version string of the training dataset (e.g., '1.0', '2.1').	required

Returns:

Type	Description
Dict[str, Any]	A dictionary representing the specified training dataset.

Raises:

Type	Description
NotFoundError	If the challenge or the specified training dataset is not found.

Source code in src/crowdcent_challenge/client.py

def get_training_dataset(self, version: str) -> Dict[str, Any]:
    """Gets details for a specific training dataset version.

    Args:
        version: The version string of the training dataset (e.g., '1.0', '2.1').

    Returns:
        A dictionary representing the specified training dataset.

    Raises:
        NotFoundError: If the challenge or the specified training dataset is not found.
    """
    response = self._request(
        "GET", f"/challenges/{self.challenge_slug}/training_data/{version}/"
    )
    return response.json()

download_training_dataset(version, dest_path)

Downloads the training data file for a specific dataset version.

Parameters:

Name	Type	Description	Default
version	str	The version string of the training dataset (e.g., '1.0', '2.1') or 'latest' to get the latest version.	required
dest_path	str	The local file path to save the downloaded dataset.	required

Raises:

Type	Description
NotFoundError	If the challenge, dataset, or its file is not found.

Source code in src/crowdcent_challenge/client.py

def download_training_dataset(self, version: str, dest_path: str):
    """Downloads the training data file for a specific dataset version.

    Args:
        version: The version string of the training dataset (e.g., '1.0', '2.1')
                or 'latest' to get the latest version.
        dest_path: The local file path to save the downloaded dataset.

    Raises:
        NotFoundError: If the challenge, dataset, or its file is not found.
    """
    if version == "latest":
        endpoint = (
            f"/challenges/{self.challenge_slug}/training_data/latest/download/"
        )
    else:
        endpoint = (
            f"/challenges/{self.challenge_slug}/training_data/{version}/download/"
        )

    logger.info(
        f"Downloading training data for challenge '{self.challenge_slug}' v{version} to {dest_path}"
    )
    response = self._request("GET", endpoint, stream=True)

    try:
        with open(dest_path, "wb") as f:
            for chunk in response.iter_content(chunk_size=8192):
                f.write(chunk)
        logger.info(f"Successfully downloaded training data to {dest_path}")
    except IOError as e:
        logger.error(f"Failed to write dataset to {dest_path}: {e}")
        raise CrowdCentAPIError(f"Failed to write dataset file: {e}") from e

list_inference_data()

Lists all inference data periods for this challenge.

Returns:

Type	Description
List[Dict[str, Any]]	A list of dictionaries, each representing an inference data period.

Raises:

Type	Description
NotFoundError	If the challenge is not found.

Source code in src/crowdcent_challenge/client.py

def list_inference_data(self) -> List[Dict[str, Any]]:
    """Lists all inference data periods for this challenge.

    Returns:
        A list of dictionaries, each representing an inference data period.

    Raises:
        NotFoundError: If the challenge is not found.
    """
    response = self._request(
        "GET", f"/challenges/{self.challenge_slug}/inference_data/"
    )
    return response.json()

get_current_inference_data()

Gets the current inference data period for this challenge.

Returns:

Type	Description
Dict[str, Any]	A dictionary representing the current inference data period.

Raises:

Type	Description
NotFoundError	If the challenge has no active inference period.

Source code in src/crowdcent_challenge/client.py

def get_current_inference_data(self) -> Dict[str, Any]:
    """Gets the current inference data period for this challenge.

    Returns:
        A dictionary representing the current inference data period.

    Raises:
        NotFoundError: If the challenge has no active inference period.
    """
    response = self._request(
        "GET", f"/challenges/{self.challenge_slug}/inference_data/current/"
    )
    return response.json()

get_inference_data(release_date)

Gets details for a specific inference data period by its release date.

Parameters:

Name	Type	Description	Default
release_date	str	The release date of the inference data in 'YYYY-MM-DD' format.	required

Returns:

Type	Description
Dict[str, Any]	A dictionary representing the specified inference data period.

Raises:

Type	Description
NotFoundError	If the challenge or the specified inference data is not found.
ClientError	If the date format is invalid.

Source code in src/crowdcent_challenge/client.py

def get_inference_data(self, release_date: str) -> Dict[str, Any]:
    """Gets details for a specific inference data period by its release date.

    Args:
        release_date: The release date of the inference data in 'YYYY-MM-DD' format.

    Returns:
        A dictionary representing the specified inference data period.

    Raises:
        NotFoundError: If the challenge or the specified inference data is not found.
        ClientError: If the date format is invalid.
    """
    # Validate date format
    try:
        datetime.strptime(release_date, "%Y-%m-%d")
    except ValueError:
        raise ClientError(
            f"Invalid date format: {release_date}. Use 'YYYY-MM-DD' format."
        )

    response = self._request(
        "GET", f"/challenges/{self.challenge_slug}/inference_data/{release_date}/"
    )
    return response.json()

download_inference_data(release_date, dest_path)

Downloads the inference features file for a specific period.

Parameters:

Name	Type	Description	Default
release_date	str	The release date of the inference data in 'YYYY-MM-DD' format or 'current' to get the current period's data.	required
dest_path	str	The local file path to save the downloaded features file.	required

Raises:

Type	Description
NotFoundError	If the challenge, inference data, or its file is not found.
ClientError	If the date format is invalid.

Source code in src/crowdcent_challenge/client.py

def download_inference_data(self, release_date: str, dest_path: str):
    """Downloads the inference features file for a specific period.

    Args:
        release_date: The release date of the inference data in 'YYYY-MM-DD' format
                     or 'current' to get the current period's data.
        dest_path: The local file path to save the downloaded features file.

    Raises:
        NotFoundError: If the challenge, inference data, or its file is not found.
        ClientError: If the date format is invalid.
    """
    if release_date == "current":
        endpoint = (
            f"/challenges/{self.challenge_slug}/inference_data/current/download/"
        )
    else:
        # Validate date format
        try:
            datetime.strptime(release_date, "%Y-%m-%d")
        except ValueError:
            raise ClientError(
                f"Invalid date format: {release_date}. Use 'YYYY-MM-DD' format."
            )

        endpoint = f"/challenges/{self.challenge_slug}/inference_data/{release_date}/download/"

    logger.info(
        f"Downloading inference data for challenge '{self.challenge_slug}' {release_date} to {dest_path}"
    )
    response = self._request("GET", endpoint, stream=True)

    try:
        with open(dest_path, "wb") as f:
            for chunk in response.iter_content(chunk_size=8192):
                f.write(chunk)
        logger.info(f"Successfully downloaded inference data to {dest_path}")
    except IOError as e:
        logger.error(f"Failed to write inference data to {dest_path}: {e}")
        raise CrowdCentAPIError(f"Failed to write inference data file: {e}") from e

list_submissions(period=None)

Lists the authenticated user's submissions for this challenge.

Parameters:

Name	Type	Description	Default
period	Optional[str]	Optional filter for submissions by period: - 'current': Only show submissions for the current active period - 'YYYY-MM-DD': Only show submissions for a specific inference period date	None

Returns:

Type	Description
List[Dict[str, Any]]	A list of dictionaries, each representing a submission.

Source code in src/crowdcent_challenge/client.py

def list_submissions(self, period: Optional[str] = None) -> List[Dict[str, Any]]:
    """Lists the authenticated user's submissions for this challenge.

    Args:
        period: Optional filter for submissions by period:
              - 'current': Only show submissions for the current active period
              - 'YYYY-MM-DD': Only show submissions for a specific inference period date

    Returns:
        A list of dictionaries, each representing a submission.
    """
    params = {}
    if period:
        params["period"] = period

    response = self._request(
        "GET", f"/challenges/{self.challenge_slug}/submissions/", params=params
    )
    return response.json()

get_submission(submission_id)

Gets details for a specific submission by its ID.

Parameters:

Name	Type	Description	Default
submission_id	int	The ID of the submission to retrieve.	required

Returns:

Type	Description
Dict[str, Any]	A dictionary representing the specified submission.

Raises:

Type	Description
NotFoundError	If the submission with the given ID is not found or doesn't belong to the user.

Source code in src/crowdcent_challenge/client.py

def get_submission(self, submission_id: int) -> Dict[str, Any]:
    """Gets details for a specific submission by its ID.

    Args:
        submission_id: The ID of the submission to retrieve.

    Returns:
        A dictionary representing the specified submission.

    Raises:
        NotFoundError: If the submission with the given ID is not found
                       or doesn't belong to the user.
    """
    response = self._request(
        "GET", f"/challenges/{self.challenge_slug}/submissions/{submission_id}/"
    )
    return response.json()

submit_predictions(file_path)

Submits a prediction file for the current active inference period of this challenge.

The file must be a Parquet file with the required prediction columns: id, pred_1M, pred_3M, pred_6M, pred_9M, pred_12M

Parameters:

Name	Type	Description	Default
file_path	str	The path to the prediction Parquet file.	required

Returns:

Type	Description
Dict[str, Any]	A dictionary representing the newly created submission.

Raises:

Type	Description
FileNotFoundError	If the specified file_path does not exist.
ClientError	If the submission is invalid (e.g., wrong format, outside submission window, already submitted, etc).

Source code in src/crowdcent_challenge/client.py

def submit_predictions(self, file_path: str) -> Dict[str, Any]:
    """Submits a prediction file for the current active inference period of this challenge.

    The file must be a Parquet file with the required prediction columns:
    id, pred_1M, pred_3M, pred_6M, pred_9M, pred_12M

    Args:
        file_path: The path to the prediction Parquet file.

    Returns:
        A dictionary representing the newly created submission.

    Raises:
        FileNotFoundError: If the specified file_path does not exist.
        ClientError: If the submission is invalid (e.g., wrong format,
                     outside submission window, already submitted, etc).
    """
    logger.info(
        f"Submitting predictions from {file_path} to challenge '{self.challenge_slug}'"
    )
    try:
        with open(file_path, "rb") as f:
            files = {
                "prediction_file": (
                    os.path.basename(file_path),
                    f,
                    "application/octet-stream",
                )
            }
            response = self._request(
                "POST",
                f"/challenges/{self.challenge_slug}/submissions/",
                files=files,
            )
        logger.info(
            f"Successfully submitted predictions to challenge '{self.challenge_slug}'"
        )
        return response.json()
    except FileNotFoundError as e:
        logger.error(f"Prediction file not found at {file_path}")
        raise FileNotFoundError(f"Prediction file not found at {file_path}") from e
    except IOError as e:
        logger.error(f"Failed to read prediction file {file_path}: {e}")
        raise CrowdCentAPIError(f"Failed to read prediction file: {e}") from e

switch_challenge(new_challenge_slug)

Switch this client to interact with a different challenge.

Parameters:

Name	Type	Description	Default
new_challenge_slug	str	The slug identifier for the new challenge.	required

Returns:

Type	Description
	None. The client is modified in-place.

Source code in src/crowdcent_challenge/client.py

def switch_challenge(self, new_challenge_slug: str):
    """Switch this client to interact with a different challenge.

    Args:
        new_challenge_slug: The slug identifier for the new challenge.

    Returns:
        None. The client is modified in-place.
    """
    self.challenge_slug = new_challenge_slug
    logger.info(f"Client switched to challenge '{new_challenge_slug}'")