diff --git a/docs/architecture/PREDICTION_MARKETS.md b/docs/architecture/PREDICTION_MARKETS.md index 50547cf6b..184e56b59 100644 --- a/docs/architecture/PREDICTION_MARKETS.md +++ b/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 diff --git a/rust_hft/prediction-markets/config/research_missions/polymarket-btc-5m.example.json b/rust_hft/prediction-markets/config/research_missions/polymarket-btc-5m.example.json index 02c7966a8..53097d8a7 100644 --- a/rust_hft/prediction-markets/config/research_missions/polymarket-btc-5m.example.json +++ b/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, diff --git a/rust_hft/prediction-markets/config/research_missions/polymarket-sol-5m.example.json b/rust_hft/prediction-markets/config/research_missions/polymarket-sol-5m.example.json index 5c1949ff1..4fcbcc2f6 100644 --- a/rust_hft/prediction-markets/config/research_missions/polymarket-sol-5m.example.json +++ b/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, diff --git a/rust_hft/prediction-markets/crates/ploy-research/src/autofactor.rs b/rust_hft/prediction-markets/crates/ploy-research/src/autofactor.rs index 1bcbae563..0a9613286 100644 --- a/rust_hft/prediction-markets/crates/ploy-research/src/autofactor.rs +++ b/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,6 +1177,13 @@ 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 @@ -1181,11 +1191,10 @@ pub struct RepricePilotMetrics { /// 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, -) -> Result<(Vec<&FactorObservationV2>, Vec<&FactorObservationV2>), AutoFactorError> { - let mut event_ends: BTreeMap<&str, Option>> = BTreeMap::new(); +pub fn split_reprice_rows_by_event_cohort<'a>( + rows: &'a [FactorObservationV2], + partition: &'a EventCohortPartition, +) -> Result, 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() { @@ -1193,16 +1202,6 @@ pub fn split_reprice_rows_by_event_cohort( "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,6 +3610,9 @@ mod tests { }) .collect::>(); 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", @@ -3617,21 +3620,24 @@ mod tests { 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::>(); - let test = test.into_iter().cloned().collect::>(); + let train = train.iter().map(|row| (*row).clone()).collect::>(); + let test = test.iter().map(|row| (*row).clone()).collect::>(); let reports = mine_domain_autofactors_from_v2( &train, target, diff --git a/rust_hft/prediction-markets/crates/ploy-research/src/bin/monday-prediction-evaluator.rs b/rust_hft/prediction-markets/crates/ploy-research/src/bin/monday-prediction-evaluator.rs index e85ca0529..e94f1d1c2 100644 --- a/rust_hft/prediction-markets/crates/ploy-research/src/bin/monday-prediction-evaluator.rs +++ b/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, test_market_ids: Vec, } @@ -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, + 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::>(); - let test_rows = test_refs.into_iter().cloned().collect::>(); + let train_rows = split + .train + .iter() + .map(|row| (*row).clone()) + .collect::>(); + let test_rows = split + .held_out + .iter() + .map(|row| (*row).clone()) + .collect::>(); 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, Vec<_>, Vec<_>, + Option, ) = { 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::>(); 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() diff --git a/rust_hft/prediction-markets/crates/ploy-research/src/event_cohort_partition.rs b/rust_hft/prediction-markets/crates/ploy-research/src/event_cohort_partition.rs new file mode 100644 index 000000000..9128f55fe --- /dev/null +++ b/rust_hft/prediction-markets/crates/ploy-research/src/event_cohort_partition.rs @@ -0,0 +1,337 @@ +use std::collections::BTreeMap; + +use chrono::Duration; +use serde::Serialize; + +use crate::prediction_loop::validate_sha256_id; +use crate::prediction_loop_fs::{canonical_json_bytes, sha256_hex}; +use crate::research_snapshot::ResearchSnapshot; + +pub const EVENT_COHORT_PARTITION_VERSION: &str = "event_cohort_partition.v1"; + +#[derive(Debug, Clone, PartialEq, Eq)] +struct EventCohortMetadata { + market_id: String, + reference_path_start_ms: i64, + reference_path_end_ms: i64, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum EventCohortExclusionReason { + ReferencePathCrossesBoundary, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub struct EventCohortExclusion { + pub market_id: String, + pub reason: EventCohortExclusionReason, +} + +/// Immutable, content-addressed assignment shared by every research task. +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub struct EventCohortPartition { + schema_version: &'static str, + authenticated_snapshot_digest: String, + common_time_boundary_ms: i64, + train_market_ids: Vec, + crossing_excluded: Vec, + held_out_market_ids: Vec, + digest: String, +} + +#[derive(Serialize)] +struct EventCohortPartitionPayload<'a> { + schema_version: &'static str, + authenticated_snapshot_digest: &'a str, + common_time_boundary_ms: i64, + train_market_ids: &'a [String], + crossing_excluded: &'a [EventCohortExclusion], + held_out_market_ids: &'a [String], +} + +impl EventCohortPartition { + /// Build once from a snapshot whose loader has rehashed every referenced + /// artifact. Consumers receive this partition, never the source rows. + pub fn from_verified_snapshot( + snapshot: &ResearchSnapshot, + symbols: &[String], + event_window_secs: i64, + common_time_boundary_ms: i64, + ) -> Result { + let authenticated_snapshot_digest = snapshot + .manifest + .snapshot_contract_hash + .as_deref() + .or(snapshot.manifest.snapshot_hash.as_deref()) + .ok_or_else(|| "verified snapshot is missing its authenticated digest".to_string())?; + validate_sha256_id( + authenticated_snapshot_digest, + "authenticated snapshot digest", + )?; + let window = Duration::try_seconds(event_window_secs) + .filter(|window| *window > Duration::zero()) + .ok_or_else(|| "event cohort window must be positive".to_string())?; + let mut event_ends = BTreeMap::new(); + for row in snapshot + .observations + .iter() + .filter(|row| symbols.contains(&row.symbol)) + .filter(|row| row.event_window_secs == event_window_secs) + { + 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 events = event_ends + .into_iter() + .map(|(market_id, event_end)| { + let event_end = event_end.ok_or_else(|| { + format!("market_id {market_id} has no consistent canonical event end") + })?; + let event_start = event_end.checked_sub_signed(window).ok_or_else(|| { + format!("market_id {market_id} reference path start overflows") + })?; + Ok(EventCohortMetadata { + market_id: market_id.to_string(), + reference_path_start_ms: event_start.timestamp_millis(), + reference_path_end_ms: event_end.timestamp_millis(), + }) + }) + .collect::, String>>()?; + Self::build( + authenticated_snapshot_digest, + events, + common_time_boundary_ms, + ) + } + + #[cfg(test)] + pub(crate) fn from_test_observations( + rows: &[crate::factors_v2::FactorObservationV2], + boundary_ms: i64, + event_window_secs: i64, + ) -> Result { + let window = Duration::seconds(event_window_secs); + let events = rows + .iter() + .map(|row| { + let event_end = row + .event_end_ts + .ok_or_else(|| "test observation lacks event end".to_string())?; + Ok(EventCohortMetadata { + market_id: row.event_id.clone(), + reference_path_start_ms: (event_end - window).timestamp_millis(), + reference_path_end_ms: event_end.timestamp_millis(), + }) + }) + .collect::, String>>()?; + let mut unique = BTreeMap::new(); + for event in events { + unique.entry(event.market_id.clone()).or_insert(event); + } + Self::build( + "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + unique.into_values(), + boundary_ms, + ) + } + + fn build( + authenticated_snapshot_digest: &str, + authenticated_events: impl IntoIterator, + common_time_boundary_ms: i64, + ) -> Result { + if common_time_boundary_ms <= 0 { + return Err( + "common time boundary must be a positive Unix millisecond timestamp".into(), + ); + } + let mut events = authenticated_events.into_iter().collect::>(); + events.sort_by(|left, right| left.market_id.cmp(&right.market_id)); + + let mut train_market_ids = Vec::new(); + let mut crossing_excluded = Vec::new(); + let mut held_out_market_ids = Vec::new(); + let mut previous_market_id: Option<&str> = None; + for event in &events { + if event.market_id.trim().is_empty() || event.market_id.trim() != event.market_id { + return Err("event cohort market_id must be a trimmed non-empty string".into()); + } + if previous_market_id == Some(event.market_id.as_str()) { + return Err(format!( + "event cohort contains duplicate market_id {}", + event.market_id + )); + } + if event.reference_path_start_ms >= event.reference_path_end_ms { + return Err(format!( + "event cohort market_id {} has an invalid reference path", + event.market_id + )); + } + previous_market_id = Some(event.market_id.as_str()); + + if event.reference_path_end_ms < common_time_boundary_ms { + train_market_ids.push(event.market_id.clone()); + } else if event.reference_path_start_ms >= common_time_boundary_ms { + held_out_market_ids.push(event.market_id.clone()); + } else { + crossing_excluded.push(EventCohortExclusion { + market_id: event.market_id.clone(), + reason: EventCohortExclusionReason::ReferencePathCrossesBoundary, + }); + } + } + let payload = EventCohortPartitionPayload { + schema_version: EVENT_COHORT_PARTITION_VERSION, + authenticated_snapshot_digest, + common_time_boundary_ms, + train_market_ids: &train_market_ids, + crossing_excluded: &crossing_excluded, + held_out_market_ids: &held_out_market_ids, + }; + let digest = format!("sha256:{}", sha256_hex(&canonical_json_bytes(&payload)?)); + Ok(Self { + schema_version: EVENT_COHORT_PARTITION_VERSION, + authenticated_snapshot_digest: authenticated_snapshot_digest.to_string(), + common_time_boundary_ms, + train_market_ids, + crossing_excluded, + held_out_market_ids, + digest, + }) + } + + pub fn common_time_boundary_ms(&self) -> i64 { + self.common_time_boundary_ms + } + + pub fn train_market_ids(&self) -> &[String] { + &self.train_market_ids + } + + pub fn crossing_excluded(&self) -> &[EventCohortExclusion] { + &self.crossing_excluded + } + + pub fn held_out_market_ids(&self) -> &[String] { + &self.held_out_market_ids + } + + pub fn digest(&self) -> &str { + &self.digest + } + + pub fn contains_train_market(&self, market_id: &str) -> bool { + self.train_market_ids + .binary_search_by(|candidate| candidate.as_str().cmp(market_id)) + .is_ok() + } + + pub fn contains_held_out_market(&self, market_id: &str) -> bool { + self.held_out_market_ids + .binary_search_by(|candidate| candidate.as_str().cmp(market_id)) + .is_ok() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const SNAPSHOT_DIGEST: &str = + "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"; + + #[test] + fn authenticated_events_build_one_stable_common_time_partition() { + let events = vec![ + EventCohortMetadata { + market_id: "held-out".to_string(), + reference_path_start_ms: 2_000, + reference_path_end_ms: 2_500, + }, + EventCohortMetadata { + market_id: "crossing".to_string(), + reference_path_start_ms: 900, + reference_path_end_ms: 1_100, + }, + EventCohortMetadata { + market_id: "train".to_string(), + reference_path_start_ms: 100, + reference_path_end_ms: 999, + }, + ]; + + let partition = + EventCohortPartition::build(SNAPSHOT_DIGEST, events.clone(), 1_000).unwrap(); + let reordered = + EventCohortPartition::build(SNAPSHOT_DIGEST, events.into_iter().rev(), 1_000).unwrap(); + + assert_eq!(partition.common_time_boundary_ms(), 1_000); + assert_eq!(partition.train_market_ids(), ["train"]); + assert_eq!( + partition.crossing_excluded(), + [EventCohortExclusion { + market_id: "crossing".to_string(), + reason: EventCohortExclusionReason::ReferencePathCrossesBoundary, + }] + ); + assert_eq!(partition.held_out_market_ids(), ["held-out"]); + assert_eq!(partition.digest(), reordered.digest()); + } + + #[test] + fn one_market_cannot_enter_more_than_one_partition() { + let duplicate = EventCohortMetadata { + market_id: "same-market".to_string(), + reference_path_start_ms: 100, + reference_path_end_ms: 999, + }; + + let error = + EventCohortPartition::build(SNAPSHOT_DIGEST, [duplicate.clone(), duplicate], 1_000) + .expect_err("duplicate market identity must fail closed"); + + assert!(error.contains("duplicate market_id same-market")); + } + + #[test] + fn reference_path_touching_or_crossing_boundary_is_excluded() { + let partition = EventCohortPartition::build( + SNAPSHOT_DIGEST, + [ + EventCohortMetadata { + market_id: "ends-at-boundary".to_string(), + reference_path_start_ms: 500, + reference_path_end_ms: 1_000, + }, + EventCohortMetadata { + market_id: "starts-at-boundary".to_string(), + reference_path_start_ms: 1_000, + reference_path_end_ms: 1_500, + }, + ], + 1_000, + ) + .unwrap(); + + assert!(partition.train_market_ids().is_empty()); + assert_eq!( + partition + .crossing_excluded() + .iter() + .map(|event| event.market_id.as_str()) + .collect::>(), + ["ends-at-boundary"] + ); + assert_eq!(partition.held_out_market_ids(), ["starts-at-boundary"]); + } +} diff --git a/rust_hft/prediction-markets/crates/ploy-research/src/factors_v2.rs b/rust_hft/prediction-markets/crates/ploy-research/src/factors_v2.rs index b9048ac19..1f59d32e5 100644 --- a/rust_hft/prediction-markets/crates/ploy-research/src/factors_v2.rs +++ b/rust_hft/prediction-markets/crates/ploy-research/src/factors_v2.rs @@ -5,6 +5,7 @@ use ploy_market_contracts::Regime; use serde::{Deserialize, Serialize}; use crate::autofactor::{LlmPriorSpec, LlmProbabilityBlendSpec}; +use crate::event_cohort_partition::EventCohortPartition; use crate::factors::{ normalized_underlying_symbol, pearson_ic, spearman_ic, FactorObservation, ResearchPmBookSnapshot, @@ -464,10 +465,11 @@ pub struct SettlementProbabilityWalkForwardOptions { /// Mission-pinned outer train/validation boundary for settlement research. /// Generic factor and token-execution reviews leave this unset. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq)] pub struct SettlementProbabilityTimeCohort { boundary: DateTime, event_window_secs: i64, + partition: Option, } impl SettlementProbabilityTimeCohort { @@ -481,8 +483,29 @@ impl SettlementProbabilityTimeCohort { Ok(Self { boundary, event_window_secs, + partition: None, }) } + + pub fn with_partition(mut self, partition: EventCohortPartition) -> Result { + if partition.common_time_boundary_ms() != self.boundary.timestamp_millis() { + return Err("event cohort partition boundary differs from settlement cohort".into()); + } + self.partition = Some(partition); + Ok(self) + } + + pub fn partition(&self) -> Option<&EventCohortPartition> { + self.partition.as_ref() + } + + pub fn boundary(&self) -> DateTime { + self.boundary + } + + pub fn event_window_secs(&self) -> i64 { + self.event_window_secs + } } impl Default for SettlementProbabilityWalkForwardOptions { @@ -3115,64 +3138,70 @@ pub fn walk_forward_settlement_probability_report_with_prior( } /// Build search reward evidence from the mission-pinned training cohort only. -/// Held-out rows are used solely to establish the first decision timestamp; -/// their labels, probabilities, execution, and metrics are never evaluated. +/// The caller must prefilter rows with the attached partition; held-out rows +/// and labels never enter this search-facing API. pub fn build_settlement_training_probability_report_with_prior( rows: &[FactorObservationV2], start: DateTime, end: DateTime, prior: Option<&LlmPriorSpec>, options: SettlementProbabilityWalkForwardOptions, -) -> SettlementTrainingProbabilityReport { - let Some(cohort) = options.time_cohort else { - return SettlementTrainingProbabilityReport { - training_cohort_id: "missing-time-cohort".to_string(), - event_count: 0, - baselines: Vec::new(), - }; +) -> Result { + let Some(cohort) = options.time_cohort.as_ref() else { + return Err("settlement training requires a time cohort".into()); }; - let cohort_id = format!( - "settlement-training-before-{}-{}s", - cohort.boundary.timestamp_millis(), - cohort.event_window_secs - ); + let Some(partition) = cohort.partition() else { + return Err("settlement training requires an event cohort partition".into()); + }; + if rows + .iter() + .any(|row| !partition.contains_train_market(&row.event_id)) + { + return Err("settlement training received a non-training event".into()); + } + let cohort_id = partition.digest().to_string(); let Some(bounds) = settlement_walk_forward_window_bounds(start, end, &options) .into_iter() .next() else { - return SettlementTrainingProbabilityReport { + return Ok(SettlementTrainingProbabilityReport { training_cohort_id: cohort_id, event_count: 0, baselines: Vec::new(), - }; + }); }; let mut rows = rows.to_vec(); rows.sort_by_key(|row| row.tick_ts); - let event_ends = event_ends_for_walk_forward(&rows, Some(&cohort)); + let event_ends = event_ends_for_walk_forward(&rows, Some(cohort)); let label_observation_times = official_label_observation_times(&rows); - let (training, _) = event_disjoint_walk_forward_slices( - &rows, - &event_ends, - &label_observation_times, - Some(&cohort), - ( - bounds.train_start, - bounds.train_end, - bounds.test_start, - bounds.test_end, - ), - ); + let training = walk_forward_time_slice(&rows, bounds.train_start, bounds.train_end) + .iter() + .filter(|row| { + event_ends + .get(row.event_id.as_str()) + .and_then(Option::as_ref) + .is_some_and(|event_end| { + *event_end >= bounds.train_start && *event_end < bounds.train_end + }) + }) + .filter(|row| { + label_observation_times + .get(row.event_id.as_str()) + .and_then(Option::as_ref) + .is_some_and(|observed_at| *observed_at <= cohort.boundary) + }) + .collect::>(); let event_count = training .iter() .map(|row| row.event_id.as_str()) .collect::>() .len(); if training.len() < options.walk_forward.review.min_observations { - return SettlementTrainingProbabilityReport { + return Ok(SettlementTrainingProbabilityReport { training_cohort_id: cohort_id, event_count, baselines: Vec::new(), - }; + }); } let probability_options = normalize_settlement_probability_report_options(options.probability.clone()); @@ -3182,11 +3211,11 @@ pub fn build_settlement_training_probability_report_with_prior( prior, probability_options, ); - SettlementTrainingProbabilityReport { + Ok(SettlementTrainingProbabilityReport { training_cohort_id: cohort_id, event_count, baselines: report.baselines, - } + }) } #[derive(Debug, Clone, Copy)] @@ -3203,7 +3232,7 @@ fn settlement_walk_forward_window_bounds( end: DateTime, options: &SettlementProbabilityWalkForwardOptions, ) -> Vec { - if let Some(cohort) = options.time_cohort { + if let Some(cohort) = options.time_cohort.as_ref() { if !(start < cohort.boundary && cohort.boundary < end) { return Vec::new(); } @@ -4256,7 +4285,7 @@ pub fn format_settlement_probability_walk_forward_report( report.options.max_test_log_loss, report.options.max_test_expected_calibration_error, )); - if let Some(cohort) = report.options.time_cohort { + if let Some(cohort) = report.options.time_cohort.as_ref() { out.push_str(&format!( "mission_time_cohort_boundary={} event_window_secs={} crossing_events=purged\n", cohort.boundary.to_rfc3339(), @@ -7828,6 +7857,13 @@ fn event_disjoint_walk_forward_slices<'a>( bounds: (DateTime, DateTime, DateTime, DateTime), ) -> (Vec<&'a FactorObservationV2>, Vec<&'a FactorObservationV2>) { let (train_start, train_end, test_start, test_end) = bounds; + let partition = match time_cohort { + Some(cohort) => match cohort.partition() { + Some(partition) => Some(partition), + None => return (Vec::new(), Vec::new()), + }, + None => None, + }; let train = walk_forward_time_slice(rows, train_start, train_end); let test = walk_forward_time_slice(rows, test_start, test_end); let ends_in = |row: &&FactorObservationV2, start, end| { @@ -7841,16 +7877,9 @@ fn event_disjoint_walk_forward_slices<'a>( .iter() .filter(|row| ends_in(row, test_start, test_end)) .filter(|row| { - time_cohort.is_none_or(|cohort| { - event_ends - .get(row.event_id.as_str()) - .and_then(Option::as_ref) - .and_then(|event_end| { - Duration::try_seconds(cohort.event_window_secs) - .and_then(|window| event_end.checked_sub_signed(window)) - }) - .is_some_and(|event_start| event_start >= cohort.boundary) - }) + partition + .as_ref() + .is_none_or(|partition| partition.contains_held_out_market(&row.event_id)) }) .collect::>(); let Some(first_test_decision) = test_rows.iter().map(|row| row.tick_ts).min() else { @@ -7860,12 +7889,9 @@ fn event_disjoint_walk_forward_slices<'a>( .iter() .filter(|row| ends_in(row, train_start, train_end)) .filter(|row| { - time_cohort.is_none_or(|cohort| { - event_ends - .get(row.event_id.as_str()) - .and_then(Option::as_ref) - .is_some_and(|event_end| *event_end < cohort.boundary) - }) + partition + .as_ref() + .is_none_or(|partition| partition.contains_train_market(&row.event_id)) }) .filter(|row| { let event_id = row.event_id.as_str(); @@ -11288,6 +11314,7 @@ mod tests { } fn governed_time_cohort_options( + rows: &[FactorObservationV2], boundary: DateTime, ) -> SettlementProbabilityWalkForwardOptions { SettlementProbabilityWalkForwardOptions { @@ -11305,7 +11332,7 @@ mod tests { component_profile: SettlementProbabilityComponentProfile::FullSurface, ..Default::default() }, - time_cohort: Some(SettlementProbabilityTimeCohort::new(boundary, 300).unwrap()), + time_cohort: Some(test_time_cohort(rows, boundary)), ..Default::default() } } @@ -11320,14 +11347,14 @@ mod tests { rows, start, end, - governed_time_cohort_options(boundary), + governed_time_cohort_options(rows, boundary), ); let verdict = walk_forward_settlement_verdict_report_with_prior( rows, start, end, None, - governed_time_cohort_options(boundary), + governed_time_cohort_options(rows, boundary), ); assert!(probability.windows.is_empty()); assert!(probability.aggregates.is_empty()); @@ -11335,6 +11362,17 @@ mod tests { assert!(verdict.aggregates.is_empty()); } + fn test_time_cohort( + rows: &[FactorObservationV2], + boundary: DateTime, + ) -> SettlementProbabilityTimeCohort { + let cohort = SettlementProbabilityTimeCohort::new(boundary, 300).unwrap(); + EventCohortPartition::from_test_observations(rows, boundary.timestamp_millis(), 300) + .map_or(cohort.clone(), |partition| { + cohort.with_partition(partition).unwrap() + }) + } + fn short_settlement_time_cohort_case() -> ( DateTime, DateTime, @@ -12537,7 +12575,7 @@ mod tests { rows.sort_by_key(|row| row.tick_ts); let event_ends = canonical_event_ends(&rows); let label_observation_times = official_label_observation_times(&rows); - let cohort = SettlementProbabilityTimeCohort::new(boundary, 300).unwrap(); + let cohort = test_time_cohort(&rows, boundary); let (train, test) = event_disjoint_walk_forward_slices( &rows, @@ -12621,7 +12659,7 @@ mod tests { let lossy_event_end = inferred_event_end(held_out_row).unwrap(); assert!(lossy_event_end < held_out_end); assert!(lossy_event_end - Duration::seconds(300) < boundary); - let cohort = SettlementProbabilityTimeCohort::new(boundary, 300).unwrap(); + let cohort = test_time_cohort(&rows, boundary); let (train, test) = event_disjoint_walk_forward_slices( &rows, &canonical_event_ends(&rows), @@ -12653,7 +12691,7 @@ mod tests { &rows, boundary - Duration::minutes(15), boundary + Duration::minutes(10), - governed_time_cohort_options(boundary), + governed_time_cohort_options(&rows, boundary), ); assert!(!probability.windows.is_empty()); assert!(probability @@ -12665,7 +12703,7 @@ mod tests { boundary - Duration::minutes(15), boundary + Duration::minutes(10), None, - governed_time_cohort_options(boundary), + governed_time_cohort_options(&rows, boundary), ); assert!(!verdict.windows.is_empty()); assert!(verdict.windows.iter().all(|window| window.test_n == 1)); @@ -12695,14 +12733,6 @@ mod tests { missing.event_end_ts = None; let mut missing_rows = vec![train_row.clone(), missing]; make_settlement_probability_eligible(&mut missing_rows); - let (legacy_train, legacy_test) = event_disjoint_walk_forward_slices( - &missing_rows, - &inferred_event_ends(&missing_rows), - &official_label_observation_times(&missing_rows), - Some(&cohort), - bounds, - ); - assert_eq!((legacy_train.len(), legacy_test.len()), (1, 1)); let (train, test) = event_disjoint_walk_forward_slices( &missing_rows, &event_ends_for_walk_forward(&missing_rows, Some(&cohort)), @@ -12724,14 +12754,6 @@ mod tests { second.event_end_ts = Some(boundary + Duration::minutes(6)); let mut conflicting_rows = vec![train_row, first, second]; make_settlement_probability_eligible(&mut conflicting_rows); - let (legacy_train, legacy_test) = event_disjoint_walk_forward_slices( - &conflicting_rows, - &inferred_event_ends(&conflicting_rows), - &official_label_observation_times(&conflicting_rows), - Some(&cohort), - bounds, - ); - assert_eq!((legacy_train.len(), legacy_test.len()), (1, 2)); let (train, test) = event_disjoint_walk_forward_slices( &conflicting_rows, &event_ends_for_walk_forward(&conflicting_rows, Some(&cohort)), @@ -12765,7 +12787,7 @@ mod tests { end, None, SettlementProbabilityWalkForwardOptions { - time_cohort: Some(SettlementProbabilityTimeCohort::new(boundary, 300).unwrap()), + time_cohort: Some(test_time_cohort(&rows, boundary)), ..options }, ); @@ -12798,21 +12820,30 @@ mod tests { fn settlement_training_report_never_scores_the_held_out_event() { let (start, boundary, end, rows, options) = short_settlement_time_cohort_case(); let options = SettlementProbabilityWalkForwardOptions { - time_cohort: Some(SettlementProbabilityTimeCohort::new(boundary, 300).unwrap()), + time_cohort: Some(test_time_cohort(&rows, boundary)), ..options }; + let partition = options.time_cohort.as_ref().unwrap().partition().unwrap(); + let training_rows = rows + .iter() + .filter(|row| partition.contains_train_market(&row.event_id)) + .cloned() + .collect::>(); + let partition_digest = partition.digest().to_string(); let report = build_settlement_training_probability_report_with_prior( - &rows, start, end, None, options, - ); + &training_rows, + start, + end, + None, + options, + ) + .unwrap(); assert_eq!(report.event_count, 1); assert!(report.baselines.iter().all(|baseline| baseline.n == 1)); assert_eq!( - report.training_cohort_id, - format!( - "settlement-training-before-{}-300s", - boundary.timestamp_millis() - ) + report.training_cohort_id, partition_digest, + "settlement search must expose the shared immutable partition digest" ); } @@ -12837,7 +12868,7 @@ mod tests { end, None, SettlementProbabilityWalkForwardOptions { - time_cohort: Some(SettlementProbabilityTimeCohort::new(boundary, 300).unwrap()), + time_cohort: Some(test_time_cohort(&rows, boundary)), ..options }, ); @@ -12870,7 +12901,7 @@ mod tests { ("after end", end + Duration::seconds(1)), ] { let options = SettlementProbabilityWalkForwardOptions { - time_cohort: Some(SettlementProbabilityTimeCohort::new(boundary, 300).unwrap()), + time_cohort: Some(test_time_cohort(&rows, boundary)), ..options.clone() }; let probability = walk_forward_settlement_probability_report_with_prior( diff --git a/rust_hft/prediction-markets/crates/ploy-research/src/lib.rs b/rust_hft/prediction-markets/crates/ploy-research/src/lib.rs index d9f74bd0f..65c66b0a5 100644 --- a/rust_hft/prediction-markets/crates/ploy-research/src/lib.rs +++ b/rust_hft/prediction-markets/crates/ploy-research/src/lib.rs @@ -7,6 +7,7 @@ pub mod backtesting; pub mod dataset; #[cfg(feature = "db")] pub mod deribit; +pub mod event_cohort_partition; pub mod event_ml; pub mod factors; pub mod factors_new; @@ -52,6 +53,10 @@ pub use deribit::{ load_deribit_feature_snapshots, load_deribit_feature_snapshots_with_timings, DeribitFeatureLoadResult, }; +pub use event_cohort_partition::{ + EventCohortExclusion, EventCohortExclusionReason, EventCohortPartition, + EVENT_COHORT_PARTITION_VERSION, +}; pub use event_ml::{ build_event_ml_strategy_handoff, build_walk_forward_report, canonical_event_ml_architecture, event_ml_architecture_markdown, event_ml_strategy_handoff_markdown, gate_matrix, @@ -132,8 +137,8 @@ pub use autofactor::{ AutoFactorDecision, AutoFactorError, AutoFactorMatrix, AutoFactorOptions, AutoFactorReport, AutoFactorRuntimeContractCatalog, AutoFactorRuntimeFormulaBlocker, AutoFactorRuntimeInputContract, AutoFactorTargetContract, AutoFactorV2Target, FactorExpr, - LlmMutationSpec, LlmPriorSpec, LlmProbabilityBlendSpec, NamedFactorExpr, RepricePilotMetrics, - RepricePilotSelection, + LlmMutationSpec, LlmPriorSpec, LlmProbabilityBlendSpec, NamedFactorExpr, + RepriceEventCohortSplit, RepricePilotMetrics, RepricePilotSelection, }; pub use backtest::{run_binary_backtest, BacktestMetrics, SimulatedFill}; pub use factors_new::{ diff --git a/rust_hft/prediction-markets/crates/ploy-research/src/prediction_loop.rs b/rust_hft/prediction-markets/crates/ploy-research/src/prediction_loop.rs index 7a4f99590..073701da5 100644 --- a/rust_hft/prediction-markets/crates/ploy-research/src/prediction_loop.rs +++ b/rust_hft/prediction-markets/crates/ploy-research/src/prediction_loop.rs @@ -234,7 +234,7 @@ pub fn current_prediction_policy_snapshot_id() -> String { format!("sha256:{:x}", digest.finalize()) } -fn prediction_policy_sources() -> [(&'static str, &'static [u8]); 39] { +fn prediction_policy_sources() -> [(&'static str, &'static [u8]); 40] { [ ( "crates/ploy-research/src/autofactor.rs", @@ -244,6 +244,10 @@ fn prediction_policy_sources() -> [(&'static str, &'static [u8]); 39] { "crates/ploy-research/src/alpha_search.rs", include_bytes!("alpha_search.rs"), ), + ( + "crates/ploy-research/src/event_cohort_partition.rs", + include_bytes!("event_cohort_partition.rs"), + ), ( "crates/ploy-research/src/factors.rs", include_bytes!("factors.rs"), @@ -4150,6 +4154,7 @@ mod tests { assert!(paths .iter() .any(|path| path.contains("monday-prediction-research"))); + assert!(paths.contains(&"crates/ploy-research/src/event_cohort_partition.rs")); assert!(paths.contains(&"crates/ploy-research/src/polymarket_evidence_projection.rs")); assert!(paths.contains(&"crates/ploy-research/src/verified_artifact_audit.rs")); assert!(paths.contains(&"crates/ploy-research/src/verified_binance_projection.rs"));