feat: add Kalshi as a market source

Makefile

@@ -118,7 +118,7 @@ all: deploy
 
 deploy: orchestration questions metadata resolve leaderboards curate-questions website baselines
 
-questions: manifold metaculus acled infer yfinance polymarket wikipedia fred dbnomics
+questions: manifold metaculus acled infer kalshi yfinance polymarket wikipedia fred dbnomics
 
 orchestration: nightly-worker-job nightly-manager-job compress_buckets
 
@@ -170,6 +170,14 @@ acled-fetch:
 acled-update-questions:
 	$(MAKE) -C src/questions/acled/update_questions || echo "* $@" >> $(MAKE_FAILURE_LOG)
 
+kalshi: kalshi-fetch kalshi-update-questions
+
+kalshi-fetch:
+	$(MAKE) -C src/orchestration/func_kalshi_fetch || echo "* $@" >> $(MAKE_FAILURE_LOG)
+
+kalshi-update-questions:
+	$(MAKE) -C src/orchestration/func_kalshi_update || echo "* $@" >> $(MAKE_FAILURE_LOG)
+
 yfinance: yfinance-fetch yfinance-update-questions
 
 yfinance-fetch:

src/_schemas.py

@@ -138,6 +138,20 @@ class Config:
         coerce = True
 
 
