Python API Client

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

crowdcent_challenge.client.ChallengeClient

Bases: DataAPI, SubmissionsAPI, SimulatorAPI, TradingAPI, BaseClient

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 — plus meta-model simulation and live trading (staff preview until Trading GA).

Source code in src/crowdcent_challenge/client/__init__.py

class ChallengeClient(DataAPI, SubmissionsAPI, SimulatorAPI, TradingAPI, BaseClient):
    """
    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 — plus meta-model simulation
    and live trading (staff preview until Trading GA).
    """

__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 https://crowdcent.com/api.	None

Source code in src/crowdcent_challenge/client/base.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
                  https://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/base.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}")

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	None. The client is modified in-place.

Source code in src/crowdcent_challenge/client/base.py

def switch_challenge(self, new_challenge_slug: str) -> None:
    """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}'")

check_auth()

Checks the presenting API key and reports its capabilities.

A cheap validity probe: use it to fail fast at startup, or to decide whether trading features should be surfaced at all.

Returns:

Type	Description
Dict[str, Any]	A dictionary with: - username: The account the key belongs to. - allow_trading: Whether this key may call mutating trading endpoints (the per-key "Allow live trading" switch). - oms_access: Whether the trading API is open to this user at all (staff preview until Trading GA).

Raises:

Type	Description
AuthenticationError	If the key is invalid or revoked.

Source code in src/crowdcent_challenge/client/base.py

def check_auth(self) -> Dict[str, Any]:
    """Checks the presenting API key and reports its capabilities.

    A cheap validity probe: use it to fail fast at startup, or to decide
    whether trading features should be surfaced at all.

    Returns:
        A dictionary with:
            - `username`: The account the key belongs to.
            - `allow_trading`: Whether this key may call mutating trading
              endpoints (the per-key "Allow live trading" switch).
            - `oms_access`: Whether the trading API is open to this user
              at all (staff preview until Trading GA).

    Raises:
        AuthenticationError: If the key is invalid or revoked.
    """
    response = self._request("GET", "/auth/check/")
    return response.json()

get_trading_accounts()

Lists your Hyperliquid trading accounts (both networks).

Returns only safe fields (status, can_trade, master_address, agent_expires_at, builder_approved, network) — key material never traverses the API.

Source code in src/crowdcent_challenge/client/trading.py

def get_trading_accounts(self) -> List[Dict[str, Any]]:
    """Lists your Hyperliquid trading accounts (both networks).

    Returns only safe fields (`status`, `can_trade`, `master_address`,
    `agent_expires_at`, `builder_approved`, `network`) — key material
    never traverses the API.
    """
    response = self._request("GET", "/trading/accounts/")
    return response.json()["accounts"]

get_mandate(network='testnet')

Gets the mandate (execution policy + weighted sleeves) for this challenge on the given network.

Raises:

Type	Description
NotFoundError	NO_MANDATE if none exists yet — create one with meth:set_mandate.

Source code in src/crowdcent_challenge/client/trading.py

def get_mandate(self, network: str = "testnet") -> Dict[str, Any]:
    """Gets the mandate (execution policy + weighted sleeves) for this
    challenge on the given network.

    Raises:
        NotFoundError: `NO_MANDATE` if none exists yet — create one with
            :py:meth:`set_mandate`.
    """
    response = self._request(
        "GET",
        f"/challenges/{self.challenge_slug}/trading/mandate/",
        params={"network": network},
    )
    return response.json()

set_mandate(mandate, network='testnet')

Creates or fully replaces the mandate for this challenge/network.

Parameters:

Name	Type	Description	Default
mandate	Dict[str, Any]	{"sleeves": [{"config" | "config_token", "weight", "label"?}, ...], ...execution knobs...}. Sleeve configs use the simulation vocabulary and validate through the same parser and tier locks as the site. Optional execution knobs (order_type, target_leverage, schedule_enabled, schedule_at_time, twap_minutes, stop_loss_pct, ...) clamp to the web's bounds.	required
network	str	"testnet" (default) or "mainnet".	'testnet'

Returns:

Type	Description
Dict[str, Any]	The saved mandate, same shape as meth:get_mandate.

Source code in src/crowdcent_challenge/client/trading.py

def set_mandate(
    self, mandate: Dict[str, Any], network: str = "testnet"
) -> Dict[str, Any]:
    """Creates or fully replaces the mandate for this challenge/network.

    Args:
        mandate: ``{"sleeves": [{"config" | "config_token", "weight",
            "label"?}, ...], ...execution knobs...}``. Sleeve configs use
            the simulation vocabulary and validate through the same
            parser and tier locks as the site. Optional execution knobs
            (`order_type`, `target_leverage`, `schedule_enabled`,
            `schedule_at_time`, `twap_minutes`, `stop_loss_pct`, ...)
            clamp to the web's bounds.
        network: "testnet" (default) or "mainnet".

    Returns:
        The saved mandate, same shape as :py:meth:`get_mandate`.
    """
    response = self._request(
        "PUT",
        f"/challenges/{self.challenge_slug}/trading/mandate/",
        json_data={**mandate, "network": network},
    )
    return response.json()

get_target_book(network='testnet')

Gets the blended target book the mandate's sleeves currently resolve to: target holdings, as_of ranking day, and per-sleeve books.

Source code in src/crowdcent_challenge/client/trading.py

def get_target_book(self, network: str = "testnet") -> Dict[str, Any]:
    """Gets the blended target book the mandate's sleeves currently
    resolve to: target holdings, `as_of` ranking day, and per-sleeve
    books."""
    response = self._request(
        "GET",
        f"/challenges/{self.challenge_slug}/trading/book/",
        params={"network": network},
    )
    return response.json()

preview_rebalance(network='testnet')

Plans a rebalance WITHOUT executing (dry run, persisted for audit).

Returns the full plan (trades, skipped_trades, gross, turnover, est_fees) plus run_id and plan_hash — show the plan to the user, get their explicit confirmation, then pass the hash to meth:execute_rebalance within 10 minutes.

Source code in src/crowdcent_challenge/client/trading.py

def preview_rebalance(self, network: str = "testnet") -> Dict[str, Any]:
    """Plans a rebalance WITHOUT executing (dry run, persisted for audit).

    Returns the full plan (`trades`, `skipped_trades`, `gross`,
    `turnover`, `est_fees`) plus `run_id` and **`plan_hash`** — show the
    plan to the user, get their explicit confirmation, then pass the
    hash to :py:meth:`execute_rebalance` within 10 minutes.
    """
    response = self._request(
        "POST",
        f"/challenges/{self.challenge_slug}/trading/rebalance/preview/",
        json_data={"network": network},
    )
    return response.json()

execute_rebalance(plan_hash, network='testnet')

Executes a LIVE rebalance on your Hyperliquid account.

Real orders; real money on mainnet. Requires a fresh (<10 min) preview's plan_hash as consent evidence, else the server answers 409 CONFIRMATION_REQUIRED. Execution re-plans fresh, so fills may differ slightly from the preview; the gross cap, minimum notional, and stale-book guards always bind server-side.

Raises:

Type	Description
ClientError	CONFIRMATION_REQUIRED (409) without a fresh matching hash; ACCOUNT_BUSY (409) if another OMS action holds the account lock — do not retry immediately.

Source code in src/crowdcent_challenge/client/trading.py

def execute_rebalance(
    self, plan_hash: str, network: str = "testnet"
) -> Dict[str, Any]:
    """Executes a LIVE rebalance on your Hyperliquid account.

    Real orders; real money on mainnet. Requires a fresh (<10 min)
    preview's `plan_hash` as consent evidence, else the server answers
    409 `CONFIRMATION_REQUIRED`. Execution re-plans fresh, so fills may
    differ slightly from the preview; the gross cap, minimum notional,
    and stale-book guards always bind server-side.

    Raises:
        ClientError: `CONFIRMATION_REQUIRED` (409) without a fresh
            matching hash; `ACCOUNT_BUSY` (409) if another OMS action
            holds the account lock — do not retry immediately.
    """
    response = self._request(
        "POST",
        f"/challenges/{self.challenge_slug}/trading/rebalance/",
        json_data={"network": network, "plan_hash": plan_hash},
    )
    return response.json()

flatten(plan_hash=None, *, preview=False, network='testnet')

Closes every position (full liquidation) — two-step like rebalancing.

Call with preview=True first to get the flatten plan and its plan_hash; then call again with that hash to execute. Flatten hashes are kind-scoped: a rebalance preview's hash never authorizes a flatten.

Source code in src/crowdcent_challenge/client/trading.py

def flatten(
    self,
    plan_hash: Optional[str] = None,
    *,
    preview: bool = False,
    network: str = "testnet",
) -> Dict[str, Any]:
    """Closes every position (full liquidation) — two-step like
    rebalancing.

    Call with ``preview=True`` first to get the flatten plan and its
    `plan_hash`; then call again with that hash to execute. Flatten
    hashes are kind-scoped: a rebalance preview's hash never authorizes
    a flatten.
    """
    if preview:
        payload: Dict[str, Any] = {"network": network, "preview": True}
    else:
        payload = {"network": network, "plan_hash": plan_hash}
    response = self._request(
        "POST",
        f"/challenges/{self.challenge_slug}/trading/flatten/",
        json_data=payload,
    )
    return response.json()

pause_trading(network='testnet')

Pauses the mandate (scheduled trading off). Works with ANY valid key — killing risk is never blocked by the trading scope.

Source code in src/crowdcent_challenge/client/trading.py

def pause_trading(self, network: str = "testnet") -> Dict[str, Any]:
    """Pauses the mandate (scheduled trading off). Works with ANY valid
    key — killing risk is never blocked by the trading scope."""
    response = self._request(
        "POST",
        f"/challenges/{self.challenge_slug}/trading/pause/",
        json_data={"network": network},
    )
    return response.json()

resume_trading(network='testnet')

Resumes the mandate (scheduled trading on). Requires a trade-enabled key — re-enabling risk is a trading action.

Source code in src/crowdcent_challenge/client/trading.py

def resume_trading(self, network: str = "testnet") -> Dict[str, Any]:
    """Resumes the mandate (scheduled trading on). Requires a
    trade-enabled key — re-enabling risk is a trading action."""
    response = self._request(
        "POST",
        f"/challenges/{self.challenge_slug}/trading/resume/",
        json_data={"network": network},
    )
    return response.json()

list_rebalance_runs(limit=10, network='testnet')

Lists recent rebalance runs (the audit trail): kind, status, release_date, result summary, error. The morning-briefing feed.

Source code in src/crowdcent_challenge/client/trading.py

def list_rebalance_runs(
    self, limit: int = 10, network: str = "testnet"
) -> List[Dict[str, Any]]:
    """Lists recent rebalance runs (the audit trail): kind, status,
    release_date, result summary, error. The morning-briefing feed."""
    response = self._request(
        "GET",
        f"/challenges/{self.challenge_slug}/trading/runs/",
        params={"network": network, "limit": limit},
    )
    return response.json()["runs"]

list_orders(status=None, network='testnet')

Lists blotter orders, newest first, optionally filtered by status (resting, filled, twap_running, ...).

Source code in src/crowdcent_challenge/client/trading.py

def list_orders(
    self, status: Optional[str] = None, network: str = "testnet"
) -> List[Dict[str, Any]]:
    """Lists blotter orders, newest first, optionally filtered by
    status (resting, filled, twap_running, ...)."""
    params: Dict[str, Any] = {"network": network}
    if status:
        params["status"] = status
    response = self._request(
        "GET",
        f"/challenges/{self.challenge_slug}/trading/orders/",
        params=params,
    )
    return response.json()["orders"]

get_simulator_capabilities()

Gets the simulator knob vocabulary for YOUR tier.

Call this before building simulation configs: it lists the allowed values per knob for the presenting key's points tier, the sweep and blend budgets, the available data range, and which features are sealed behind higher tiers.

Returns:

Type	Description
Dict[str, Any]	A dictionary with tier, data, config (allowed values per
Dict[str, Any]	knob), sealed_features, sweep (budgets + sweepable knobs),
Dict[str, Any]	blend, benchmark_trials, and include_options.

Source code in src/crowdcent_challenge/client/simulator.py

def get_simulator_capabilities(self) -> Dict[str, Any]:
    """Gets the simulator knob vocabulary for YOUR tier.

    Call this before building simulation configs: it lists the allowed
    values per knob for the presenting key's points tier, the sweep and
    blend budgets, the available data range, and which features are
    sealed behind higher tiers.

    Returns:
        A dictionary with `tier`, `data`, `config` (allowed values per
        knob), `sealed_features`, `sweep` (budgets + sweepable knobs),
        `blend`, `benchmark_trials`, and `include_options`.
    """
    response = self._request("GET", f"/challenges/{self.challenge_slug}/simulator/")
    return response.json()

run_simulation(config=None, *, config_token=None, include=None, benchmark_trials=0)

Backtests one portfolio configuration on the live meta-model.

Runs the exact engine behind the site's Simulation tab: the simulator trades the meta-model's published rankings as a long/short portfolio with your chosen construction knobs (cohort sizes, rebalance cadence, weighting scheme, fees, funding, ...).

Knobs above your tier are silently clamped to their accessible values, identical to the web UI. Check the echoed config and the locked list in the response to see what was clamped.

Parameters:

Name	Type	Description	Default
config	Optional[Dict[str, Any]]	SimulationConfig field names with JSON scalars, e.g. {"n_long": 10, "n_short": 10, "rebalance_days": "10t", "weighting": "inv_vol", "include_funding": True}. Omitted knobs use the site's defaults.	None
config_token	Optional[str]	Alternatively, a compact config token from a previous response or a site URL — reproduces that exact config. Mutually exclusive with config.	None
include	Optional[List[str]]	Optional extras: any of "curve" (daily series), "holdings" (current book), "monthly" (returns grid), "contributions" (per-asset P&L attribution). Defaults to none, keeping responses compact.	None
benchmark_trials	int	0, 25, or 100 — score the signal against that many random-ranking portfolios with identical construction.	0

Returns:

Type	Description
Dict[str, Any]	A dictionary with the clamped config echo, locked,
Dict[str, Any]	config_token, web_url (the site pre-loaded with this exact
Dict[str, Any]	config), as_of, n_days, stats, is_stats, oos_stats,
Dict[str, Any]	plus any include extras and benchmark results.

Example

result = client.run_simulation(
    config={"n_long": 10, "n_short": 10, "weighting": "inv_vol"}
)
result["stats"]["sharpe"]  # 1.42

Source code in src/crowdcent_challenge/client/simulator.py

def run_simulation(
    self,
    config: Optional[Dict[str, Any]] = None,
    *,
    config_token: Optional[str] = None,
    include: Optional[List[str]] = None,
    benchmark_trials: int = 0,
) -> Dict[str, Any]:
    """Backtests one portfolio configuration on the live meta-model.

    Runs the exact engine behind the site's Simulation tab: the
    simulator trades the meta-model's published rankings as a long/short
    portfolio with your chosen construction knobs (cohort sizes,
    rebalance cadence, weighting scheme, fees, funding, ...).

    Knobs above your tier are silently clamped to their accessible
    values, identical to the web UI. Check the echoed `config` and the
    `locked` list in the response to see what was clamped.

    Args:
        config: SimulationConfig field names with JSON scalars, e.g.
            ``{"n_long": 10, "n_short": 10, "rebalance_days": "10t",
            "weighting": "inv_vol", "include_funding": True}``. Omitted
            knobs use the site's defaults.
        config_token: Alternatively, a compact config token from a
            previous response or a site URL — reproduces that exact
            config. Mutually exclusive with `config`.
        include: Optional extras: any of `"curve"` (daily series),
            `"holdings"` (current book), `"monthly"` (returns grid),
            `"contributions"` (per-asset P&L attribution). Defaults to
            none, keeping responses compact.
        benchmark_trials: 0, 25, or 100 — score the signal against that
            many random-ranking portfolios with identical construction.

    Returns:
        A dictionary with the clamped `config` echo, `locked`,
        `config_token`, `web_url` (the site pre-loaded with this exact
        config), `as_of`, `n_days`, `stats`, `is_stats`, `oos_stats`,
        plus any `include` extras and `benchmark` results.

    Example:
        ```python
        result = client.run_simulation(
            config={"n_long": 10, "n_short": 10, "weighting": "inv_vol"}
        )
        result["stats"]["sharpe"]  # 1.42
        ```
    """
    payload: Dict[str, Any] = {}
    if config is not None:
        payload["config"] = config
    if config_token is not None:
        payload["config_token"] = config_token
    if include:
        payload["include"] = include
    if benchmark_trials:
        payload["benchmark_trials"] = benchmark_trials
    response = self._request(
        "POST",
        f"/challenges/{self.challenge_slug}/simulator/run/",
        json_data=payload,
    )
    return response.json()

run_sweep(config, sweep, *, on_chunk=None)

Grid-searches portfolio configurations on the meta-model.

The server runs one cost-bounded chunk per request; this method loops offset -> next_offset transparently until the whole grid has run, so you always get the complete result set back.

How to read a sweep: plateaus, not peaks. A lone bright cell is luck; a bright region is structure. Prefer configurations whose neighbors also perform, and weight out-of-sample stats (oos_stats) over in-sample when picking a candidate.

Parameters:

Name	Type	Description	Default
config	Dict[str, Any]	The base configuration (same vocabulary as meth:run_simulation); swept knobs override it.	required
sweep	Dict[str, List[Any]]	Mapping of sweepable knob -> list of values, e.g. {"n_long": [5, 10, 20], "rebalance_days": ["5t", "10t"]}. See meth:get_simulator_capabilities for your tier's sweepable knobs, allowed values, and grid budget (96 configs at Contender tier, 24 below).	required
on_chunk	Optional[Callable[[List[Dict[str, Any]], int], None]]	Optional callable on_chunk(results_so_far, total) invoked after each server chunk — useful for progress bars.	None

Returns:

Type	Description
Dict[str, Any]	A dictionary with total and results: one entry per grid cell
Dict[str, Any]	(in deterministic grid order), each holding config_token,
Dict[str, Any]	params (the swept values), and stats/is_stats/oos_stats
Dict[str, Any]	or an error.

Raises:

Type	Description
ClientError	SWEEP_TOO_LARGE if the grid exceeds your tier's budget; INVALID_SWEEP_KNOB/TIER_REQUIRED for bad knobs.

Source code in src/crowdcent_challenge/client/simulator.py

def run_sweep(
    self,
    config: Dict[str, Any],
    sweep: Dict[str, List[Any]],
    *,
    on_chunk: Optional[Callable[[List[Dict[str, Any]], int], None]] = None,
) -> Dict[str, Any]:
    """Grid-searches portfolio configurations on the meta-model.

    The server runs one cost-bounded chunk per request; this method
    loops `offset` -> `next_offset` transparently until the whole grid
    has run, so you always get the complete result set back.

    How to read a sweep: **plateaus, not peaks.** A lone bright cell is
    luck; a bright region is structure. Prefer configurations whose
    neighbors also perform, and weight out-of-sample stats (`oos_stats`)
    over in-sample when picking a candidate.

    Args:
        config: The base configuration (same vocabulary as
            :py:meth:`run_simulation`); swept knobs override it.
        sweep: Mapping of sweepable knob -> list of values, e.g.
            ``{"n_long": [5, 10, 20], "rebalance_days": ["5t", "10t"]}``.
            See :py:meth:`get_simulator_capabilities` for your tier's
            sweepable knobs, allowed values, and grid budget (96 configs
            at Contender tier, 24 below).
        on_chunk: Optional callable ``on_chunk(results_so_far, total)``
            invoked after each server chunk — useful for progress bars.

    Returns:
        A dictionary with `total` and `results`: one entry per grid cell
        (in deterministic grid order), each holding `config_token`,
        `params` (the swept values), and `stats`/`is_stats`/`oos_stats`
        or an `error`.

    Raises:
        ClientError: `SWEEP_TOO_LARGE` if the grid exceeds your tier's
            budget; `INVALID_SWEEP_KNOB`/`TIER_REQUIRED` for bad knobs.
    """
    results: List[Dict[str, Any]] = []
    offset = 0
    total = None
    while True:
        response = self._request(
            "POST",
            f"/challenges/{self.challenge_slug}/simulator/sweep/",
            json_data={"config": config, "sweep": sweep, "offset": offset},
        )
        body = response.json()
        total = body["total"]
        results.extend(body["results"])
        if on_chunk is not None:
            on_chunk(results, total)
        if body.get("next_offset") is None:
            break
        offset = body["next_offset"]
    return {"total": total, "results": results}

run_blend(sleeves)

Blends weighted sleeves into one ensemble book and evaluates it.

Each sleeve runs once and the weighted blend is formed on date-aligned daily returns. Fail-closed: any sleeve error fails the whole blend. Sleeve count is capped by tier (5 at Contender, 3 below).

Parameters:

Name	Type	Description	Default
sleeves	List[Dict[str, Any]]	A list of {"config": {...} | "config_token": "...", "weight": float, "label": str?} dictionaries.	required

Returns:

Type	Description
Dict[str, Any]	A dictionary with blend stats/is_stats/oos_stats, the
Dict[str, Any]	sleeve correlation matrix, and per-sleeve stats (computed on
Dict[str, Any]	the aligned window so they are directly comparable).

Source code in src/crowdcent_challenge/client/simulator.py

def run_blend(self, sleeves: List[Dict[str, Any]]) -> Dict[str, Any]:
    """Blends weighted sleeves into one ensemble book and evaluates it.

    Each sleeve runs once and the weighted blend is formed on
    date-aligned daily returns. Fail-closed: any sleeve error fails the
    whole blend. Sleeve count is capped by tier (5 at Contender, 3
    below).

    Args:
        sleeves: A list of ``{"config": {...} | "config_token": "...",
            "weight": float, "label": str?}`` dictionaries.

    Returns:
        A dictionary with blend `stats`/`is_stats`/`oos_stats`, the
        sleeve `correlation` matrix, and per-sleeve stats (computed on
        the aligned window so they are directly comparable).
    """
    response = self._request(
        "POST",
        f"/challenges/{self.challenge_slug}/simulator/blend/",
        json_data={"sleeves": sleeves},
    )
    return response.json()

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/submissions.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/submissions.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='submission.parquet', df=None, slot=1, queue_next=True, temp=True, max_retries=3, retry_delay=1.0, is_experimental=False, notes='')

Submit predictions for this challenge.

If a submission window is currently open, the prediction is submitted immediately. If no window is open, the prediction is queued and will be automatically submitted when the next window opens.

You can provide either a file path to an existing Parquet file or a DataFrame that will be temporarily saved as Parquet for submission.

The data must contain the required prediction columns specified by the challenge (e.g., id, pred_10d, pred_30d).

Parameters:

Name	Type	Description	Default
file_path	str	Optional path to an existing prediction Parquet file.	'submission.parquet'
df	Optional[IntoFrameT]	Optional DataFrame with the prediction columns. If provided, it will be temporarily saved as Parquet for submission.	None
slot	int	Submission slot number (1-based).	1
queue_next	bool	Whether to also queue this submission for the next period (auto-rollover). Defaults to True. When submitting during an open window, this queues a copy for the following period.	True
temp	bool	Whether to save the DataFrame to a temporary file.	True
max_retries	int	Maximum number of retry attempts for connection errors (default: 3).	3
retry_delay	float	Initial delay between retries in seconds (default: 1.0).	1.0
is_experimental	bool	Mark this submission as experimental. Experimental submissions are scored normally and receive a shadow percentile against the non-experimental competitive field, but are excluded from the leaderboard, the meta-model, and CC Points performance adjustment. Constraint: at least one slot per period must have a non-experimental submission; submitting experimental without a non-experimental sibling in another slot is rejected. Defaults to False.	False
notes	str	Free-text annotation for this submission, max 2000 characters. Notes are private to the submission owner. Defaults to "".	''

Returns:

Type	Description
Dict[str, Any]	A dictionary with submission details. The shape depends on context:
Dict[str, Any]	Window open (immediate submission): Contains submission fields like id, status, slot, submitted_at, is_experimental, notes, plus queued_for_next (bool). If the queue copy was rejected, queue_error_code and queue_error are populated.
Dict[str, Any]	Window closed (queued): Contains status: "queued", slot, challenge, is_experimental, notes, and a message describing when it will be submitted.

Raises:

Type	Description
ValueError	If neither file_path nor df is provided, or if both are provided.
FileNotFoundError	If the specified file_path does not exist.
ClientError	If the submission is invalid (e.g., wrong format, missing columns, experimental constraint violated).

Examples:

Submit from a DataFrame

client.submit_predictions(df=predictions_df)

Submit from a file

client.submit_predictions(file_path="predictions.parquet")

Submit and opt-out of auto-queueing for next period

client.submit_predictions(df=predictions_df, queue_next=False)

Submit an experimental prediction with a note

client.submit_predictions( df=predictions_df, slot=2, is_experimental=True, notes="2-layer transformer w/ sector embeddings", )

Source code in src/crowdcent_challenge/client/submissions.py

@nw.narwhalify
def submit_predictions(
    self,
    file_path: str = "submission.parquet",
    df: Optional[IntoFrameT] = None,
    slot: int = 1,
    queue_next: bool = True,
    temp: bool = True,
    max_retries: int = 3,
    retry_delay: float = 1.0,
    is_experimental: bool = False,
    notes: str = "",
) -> Dict[str, Any]:
    """Submit predictions for this challenge.

    If a submission window is currently open, the prediction is submitted immediately.
    If no window is open, the prediction is queued and will be automatically submitted
    when the next window opens.

    You can provide either a file path to an existing Parquet file or a DataFrame
    that will be temporarily saved as Parquet for submission.

    The data must contain the required prediction columns specified by the challenge
    (e.g., id, pred_10d, pred_30d).

    Args:
        file_path: Optional path to an existing prediction Parquet file.
        df: Optional DataFrame with the prediction columns. If provided,
            it will be temporarily saved as Parquet for submission.
        slot: Submission slot number (1-based).
        queue_next: Whether to also queue this submission for the next period
            (auto-rollover). Defaults to True. When submitting during an open
            window, this queues a copy for the following period.
        temp: Whether to save the DataFrame to a temporary file.
        max_retries: Maximum number of retry attempts for connection errors (default: 3).
        retry_delay: Initial delay between retries in seconds (default: 1.0).
        is_experimental: Mark this submission as experimental. Experimental
            submissions are scored normally and receive a shadow percentile
            against the non-experimental competitive field, but are excluded
            from the leaderboard, the meta-model, and CC Points performance
            adjustment. **Constraint:** at least one slot per period must
            have a non-experimental submission; submitting experimental
            without a non-experimental sibling in another slot is rejected.
            Defaults to False.
        notes: Free-text annotation for this submission, max 2000 characters.
            Notes are private to the submission owner. Defaults to "".

    Returns:
        A dictionary with submission details. The shape depends on context:

        - **Window open (immediate submission)**: Contains submission fields
            like `id`, `status`, `slot`, `submitted_at`, `is_experimental`,
            `notes`, plus `queued_for_next` (bool). If the queue copy was
            rejected, `queue_error_code` and `queue_error` are populated.
        - **Window closed (queued)**: Contains `status: "queued"`, `slot`,
            `challenge`, `is_experimental`, `notes`, and a `message`
            describing when it will be submitted.

    Raises:
        ValueError: If neither file_path nor df is provided, or if both are provided.
        FileNotFoundError: If the specified file_path does not exist.
        ClientError: If the submission is invalid (e.g., wrong format, missing columns,
            experimental constraint violated).

    Examples:
        # Submit from a DataFrame
        client.submit_predictions(df=predictions_df)

        # Submit from a file
        client.submit_predictions(file_path="predictions.parquet")

        # Submit and opt-out of auto-queueing for next period
        client.submit_predictions(df=predictions_df, queue_next=False)

        # Submit an experimental prediction with a note
        client.submit_predictions(
            df=predictions_df,
            slot=2,
            is_experimental=True,
            notes="2-layer transformer w/ sector embeddings",
        )
    """
    if df is not None:
        df.write_parquet(file_path)
        logger.info(f"Wrote DataFrame to temporary file: {file_path}")

    logger.info(
        f"Submitting predictions from {file_path} to challenge '{self.challenge_slug}' (Slot: {slot or '1'})"
    )

    try:
        with open(file_path, "rb") as f:
            files = {
                "prediction_file": (
                    os.path.basename(file_path),
                    f,
                    "application/octet-stream",
                )
            }
            data_payload = {
                "slot": str(slot),
                "also_queue_next": str(queue_next).lower(),
                "is_experimental": str(is_experimental).lower(),
                "notes": notes,
            }
            response = self._request(
                "POST",
                f"/challenges/{self.challenge_slug}/submissions/",
                files=files,
                data=data_payload,
                max_retries=max_retries,
                retry_delay=retry_delay,
            )

        resp_data = response.json()

        # 202=queued, 200=updated, 201=created
        msg = {202: "queued", 200: "updated", 201: "created"}.get(
            response.status_code, "submitted"
        )
        exp_label = " (experimental)" if is_experimental else ""
        logger.info(f"Submission {msg} (slot {slot}){exp_label}")
        if resp_data.get("queued_for_next"):
            logger.info("Also queued for next period.")
        elif resp_data.get("queue_error_code"):
            logger.warning(
                "Live submission saved but queue copy was rejected: "
                f"{resp_data['queue_error_code']} - {resp_data.get('queue_error')}"
            )

        return resp_data
    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
    finally:
        # Clean up the temporary file if we created one
        if df is not None and temp:
            try:
                os.unlink(file_path)
                logger.debug(f"Cleaned up temporary file: {file_path}")
            except Exception as e:
                logger.warning(
                    f"Failed to clean up temporary file {file_path}: {e}"
                )

get_performance(user=None, scored_only=True, slot=None)

Get performance history for a user (defaults to authenticated user).

Fetches submissions with their scores and percentiles, flattens the nested score data, and returns a list ready to wrap in pandas/polars.

Parameters:

Name	Type	Description	Default
user	Optional[str]	Username to fetch performance for. If None (default), fetches performance for the authenticated user. Note: Fetching other users' performance is not yet supported.	None
scored_only	bool	If True (default), only include submissions that have been scored. For pending submissions, only the most recent (partially resolved) score is available — daily granularity is not currently exposed by the API.	True
slot	Optional[int]	Optional slot filter. If provided, only include submissions from this slot.	None

Returns:

Type	Description
List[Dict[str, Any]]	A list of dictionaries, each containing:
List[Dict[str, Any]]	id: Submission ID
List[Dict[str, Any]]	slot: Submission slot number
List[Dict[str, Any]]	release_date: The inference period date (ISO string)
List[Dict[str, Any]]	submitted_at: When the submission was made (ISO string)
List[Dict[str, Any]]	status: Submission status ("pending" or "evaluated")
List[Dict[str, Any]]	is_experimental: Whether this submission was marked experimental (bool)
List[Dict[str, Any]]	notes: Free-text note attached at submit time (str, may be empty)
List[Dict[str, Any]]	score_*: Individual score metrics (e.g., score_spearman_10d)
List[Dict[str, Any]]	percentile_*: Individual percentile metrics (e.g., percentile_spearman_10d)
List[Dict[str, Any]]	composite_percentile: Overall percentile ranking (if available)

Note

• For submissions with status="pending", scores reflect the most recent partial evaluation (e.g., day 5 of a 10-day prediction). Daily score progression is tracked server-side but not yet available via the API.

• Percentile fields (e.g., composite_percentile=0.75) indicate rank relative to all participants — 0.75 means outperforming 75% of submissions for that period.

• Experimental submissions are included in the result. They are scored against the non-experimental competitive field (shadow percentiles) but excluded from leaderboards, the meta-model, and CC Points performance adjustment. Filter them out client-side if you want only competitive history: [r for r in rows if not r["is_experimental"]].

Example

client = ChallengeClient("momentum-alpha") history = client.get_performance() import pandas as pd df = pd.DataFrame(history)

Source code in src/crowdcent_challenge/client/submissions.py

def get_performance(
    self,
    user: Optional[str] = None,
    scored_only: bool = True,
    slot: Optional[int] = None,
) -> List[Dict[str, Any]]:
    """Get performance history for a user (defaults to authenticated user).

    Fetches submissions with their scores and percentiles, flattens the
    nested score data, and returns a list ready to wrap in pandas/polars.

    Args:
        user: Username to fetch performance for. If None (default), fetches
            performance for the authenticated user.
            *Note: Fetching other users' performance is not yet supported.*
        scored_only: If True (default), only include submissions that have been scored.
            For pending submissions, only the most recent (partially resolved) score
            is available — daily granularity is not currently exposed by the API.
        slot: Optional slot filter. If provided, only include submissions from this slot.

    Returns:
        A list of dictionaries, each containing:
        - id: Submission ID
        - slot: Submission slot number
        - release_date: The inference period date (ISO string)
        - submitted_at: When the submission was made (ISO string)
        - status: Submission status ("pending" or "evaluated")
        - is_experimental: Whether this submission was marked experimental (bool)
        - notes: Free-text note attached at submit time (str, may be empty)
        - score_*: Individual score metrics (e.g., score_spearman_10d)
        - percentile_*: Individual percentile metrics (e.g., percentile_spearman_10d)
        - composite_percentile: Overall percentile ranking (if available)

    Note:
        - For submissions with status="pending", scores reflect the most recent
          partial evaluation (e.g., day 5 of a 10-day prediction). Daily score
          progression is tracked server-side but not yet available via the API.

        - Percentile fields (e.g., composite_percentile=0.75) indicate rank
          relative to all participants — 0.75 means outperforming 75% of
          submissions for that period.

        - Experimental submissions are included in the result. They are
          scored against the non-experimental competitive field (shadow
          percentiles) but excluded from leaderboards, the meta-model, and
          CC Points performance adjustment. Filter them out client-side if
          you want only competitive history:
          ``[r for r in rows if not r["is_experimental"]]``.

    Example:
        >>> client = ChallengeClient("momentum-alpha")
        >>> history = client.get_performance()
        >>> import pandas as pd
        >>> df = pd.DataFrame(history)
    """
    if user is not None:
        raise NotImplementedError(
            "Fetching performance for specific users is not yet supported via the API. "
            "Leave `user=None` to fetch your own performance."
        )

    logger.info(f"Fetching submission history for '{self.challenge_slug}'...")
    submissions = self.list_submissions()

    if not submissions:
        logger.info("No submissions found.")
        return []

    rows = []
    for sub in submissions:
        # Skip unscored if requested
        if scored_only and not sub.get("score_details"):
            continue

        # Skip if slot filter doesn't match
        if slot is not None and sub.get("slot") != slot:
            continue

        row = {
            "id": sub.get("id"),
            "slot": sub.get("slot"),
            "release_date": sub.get("inference_data_release_date", "")[:10]
            if sub.get("inference_data_release_date")
            else None,
            "submitted_at": sub.get("submitted_at"),
            "status": sub.get("status"),
            "is_experimental": sub.get("is_experimental", False),
            "notes": sub.get("notes", "") or "",
        }

        # Flatten score_details (avoid redundant prefix if key already contains it)
        score_details = sub.get("score_details") or {}
        for key, value in score_details.items():
            col = key if "score" in key else f"score_{key}"
            row[col] = value

        # Flatten percentile_details (avoid redundant prefix if key already contains it)
        percentile_details = sub.get("percentile_details") or {}
        for key, value in percentile_details.items():
            col = key if "percentile" in key else f"percentile_{key}"
            row[col] = value

        rows.append(row)

    # Sort by release_date descending (most recent first)
    rows.sort(key=lambda x: x.get("release_date") or "", reverse=True)

    logger.info(f"Loaded {len(rows)} scored submissions.")
    return rows

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/data.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/data.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_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') or the special value "latest" to get the latest version.	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/data.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')
                 or the special value ``"latest"`` to get the latest version.

    Returns:
        A dictionary representing the specified training dataset.

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

    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/data.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":
        latest_info = self.get_training_dataset("latest")
        version = latest_info["version"]

    endpoint = (
        f"/challenges/{self.challenge_slug}/training_data/{version}/download/"
    )
    self._download_file(endpoint, dest_path, f"training data v{version}")

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/data.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_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. You can also pass the special values: - "current" to fetch the current active inference period - "latest" to fetch the most recently available inference period	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/data.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.
                      You can also pass the special values:
                      - ``"current"`` to fetch the current active inference period
                      - ``"latest"`` to fetch the most recently *available* inference period

    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.
    """
    if release_date == "current":
        response = self._request(
            "GET", f"/challenges/{self.challenge_slug}/inference_data/current/"
        )
        return response.json()

    if release_date == "latest":
        # Simply resolve via list_inference_data(); avoid noisy probe.
        periods = self.list_inference_data()
        if not periods:
            raise NotFoundError(
                "No inference data periods found for this challenge."
            )

        latest_period = max(periods, key=lambda p: p["release_date"])
        release_date_iso = latest_period["release_date"]
        release_date = release_date_iso.split("T")[0]

    # Validate date format for explicit dates
    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, poll=True, poll_interval=30, timeout=900)

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 the special values "current" or "latest".	required
dest_path	str	The local file path to save the downloaded features file.	required
poll	bool	Whether to wait for the inference data to be available before downloading.	True
poll_interval	int	Seconds to wait between retries when polling.	30
timeout	Optional[int]	Maximum seconds to wait before raising :class:TimeoutError. None waits indefinitely.	900

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/data.py

def download_inference_data(
    self,
    release_date: str,
    dest_path: str,
    poll: bool = True,
    poll_interval: int = 30,
    timeout: Optional[int] = 900,
):
    """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 the special values ``"current"`` or ``"latest"``.
        dest_path: The local file path to save the downloaded features file.
        poll: Whether to wait for the inference data to be available before downloading.
        poll_interval: Seconds to wait between retries when polling.
        timeout: Maximum seconds to wait before raising :class:`TimeoutError`.
            ``None`` waits indefinitely.

    Raises:
        NotFoundError: If the challenge, inference data, or its file is not found.
        ClientError: If the date format is invalid.
    """
    if release_date == "current":
        # If polling is enabled, delegate to wait_for_inference_data which wraps
        # this method and adds retry logic. Otherwise attempt a single direct
        # download request.
        if poll:
            self.wait_for_inference_data(dest_path, poll_interval, timeout)
            return

        # Polling disabled → attempt once and propagate NotFoundError on 404.
        endpoint = (
            f"/challenges/{self.challenge_slug}/inference_data/current/download/"
        )
    else:
        if release_date == "latest":
            latest_info = self.get_inference_data("latest")
            release_date_iso = latest_info.get("release_date")
            release_date = (
                release_date_iso.split("T")[0] if release_date_iso else None
            )
            if not release_date:
                raise CrowdCentAPIError(
                    "Malformed response when resolving latest inference period."
                )

        # Validate date format after any resolution.
        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/"

    self._download_file(endpoint, dest_path, f"inference data {release_date}")

