From ae814c341e0c1cdb147a6ad853faf003df94b5c4 Mon Sep 17 00:00:00 2001 From: Kunthawat Greethong Date: Thu, 27 Aug 2026 12:17:21 +0700 Subject: [PATCH] [verified] Factor-learning validation gate (no auto-apply) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a strict holdout/walk-forward + baseline gate to factor-weight learning, per the P4 guardrail: learned weights are never auto-applied until minimum sample, holdout/walk-forward, and baseline comparison all pass. - backend/app/weight_learning.py: - FactorLearning gained ic_train / ic_holdout / validated / gate_notes. - apply_validation_gate(...) splits a chronological IC series into train + holdout and only marks validated=True when: total >= MIN_SAMPLE_PERIODS, each window >= its minimum, train AND holdout IC are positive (beat the BASELINE_IC=0) and agree in sign, and the pooled |t| > MIN_IC_TSTAT. - apply_weight_update now keeps new_weight == old_weight for any factor that is not validated (no auto-apply); only validated factors move. - learn_momentum_gated(...) builds PIT momentum ICs then applies the gate. - backend/app/__init__.py: /api/v1/learning/momentum uses the gated learner and surfaces ic_train/ic_holdout/validated/gate_notes. - tests: gate (16) via rewritten suite — full backend 286 passed. Live probe on current price archive: validated=false with gate_note 'IC not above baseline (0.0711/-0.1143)' — momentum is not validated, weight stays unchanged (new_weight=None). --- backend/app/__init__.py | 11 ++- backend/app/weight_learning.py | 135 +++++++++++++++++++++++++- backend/tests/test_weight_learning.py | 101 ++++++++++++------- 3 files changed, 209 insertions(+), 38 deletions(-) diff --git a/backend/app/__init__.py b/backend/app/__init__.py index aa888ca..d9a1b09 100644 --- a/backend/app/__init__.py +++ b/backend/app/__init__.py @@ -920,9 +920,12 @@ def create_app(config: dict[str, Any] | None = None) -> Flask: Runs a strictly point-in-time IC analysis over the price history: does 12-1 momentum at month t predict 3m forward returns across the - cross-section? Reports mean IC, t-stat, n periods, and a suggested - weight delta. Honest: returns 503 when no price snapshot exists. - Macro/demographic factors are not yet attributable (no vintages). + cross-section? Applies a split-sample holdout validation gate — + reports mean/train/holdout IC, t-stat, n periods, `validated`, why not + (gate_notes), and a suggested weight. The learned weight is NEVER + auto-applied: `validated=false` keeps the weight unchanged. Honest: + 503 when no price snapshot exists. Macro/demographic factors are not + yet attributable (no vintages). """ from app import weight_learning as wl from app import simulation as sim @@ -934,7 +937,7 @@ def create_app(config: dict[str, Any] | None = None) -> Flask: return jsonify({"error": f"price snapshot: {exc}"}), 503 symbols = sorted(series.keys()) try: - learning = wl.learn_momentum(series, symbols, start, end) + learning = wl.learn_momentum_gated(series, symbols, start, end) except (wl.WeightLearningError, ValueError) as exc: return jsonify({"error": str(exc)}), 400 return jsonify({"factor": learning.to_dict(), "window": {"start": start, "end": end}}) diff --git a/backend/app/weight_learning.py b/backend/app/weight_learning.py index 41b8137..335e946 100644 --- a/backend/app/weight_learning.py +++ b/backend/app/weight_learning.py @@ -45,6 +45,10 @@ class FactorLearning: old_weight: Optional[float] = None new_weight: Optional[float] = None blocked: bool = False # True when no PIT historical series is available + ic_train: Optional[float] = None # walk-forward train-window mean IC + ic_holdout: Optional[float] = None # holdout-window mean IC (out-of-sample) + validated: bool = False # True only when the gate fully passes + gate_notes: list = field(default_factory=list) # why not validated def to_dict(self) -> dict: return { @@ -53,9 +57,13 @@ class FactorLearning: "ic_mean": self.ic_mean, "ic_std": self.ic_std, "ic_tstat": self.ic_tstat, + "ic_train": self.ic_train, + "ic_holdout": self.ic_holdout, "old_weight": self.old_weight, "new_weight": self.new_weight, "blocked": self.blocked, + "validated": self.validated, + "gate_notes": self.gate_notes, } @@ -166,6 +174,77 @@ def learn_factor_series(period_ics: list[float]) -> FactorLearning: return res +# Default gate thresholds (honest, conservative — from the user's P4 guardrail: +# no auto-apply until minimum sample + holdout/walk-forward + baseline pass). +MIN_SAMPLE_PERIODS = 12 # at least a year of monthly periods for any learn +MIN_HOLDOUT_PERIODS = 6 # holdout window must be meaningful, not trivial +MIN_TRAIN_PERIODS = 6 # train window must be non-trivial +# A factor "beats baseline (random)" when both train and holdout IC are +# positive with a t-stat magnitude beyond this (else it is indistinguishable +# from noise and must not move weights). +MIN_IC_TSTAT = 1.0 +BASELINE_IC = 0.0 # naive benchmark: no predictive power (IC = 0) + + +def apply_validation_gate(learning: FactorLearning, + train_ics: list[float], + holdout_ics: list[float]) -> FactorLearning: + """Apply the strict holdout/walk-forward + baseline gate. + + Sets ``ic_train`` / ``ic_holdout`` and decides ``validated``. A factor is + only ``validated=True`` when ALL of: + + - total periods >= MIN_SAMPLE_PERIODS + - train and holdout windows each >= their minimums + - train IC and holdout IC agree in sign AND are positive (beat baseline) + - the pooled IC is significant enough (|t| > MIN_IC_TSTAT) + + This is intentionally strict: unvalidated factors keep their weight (the + caller must never auto-apply). + """ + total = len(train_ics) + len(holdout_ics) + notes: list[str] = [] + if total < MIN_SAMPLE_PERIODS: + notes.append(f"sample too small ({total} < {MIN_SAMPLE_PERIODS})") + + def _mean(xs): + return round(statistics.fmean(xs), 4) if xs else None + learning.ic_train = _mean(train_ics) + learning.ic_holdout = _mean(holdout_ics) + + validated = True + if len(train_ics) < MIN_TRAIN_PERIODS: + notes.append(f"train too small ({len(train_ics)} < {MIN_TRAIN_PERIODS})") + validated = False + if len(holdout_ics) < MIN_HOLDOUT_PERIODS: + notes.append(f"holdout too small ({len(holdout_ics)} < {MIN_HOLDOUT_PERIODS})") + validated = False + # train & holdout must both be positive (beat baseline) and agree in sign + if validated: + if not train_ics or not holdout_ics: + notes.append("missing train or holdout IC") + validated = False + elif learning.ic_train <= BASELINE_IC or learning.ic_holdout <= BASELINE_IC: + notes.append(f"IC not above baseline ({learning.ic_train}/{learning.ic_holdout})") + validated = False + elif (learning.ic_train < 0) != (learning.ic_holdout < 0): + notes.append("train and holdout IC disagree in sign") + validated = False + # significance: pooled t-stat must clear the bar + pooled = (train_ics + holdout_ics) if validated else [] + if validated and pooled: + m = statistics.fmean(pooled) + sd = statistics.pstdev(pooled) + t = (m / (sd / math.sqrt(len(pooled)))) if sd else None + if t is None or abs(t) <= MIN_IC_TSTAT: + notes.append(f"IC not significant (|t|={t} <= {MIN_IC_TSTAT})") + validated = False + + learning.validated = validated + learning.gate_notes = notes + return learning + + # --------------------------------------------------------------------------- # Momentum factor learning (real PIT demo) # --------------------------------------------------------------------------- @@ -203,12 +282,66 @@ def apply_weight_update(learning: FactorLearning, shrink: float = DEFAULT_SHRINK """Fold a learned IC into the factor's weight (in place). new = clip(old * (1 + shrink * ic_mean), min_w, max_w). - Blocked / no-data factors keep their weight (new == old). + + **Never auto-applies an unvalidated factor.** A weight only moves when the + factor is ``validated`` (passed minimum sample + holdout/walk-forward + + baseline via ``apply_validation_gate``). Blocked, no-data, insufficient, or + non-significant factors keep their weight (new == old) and stay + ``validated=False``. """ if learning.old_weight is None: return if learning.blocked or learning.ic_mean is None or learning.n_periods < 3: learning.new_weight = learning.old_weight return + if not learning.validated: + # gate not passed -> do NOT move the weight (honest, no auto-apply) + learning.new_weight = learning.old_weight + return nw = learning.old_weight * (1.0 + shrink * learning.ic_mean) learning.new_weight = round(min(max(nw, min_w), max_w), 4) + + +def learn_momentum_gated(series: dict, symbols: list[str], start: str, end: str, + step_days: int = 21, forward_months: int = FORWARD_MONTHS, + holdout_fraction: float = 0.3) -> FactorLearning: + """Learn 12-1 momentum with a strict holdout validation gate. + + Builds per-period PIT ICs across [start, end], then splits chronologically + into train (first 1-holdout_fraction) and holdout (last fraction) windows + and applies ``apply_validation_gate``. Returns a FactorLearning whose + ``validated`` is True only if the factor clears the gate; unvalidated runs + keep their weight (the caller must not auto-apply). + """ + s = dt.date.fromisoformat(start) + e = dt.date.fromisoformat(end) + all_ics: list[float] = [] + cur = s + while cur <= e: + fv: dict[str, float] = {} + fr: dict[str, float] = {} + for sym in symbols: + m = momentum_at(series, sym, cur) + f = _forward_return(series, sym, cur, forward_months) + if m is not None and f is not None: + fv[sym] = m + fr[sym] = f + ic = spearman_ic(fv, fr) + if ic is not None: + all_ics.append(ic) + cur += dt.timedelta(days=step_days) + + res = learn_factor_series(all_ics) + res.factor_key = "momentum_12_1" + + # chronologically split train / holdout (walk-forward style) + holdout_n = max(0, int(round(len(all_ics) * holdout_fraction))) + if holdout_n > 0: + split = len(all_ics) - holdout_n + train_ics = all_ics[:split] + holdout_ics = all_ics[split:] + res = apply_validation_gate(res, train_ics, holdout_ics) + else: + res.gate_notes.append("no holdout window (too few periods)") + res.validated = False + return res diff --git a/backend/tests/test_weight_learning.py b/backend/tests/test_weight_learning.py index bc94d74..8894e80 100644 --- a/backend/tests/test_weight_learning.py +++ b/backend/tests/test_weight_learning.py @@ -1,4 +1,4 @@ -"""Tests for the factor-weight learning loop (P4).""" +"""Tests for the factor-weight learning loop (P4) + validation gate.""" from __future__ import annotations @@ -10,19 +10,14 @@ from app import weight_learning as wl class SpearmanICTest(unittest.TestCase): def test_perfect_positive(self): - # Factor and forward returns perfectly rank-aligned -> IC = 1. fv = {"A": 1.0, "B": 2.0, "C": 3.0, "D": 4.0} fr = {"A": 0.1, "B": 0.2, "C": 0.3, "D": 0.4} - ic = wl.spearman_ic(fv, fr) - self.assertIsNotNone(ic) - self.assertTrue(ic is not None and abs(ic - 1.0) < 1e-5) + self.assertTrue(abs(wl.spearman_ic(fv, fr) - 1.0) < 1e-5) def test_inverse_is_negative_one(self): fv = {"A": 1.0, "B": 2.0, "C": 3.0, "D": 4.0} fr = {"A": 0.4, "B": 0.3, "C": 0.2, "D": 0.1} - ic = wl.spearman_ic(fv, fr) - self.assertIsNotNone(ic) - self.assertTrue(ic is not None and abs(ic + 1.0) < 1e-5) + self.assertTrue(abs(wl.spearman_ic(fv, fr) + 1.0) < 1e-5) def test_too_few_symbols_returns_none(self): self.assertIsNone(wl.spearman_ic({"A": 1.0}, {"A": 0.1})) @@ -30,75 +25,115 @@ class SpearmanICTest(unittest.TestCase): def test_invalid_symbols_excluded(self): fv = {"A": 1.0, "B": 2.0, "C": 3.0, "D": 4.0, "E": float("nan")} fr = {"A": 0.1, "B": 0.2, "C": 0.3, "D": 0.4, "E": 0.5} - ic = wl.spearman_ic(fv, fr) - self.assertIsNotNone(ic) - self.assertTrue(ic is not None and abs(ic - 1.0) < 1e-5) + self.assertTrue(abs(wl.spearman_ic(fv, fr) - 1.0) < 1e-5) -class WeightUpdateTest(unittest.TestCase): - def test_positive_ic_raises_weight(self): - l = wl.FactorLearning("f", n_periods=12, ic_mean=0.3, - old_weight=1.0) +class ValidationGateTest(unittest.TestCase): + def test_passes_when_both_windows_positive_and_significant(self): + l = wl.learn_factor_series([0.2] * 10) + # train (10 positive, all strong) + holdout (6 positive) -> validated + train = [0.2] * 10 + holdout = [0.3] * 6 + res = wl.apply_validation_gate(l, train, holdout) + self.assertTrue(res.validated) + self.assertEqual(res.gate_notes, []) + self.assertIsNotNone(res.ic_train) + self.assertIsNotNone(res.ic_holdout) + + def test_fails_when_sample_too_small(self): + l = wl.learn_factor_series([0.2] * 6) + res = wl.apply_validation_gate(l, [0.2] * 3, [0.3] * 3) + self.assertFalse(res.validated) + self.assertTrue(any("sample too small" in n for n in res.gate_notes)) + + def test_fails_when_holdout_not_above_baseline(self): + l = wl.learn_factor_series([0.2] * 10) + res = wl.apply_validation_gate(l, [0.2] * 10, [-0.1] * 6) + self.assertFalse(res.validated) + self.assertTrue(any("baseline" in n for n in res.gate_notes)) + + def test_fails_when_train_and_holdout_disagree_in_sign(self): + l = wl.learn_factor_series([0.2] * 10) + res = wl.apply_validation_gate(l, [0.2] * 10, [-0.05] * 6) + self.assertFalse(res.validated) + + def test_fails_when_ic_not_significant(self): + l = wl.learn_factor_series([0.02] * 10) + res = wl.apply_validation_gate(l, [0.02] * 10, [0.02] * 6) + self.assertFalse(res.validated) # tiny IC, |t| below bar + + +class WeightUpdateGateTest(unittest.TestCase): + def test_unvalidated_factor_never_auto_applies(self): + # positive IC but NOT validated -> weight must NOT move + l = wl.FactorLearning("f", n_periods=20, ic_mean=0.3, old_weight=1.0) + wl.apply_weight_update(l, shrink=0.5) + self.assertEqual(l.new_weight, 1.0) # unchanged (no auto-apply) + self.assertFalse(l.validated) + + def test_validated_positive_ic_raises_weight(self): + l = wl.FactorLearning("f", n_periods=20, ic_mean=0.3, old_weight=1.0, + validated=True) wl.apply_weight_update(l, shrink=0.5) self.assertIsNotNone(l.new_weight) - self.assertTrue(l.new_weight is not None and l.new_weight > 1.0) + self.assertTrue(l.new_weight > 1.0) - def test_negative_ic_lowers_weight(self): - l = wl.FactorLearning("f", n_periods=12, ic_mean=-0.4, - old_weight=1.0) + def test_validated_negative_ic_lowers_weight(self): + l = wl.FactorLearning("f", n_periods=20, ic_mean=-0.4, old_weight=1.0, + validated=True) wl.apply_weight_update(l, shrink=0.5) self.assertIsNotNone(l.new_weight) - self.assertTrue(l.new_weight is not None and l.new_weight < 1.0) + self.assertTrue(l.new_weight < 1.0) def test_clamped_to_bounds(self): - l = wl.FactorLearning("f", n_periods=12, ic_mean=10.0, old_weight=1.0) + l = wl.FactorLearning("f", n_periods=20, ic_mean=10.0, old_weight=1.0, + validated=True) wl.apply_weight_update(l, shrink=1.0, max_w=3.0) self.assertEqual(l.new_weight, 3.0) def test_blocked_keeps_weight(self): - l = wl.FactorLearning("f", n_periods=0, ic_mean=None, - old_weight=1.0, blocked=True) + l = wl.FactorLearning("f", n_periods=0, ic_mean=None, old_weight=1.0, + blocked=True) wl.apply_weight_update(l) self.assertEqual(l.new_weight, 1.0) def test_no_old_weight_returns(self): - l = wl.FactorLearning("f", n_periods=12, ic_mean=0.2, old_weight=None) + l = wl.FactorLearning("f", n_periods=12, ic_mean=0.2, old_weight=None, + validated=True) wl.apply_weight_update(l) self.assertIsNone(l.new_weight) -class MomentumLearningTest(unittest.TestCase): +class MomentumGatedLearningTest(unittest.TestCase): @staticmethod def _series(): - # A trends up strongly (positive momentum), B flat. def bars(base, drift): out = [] - for i in range(500): + for i in range(600): d = (dt.date(2024, 1, 1) + dt.timedelta(days=i)).isoformat() out.append({"date": d, "adjusted_close": base + drift * i}) return out return {"A": {"bars": bars(10.0, 0.05)}, "B": {"bars": bars(20.0, 0.0)}} - def test_learn_runs_without_error(self): - res = wl.learn_momentum(self._series(), ["A", "B"], - "2025-06-01", "2026-06-01") + def test_gated_runs_and_returns_validated_flag(self): + res = wl.learn_momentum_gated(self._series(), ["A", "B"], + "2025-01-01", "2026-06-01") self.assertIsInstance(res, wl.FactorLearning) + self.assertIn(res.validated, (True, False)) # always a boolean verdict self.assertGreaterEqual(res.n_periods, 0) class AddMonthsTest(unittest.TestCase): def test_clamps_day_to_end_of_month(self): - # Jan 31 + 1 month must clamp to Feb 28/29, not raise. self.assertEqual(wl._add_months(dt.date(2026, 1, 31), 1), dt.date(2026, 2, 28)) self.assertEqual(wl._add_months(dt.date(2026, 1, 31), 2), dt.date(2026, 3, 31)) - self.assertEqual(wl._add_months(dt.date(2024, 1, 31), 1), dt.date(2024, 2, 29)) # leap + self.assertEqual(wl._add_months(dt.date(2024, 1, 31), 1), dt.date(2024, 2, 29)) class LearnFactorSeriesTest(unittest.TestCase): def test_aggregates_ics(self): res = wl.learn_factor_series([0.1, 0.2, 0.3, 0.4]) self.assertEqual(res.n_periods, 4) - self.assertIsNotNone(res.ic_mean) self.assertTrue(res.ic_mean is not None and abs(res.ic_mean - 0.25) < 1e-6) self.assertIsNotNone(res.ic_tstat)