feat(research): share immutable event cohort partition

docs/architecture/PREDICTION_MARKETS.md

@@ -166,6 +166,10 @@ walk-forward evaluator, including the no-prior baseline turn. A missing or
 invalid boundary fails closed; generic factor and token-execution reviews do
 not inherit this settlement split implicitly.
 
+Authenticated event metadata is assigned once to a content-addressed
+`EventCohortPartition` shared by settlement, UP, and DOWN tasks. Crossing events
+are excluded whole; held-out rows and labels remain inaccessible during search.
+
 The broader governed baseline remains BTC/SOL five-minute settlement research
 over the retained one-second full-visible-depth L2 snapshots. The
 continuously-ready catalog and Mission contract above admits only BTC

rust_hft/prediction-markets/config/research_missions/polymarket-btc-5m.example.json

@@ -15,7 +15,7 @@
   "horizon": "5m",
   "time_cohort_boundary_ms": 0,
   "prompt_snapshot_id": "sha256:2b55ba0e724dfc9f5a040911e397058a09c6229e42064e4998739d617b368dcb",
-  "search_policy_snapshot_id": "sha256:c6a3c66c2d6741d65dd9607df1827a5a3b9e4c1cf364f1bb166e706d041f50a6",
+  "search_policy_snapshot_id": "sha256:bd5d11fd6391c60be0ece49bcd9bb6d560121bb96e552cf86fb990782e0f4a39",
   "search_budget": {
     "max_candidates": 6,
     "max_llm_calls": 2,

rust_hft/prediction-markets/config/research_missions/polymarket-sol-5m.example.json

@@ -15,7 +15,7 @@
   "horizon": "5m",
   "time_cohort_boundary_ms": 0,
   "prompt_snapshot_id": "sha256:0816ebccf4c75ee6bdcfe315b253c84ddf1808a38687ebe3b924b87bd72a52a9",
-  "search_policy_snapshot_id": "sha256:c6a3c66c2d6741d65dd9607df1827a5a3b9e4c1cf364f1bb166e706d041f50a6",
+  "search_policy_snapshot_id": "sha256:bd5d11fd6391c60be0ece49bcd9bb6d560121bb96e552cf86fb990782e0f4a39",
   "search_budget": {
     "max_candidates": 6,
     "max_llm_calls": 2,

rust_hft/prediction-markets/crates/ploy-research/src/autofactor.rs

@@ -3,10 +3,13 @@ use std::error::Error;
 use std::fmt;
 use std::sync::OnceLock;
 
-use chrono::{DateTime, Datelike, Duration, Utc};
+use chrono::Datelike;
+#[cfg(test)]
+use chrono::{Duration, Utc};
 use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
 
+use crate::event_cohort_partition::EventCohortPartition;
 use crate::factors::{pearson_ic, spearman_ic};
 use crate::factors_v2::{FactorObservationV2, ReviewSide};
 
@@ -1174,35 +1177,31 @@ pub struct RepricePilotMetrics {
     pub positive_event_ratio: f64,
 }
 
+#[derive(Debug)]
+pub struct RepriceEventCohortSplit<'a> {
+    pub partition: &'a EventCohortPartition,
+    pub train: Vec<&'a FactorObservationV2>,
+    pub held_out: Vec<&'a FactorObservationV2>,
+}
+
 /// Split complete five-minute market episodes at `boundary`.
 ///
 /// The verified Polymarket projection carries its exact `market_id` through
 /// `FactorObservationV2::event_id`; that market identity is the episode key.
 /// Each episode must carry one stable Up token and one stable Down token.
 /// Events crossing the boundary and events without one consistent canonical
 /// end are excluded.
-pub fn split_reprice_rows_by_event_cohort(
-    rows: &[FactorObservationV2],
-    boundary: DateTime<Utc>,
-) -> Result<(Vec<&FactorObservationV2>, Vec<&FactorObservationV2>), AutoFactorError> {
-    let mut event_ends: BTreeMap<&str, Option<DateTime<Utc>>> = BTreeMap::new();
+pub fn split_reprice_rows_by_event_cohort<'a>(
+    rows: &'a [FactorObservationV2],
+    partition: &'a EventCohortPartition,
+) -> Result<RepriceEventCohortSplit<'a>, AutoFactorError> {
     let mut event_tokens: BTreeMap<&str, (BTreeSet<&str>, BTreeSet<&str>)> = BTreeMap::new();
     for row in rows {
         if row.event_id.trim().is_empty() || row.pm_token_id.trim().is_empty() {
             return Err(AutoFactorError::IdentityMismatch(
                 "reprice pilot requires a non-empty market_id event_id and token id".to_string(),
             ));
         }
-        match event_ends.entry(row.event_id.as_str()) {
-            std::collections::btree_map::Entry::Vacant(entry) => {
-                entry.insert(row.event_end_ts);
-            }
-            std::collections::btree_map::Entry::Occupied(mut entry) => {
-                if *entry.get() != row.event_end_ts {
-                    entry.insert(None);
-                }
-            }
-        }
         let tokens = event_tokens.entry(row.event_id.as_str()).or_default();
         match row.side {
             ReviewSide::Up => {
@@ -1221,27 +1220,31 @@ pub fn split_reprice_rows_by_event_cohort(
         }
     }
 
-    let window = Duration::seconds(300);
+    if rows.iter().any(|row| {
+        !partition.contains_train_market(&row.event_id)
+            && !partition.contains_held_out_market(&row.event_id)
+            && !partition
+                .crossing_excluded()
+                .iter()
+                .any(|event| event.market_id == row.event_id)
+    }) {
+        return Err(AutoFactorError::IdentityMismatch(
+            "reprice row market_id is absent from the shared partition".into(),
+        ));
+    }
     let train = rows
         .iter()
-        .filter(|row| {
-            event_ends
-                .get(row.event_id.as_str())
-                .and_then(Option::as_ref)
-                .is_some_and(|event_end| *event_end < boundary)
-        })
+        .filter(|row| partition.contains_train_market(&row.event_id))
         .collect();
-    let test = rows
+    let held_out = rows
         .iter()
-        .filter(|row| {
-            event_ends
-                .get(row.event_id.as_str())
-                .and_then(Option::as_ref)
-                .and_then(|event_end| event_end.checked_sub_signed(window))
-                .is_some_and(|event_start| event_start >= boundary)
-        })
+        .filter(|row| partition.contains_held_out_market(&row.event_id))
         .collect();
-    Ok((train, test))
+    Ok(RepriceEventCohortSplit {
+        partition,
+        train,
+        held_out,
+    })
 }
 
 /// Fit one factor's threshold using training rows only. `target` must be a
@@ -3586,15 +3589,12 @@ mod tests {
                 -1.0,
             ),
         ];
-        let excluded_rows = [
-            row(
-                "crossing",
-                Some(boundary + Duration::seconds(1)),
-                50.0,
-                50.0,
-            ),
-            row("missing-end", None, 60.0, 60.0),
-        ];
+        let excluded_rows = [row(
+            "crossing",
+            Some(boundary + Duration::seconds(1)),
+            50.0,
+            50.0,
+        )];
         let mut rows = train_rows
             .into_iter()
             .chain(test_rows)
@@ -3610,28 +3610,34 @@ mod tests {
             })
             .collect::<Vec<_>>();
         rows.extend(down_rows);