wait_for_inference_data(dest_path, poll_interval=30, timeout=900)

Waits for the current inference data release to appear and downloads it.

The internal data-generation pipeline begins around 14:00 UTC, but the public inference file becomes available only after it passes data-quality checks. This helper repeatedly calls meth:download_inference_data with release_date="current" until the file is ready (HTTP 404s are silently retried).

Parameters:

Name	Type	Description	Default
dest_path	str	Local path where the parquet file will be saved once available.	required
poll_interval	int	Seconds to wait between retries.	30
timeout	Optional[int]	Maximum seconds to wait before raising :class:TimeoutError. None waits indefinitely.	900

Raises:

Type	Description
TimeoutError	If timeout seconds pass without a successful download.
CrowdCentAPIError	For unrecoverable errors returned by the API.

Source code in src/crowdcent_challenge/client/data.py

def wait_for_inference_data(
    self,
    dest_path: str,
    poll_interval: int = 30,
    timeout: Optional[int] = 900,
) -> None:
    """Waits for the *current* inference data release to appear and downloads it.

    The internal data-generation pipeline begins around 14:00 UTC, but the
    public inference file becomes available only after it passes data-quality
    checks. This helper repeatedly calls
    :py:meth:`download_inference_data` with ``release_date="current"`` until
    the file is ready (HTTP 404s are silently retried).

    Args:
        dest_path: Local path where the parquet file will be saved once available.
        poll_interval: Seconds to wait between retries.
        timeout: Maximum seconds to wait before raising :class:`TimeoutError`.
            ``None`` waits indefinitely.

    Raises:
        TimeoutError: If *timeout* seconds pass without a successful download.
        CrowdCentAPIError: For unrecoverable errors returned by the API.
    """
    start_time = time.time()
    attempts = 0

    while True:
        attempts += 1
        try:
            # Try to download the *current* period *once*. Pass poll=False to avoid
            # the mutual recursion between `wait_for_inference_data` and
            # `download_inference_data` which would otherwise trigger an infinite
            # loop when the file is not yet available.
            self.download_inference_data("current", dest_path, poll=False)
            logger.info(
                f"Successfully downloaded inference data after {attempts} attempt(s) to {dest_path}"
            )
            return  # Success – exit the loop
        except NotFoundError:
            # File not published yet – check timeout and sleep before retrying.
            elapsed = time.time() - start_time
            if timeout is not None and elapsed >= timeout:
                raise TimeoutError(
                    f"Inference data was not available after waiting {timeout} seconds."
                )
            logger.debug(
                f"Inference data not yet available (attempt {attempts}). "
                f"Sleeping {poll_interval}s before retrying."
            )
            time.sleep(poll_interval)

