experiment-os: first-cohort registrations (5 experiments) (+test)

Plan 006 step 4. experiments/*.json for time_stop_10m_v1, probe_add_v1,
asym_exit_v1, rank_one_bullet_v1, pm1_v1 — bars verbatim from the plan table,
production population, fidelity floors {0.25, 3, 1} from the research doc. Also
hardened evaluate(): n==0 and any uncomputable bar metric -> INCONCLUSIVE (not a
false FAIL). Cohort registers + survives run->adjudicate end-to-end. 25 tests green.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

experiments/asym_exit_v1.json

@@ -0,0 +1,9 @@
+{
+  "id": "asym_exit_v1",
+  "hypothesis": "An asymmetric exit (volatility-fractional stop with a spread floor, no profit target, trail to EOD) captures the published fat-right-tail edge better than the current managed exit.",
+  "population": {"min_entry_gap_pct": 0.10, "min_day_volume": 300000, "price_min": 2.0, "price_max": 20.0},
+  "intervention": {"kind": "asym_exit", "atr_frac": 0.25, "no_target": true, "trail": "eod"},
+  "bars": {"expR_delta_min": 0.10, "min_n": 300},
+  "futility": {"min_n": 500, "expR_delta_max": -0.02},
+  "fidelity": {"atr_frac": 0.25, "spread_mult": 3, "adverse_ticks": 1}
+}

experiments/pm1_v1.json

@@ -0,0 +1,9 @@
+{
+  "id": "pm1_v1",
+  "hypothesis": "The premarket-setup (pm1) detector's continuation clears its pre-registered promotion bar (median +10% upside, drawdown better than -6%, +1R rate > 50% overall and on the >=50%-gap subset).",
+  "population": {},
+  "intervention": {"kind": "pm1_bridge"},
+  "bars": {"median_up_60m_min": 0.10, "median_dd_60m_min": -0.06, "rate_1r_min": 0.50, "big_gap_rate_1r_min": 0.50},
+  "futility": null,
+  "fidelity": {"atr_frac": 0.25, "spread_mult": 3, "adverse_ticks": 1}
+}

experiments/probe_add_v1.json

@@ -0,0 +1,9 @@
+{
+  "id": "probe_add_v1",
+  "hypothesis": "Entering half-size at the fill and adding the other half only on a confirmed new high above the trigger bar (within 10 min) improves blended R over full-size-at-fill.",
+  "population": {"min_entry_gap_pct": 0.10, "min_day_volume": 300000, "price_min": 2.0, "price_max": 20.0},
+  "intervention": {"kind": "probe_add", "probe_frac": 0.5, "confirm_minutes": 10},
+  "bars": {"expR_delta_min": 0.05, "min_n": 300},
+  "futility": {"min_n": 500, "expR_delta_max": -0.02},
+  "fidelity": {"atr_frac": 0.25, "spread_mult": 3, "adverse_ticks": 1}
+}

experiments/rank_one_bullet_v1.json

@@ -0,0 +1,9 @@
+{
+  "id": "rank_one_bullet_v1",
+  "hypothesis": "Concentrating each session on only the single top-ranked fire beats spreading risk across every fire that day.",
+  "population": {"min_entry_gap_pct": 0.10, "min_day_volume": 300000, "price_min": 2.0, "price_max": 20.0},
+  "intervention": {"kind": "rank_one_bullet", "top_n": 1},
+  "bars": {"expR_delta_min": 0.10, "min_n": 100},
+  "futility": {"min_n": 200, "expR_delta_max": -0.05},
+  "fidelity": {"atr_frac": 0.25, "spread_mult": 3, "adverse_ticks": 1}
+}

experiments/time_stop_10m_v1.json

@@ -0,0 +1,9 @@
+{
+  "id": "time_stop_10m_v1",
+  "hypothesis": "A 10-minute time-stop that scratches unconfirmed entries (no new post-entry high and < +0.5R) beats letting them run to the full opening-range stop.",
+  "population": {"min_entry_gap_pct": 0.10, "min_day_volume": 300000, "price_min": 2.0, "price_max": 20.0},
+  "intervention": {"kind": "time_stop", "minutes": 10, "confirm": "new_high_or_0.5R"},
+  "bars": {"expR_delta_min": 0.05, "min_n": 300},
+  "futility": {"min_n": 500, "expR_delta_max": -0.02},
+  "fidelity": {"atr_frac": 0.25, "spread_mult": 3, "adverse_ticks": 1}
+}

research/experiments/adjudicate.py

@@ -93,11 +93,18 @@ def evaluate(runs: list[dict], bars: dict, futility: dict | None):
 
     if futility and n >= int(futility.get("min_n", 1 << 60)):
         conds = [c for c in futility if c != "min_n"]
+        # _meets is False on a None metric, so futility never fires on no data.
         if conds and all(_meets(c, futility[c], metrics) for c in conds):
             return "KILLED", metrics, n
 
+    if n == 0:
+        return "INCONCLUSIVE", metrics, n
     if n >= int(bars.get("min_n", 0)):
         bar_conds = [c for c in bars if c != "min_n"]
+        # a bar whose metric is uncomputable (None) is not a DECISIVE failure —
+        # can't conclude, so INCONCLUSIVE rather than a false FAIL.
+        if any(metrics.get(_metric_name(c)) is None for c in bar_conds):
+            return "INCONCLUSIVE", metrics, n
         if bar_conds and all(_meets(c, bars[c], metrics) for c in bar_conds):
             return "PASS", metrics, n
         return "FAIL", metrics, n

tests/unit/experiments/test_cohort.py

@@ -0,0 +1,42 @@
+"""First cohort: all five experiments/*.json register and survive a full
+run -> adjudicate cycle end-to-end on an isolated store (substitutes the plan's
+'run/adjudicate against a dev DB' verify; the real verdicts land when the owner
+first schedules the harness against production)."""
+
+import glob
+import json
+import os
+
+from research.experiments import adjudicate, registry, replay
+
+REPO = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", ".."))
+COHORT = os.path.join(REPO, "experiments")
+
+EXPECTED = {"time_stop_10m_v1", "probe_add_v1", "asym_exit_v1",
+            "rank_one_bullet_v1", "pm1_v1"}
+
+
+def test_five_json_files_present_and_valid():
+    files = glob.glob(os.path.join(COHORT, "*.json"))
+    ids = {json.load(open(f))["id"] for f in files}
+    assert EXPECTED <= ids, f"missing cohort files: {EXPECTED - ids}"
+    # every cohort file carries fidelity floors (the review bar)
+    for f in files:
+        spec = json.load(open(f))
+        if spec["id"] in EXPECTED:
+            for k in ("atr_frac", "spread_mult", "adverse_ticks"):
+                assert k in spec["fidelity"], f"{spec['id']} missing fidelity.{k}"
+
+
+def test_cohort_registers_and_adjudicates(store):
+    for exp_id in EXPECTED:
+        registry.register(store.con, os.path.join(COHORT, f"{exp_id}.json"))
+    assert {e["id"] for e in registry.list_experiments(store.con)} == EXPECTED
+
+    # empty history -> run inserts nothing; adjudicate yields INCONCLUSIVE for all
+    # (pm1 has no labeled setups seeded here) without raising.
+    replay.run(store)
+    out = adjudicate.adjudicate(store)
+    verdicts = {o["experiment_id"]: o["verdict"] for o in out}
+    assert set(verdicts) == EXPECTED
+    assert all(v == "INCONCLUSIVE" for v in verdicts.values()), verdicts