+        let partition =
+            EventCohortPartition::from_test_observations(&rows, boundary.timestamp_millis(), 300)
+                .unwrap();
         assert!(split_reprice_rows_by_event_cohort(
             &[row(
                 "unpaired-market",
                 Some(boundary - Duration::seconds(1)),
                 1.0,
                 1.0,
             )],
-            boundary,
+            &partition,
         )
         .expect_err("a market episode without both tokens must fail closed")
         .to_string()
         .contains("exactly one Up and one Down token"));
 
-        let (train, test) =
-            split_reprice_rows_by_event_cohort(&rows, boundary).expect("paired market episodes");
+        let split =
+            split_reprice_rows_by_event_cohort(&rows, &partition).expect("paired market episodes");
+        assert_eq!(split.partition.digest(), partition.digest());
+        let train = &split.train;
+        let test = &split.held_out;
         assert_eq!(train.len(), 8);
         assert_eq!(test.len(), 4);
         assert!(train.iter().all(|row| row.event_id.starts_with("train")));
         assert!(test.iter().all(|row| row.event_id.starts_with("test")));
 
-        let train = train.into_iter().cloned().collect::<Vec<_>>();
-        let test = test.into_iter().cloned().collect::<Vec<_>>();
+        let train = train.iter().map(|row| (*row).clone()).collect::<Vec<_>>();
+        let test = test.iter().map(|row| (*row).clone()).collect::<Vec<_>>();
         let reports = mine_domain_autofactors_from_v2(
             &train,
             target,

rust_hft/prediction-markets/crates/ploy-research/src/bin/monday-prediction-evaluator.rs

@@ -43,8 +43,8 @@ use ploy_research::{
     write_alpha_search_artifacts_with_state_and_runtime_feedback,
     write_side_bound_alpha_search_artifacts_with_state_and_runtime_feedback,
     AlphaSearchArtifactSummary, AlphaSearchRuntimeFeedback, AlphaZooSnapshot, AutoFactorOptions,
-    AutoFactorV2Target, FactorComboV1Options, FactorObservation, FactorObservationV2,
-    FactorReviewOptions, FactorStabilityOptions, FactorWalkForwardOptions,
+    AutoFactorV2Target, EventCohortPartition, FactorComboV1Options, FactorObservation,
+    FactorObservationV2, FactorReviewOptions, FactorStabilityOptions, FactorWalkForwardOptions,
     FillabilityReviewOptions, FullDepthExecutionMatrixOptions, FullDepthExecutionMatrixReport,
     LiquidityGateV1Options, LiquidityGatedAlphaV1Options, LlmPriorSpec,
     MetaLabelWalkForwardOptions, RepricePilotMetrics, RepricePilotSelection, RepricingIcOptions,
@@ -127,6 +127,7 @@ struct RepricePilotSearchArtifact {
 #[derive(Clone, serde::Serialize)]
 struct RepricePilotEpisodeCohorts {
     key: &'static str,
+    partition_digest: String,
     train_market_ids: Vec<String>,
     test_market_ids: Vec<String>,
 }
@@ -443,7 +444,7 @@ fn sorted_distinct_reprice_pilot_market_ids<'a>(
 #[allow(clippy::too_many_arguments)]
 fn run_reprice_pilot_10s(
     rows: &[FactorObservationV2],
-    boundary: DateTime<Utc>,
+    partition: &EventCohortPartition,
     alpha_search_output_dir: &Path,
     report_output_dir: &Path,
     snapshot_hash: &str,
@@ -458,10 +459,18 @@ fn run_reprice_pilot_10s(
     min_observations: usize,
     min_full_depth_entry_fill_rate: f64,
 ) -> Result<[PathBuf; 2], String> {
-    let (train_refs, test_refs) = split_reprice_rows_by_event_cohort(rows, boundary)
+    let split = split_reprice_rows_by_event_cohort(rows, partition)
         .map_err(|error| format!("validate reprice market episodes: {error}"))?;
-    let train_rows = train_refs.into_iter().cloned().collect::<Vec<_>>();
-    let test_rows = test_refs.into_iter().cloned().collect::<Vec<_>>();
+    let train_rows = split
+        .train
+        .iter()
+        .map(|row| (*row).clone())
+        .collect::<Vec<_>>();
+    let test_rows = split
+        .held_out
+        .iter()
+        .map(|row| (*row).clone())
+        .collect::<Vec<_>>();
     let excluded_rows = rows
         .len()
         .saturating_sub(train_rows.len().saturating_add(test_rows.len()));
@@ -474,6 +483,7 @@ fn run_reprice_pilot_10s(
     }
     let episode_cohorts = RepricePilotEpisodeCohorts {
         key: "polymarket market_id carried as FactorObservationV2.event_id",
+        partition_digest: split.partition.digest().to_string(),
         train_market_ids: sorted_distinct_reprice_pilot_market_ids(
             train_rows.iter().map(|row| row.event_id.as_str()),
         ),
@@ -1524,10 +1534,11 @@ async fn main() {
     let snapshot_source_kind: String;
     let settlement_component_profile: SettlementProbabilityComponentProfile;
     let include_deribit: bool;
-    let (observations, deribit_snapshots, all_pm_book_snapshots): (
+    let (observations, deribit_snapshots, all_pm_book_snapshots, event_cohort_partition): (
         Vec<FactorObservation>,
         Vec<_>,
         Vec<_>,
+        Option<EventCohortPartition>,
     ) = {
         let started = std::time::Instant::now();
         let snapshot =
@@ -1590,6 +1601,18 @@ async fn main() {
         );
         snapshot_data_audit_status = snapshot.manifest.data_audit_status.clone();
         include_deribit = snapshot.manifest.include_deribit;
+        let event_cohort_partition = settlement_time_cohort
+            .as_ref()
+            .map(|cohort| {
+                EventCohortPartition::from_verified_snapshot(
+                    &snapshot,
+                    &symbols,
+                    cohort.event_window_secs(),
+                    cohort.boundary().timestamp_millis(),
+                )
+            })
+            .transpose()
+            .unwrap_or_else(|reason| panic!("verified snapshot cohort invalid: {reason}"));
         snapshot_provenance = format!(
             "# Snapshot\nsnapshot_schema={}\nsnapshot_hash={}\nsnapshot_contract_hash={}\nsnapshot_generated_at={}\nsnapshot_optimizer_data_dir={}\nsnapshot_data_requirements={}\nsnapshot_data_audit_status={}\nsnapshot_data_audit_report={}\nsnapshot_include_deribit={}\n",
             snapshot.manifest.schema_version,
@@ -1652,7 +1675,21 @@ async fn main() {
             deribit_snapshots.len(),
             pm_book_snapshots.len()
         );
-        (observations, deribit_snapshots, pm_book_snapshots)
+        (
+            observations,
+            deribit_snapshots,
+            pm_book_snapshots,
+            event_cohort_partition,
+        )
+    };
+    let settlement_time_cohort = match (settlement_time_cohort, &event_cohort_partition) {
+        (Some(cohort), Some(partition)) => Some(
+            cohort
+                .with_partition(partition.clone())
+                .unwrap_or_else(|reason| panic!("attach event cohort partition: {reason}")),
+        ),
+        (None, None) => None,
+        _ => panic!("time cohort and event cohort partition must be present together"),
     };
 
     if report_output_dir.is_some() {
@@ -1745,8 +1782,16 @@ async fn main() {
         let prior = governed_prediction_prior
             .filter(|prior| prior.probability_blends.len() == 1)
             .expect("training candidate prior was validated");
+        let partition = event_cohort_partition
+            .as_ref()
+            .expect("prediction MCTS training requires the verified event partition");
+        let training_rows = autofactor_rows
+            .iter()
+            .filter(|row| partition.contains_train_market(&row.event_id))
+            .cloned()
+            .collect::<Vec<_>>();
         let training = build_settlement_training_probability_report_with_prior(
-            &autofactor_rows,
+            &training_rows,
             start,
             end,
             Some(prior),
@@ -1757,10 +1802,11 @@ async fn main() {
                     component_profile: settlement_component_profile,
                     ..Default::default()
                 },
-                time_cohort: settlement_time_cohort,
+                time_cohort: settlement_time_cohort.clone(),
                 ..Default::default()
             },
-        );
+        )
+        .unwrap_or_else(|reason| panic!("build settlement training evidence: {reason}"));
         let model = format!("q_llm_{}", candidate.probability_blend.name);
         let metrics = training
             .baselines
@@ -1800,14 +1846,13 @@ async fn main() {
     if reprice_pilot_10s {
         let boundary_ms = time_cohort_boundary_ms
             .expect("--reprice-pilot-10s validated --time-cohort-boundary-ms");
-        let boundary = Utc
-            .timestamp_millis_opt(boundary_ms)
-            .single()
-            .expect("--reprice-pilot-10s validated time cohort boundary");
+        let partition = event_cohort_partition
+            .as_ref()
+            .expect("reprice pilot requires the verified event partition");
         println!("{snapshot_provenance}");
         let paths = run_reprice_pilot_10s(
             &autofactor_rows,
-            boundary,
+            partition,
             Path::new(
                 alpha_search_output_dir
                     .as_deref()