download_meta_model(dest_path)

Downloads the consolidated meta-model file for this challenge.

The meta-model is typically an aggregation (e.g., average) of all valid submissions for past inference periods.

Parameters:

Name	Type	Description	Default
dest_path	str	The local file path to save the downloaded meta-model.	required

Raises:

Type	Description
NotFoundError	If the challenge or its meta-model file is not found.
CrowdCentAPIError	For issues during download or file writing.
PermissionDenied	If the meta-model is not public and user lacks permission.

Source code in src/crowdcent_challenge/client/data.py

def download_meta_model(self, dest_path: str):
    """Downloads the consolidated meta-model file for this challenge.

    The meta-model is typically an aggregation (e.g., average) of all valid
    submissions for past inference periods.

    Args:
        dest_path: The local file path to save the downloaded meta-model.

    Raises:
        NotFoundError: If the challenge or its meta-model file is not found.
        CrowdCentAPIError: For issues during download or file writing.
        PermissionDenied: If the meta-model is not public and user lacks permission.
    """
    endpoint = f"/challenges/{self.challenge_slug}/meta_model/download/"
    self._download_file(endpoint, dest_path, "meta-model")

get_training_dataset_url(version)

Returns a signed, time-limited download URL for a training dataset.

The URL twin of meth:download_training_dataset for callers without a local filesystem (hosted agents): fetch it yourself with any HTTP client.

