diff --git a/Makefile b/Makefile index f4dc7601..bae0330b 100644 --- a/Makefile +++ b/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: diff --git a/src/_schemas.py b/src/_schemas.py index e380d597..b6ed436b 100644 --- a/src/_schemas.py +++ b/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. diff --git a/src/curate_questions/create_question_set/main.py b/src/curate_questions/create_question_set/main.py index eb2b9f2c..82c675a6 100644 --- a/src/curate_questions/create_question_set/main.py +++ b/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( diff --git a/src/helpers/kalshi.py b/src/helpers/kalshi.py new file mode 100644 index 00000000..17dc3788 --- /dev/null +++ b/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 diff --git a/src/helpers/metadata_llm.py b/src/helpers/metadata_llm.py index 7eb6f3da..91035c83 100644 --- a/src/helpers/metadata_llm.py +++ b/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 diff --git a/src/helpers/question_curation.py b/src/helpers/question_curation.py index 93fedd88..82e3c11f 100644 --- a/src/helpers/question_curation.py +++ b/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, diff --git a/src/orchestration/func_kalshi_fetch/Makefile b/src/orchestration/func_kalshi_fetch/Makefile new file mode 100644 index 00000000..01a9c9bc --- /dev/null +++ b/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 diff --git a/src/orchestration/func_kalshi_fetch/main.py b/src/orchestration/func_kalshi_fetch/main.py new file mode 100644 index 00000000..3b09df3a --- /dev/null +++ b/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) diff --git a/src/orchestration/func_kalshi_fetch/requirements.txt b/src/orchestration/func_kalshi_fetch/requirements.txt new file mode 100644 index 00000000..980fca32 --- /dev/null +++ b/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 diff --git a/src/orchestration/func_kalshi_update/Makefile b/src/orchestration/func_kalshi_update/Makefile new file mode 100644 index 00000000..9b47f7c6 --- /dev/null +++ b/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 diff --git a/src/orchestration/func_kalshi_update/main.py b/src/orchestration/func_kalshi_update/main.py new file mode 100644 index 00000000..663abf43 --- /dev/null +++ b/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) diff --git a/src/orchestration/func_kalshi_update/requirements.txt b/src/orchestration/func_kalshi_update/requirements.txt new file mode 100644 index 00000000..980fca32 --- /dev/null +++ b/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 diff --git a/src/sources/_metadata.py b/src/sources/_metadata.py index 0a8bfc92..54e79293 100644 --- a/src/sources/_metadata.py +++ b/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": ( diff --git a/src/sources/kalshi.py b/src/sources/kalshi.py new file mode 100644 index 00000000..44aa39d7 --- /dev/null +++ b/src/sources/kalshi.py @@ -0,0 +1,855 @@ +"""Kalshi question source.""" + +import logging +import time +from collections import Counter +from datetime import date, timedelta +from typing import Any, ClassVar, TypedDict + +import backoff +import certifi +import numpy as np +import pandas as pd +import pandera.pandas as pa +import requests +from pandera.typing import DataFrame + +from _fb_types import UpdateResult +from _schemas import KalshiFetchFrame, QuestionFrame, ResolutionFrame +from helpers import constants, data_utils, dates +from helpers import kalshi as kalshi_helpers +from helpers import question_curation + +from ._market import MarketSource + +logger = logging.getLogger(__name__) + +_KALSHI_API_BASE = "https://api.elections.kalshi.com/trade-api/v2" + +# Liquidity floors. Any binary market that clears these and resolves within the window qualifies, +# regardless of its event category (Kalshi exposes ~16 categories). The thresholds are calibrated +# so the all-category pool approximates Polymarket's active question count. +_MIN_VOLUME = 10_000 +_MIN_OPEN_INTEREST = 1000 +_MAX_RESOLUTION_DATE_IN_DAYS = 365 * 2 +_QUESTION_LIMIT = 5000 + +# Per-category cap applied in fetch(). Kalshi's liquid universe is dominated by a few high-volume +# categories (notably Sports, ~50% of the pool), so without a cap the pool would be flooded by one +# category. Capping each category keeps the discovered pool representative across all categories +# while smaller categories are taken in full. +_MAX_PER_CATEGORY = 1200 + +# Kalshi's basic read tier currently permits 20 ordinary GET requests per second. Keep a +# conservative 10 requests/second ceiling so retries and small provider-side timing differences +# have headroom. +_MIN_REQUEST_INTERVAL = 0.1 +_CANDLESTICK_PERIOD_INTERVAL = 60 # hourly candlesticks for UTC day-end values +_MAX_CANDLESTICKS_PER_REQUEST = 5000 +# Kalshi counts both ends of the requested range. Keep each window one period below the limit, +# then advance by one second so consecutive windows cannot omit or duplicate a boundary. +_MAX_CANDLESTICK_RANGE_SECONDS = ( + (_MAX_CANDLESTICKS_PER_REQUEST - 1) * _CANDLESTICK_PERIOD_INTERVAL * 60 +) +_RESOLVED_STATUSES = {"finalized"} + + +class _DiscoveredMarket(TypedDict): + """Kalshi fields retained from event discovery.""" + + category: str + event_ticker: str + series_ticker: str + + +class MarketNotFoundError(Exception): + """Raised when a Kalshi market is absent from both live and historical APIs. + + Kalshi documents short-lived 404s while newly created markets propagate. API methods therefore + retry this exception before ``update()`` treats the market as unavailable for the current run. + """ + + def __init__(self, ticker: str): + """Initialize the error with the ticker that could not be found.""" + self.ticker = ticker + super().__init__(f"Kalshi market not found for ticker {ticker}.") + + +class KalshiSource(MarketSource): + """Kalshi prediction market source.""" + + name: ClassVar[str] = "kalshi" + + def __init__(self) -> None: + """Initialize request-throttling state.""" + super().__init__() + self._last_request_time = 0.0 + + # ------------------------------------------------------------------ + # Public: fetch + # ------------------------------------------------------------------ + + @pa.check_types + def fetch( + self, + *, + today: date | None = None, + **kwargs: Any, + ) -> DataFrame[KalshiFetchFrame]: + """Discover eligible Kalshi market tickers via the events endpoint. + + Paginates open events, keeping every liquid binary market that resolves within the window + from the freeze window to ``_MAX_RESOLUTION_DATE_IN_DAYS`` out, in any category. The + discovered pool is then balanced across categories (``_balance_categories``) so a few + high-volume categories do not dominate. The upper date bound keeps the pool to markets + that actually resolve, rather than the perpetual novelty markets (closing decades out) that + otherwise clear the liquidity floors on cumulative volume. + + Args: + today (date | None): Reference date for the min/max resolution dates. Defaults to + today, computed once here and threaded through so every page shares the same + reference instead of each recomputing "today". + """ + if today is None: + today = dates.get_date_today() + min_resolution_date = today + timedelta(days=question_curation.FREEZE_WINDOW_IN_DAYS) + max_resolution_date = today + timedelta(days=_MAX_RESOLUTION_DATE_IN_DAYS) + discovered_markets = self._search_markets( + min_resolution_date=min_resolution_date, + max_resolution_date=max_resolution_date, + ) + logger.info( + f"Discovered {len(discovered_markets)} candidate market tickers across all categories." + ) + ids = self._balance_categories( + {ticker: metadata["category"] for ticker, metadata in discovered_markets.items()} + ) + logger.info( + f"Kept {len(ids)} tickers after applying the per-category cap of {_MAX_PER_CATEGORY}." + ) + rows = [ + { + "id": ticker, + "event_ticker": discovered_markets[ticker]["event_ticker"], + "series_ticker": discovered_markets[ticker]["series_ticker"], + } + for ticker in sorted(ids) + ] + return pd.DataFrame(rows, columns=["id", "event_ticker", "series_ticker"]) + + # ------------------------------------------------------------------ + # Public: update + # ------------------------------------------------------------------ + + @pa.check_types + def update( + self, + dfq: DataFrame[QuestionFrame], + dff: DataFrame[KalshiFetchFrame], + *, + existing_resolution_files: dict[str, pd.DataFrame] | None = None, + existing_resolution_ids: set[str] | None = None, + ) -> UpdateResult: + """Process fetched tickers into updated questions and resolution files. + + For each new ticker in dff, appends to dfq. Then for each unresolved question, fetches + market details and builds/updates resolution files. Finally regenerates missing resolution + files for resolved questions. + + Args: + dfq (DataFrame[QuestionFrame]): Existing questions. + dff (DataFrame[KalshiFetchFrame]): Freshly fetched market tickers. + existing_resolution_files (dict | None): Per-question existing resolution data. + existing_resolution_ids (set[str] | None): Bare IDs that already have a resolution + file in storage. + """ + existing_resolution_files = existing_resolution_files or {} + existing_resolution_ids = existing_resolution_ids or set() + persisted_dfq = dfq.copy(deep=True) + persisted_ids = set(persisted_dfq["id"].astype(str)) + nullified_ids = self.get_nullified_ids() + if nullified_ids: + nullified_mask = dfq["id"].astype(str).isin(nullified_ids) + dfq.loc[nullified_mask, "resolved"] = True + dfq.loc[nullified_mask, "freeze_datetime_value"] = "N/A" + resolution_files: dict[str, pd.DataFrame] = {} + not_found_ids: set[str] = set() + update_datetime = dates.get_datetime_today() + candlesticks_end_ts = int(update_datetime.timestamp()) + yesterday = update_datetime.date() - timedelta(days=1) + routing_by_id = {str(row["id"]): row for row in dff.to_dict(orient="records")} + + # --- Append new tickers from dff to dfq (capped to keep the pool bounded) --- + newly_added_ids: set[str] = set() + new_ids = dff[~dff["id"].isin(dfq["id"]) & ~dff["id"].astype(str).isin(nullified_ids)]["id"] + if not new_ids.empty: + df_new = pd.DataFrame({"id": new_ids}).assign( + **{col: None for col in dfq.columns if col != "id"} + ) + df_new["resolved"] = False + df_new["freeze_datetime_value_explanation"] = "The market price." + df_new["market_info_resolution_datetime"] = "N/A" + + # Cap new additions so the unresolved pool stays under _QUESTION_LIMIT + max_to_add = _QUESTION_LIMIT - len(dfq[dfq["resolved"] == False]) # noqa: E712 + if max_to_add > 0: + # Random sample (not head()) when the cap binds, so the alphabetically-first + # tickers aren't systematically favoured (fetch() sorts ids). fetch() already + # balances categories, so a uniform sample here preserves that balance. + if len(df_new) > max_to_add: + df_new = df_new.sample(n=max_to_add) + # Track which tickers are brand-new this run: the append above seeds them with None + # placeholders that the loop below fills in. Any that 404 before being populated must + # be dropped rather than persisted (see the cleanup after the loops). + newly_added_ids = set(df_new["id"].astype(str)) + dfq = pd.concat([dfq, df_new], ignore_index=True) + + # Fetch market details once, then use sibling title counts to decide whether a child label + # is needed to distinguish questions within the same event. + dfq["resolved"] = dfq["resolved"].astype(bool) + unresolved_rows = list(dfq[~dfq["resolved"]].iterrows()) + market_details: dict[str, dict] = {} + for _index, row in unresolved_rows: + question_id = str(row["id"]) + try: + market_details[question_id] = self._get_market(question_id) + except MarketNotFoundError: + not_found_ids.add(question_id) + + title_counts = Counter( + (market["event_ticker"], market["title"].strip().casefold()) + for market in market_details.values() + ) + + # --- Update all unresolved questions --- + for index, row in unresolved_rows: + question_id = str(row["id"]) + market = market_details.get(question_id) + if market is None: + continue + + resolution_window = self._resolution_window(market) + if resolution_window is None: + raise ValueError(f"Invalid Kalshi resolution window for {row['id']}.") + earliest_resolution_time, _ = resolution_window + + # Assign market details to dfq row + question_key = (market["event_ticker"], market["title"].strip().casefold()) + dfq.at[index, "question"] = self._question_text( + market, include_yes_label=title_counts[question_key] > 1 + ) + dfq.at[index, "background"] = "N/A" + dfq.at[index, "market_info_resolution_criteria"] = self._resolution_criteria(market) + dfq.at[index, "market_info_open_datetime"] = dates.convert_zulu_to_iso( + market["open_time"] + ) + dfq.at[index, "market_info_close_datetime"] = dates.convert_zulu_to_iso( + earliest_resolution_time + ) + routing = routing_by_id.get(str(row["id"])) + if routing is not None: + dfq.at[index, "url"] = self._market_url( + routing["series_ticker"], routing["event_ticker"] + ) + if self._is_resolved(market): + dfq.at[index, "resolved"] = True + dfq.at[index, "market_info_resolution_datetime"] = self._resolution_datetime(market) + dfq.at[index, "forecast_horizons"] = "N/A" + + # Build resolution file + existing_df = existing_resolution_files.get(row["id"]) + try: + df_res = self._build_resolution_df( + market=market, + market_info_resolution_datetime=dfq.at[ + index, "market_info_resolution_datetime" + ], + candlesticks_end_ts=candlesticks_end_ts, + yesterday=yesterday, + existing_df=existing_df, + ) + except MarketNotFoundError: + not_found_ids.add(str(row["id"])) + continue + if df_res is not None: + dfq.at[index, "freeze_datetime_value"] = df_res["value"].iloc[-1] + # if rebuilt, then write; else - skip + if df_res is not existing_df: + logger.info(f"Rebuilt, will write - id={row['id']}") + resolution_files[row["id"]] = df_res + else: + logger.info(f"Skipped writing to resolution files, not changed -id={row['id']}") + else: + logger.warning( + f"No resolution file built for id={row['id']} " + "(no candlesticks / no usable price data)." + ) + + # --- Regenerate missing resolution files for resolved questions --- + for _index, row in dfq[dfq["resolved"]].iterrows(): + question_id = str(row["id"]) + if question_id in nullified_ids: + continue + if question_id not in existing_resolution_ids and row["id"] not in resolution_files: + try: + market = self._get_market(row["id"]) + except MarketNotFoundError: + not_found_ids.add(question_id) + continue + try: + df_res = self._build_resolution_df( + market=market, + market_info_resolution_datetime=row["market_info_resolution_datetime"], + candlesticks_end_ts=candlesticks_end_ts, + yesterday=yesterday, + existing_df=None, + ) + except MarketNotFoundError: + not_found_ids.add(question_id) + continue + if df_res is not None: + resolution_files[row["id"]] = df_res + else: + logger.warning( + f"No resolution file built for resolved id={row['id']} " + "(no candlesticks / no usable price data)." + ) + + # Drop brand-new tickers that were absent from both APIs before they were ever populated. + # Persisted questions are restored after any partial refresh and quarantined from curation + # with an N/A price; because they remain unresolved, a later nightly run can recover them. + orphan_ids = newly_added_ids.intersection(not_found_ids) + if orphan_ids: + dfq = dfq[~dfq["id"].isin(orphan_ids)].reset_index(drop=True) + + persisted_not_found_ids = persisted_ids.intersection(not_found_ids) + for question_id in persisted_not_found_ids: + current_index = dfq.index[dfq["id"].astype(str) == question_id][0] + persisted_index = persisted_dfq.index[persisted_dfq["id"].astype(str) == question_id][0] + dfq.loc[current_index, :] = persisted_dfq.loc[persisted_index, :] + dfq.at[current_index, "freeze_datetime_value"] = "N/A" + + if not_found_ids: + logger.warning( + f"{len(not_found_ids)} question(s) were absent from both live and historical APIs " + f"after retries: {sorted(not_found_ids)}. Dropped {len(orphan_ids)} never-populated " + f"new ticker(s) and quarantined {len(persisted_not_found_ids)} persisted " + "question(s) with an N/A price." + ) + + return UpdateResult( + dfq=dfq, + resolution_files=resolution_files, + ) + + # ------------------------------------------------------------------ + # Private: request throttling + # ------------------------------------------------------------------ + + def _throttle(self) -> None: + """Sleep if needed to keep consecutive Kalshi requests under the read limit.""" + elapsed = time.monotonic() - self._last_request_time + if elapsed < _MIN_REQUEST_INTERVAL: + time.sleep(_MIN_REQUEST_INTERVAL - elapsed) + self._last_request_time = time.monotonic() + + # ------------------------------------------------------------------ + # Private: events (search) API + # ------------------------------------------------------------------ + + @backoff.on_exception( + backoff.expo, + requests.exceptions.RequestException, + max_time=500, + on_backoff=data_utils.print_error_info_handler, + ) + def _call_search_endpoint( + self, + *, + min_resolution_date: date, + max_resolution_date: date | None = None, + cursor: str | None = None, + ) -> tuple[dict[str, _DiscoveredMarket], str | None]: + """Fetch one page of open events (with nested markets) and return qualifying tickers. + + Returns each qualifying market's parent category and routing identifiers so fetch() can + balance the pool and retain enough information to build its public Kalshi URL. + """ + endpoint = f"{_KALSHI_API_BASE}/events" + params: dict[str, Any] = { + "status": "open", + "with_nested_markets": "true", + "limit": 200, + } + if cursor: + params["cursor"] = cursor + + self._throttle() + response = requests.get(endpoint, params=params, verify=certifi.where()) + if not response.ok: + logger.error( + f"Request to endpoint failed for {endpoint}: {response.status_code} Error. " + f"{response.text}" + ) + response.raise_for_status() + + data = response.json() + try: + discovered_markets: dict[str, _DiscoveredMarket] = {} + for event in data["events"]: + category = event.get("category") or "Uncategorized" + event_ticker = event["event_ticker"] + series_ticker = event["series_ticker"] + for market in event["markets"]: + if market["event_ticker"] != event_ticker: + logger.warning( + f"Skipping Kalshi market {market['ticker']} because its event_ticker " + f"does not match parent event {event_ticker}." + ) + continue + if self._market_qualifies( + market, + min_resolution_date=min_resolution_date, + max_resolution_date=max_resolution_date, + ): + discovered_markets[market["ticker"]] = { + "category": category, + "event_ticker": event_ticker, + "series_ticker": series_ticker, + } + return discovered_markets, data["cursor"] + except KeyError as error: + raise ValueError( + f"Kalshi events API response is missing required field {error.args[0]!r}." + ) from error + + def _search_markets( + self, + *, + min_resolution_date: date, + max_resolution_date: date | None = None, + ) -> dict[str, _DiscoveredMarket]: + """Discover market tickers and parent metadata by paginating all open events.""" + logger.info("Calling Kalshi events endpoint") + discovered_markets: dict[str, _DiscoveredMarket] = {} + cursor: str | None = None + while True: + page, cursor = self._call_search_endpoint( + min_resolution_date=min_resolution_date, + max_resolution_date=max_resolution_date, + cursor=cursor, + ) + discovered_markets.update(page) + if not cursor: + break + return discovered_markets + + @staticmethod + def _balance_categories(ticker_categories: dict[str, str]) -> list[str]: + """Cap each category to ``_MAX_PER_CATEGORY`` tickers, sampling randomly within a category. + + Kalshi's liquid universe is dominated by a few high-volume categories (notably Sports), so + without a cap the pool would be flooded by one category. Capping keeps the pool + representative across all categories while smaller categories are taken in full. Sampling is + random within a category (not by ticker name) so the selection is not alphabetically biased + and rotates across nightly runs. + + Args: + ticker_categories (dict[str, str]): Discovered market tickers mapped to their event + category. + """ + if not ticker_categories: + return [] + df = pd.DataFrame( + {"id": list(ticker_categories), "category": list(ticker_categories.values())} + ) + df["category"] = df["category"].fillna("Uncategorized") + kept = [ + group.sample(n=_MAX_PER_CATEGORY) if len(group) > _MAX_PER_CATEGORY else group + for _, group in df.groupby("category") + ] + return pd.concat(kept, ignore_index=True)["id"].tolist() + + @staticmethod + def _market_qualifies( + market: dict, + *, + min_resolution_date: date, + max_resolution_date: date | None = None, + ) -> bool: + """Return True if a market is a liquid binary market resolving within the target window. + + A market qualifies when it is active, binary, sufficiently liquid (volume and open + interest), its earliest credible resolution is on or after ``min_resolution_date``, and + its latest resolution bound is (when set) no later than ``max_resolution_date``. Category + is not a criterion -- every category is eligible and the pool is balanced across + categories afterwards in fetch(). + """ + if market["status"] != "active": + return False + if market["market_type"] != "binary": + return False + if float(market["volume_fp"]) < _MIN_VOLUME: + return False + if float(market["open_interest_fp"]) < _MIN_OPEN_INTEREST: + return False + resolution_window = KalshiSource._resolution_window(market) + if resolution_window is None: + return False + earliest_resolution_time, latest_resolution_time = resolution_window + earliest_resolution_date = dates.convert_zulu_to_datetime(earliest_resolution_time).date() + latest_resolution_date = dates.convert_zulu_to_datetime(latest_resolution_time).date() + if earliest_resolution_date < min_resolution_date: + return False + if max_resolution_date is not None and latest_resolution_date > max_resolution_date: + return False + return True + + # ------------------------------------------------------------------ + # Private: market detail API + # ------------------------------------------------------------------ + + @backoff.on_exception( + backoff.expo, + MarketNotFoundError, + max_tries=3, + on_backoff=data_utils.print_error_info_handler, + ) + @backoff.on_exception( + backoff.expo, + requests.exceptions.RequestException, + max_time=200, + max_tries=10, + factor=2, + base=2, + on_backoff=data_utils.print_error_info_handler, + ) + def _get_market(self, ticker: str) -> dict: + """Fetch market details from the live API, falling back to historical storage.""" + logger.info(f"Calling market endpoint for {ticker}") + endpoint = f"{_KALSHI_API_BASE}/markets/{ticker}" + self._throttle() + response = requests.get(endpoint, verify=certifi.where()) + if response.status_code == 404: + endpoint = f"{_KALSHI_API_BASE}/historical/markets/{ticker}" + self._throttle() + response = requests.get(endpoint, verify=certifi.where()) + if response.status_code == 404: + logger.warning(f"Market {ticker} was absent from both live and historical APIs.") + raise MarketNotFoundError(ticker) + if not response.ok: + logger.error(f"Request to market endpoint failed for {ticker}.") + response.raise_for_status() + return response.json()["market"] + + # ------------------------------------------------------------------ + # Private: candlesticks API + # ------------------------------------------------------------------ + + @backoff.on_exception( + backoff.expo, + MarketNotFoundError, + max_tries=3, + on_backoff=data_utils.print_error_info_handler, + ) + @backoff.on_exception( + backoff.expo, + requests.exceptions.RequestException, + max_time=200, + max_tries=10, + factor=2, + base=2, + on_backoff=data_utils.print_error_info_handler, + ) + def _get_market_candlesticks( + self, + ticker: str, + *, + start_ts: int, + end_ts: int, + ) -> list[dict]: + """Fetch hourly candles in bounded windows, with historical fallback.""" + logger.info(f"Calling candlesticks endpoint for {ticker}") + series_ticker = self._series_ticker(ticker) + historical = False + candles_by_end_ts: dict[int, dict] = {} + window_start = start_ts + + while window_start <= end_ts: + window_end = min(window_start + _MAX_CANDLESTICK_RANGE_SECONDS, end_ts) + if historical: + endpoint = f"{_KALSHI_API_BASE}/historical/markets/{ticker}/candlesticks" + else: + endpoint = ( + f"{_KALSHI_API_BASE}/series/{series_ticker}/markets/{ticker}/candlesticks" + ) + params: dict[str, Any] = { + "start_ts": window_start, + "end_ts": window_end, + "period_interval": _CANDLESTICK_PERIOD_INTERVAL, + } + self._throttle() + response = requests.get(endpoint, params=params, verify=certifi.where()) + if not historical and response.status_code == 404: + historical = True + endpoint = f"{_KALSHI_API_BASE}/historical/markets/{ticker}/candlesticks" + self._throttle() + response = requests.get(endpoint, params=params, verify=certifi.where()) + if response.status_code == 404: + logger.warning( + f"Candlesticks for {ticker} were absent from both live and historical APIs." + ) + raise MarketNotFoundError(ticker) + if not response.ok: + logger.error(f"Request to candlesticks endpoint failed for {ticker}.") + response.raise_for_status() + for candle in response.json().get("candlesticks", []): + normalized_candle = candle.copy() + normalized_price = candle.get("price", {}).copy() + if "close_dollars" not in normalized_price and "close" in normalized_price: + normalized_price["close_dollars"] = normalized_price["close"] + normalized_candle["price"] = normalized_price + candles_by_end_ts[normalized_candle["end_period_ts"]] = normalized_candle + + if window_end == end_ts: + break + window_start = window_end + 1 + + return [candles_by_end_ts[end_period_ts] for end_period_ts in sorted(candles_by_end_ts)] + + # ------------------------------------------------------------------ + # Private: resolution file building + # ------------------------------------------------------------------ + + def _build_resolution_df( + self, + market: dict, + market_info_resolution_datetime: str, + *, + candlesticks_end_ts: int, + yesterday: date, + existing_df: pd.DataFrame | None = None, + ) -> DataFrame[ResolutionFrame] | None: + """Build or update a resolution file for a single market.""" + ticker = market["ticker"] + resolved = self._is_resolved(market) + existing_last_date: date | None = None + + # --- Already up-to-date check --- + if existing_df is not None and not existing_df.empty: + existing_last_date = pd.to_datetime(existing_df["date"].max()).date() + if resolved: + resolved_date = pd.Timestamp(market_info_resolution_datetime).date() + if existing_last_date >= resolved_date: + # An unresolved run may already have stored a probability on the eventual + # settlement date. Replace it (and any later rows) with the terminal result. + df = existing_df.copy() + df["date"] = pd.to_datetime(df["date"]) + df = df[df["date"].dt.date < resolved_date].reset_index(drop=True) + df.loc[len(df)] = { + "id": ticker, + "date": pd.Timestamp(resolved_date), + "value": self._get_resolved_market_value(market), + } + df = df[["id", "date", "value"]].astype( + dtype=constants.RESOLUTION_FILE_COLUMN_DTYPE + ) + if df.equals(existing_df): + return existing_df + return df + elif existing_last_date >= yesterday: + return existing_df + + # --- Fetch hourly candlesticks and build a UTC end-of-day series --- + candlesticks_start_ts = max( + constants.BENCHMARK_START_DATE_EPOCHTIME, + int(dates.convert_zulu_to_datetime(market["open_time"]).timestamp()), + ) + if existing_last_date is not None: + candlesticks_start_ts = max( + candlesticks_start_ts, + dates.convert_iso_date_to_epoch_time(existing_last_date + timedelta(days=1)), + ) + candles = self._get_market_candlesticks( + ticker, + start_ts=candlesticks_start_ts, + end_ts=candlesticks_end_ts, + ) + df = pd.DataFrame( + [ + { + "datetime": dates.convert_epoch_time_in_sec_to_iso(candle["end_period_ts"]), + "value": float(candle["price"]["close_dollars"]), + } + for candle in candles + if candle.get("price", {}).get("close_dollars") is not None + ] + ) + if not df.empty: + df["datetime"] = pd.to_datetime(df["datetime"], utc=True) + df = df.sort_values(by="datetime") + + # A stored date D represents the last market price at or before D+1 00:00 UTC. Build + # those boundaries explicitly so a later rebuild cannot assign post-deadline trading + # to D. An exact-boundary hourly candle is eligible because it closes the prior hour. + first_boundary = df["datetime"].min().ceil("D") + last_boundary = pd.Timestamp(yesterday + timedelta(days=1), tz="UTC") + if first_boundary <= last_boundary: + boundaries = pd.DataFrame( + {"boundary": pd.date_range(first_boundary, last_boundary, freq="D")} + ) + df = pd.merge_asof( + boundaries, + df, + left_on="boundary", + right_on="datetime", + direction="backward", + allow_exact_matches=True, + ) + df["date"] = df["boundary"].dt.date.apply( + lambda boundary_date: boundary_date - timedelta(days=1) + ) + df = df[["date", "value"]] + else: + df = pd.DataFrame(columns=["date", "value"]) + else: + df = pd.DataFrame(columns=["date", "value"]) + + if existing_df is not None and not existing_df.empty: + existing = existing_df[["date", "value"]].copy() + existing["date"] = pd.to_datetime(existing["date"]).dt.date + df = pd.concat([existing, df], ignore_index=True) + df = ( + df.drop_duplicates(subset="date", keep="last") + .sort_values(by="date") + .reset_index(drop=True) + ) + if df.empty: + return None + + # --- Forward-fill missing dates --- + date_range = pd.date_range(start=df["date"].min(), end=yesterday, freq="D") + if resolved: + resolved_date = pd.Timestamp(market_info_resolution_datetime).date() + df = df[df["date"] < resolved_date] + df.loc[len(df)] = { + "date": resolved_date, + "value": self._get_resolved_market_value(market), + } + date_range = pd.date_range(start=df["date"].min(), end=resolved_date, freq="D") + + df_dates = pd.DataFrame(date_range, columns=["date"]) + df_dates["date"] = df_dates["date"].dt.date + df = pd.merge(left=df_dates, right=df, on="date", how="left") + + if resolved: + # Don't forward-fill last row (could be NaN for a void/ambiguous resolution) + df.iloc[:-1] = df.iloc[:-1].ffill() + else: + df = df.ffill() + + df["id"] = ticker + df["date"] = pd.to_datetime(df["date"]) + df = df[df["date"].dt.date >= constants.BENCHMARK_START_DATE_DATETIME_DATE] + return df[["id", "date", "value"]].astype(dtype=constants.RESOLUTION_FILE_COLUMN_DTYPE) + + # ------------------------------------------------------------------ + # Private: market helpers + # ------------------------------------------------------------------ + + @staticmethod + def _resolution_window(market: dict) -> tuple[str, str] | None: + """Return the credible earliest and latest resolution timestamps. + + Kalshi's close time can be a late postponement bound rather than the time the outcome is + expected to become known. Required timestamp keys are accessed directly so absent API + fields still fail loudly; present but unusable or inconsistent timestamps make the market + ineligible. + """ + timestamps = { + "close": market["close_time"], + "expected": market["expected_expiration_time"], + "latest": market["latest_expiration_time"], + } + occurrence_datetime = market.get("occurrence_datetime") + if not all( + isinstance(timestamp, str) and timestamp.strip() for timestamp in timestamps.values() + ): + return None + timestamps = {name: timestamp.strip() for name, timestamp in timestamps.items()} + if occurrence_datetime is not None: + if not isinstance(occurrence_datetime, str): + return None + if occurrence_datetime.strip(): + timestamps["occurrence"] = occurrence_datetime.strip() + + try: + parsed_datetimes = { + name: dates.convert_zulu_to_datetime(timestamp) + for name, timestamp in timestamps.items() + } + except (TypeError, ValueError): + return None + if any( + parsed_datetime.utcoffset() is None for parsed_datetime in parsed_datetimes.values() + ): + return None + + latest_field = max(("close", "latest"), key=parsed_datetimes.__getitem__) + latest_resolution_datetime = parsed_datetimes[latest_field] + if parsed_datetimes["expected"] > latest_resolution_datetime: + return None + if ( + parsed_datetimes.get("occurrence", latest_resolution_datetime) + > latest_resolution_datetime + ): + return None + + return kalshi_helpers.earliest_curation_datetime(market), timestamps[latest_field] + + @staticmethod + def _get_resolved_market_value(market: dict) -> float: + """Map resolution outcome to numeric value. + + yes -> 1, no -> 0, anything else (scalar, void) -> NaN + """ + return {"yes": 1, "no": 0}.get(market.get("result", ""), np.nan) + + @staticmethod + def _is_resolved(market: dict) -> bool: + """Return True if the market has reached a terminal (resolved) status.""" + return market.get("status") in _RESOLVED_STATUSES + + @staticmethod + def _question_text(market: dict, *, include_yes_label: bool) -> str: + """Add the child Yes label only when sibling market titles repeat.""" + if include_yes_label: + return f'{market["title"]} [Yes: {market["yes_sub_title"]}]' + return market["title"] + + @staticmethod + def _market_url(series_ticker: str, event_ticker: str) -> str: + """Build a public Kalshi URL from structured parent identifiers.""" + return f"https://kalshi.com/markets/{series_ticker.lower()}/x/{event_ticker.lower()}" + + @staticmethod + def _resolution_criteria(market: dict) -> str: + """Join the market's primary and secondary rules into a resolution criteria string.""" + parts = [market.get("rules_primary"), market.get("rules_secondary")] + parts = [part for part in parts if part] + return " ".join(parts) if parts else "N/A" + + @staticmethod + def _resolution_datetime(market: dict) -> str: + """Return the resolution datetime as ISO, preferring settlement over expiration/close.""" + ts = ( + market.get("settlement_ts") + or market.get("expected_expiration_time") + or market["close_time"] + ) + return dates.convert_zulu_to_iso(ts) + + @staticmethod + def _series_ticker(ticker: str) -> str: + """Derive the Kalshi series ticker (the prefix before the first dash).""" + return ticker.split("-")[0] diff --git a/src/sources/registry.py b/src/sources/registry.py index 8cc70e9a..37261656 100644 --- a/src/sources/registry.py +++ b/src/sources/registry.py @@ -11,6 +11,7 @@ from .dbnomics import DbnomicsSource from .fred import FredSource from .infer import InferSource +from .kalshi import KalshiSource from .manifold import ManifoldSource from .metaculus import MetaculusSource from .polymarket import PolymarketSource @@ -22,6 +23,7 @@ _dbnomics = DbnomicsSource() _fred = FredSource() _infer = InferSource() +_kalshi = KalshiSource() _manifold = ManifoldSource() _metaculus = MetaculusSource() _polymarket = PolymarketSource() @@ -35,6 +37,7 @@ _dbnomics, _fred, _infer, + _kalshi, _manifold, _metaculus, _polymarket, diff --git a/src/tests/conftest.py b/src/tests/conftest.py index cbc2f584..04700e8a 100644 --- a/src/tests/conftest.py +++ b/src/tests/conftest.py @@ -9,6 +9,7 @@ from sources.acled import AcledSource from sources.fred import FredSource +from sources.kalshi import KalshiSource from sources.manifold import ManifoldSource from sources.metaculus import MetaculusSource from sources.polymarket import PolymarketSource @@ -85,6 +86,12 @@ def metaculus_source(): return src +@pytest.fixture() +def kalshi_source(): + """Return a KalshiSource instance.""" + return KalshiSource() + + @pytest.fixture() def polymarket_source(): """Return a PolymarketSource instance.""" @@ -347,6 +354,72 @@ def make_metaculus_fetch_df(ids): return pd.DataFrame({"id": [str(i) for i in ids]}) +# --------------------------------------------------------------------------- +# Kalshi-specific factories +# --------------------------------------------------------------------------- + + +def make_kalshi_api_market(**overrides): + """Build a realistic Kalshi market dict as returned by /markets/{ticker}.""" + base = { + "ticker": "KXTEST-001", + "event_ticker": "KXTEST", + "title": "Will X happen by 2026?", + "yes_sub_title": "X happens", + "market_type": "binary", + "open_time": "2025-06-01T00:00:00Z", + "occurrence_datetime": "2026-12-01T00:00:00Z", + "close_time": "2026-12-01T00:00:00Z", + "status": "active", + "result": "", + "volume_fp": "10000.00", + "volume_24h_fp": "500.00", + "open_interest_fp": "2000.00", + "rules_primary": "Resolves Yes if X happens.", + "rules_secondary": "", + "settlement_ts": None, + "expected_expiration_time": "2026-12-01T00:00:00Z", + "latest_expiration_time": "2026-12-08T00:00:00Z", + } + base.update(overrides) + return base + + +def make_kalshi_event(markets=None, category="Economics", **overrides): + """Build a Kalshi event dict with nested markets as returned by /events.""" + base = { + "event_ticker": "KXTEST", + "series_ticker": "KXTEST", + "category": category, + "title": "Test event", + "markets": markets if markets is not None else [make_kalshi_api_market()], + } + base.update(overrides) + return base + + +def make_kalshi_candlestick(end_period_ts, close_dollars=None, **overrides): + """Build a candlestick dict as returned by the candlesticks endpoint. + + ``price`` is empty ({}) when no trade occurred during the period, matching the API. + """ + price = {} if close_dollars is None else {"close_dollars": str(close_dollars)} + base = { + "end_period_ts": end_period_ts, + "price": price, + "volume_fp": "100.00", + "open_interest_fp": "500.00", + } + base.update(overrides) + return base + + +def make_kalshi_fetch_df(rows): + """Build a DataFrame matching KalshiFetchFrame schema.""" + records = [{"event_ticker": "KXTEST", "series_ticker": "KXTEST", **row} for row in rows] + return pd.DataFrame(records, columns=["id", "event_ticker", "series_ticker"]) + + # --------------------------------------------------------------------------- # Polymarket-specific factories # --------------------------------------------------------------------------- diff --git a/src/tests/test_kalshi.py b/src/tests/test_kalshi.py new file mode 100644 index 00000000..30bfc615 --- /dev/null +++ b/src/tests/test_kalshi.py @@ -0,0 +1,1647 @@ +"""Tests for KalshiSource fetch/update logic.""" + +from datetime import date, datetime, timedelta, timezone +from unittest.mock import Mock, patch + +import numpy as np +import pandas as pd +import pytest + +from _fb_types import NullifiedQuestion, SourceQuestionBank +from _schemas import KalshiFetchFrame, QuestionFrame, ResolutionFrame +from curate_questions.create_question_set import main as create_question_set +from helpers import kalshi as kalshi_helpers +from resolve._impute import impute_missing_forecasts +from resolve._prepare import check_and_prepare_forecast_file, set_resolution_dates +from resolve.explode_question_set import explode_question_set +from resolve.resolve_all import resolve_all +from sources.kalshi import KalshiSource, MarketNotFoundError +from sources.registry import SOURCES + +from .conftest import ( + make_kalshi_api_market, + make_kalshi_candlestick, + make_kalshi_event, + make_kalshi_fetch_df, + make_question_df, + make_question_set_df, + make_resolution_df, +) + + +def _ts(year, month, day, hour=0): + """Return the unix timestamp (seconds) for a UTC datetime.""" + return int(datetime(year, month, day, hour, tzinfo=timezone.utc).timestamp()) + + +def _update_boundaries(today: date) -> dict[str, int | date]: + """Return the timestamp and date boundaries pinned at the start of an update.""" + update_datetime = datetime(today.year, today.month, today.day, tzinfo=timezone.utc) + return { + "candlesticks_end_ts": int(update_datetime.timestamp()), + "yesterday": today - timedelta(days=1), + } + + +def _discovered_market( + category: str, + event_ticker: str = "KXTEST", + series_ticker: str = "KXTEST", +) -> dict[str, str]: + """Build retained Kalshi event metadata for fetch tests.""" + return { + "category": category, + "event_ticker": event_ticker, + "series_ticker": series_ticker, + } + + +# --------------------------------------------------------------------------- +# _get_resolved_market_value (pure, no mocking) +# --------------------------------------------------------------------------- + + +class TestGetResolvedMarketValue: + """Tests for KalshiSource._get_resolved_market_value static method.""" + + def test_yes_resolution(self): + """'yes' result returns 1.""" + assert KalshiSource._get_resolved_market_value(make_kalshi_api_market(result="yes")) == 1 + + def test_no_resolution(self): + """'no' result returns 0.""" + assert KalshiSource._get_resolved_market_value(make_kalshi_api_market(result="no")) == 0 + + def test_empty_result_is_nan(self): + """Empty result returns NaN.""" + assert np.isnan(KalshiSource._get_resolved_market_value(make_kalshi_api_market(result=""))) + + def test_scalar_result_is_nan(self): + """A non yes/no result returns NaN.""" + result = KalshiSource._get_resolved_market_value(make_kalshi_api_market(result="scalar")) + assert np.isnan(result) + + +# --------------------------------------------------------------------------- +# _is_resolved / _series_ticker (pure) +# --------------------------------------------------------------------------- + + +class TestMarketHelpers: + """Tests for small Kalshi static helpers.""" + + def test_is_resolved_finalized(self): + """A finalized market is treated as resolved.""" + market = make_kalshi_api_market(status="finalized") + assert KalshiSource._is_resolved(market) is True + + def test_is_resolved_non_terminal(self): + """Markets that can still change are not yet resolved.""" + for status in [ + "initialized", + "active", + "inactive", + "closed", + "determined", + "disputed", + "amended", + ]: + assert KalshiSource._is_resolved(make_kalshi_api_market(status=status)) is False + + def test_series_ticker(self): + """Series ticker is the prefix before the first dash.""" + assert KalshiSource._series_ticker("KXWCSPREAD-26JUN18CANQAT-CAN6") == "KXWCSPREAD" + + def test_resolution_criteria_joins_rules(self): + """Primary and secondary rules are joined, empties dropped.""" + market = make_kalshi_api_market(rules_primary="Primary.", rules_secondary="Secondary.") + assert KalshiSource._resolution_criteria(market) == "Primary. Secondary." + + def test_resolution_criteria_na_when_empty(self): + """No rules yields 'N/A'.""" + market = make_kalshi_api_market(rules_primary="", rules_secondary="") + assert KalshiSource._resolution_criteria(market) == "N/A" + + def test_resolution_datetime_prefers_settlement(self): + """settlement_ts is preferred over expiration/close.""" + market = make_kalshi_api_market(settlement_ts="2026-01-13T05:00:00Z") + assert KalshiSource._resolution_datetime(market).startswith("2026-01-13") + + +# --------------------------------------------------------------------------- +# _build_resolution_df (mock _get_market_candlesticks) +# --------------------------------------------------------------------------- + + +class TestBuildResolutionDf: + """Tests for KalshiSource._build_resolution_df.""" + + @patch.object(KalshiSource, "_get_market_candlesticks") + def test_already_up_to_date(self, mock_candles, kalshi_source, freeze_today): + """Skips API call if existing data covers through yesterday.""" + freeze_today(date(2026, 1, 15)) + existing = make_resolution_df( + [ + {"id": "KXTEST-001", "date": "2024-06-01", "value": 0.5}, + {"id": "KXTEST-001", "date": "2026-01-14", "value": 0.6}, + ] + ) + market = make_kalshi_api_market() + result = kalshi_source._build_resolution_df( + market=market, + market_info_resolution_datetime="N/A", + existing_df=existing, + **_update_boundaries(date(2026, 1, 15)), + ) + + assert result.equals(existing) + mock_candles.assert_not_called() + + @patch.object(KalshiSource, "_get_market_candlesticks") + def test_basic_unresolved_market(self, mock_candles, kalshi_source, freeze_today): + """Builds a valid time series from candlesticks for an unresolved market.""" + freeze_today(date(2026, 1, 15)) + mock_candles.return_value = [ + make_kalshi_candlestick(_ts(2026, 1, 10), close_dollars="0.40"), + make_kalshi_candlestick(_ts(2026, 1, 12), close_dollars="0.60"), + ] + result = kalshi_source._build_resolution_df( + market=make_kalshi_api_market(), + market_info_resolution_datetime="N/A", + existing_df=None, + **_update_boundaries(date(2026, 1, 15)), + ) + + assert result is not None + assert (result["id"] == "KXTEST-001").all() + ResolutionFrame.validate(result) + # Midnight candles close the preceding UTC dates; missing dates are forward-filled. + assert len(result) >= 5 + + @patch.object(KalshiSource, "_get_market_candlesticks") + def test_empty_candles_returns_none(self, mock_candles, kalshi_source, freeze_today): + """No candlesticks returns None.""" + freeze_today(date(2026, 1, 15)) + mock_candles.return_value = [] + result = kalshi_source._build_resolution_df( + market=make_kalshi_api_market(), + market_info_resolution_datetime="N/A", + existing_df=None, + **_update_boundaries(date(2026, 1, 15)), + ) + assert result is None + + @patch.object(KalshiSource, "_get_market_candlesticks") + def test_no_trade_candles_returns_none(self, mock_candles, kalshi_source, freeze_today): + """Candlesticks with no trades (empty price) return None.""" + freeze_today(date(2026, 1, 15)) + mock_candles.return_value = [ + make_kalshi_candlestick(_ts(2026, 1, 10)), + make_kalshi_candlestick(_ts(2026, 1, 12)), + ] + result = kalshi_source._build_resolution_df( + market=make_kalshi_api_market(), + market_info_resolution_datetime="N/A", + existing_df=None, + **_update_boundaries(date(2026, 1, 15)), + ) + assert result is None + + @patch.object(KalshiSource, "_get_market_candlesticks") + def test_forward_fills_gaps(self, mock_candles, kalshi_source, freeze_today): + """Missing dates between candlesticks are forward-filled.""" + freeze_today(date(2026, 1, 15)) + mock_candles.return_value = [ + make_kalshi_candlestick(_ts(2026, 1, 10), close_dollars="0.30"), + make_kalshi_candlestick(_ts(2026, 1, 14), close_dollars="0.80"), + ] + result = kalshi_source._build_resolution_df( + market=make_kalshi_api_market(), + market_info_resolution_datetime="N/A", + existing_df=None, + **_update_boundaries(date(2026, 1, 15)), + ) + + dates_in_df = pd.to_datetime(result["date"]).dt.date.tolist() + assert date(2026, 1, 11) in dates_in_df + assert date(2026, 1, 12) in dates_in_df + assert date(2026, 1, 13) in dates_in_df + + @patch.object(KalshiSource, "_get_market_candlesticks") + def test_resolved_truncates_at_resolution(self, mock_candles, kalshi_source, freeze_today): + """Resolved market: data truncated at resolution date, final row has resolved value.""" + freeze_today(date(2026, 1, 15)) + mock_candles.return_value = [ + make_kalshi_candlestick(_ts(2026, 1, 10), close_dollars="0.40"), + make_kalshi_candlestick(_ts(2026, 1, 12), close_dollars="0.60"), + make_kalshi_candlestick(_ts(2026, 1, 14), close_dollars="0.90"), + ] + market = make_kalshi_api_market(status="finalized", result="yes") + result = kalshi_source._build_resolution_df( + market=market, + market_info_resolution_datetime="2026-01-13T12:00:00+00:00", + existing_df=None, + **_update_boundaries(date(2026, 1, 15)), + ) + + assert result is not None + last_date = pd.to_datetime(result["date"].iloc[-1]).date() + assert last_date == date(2026, 1, 13) + assert float(result["value"].iloc[-1]) == 1.0 + all_dates = pd.to_datetime(result["date"]).dt.date + assert all(d <= date(2026, 1, 13) for d in all_dates) + + @patch.object(KalshiSource, "_get_market_candlesticks") + def test_resolved_void_nan_last_row(self, mock_candles, kalshi_source, freeze_today): + """Void resolution (empty result) on a terminal market: last row is NaN.""" + freeze_today(date(2026, 1, 15)) + mock_candles.return_value = [ + make_kalshi_candlestick(_ts(2026, 1, 10), close_dollars="0.40"), + make_kalshi_candlestick(_ts(2026, 1, 12), close_dollars="0.60"), + ] + market = make_kalshi_api_market(status="finalized", result="") + result = kalshi_source._build_resolution_df( + market=market, + market_info_resolution_datetime="2026-01-13T12:00:00+00:00", + existing_df=None, + **_update_boundaries(date(2026, 1, 15)), + ) + + assert result is not None + assert np.isnan(float(result["value"].iloc[-1])) + + @patch.object(KalshiSource, "_get_market_candlesticks") + def test_filters_future_candles(self, mock_candles, kalshi_source, freeze_today): + """Candlesticks after yesterday's UTC boundary cannot enter the daily series.""" + freeze_today(date(2026, 1, 15)) + mock_candles.return_value = [ + make_kalshi_candlestick(_ts(2026, 1, 15), close_dollars="0.50"), + make_kalshi_candlestick(_ts(2026, 1, 16), close_dollars="0.90"), + ] + result = kalshi_source._build_resolution_df( + market=make_kalshi_api_market(), + market_info_resolution_datetime="N/A", + existing_df=None, + **_update_boundaries(date(2026, 1, 15)), + ) + + assert result is not None + all_dates = pd.to_datetime(result["date"]).dt.date + assert all(d <= date(2026, 1, 14) for d in all_dates) + assert date(2026, 1, 15) not in all_dates.tolist() + + @patch.object(KalshiSource, "_get_market_candlesticks") + def test_exact_utc_boundary_closes_previous_day( + self, mock_candles, kalshi_source, freeze_today + ): + """A candle ending exactly at UTC midnight supplies the preceding day's value.""" + freeze_today(date(2026, 1, 15)) + mock_candles.return_value = [ + make_kalshi_candlestick(_ts(2026, 1, 12), close_dollars="0.55"), + ] + result = kalshi_source._build_resolution_df( + market=make_kalshi_api_market(), + market_info_resolution_datetime="N/A", + existing_df=None, + **_update_boundaries(date(2026, 1, 15)), + ) + + result_dates = pd.to_datetime(result["date"]).dt.date + assert result_dates.min() == date(2026, 1, 11) + first_val = result.loc[result_dates == date(2026, 1, 11), "value"].iloc[0] + assert float(first_val) == 0.55 + + @patch.object(KalshiSource, "_get_market_candlesticks") + def test_utc_cutoff_is_invariant_to_later_rebuild( + self, mock_candles, kalshi_source, freeze_today + ): + """Post-midnight trading cannot retroactively change the prior UTC date.""" + freeze_today(date(2026, 7, 24)) + candles = [ + make_kalshi_candlestick(_ts(2026, 7, 23, hour=4), close_dollars="0.51"), + make_kalshi_candlestick(_ts(2026, 7, 24, hour=0), close_dollars="0.50"), + make_kalshi_candlestick(_ts(2026, 7, 24, hour=4), close_dollars="0.49"), + ] + mock_candles.side_effect = lambda _ticker, **kwargs: [ + candle for candle in candles if candle["end_period_ts"] <= kwargs["end_ts"] + ] + market = make_kalshi_api_market(open_time="2026-07-22T00:00:00Z") + + at_cutoff = kalshi_source._build_resolution_df( + market=market, + market_info_resolution_datetime="N/A", + candlesticks_end_ts=_ts(2026, 7, 24, hour=0), + yesterday=date(2026, 7, 23), + ) + later_rebuild = kalshi_source._build_resolution_df( + market=market, + market_info_resolution_datetime="N/A", + candlesticks_end_ts=_ts(2026, 7, 24, hour=6), + yesterday=date(2026, 7, 23), + ) + + at_cutoff_value = at_cutoff.loc[ + pd.to_datetime(at_cutoff["date"]).dt.date == date(2026, 7, 23), "value" + ].iloc[0] + later_value = later_rebuild.loc[ + pd.to_datetime(later_rebuild["date"]).dt.date == date(2026, 7, 23), "value" + ].iloc[0] + assert float(at_cutoff_value) == 0.50 + assert float(later_value) == 0.50 + + +# --------------------------------------------------------------------------- +# _call_search_endpoint (mock requests.get) +# --------------------------------------------------------------------------- + + +class TestCallSearchEndpoint: + """Tests for KalshiSource._call_search_endpoint.""" + + def _mock_response(self, events, cursor=None): + resp = Mock() + resp.ok = True + resp.json.return_value = {"events": events, "cursor": cursor} + resp.raise_for_status = Mock() + return resp + + @patch("sources.kalshi.requests.get") + def test_basic_returns_qualifying_tickers(self, mock_get, kalshi_source): + """Returns qualifying tickers with category and parent routing metadata.""" + events = [ + make_kalshi_event( + category="Economics", + markets=[ + make_kalshi_api_market(ticker="A"), + make_kalshi_api_market(ticker="B"), + ], + ) + ] + mock_get.return_value = self._mock_response(events) + tickers, cursor = kalshi_source._call_search_endpoint(min_resolution_date=date(2026, 1, 25)) + assert tickers == { + "A": _discovered_market("Economics"), + "B": _discovered_market("Economics"), + } + assert cursor is None + + @patch("sources.kalshi.requests.get") + def test_skips_market_with_mismatched_parent_event(self, mock_get, kalshi_source): + """A market cannot inherit routing metadata from a different parent event.""" + events = [ + make_kalshi_event( + event_ticker="EVENT-A", + markets=[make_kalshi_api_market(ticker="A", event_ticker="EVENT-B")], + ) + ] + mock_get.return_value = self._mock_response(events) + + tickers, _ = kalshi_source._call_search_endpoint(min_resolution_date=date(2026, 1, 25)) + + assert tickers == {} + + @patch("sources.kalshi.requests.get") + def test_filters_non_binary(self, mock_get, kalshi_source): + """Scalar markets are excluded.""" + events = [ + make_kalshi_event( + category="Economics", + markets=[ + make_kalshi_api_market(ticker="bin", market_type="binary"), + make_kalshi_api_market(ticker="scal", market_type="scalar"), + ], + ) + ] + mock_get.return_value = self._mock_response(events) + ids, _ = kalshi_source._call_search_endpoint(min_resolution_date=date(2026, 1, 25)) + assert set(ids) == {"bin"} + + @patch("sources.kalshi.requests.get") + def test_filters_non_active_child_market(self, mock_get, kalshi_source): + """An open event contributes only child markets that are themselves active.""" + events = [ + make_kalshi_event( + markets=[ + make_kalshi_api_market(ticker="active", status="active"), + make_kalshi_api_market(ticker="closed", status="closed"), + ] + ) + ] + mock_get.return_value = self._mock_response(events) + + ids, _ = kalshi_source._call_search_endpoint(min_resolution_date=date(2026, 1, 25)) + + assert set(ids) == {"active"} + + @patch("sources.kalshi.requests.get") + def test_filters_low_volume(self, mock_get, kalshi_source): + """Markets with volume below the floor are excluded.""" + events = [ + make_kalshi_event( + category="Economics", + markets=[ + make_kalshi_api_market(ticker="low", volume_fp="100.00"), + make_kalshi_api_market(ticker="ok", volume_fp="10000.00"), + ], + ) + ] + mock_get.return_value = self._mock_response(events) + ids, _ = kalshi_source._call_search_endpoint(min_resolution_date=date(2026, 1, 25)) + assert set(ids) == {"ok"} + + @patch("sources.kalshi.requests.get") + def test_filters_low_open_interest(self, mock_get, kalshi_source): + """Markets with open interest below the floor are excluded.""" + events = [ + make_kalshi_event( + category="Economics", + markets=[ + make_kalshi_api_market(ticker="low", open_interest_fp="10.00"), + make_kalshi_api_market(ticker="ok", open_interest_fp="2000.00"), + ], + ) + ] + mock_get.return_value = self._mock_response(events) + ids, _ = kalshi_source._call_search_endpoint(min_resolution_date=date(2026, 1, 25)) + assert set(ids) == {"ok"} + + @patch("sources.kalshi.requests.get") + def test_filters_close_before_min_resolution(self, mock_get, kalshi_source): + """An early trading close remains an earliest plausible resolution bound.""" + events = [ + make_kalshi_event( + category="Economics", + markets=[ + make_kalshi_api_market(ticker="soon", close_time="2026-01-20T00:00:00Z"), + make_kalshi_api_market(ticker="ok", close_time="2026-03-01T00:00:00Z"), + ], + ) + ] + mock_get.return_value = self._mock_response(events) + ids, _ = kalshi_source._call_search_endpoint(min_resolution_date=date(2026, 1, 25)) + assert set(ids) == {"ok"} + + @patch("sources.kalshi.requests.get") + def test_filters_sports_event_before_min_despite_later_close(self, mock_get, kalshi_source): + """A postponement close cannot hide a sports outcome expected before the minimum.""" + events = [ + make_kalshi_event( + category="Sports", + markets=[ + make_kalshi_api_market( + ticker="postponement-bound", + occurrence_datetime="2026-07-25T05:00:00Z", + expected_expiration_time="2026-07-25T05:00:00Z", + close_time="2026-08-08T02:00:00Z", + latest_expiration_time="2026-08-08T02:00:00Z", + ), + make_kalshi_api_market( + ticker="aligned", + occurrence_datetime=None, + expected_expiration_time="2026-08-05T05:00:00Z", + close_time="2026-08-05T05:00:00Z", + latest_expiration_time="2026-08-08T02:00:00Z", + ), + ], + ) + ] + mock_get.return_value = self._mock_response(events) + + ids, _ = kalshi_source._call_search_endpoint( + min_resolution_date=date(2026, 8, 2), + max_resolution_date=date(2026, 8, 31), + ) + + assert set(ids) == {"aligned"} + + @patch("sources.kalshi.requests.get") + def test_filters_stale_expected_expiration_after_latest_bound(self, mock_get, kalshi_source): + """A stale child-contract expiration beyond the latest bound is rejected.""" + market = make_kalshi_api_market( + ticker="stale-ladder", + occurrence_datetime="2026-08-01T00:00:00Z", + close_time="2026-08-01T00:00:00Z", + latest_expiration_time="2026-08-08T00:00:00Z", + expected_expiration_time="2027-01-01T00:00:00Z", + ) + mock_get.return_value = self._mock_response([make_kalshi_event(markets=[market])]) + + ids, _ = kalshi_source._call_search_endpoint( + min_resolution_date=date(2026, 7, 1), + max_resolution_date=date(2027, 12, 31), + ) + + assert ids == {} + + @patch("sources.kalshi.requests.get") + def test_filters_close_after_max_resolution(self, mock_get, kalshi_source): + """Either provider latest bound can exclude a market beyond the maximum date.""" + events = [ + make_kalshi_event( + category="Economics", + markets=[ + make_kalshi_api_market(ticker="ok", close_time="2026-06-01T00:00:00Z"), + make_kalshi_api_market(ticker="far-close", close_time="2099-01-01T00:00:00Z"), + make_kalshi_api_market( + ticker="far-latest", + occurrence_datetime="2026-06-01T00:00:00Z", + expected_expiration_time="2026-06-01T00:00:00Z", + close_time="2026-06-01T00:00:00Z", + latest_expiration_time="2099-01-01T00:00:00Z", + ), + ], + ) + ] + mock_get.return_value = self._mock_response(events) + ids, _ = kalshi_source._call_search_endpoint( + min_resolution_date=date(2026, 1, 25), + max_resolution_date=date(2028, 1, 1), + ) + assert set(ids) == {"ok"} + + @pytest.mark.parametrize( + ("field", "value"), + [ + ("expected_expiration_time", ""), + ("latest_expiration_time", "not-a-date"), + ("occurrence_datetime", "2026-12-01T00:00:00"), + ], + ) + @patch("sources.kalshi.requests.get") + def test_filters_unusable_timing(self, mock_get, kalshi_source, field, value): + """Empty, malformed, or timezone-naive timestamps are ineligible.""" + market = make_kalshi_api_market(**{field: value}) + mock_get.return_value = self._mock_response([make_kalshi_event(markets=[market])]) + + ids, _ = kalshi_source._call_search_endpoint(min_resolution_date=date(2026, 1, 25)) + + assert ids == {} + + @patch("sources.kalshi.requests.get") + def test_any_category_included(self, mock_get, kalshi_source): + """A liquid market in any category is included (no category whitelist).""" + events = [ + make_kalshi_event( + category="Sports", + markets=[make_kalshi_api_market(ticker="sporty", volume_24h_fp="10.00")], + ) + ] + mock_get.return_value = self._mock_response(events) + tickers, _ = kalshi_source._call_search_endpoint(min_resolution_date=date(2026, 1, 25)) + assert tickers == {"sporty": _discovered_market("Sports")} + + @patch("sources.kalshi.requests.get") + def test_returns_ticker_to_category_mapping(self, mock_get, kalshi_source): + """Each qualifying ticker retains its category and parent routing metadata.""" + events = [ + make_kalshi_event( + category="Crypto", + markets=[make_kalshi_api_market(ticker="btc")], + ), + make_kalshi_event( + category="Politics", + markets=[make_kalshi_api_market(ticker="election")], + ), + ] + mock_get.return_value = self._mock_response(events) + tickers, _ = kalshi_source._call_search_endpoint(min_resolution_date=date(2026, 1, 25)) + assert tickers == { + "btc": _discovered_market("Crypto"), + "election": _discovered_market("Politics"), + } + + @patch("sources.kalshi.requests.get") + def test_cursor_passed_through(self, mock_get, kalshi_source): + """The next-page cursor is returned and an incoming cursor is sent in params.""" + mock_get.return_value = self._mock_response([], cursor="next_page") + _, cursor = kalshi_source._call_search_endpoint( + min_resolution_date=date(2026, 1, 25), cursor="cur1" + ) + assert cursor == "next_page" + assert mock_get.call_args.kwargs["params"]["cursor"] == "cur1" + + @patch("sources.kalshi.requests.get") + def test_missing_category_uses_uncategorized(self, mock_get, kalshi_source): + """A market remains eligible when its event omits the deprecated category field.""" + event = make_kalshi_event(markets=[make_kalshi_api_market(ticker="uncategorized")]) + del event["category"] + mock_get.return_value = self._mock_response([event]) + + tickers, _ = kalshi_source._call_search_endpoint(min_resolution_date=date(2026, 1, 25)) + + assert tickers == {"uncategorized": _discovered_market("Uncategorized")} + + @pytest.mark.parametrize( + ("location", "missing_field"), + [ + ("response", "events"), + ("response", "cursor"), + ("event", "markets"), + ("event", "event_ticker"), + ("event", "series_ticker"), + ("market", "ticker"), + ("market", "event_ticker"), + ("market", "status"), + ("market", "market_type"), + ("market", "volume_fp"), + ("market", "open_interest_fp"), + ("market", "close_time"), + ("market", "expected_expiration_time"), + ("market", "latest_expiration_time"), + ], + ) + @patch("sources.kalshi.requests.get") + def test_missing_required_field_fails_loudly( + self, mock_get, kalshi_source, location, missing_field + ): + """A malformed successful response raises instead of silently losing markets.""" + market = make_kalshi_api_market() + event = make_kalshi_event(markets=[market]) + data = {"events": [event], "cursor": None} + target = {"response": data, "event": event, "market": market}[location] + del target[missing_field] + + response = Mock() + response.ok = True + response.json.return_value = data + mock_get.return_value = response + + with pytest.raises( + ValueError, + match=rf"Kalshi events API response is missing required field '{missing_field}'", + ): + kalshi_source._call_search_endpoint(min_resolution_date=date(2026, 1, 25)) + + +# --------------------------------------------------------------------------- +# _get_market / _get_market_candlesticks (mock requests.get) +# --------------------------------------------------------------------------- + + +class TestGetMarket: + """Tests for KalshiSource._get_market.""" + + @patch("sources.kalshi.time.sleep") + @patch("sources.kalshi.requests.get") + def test_returns_market_object(self, mock_get, mock_sleep, kalshi_source): + """Unwraps and returns the 'market' object.""" + resp = Mock() + resp.ok = True + resp.json.return_value = {"market": make_kalshi_api_market(ticker="KXTEST-001")} + mock_get.return_value = resp + + result = kalshi_source._get_market("KXTEST-001") + assert result["ticker"] == "KXTEST-001" + + @patch("sources.kalshi.time.sleep") + @patch("sources.kalshi.requests.get") + def test_live_404_falls_back_to_historical_market(self, mock_get, mock_sleep, kalshi_source): + """An archived market is loaded from the historical detail endpoint.""" + live = Mock(status_code=404, ok=False) + historical = Mock(status_code=200, ok=True) + historical.json.return_value = {"market": make_kalshi_api_market(ticker="KXARCHIVED-001")} + mock_get.side_effect = [live, historical] + + result = kalshi_source._get_market("KXARCHIVED-001") + + assert result["ticker"] == "KXARCHIVED-001" + assert "/historical/markets/KXARCHIVED-001" in mock_get.call_args.args[0] + + @patch("sources.kalshi.time.sleep") + @patch("sources.kalshi.requests.get") + def test_transient_dual_404_is_retried(self, mock_get, mock_sleep, kalshi_source): + """A newly created market can become available during bounded 404 retries.""" + missing = Mock(status_code=404, ok=False) + available = Mock(status_code=200, ok=True) + available.json.return_value = {"market": make_kalshi_api_market(ticker="KXNEW-001")} + mock_get.side_effect = [missing, missing, available] + + result = kalshi_source._get_market("KXNEW-001") + + assert result["ticker"] == "KXNEW-001" + + @patch("sources.kalshi.time.sleep") + @patch("sources.kalshi.requests.get") + def test_dual_404_raises_market_not_found(self, mock_get, mock_sleep, kalshi_source): + """Repeated absence from both API partitions is reported to update().""" + missing = Mock(status_code=404, ok=False) + mock_get.return_value = missing + + with pytest.raises(MarketNotFoundError): + kalshi_source._get_market("KXGONE-001") + + +class TestGetMarketCandlesticks: + """Tests for KalshiSource._get_market_candlesticks.""" + + @patch("sources.kalshi.time.sleep") + @patch("sources.kalshi.requests.get") + def test_builds_series_url_and_returns_candles( + self, mock_get, mock_sleep, kalshi_source, freeze_today + ): + """Uses the series-derived URL and returns the candlesticks list.""" + freeze_today(date(2026, 1, 15)) + resp = Mock() + resp.ok = True + resp.json.return_value = { + "candlesticks": [make_kalshi_candlestick(_ts(2026, 1, 10), close_dollars="0.5")] + } + mock_get.return_value = resp + + result = kalshi_source._get_market_candlesticks( + "KXWCSPREAD-26JUN18CANQAT-CAN6", + start_ts=_ts(2026, 1, 9), + end_ts=_ts(2026, 1, 15), + ) + assert len(result) == 1 + url = mock_get.call_args[0][0] + assert "/series/KXWCSPREAD/markets/KXWCSPREAD-26JUN18CANQAT-CAN6/candlesticks" in url + assert mock_get.call_args.kwargs["params"]["period_interval"] == 60 + + @patch("sources.kalshi.time.sleep") + @patch("sources.kalshi.requests.get") + def test_historical_candles_are_normalized(self, mock_get, mock_sleep, kalshi_source): + """Archived price.close values use the same shape as live close_dollars values.""" + live = Mock(status_code=404, ok=False) + historical = Mock(status_code=200, ok=True) + historical.json.return_value = { + "candlesticks": [ + { + "end_period_ts": _ts(2026, 1, 10), + "price": {"close": "0.50"}, + } + ] + } + mock_get.side_effect = [live, historical] + + candles = kalshi_source._get_market_candlesticks( + "KXARCHIVED-001", + start_ts=_ts(2026, 1, 9), + end_ts=_ts(2026, 1, 15), + ) + + assert candles[0]["price"]["close_dollars"] == "0.50" + assert "/historical/markets/KXARCHIVED-001/candlesticks" in mock_get.call_args.args[0] + + @patch("sources.kalshi.time.sleep") + @patch("sources.kalshi.requests.get") + def test_long_hourly_history_is_fetched_in_bounded_windows( + self, mock_get, mock_sleep, kalshi_source + ): + """Histories over Kalshi's 5,000-candle limit are merged without duplicates.""" + first = Mock() + first.ok = True + first.json.return_value = { + "candlesticks": [ + make_kalshi_candlestick(1000, close_dollars="0.30"), + make_kalshi_candlestick(2000, close_dollars="0.40"), + ] + } + second = Mock() + second.ok = True + second.json.return_value = { + "candlesticks": [ + make_kalshi_candlestick(2000, close_dollars="0.40"), + make_kalshi_candlestick(3000, close_dollars="0.50"), + ] + } + mock_get.side_effect = [first, second] + end_ts = 100 + 5001 * 60 * 60 + + candles = kalshi_source._get_market_candlesticks( + "KXTEST-001", + start_ts=100, + end_ts=end_ts, + ) + + assert [candle["end_period_ts"] for candle in candles] == [1000, 2000, 3000] + assert mock_get.call_count == 2 + first_params = mock_get.call_args_list[0].kwargs["params"] + second_params = mock_get.call_args_list[1].kwargs["params"] + assert first_params["end_ts"] - first_params["start_ts"] < 5000 * 60 * 60 + assert second_params["start_ts"] == first_params["end_ts"] + 1 + assert second_params["end_ts"] == end_ts + + @patch("sources.kalshi.time.sleep") + @patch("sources.kalshi.requests.get") + def test_empty_candles(self, mock_get, mock_sleep, kalshi_source, freeze_today): + """Missing candlesticks key returns empty list.""" + freeze_today(date(2026, 1, 15)) + resp = Mock() + resp.ok = True + resp.json.return_value = {"ticker": "KXTEST-001"} + mock_get.return_value = resp + assert ( + kalshi_source._get_market_candlesticks( + "KXTEST-001", + start_ts=_ts(2026, 1, 14), + end_ts=_ts(2026, 1, 15), + ) + == [] + ) + + +# --------------------------------------------------------------------------- +# Request throttling +# --------------------------------------------------------------------------- + + +class TestRequestThrottling: + """Tests that consecutive Kalshi API requests are paced across endpoints.""" + + @patch("sources.kalshi.time.sleep") + @patch("sources.kalshi.time.monotonic", side_effect=[100.0, 100.0, 100.04, 100.1]) + @patch("sources.kalshi.requests.get") + def test_consecutive_requests_share_rate_limit( + self, mock_get, _mock_monotonic, mock_sleep, kalshi_source + ): + """A market request followed by a candle request waits for the remaining interval.""" + market_response = Mock() + market_response.ok = True + market_response.status_code = 200 + market_response.json.return_value = {"market": make_kalshi_api_market(ticker="KXTEST-001")} + candle_response = Mock() + candle_response.ok = True + candle_response.json.return_value = {"candlesticks": []} + mock_get.side_effect = [market_response, candle_response] + + kalshi_source._get_market("KXTEST-001") + kalshi_source._get_market_candlesticks( + "KXTEST-001", + start_ts=_ts(2026, 1, 14), + end_ts=_ts(2026, 1, 15), + ) + + mock_sleep.assert_called_once_with(pytest.approx(0.06)) + assert mock_get.call_count == 2 + + +# --------------------------------------------------------------------------- +# fetch() (mock _search_markets) +# --------------------------------------------------------------------------- + + +class TestFetch: + """Tests for KalshiSource.fetch.""" + + @patch.object(KalshiSource, "_search_markets") + def test_basic_fetch(self, mock_search, kalshi_source): + """Returns sorted IDs with their structured parent identifiers.""" + mock_search.return_value = { + "id_b": _discovered_market("Sports", "event_b", "series_b"), + "id_a": _discovered_market("Economics", "event_a", "series_a"), + "id_c": _discovered_market("Crypto", "event_c", "series_c"), + } + dff = kalshi_source.fetch() + + assert dff["id"].tolist() == ["id_a", "id_b", "id_c"] + assert dff[["event_ticker", "series_ticker"]].to_dict("records") == [ + {"event_ticker": "event_a", "series_ticker": "series_a"}, + {"event_ticker": "event_b", "series_ticker": "series_b"}, + {"event_ticker": "event_c", "series_ticker": "series_c"}, + ] + KalshiFetchFrame.validate(dff) + + @patch.object(KalshiSource, "_search_markets") + def test_empty_results(self, mock_search, kalshi_source): + """Empty search returns empty valid frame.""" + mock_search.return_value = {} + dff = kalshi_source.fetch() + + assert len(dff) == 0 + KalshiFetchFrame.validate(dff) + + @patch.object(KalshiSource, "_search_markets") + def test_fetch_caps_dominant_category(self, mock_search, kalshi_source, monkeypatch): + """fetch() balances the pool so a dominant category cannot flood it.""" + monkeypatch.setattr("sources.kalshi._MAX_PER_CATEGORY", 2) + # 5 Sports (over the cap of 2) and 1 each of two other categories (kept in full). + mock_search.return_value = { + **{ + f"sport_{i}": _discovered_market("Sports", f"sport_event_{i}", "sport_series") + for i in range(5) + }, + "econ_0": _discovered_market("Economics", "econ_event", "econ_series"), + "crypto_0": _discovered_market("Crypto", "crypto_event", "crypto_series"), + } + dff = kalshi_source.fetch() + + kept = set(dff["id"]) + assert len([i for i in kept if i.startswith("sport_")]) == 2 # Sports capped + assert "econ_0" in kept and "crypto_0" in kept # small categories kept in full + assert len(dff) == 4 + KalshiFetchFrame.validate(dff) + + +class TestBalanceCategories: + """Tests for KalshiSource._balance_categories.""" + + def test_empty_returns_empty(self): + """No discovered tickers returns an empty list.""" + assert KalshiSource._balance_categories({}) == [] + + def test_small_categories_kept_in_full(self, monkeypatch): + """Every category under the cap is kept entirely.""" + monkeypatch.setattr("sources.kalshi._MAX_PER_CATEGORY", 10) + mapping = {"a": "Sports", "b": "Economics", "c": "Crypto"} + assert set(KalshiSource._balance_categories(mapping)) == {"a", "b", "c"} + + def test_over_cap_category_is_downsampled(self, monkeypatch): + """A category above the cap is reduced to exactly the cap; others untouched.""" + monkeypatch.setattr("sources.kalshi._MAX_PER_CATEGORY", 3) + mapping = {**{f"s{i}": "Sports" for i in range(10)}, "e0": "Economics"} + kept = KalshiSource._balance_categories(mapping) + assert len([i for i in kept if i.startswith("s")]) == 3 + assert "e0" in kept + + +# --------------------------------------------------------------------------- +# final curation eligibility refresh +# --------------------------------------------------------------------------- + + +class TestRefreshCurationCandidates: + """Tests for the live Kalshi eligibility check immediately before curation.""" + + @patch("helpers.kalshi._get_markets_by_ticker") + def test_refreshes_timing_and_keeps_only_active_markets(self, mock_get_markets): + """Curation uses current timing and excludes non-active or missing markets.""" + mock_get_markets.return_value = [ + make_kalshi_api_market( + ticker="ACTIVE", + status="active", + occurrence_datetime="2026-08-01T05:00:00Z", + expected_expiration_time="2026-08-02T05:00:00Z", + close_time="2026-08-03T05:00:00Z", + ), + make_kalshi_api_market(ticker="DETERMINED", status="determined"), + ] + dfq = make_question_df( + [ + {"id": "ACTIVE"}, + {"id": "DETERMINED"}, + {"id": "MISSING"}, + ] + ) + + refreshed = kalshi_helpers.refresh_curation_candidates(dfq) + + assert refreshed["id"].tolist() == ["ACTIVE"] + assert refreshed.iloc[0]["market_info_close_datetime"] == "2026-08-01T05:00:00+00:00" + assert dfq["id"].tolist() == ["ACTIVE", "DETERMINED", "MISSING"] + + +# --------------------------------------------------------------------------- +# update() (mock _get_market + _build_resolution_df) +# --------------------------------------------------------------------------- + + +class TestUpdate: + """Tests for KalshiSource.update.""" + + def test_uses_one_time_cutoff_for_all_markets(self, kalshi_source): + """Every candlestick request in one update uses the same time cutoff.""" + response = Mock() + response.ok = True + response.json.return_value = { + "candlesticks": [make_kalshi_candlestick(_ts(2026, 1, 12), close_dollars="0.50")] + } + update_datetimes = [ + datetime(2026, 1, 15, 23, 59, 58, tzinfo=timezone.utc), + datetime(2026, 1, 15, 23, 59, 59, tzinfo=timezone.utc), + datetime(2026, 1, 16, 0, 0, 0, tzinfo=timezone.utc), + datetime(2026, 1, 16, 0, 0, 1, tzinfo=timezone.utc), + ] + ids = ["KXTEST-001", "KXTEST-002"] + dfq = make_question_df([{"id": ticker, "resolved": False} for ticker in ids]) + dff = make_kalshi_fetch_df([{"id": ticker} for ticker in ids]) + + with ( + patch.object( + KalshiSource, + "_get_market", + side_effect=lambda ticker: make_kalshi_api_market( + ticker=ticker, + open_time="2026-01-10T00:00:00Z", + ), + ), + patch("sources.kalshi.requests.get", return_value=response) as mock_get, + patch("sources.kalshi.time.sleep"), + patch( + "sources.kalshi.dates.get_datetime_today", + side_effect=update_datetimes, + ), + ): + kalshi_source.update(dfq, dff) + + request_end_timestamps = [ + call.kwargs["params"]["end_ts"] for call in mock_get.call_args_list + ] + expected_end_ts = int(update_datetimes[0].timestamp()) + assert request_end_timestamps == [expected_end_ts, expected_end_ts] + + @patch.object(KalshiSource, "_build_resolution_df") + @patch.object(KalshiSource, "_get_market") + def test_new_id_appended(self, mock_market, mock_build, kalshi_source): + """Tickers in dff not in dfq get appended with defaults.""" + mock_market.return_value = make_kalshi_api_market(ticker="new_001") + mock_build.return_value = make_resolution_df( + [{"id": "new_001", "date": "2024-06-01", "value": 0.5}] + ) + dfq = make_question_df([{"id": "existing_001"}]) + dff = make_kalshi_fetch_df([{"id": "new_001"}]) + + result = kalshi_source.update(dfq, dff) + + assert "new_001" in result.dfq["id"].values + assert len(result.dfq) == 2 + new_row = result.dfq[result.dfq["id"] == "new_001"].iloc[0] + assert new_row["freeze_datetime_value_explanation"] == "The market price." + + @patch.object(KalshiSource, "_build_resolution_df") + @patch.object(KalshiSource, "_get_market") + def test_existing_unresolved_updated(self, mock_market, mock_build, kalshi_source): + """Unresolved question fields are updated from market details.""" + mock_market.return_value = make_kalshi_api_market( + ticker="KXTEST-001", + title="Updated question text", + yes_sub_title="Specific Yes outcome", + rules_primary="New rules", + ) + mock_build.return_value = make_resolution_df( + [{"id": "KXTEST-001", "date": "2024-06-01", "value": 0.65}] + ) + dfq = make_question_df([{"id": "KXTEST-001", "resolved": False}]) + dff = make_kalshi_fetch_df([{"id": "KXTEST-001"}]) + + result = kalshi_source.update(dfq, dff) + + row = result.dfq[result.dfq["id"] == "KXTEST-001"].iloc[0] + assert row["question"] == "Updated question text" + assert row["market_info_resolution_criteria"] == "New rules" + assert row["url"] == "https://kalshi.com/markets/kxtest/x/kxtest" + + @patch.object(KalshiSource, "_build_resolution_df") + @patch.object(KalshiSource, "_get_market") + def test_new_question_url_uses_structured_parent_identifiers( + self, mock_market, mock_build, kalshi_source + ): + """A first update builds the public URL without parsing the child market ticker.""" + mock_market.return_value = make_kalshi_api_market( + ticker="APPLEFOLD-26DEC31", + event_ticker="APPLEFOLD", + ) + mock_build.return_value = make_resolution_df( + [{"id": "APPLEFOLD-26DEC31", "date": "2026-08-01", "value": 0.8}] + ) + dff = make_kalshi_fetch_df( + [ + { + "id": "APPLEFOLD-26DEC31", + "event_ticker": "APPLEFOLD", + "series_ticker": "KXAPPLEFOLD", + } + ] + ) + + result = kalshi_source.update( + make_question_df([{"id": "existing", "resolved": True}]), + dff, + existing_resolution_ids={"existing"}, + ) + + row = result.dfq[result.dfq["id"] == "APPLEFOLD-26DEC31"].iloc[0] + assert row["url"] == "https://kalshi.com/markets/kxapplefold/x/applefold" + + @patch.object(KalshiSource, "_build_resolution_df") + @patch.object(KalshiSource, "_get_market") + def test_existing_question_absent_from_fetch_keeps_url( + self, mock_market, mock_build, kalshi_source + ): + """A later capped fetch cannot replace an existing URL with a guessed route.""" + mock_market.return_value = make_kalshi_api_market(ticker="KXTEST-001") + mock_build.return_value = make_resolution_df( + [{"id": "KXTEST-001", "date": "2026-08-01", "value": 0.5}] + ) + expected_url = "https://kalshi.com/markets/kxtest/x/kxtest" + dfq = make_question_df([{"id": "KXTEST-001", "url": expected_url}]) + + result = kalshi_source.update(dfq, make_kalshi_fetch_df([])) + + assert result.dfq.iloc[0]["url"] == expected_url + + @patch.object(KalshiSource, "_build_resolution_df") + @patch.object(KalshiSource, "_get_market") + def test_persists_earliest_resolution_for_shared_curation_filter( + self, mock_market, mock_build, kalshi_source, monkeypatch + ): + """The existing close-field filter sees Kalshi's earliest plausible resolution.""" + mock_market.return_value = make_kalshi_api_market( + ticker="KXSPORTS-001", + occurrence_datetime="2026-07-25T05:00:00Z", + expected_expiration_time="2026-07-25T05:00:00Z", + close_time="2026-08-08T02:00:00Z", + latest_expiration_time="2026-08-08T02:00:00Z", + ) + mock_build.return_value = make_resolution_df( + [{"id": "KXSPORTS-001", "date": "2026-07-23", "value": 0.5}] + ) + dfq = make_question_df([{"id": "KXSPORTS-001", "resolved": False}]) + dff = make_kalshi_fetch_df([{"id": "KXSPORTS-001"}]) + + result = kalshi_source.update(dfq, dff) + row = result.dfq.loc[result.dfq["id"] == "KXSPORTS-001"].iloc[0] + + assert row["market_info_close_datetime"] == "2026-07-25T05:00:00+00:00" + monkeypatch.setattr( + create_question_set.question_curation, + "FREEZE_DATETIME", + datetime(2026, 7, 23, tzinfo=timezone.utc), + ) + monkeypatch.setattr(create_question_set.question_curation, "FREEZE_WINDOW_IN_DAYS", 10) + curated = create_question_set.drop_questions_that_resolve_too_soon( + source="kalshi", dfq=result.dfq + ) + assert curated.empty + + @patch.object(KalshiSource, "_build_resolution_df") + @patch.object(KalshiSource, "_get_market") + def test_distinct_sibling_titles_omit_participant_labels( + self, mock_market, mock_build, kalshi_source + ): + """Complete sibling titles are not narrowed by non-binding participant labels.""" + markets = { + "GOVPARTYAZ-26-D": make_kalshi_api_market( + ticker="GOVPARTYAZ-26-D", + event_ticker="GOVPARTYAZ-26", + title="Will the Democratic party win the governorship in Arizona", + yes_sub_title="Katie Hobbs", + rules_primary=( + "If a representative of the Democratic party is inaugurated as the governor " + "of Arizona pursuant to the 2026 election, then the market resolves to Yes." + ), + ), + "GOVPARTYAZ-26-R": make_kalshi_api_market( + ticker="GOVPARTYAZ-26-R", + event_ticker="GOVPARTYAZ-26", + title="Will the Republican party win the governorship in Arizona", + yes_sub_title="Andy Biggs", + rules_primary=( + "If a representative of the Republican party is inaugurated as the governor " + "of Arizona pursuant to the 2026 election, then the market resolves to Yes." + ), + ), + } + mock_market.side_effect = lambda ticker: markets[ticker] + mock_build.side_effect = lambda market, *args, **kwargs: make_resolution_df( + [{"id": market["ticker"], "date": "2026-08-01", "value": 0.5}] + ) + ids = list(markets) + dfq = make_question_df([{"id": ticker, "resolved": False} for ticker in ids]) + dff = make_kalshi_fetch_df( + [{"id": ticker, "event_ticker": "GOVPARTYAZ-26"} for ticker in ids] + ) + + result = kalshi_source.update(dfq, dff) + + rows = result.dfq.set_index("id") + assert rows.at["GOVPARTYAZ-26-D", "question"] == ( + "Will the Democratic party win the governorship in Arizona" + ) + assert rows.at["GOVPARTYAZ-26-R", "question"] == ( + "Will the Republican party win the governorship in Arizona" + ) + + @patch.object(KalshiSource, "_build_resolution_df") + @patch.object(KalshiSource, "_get_market") + def test_child_outcomes_produce_distinct_questions( + self, mock_market, mock_build, kalshi_source + ): + """Sibling contracts with a shared title retain distinct public questions.""" + markets = { + "KXSPACEX-120": make_kalshi_api_market( + ticker="KXSPACEX-120", + event_ticker="KXSPACEX", + title="How many launches will SpaceX have in 2026?", + yes_sub_title="Above 120", + ), + "KXSPACEX-140": make_kalshi_api_market( + ticker="KXSPACEX-140", + event_ticker="KXSPACEX", + title="How many launches will SpaceX have in 2026?", + yes_sub_title="Above 140", + ), + } + mock_market.side_effect = lambda ticker: markets[ticker] + mock_build.side_effect = lambda market, *args, **kwargs: make_resolution_df( + [{"id": market["ticker"], "date": "2024-06-01", "value": 0.5}] + ) + ids = list(markets) + dfq = make_question_df([{"id": ticker, "resolved": False} for ticker in ids]) + dff = make_kalshi_fetch_df([{"id": ticker} for ticker in ids]) + + result = kalshi_source.update(dfq, dff) + + questions = result.dfq.set_index("id")["question"] + assert questions["KXSPACEX-120"] == ( + "How many launches will SpaceX have in 2026? [Yes: Above 120]" + ) + assert questions["KXSPACEX-140"] == ( + "How many launches will SpaceX have in 2026? [Yes: Above 140]" + ) + assert questions.nunique() == 2 + + @patch.object(KalshiSource, "_build_resolution_df") + @patch.object(KalshiSource, "_get_market") + def test_market_becomes_resolved(self, mock_market, mock_build, kalshi_source): + """Market with a terminal status marks the dfq row as resolved.""" + mock_market.return_value = make_kalshi_api_market( + ticker="KXTEST-001", + status="finalized", + result="yes", + settlement_ts="2026-01-13T05:00:00Z", + ) + mock_build.return_value = make_resolution_df( + [{"id": "KXTEST-001", "date": "2024-06-01", "value": 1.0}] + ) + dfq = make_question_df([{"id": "KXTEST-001", "resolved": False}]) + dff = make_kalshi_fetch_df([{"id": "KXTEST-001"}]) + + result = kalshi_source.update(dfq, dff) + + row = result.dfq[result.dfq["id"] == "KXTEST-001"].iloc[0] + assert bool(row["resolved"]) is True + assert "2026-01-13" in str(row["market_info_resolution_datetime"]) + + @pytest.mark.parametrize(("outcome", "expected"), [("yes", 1.0), ("no", 0.0)]) + def test_finalization_replaces_stale_settlement_probability( + self, outcome, expected, kalshi_source + ): + """Finalization emits a file replacing a same-date probability with the outcome.""" + ticker = "KXTEST-001" + resolved_date = date(2026, 1, 13) + existing = make_resolution_df( + [ + {"id": ticker, "date": "2026-01-12", "value": 0.55}, + {"id": ticker, "date": "2026-01-13", "value": 0.63}, + ] + ) + market = make_kalshi_api_market( + ticker=ticker, + status="finalized", + result=outcome, + settlement_ts="2026-01-13T05:00:00Z", + ) + dfq = make_question_df([{"id": ticker, "resolved": False}]) + dff = make_kalshi_fetch_df([{"id": ticker}]) + + with ( + patch.object(KalshiSource, "_get_market", return_value=market), + patch.object(KalshiSource, "_get_market_candlesticks", return_value=[]), + ): + result = kalshi_source.update( + dfq, + dff, + existing_resolution_files={ticker: existing}, + existing_resolution_ids={ticker}, + ) + + corrected = result.resolution_files[ticker] + corrected_dates = pd.to_datetime(corrected["date"]).dt.date + row = result.dfq.loc[result.dfq["id"] == ticker].iloc[0] + assert bool(row["resolved"]) is True + assert float(row["freeze_datetime_value"]) == expected + assert corrected_dates.max() == resolved_date + assert corrected_dates.tolist().count(resolved_date) == 1 + assert float(corrected["value"].iloc[-1]) == expected + + @patch.object(KalshiSource, "_build_resolution_df") + @patch.object(KalshiSource, "_get_market") + def test_resolution_file_stored(self, mock_market, mock_build, kalshi_source): + """Resolution file from _build_resolution_df is in result.""" + mock_market.return_value = make_kalshi_api_market(ticker="KXTEST-001") + mock_build.return_value = make_resolution_df( + [{"id": "KXTEST-001", "date": "2024-06-01", "value": 0.5}] + ) + dfq = make_question_df([{"id": "KXTEST-001", "resolved": False}]) + dff = make_kalshi_fetch_df([{"id": "KXTEST-001"}]) + + result = kalshi_source.update(dfq, dff) + assert "KXTEST-001" in result.resolution_files + + @patch.object(KalshiSource, "_build_resolution_df") + @patch.object(KalshiSource, "_get_market") + def test_freeze_datetime_value_set(self, mock_market, mock_build, kalshi_source): + """freeze_datetime_value is set to last value of resolution df.""" + mock_market.return_value = make_kalshi_api_market(ticker="KXTEST-001") + mock_build.return_value = make_resolution_df( + [ + {"id": "KXTEST-001", "date": "2024-06-01", "value": 0.3}, + {"id": "KXTEST-001", "date": "2024-06-02", "value": 0.75}, + ] + ) + dfq = make_question_df([{"id": "KXTEST-001", "resolved": False}]) + dff = make_kalshi_fetch_df([{"id": "KXTEST-001"}]) + + result = kalshi_source.update(dfq, dff) + row = result.dfq[result.dfq["id"] == "KXTEST-001"].iloc[0] + assert str(row["freeze_datetime_value"]) == "0.75" + + @patch.object(KalshiSource, "_build_resolution_df") + @patch.object(KalshiSource, "_get_market") + def test_build_resolution_returns_none(self, mock_market, mock_build, kalshi_source): + """_build_resolution_df returning None: no resolution file stored.""" + mock_market.return_value = make_kalshi_api_market(ticker="KXTEST-001") + mock_build.return_value = None + dfq = make_question_df([{"id": "KXTEST-001", "resolved": False}]) + dff = make_kalshi_fetch_df([{"id": "KXTEST-001"}]) + + result = kalshi_source.update(dfq, dff) + assert "KXTEST-001" not in (result.resolution_files or {}) + + @patch.object(KalshiSource, "_build_resolution_df") + @patch.object(KalshiSource, "_get_market") + def test_regenerates_missing_resolved_files(self, mock_market, mock_build, kalshi_source): + """Resolved questions missing from storage get resolution files regenerated.""" + mock_market.return_value = make_kalshi_api_market( + ticker="KXTEST-001", status="finalized", result="yes" + ) + mock_build.return_value = make_resolution_df( + [{"id": "KXTEST-001", "date": "2024-06-01", "value": 1.0}] + ) + dfq = make_question_df( + [ + { + "id": "KXTEST-001", + "resolved": True, + "market_info_resolution_datetime": "2024-07-01T00:00:00+00:00", + } + ] + ) + dff = make_kalshi_fetch_df([{"id": "KXTEST-001"}]) + + result = kalshi_source.update(dfq, dff, existing_resolution_ids=set()) + assert "KXTEST-001" in result.resolution_files + + @patch.object(KalshiSource, "_build_resolution_df") + @patch.object(KalshiSource, "_get_market") + def test_skips_resolved_already_in_storage(self, mock_market, mock_build, kalshi_source): + """Resolved questions with files in storage are not re-fetched.""" + dfq = make_question_df( + [ + { + "id": "KXTEST-001", + "resolved": True, + "market_info_resolution_datetime": "2024-07-01T00:00:00+00:00", + } + ] + ) + dff = make_kalshi_fetch_df([{"id": "KXTEST-001"}]) + + result = kalshi_source.update(dfq, dff, existing_resolution_ids={"KXTEST-001"}) + mock_market.assert_not_called() + assert "KXTEST-001" not in (result.resolution_files or {}) + + @patch.object(KalshiSource, "_build_resolution_df") + @patch.object(KalshiSource, "_get_market") + def test_caps_new_questions(self, mock_market, mock_build, kalshi_source, monkeypatch): + """New tickers exceeding the unresolved-pool cap are not all added.""" + question_limit = 2 + monkeypatch.setattr("sources.kalshi._QUESTION_LIMIT", question_limit) + mock_market.return_value = make_kalshi_api_market() + mock_build.return_value = make_resolution_df( + [{"id": "x", "date": "2024-06-01", "value": 0.5}] + ) + new_ids = [f"new_{i}" for i in range(5)] + dfq = make_question_df([{"id": "existing"}]) + dff = make_kalshi_fetch_df([{"id": ticker} for ticker in new_ids]) + + result = kalshi_source.update(dfq, dff) + + result_ids = set(result.dfq["id"]) + assert len(result.dfq) == question_limit + assert "existing" in result_ids + assert len(result_ids.intersection(new_ids)) == 1 + + @patch.object(KalshiSource, "_build_resolution_df") + @patch.object(KalshiSource, "_get_market") + def test_output_schema_valid(self, mock_market, mock_build, kalshi_source): + """Output dfq passes QuestionFrame validation.""" + mock_market.return_value = make_kalshi_api_market(ticker="new_001") + mock_build.return_value = make_resolution_df( + [{"id": "new_001", "date": "2024-06-01", "value": 0.5}] + ) + dfq = make_question_df([{"id": "existing_001"}]) + dff = make_kalshi_fetch_df([{"id": "new_001"}]) + + result = kalshi_source.update(dfq, dff) + QuestionFrame.validate(result.dfq) + + @patch.object(KalshiSource, "_build_resolution_df") + @patch.object(KalshiSource, "_get_market") + def test_quarantines_persisted_market_not_found(self, mock_market, mock_build, kalshi_source): + """A persisted dual-404 market is retained but made ineligible for curation.""" + mock_market.side_effect = MarketNotFoundError("KXTEST-001") + dfq = make_question_df([{"id": "KXTEST-001", "resolved": False}]) + dff = make_kalshi_fetch_df([{"id": "KXTEST-001"}]) + + result = kalshi_source.update(dfq, dff) + + row = result.dfq.loc[result.dfq["id"] == "KXTEST-001"].iloc[0] + assert bool(row["resolved"]) is False + assert row["freeze_datetime_value"] == "N/A" + assert "KXTEST-001" not in (result.resolution_files or {}) + mock_build.assert_not_called() + + @patch.object(KalshiSource, "_build_resolution_df") + @patch.object(KalshiSource, "_get_market") + def test_candle_miss_restores_persisted_question(self, mock_market, mock_build, kalshi_source): + """A dual-404 candle miss cannot leave a partially refreshed question behind.""" + mock_market.return_value = make_kalshi_api_market( + ticker="KXTEST-001", + title="New title", + status="finalized", + settlement_ts="2026-01-13T05:00:00Z", + ) + mock_build.side_effect = MarketNotFoundError("KXTEST-001") + dfq = make_question_df( + [{"id": "KXTEST-001", "question": "Persisted title", "resolved": False}] + ) + dff = make_kalshi_fetch_df([{"id": "KXTEST-001"}]) + + result = kalshi_source.update(dfq, dff) + + row = result.dfq.loc[result.dfq["id"] == "KXTEST-001"].iloc[0] + assert row["question"] == "Persisted title" + assert bool(row["resolved"]) is False + assert row["freeze_datetime_value"] == "N/A" + + @patch.object(KalshiSource, "_build_resolution_df") + @patch.object(KalshiSource, "_get_market") + def test_nullified_market_is_skipped_and_does_not_consume_cap( + self, mock_market, mock_build, kalshi_source, monkeypatch + ): + """A confirmed permanent orphan is retained without blocking a new candidate.""" + monkeypatch.setattr("sources.kalshi._QUESTION_LIMIT", 1) + monkeypatch.setattr( + kalshi_source, + "nullified_questions", + [NullifiedQuestion(id="KXNULL-001", nullification_start_date=date(2026, 1, 1))], + ) + mock_market.return_value = make_kalshi_api_market(ticker="KXNEW-001") + mock_build.return_value = make_resolution_df( + [{"id": "KXNEW-001", "date": "2026-01-10", "value": 0.5}] + ) + dfq = make_question_df([{"id": "KXNULL-001", "resolved": False}]) + dff = make_kalshi_fetch_df([{"id": "KXNEW-001"}]) + + result = kalshi_source.update(dfq, dff) + + nullified = result.dfq.loc[result.dfq["id"] == "KXNULL-001"].iloc[0] + assert bool(nullified["resolved"]) is True + assert nullified["freeze_datetime_value"] == "N/A" + assert "KXNEW-001" in result.dfq["id"].values + assert all(call.args[0] != "KXNULL-001" for call in mock_market.call_args_list) + + @patch.object(KalshiSource, "_build_resolution_df") + @patch.object(KalshiSource, "_get_market") + def test_new_ticker_not_found_is_dropped(self, mock_market, mock_build, kalshi_source): + """A brand-new ticker that 404s is dropped, not persisted as a null-filled row. + + The append step seeds new rows with None placeholders; if a brand-new ticker 404s before it + is populated, keeping that row would fail QuestionFrame's non-nullable columns on the next + run's input validation. Existing questions that 404 keep their prior data (see + test_quarantines_persisted_market_not_found). + """ + mock_market.side_effect = MarketNotFoundError("new_404") + # existing_001 is resolved and already in storage, so it is never re-fetched; only the + # brand-new ticker reaches _get_market (and 404s). + dfq = make_question_df([{"id": "existing_001", "resolved": True}]) + dff = make_kalshi_fetch_df([{"id": "new_404"}]) + + result = kalshi_source.update(dfq, dff, existing_resolution_ids={"existing_001"}) + + assert "new_404" not in result.dfq["id"].values + assert "existing_001" in result.dfq["id"].values + # The persisted frame must still satisfy the (non-nullable) QuestionFrame contract. + QuestionFrame.validate(result.dfq) + mock_build.assert_not_called() + + +def test_update_driver_downloads_only_unresolved_resolution_histories(): + """The nightly update downloads history contents only for unresolved questions.""" + from orchestration.func_kalshi_update import main as update_main + + dfq = make_question_df( + [ + {"id": "active", "resolved": False}, + {"id": "nullified", "resolved": False}, + {"id": "finalized", "resolved": True}, + ] + ) + dff = make_kalshi_fetch_df([{"id": "active"}]) + + with ( + patch.object( + update_main.data_utils, + "get_data_from_cloud_storage", + return_value=(dfq, dff), + ), + patch.object(update_main.data_utils, "upload_questions"), + patch.object( + update_main._source_io.gcp.storage, + "download_no_error_message_on_404", + ) as mock_download, + patch.object(update_main._source_io.os.path, "exists", return_value=False), + patch.object( + update_main._source_io, + "list_existing_resolution_ids", + return_value={"active", "finalized"}, + ), + patch.object(update_main, "KalshiSource") as mock_source_class, + ): + mock_source_class.return_value.get_nullified_ids.return_value = {"nullified"} + mock_source_class.return_value.update.return_value = Mock( + dfq=dfq, + resolution_files={}, + ) + + update_main.driver(None) + + downloaded_paths = [call.kwargs["filename"] for call in mock_download.call_args_list] + assert downloaded_paths == ["kalshi/active.jsonl"] + + +# --------------------------------------------------------------------------- +# End-to-end resolution (constraint 4) +# --------------------------------------------------------------------------- + + +class TestKalshiEndToEnd: + """Question set -> explode -> resolve_all -> dummy forecasts -> impute.""" + + def test_resolution_passes(self, freeze_today): + """A Kalshi question set resolves and dummy forecasts flow through imputation.""" + freeze_today(date(2025, 2, 1)) + + # Build a question set: 2 standard + 1 combo Kalshi question, plus a data + # question to seed the shared resolution date. + question_set_df = make_question_set_df( + [ + {"id": "m1", "source": "kalshi", "resolution_dates": "N/A"}, + {"id": "m2", "source": "kalshi", "resolution_dates": "N/A"}, + {"id": ("m1", "m2"), "source": "kalshi", "resolution_dates": "N/A"}, + {"id": "d1", "source": "fred", "resolution_dates": ["2025-01-08"]}, + ] + ) + + exploded = explode_question_set(question_set_df, "2025-01-01") + exploded = exploded[exploded["source"] == "kalshi"].copy() + assert len(exploded) > 0 + + # Question bank: market resolves to yesterday's (Jan 31) value. + dfq = make_question_df([{"id": "m1", "resolved": False}, {"id": "m2", "resolved": False}]) + dfr = make_resolution_df( + [ + {"id": "m1", "date": "2025-01-01", "value": 0.3}, + {"id": "m1", "date": "2025-01-08", "value": 0.5}, + {"id": "m1", "date": "2025-01-31", "value": 0.7}, + {"id": "m2", "date": "2025-01-01", "value": 0.4}, + {"id": "m2", "date": "2025-01-08", "value": 0.6}, + {"id": "m2", "date": "2025-01-31", "value": 0.8}, + ] + ) + question_bank = {"kalshi": SourceQuestionBank(dfq=dfq, dfr=dfr)} + + resolved, _ = resolve_all( + exploded, + question_bank=question_bank, + sources={"kalshi": SOURCES["kalshi"]}, + forecast_due_date=date(2025, 1, 1), + ) + assert len(resolved) > 0 + assert resolved["resolved_to"].notna().all() + + # Dummy forecasts: m1 provided, m2 missing (to exercise imputation). + forecast_df = pd.DataFrame( + { + "id": ["m1", "m2"], + "source": ["kalshi", "kalshi"], + "direction": [(), ()], + "forecast": [0.65, np.nan], + "resolution_date": ["2025-01-08", "2025-01-08"], + } + ) + prepared = check_and_prepare_forecast_file(forecast_df, "2025-01-01", "test_org") + merged = set_resolution_dates(prepared, resolved) + result = impute_missing_forecasts(merged, "test_org", "test_model_org", "test_model") + + m1_rows = result[result["id"] == "m1"] + assert len(m1_rows) > 0 + assert m1_rows.iloc[0]["forecast"] == 0.65 + assert bool(m1_rows.iloc[0]["imputed"]) is False + + m2_rows = result[result["id"] == "m2"] + assert len(m2_rows) > 0 + assert m2_rows.iloc[0]["forecast"] == 0.5 + assert bool(m2_rows.iloc[0]["imputed"]) is True diff --git a/src/tests/test_types_and_schemas.py b/src/tests/test_types_and_schemas.py index 47b7e1d1..412c53fa 100644 --- a/src/tests/test_types_and_schemas.py +++ b/src/tests/test_types_and_schemas.py @@ -115,6 +115,7 @@ def test_reasoning_column_is_required(self): "dbnomics": SourceType.DATASET, "fred": SourceType.DATASET, "infer": SourceType.MARKET, + "kalshi": SourceType.MARKET, "manifold": SourceType.MARKET, "metaculus": SourceType.MARKET, "polymarket": SourceType.MARKET, diff --git a/src/www.forecastbench.org/about/index.md b/src/www.forecastbench.org/about/index.md index 1a2858e3..e64d864c 100644 --- a/src/www.forecastbench.org/about/index.md +++ b/src/www.forecastbench.org/about/index.md @@ -16,7 +16,7 @@ footer_scripts:

We use two types of binary prediction questions:

ForecastBench operates as a fully automated, dynamic system. New forecasting rounds occur every two weeks, with each round generating 500 questions split evenly between market and dataset questions. The leaderboard is updated nightly as new data becomes available and market questions resolve over time, allowing us to continuously track forecasting performance.