fix: remove infer as a source

Makefile

@@ -157,10 +157,7 @@ metaculus-fetch:
 metaculus-update-questions:
 	$(MAKE) -C src/orchestration/func_metaculus_update || echo "* $@" >> $(MAKE_FAILURE_LOG)
 
-infer: infer-fetch infer-update-questions
-
-infer-fetch:
-	$(MAKE) -C src/orchestration/func_infer_fetch || echo "* $@" >> $(MAKE_FAILURE_LOG)
+infer: infer-update-questions
 
 infer-update-questions:
 	$(MAKE) -C src/orchestration/func_infer_update || echo "* $@" >> $(MAKE_FAILURE_LOG)

src/_schemas.py

@@ -100,14 +100,6 @@ class Config:
         coerce = False
 
 
-class InferFetchFrame(QuestionFrame):
-    """Output of InferSource.fetch(). QuestionFrame plus transient fields for update()."""
-
-    fetch_datetime: Series[str]
-    probability: Series[object] = pa.Field(nullable=True)
-    nullify_question: Series[bool]
-
-
 class PolymarketFetchFrame(QuestionFrame):
     """Output of PolymarketSource.fetch(). QuestionFrame plus transient fields for update()."""
 

src/curate_questions/create_question_set/main.py

@@ -163,6 +163,7 @@ def process_questions(
             df_all_available,
             total_market_requested,
             source_name="ALL SOURCES",
+            source_counts=[(source, got, want) for source, got, want, _ in source_summaries],
         )
 
     log_sampling_summary(
@@ -561,6 +562,7 @@ def plot_sampling_distribution(
     df_available: pd.DataFrame,
     n_target: int,
     source_name: str | None = None,
+    source_counts: list[tuple[str, int, int]] | None = None,
 ) -> None:
     """Plot the realized vs expected distribution for market question sampling.
 
@@ -572,6 +574,8 @@ def plot_sampling_distribution(
         df_available (pd.DataFrame): All available questions (must include bin columns)
         n_target (int): Number of questions requested
         source_name (str | None): Name for the source (used in chart title)
+        source_counts (list[tuple[str, int, int]] | None): (source, sampled, target) per source,
+            charted as an extra row when given
     """
     if not env.RUNNING_LOCALLY:
         return
@@ -589,11 +593,19 @@ def plot_sampling_distribution(
     title = "Sampling Distribution"
     if source_name:
         title = f"Sampling Distribution: {source_name}"
+    title += (
+        f"<br><sub>{len(df_sampled):,}/{n_target:,} sampled "
+        f"from {len(df_available):,} available</sub>"
+    )
+
+    subplot_titles = ["Market Value", "Time Horizon", "Category"]
+    if source_counts:
+        subplot_titles.append("Questions by Source")
 
     fig = make_subplots(
-        rows=3,
+        rows=len(subplot_titles),
         cols=1,
-        subplot_titles=("Market Value", "Time Horizon", "Category"),
+        subplot_titles=tuple(subplot_titles),
         vertical_spacing=0.1,
     )
 
@@ -733,9 +745,43 @@ def add_line_chart(
     )
     fig.update_xaxes(tickangle=45, row=3, col=1)
 
+    # Number of questions sampled from each source
+    if source_counts:
+        source_row = len(subplot_titles)
+        source_labels = [source for source, _, _ in source_counts]
+        # The target bar is never shorter than the selected bar, so labeling it keeps the
+        # sampled/target counts clear of both bars.
+        fig.add_trace(
+            go.Bar(
+                name="Target",
+                x=source_labels,
+                y=[target for _, _, target in source_counts],
+                text=[f"{sampled}/{target}" for _, sampled, target in source_counts],
+                textposition="outside",
+                marker=dict(color="coral", opacity=0.5),
+                legendgroup="target",
+                showlegend=False,
+            ),
+            row=source_row,
+            col=1,
+        )
+        fig.add_trace(
+            go.Bar(
+                name="Selected",
+                x=source_labels,
+                y=[sampled for _, sampled, _ in source_counts],
+                marker=dict(color="steelblue"),
+                legendgroup="selected",
+                showlegend=False,
+            ),
+            row=source_row,
+            col=1,
+        )
+        fig.update_xaxes(tickangle=45, row=source_row, col=1)
+
     fig.update_layout(
         title_text=title,
-        height=1000,
+        height=250 + 250 * len(subplot_titles),
         barmode="overlay",
         legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1),
     )

src/helpers/keys.py

@@ -35,7 +35,6 @@ def get_secret_that_may_not_exist(secret_name, version_id="latest"):
 # QUESTION MARKET SOURCES
 API_KEY_METACULUS = get_secret(secret_name="API_KEY_METACULUS")
 API_KEY_POLYMARKET = get_secret("API_KEY_POLYMARKET")
-API_KEY_INFER = get_secret("API_KEY_INFER")
 
 # WORKFLOW BOT
 API_SLACK_BOT_NOTIFICATION = get_secret(secret_name="API_SLACK_BOT_NOTIFICATION")

src/helpers/question_curation.py

@@ -9,7 +9,6 @@
     dates,
     dbnomics,
     fred,
-    infer,
     manifold,
     metaculus,
     polymarket,
@@ -28,8 +27,9 @@
 
 
 FREEZE_QUESTION_MARKET_SOURCES = {
-    # If market sources are ever removed, the key must be added to MARKET_SOURCES in
-    # `helpers/resolution.py` as the resolution code needs all old market sources.
+    # The market sources we sample questions from. Dropping a source here stops sampling it
+    # without affecting resolution: `helpers/resolution.py` takes its source lists from
+    # `sources/_metadata.py`, which holds every market source we've ever published questions for.
     "manifold": {
         "name": "Manifold",
         "source_intro": manifold.SOURCE_INTRO,
@@ -40,11 +40,6 @@
         "source_intro": metaculus.SOURCE_INTRO,
         "resolution_criteria": metaculus.RESOLUTION_CRITERIA,
     },
-    "infer": {
-        "name": "INFER",
-        "source_intro": infer.SOURCE_INTRO,
-        "resolution_criteria": infer.RESOLUTION_CRITERIA,
-    },
     "polymarket": {
         "name": "Polymarket",
         "source_intro": polymarket.SOURCE_INTRO,
@@ -84,7 +79,6 @@
 
 DATA_SOURCES = list(FREEZE_QUESTION_DATA_SOURCES.keys())
 MARKET_SOURCES = list(FREEZE_QUESTION_MARKET_SOURCES.keys())
-ALL_SOURCES = DATA_SOURCES + MARKET_SOURCES
 
 FREEZE_WINDOW_IN_DAYS = 10
 

src/leaderboard/main.py

@@ -34,14 +34,14 @@
     decorator,
     env,
     git,
-    question_curation,
     resolution,
     slack,
 )
 from llm_forecaster.forecast_variants import (  # noqa: E402
     ALL_FORECAST_VARIANT_KEYS_WITH_CONTEXT,
     ALL_FORECAST_VARIANT_KEYS_WITHOUT_CONTEXT,
 )
+from sources import DATASET_SOURCE_NAMES, MARKET_SOURCE_NAMES  # noqa: E402
 
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
@@ -382,15 +382,15 @@ def download_question_set_save_in_cache(
 
 
 def get_dataset_mask(df: pd.DataFrame) -> pd.Series:
-    """Generate boolean masks for market questions.
+    """Generate boolean masks for dataset questions.
 
     Args:
         df (pd.DataFrame): The forecast set.
 
     Returns:
-        pd.Series: questions from DATA_SOURCES
+        pd.Series: questions from DATASET_SOURCE_NAMES
     """
-    return df["source"].isin(question_curation.DATA_SOURCES)
+    return df["source"].isin(DATASET_SOURCE_NAMES)
 
 
 def get_market_mask(df: pd.DataFrame) -> pd.Series:
@@ -400,9 +400,9 @@ def get_market_mask(df: pd.DataFrame) -> pd.Series:
         df (pd.DataFrame): The forecast set.
 
     Returns:
-        pd.Series: all questions from MARKET_SOURCES.
+        pd.Series: all questions from MARKET_SOURCE_NAMES.
     """
-    return df["source"].isin(question_curation.MARKET_SOURCES)
+    return df["source"].isin(MARKET_SOURCE_NAMES)
 
 
 def get_masks(df: pd.DataFrame) -> Dict[str, pd.Series]:
@@ -413,8 +413,8 @@ def get_masks(df: pd.DataFrame) -> Dict[str, pd.Series]:
 
     Returns:
         Dict[str, pd.Series]: Mapping of mask names to boolean Series:
-            - "dataset":      questions from DATA_SOURCES that are resolved.
-            - "market":       all questions from MARKET_SOURCES.
+            - "dataset":      questions from DATASET_SOURCE_NAMES that are resolved.
+            - "market":       all questions from MARKET_SOURCE_NAMES.
             - "market_resolved":   market questions that are resolved.
             - "market_unresolved": market questions that are unresolved.
     """

src/metadata/validate_questions/main.py

@@ -117,7 +117,6 @@ def driver(_):
         dfmeta = dfmeta[dfmeta["source"] != source]
 
         if source in question_curation.DATA_SOURCES + [
-            "infer",
             "metaculus",
         ]:
             dfq["valid_question"] = True

src/nightly_update_workflow/manager/main.py

@@ -9,11 +9,22 @@
 
 sys.path.append(os.path.join(os.path.dirname(__file__), "../.."))
 from helpers import cloud_run, constants, env, question_curation, slack  # noqa: E402
+from sources import ALL_SOURCE_NAMES  # noqa: E402
 
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
 
+def get_fetch_and_update_task_count() -> int:
+    """Return the number of worker tasks needed to fetch and update every source.
+
+    The worker runs one task per source, so this must cover the longest job list the worker can
+    produce on any day. Sized from the source metadata rather than the question-curation lists,
+    which only cover the sources we still sample questions from.
+    """
+    return len(ALL_SOURCE_NAMES)
+
+
 def call_worker(dict_to_use, task_count, timeout=cloud_run.timeout_1h):
     """Make main() easier to read."""
     return cloud_run.call_worker(
@@ -33,7 +44,7 @@ def summarize_question_bank():
         lines=True,
     )[["id", "source", "valid_question"]]
     df = pd.DataFrame()
-    for source in sorted(question_curation.ALL_SOURCES):
+    for source in ALL_SOURCE_NAMES:
         logger.info(f"downloading {source} question file.")
         dfq = pd.read_json(
             f"gs://{env.QUESTION_BANK_BUCKET}/{source}_questions.jsonl",
@@ -105,9 +116,7 @@ def main():
 
     dict_to_use = "fetch_and_update"
     timeout_fetch_and_update = cloud_run.timeout_1h * 6
-    task_count = len(question_curation.FREEZE_QUESTION_DATA_SOURCES) + len(
-        question_curation.FREEZE_QUESTION_MARKET_SOURCES
-    )
+    task_count = get_fetch_and_update_task_count()
     operation = call_worker(
         dict_to_use=dict_to_use,
         task_count=task_count,

src/nightly_update_workflow/worker/main.py

@@ -5,6 +5,7 @@
 
 sys.path.append(os.path.join(os.path.dirname(__file__), "../.."))
 from helpers import cloud_run, dates, question_curation  # noqa: E402
+from sources import ALL_SOURCE_NAMES, SOURCE_METADATA  # noqa: E402
 
 metadata = [
     [
@@ -88,29 +89,24 @@ def get_publish_question_set_make_llm_baseline():
 
 def get_fetch_and_update():
     """Dynamically add acled to list of functions to call dending on the day of the week."""
-    sources = [
-        "dbnomics",
-        "fred",
-        "infer",
-        "manifold",
-        "metaculus",
-        "polymarket",
-        "wikipedia",
-        "yfinance",
-    ]
+    sources = [source for source in ALL_SOURCE_NAMES if source != "acled"]
     day_of_week = dates.get_datetime_today().strftime("%A")
     if day_of_week in ["Wednesday"]:
         # Fetch ACLED data on Wednesdays. See Issue #115.
         sources += [
             "acled",
         ]
-    return [
-        [
-            (f"func-data-{source}-fetch", True, cloud_run.timeout_1h * 3, 1),
-            (f"func-data-{source}-update-questions", True, cloud_run.timeout_1h * 3, 1),
-        ]
-        for source in sources
-    ]
+    jobs = []
+    for source in sources:
+        group = []
+        if SOURCE_METADATA[source]["run_fetch"]:
+            group.append((f"func-data-{source}-fetch", True, cloud_run.timeout_1h * 3, 1))
+        if SOURCE_METADATA[source]["run_update"]:
+            group.append(
+                (f"func-data-{source}-update-questions", True, cloud_run.timeout_1h * 3, 1)
+            )
+        jobs.append(group)
+    return jobs
 
 
 def sequential_cloud_run_jobs(functions_to_call):

src/orchestration/_io.py

@@ -25,7 +25,7 @@
 from _schemas import AcledResolutionFrame, QuestionFrame, ResolutionFrame
 from helpers import data_utils, dates, env
 from helpers.run_mode import RunMode
-from sources import ALL_SOURCE_NAMES, MARKET_SOURCE_NAMES
+from sources import ALL_SOURCE_NAMES, MARKET_SOURCE_NAMES, SOURCE_METADATA
 from sources._base import BaseSource
 
 logging.basicConfig(level=logging.INFO)
@@ -110,6 +110,10 @@ def load_question_bank(sources_to_get: list[str] | None = None) -> QuestionBank:
     # Check market dfq files are up-to-date
     any_out_of_date_dfq = False
     for source in MARKET_SOURCE_NAMES:
+        if not SOURCE_METADATA[source]["run_fetch"]:
+            # Sources we no longer fetch have an intentionally frozen dfq and hence will be
+            # out of date.
+            continue
         last_updated_dfq = data_utils.get_last_modified_time_of_dfq_from_cloud_storage(source)
         any_out_of_date_dfq |= last_updated_dfq is None or last_updated_dfq.date() < today
         if last_updated_dfq is None or last_updated_dfq.date() < today: