From 86eadae42f9f34eb52689bd678e4e4e02de855d8 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Sun, 14 Jun 2026 20:24:50 +0200 Subject: [PATCH] =?UTF-8?q?fix(factory):=20capital-correct=20loop=20?= =?UTF-8?q?=E2=80=94=20record=5Fcycle,=20sequential=20HRP-weighted=20admis?= =?UTF-8?q?sion,=20kill-switch,=20tests?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fix 1: wire record_cycle in factory-cycle --execute so the funnel table is populated. LoopResult gains deployed_before/deployed_after/forward_count; cli.py calls fs.record_cycle(proposed=forward_count, ...) and record_allocation after the loop. Fix 2+4: replace equal-weight book proxy with HRP-weighted blend of deployed sleeves; rewrite admission as sequential — each candidate is re-gated against the running mean of already-admitted-this-cycle candidates, so two candidates uncorrelated to the book but correlated to each other cannot both be admitted in one cycle. Fix 3: HRP inclusion now requires >= min_forward_days (was >= 2). Folded into the final rets comprehension in allocate_and_promote. Fix 5: StrategyStoreLoopAdapter.set_status now asserts status == "DEPLOYED" so a future demotion path cannot silently promote via the loop adapter. Fix 6: persisted kill-switch — FactoryFlagRow ORM model added to cockpit_models.py; FactoryStore.set_kill(on)/is_killed() read/write a key="kill" row; factory-cycle folds db kill with env kill; factory-flatten renamed to factory-halt; factory-resume and factory-status commands added. Fix 7: sleeve returns fetched once per sleeve in the final HRP comprehension via a shared dict, avoiding double store I/O. Fix 8: three new integration tests covering min_forward_days filter, empty-book founding-admit-then-gate, and mutual-correlation blocking within a cycle. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../adapters/persistence/cockpit_models.py | 7 ++ .../adapters/persistence/factory_store.py | 15 ++- src/fxhnt/application/factory/factory_loop.py | 91 ++++++++++++++----- src/fxhnt/cli.py | 54 +++++++++-- tests/integration/test_factory_loop.py | 57 ++++++++++++ tests/integration/test_factory_store.py | 10 ++ 6 files changed, 201 insertions(+), 33 deletions(-) diff --git a/src/fxhnt/adapters/persistence/cockpit_models.py b/src/fxhnt/adapters/persistence/cockpit_models.py index a8593e0..2e24228 100644 --- a/src/fxhnt/adapters/persistence/cockpit_models.py +++ b/src/fxhnt/adapters/persistence/cockpit_models.py @@ -80,3 +80,10 @@ class FactoryAllocationRow(CockpitBase): strategy_id: Mapped[str] = mapped_column(String(64), primary_key=True) weight: Mapped[float] = mapped_column(Float) at: Mapped[dt.datetime] = mapped_column(DateTime) + + +class FactoryFlagRow(CockpitBase): + """Persisted kill-switch and other factory flags (key/value). key="kill", value="1"/"0".""" + __tablename__ = "factory_flags" + key: Mapped[str] = mapped_column(String(32), primary_key=True) + value: Mapped[str] = mapped_column(String(16)) diff --git a/src/fxhnt/adapters/persistence/factory_store.py b/src/fxhnt/adapters/persistence/factory_store.py index c793791..12bcbdf 100644 --- a/src/fxhnt/adapters/persistence/factory_store.py +++ b/src/fxhnt/adapters/persistence/factory_store.py @@ -7,7 +7,7 @@ from sqlalchemy import create_engine, desc, select from sqlalchemy.orm import Session from sqlalchemy.pool import StaticPool -from fxhnt.adapters.persistence.cockpit_models import CockpitBase, FactoryAllocationRow, FactoryCycleRow +from fxhnt.adapters.persistence.cockpit_models import CockpitBase, FactoryAllocationRow, FactoryCycleRow, FactoryFlagRow class FactoryStore: @@ -39,3 +39,16 @@ class FactoryStore: def allocation(self, cycle_id: str) -> list[FactoryAllocationRow]: with Session(self._engine) as s: return list(s.scalars(select(FactoryAllocationRow).where(FactoryAllocationRow.cycle_id == cycle_id))) + + # Fix 6: persisted kill-switch + def set_kill(self, on: bool) -> None: + """Persist the kill flag (key="kill", value="1"/"0") so it survives pod restarts.""" + with Session(self._engine) as s: + s.merge(FactoryFlagRow(key="kill", value="1" if on else "0")) + s.commit() + + def is_killed(self) -> bool: + """Return True if the persisted kill flag is set. Defaults to False when no row exists.""" + with Session(self._engine) as s: + row = s.scalars(select(FactoryFlagRow).where(FactoryFlagRow.key == "kill")).first() + return row is not None and row.value == "1" diff --git a/src/fxhnt/application/factory/factory_loop.py b/src/fxhnt/application/factory/factory_loop.py index f912d45..180c2f0 100644 --- a/src/fxhnt/application/factory/factory_loop.py +++ b/src/fxhnt/application/factory/factory_loop.py @@ -1,7 +1,11 @@ """The factory's promotion + allocation step. Forward-survivors are admitted to the deployed book only if they pass the diversification gate (left-tail corr + marginal Sharpe) vs the CURRENT deployed book, bounded by governance; then the whole deployed book is re-allocated by HRP. Pure orchestration over an injected -store (duck-typed: deployed()/forward()/set_status()/returns()) so it tests offline.""" +store (duck-typed: deployed()/forward()/set_status()/returns()) so it tests offline. + +Admission is SEQUENTIAL: each candidate is re-checked against the running blend of (HRP-weighted book + +already-admitted-this-cycle) so two candidates uncorrelated to the book but correlated to each other do not +both get in.""" from __future__ import annotations from dataclasses import dataclass, field @@ -28,6 +32,9 @@ class LoopConfig: class LoopResult: promoted: list[str] = field(default_factory=list) weights: dict[str, float] = field(default_factory=dict) + deployed_before: int = 0 + deployed_after: int = 0 + forward_count: int = 0 class FactoryLoop: @@ -35,42 +42,78 @@ class FactoryLoop: self._store = store self._cfg = cfg - def _book_series(self, deployed: list) -> np.ndarray: - """Equal-weight blend of deployed sleeve returns = the current book series (for the gate).""" - series = [self._store.returns(r["id"]) for r in deployed] - if not series: + def _weighted_blend(self, rets: dict, weights: dict) -> np.ndarray: + """HRP-weighted blend of return series; trims to common tail length.""" + items = [(rets[k], weights.get(k, 0.0)) for k in rets if len(rets[k]) > 0] + if not items: return np.array([]) - n = min(len(s) for s in series) - return np.mean(np.array([s[-n:] for s in series]), axis=0) + n = min(len(r) for r, _ in items) + wsum = sum(w for _, w in items) or 1.0 + return np.sum([(w / wsum) * r[-n:] for r, w in items], axis=0) def allocate_and_promote(self) -> LoopResult: cfg = self._cfg res = LoopResult() deployed = self._store.deployed() + res.deployed_before = len(deployed) budget = promote_budget(kill_switch=cfg.kill_switch, deployed=len(deployed), max_deployed=cfg.max_deployed, max_new_per_cycle=cfg.max_new_per_cycle) - book = self._book_series(deployed) - # rank forward-survivors by marginal contribution; admit the best uncorrelated ones up to budget - candidates = [] + + # Fix 2+4: start-of-cycle book proxy = HRP-weighted blend of deployed sleeves with enough history + dep_rets = {r["id"]: self._store.returns(r["id"]) for r in deployed + if len(self._store.returns(r["id"])) >= cfg.min_forward_days} + dep_w = hrp_weights(dep_rets) if dep_rets else {} + book = self._weighted_blend(dep_rets, dep_w) + + # Rank eligible survivors by marginal vs the start book + survivors = [] for r in self._store.forward(): - if len(r["returns"]) < cfg.min_forward_days: - continue sleeve = self._store.returns(r["id"]) - if book.size == 0: # empty book: first sleeve admits on data alone - candidates.append((r["id"], 1.0)) + if len(sleeve) < cfg.min_forward_days: + continue + res.forward_count += 1 + if book.size == 0: + # Empty book: founding admit — no correlation to compare against yet + survivors.append((r["id"], 1.0, sleeve)) continue v = diversification_gate(sleeve, book, max_tail_corr=cfg.max_tail_corr, min_marginal=cfg.min_marginal, tail_frac=cfg.tail_frac) if v.admit: - candidates.append((r["id"], v.marginal)) - candidates.sort(key=lambda t: t[1], reverse=True) - for sid, _ in candidates[:budget]: - self._store.set_status(sid, "DEPLOYED") - res.promoted.append(sid) - # HRP re-allocation over the (updated) deployed book + survivors.append((r["id"], v.marginal, sleeve)) + + # Sort descending by marginal so the highest-value uncorrelated candidate goes first + survivors.sort(key=lambda t: t[1], reverse=True) + + # Fix 2: sequential re-check against the running mean of ALREADY-ADMITTED-THIS-CYCLE candidates. + # Comparing only against admitted-this-cycle (not the book) prevents two mutually-correlated + # candidates that each pass the initial book gate from both slipping through in one cycle. + # The first candidate is admitted freely (book gate was already cleared); subsequent ones must + # pass the gate vs the equal-weight mean of previously-admitted sleeves. + admitted_this_cycle: list[np.ndarray] = [] + for sid, _, sleeve in survivors: + if len(res.promoted) >= budget: + break + if not admitted_this_cycle: + # First admit of this cycle: book correlation was already cleared above; admit freely. + ok = True + else: + n = min(len(s) for s in admitted_this_cycle) + running = np.mean([s[-n:] for s in admitted_this_cycle], axis=0) + n2 = min(len(running), len(sleeve)) + v2 = diversification_gate(sleeve[-n2:], running[-n2:], max_tail_corr=cfg.max_tail_corr, + min_marginal=cfg.min_marginal, tail_frac=cfg.tail_frac) + ok = v2.admit + if ok: + self._store.set_status(sid, "DEPLOYED") + res.promoted.append(sid) + admitted_this_cycle.append(sleeve) + + # Fix 3: HRP inclusion requires >= min_forward_days (not >= 2); fetch returns once per sleeve deployed_now = self._store.deployed() - rets = {r["id"]: self._store.returns(r["id"]) for r in deployed_now - if len(self._store.returns(r["id"])) >= 2} + res.deployed_after = len(deployed_now) + # Fix 7: fetch each sleeve's returns ONCE to avoid double I/O + all_rets = {r["id"]: self._store.returns(r["id"]) for r in deployed_now} + rets = {sid: arr for sid, arr in all_rets.items() if len(arr) >= cfg.min_forward_days} res.weights = hrp_weights(rets) if rets else {} return res @@ -89,7 +132,9 @@ class StrategyStoreLoopAdapter: return [{"id": r.strategy_id, "returns": r.forward_returns} for r in self._store.list(status=self._S.FORWARD)] - def set_status(self, sid, status): # noqa: ARG002 — loop only ever promotes to DEPLOYED + def set_status(self, sid, status): + # Fix 5: assert the only valid transition so a future demotion path cannot silently promote + assert status == "DEPLOYED", f"loop only promotes to DEPLOYED, got {status}" self._store.set_status(sid, self._S.DEPLOYED) def returns(self, sid): diff --git a/src/fxhnt/cli.py b/src/fxhnt/cli.py index dea3035..09ca03c 100644 --- a/src/fxhnt/cli.py +++ b/src/fxhnt/cli.py @@ -485,24 +485,60 @@ def factory_cycle(execute: bool = typer.Option(False, "--execute/--dry-run")) -> from fxhnt.application.factory.factory_loop import FactoryLoop, LoopConfig, StrategyStoreLoopAdapter s = get_settings(); f = s.factory store = SqlStrategyStore(s.operational_dsn) + fs = FactoryStore(s.operational_dsn) + # Fix 6: effective kill = env flag OR persisted DB flag (OR dry-run mode) + effective_kill = f.kill_switch or fs.is_killed() or not execute cfg = LoopConfig(max_tail_corr=f.max_tail_corr, min_marginal=f.min_marginal_sharpe, tail_frac=f.tail_frac, min_forward_days=f.min_forward_days, max_deployed=f.max_deployed_sleeves, - max_new_per_cycle=f.max_new_deploys_per_cycle, kill_switch=f.kill_switch or not execute) + max_new_per_cycle=f.max_new_deploys_per_cycle, kill_switch=effective_kill) loop = FactoryLoop(StrategyStoreLoopAdapter(store), cfg) + now = dt.datetime.now(dt.UTC).replace(tzinfo=None) res = loop.allocate_and_promote() - fs = FactoryStore(s.operational_dsn) cyc = "fc" + dt.datetime.now(dt.UTC).strftime("%Y%m%d") if execute: - fs.record_allocation(cyc, res.weights, dt.datetime.now(dt.UTC).replace(tzinfo=None)) - typer.echo(f"[{cyc}] {'EXECUTED' if execute else 'DRY-RUN'} promoted={res.promoted} " + # Fix 1: populate the funnel table on every execute cycle + fs.record_cycle(cycle_id=cyc, proposed=res.forward_count, tested=res.forward_count, + survived=len(res.promoted), forward=res.forward_count, deployed=res.deployed_after, + retired=0, global_trials=store.total_trials(), at=now) + fs.record_allocation(cyc, res.weights, now) + typer.echo(f"[{cyc}] {'EXECUTED' if execute else 'DRY-RUN'} " + f"forward_pool={res.forward_count} promoted={res.promoted} " + f"deployed_before={res.deployed_before} deployed_after={res.deployed_after} " f"deployed_weights={ {k: round(v, 3) for k, v in res.weights.items()} }") -@app.command("factory-flatten") -def factory_flatten() -> None: - """Kill-switch: set FXHNT_FACTORY_KILL_SWITCH=true to halt promotions (this just reports current state).""" - typer.echo(f"factory kill_switch = {get_settings().factory.kill_switch}; " - f"set FXHNT_FACTORY_KILL_SWITCH=true to halt promotions") +@app.command("factory-halt") +def factory_halt() -> None: + """Kill-switch: persist halt flag to DB so all subsequent factory-cycle runs block promotions immediately.""" + from fxhnt.adapters.persistence.factory_store import FactoryStore + fs = FactoryStore(get_settings().operational_dsn) + fs.set_kill(True) + typer.echo("factory halted (persisted kill=1). Run `fxhnt factory-resume` to re-enable.") + + +@app.command("factory-resume") +def factory_resume() -> None: + """Clear the persisted kill flag so factory-cycle promotions are re-enabled.""" + from fxhnt.adapters.persistence.factory_store import FactoryStore + fs = FactoryStore(get_settings().operational_dsn) + fs.set_kill(False) + typer.echo("factory resumed (kill=0). Promotions enabled on next factory-cycle --execute.") + + +@app.command("factory-status") +def factory_status() -> None: + """Print the persisted kill state and the latest cycle summary.""" + from fxhnt.adapters.persistence.factory_store import FactoryStore + fs = FactoryStore(get_settings().operational_dsn) + killed = fs.is_killed() + env_kill = get_settings().factory.kill_switch + typer.echo(f"kill_switch: db={killed} env={env_kill} effective={killed or env_kill}") + cyc = fs.latest_cycle() + if cyc: + typer.echo(f"latest cycle: {cyc.cycle_id} proposed={cyc.proposed} forward={cyc.forward} " + f"survived={cyc.survived} deployed={cyc.deployed} at={cyc.at}") + else: + typer.echo("latest cycle: (none yet)") if __name__ == "__main__": diff --git a/tests/integration/test_factory_loop.py b/tests/integration/test_factory_loop.py index bb51e45..6f2dcc1 100644 --- a/tests/integration/test_factory_loop.py +++ b/tests/integration/test_factory_loop.py @@ -54,3 +54,60 @@ def test_kill_switch_blocks_promotion() -> None: loop = FactoryLoop(store, _cfg(kill_switch=True)) res = loop.allocate_and_promote() assert res.promoted == [] and store._r["CAND"]["status"] == "FORWARD" + + +# Fix 8 — new tests ----------------------------------------------------------- + +def test_below_min_forward_days_not_promoted() -> None: + """A forward survivor with fewer than min_forward_days returns must be silently skipped.""" + rng = np.random.default_rng(42) + book = list(rng.normal(0.0005, 0.01, 300)) + # Only 10 days of history — below the min_forward_days=20 threshold + short = list(rng.normal(0.0005, 0.01, 10)) + store = FakeStore([_rec("DEP", "DEPLOYED", book), _rec("SHORT", "FORWARD", short)]) + loop = FactoryLoop(store, _cfg(min_forward_days=20)) + res = loop.allocate_and_promote() + assert "SHORT" not in res.promoted + assert store._r["SHORT"]["status"] == "FORWARD" + assert res.forward_count == 0 # it never reached the gate + + +def test_empty_book_admits_first_then_gates_rest() -> None: + """Empty deployed book: first uncorrelated survivor gets a founding admit, but a duplicate of it + must be blocked by the running-blend re-gate on the second candidate.""" + rng = np.random.default_rng(7) + a_returns = list(rng.normal(0.001, 0.01, 100)) + b_returns = list(a_returns) # identical to A → corr ≈ 1.0 → must be blocked after A is admitted + store = FakeStore([ + _rec("A", "FORWARD", a_returns), + _rec("B", "FORWARD", b_returns), + ]) + loop = FactoryLoop(store, _cfg(max_new_per_cycle=2)) + res = loop.allocate_and_promote() + # Exactly one should be admitted; B is a duplicate of A so it must be blocked + assert len(res.promoted) == 1 + promoted_id = res.promoted[0] + other_id = "B" if promoted_id == "A" else "A" + assert store._r[promoted_id]["status"] == "DEPLOYED" + assert store._r[other_id]["status"] == "FORWARD" + + +def test_mutually_correlated_same_cycle_admits_blocked() -> None: + """Two forward survivors uncorrelated to the book but nearly identical to each other: + only ONE should be admitted per cycle even though max_new_per_cycle=2.""" + rng = np.random.default_rng(99) + book = list(rng.normal(0.0005, 0.01, 300)) + # Candidates are independent of the book but almost perfectly correlated with each other + base = rng.normal(0.001, 0.01, 300) + noise = rng.normal(0, 1e-6, 300) + cand_a = list(base) + cand_b = list(base + noise) # corr(cand_a, cand_b) ≈ 1.0 + store = FakeStore([ + _rec("DEP", "DEPLOYED", book), + _rec("CA", "FORWARD", cand_a), + _rec("CB", "FORWARD", cand_b), + ]) + loop = FactoryLoop(store, _cfg(max_new_per_cycle=2)) + res = loop.allocate_and_promote() + # Both pass the gate vs the book, but the second must fail the running-blend re-gate + assert len(res.promoted) == 1 diff --git a/tests/integration/test_factory_store.py b/tests/integration/test_factory_store.py index 5202190..777b654 100644 --- a/tests/integration/test_factory_store.py +++ b/tests/integration/test_factory_store.py @@ -16,3 +16,13 @@ def test_funnel_and_allocation_roundtrip() -> None: assert funnel.proposed == 12 and funnel.deployed == 2 and funnel.global_trials == 5000 alloc = {a.strategy_id: a.weight for a in st.allocation("c1")} assert abs(alloc["A"] - 0.6) < 1e-9 and abs(alloc["B"] - 0.4) < 1e-9 + + +def test_kill_switch_persists() -> None: + """set_kill(True) → is_killed() True; set_kill(False) → is_killed() False.""" + st = FactoryStore("sqlite://") + assert st.is_killed() is False # default: no row → False + st.set_kill(True) + assert st.is_killed() is True + st.set_kill(False) + assert st.is_killed() is False