From 8951c7a89cdb7fb731c47c23e7b8d96c97202445 Mon Sep 17 00:00:00 2001 From: Cody Date: Sat, 6 Jun 2026 14:56:29 +0000 Subject: [PATCH 1/2] feat(scan): add stats.momentum.tsmom@1 intraday momentum detector New single-leg ANOM scan measuring intraday time-series momentum / return continuation as a tradeable signal (sign + magnitude + significance + hit-rate), not just an autocorrelation p-value. Per horizon k, the kernel partitions the return series into non-overlapping k-blocks and fits OLS of next-block on past-block return; the slope is the continuation coefficient (>0 momentum, <0 reversion) with a Student-t t-stat, a directional hit-rate, and the sign(past)*next TSMOM mean. The output frames the dominant hold horizon (selected_hold_bars) and per-k turnover (1/k) so the Quant agent can read natural turnover off the finding. Optional ex-ante trailing-vol scaling (default on). - Registered alphabetically in register_anom_scans (Pattern E); dispatchable on all three surfaces via the shared registry. - Kernel + scan unit tests, happy-path integration test, and a float-free insta schema snapshot. rustfmt clean. Co-Authored-By: Claude Opus 4.8 --- crates/miner-core/src/scan/anom/mod.rs | 5 + .../miner-core/src/scan/anom/tsmom/kernel.rs | 445 ++++++++++ crates/miner-core/src/scan/anom/tsmom/mod.rs | 811 ++++++++++++++++++ crates/miner-core/tests/scan_tsmom.rs | 202 +++++ .../snapshots/scan_tsmom__tsmom_schema.snap | 54 ++ 5 files changed, 1517 insertions(+) create mode 100644 crates/miner-core/src/scan/anom/tsmom/kernel.rs create mode 100644 crates/miner-core/src/scan/anom/tsmom/mod.rs create mode 100644 crates/miner-core/tests/scan_tsmom.rs create mode 100644 crates/miner-core/tests/snapshots/scan_tsmom__tsmom_schema.snap diff --git a/crates/miner-core/src/scan/anom/mod.rs b/crates/miner-core/src/scan/anom/mod.rs index 152782a..d30e5ef 100644 --- a/crates/miner-core/src/scan/anom/mod.rs +++ b/crates/miner-core/src/scan/anom/mod.rs @@ -25,6 +25,7 @@ pub mod meanrev; pub mod outliers; pub mod returns; pub mod summary; +pub mod tsmom; pub mod variance_ratio; pub mod vol; @@ -38,6 +39,7 @@ pub use meanrev::OuHalfLifeScan; pub use outliers::OutliersZAndMadScan; pub use returns::ReturnsProfileScan; pub use summary::SummaryWelfordScan; +pub use tsmom::TsmomScan; pub use variance_ratio::VarianceRatioScan; pub use vol::VolRollingScan; @@ -55,6 +57,7 @@ pub fn register_anom_scans(r: &mut Registry) { // stats.drawdown.profile <- Plan 04-04 // stats.heteroskedasticity.arch_lm <- Plan 04-06 // stats.meanrev.ou_halflife <- RAD-3627 + // stats.momentum.tsmom <- RAD-3839 // stats.normality.jarque_bera <- Plan 04-06 // stats.outliers.z_and_mad <- Plan 04-04 // stats.returns.profile <- Plan 04-03 @@ -67,6 +70,7 @@ pub fn register_anom_scans(r: &mut Registry) { r.register(Box::new(DrawdownProfileScan)); r.register(Box::new(ArchLmScan)); r.register(Box::new(OuHalfLifeScan)); + r.register(Box::new(TsmomScan)); r.register(Box::new(JarqueBeraScan)); r.register(Box::new(OutliersZAndMadScan)); r.register(Box::new(ReturnsProfileScan)); @@ -105,6 +109,7 @@ mod tests { r.get("stats.meanrev.ou_halflife", 1).is_some(), "RAD-3627 ou_halflife" ); + assert!(r.get("stats.momentum.tsmom", 1).is_some(), "RAD-3839 tsmom"); assert!( r.get("stats.normality.jarque_bera", 1).is_some(), "ANOM-09 jarque_bera" diff --git a/crates/miner-core/src/scan/anom/tsmom/kernel.rs b/crates/miner-core/src/scan/anom/tsmom/kernel.rs new file mode 100644 index 0000000..36d5b5e --- /dev/null +++ b/crates/miner-core/src/scan/anom/tsmom/kernel.rs @@ -0,0 +1,445 @@ +//! Pure intraday time-series-momentum (TSMOM) kernel for ANOM-12 — `tsmom`. +//! +//! Pattern analog: [`crate::scan::anom::variance_ratio::kernel`] and the +//! sibling `adf/kernel.rs` / `kpss/kernel.rs` — private `pub(crate)` pure +//! functions over `&[f64]` with a sibling `#[cfg(test)] mod tests` block. No +//! IO, no `serde_json`, no `Reader` calls. +//! +//! ## Reference +//! +//! Time-series momentum (return continuation): Moskowitz, T. J., Ooi, Y. H. & +//! Pedersen, L. H. (2012), "Time Series Momentum", Journal of Financial +//! Economics 104(2), 228-250. The canonical TSMOM signal trades the sign of +//! the trailing-`k` return and holds for the next `k` bars; the continuation +//! coefficient is the slope of next-`k` return regressed on past-`k` return. +//! +//! ## Algorithm (per horizon `k`) +//! +//! 1. **Non-overlapping `k`-blocks.** Partition the return series into +//! consecutive blocks of `k` bars (dropping the trailing remainder); block +//! `b` carries `R_b = Σ r_t` over its `k` returns. Non-overlapping blocks +//! keep the `(past, next)` pairs (near-)independent so the OLS t-stat is +//! honest — overlapping windows would inflate it via induced +//! autocorrelation. +//! 2. **Continuation regression.** Form pairs `(R_b, R_{b+1})` and fit OLS +//! `R_{b+1} = α + β·R_b`. The slope `β` is the continuation coefficient: +//! `β > 0` ⇒ momentum (the past block predicts the next in the same +//! direction), `β < 0` ⇒ mean reversion. Under a random walk `β ≈ 0`. +//! 3. **t-stat / p-value.** `t = β / SE(β)`, two-sided p from Student-t with +//! `df = m - 2` (`m` = number of pairs). +//! 4. **Hit-rate.** Fraction of pairs whose past and next blocks share sign +//! (`R_b · R_{b+1} > 0`) — the directional accuracy of the signal. +//! 5. **TSMOM mean.** Mean of `sign(R_b)·R_{b+1}` — the average per-block +//! return to holding the next block in the direction of the past one. +//! +//! Vol-normalisation of the input return series (the `scaling` param) is +//! applied by the caller via [`vol_normalize`] before the per-`k` blocks are +//! formed; a global rescale would leave `β`/`t`/hit-rate unchanged (OLS slope +//! is scale-invariant), so the normalisation is deliberately *time-varying* +//! (ex-ante trailing vol). + +#![cfg_attr(any(test, debug_assertions), allow(clippy::float_cmp))] + +use statrs::distribution::{ContinuousCDF, StudentsT}; + +/// Trailing window (in bars) for the ex-ante volatility estimate used by +/// [`vol_normalize`]. 20 bars is a standard intraday lookback. +pub(crate) const VOL_WINDOW: usize = 20; + +#[derive(Debug, Clone, PartialEq)] +pub(crate) struct TsmomResult { + /// Continuation coefficient: OLS slope of next-block on past-block return. + pub continuation_coef: f64, + /// t-statistic of the continuation coefficient. + pub t_stat: f64, + /// Two-sided p-value of the continuation coefficient (Student-t, df=m-2). + pub p_value: f64, + /// Directional hit-rate: fraction of block pairs sharing sign. + pub hit_rate: f64, + /// Mean per-block return to the sign(past)·next TSMOM rule. + pub tsmom_mean: f64, + /// Number of `(past, next)` block pairs used. + pub n_pairs: usize, +} + +/// Sample standard deviation (unbiased, ddof=1). Returns `0.0` for `< 2` +/// elements. +#[inline] +#[must_use] +#[allow(clippy::cast_precision_loss, reason = "element count << 2^52")] +pub(crate) fn sample_std(xs: &[f64]) -> f64 { + let n = xs.len(); + if n < 2 { + return 0.0; + } + let n_f = n as f64; + let mean = xs.iter().sum::() / n_f; + let var = xs.iter().map(|x| (x - mean) * (x - mean)).sum::() / (n_f - 1.0); + var.sqrt() +} + +/// Ex-ante (look-ahead-free) volatility normalisation: divide each return by +/// the sample std of the *strictly preceding* returns in a trailing `window`. +/// +/// The first one/two bars (fewer than two past observations) fall back to the +/// full-series std — a bounded warmup approximation acceptable for a +/// measurement scan. A constant series (global std `0`) is normalised against +/// a tiny positive floor; the caller's degenerate-variance guard rejects it +/// downstream regardless. +#[inline] +#[must_use] +pub(crate) fn vol_normalize(returns: &[f64], window: usize) -> Vec { + let n = returns.len(); + if n == 0 { + return Vec::new(); + } + let global = { + let g = sample_std(returns); + if g > 0.0 && g.is_finite() { + g + } else { + f64::MIN_POSITIVE + } + }; + let mut out = Vec::with_capacity(n); + for i in 0..n { + let lo = i.saturating_sub(window); + let past = &returns[lo..i]; + let sigma = if past.len() >= 2 { + let s = sample_std(past); + if s > 0.0 && s.is_finite() { s } else { global } + } else { + global + }; + out.push(returns[i] / sigma); + } + out +} + +/// Compute the TSMOM continuation statistics for a return series at horizon +/// `k` using non-overlapping `k`-blocks. +/// +/// The caller guarantees `k >= 1` and `returns.len() / k >= 4` (at least three +/// `(past, next)` block pairs so the OLS has `df = m - 2 >= 1`). +#[inline] +#[allow(clippy::cast_precision_loss, reason = "block / pair counts << 2^52")] +#[allow( + clippy::similar_names, + reason = "sx/sy/sxx/sxy/dx/dy are the canonical OLS accumulator names" +)] +pub(crate) fn tsmom_continuation(returns: &[f64], k: usize) -> Result { + let n = returns.len(); + if k < 1 { + return Err(format!("tsmom: k must be >= 1; got k={k}")); + } + let num_blocks = n / k; + if num_blocks < 4 { + return Err(format!( + "tsmom: k={k} too large for n={n} (need n >= 4*k for >= 3 block pairs)" + )); + } + + // Step 1 — non-overlapping block sums (drop the trailing remainder). + let mut blocks: Vec = Vec::with_capacity(num_blocks); + for b in 0..num_blocks { + let sum: f64 = returns[b * k..(b + 1) * k].iter().sum(); + blocks.push(sum); + } + + // Step 2 — (past, next) pairs: x = R_b, y = R_{b+1}. + let m = num_blocks - 1; + let m_f = m as f64; + let mut sx = 0.0_f64; + let mut sy = 0.0_f64; + for j in 0..m { + sx += blocks[j]; + sy += blocks[j + 1]; + } + let xbar = sx / m_f; + let ybar = sy / m_f; + + let mut sxx = 0.0_f64; + let mut sxy = 0.0_f64; + let mut hits = 0usize; + let mut tsmom_sum = 0.0_f64; + for j in 0..m { + let x = blocks[j]; + let y = blocks[j + 1]; + let dx = x - xbar; + let dy = y - ybar; + sxx += dx * dx; + sxy += dx * dy; + if x * y > 0.0 { + hits += 1; + } + // Directional TSMOM return: hold the next block in the sign of past. + if x > 0.0 { + tsmom_sum += y; + } else if x < 0.0 { + tsmom_sum -= y; + } + } + + if sxx <= 0.0 || !sxx.is_finite() { + return Err(format!( + "tsmom: degenerate past-block variance (constant blocks?) at k={k}" + )); + } + + // Step 3 — OLS slope + t-stat. + let beta = sxy / sxx; + let alpha = ybar - beta * xbar; + let mut sse = 0.0_f64; + for j in 0..m { + let resid = blocks[j + 1] - alpha - beta * blocks[j]; + sse += resid * resid; + } + let df = m - 2; // m >= 3 guaranteed (num_blocks >= 4). + let s2 = sse / df as f64; + let se_beta = (s2 / sxx).sqrt(); + // se_beta == 0 (perfect fit, beta != 0) -> t = ±inf -> p -> 0; handled in + // two_sided_p_value. + let t_stat = beta / se_beta; + let p_value = two_sided_p_value(t_stat, df); + + Ok(TsmomResult { + continuation_coef: beta, + t_stat, + p_value, + hit_rate: hits as f64 / m_f, + tsmom_mean: tsmom_sum / m_f, + n_pairs: m, + }) +} + +/// Two-sided Student-t p-value for a t-statistic with `df` degrees of freedom. +/// A finite `t` runs the standard path (`t == 0` falls out as `p = 1` since +/// `cdf(0) = 0.5`); non-finite `|t|` (perfect fit) ⇒ 0.0; NaN ⇒ NaN. +#[inline] +#[allow(clippy::cast_precision_loss, reason = "df << 2^52")] +fn two_sided_p_value(t: f64, df: usize) -> f64 { + if !t.is_finite() { + return if t.is_nan() { f64::NAN } else { 0.0 }; + } + let dist = StudentsT::new(0.0, 1.0, df as f64).expect("students-t df >= 1"); + let upper_tail = 1.0 - dist.cdf(t.abs()); + (2.0 * upper_tail).clamp(0.0, 1.0) +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[cfg(test)] +#[allow(clippy::cast_lossless)] +mod tests { + use super::*; + + fn approx_eq(a: f64, b: f64, tol: f64) -> bool { + (a - b).abs() <= tol + } + + /// Deterministic LCG noise in [-0.5, 0.5] — an IID white-noise return + /// series (random walk in log-price), so continuation `β ≈ 0`. + #[allow(clippy::cast_possible_truncation)] + fn lcg_returns(n: usize, seed: u64) -> Vec { + let mut s = seed as u32; + let mut out = Vec::with_capacity(n); + for _ in 0..n { + s = s.wrapping_mul(1_664_525).wrapping_add(1_013_904_223); + out.push(f64::from(s) / f64::from(u32::MAX) - 0.5); + } + out + } + + /// AR(1) momentum series `r_t = phi·r_{t-1} + eps_t` with `phi > 0` — + /// positive serial dependence, so block-to-block continuation `β > 0`. + #[allow(clippy::cast_possible_truncation)] + fn ar1_returns(n: usize, seed: u64, phi: f64) -> Vec { + let eps = lcg_returns(n, seed); + let mut out = Vec::with_capacity(n); + let mut prev = 0.0_f64; + for &e in &eps { + let r = phi * prev + e; + out.push(r); + prev = r; + } + out + } + + // -- two_sided_p_value ----------------------------------------------- + + #[test] + fn p_value_at_zero_is_unity() { + assert!(approx_eq(two_sided_p_value(0.0, 10), 1.0, 1e-12)); + } + + #[test] + fn p_value_large_t_is_tiny() { + let p = two_sided_p_value(8.0, 30); + assert!(p < 0.01, "p({}) = {p}", 8.0); + } + + #[test] + fn p_value_infinite_t_is_zero() { + assert!(approx_eq(two_sided_p_value(f64::INFINITY, 10), 0.0, 1e-12)); + } + + #[test] + fn p_value_nan_in_nan_out() { + assert!(two_sided_p_value(f64::NAN, 10).is_nan()); + } + + #[test] + fn p_value_symmetric_in_sign() { + let a = two_sided_p_value(1.7, 25); + let b = two_sided_p_value(-1.7, 25); + assert!(approx_eq(a, b, 1e-12)); + } + + // -- sample_std ------------------------------------------------------ + + #[test] + fn sample_std_known_value() { + // [2,4,4,4,5,5,7,9] has sample std 2.138... (ddof=1). + let s = sample_std(&[2.0, 4.0, 4.0, 4.0, 5.0, 5.0, 7.0, 9.0]); + assert!(approx_eq(s, 2.138_089_935_299_395, 1e-9), "std = {s}"); + } + + #[test] + fn sample_std_too_few_is_zero() { + assert_eq!(sample_std(&[]), 0.0); + assert_eq!(sample_std(&[3.0]), 0.0); + } + + // -- vol_normalize --------------------------------------------------- + + #[test] + fn vol_normalize_length_invariant() { + let r = lcg_returns(100, 1); + let v = vol_normalize(&r, VOL_WINDOW); + assert_eq!(v.len(), r.len()); + assert!(v.iter().all(|x| x.is_finite())); + } + + #[test] + fn vol_normalize_empty() { + assert!(vol_normalize(&[], VOL_WINDOW).is_empty()); + } + + /// Global rescale of the input does NOT change the normalised series' + /// continuation statistics — but the time-varying normalisation itself is + /// not a global rescale, so it generally changes them. Here we just pin + /// that normalising white noise yields a finite, roughly unit-scale + /// series. + #[test] + fn vol_normalize_unit_scale_on_white_noise() { + let r = lcg_returns(500, 7); + let v = vol_normalize(&r, VOL_WINDOW); + let s = sample_std(&v); + assert!(s > 0.3 && s < 4.0, "normalised std = {s} should be O(1)"); + } + + // -- tsmom_continuation: guards -------------------------------------- + + #[test] + fn tsmom_k_zero_errors() { + let r = lcg_returns(40, 1); + assert!(tsmom_continuation(&r, 0).is_err()); + } + + #[test] + fn tsmom_k_too_large_errors() { + // n=10, k=3 -> num_blocks=3 < 4 -> error. + let r = lcg_returns(10, 1); + assert!(tsmom_continuation(&r, 3).is_err()); + } + + #[test] + fn tsmom_constant_blocks_errors() { + let r = vec![0.0_f64; 40]; + assert!( + tsmom_continuation(&r, 1).is_err(), + "constant -> sxx=0 -> Err" + ); + } + + // -- tsmom_continuation: behaviour ----------------------------------- + + /// White-noise returns: continuation coefficient ≈ 0 and the directional + /// hit-rate ≈ 0.5. (Both are robust to the seed — `β` sits ~7σ inside the + /// tolerance and the hit-rate ~9σ — unlike a `p > 0.05` check which has an + /// inherent ~5% per-seed false-positive rate under the null.) + #[test] + fn tsmom_white_noise_no_continuation() { + let r = lcg_returns(2000, 42); + let res = tsmom_continuation(&r, 1).expect("ok"); + assert!( + res.continuation_coef.abs() < 0.15, + "white-noise β = {} should be ≈ 0", + res.continuation_coef + ); + assert!( + (res.hit_rate - 0.5).abs() < 0.1, + "white-noise hit-rate = {} should be ≈ 0.5", + res.hit_rate + ); + assert!( + res.p_value >= 0.0 && res.p_value <= 1.0, + "p in [0,1]: {}", + res.p_value + ); + } + + /// AR(1) positive momentum: continuation coefficient significantly > 0 at + /// the matching short horizon (k=1). + #[test] + fn tsmom_ar1_positive_continuation_significant() { + let r = ar1_returns(2000, 99, 0.5); + let res = tsmom_continuation(&r, 1).expect("ok"); + assert!( + res.continuation_coef > 0.0, + "AR(1) β = {} should be > 0", + res.continuation_coef + ); + assert!( + res.p_value < 0.01, + "AR(1) p = {} should be significant", + res.p_value + ); + assert!( + res.hit_rate > 0.5, + "AR(1) hit-rate = {} should beat a coin flip", + res.hit_rate + ); + assert!(res.tsmom_mean > 0.0, "AR(1) TSMOM mean should be positive"); + } + + /// Perfect anti-correlation at the block scale: alternating block signs + /// yield a negative continuation coefficient (mean reversion). + #[test] + fn tsmom_mean_reverting_negative_continuation() { + // Each bar is a one-element block at k=1; alternating ±1 with a tiny + // perturbation so blocks are not perfectly collinear. + let mut r = Vec::new(); + for i in 0..400 { + let base = if i % 2 == 0 { 1.0 } else { -1.0 }; + let jitter = ((i as f64) * 0.01).sin() * 0.05; + r.push(base + jitter); + } + let res = tsmom_continuation(&r, 1).expect("ok"); + assert!( + res.continuation_coef < 0.0, + "alternating β = {} should be < 0 (reversion)", + res.continuation_coef + ); + } + + #[test] + fn tsmom_n_pairs_matches_blocks() { + let r = lcg_returns(100, 3); + // k=10 -> num_blocks=10 -> m=9 pairs. + let res = tsmom_continuation(&r, 10).expect("ok"); + assert_eq!(res.n_pairs, 9); + } +} diff --git a/crates/miner-core/src/scan/anom/tsmom/mod.rs b/crates/miner-core/src/scan/anom/tsmom/mod.rs new file mode 100644 index 0000000..2543ef2 --- /dev/null +++ b/crates/miner-core/src/scan/anom/tsmom/mod.rs @@ -0,0 +1,811 @@ +//! `TsmomScan` — ANOM-12 intraday time-series momentum / return continuation. +//! +//! Pattern analog: [`crate::scan::anom::variance_ratio::VarianceRatioScan`] +//! (Plan 04-05 sibling) — multi-`k` ANOM scan emitting parallel arrays in +//! `effect.extra` (Pattern A from `04-PATTERNS.md`). Tier-2 build for +//! RAD-3545 / RAD-3839. +//! +//! ## Reference +//! +//! Moskowitz, T. J., Ooi, Y. H. & Pedersen, L. H. (2012), "Time Series +//! Momentum", Journal of Financial Economics 104(2), 228-250. Unlike +//! `stats.autocorr.ljung_box` (which only reports an autocorrelation +//! p-value), this scan frames the dependence of the next-`k`-bar return on the +//! past-`k`-bar return as a *tradeable continuation signal*: sign + magnitude +//! (`β`) + significance (t-stat / p) + directional hit-rate, plus the hold +//! horizon (which `k` persists) so the Quant agent can read natural turnover +//! straight off the finding. +//! +//! ## D4-02 surface +//! +//! - `id = "stats.momentum.tsmom"`, `version = 1`, `arity = ScanArity::Single`. +//! - `params`: optional `k_values` (lookback/holding horizons in bars, default +//! `[1, 5, 10, 20]`; each `k >= 1`), optional `scaling` (vol-normalised +//! returns, default `true`). +//! - `effect.metric = "tsmom_continuation"`, `effect.value = continuation +//! coefficient at the selected hold horizon` (the `k` whose positive +//! continuation is most significant; falls back to the strongest |t-stat| +//! when no horizon shows positive continuation). +//! - `effect.p_value = p of the continuation coefficient at the selected k`. +//! - `effect.effect_size = {kind: "hit_rate", value: hit-rate at selected k}`. +//! - `effect.extra = {continuation_coefs, hit_rates, k_values, p_values, +//! selected_hold_bars, t_stats, tsmom_means, turnover_per_bar}` (alphabetical +//! `BTreeMap` order) — parallel per-`k` arrays plus the single-element +//! `selected_hold_bars` framing the dominant hold horizon. +//! - `raw.series = {returns, timestamps_ms}` (the raw log returns; `scaling` +//! is a reproducible transform applied to the analysis copy). +//! +//! ## Determinism +//! +//! The `k`-grid loop is SEQUENTIAL (not `par_iter`) — same Pitfall 4 +//! discipline as the variance-ratio / ADF scans. +//! +//! ## Registration +//! +//! Appended inside [`crate::scan::anom::register_anom_scans`] (Pattern E — +//! `crates/miner-core/src/scan/registry.rs` is NOT modified). + +use std::collections::BTreeMap; +use std::sync::atomic::Ordering; + +use chrono::Utc; +use serde_json::Value as JsonValue; + +use crate::findings::{ + DataSlice, Effect, EffectSize, Finding, FindingSink, Raw, RawArray, ResultFinding, Source, +}; +use crate::scan::primitives::raw_array::f64_slice_to_raw_array; +use crate::scan::primitives::returns::log_returns; +use crate::scan::{Scan, ScanArity, ScanCtx, ScanError, ScanFindingShape, ScanRequest}; + +pub mod kernel; + +/// ANOM-12 — intraday time-series momentum (return continuation) scan. +pub struct TsmomScan; + +const SCAN_ID: &str = "stats.momentum.tsmom"; +const SCAN_VERSION: u32 = 1; +const EFFECT_METRIC: &str = "tsmom_continuation"; + +const DEFAULT_K_VALUES: &[i64] = &[1, 5, 10, 20]; + +impl Scan for TsmomScan { + fn id(&self) -> &'static str { + SCAN_ID + } + + fn version(&self) -> u32 { + SCAN_VERSION + } + + fn arity(&self) -> ScanArity { + ScanArity::Single + } + + fn param_schema(&self) -> JsonValue { + serde_json::json!({ + "$schema": "http://json-schema.org/draft-07/schema#", + "type": "object", + "properties": { + "k_values": { + "type": "array", + "items": { "type": "integer", "minimum": 1 }, + "default": [1, 5, 10, 20], + "description": "Lookback/holding horizons in bars; each k >= 1. Continuation is measured over non-overlapping k-blocks." + }, + "scaling": { + "type": "boolean", + "default": true, + "description": "If true (default), normalise each return by a trailing ex-ante volatility before forming blocks (vol-scaled momentum). If false, use raw log returns." + } + }, + "additionalProperties": false + }) + } + + fn finding_fields(&self) -> ScanFindingShape { + ScanFindingShape { + effect_extra_keys: &[ + "continuation_coefs", + "hit_rates", + "k_values", + "p_values", + "selected_hold_bars", + "t_stats", + "tsmom_means", + "turnover_per_bar", + ], + raw_series_keys: &["returns", "timestamps_ms"], + } + } + + #[allow( + clippy::too_many_lines, + reason = "Scan::run is the linear dispatch + envelope build path; splitting into helpers obscures the 7-step Pattern A structure" + )] + fn run( + &self, + ctx: &ScanCtx<'_>, + req: &ScanRequest, + sink: &mut dyn FindingSink, + ) -> Result<(), ScanError> { + // Step 1 — cancel at entry. + if ctx.cancel.load(Ordering::Relaxed) { + return Ok(()); + } + + // Step 2 — N guard. Need at least 2 closes for log_returns. + let n_closes = ctx.bars.close.len(); + if n_closes < 2 { + return Err(ScanError::Kernel(format!( + "stats.momentum.tsmom: need n >= 2 closes; got n={n_closes} (InsufficientData)" + ))); + } + + // Step 3 — resolve params. + let k_values = resolve_k_values(req)?; + let scaling = resolve_scaling(req)?; + + // Step 4 — compute returns + validate k_values vs returns length. + let returns = log_returns(&ctx.bars.close); + let n_returns = returns.len(); + // Smallest meaningful horizon (k=1) needs >= 4 blocks => >= 4 returns. + if n_returns < 4 { + return Err(ScanError::Kernel(format!( + "stats.momentum.tsmom: need >= 4 returns; got {n_returns} (InsufficientData)" + ))); + } + for &k in &k_values { + if k < 1 { + return Err(ScanError::Kernel(format!( + "stats.momentum.tsmom: k_values entries must be >= 1; got {k}" + ))); + } + if n_returns / k < 4 { + return Err(ScanError::Kernel(format!( + "stats.momentum.tsmom: k={k} too large for n_returns={n_returns} (need n_returns >= 4*k)" + ))); + } + } + + // Analysis series: ex-ante vol-normalised when scaling is on. A global + // rescale is OLS scale-invariant, so the normalisation is deliberately + // time-varying (trailing vol) to genuinely vol-scale the signal. + let analysis: Vec = if scaling { + kernel::vol_normalize(&returns, kernel::VOL_WINDOW) + } else { + returns.clone() + }; + + // Step 5 — kernel calls (sequential k loop — Pitfall 4 determinism). + let mut continuation_coefs: Vec = Vec::with_capacity(k_values.len()); + let mut t_stats: Vec = Vec::with_capacity(k_values.len()); + let mut p_values: Vec = Vec::with_capacity(k_values.len()); + let mut hit_rates: Vec = Vec::with_capacity(k_values.len()); + let mut tsmom_means: Vec = Vec::with_capacity(k_values.len()); + let mut turnover_per_bar: Vec = Vec::with_capacity(k_values.len()); + for &k in &k_values { + let res = kernel::tsmom_continuation(&analysis, k).map_err(ScanError::Kernel)?; + continuation_coefs.push(res.continuation_coef); + t_stats.push(res.t_stat); + p_values.push(res.p_value); + hit_rates.push(res.hit_rate); + tsmom_means.push(res.tsmom_mean); + turnover_per_bar.push(1.0 / usize_to_f64(k)); + } + + let k_values_f64: Vec = k_values.iter().map(|k| usize_to_f64(*k)).collect(); + + // Step 6 — select the hold horizon that best persists (AC3 framing). + let sel = select_hold_index(&continuation_coefs, &t_stats); + let selected_hold_bars = vec![k_values_f64[sel]]; + + // Step 7 — build raw.series (raw log returns + parallel timestamps). + #[allow( + clippy::cast_precision_loss, + reason = "epoch-ms fits exactly in f64 mantissa for realistic timestamps" + )] + let ts_ms: Vec = ctx + .bars + .ts_open_utc + .iter() + .skip(1) + .map(|t| t.timestamp_millis() as f64) + .collect(); + + // Envelope construction. effect.value = continuation at the selected + // (dominant) hold horizon; p / hit-rate are taken at that same k. + let mut extra: BTreeMap = BTreeMap::new(); + extra.insert( + "continuation_coefs".into(), + f64_slice_to_raw_array(&continuation_coefs), + ); + extra.insert("hit_rates".into(), f64_slice_to_raw_array(&hit_rates)); + extra.insert("k_values".into(), f64_slice_to_raw_array(&k_values_f64)); + extra.insert("p_values".into(), f64_slice_to_raw_array(&p_values)); + extra.insert( + "selected_hold_bars".into(), + f64_slice_to_raw_array(&selected_hold_bars), + ); + extra.insert("t_stats".into(), f64_slice_to_raw_array(&t_stats)); + extra.insert("tsmom_means".into(), f64_slice_to_raw_array(&tsmom_means)); + extra.insert( + "turnover_per_bar".into(), + f64_slice_to_raw_array(&turnover_per_bar), + ); + + let effect = Effect { + metric: EFFECT_METRIC.to_string(), + value: continuation_coefs[sel], + p_value: Some(p_values[sel]), + #[allow( + clippy::cast_possible_truncation, + reason = "n_returns <= u64 on all supported targets" + )] + n: Some(n_returns as u64), + ci95: None, + // Tradeable effect size: directional hit-rate at the selected k. + effect_size: Some(EffectSize { + kind: "hit_rate".to_string(), + value: hit_rates[sel], + }), + extra, + }; + + let mut series_map: BTreeMap = BTreeMap::new(); + series_map.insert("returns".into(), f64_slice_to_raw_array(&returns)); + series_map.insert("timestamps_ms".into(), f64_slice_to_raw_array(&ts_ms)); + let raw_block = Raw::new(series_map).map_err(|m| ScanError::Kernel(m.to_string()))?; + + let sources: Vec = req + .instruments + .iter() + .map(|spec| Source { + source_id: ctx.bars.source_id.clone(), + symbol: spec.symbol.clone(), + side: spec.side.as_str().to_string(), + timeframe: req.timeframe.as_str().to_string(), + }) + .collect(); + + let finding = ResultFinding { + schema_version: 1, + scan_id_at_version: format!("{SCAN_ID}@{SCAN_VERSION}"), + param_hash: req.param_hash.as_str().to_string(), + code_revision: ctx.code_revision.to_string(), + data_slice: DataSlice { + range: req.sub_range.clone(), + gap_manifest_ref: None, + gap_manifest: ctx.gap_manifest.cloned(), + sources, + }, + dsr: None, + fdr_q: None, + run_id: ctx.run_id, + produced_at_utc: Utc::now(), + params: req.resolved_params.clone(), + effect, + raw: Some(raw_block), + repro: None, + }; + + sink.write_envelope(&Finding::Result(finding))?; + Ok(()) + } +} + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +/// Select the hold horizon (index into the k-grid) that best persists: +/// the largest POSITIVE continuation t-stat (strongest momentum). When no +/// horizon shows positive continuation, fall back to the largest `|t-stat|`. +/// Ties resolve to the earliest index (smallest `k`). +fn select_hold_index(coefs: &[f64], t_stats: &[f64]) -> usize { + let mut best_pos: Option<(usize, f64)> = None; + for (i, (&c, &t)) in coefs.iter().zip(t_stats).enumerate() { + if c > 0.0 && t.is_finite() { + match best_pos { + Some((_, bt)) if bt >= t => {} + _ => best_pos = Some((i, t)), + } + } + } + if let Some((i, _)) = best_pos { + return i; + } + let mut best = (0usize, f64::NEG_INFINITY); + for (i, &t) in t_stats.iter().enumerate() { + let a = if t.is_finite() { t.abs() } else { 0.0 }; + if a > best.1 { + best = (i, a); + } + } + best.0 +} + +fn resolve_k_values(req: &ScanRequest) -> Result, ScanError> { + let raw = req.resolved_params.get("k_values"); + let arr = match raw { + None => { + return Ok(DEFAULT_K_VALUES + .iter() + .map(|i| usize::try_from(*i).expect("defaults positive")) + .collect()); + } + Some(v) => v.as_array().ok_or_else(|| { + ScanError::Kernel(format!( + "stats.momentum.tsmom: k_values must be an array; got {v}" + )) + })?, + }; + if arr.is_empty() { + return Err(ScanError::Kernel( + "stats.momentum.tsmom: k_values must be non-empty".into(), + )); + } + let mut out: Vec = Vec::with_capacity(arr.len()); + for v in arr { + let i = v.as_i64().ok_or_else(|| { + ScanError::Kernel(format!( + "stats.momentum.tsmom: k_values entries must be integers; got {v}" + )) + })?; + if i < 1 { + return Err(ScanError::Kernel(format!( + "stats.momentum.tsmom: k_values entries must be >= 1; got {i}" + ))); + } + let u = usize::try_from(i).map_err(|_| { + ScanError::Kernel(format!( + "stats.momentum.tsmom: k_values entry {i} out of usize range" + )) + })?; + out.push(u); + } + Ok(out) +} + +fn resolve_scaling(req: &ScanRequest) -> Result { + let raw = req.resolved_params.get("scaling"); + match raw { + None => Ok(true), + Some(v) => v.as_bool().ok_or_else(|| { + ScanError::Kernel(format!( + "stats.momentum.tsmom: scaling must be a boolean; got {v}" + )) + }), + } +} + +#[allow( + clippy::cast_precision_loss, + reason = "k << 2^52 for any realistic horizon" +)] +#[inline] +fn usize_to_f64(i: usize) -> f64 { + i as f64 +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::*; + use crate::aggregator::{BarFrame, Timeframe}; + use crate::engine::gap_policy::GapPolicyKind; + use crate::findings::TimeRange; + use crate::findings::run_id::RunId; + use crate::findings::sink::VecSink; + use crate::reader::{Blake3Hex, ClosedRangeUtc, InstrumentSpec, Side}; + use chrono::{DateTime, Duration, TimeZone}; + use std::sync::Arc; + use std::sync::atomic::AtomicBool; + + fn blake3_hex_zero() -> Blake3Hex { + let bytes: [u8; 64] = [b'0'; 64]; + Blake3Hex::from_hex_bytes(&bytes) + } + + /// LCG white-noise returns -> random-walk closes via `exp` accumulation. + #[allow(clippy::cast_possible_truncation)] + fn random_walk_closes(n: usize, seed: u64) -> Vec { + let mut s = seed as u32; + let mut out = Vec::with_capacity(n); + let mut price = 1.0_f64; + out.push(price); + for _ in 1..n { + s = s.wrapping_mul(1_664_525).wrapping_add(1_013_904_223); + let eps = (f64::from(s) / f64::from(u32::MAX) - 0.5) * 0.01; + price *= eps.exp(); + out.push(price); + } + out + } + + /// AR(1) positive-momentum log-returns -> closes via `exp` accumulation, + /// so `log_returns(closes)` recovers the AR(1) series. + #[allow(clippy::cast_possible_truncation)] + fn ar1_momentum_closes(n: usize, seed: u64, phi: f64) -> Vec { + let mut s = seed as u32; + let mut out = Vec::with_capacity(n); + let mut price = 1.0_f64; + out.push(price); + let mut prev = 0.0_f64; + for _ in 1..n { + s = s.wrapping_mul(1_664_525).wrapping_add(1_013_904_223); + let eps = (f64::from(s) / f64::from(u32::MAX) - 0.5) * 0.01; + let r = phi * prev + eps; + price *= r.exp(); + out.push(price); + prev = r; + } + out + } + + fn bar_frame_from_closes(closes: Vec) -> BarFrame { + let start = Utc.with_ymd_and_hms(2024, 1, 1, 0, 0, 0).unwrap(); + let n = closes.len(); + let ts_open: Vec> = (0..n) + .map(|i| { + let i_i64 = i64::try_from(i).expect("fits in i64"); + start + Duration::minutes(15 * i_i64) + }) + .collect(); + let ts_close: Vec> = + ts_open.iter().map(|t| *t + Duration::minutes(15)).collect(); + let opens = closes.clone(); + let highs: Vec = closes.iter().map(|c| c + 0.001).collect(); + let lows: Vec = closes.iter().map(|c| c - 0.001).collect(); + let vols = vec![1.0; n]; + BarFrame { + source_id: "dukascopy".into(), + symbol: "EURUSD".into(), + side: Side::Bid, + tf: Timeframe::Tf15m, + ts_open_utc: ts_open, + ts_close_utc: ts_close, + open: opens, + high: highs, + low: lows, + close: closes, + tick_volume: vols, + } + } + + fn sample_request_with_params(params: serde_json::Value) -> ScanRequest { + let start = Utc.with_ymd_and_hms(2024, 1, 1, 0, 0, 0).unwrap(); + let end = Utc.with_ymd_and_hms(2024, 1, 10, 0, 0, 0).unwrap(); + ScanRequest { + scan_id: SCAN_ID.into(), + version: SCAN_VERSION, + instruments: vec![InstrumentSpec { + symbol: "EURUSD".into(), + side: Side::Bid, + }], + timeframe: Timeframe::Tf15m, + window: ClosedRangeUtc { start, end }, + sub_range: TimeRange { + start_utc: start, + end_utc: end, + }, + gap_policy: GapPolicyKind::ContinuousOnly, + resolved_params: params, + param_hash: blake3_hex_zero(), + dry_run: false, + master_seed: None, + job_seed: None, + bootstrap_method: None, + bootstrap_n: None, + null_method: None, + null_n: None, + sleep_after_first_finding_ms: None, + } + } + + fn make_ctx(bars: &BarFrame, cancel: Arc) -> ScanCtx<'_> { + ScanCtx { + bars, + bars_pair: None, + gap_manifest: None, + run_id: RunId::new(), + code_revision: "abc1234", + cancel, + sleep_after_first_finding_ms: None, + } + } + + fn parse_sink_to_findings(sink: &VecSink) -> Vec { + sink.0 + .split(|b| *b == b'\n') + .filter(|line| !line.is_empty()) + .map(|line| serde_json::from_slice::(line).expect("parse")) + .collect() + } + + fn read_f64(arr: &RawArray, idx: usize) -> f64 { + let off = idx * 8; + let mut buf = [0u8; 8]; + buf.copy_from_slice(&arr.data.0[off..off + 8]); + f64::from_le_bytes(buf) + } + + // ----------------------------------------------------------------------- + + #[test] + fn tsmom_id_and_version() { + assert_eq!(TsmomScan.id(), "stats.momentum.tsmom"); + assert_eq!(TsmomScan.version(), 1); + } + + #[test] + fn tsmom_arity_is_single() { + assert_eq!(TsmomScan.arity(), ScanArity::Single); + } + + #[test] + fn tsmom_param_schema() { + let schema = TsmomScan.param_schema(); + assert_eq!(schema["type"], "object"); + assert_eq!(schema["properties"]["scaling"]["default"], true); + assert_eq!( + schema["properties"]["k_values"]["default"], + serde_json::json!([1, 5, 10, 20]) + ); + assert_eq!(schema["additionalProperties"], false); + } + + #[test] + fn tsmom_default_k_values_shapes() { + let bars = bar_frame_from_closes(random_walk_closes(600, 1)); + let mut sink = VecSink::new(); + let req = sample_request_with_params(serde_json::json!({})); + let ctx = make_ctx(&bars, Arc::new(AtomicBool::new(false))); + TsmomScan.run(&ctx, &req, &mut sink).expect("ok"); + let findings = parse_sink_to_findings(&sink); + let Finding::Result(r) = &findings[0] else { + panic!("expected Result"); + }; + for key in [ + "continuation_coefs", + "hit_rates", + "k_values", + "p_values", + "t_stats", + "tsmom_means", + "turnover_per_bar", + ] { + assert_eq!(r.effect.extra[key].shape, vec![4], "{key} length"); + } + // selected_hold_bars is a single-element array. + assert_eq!(r.effect.extra["selected_hold_bars"].shape, vec![1]); + } + + #[test] + fn tsmom_emits_one_result() { + let bars = bar_frame_from_closes(random_walk_closes(600, 2)); + let mut sink = VecSink::new(); + let req = sample_request_with_params(serde_json::json!({})); + let ctx = make_ctx(&bars, Arc::new(AtomicBool::new(false))); + TsmomScan.run(&ctx, &req, &mut sink).expect("ok"); + let findings = parse_sink_to_findings(&sink); + assert_eq!(findings.len(), 1); + assert!(matches!(&findings[0], Finding::Result(_))); + } + + #[test] + fn tsmom_result_envelope_shape() { + let bars = bar_frame_from_closes(random_walk_closes(600, 3)); + let mut sink = VecSink::new(); + let req = sample_request_with_params(serde_json::json!({"k_values": [1, 5]})); + let ctx = make_ctx(&bars, Arc::new(AtomicBool::new(false))); + TsmomScan.run(&ctx, &req, &mut sink).expect("ok"); + let findings = parse_sink_to_findings(&sink); + let Finding::Result(r) = &findings[0] else { + panic!("expected Result"); + }; + assert_eq!(r.scan_id_at_version, "stats.momentum.tsmom@1"); + assert_eq!(r.effect.metric, "tsmom_continuation"); + assert!( + r.effect.p_value.is_some(), + "headline continuation p present" + ); + let extra_keys: Vec<&str> = r.effect.extra.keys().map(String::as_str).collect(); + assert_eq!( + extra_keys, + vec![ + "continuation_coefs", + "hit_rates", + "k_values", + "p_values", + "selected_hold_bars", + "t_stats", + "tsmom_means", + "turnover_per_bar", + ] + ); + let raw = r.raw.as_ref().expect("raw"); + let raw_keys: Vec<&str> = raw.series.keys().map(String::as_str).collect(); + assert_eq!(raw_keys, vec!["returns", "timestamps_ms"]); + // effect_size carries the tradeable hit-rate. + let es = r.effect.effect_size.as_ref().expect("effect_size"); + assert_eq!(es.kind, "hit_rate"); + assert!(es.value >= 0.0 && es.value <= 1.0); + } + + /// AC2 (momentum half): injected positive serial dependence yields a + /// significant POSITIVE continuation at the matching short horizon. + #[test] + fn tsmom_ar1_significant_positive_continuation() { + let bars = bar_frame_from_closes(ar1_momentum_closes(3000, 99, 0.5)); + let mut sink = VecSink::new(); + // Disable scaling so the assertion reads the raw-return continuation. + let req = + sample_request_with_params(serde_json::json!({"k_values": [1], "scaling": false})); + let ctx = make_ctx(&bars, Arc::new(AtomicBool::new(false))); + TsmomScan.run(&ctx, &req, &mut sink).expect("ok"); + let findings = parse_sink_to_findings(&sink); + let Finding::Result(r) = &findings[0] else { + panic!("expected Result"); + }; + let coef = read_f64(&r.effect.extra["continuation_coefs"], 0); + let p = read_f64(&r.effect.extra["p_values"], 0); + assert!(coef > 0.0, "AR(1) continuation = {coef} should be > 0"); + assert!(p < 0.01, "AR(1) p = {p} should be significant"); + // Headline reflects the same (only) horizon. + assert!(r.effect.value > 0.0); + assert_eq!(read_f64(&r.effect.extra["selected_hold_bars"], 0), 1.0); + } + + /// AC2 (random-walk half): a random walk yields ~zero continuation and a + /// directional hit-rate ≈ 0.5 at the matching horizon. (Robust to the seed + /// — a `p > 0.05` check would carry an inherent ~5% per-seed false-positive + /// rate under the null; the near-zero coefficient + 0.5 hit-rate are the + /// substantive non-significance evidence and sit many σ inside tolerance.) + #[test] + fn tsmom_random_walk_near_zero_continuation() { + let bars = bar_frame_from_closes(random_walk_closes(3000, 7)); + let mut sink = VecSink::new(); + let req = + sample_request_with_params(serde_json::json!({"k_values": [1], "scaling": false})); + let ctx = make_ctx(&bars, Arc::new(AtomicBool::new(false))); + TsmomScan.run(&ctx, &req, &mut sink).expect("ok"); + let findings = parse_sink_to_findings(&sink); + let Finding::Result(r) = &findings[0] else { + panic!("expected Result"); + }; + let coef = read_f64(&r.effect.extra["continuation_coefs"], 0); + let hit = read_f64(&r.effect.extra["hit_rates"], 0); + let p = read_f64(&r.effect.extra["p_values"], 0); + assert!(coef.abs() < 0.15, "random-walk continuation = {coef} ≈ 0"); + assert!( + (hit - 0.5).abs() < 0.1, + "random-walk hit-rate = {hit} ≈ 0.5" + ); + assert!((0.0..=1.0).contains(&p), "p in [0,1]: {p}"); + } + + #[test] + fn tsmom_turnover_is_reciprocal_k() { + let bars = bar_frame_from_closes(random_walk_closes(600, 11)); + let mut sink = VecSink::new(); + let req = sample_request_with_params(serde_json::json!({"k_values": [2, 10]})); + let ctx = make_ctx(&bars, Arc::new(AtomicBool::new(false))); + TsmomScan.run(&ctx, &req, &mut sink).expect("ok"); + let findings = parse_sink_to_findings(&sink); + let Finding::Result(r) = &findings[0] else { + panic!("expected Result"); + }; + assert!((read_f64(&r.effect.extra["turnover_per_bar"], 0) - 0.5).abs() < 1e-12); + assert!((read_f64(&r.effect.extra["turnover_per_bar"], 1) - 0.1).abs() < 1e-12); + } + + #[test] + fn tsmom_cancellation() { + let bars = bar_frame_from_closes(random_walk_closes(64, 4)); + let mut sink = VecSink::new(); + let req = sample_request_with_params(serde_json::json!({})); + let ctx = make_ctx(&bars, Arc::new(AtomicBool::new(true))); + TsmomScan.run(&ctx, &req, &mut sink).expect("ok"); + assert!(sink.0.is_empty()); + } + + #[test] + fn tsmom_invalid_k_zero() { + let bars = bar_frame_from_closes(random_walk_closes(200, 5)); + let mut sink = VecSink::new(); + let req = sample_request_with_params(serde_json::json!({"k_values": [0]})); + let ctx = make_ctx(&bars, Arc::new(AtomicBool::new(false))); + let err = TsmomScan + .run(&ctx, &req, &mut sink) + .expect_err("reject k=0"); + assert!(matches!(err, ScanError::Kernel(_))); + } + + #[test] + fn tsmom_k_too_large_rejected() { + // 20 closes -> 19 returns. k=5 -> 19/5 = 3 blocks < 4 -> error. + let bars = bar_frame_from_closes(random_walk_closes(20, 6)); + let mut sink = VecSink::new(); + let req = sample_request_with_params(serde_json::json!({"k_values": [5]})); + let ctx = make_ctx(&bars, Arc::new(AtomicBool::new(false))); + let err = TsmomScan + .run(&ctx, &req, &mut sink) + .expect_err("reject k too large"); + assert!(matches!(err, ScanError::Kernel(_))); + } + + #[test] + fn tsmom_n_too_small_emits_scan_error() { + let bars = bar_frame_from_closes(vec![1.0, 1.01, 1.02]); // 2 returns < 4. + let mut sink = VecSink::new(); + let req = sample_request_with_params(serde_json::json!({})); + let ctx = make_ctx(&bars, Arc::new(AtomicBool::new(false))); + let err = TsmomScan + .run(&ctx, &req, &mut sink) + .expect_err("reject small n"); + match err { + ScanError::Kernel(msg) => assert!(msg.contains("InsufficientData"), "{msg}"), + other => panic!("expected Kernel; got {other:?}"), + } + } + + #[test] + fn tsmom_invalid_k_values_non_array() { + let bars = bar_frame_from_closes(random_walk_closes(200, 7)); + let mut sink = VecSink::new(); + let req = sample_request_with_params(serde_json::json!({"k_values": "garbage"})); + let ctx = make_ctx(&bars, Arc::new(AtomicBool::new(false))); + let err = TsmomScan + .run(&ctx, &req, &mut sink) + .expect_err("reject non-array"); + assert!(matches!(err, ScanError::Kernel(_))); + } + + #[test] + fn tsmom_invalid_scaling_non_bool() { + let bars = bar_frame_from_closes(random_walk_closes(200, 8)); + let mut sink = VecSink::new(); + let req = sample_request_with_params(serde_json::json!({"scaling": "yes"})); + let ctx = make_ctx(&bars, Arc::new(AtomicBool::new(false))); + let err = TsmomScan + .run(&ctx, &req, &mut sink) + .expect_err("reject non-bool scaling"); + assert!(matches!(err, ScanError::Kernel(_))); + } + + #[test] + fn tsmom_selected_hold_is_one_of_k() { + let bars = bar_frame_from_closes(ar1_momentum_closes(3000, 21, 0.4)); + let mut sink = VecSink::new(); + let req = sample_request_with_params(serde_json::json!({"k_values": [1, 5, 10, 20]})); + let ctx = make_ctx(&bars, Arc::new(AtomicBool::new(false))); + TsmomScan.run(&ctx, &req, &mut sink).expect("ok"); + let findings = parse_sink_to_findings(&sink); + let Finding::Result(r) = &findings[0] else { + panic!("expected Result"); + }; + let sel = read_f64(&r.effect.extra["selected_hold_bars"], 0); + assert!( + [1.0, 5.0, 10.0, 20.0].contains(&sel), + "selected hold {sel} must be one of the k grid" + ); + } + + #[test] + fn tsmom_select_hold_index_prefers_positive_max_t() { + // coefs: [+, -, +], t: [1.0, 5.0, 3.0] -> pick index 2 (positive, max t). + let idx = select_hold_index(&[0.1, -0.2, 0.3], &[1.0, 5.0, 3.0]); + assert_eq!(idx, 2); + } + + #[test] + fn tsmom_select_hold_index_fallback_abs_t() { + // All negative coefs -> fall back to largest |t| (index 1). + let idx = select_hold_index(&[-0.1, -0.2, -0.05], &[1.0, -6.0, 2.0]); + assert_eq!(idx, 1); + } +} diff --git a/crates/miner-core/tests/scan_tsmom.rs b/crates/miner-core/tests/scan_tsmom.rs new file mode 100644 index 0000000..d8e026f --- /dev/null +++ b/crates/miner-core/tests/scan_tsmom.rs @@ -0,0 +1,202 @@ +//! RAD-3839 — ANOM-12 `stats.momentum.tsmom@1` happy-path integration test. +//! +//! Pattern analog: `crates/miner-core/tests/scan_variance_ratio.rs` (sibling +//! multi-`k` ANOM scan). Asserts the full envelope shape inline, then pins a +//! float-free `insta` *schema* snapshot (scan-id, metric, effect-size kind, +//! parallel-array names + shapes, raw-series names + shapes, params). The +//! schema snapshot is deterministic regardless of the underlying float math +//! (the continuation coefficients / p-values are exercised for sign and +//! significance by the unit tests in the scan module). + +#![allow(clippy::cast_precision_loss, clippy::too_many_lines)] + +mod common; + +use std::sync::Arc; +use std::sync::atomic::AtomicBool; + +use chrono::{Duration, TimeZone, Utc}; + +use miner_core::aggregator::{BarFrame, Timeframe}; +use miner_core::engine::gap_policy::GapPolicyKind; +use miner_core::engine::param_hash; +use miner_core::findings::{Finding, RunId, TimeRange}; +use miner_core::reader::{ClosedRangeUtc, InstrumentSpec, Side}; +use miner_core::scan::anom::TsmomScan; +use miner_core::scan::{Scan, ScanCtx, ScanRequest}; + +use common::BufferSink; + +/// Random-walk close series: close[i] = close[i-1] * exp(eps), eps IID — so +/// `log_returns` are white noise and the continuation coefficient is ≈ 0. +#[allow(clippy::cast_possible_truncation)] +fn random_walk_closes(n: usize, seed: u64) -> Vec { + let mut s = seed as u32; + let mut out = Vec::with_capacity(n); + let mut price = 1.0_f64; + out.push(price); + for _ in 1..n { + s = s.wrapping_mul(1_664_525).wrapping_add(1_013_904_223); + let eps = (f64::from(s) / f64::from(u32::MAX) - 0.5) * 0.01; + price *= eps.exp(); + out.push(price); + } + out +} + +fn build_bar_frame_from_closes(close: &[f64]) -> BarFrame { + let start = Utc.with_ymd_and_hms(2024, 1, 1, 0, 0, 0).unwrap(); + let n = close.len(); + let ts_open: Vec> = (0..n) + .map(|i| { + let i_i64 = i64::try_from(i).expect("fits in i64"); + start + Duration::minutes(15 * i_i64) + }) + .collect(); + let ts_close: Vec> = + ts_open.iter().map(|t| *t + Duration::minutes(15)).collect(); + let opens: Vec = close.to_vec(); + let highs: Vec = close.iter().map(|c| c + 0.001).collect(); + let lows: Vec = close.iter().map(|c| c - 0.001).collect(); + let vols = vec![1.0; n]; + BarFrame { + source_id: "dukascopy".into(), + symbol: "EURUSD".into(), + side: Side::Bid, + tf: Timeframe::Tf15m, + ts_open_utc: ts_open, + ts_close_utc: ts_close, + open: opens, + high: highs, + low: lows, + close: close.to_vec(), + tick_volume: vols, + } +} + +#[test] +fn scan_tsmom_happy_path() { + let closes = random_walk_closes(600, 42); + let bars = build_bar_frame_from_closes(&closes); + + let resolved_params = serde_json::json!({"k_values": [1, 5, 10, 20], "scaling": true}); + let param_hash = param_hash::param_hash(&resolved_params).expect("ok"); + let window_start = Utc.with_ymd_and_hms(2024, 1, 1, 0, 0, 0).unwrap(); + let window_end = Utc.with_ymd_and_hms(2024, 1, 10, 0, 0, 0).unwrap(); + let req = ScanRequest { + scan_id: "stats.momentum.tsmom".into(), + version: 1, + instruments: vec![InstrumentSpec { + symbol: "EURUSD".into(), + side: Side::Bid, + }], + timeframe: Timeframe::Tf15m, + window: ClosedRangeUtc { + start: window_start, + end: window_end, + }, + sub_range: TimeRange { + start_utc: window_start, + end_utc: window_end, + }, + gap_policy: GapPolicyKind::ContinuousOnly, + resolved_params, + param_hash, + dry_run: false, + master_seed: None, + job_seed: None, + bootstrap_method: None, + bootstrap_n: None, + null_method: None, + null_n: None, + #[cfg(any(test, feature = "test-internal"))] + sleep_after_first_finding_ms: None, + }; + let ctx = ScanCtx { + bars: &bars, + bars_pair: None, + gap_manifest: None, + run_id: RunId::new(), + code_revision: "test-rev-abc1234", + cancel: Arc::new(AtomicBool::new(false)), + sleep_after_first_finding_ms: None, + }; + + let mut sink = BufferSink::new(); + TsmomScan.run(&ctx, &req, &mut sink).expect("scan ok"); + + let findings = common::parse_findings(&sink.0); + assert_eq!(findings.len(), 1, "exactly one envelope"); + let Finding::Result(ref r) = findings[0] else { + panic!("expected Result"); + }; + assert_eq!(r.scan_id_at_version, "stats.momentum.tsmom@1"); + assert_eq!(r.effect.metric, "tsmom_continuation"); + assert!( + r.effect.p_value.is_some(), + "TSMOM reports a headline continuation p at the selected hold horizon" + ); + assert_eq!(r.effect.n, Some(599), "n = closes - 1 log returns"); + + let extra_keys: Vec<&str> = r.effect.extra.keys().map(String::as_str).collect(); + assert_eq!( + extra_keys, + vec![ + "continuation_coefs", + "hit_rates", + "k_values", + "p_values", + "selected_hold_bars", + "t_stats", + "tsmom_means", + "turnover_per_bar", + ] + ); + for key in [ + "continuation_coefs", + "hit_rates", + "k_values", + "p_values", + "t_stats", + "tsmom_means", + "turnover_per_bar", + ] { + assert_eq!(r.effect.extra[key].shape, vec![4], "{key} length mismatch"); + } + assert_eq!( + r.effect.extra["selected_hold_bars"].shape, + vec![1], + "selected_hold_bars is a single-element array" + ); + + let es = r.effect.effect_size.as_ref().expect("effect_size present"); + assert_eq!(es.kind, "hit_rate"); + assert!(es.value >= 0.0 && es.value <= 1.0, "hit-rate in [0,1]"); + + let raw = r.raw.as_ref().expect("raw present"); + let raw_keys: Vec<&str> = raw.series.keys().map(String::as_str).collect(); + assert_eq!(raw_keys, vec!["returns", "timestamps_ms"]); + + // Float-free schema snapshot — pins the catalogue-facing shape without + // coupling to the exact continuation/p-value floats (those are covered by + // the scan-module unit tests for sign + significance). + let mut extra_shapes = serde_json::Map::new(); + for (k, arr) in &r.effect.extra { + extra_shapes.insert(k.clone(), serde_json::json!(arr.shape)); + } + let mut raw_shapes = serde_json::Map::new(); + for (k, arr) in &raw.series { + raw_shapes.insert(k.clone(), serde_json::json!(arr.shape)); + } + let schema = serde_json::json!({ + "scan_id_at_version": r.scan_id_at_version, + "effect_metric": r.effect.metric, + "effect_size_kind": es.kind, + "headline_p_value_present": r.effect.p_value.is_some(), + "n": r.effect.n, + "effect_extra": serde_json::Value::Object(extra_shapes), + "raw_series": serde_json::Value::Object(raw_shapes), + "params": r.params, + }); + insta::assert_json_snapshot!("tsmom_schema", schema); +} diff --git a/crates/miner-core/tests/snapshots/scan_tsmom__tsmom_schema.snap b/crates/miner-core/tests/snapshots/scan_tsmom__tsmom_schema.snap new file mode 100644 index 0000000..302877a --- /dev/null +++ b/crates/miner-core/tests/snapshots/scan_tsmom__tsmom_schema.snap @@ -0,0 +1,54 @@ +--- +source: crates/miner-core/tests/scan_tsmom.rs +expression: schema +--- +{ + "effect_extra": { + "continuation_coefs": [ + 4 + ], + "hit_rates": [ + 4 + ], + "k_values": [ + 4 + ], + "p_values": [ + 4 + ], + "selected_hold_bars": [ + 1 + ], + "t_stats": [ + 4 + ], + "tsmom_means": [ + 4 + ], + "turnover_per_bar": [ + 4 + ] + }, + "effect_metric": "tsmom_continuation", + "effect_size_kind": "hit_rate", + "headline_p_value_present": true, + "n": 599, + "params": { + "k_values": [ + 1, + 5, + 10, + 20 + ], + "scaling": true + }, + "raw_series": { + "returns": [ + 599 + ], + "timestamps_ms": [ + 599 + ] + }, + "scan_id_at_version": "stats.momentum.tsmom@1" +} From 76716a3d63a6585c940a6df40cf9817e5d40e324 Mon Sep 17 00:00:00 2001 From: Cody Date: Sat, 6 Jun 2026 15:35:08 +0000 Subject: [PATCH 2/2] style(scan): suppress clippy lints in tsmom kernel/dispatch Add #[allow(clippy::many_single_char_names)] to tsmom_continuation (n/k/m OLS names, x/y regressor/regressand) and rename the k-loop result binding res -> k_result to clear clippy::similar_names vs the run() req parameter. Pure lint hygiene; no behaviour change. --- crates/miner-core/src/scan/anom/tsmom/kernel.rs | 4 ++++ crates/miner-core/src/scan/anom/tsmom/mod.rs | 12 ++++++------ 2 files changed, 10 insertions(+), 6 deletions(-) diff --git a/crates/miner-core/src/scan/anom/tsmom/kernel.rs b/crates/miner-core/src/scan/anom/tsmom/kernel.rs index 36d5b5e..76e7f37 100644 --- a/crates/miner-core/src/scan/anom/tsmom/kernel.rs +++ b/crates/miner-core/src/scan/anom/tsmom/kernel.rs @@ -127,6 +127,10 @@ pub(crate) fn vol_normalize(returns: &[f64], window: usize) -> Vec { clippy::similar_names, reason = "sx/sy/sxx/sxy/dx/dy are the canonical OLS accumulator names" )] +#[allow( + clippy::many_single_char_names, + reason = "n/k/m are canonical OLS/TSMOM names; x/y are the standard regressor/regressand pair" +)] pub(crate) fn tsmom_continuation(returns: &[f64], k: usize) -> Result { let n = returns.len(); if k < 1 { diff --git a/crates/miner-core/src/scan/anom/tsmom/mod.rs b/crates/miner-core/src/scan/anom/tsmom/mod.rs index 2543ef2..e37a671 100644 --- a/crates/miner-core/src/scan/anom/tsmom/mod.rs +++ b/crates/miner-core/src/scan/anom/tsmom/mod.rs @@ -185,12 +185,12 @@ impl Scan for TsmomScan { let mut tsmom_means: Vec = Vec::with_capacity(k_values.len()); let mut turnover_per_bar: Vec = Vec::with_capacity(k_values.len()); for &k in &k_values { - let res = kernel::tsmom_continuation(&analysis, k).map_err(ScanError::Kernel)?; - continuation_coefs.push(res.continuation_coef); - t_stats.push(res.t_stat); - p_values.push(res.p_value); - hit_rates.push(res.hit_rate); - tsmom_means.push(res.tsmom_mean); + let k_result = kernel::tsmom_continuation(&analysis, k).map_err(ScanError::Kernel)?; + continuation_coefs.push(k_result.continuation_coef); + t_stats.push(k_result.t_stat); + p_values.push(k_result.p_value); + hit_rates.push(k_result.hit_rate); + tsmom_means.push(k_result.tsmom_mean); turnover_per_bar.push(1.0 / usize_to_f64(k)); }