[verified] Factor-learning validation gate (no auto-apply)
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).
This commit is contained in:
Kunthawat Greethong
@@ -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}})
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user