condorgame 0.3.0

A package for participating in the Condor Game on CrunchDAO

Condor Game

Condor game is a real-time probabilistic forecasting challenge hosted by CrunchDAO at crunchdao.com

The goal is to anticipate how asset prices will evolve by providing not a single forecasted value, but a full probability distribution over the future price change at multiple forecast horizons and steps.

The current crypto assets to model are Bitcoin (BTC), Ethereum (ETH), Solana (SOL) and Tether Gold (XAU).

Install

pip install condorgame

What You Must Predict

Trackers must predict the probability distribution of price changes, defined as:

$$ r_{t,k} = P_t - P_{t-k} $$

For each defined step $k$ (e.g., 5 minutes, 1 hour, …), your tracker must return a full probability density function (PDF) over the future price change $r_{t,k}$.

Visualize the challenge

The Condor game is evaluated on incremental return predictions, not raw prices.
Incremental returns capture the relative change in price and produce a stationary series that is easier to model and compare across assets.

Below is an example of a density forecast over incremental returns for the next 24h at 5-minute intervals:

Below is a minimal example showing what your tracker might return:

>>> model.predict(asset="SOL", horizon=86400, step=300)
[
    {
        "step": (k + 1) * step,
        "prediction": {
            "type": "builtin",
            "name": "norm",
            "params": {
                "loc": -0.01,       # mean return
                "scale": 0.4        # standard deviation of return
            }
        }
    }
    for k in range(0, horizon // step)
]

Here is the return forecast mapped into price space:

Create your Tracker

A tracker is a model that processes real-time asset data to predict future price changes. It uses past prices to generate a probabilistic forecast of incremental returns. You can use the data provided by the challenge or any other datasets to improve your predictions.

It operates incrementally: prices are pushed to the tracker as they arrive and predictions are requested at specific times by the framework.

To create your tracker, you need to define a class that implements the TrackerBase interface, which already handles:

• price storage and alignment via PriceStore
• multi-resolution forecasting through predict_all()

As a participant, you only need to implement one method: predict().

1. Price data handling (already provided)

   Each tracker instance contains a PriceStore (self.prices) that:

   • stores recent historical prices per asset
   • maintains a rolling time window (30 days)
   • provides convenient accessors such as:
     • get_last_price()
     • get_prices(asset, days, resolution)
     • get_closest_price(asset, timestamp)

   The framework automatically updates the PriceStore by calling tick(self, data: PriceData) before any prediction request.

   Data example:

    data = {
          "BTC": [(timestamp1, price1), (timestamp2, price2)],
          "SOL": [(timestamp1, price1)],
      }
   

   When it's called:

   • Typically every minute or when new data is available
   • Before any prediction request
   • Can be called multiple times before a predict

2. Required method: predict(self, asset: str, horizon: int, step: int)
   This is the only method you must implement.

   It must return a sequence of predictive density distributions for the incremental price change of an asset:

   • Forecast horizon: horizon seconds into the future
   • Temporal resolution: one density every step seconds
   • Output length: horizon // step

   Each density prediction must comply with the density_pdf specification.

3. Multi-step forecasts (handled automatically)

   You do not need to implement multi-step logic.

   The framework will automatically call your predict() method multiple times via predict_all(asset, horizon, steps) to construct forecasts at different temporal resolutions.

You can refer to the Tracker examples for guidance.

class GaussianStepTracker(TrackerBase):
    """
    A benchmark tracker that models *future incremental returns* as Gaussian-distributed.

    For each forecast step, the tracker returns a normal distribution
    r_{t,step} ~ N(a · mu, √a · sigma) where:
        - mu    = mean historical return
        - sigma = std historical return
        - a = (step / 300) represents the ratio of the forecast step duration to the historical 5-minute return interval.

    This is not a price-distribution; it is a distribution over 
    incremental returns between consecutive steps.
    """
    def __init__(self):
        super().__init__()

    def predict(self, asset: str, horizon: int, step: int):

        # Retrieve recent historical prices sampled at 5-minute resolution
        pairs = self.prices.get_prices(asset, days=5, resolution=300)
        if not pairs:
            return []

        _, past_prices = zip(*pairs)

        if len(past_prices) < 3:
            return []

        # Compute historical incremental returns (price differences)
        returns = np.diff(past_prices)

        # Estimate drift (mean return) and volatility (std dev of returns)
        mu = float(np.mean(returns))
        sigma = float(np.std(returns))

        if sigma <= 0:
            return []

        num_segments = horizon // step

        # Produce one Gaussian for each future time step
        # The returned list must be compatible with the `density_pdf` library.
        distributions = []
        for k in range(1, num_segments + 1):
            distributions.append({
                "step": k * step,
                "type": "mixture",
                "components": [{
                    "density": {
                        "type": "builtin",             # Note: use 'builtin' instead of 'scipy' for speed
                        "name": "norm",  
                        "params": {
                            "loc": (step/300) * mu, 
                            "scale": np.sqrt(step/300) * sigma}
                    },
                    "weight": 1
                }]
            })

        return distributions

Prediction Phase

In each prediction round, players must submit a set of density forecasts.

A prediction round is defined by one asset, one forecast horizon and one or more step resolutions.

• A 24-hour horizon forecast with {5-minute, 1-hour, 6-hour, 24-hour} increments is triggered hourly for each asset.
• A 1-hour horizon forecast with {1-minute, 5-minute, 15-minute, 30-minute, 1-hour} increments is triggered every 12 minutes for each asset.

All required forecasts for a prediction round must be generated within 40 seconds.

Scoring

• Once the full horizon has passed, each prediction is scored using a CRPS scoring function.
• A lower CRPS score reflects more accurate predictions.
• Leaderboard ranking is based on a 7-day rolling average of scores across all assets and horizons, evaluated relative to other participants:
  • for each prediction round, the best CRPS score receives a normalized score of 1
  • the worst 5% of CRPS scores receive a score of 0

Check your Tracker performance

TrackerEvaluator allows you to track your model's performance over time locally before participating in the live game. It maintains:

• Overall CRPS score
• Recent CRPS score
• Quarantine predictions (predictions stored and evaluated at a later time)

A lower CRPS score reflects more accurate predictions.

from condorgame.tracker_evaluator import TrackerEvaluator
from condorgame.examples.benchmarktracker import GaussianStepTracker  # Your custom tracker

# Initialize the tracker evaluator with your custom GaussianStepTracker
tracker_evaluator = TrackerEvaluator(GaussianStepTracker())
# Feed a new price tick for SOL
tracker_evaluator.tick({"SOL": [(ts, price)]})
# You will generate predictive densities for SOL over a 24-hour period (86400s) at multiple step resolutions: 5 minutes, 1 hour, 6 hours and 24 hours
predictions = tracker_evaluator.predict("SOL", horizon=3600*24,
                                        steps=[300, 3600, 3600*6, 3600*24])

print(f"My overall CRPS score: {tracker_evaluator.overall_crps_score("SOL"):.4f}")
print(f"My recent CRPS score: {tracker_evaluator.recent_crps_score("SOL"):.4f}")

Tracker examples

See Tracker examples. There are:

• Quickstarter Notebooks
• Self-contained examples

General Condor Game Advice

The Condor game challenges you to predict the asset location using probabilistic forecasting.

Probabilistic Forecasting

Probabilistic forecasting provides a distribution of possible future values rather than a single point estimate, allowing for uncertainty quantification. Instead of predicting only the most likely outcome, it estimates a range of potential outcomes along with their probabilities by outputting a probability distribution.

A probabilistic forecast models the conditional probability distribution of a future value $(Y_t)$ given past observations $(\mathcal{H}_{t-1})$. This can be expressed as:

$$P(Y_t \mid \mathcal{H}_{t-1})$$

where $(\mathcal{H}_{t-1})$ represents the historical data up to time $(t-1)$. Instead of a single prediction $(\hat{Y}t)$, the model estimates a full probability distribution $(f(Y_t \mid \mathcal{H}{t-1}))$, which can take different parametric forms, such as a Gaussian:

$$Y_t \mid \mathcal{H}_{t-1} \sim \mathcal{N}(\mu_t, \sigma_t^2)$$

where $(\mu_t)$ is the predicted mean and $(\sigma_t^2)$ represents the uncertainty in the forecast.

Probabilistic forecasting can be handled through various approaches, including variance forecasters, quantile forecasters, interval forecasters or distribution forecasters, each capturing uncertainty differently.

For example, you can try to forecast the target location by a gaussian density function (or a mixture), thus the model output follows the form:

{
   "density": {
      "name": "normal",
      "params": {"loc": y_mean, "scale": y_var}
   },
   "weight": weight
}

A mixture density, such as the gaussion mixture $\sum_{i=1}^{K} w_i \mathcal{N}(Y_t | \mu_i, \sigma_i^2)$ allows for capturing multi-modal distributions and approximate more complex distributions.

Additional Resources

• Literature
• Useful Python packages