+class KalshiFetchFrame(pa.DataFrameModel):
+    """Output of KalshiSource.fetch(). Market and parent routing tickers."""
+
+    id: Series[str]
+    event_ticker: Series[str]
+    series_ticker: Series[str]
+
+    class Config:
+        """Schema configuration."""
+
+        strict = False
+        coerce = True
+
+
 class AcledResolutionFrame(pa.DataFrameModel):
     """ACLED-specific: aggregated events by country and date.
 

src/curate_questions/create_question_set/main.py

@@ -33,6 +33,7 @@
     data_utils,
     decorator,
     env,
+    kalshi,
     question_curation,
 )
 
@@ -1220,6 +1221,8 @@ def driver(_: None) -> None:
             dfq = drop_missing_freeze_datetime(dfq)
             dfq = dfq[dfq["category"] != "Other"]
             dfq = dfq[~dfq["resolved"]]
+            if source == "kalshi":
+                dfq = kalshi.refresh_curation_candidates(dfq)
             dfq = drop_questions_that_resolve_too_soon(source=source, dfq=dfq)
             dfq["source_intro"] = QUESTIONS[source]["source_intro"]
             dfq["resolution_criteria"] = dfq["url"].apply(

src/helpers/kalshi.py

@@ -0,0 +1,75 @@
+"""Kalshi-specific curation helpers and source metadata."""
+
+import json
+import logging
+from urllib.parse import urlencode
+from urllib.request import urlopen
+
+import pandas as pd
+
+from sources._metadata import SOURCE_METADATA
+
+from . import dates
+
+logger = logging.getLogger(__name__)
+
+_KALSHI_API_BASE = "https://api.elections.kalshi.com/trade-api/v2"
+_MARKETS_BATCH_SIZE = 100
+
+SOURCE_INTRO = SOURCE_METADATA["kalshi"]["source_intro"]
+RESOLUTION_CRITERIA = SOURCE_METADATA["kalshi"]["resolution_criteria"]
+
+
+def earliest_curation_datetime(market: dict) -> str:
+    """Return the earliest structured time when a market may cease to be forecastable."""
+    candidates = [market["close_time"].strip(), market["expected_expiration_time"].strip()]
+    occurrence_datetime = market.get("occurrence_datetime")
+    if occurrence_datetime is not None and occurrence_datetime.strip():
+        candidates.append(occurrence_datetime.strip())
+    return min(candidates, key=dates.convert_zulu_to_datetime)
+
+
+def _get_markets_by_ticker(tickers: list[str]) -> list[dict]:
+    """Fetch current market records for a bounded batch of tickers."""
+    query = urlencode({"tickers": ",".join(tickers), "limit": _MARKETS_BATCH_SIZE})
+    endpoint = f"{_KALSHI_API_BASE}/markets?{query}"
+    with urlopen(endpoint, timeout=10) as response:
+        return json.load(response)["markets"]
+
+
+def refresh_curation_candidates(dfq: pd.DataFrame) -> pd.DataFrame:
+    """Keep active Kalshi candidates and refresh their structured curation cutoffs."""
+    if dfq.empty:
+        return dfq.copy()
+
+    candidate_tickers = list(dict.fromkeys(dfq["id"].astype(str)))
+    markets_by_ticker: dict[str, dict] = {}
+    for start in range(0, len(candidate_tickers), _MARKETS_BATCH_SIZE):
+        batch = candidate_tickers[start : start + _MARKETS_BATCH_SIZE]
+        markets_by_ticker.update(
+            {market["ticker"]: market for market in _get_markets_by_ticker(batch)}
+        )
+
+    missing_tickers = set(candidate_tickers) - set(markets_by_ticker)
+    non_active_tickers = {
+        ticker for ticker, market in markets_by_ticker.items() if market["status"] != "active"
+    }
+    if missing_tickers:
+        logger.warning(
+            f"Dropping {len(missing_tickers)} Kalshi curation candidate(s) absent from the "
+            "current markets endpoint."
+        )
+    if non_active_tickers:
+        logger.info(
+            f"Dropping {len(non_active_tickers)} Kalshi curation candidate(s) that are no longer "
+            "active."
+        )
+
+    active_tickers = set(markets_by_ticker) - non_active_tickers
+    refreshed = dfq[dfq["id"].astype(str).isin(active_tickers)].copy()
+    for index, row in refreshed.iterrows():
+        market = markets_by_ticker[str(row["id"])]
+        refreshed.at[index, "market_info_close_datetime"] = dates.convert_zulu_to_iso(
+            earliest_curation_datetime(market)
+        )
+    return refreshed

src/helpers/metadata_llm.py

@@ -3,10 +3,7 @@
 from functools import cache
 
 from utils.llm import model_runs
-from utils.llm.model_registry import (
-    configure_api_keys,
-    validate_provider_keys,
-)
+from utils.llm.model_registry import configure_api_keys, validate_provider_keys
 
 from .openai_safety import get_openai_safety_identifier
 

src/helpers/question_curation.py

@@ -9,6 +9,7 @@
     dates,
     dbnomics,
     fred,
+    kalshi,
     manifold,
     metaculus,
     polymarket,
@@ -40,6 +41,11 @@
         "source_intro": metaculus.SOURCE_INTRO,
         "resolution_criteria": metaculus.RESOLUTION_CRITERIA,
     },
+    "kalshi": {
+        "name": "Kalshi",
+        "source_intro": kalshi.SOURCE_INTRO,
+        "resolution_criteria": kalshi.RESOLUTION_CRITERIA,
+    },
     "polymarket": {
         "name": "Polymarket",
         "source_intro": polymarket.SOURCE_INTRO,

src/orchestration/func_kalshi_fetch/Makefile

@@ -0,0 +1,37 @@
+all :
+	$(MAKE) clean
+	$(MAKE) deploy
+
+.PHONY : all clean deploy
+
+UPLOAD_DIR = upload
+ROOT_DIR ?= $(abspath ../../..)/
+include $(ROOT_DIR)orchestration_upload.mk
+
+.gcloudignore:
+	cp -r $(ROOT_DIR)src/helpers/.gcloudignore .
+
+Dockerfile: $(ROOT_DIR)src/helpers/Dockerfile.template
+	sed \
+		-e 's/REGION/$(CLOUD_DEPLOY_REGION)/g' \
+		-e 's/STACK/google-22-full/g' \
+		-e 's/PYTHON_VERSION/python312/g' \
+		$< > Dockerfile
+
+deploy : main.py .gcloudignore requirements.txt Dockerfile
+	$(stage-orchestration-upload)
+	gcloud run jobs deploy \
+		func-data-kalshi-fetch \
+		--project $(CLOUD_PROJECT) \
+		--region $(CLOUD_DEPLOY_REGION) \
+		--tasks 1 \
+		--parallelism 1 \
+		--task-timeout 1800s \
+		--memory 1Gi \
+		--max-retries 0 \
+		--service-account $(QUESTION_BANK_BUCKET_SERVICE_ACCOUNT) \
+		--set-env-vars $(DEFAULT_CLOUD_FUNCTION_ENV_VARS) \
+		--source $(UPLOAD_DIR)
+
+clean :
+	rm -rf $(UPLOAD_DIR) .gcloudignore Dockerfile

src/orchestration/func_kalshi_fetch/main.py

@@ -0,0 +1,28 @@
+"""Kalshi fetch entry point."""
+
+import logging
+from typing import Any
+
+from helpers import decorator
+from orchestration import _source_io
+from sources.kalshi import KalshiSource
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+SOURCE = "kalshi"
+
+
+@decorator.log_runtime
+def driver(_: Any) -> None:
+    """Fetch Kalshi market tickers and upload to question bank."""
+    source = KalshiSource()
+
+    dff = source.fetch()
+
+    _source_io.write_fetch_output(SOURCE, dff)
+    logger.info("Done.")
+
+
+if __name__ == "__main__":
+    driver(None)

src/orchestration/func_kalshi_fetch/requirements.txt

@@ -0,0 +1,9 @@
+google-cloud-storage
+google-cloud-secret-manager
+pandas>=2.2.2,<3.0
+pandera
+scipy
+requests
+certifi
+backoff
+numpy

src/orchestration/func_kalshi_update/Makefile

@@ -0,0 +1,37 @@
+all :
+	$(MAKE) clean
+	$(MAKE) deploy
+
+.PHONY : all clean deploy
+
+UPLOAD_DIR = upload
+ROOT_DIR ?= $(abspath ../../..)/
+include $(ROOT_DIR)orchestration_upload.mk
+
+.gcloudignore:
+	cp -r $(ROOT_DIR)src/helpers/.gcloudignore .
+
+Dockerfile: $(ROOT_DIR)src/helpers/Dockerfile.template
+	sed \
+		-e 's/REGION/$(CLOUD_DEPLOY_REGION)/g' \
+		-e 's/STACK/google-22-full/g' \
+		-e 's/PYTHON_VERSION/python312/g' \
+		$< > Dockerfile
+
+deploy : main.py .gcloudignore requirements.txt Dockerfile
+	$(stage-orchestration-upload)
+	gcloud run jobs deploy \
+		func-data-kalshi-update-questions \
+		--project $(CLOUD_PROJECT) \
+		--region $(CLOUD_DEPLOY_REGION) \
+		--tasks 1 \
+		--parallelism 1 \
+		--task-timeout 3h \
+		--memory 4Gi \
+		--max-retries 0 \
+		--service-account $(QUESTION_BANK_BUCKET_SERVICE_ACCOUNT) \
+		--set-env-vars $(DEFAULT_CLOUD_FUNCTION_ENV_VARS) \
+		--source $(UPLOAD_DIR)
+
+clean :
+	rm -rf $(UPLOAD_DIR) .gcloudignore Dockerfile

src/orchestration/func_kalshi_update/main.py

@@ -0,0 +1,52 @@
+"""Kalshi update entry point."""
+
+import logging
+from typing import Any
+
+from helpers import data_utils, decorator
+from orchestration import _source_io
+from sources.kalshi import KalshiSource
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+SOURCE = "kalshi"
+
+
+@decorator.log_runtime
+def driver(_: Any) -> None:
+    """Update Kalshi questions and resolution files."""
+    source = KalshiSource()
+
+    dfq, dff = data_utils.get_data_from_cloud_storage(
+        SOURCE, return_question_data=True, return_fetch_data=True
+    )
+
+    logger.info("Loading existing resolution files...")
+    nullified_ids = source.get_nullified_ids()
+    unresolved_mask = ~dfq["resolved"] & ~dfq["id"].astype(str).isin(nullified_ids)
+    unresolved_ids = dfq.loc[unresolved_mask, "id"].astype(str).tolist()
+    existing_resolution_files = _source_io.load_existing_resolution_files(
+        SOURCE, ids=unresolved_ids
+    )
+    logger.info(f"Loaded {len(existing_resolution_files)} resolution files")
+
+    existing_resolution_ids = _source_io.list_existing_resolution_ids(SOURCE)
+
+    result = source.update(
+        dfq,
+        dff,
+        existing_resolution_files=existing_resolution_files,
+        existing_resolution_ids=existing_resolution_ids,
+    )
+
+    logger.info("Uploading to GCP...")
+    data_utils.upload_questions(result.dfq, SOURCE)
+    if result.resolution_files:
+        _source_io.upload_resolution_files(SOURCE, result.resolution_files)
+
+    logger.info("Done.")
+
+
+if __name__ == "__main__":
+    driver(None)

src/orchestration/func_kalshi_update/requirements.txt

@@ -0,0 +1,9 @@
+google-cloud-storage
+google-cloud-secret-manager
+pandas>=2.2.2,<3.0
+pandera
+scipy
+requests
+certifi
+backoff
+numpy

src/sources/_metadata.py

@@ -120,6 +120,19 @@
         # implemented in a second pass.
         "run_fetch": False,
     },
+    "kalshi": {
+        "source_type": SourceType.MARKET,
+        "source_intro": (
+            "We would like you to predict the outcome of a prediction market. A prediction "
+            "market, in this context, is the aggregate of predictions submitted by users on the "
+            "website Kalshi. You're going to predict the probability that the market will "
+            "resolve as 'Yes'."
+        ),
+        "resolution_criteria": "Resolves to the outcome of the question found at {url}.",
+        # Add a ticker only after confirming that it remains absent from both Kalshi's live and
+        # historical APIs. A live 404 alone usually means the market has been archived.
+        "nullified_questions": [],
+    },
     "manifold": {
         "source_type": SourceType.MARKET,
         "source_intro": (