Parameters:

Name	Type	Description	Default
version	str	The version string, or "latest".	required

Source code in src/crowdcent_challenge/client/data.py

def get_training_dataset_url(self, version: str) -> str:
    """Returns a signed, time-limited download URL for a training dataset.

    The URL twin of :py:meth:`download_training_dataset` for callers
    without a local filesystem (hosted agents): fetch it yourself with
    any HTTP client.

    Args:
        version: The version string, or ``"latest"``.
    """
    if version == "latest":
        version = self.get_training_dataset("latest")["version"]
    return self._signed_url(
        f"/challenges/{self.challenge_slug}/training_data/{version}/download/"
    )

get_inference_data_url(release_date)

Returns a signed, time-limited download URL for inference features.

Parameters:

Name	Type	Description	Default
release_date	str	YYYY-MM-DD, "current", or "latest".	required

Source code in src/crowdcent_challenge/client/data.py

def get_inference_data_url(self, release_date: str) -> str:
    """Returns a signed, time-limited download URL for inference features.

    Args:
        release_date: ``YYYY-MM-DD``, ``"current"``, or ``"latest"``.
    """
    if release_date == "latest":
        release_date = self.get_inference_data("latest")["release_date"].split("T")[
            0
        ]
    if release_date == "current":
        endpoint = (
            f"/challenges/{self.challenge_slug}/inference_data/current/download/"
        )
    else:
        endpoint = (
            f"/challenges/{self.challenge_slug}/inference_data/"
            f"{release_date}/download/"
        )
    return self._signed_url(endpoint)

get_meta_model_url()

Returns a signed, time-limited download URL for the consolidated meta-model file.

Source code in src/crowdcent_challenge/client/data.py

def get_meta_model_url(self) -> str:
    """Returns a signed, time-limited download URL for the consolidated
    meta-model file."""
    return self._signed_url(
        f"/challenges/{self.challenge_slug}/meta_model/download/"
    )