# Declarative Strategy-Level Allocation Policy + Capital Ceiling (Sub-project B2a) Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** A versioned in-code `AllocationPolicy` + a pure `compute_allocation`/`target_capital` that turn the deploy-tier strategies into ex-ante-sized per-strategy target weights (equal-weight the marginal-Sharpe-gate survivors; ex-ante vol-target leverage from the FULL-history vol, never trailing; a rare latching aggregate killswitch) capped by a hard human `capital_ceiling`, recorded + policy-version-stamped.

**Architecture:** `allocation_policy.py` holds the versioned policy + content-hash (mirrors B1's gate_policy). `allocation_engine.py` composes pure helpers (full-history vol, equal-weight marginal-Sharpe prune, killswitch latch check, reusing B1's `promotion_corr.deploy_book_returns` for the equal-weight aggregate) into `compute_allocation` → per-strategy weights and `target_capital` → per-strategy dollars capped at the ceiling. A nightly step records the snapshot to `strategy_allocation`. No capital is moved (that is sub-project C).

**Tech Stack:** Python 3.12, numpy, dataclasses, SQLAlchemy 2.0 ORM (sqlite tests / Postgres prod), pytest (`uv run pytest`), ruff (E,F,I,UP,B,SIM; line-length 120).

## Global Constraints

- Spec: `docs/superpowers/specs/2026-07-09-declarative-allocation-policy-b2a-design.md` (authoritative).
- Builds on A (merged: `forward_nav` records via `ForwardNavRepo.nav_history`) + B1 (merged: deploy-tier set = registry `tier=="deploy"` ∪ `promoted_strategy_ids()`; `promotion_corr.deploy_book_returns` for the equal-weight per-date aggregate).
- **NO reactive de-sizing** (the load-bearing rule): the ex-ante leverage `K` derives from the FULL-history vol (stable), NEVER a trailing window. The only drawdown-reactive element is the rare, latching aggregate killswitch. Do NOT reuse `book_allocator.combine_book` (its `kelly · target_vol / trailing_port_vol` is the rejected reactive path).
- REAL MONEY: `capital_ceiling` is a HARD cap — total deployed gross never exceeds it; err safe (exclude a short-history strategy, deploy nothing when no eligible strategies, flatten on killswitch). No capital is MOVED by B2a (targets only).
- Policy lives in git-committed code, never runtime/DB. A change = commit + `ALLOCATION_POLICY_VERSION` bump; the content-derived `allocation_policy_hash` self-check forces the bump.
- Determinism: `compute_allocation`/`target_capital` are pure — no wall-clock, no I/O; recomputable from (policy + records).
- Idempotency: DB writes are `Session.merge` upserts. Never save tests to the repo root; tests under `tests/unit/` or `tests/integration/`. Commit after every task; messages end with the `Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>` trailer.

---

### Task 1: `allocation_policy.py` — the versioned allocation policy + hash

**Files:**
- Create: `src/fxhnt/application/allocation_policy.py`
- Test: `tests/unit/test_allocation_policy.py`

**Interfaces:**
- Produces: `AllocationPolicy` (frozen dataclass, fields below), `ALLOCATION_POLICY`, `ALLOCATION_POLICY_VERSION: int = 1`, `allocation_policy_hash(policy, version) -> str`.

- [ ] **Step 1: Write the failing test**

```python
# tests/unit/test_allocation_policy.py
from fxhnt.application.allocation_policy import (
    ALLOCATION_POLICY, ALLOCATION_POLICY_VERSION, AllocationPolicy, allocation_policy_hash)


def test_policy_defaults():
    p = ALLOCATION_POLICY
    assert (p.target_vol, p.kelly_fraction, p.max_leverage, p.max_strategy_weight) == (0.12, 0.5, 1.5, 0.5)
    assert (p.marginal_gate, p.min_history_days) == (True, 60)
    assert (p.kill_dd, p.reenter_dd) == (0.25, 0.10)
    assert p.capital_ceiling == 35_000.0


def test_hash_content_derived_and_pinned():
    assert allocation_policy_hash(ALLOCATION_POLICY, ALLOCATION_POLICY_VERSION) == \
        allocation_policy_hash(ALLOCATION_POLICY, ALLOCATION_POLICY_VERSION)                       # stable
    assert allocation_policy_hash(AllocationPolicy(target_vol=0.10), ALLOCATION_POLICY_VERSION) != \
        allocation_policy_hash(ALLOCATION_POLICY, ALLOCATION_POLICY_VERSION)                       # field change
    assert allocation_policy_hash(ALLOCATION_POLICY, 2) != allocation_policy_hash(ALLOCATION_POLICY, 1)
    assert allocation_policy_hash(ALLOCATION_POLICY, ALLOCATION_POLICY_VERSION) == _EXPECTED_HASH   # noqa: F821
```

- [ ] **Step 2: Run test to verify it fails**

Run: `uv run pytest tests/unit/test_allocation_policy.py -q`
Expected: FAIL with `ModuleNotFoundError: No module named 'fxhnt.application.allocation_policy'`.

- [ ] **Step 3: Implement `allocation_policy.py`**

```python
# src/fxhnt/application/allocation_policy.py
"""The declarative, versioned strategy-level ALLOCATION policy (sub-project B2a).

Turns the deploy-tier strategies into ex-ante-sized target weights. Single source for the allocation knobs
(previously scattered as book_allocator function defaults + config settings). A change = a code commit + an
ALLOCATION_POLICY_VERSION bump; the content-derived allocation_policy_hash self-check forces the bump (same
discipline as A's definition_hash and B1's policy_hash). The capital_ceiling — the HARD human gate on total
deployed gross — lives in this versioned policy, so a ceiling change is audited. NO reactive de-sizing: the
sizing derives from full-history vol, and the only drawdown-reactive element is the rare aggregate killswitch."""
from __future__ import annotations

import hashlib
import json
from dataclasses import asdict, dataclass


@dataclass(frozen=True, slots=True)
class AllocationPolicy:
    target_vol: float = 0.12          # ex-ante annualised vol target for the book
    kelly_fraction: float = 0.5       # fractional-Kelly scalar
    max_leverage: float = 1.5         # hard cap on the ex-ante leverage K
    max_strategy_weight: float = 0.5  # per-strategy weight cap (before leverage)
    marginal_gate: bool = True        # leave-one-out equal-weight marginal-Sharpe quality filter
    min_history_days: int = 60        # a strategy needs >= this many return obs for a stable ex-ante vol
    kill_dd: float = 0.25             # RARE aggregate killswitch: flatten beyond this drawdown (latching)
    reenter_dd: float = 0.10          # re-enter once the drawdown recovers inside this
    capital_ceiling: float = 35_000.0  # HARD fund-level dollar cap on total deployed gross (the human gate)


ALLOCATION_POLICY = AllocationPolicy()
ALLOCATION_POLICY_VERSION = 1


def allocation_policy_hash(policy: AllocationPolicy, version: int) -> str:
    canonical = json.dumps(asdict(policy), sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(f"{canonical}|v{version}".encode()).hexdigest()
```

- [ ] **Step 4: Pin the hash + run**

Run: `uv run python -c "from fxhnt.application.allocation_policy import ALLOCATION_POLICY, ALLOCATION_POLICY_VERSION, allocation_policy_hash; print(allocation_policy_hash(ALLOCATION_POLICY, ALLOCATION_POLICY_VERSION))"`
Paste the value as `_EXPECTED_HASH = "<value>"` near the top of the test file.
Run: `uv run pytest tests/unit/test_allocation_policy.py -q` → Expected: PASS (2 passed).

- [ ] **Step 5: Commit**

```bash
git add src/fxhnt/application/allocation_policy.py tests/unit/test_allocation_policy.py
git commit -F - <<'EOF'
feat(alloc): versioned AllocationPolicy + content-derived hash + capital_ceiling (B2a)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
EOF
```

---

### Task 2: allocation engine — pure helpers (full-history vol, equal-weight marginal prune, killswitch latch)

**Files:**
- Create: `src/fxhnt/application/allocation_engine.py`
- Test: `tests/unit/test_allocation_engine_helpers.py`

**Interfaces:**
- Consumes: `promotion_corr.deploy_book_returns` (B1, the equal-weight per-date aggregate), `per_period_sharpe` (`fxhnt.domain.gauntlet`).
- Produces:
  - `full_history_vol(returns: dict[str, float]) -> float` — annualised (√365) std of the return values over the WHOLE series; `0.0` if `< 2` obs.
  - `equal_weight_marginal_prune(returns_by_strategy: dict[str, dict[str, float]]) -> set[str]` — leave-one-out: bench a strategy only when removing it raises the EQUAL-WEIGHT book's per-period Sharpe; shrinks to ≥1.
  - `killswitch_active(book_returns: dict[str, float], kill_dd: float, reenter_dd: float) -> bool` — the final latch state after folding the drawdown killswitch over the sorted series.

- [ ] **Step 1: Write the failing test**

```python
# tests/unit/test_allocation_engine_helpers.py
import numpy as np

from fxhnt.application.allocation_engine import (
    equal_weight_marginal_prune, full_history_vol, killswitch_active)


def test_full_history_vol_is_annualised_std():
    r = {f"2026-06-{d:02d}": v for d, v in zip(range(1, 6), [0.01, -0.01, 0.02, -0.02, 0.0])}
    assert abs(full_history_vol(r) - float(np.std([0.01, -0.01, 0.02, -0.02, 0.0]) * np.sqrt(365.0))) < 1e-12
    assert full_history_vol({"2026-06-01": 0.01}) == 0.0        # < 2 obs → 0


def test_marginal_prune_benches_non_additive_strategy():
    dates = [f"2026-06-{d:02d}" for d in range(1, 41)]
    good = {d: (0.01 if i % 2 else -0.008) for i, d in enumerate(dates)}     # positive-Sharpe
    noise = {d: (0.03 if i % 3 == 0 else -0.02) for i, d in enumerate(dates)}  # drags the book
    kept = equal_weight_marginal_prune({"good": good, "noise": noise})
    assert "good" in kept and "noise" not in kept


def test_killswitch_latches_on_deep_dd_and_reenters():
    # a −30% run then recovery; kill at 0.25, reenter at 0.10
    down = {f"2026-06-{d:02d}": -0.04 for d in range(1, 11)}     # ~ -33% cumulative → breach 0.25
    assert killswitch_active(down, kill_dd=0.25, reenter_dd=0.10) is True
    calm = {f"2026-06-{d:02d}": 0.001 for d in range(1, 11)}
    assert killswitch_active(calm, kill_dd=0.25, reenter_dd=0.10) is False
```

- [ ] **Step 2: Run test to verify it fails**

Run: `uv run pytest tests/unit/test_allocation_engine_helpers.py -q`
Expected: FAIL with `ModuleNotFoundError: No module named 'fxhnt.application.allocation_engine'`.

- [ ] **Step 3: Implement the helpers**

```python
# src/fxhnt/application/allocation_engine.py
"""The pure, ex-ante strategy-level allocation engine (sub-project B2a).

compute_allocation turns each live strategy's return series (from A's record) into ex-ante-sized target
weights: equal-weight the marginal-Sharpe-gate survivors, size to a target vol using the FULL-history vol
(stable — NOT trailing, so a recent vol spike does not change the leverage), flatten on a rare latching
aggregate drawdown killswitch. target_capital caps total deployed gross at the policy's hard capital_ceiling.
No reactive de-sizing; the killswitch is the only drawdown-reactive element."""
from __future__ import annotations

import numpy as np

from fxhnt.application.allocation_policy import AllocationPolicy
from fxhnt.application.promotion_corr import deploy_book_returns
from fxhnt.domain.gauntlet import per_period_sharpe

_ANN = np.sqrt(365.0)


def full_history_vol(returns: dict[str, float]) -> float:
    vals = list(returns.values())
    if len(vals) < 2:
        return 0.0
    return float(np.std(np.array(vals)) * _ANN)


def _book_sharpe(returns_by_strategy: dict[str, dict[str, float]]) -> float:
    book = deploy_book_returns(returns_by_strategy)   # equal-weight per-date average
    if len(book) < 2:
        return 0.0
    return float(per_period_sharpe(np.array([book[d] for d in sorted(book)])))


def equal_weight_marginal_prune(returns_by_strategy: dict[str, dict[str, float]]) -> set[str]:
    kept = set(returns_by_strategy)
    while len(kept) > 1:
        base = _book_sharpe({e: returns_by_strategy[e] for e in kept})
        drop = next((e for e in kept
                     if _book_sharpe({x: returns_by_strategy[x] for x in kept - {e}}) > base + 1e-9), None)
        if drop is None:
            break
        kept.discard(drop)
    return kept


def killswitch_active(book_returns: dict[str, float], kill_dd: float, reenter_dd: float) -> bool:
    killed = False
    shadow_eq = 1.0
    peak = 1.0
    for d in sorted(book_returns):
        shadow_eq *= (1.0 + book_returns[d])
        peak = max(peak, shadow_eq)
        dd = (peak - shadow_eq) / peak if peak > 0 else 0.0
        if not killed and dd > kill_dd:
            killed = True
        elif killed and dd < reenter_dd:
            killed = False
    return killed
```

- [ ] **Step 4: Run tests → PASS. Ruff.**

Run: `uv run pytest tests/unit/test_allocation_engine_helpers.py -q` → PASS (3 passed).
Run: `uv run ruff check src/fxhnt/application/allocation_engine.py tests/unit/test_allocation_engine_helpers.py`

- [ ] **Step 5: Commit**

```bash
git add src/fxhnt/application/allocation_engine.py tests/unit/test_allocation_engine_helpers.py
git commit -F - <<'EOF'
feat(alloc): pure allocation helpers — full-history vol, equal-weight marginal prune, killswitch latch (B2a)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
EOF
```

---

### Task 3: `compute_allocation` — the ex-ante composition (+ the anti-reactive invariant)

**Files:**
- Modify: `src/fxhnt/application/allocation_engine.py` (add `compute_allocation`)
- Test: `tests/unit/test_compute_allocation.py`

**Interfaces:**
- Consumes: the Task-2 helpers, `AllocationPolicy` (Task 1), `deploy_book_returns`.
- Produces: `compute_allocation(returns_by_strategy: dict[str, dict[str, float]], policy: AllocationPolicy) -> dict[str, float]` — per-strategy target weights (ex-ante-sized, gated, killswitch-aware).

- [ ] **Step 1: Write the failing test**

```python
# tests/unit/test_compute_allocation.py
from fxhnt.application.allocation_policy import AllocationPolicy
from fxhnt.application.allocation_engine import compute_allocation


def _series(n, ret):  # n days of a constant-ish return
    return {f"2026-{(6 + d // 30):02d}-{(d % 30) + 1:02d}": ret for d in range(n)}


def test_excludes_short_history_and_empty_returns_empty():
    pol = AllocationPolicy(min_history_days=60, marginal_gate=False)
    assert compute_allocation({"short": _series(10, 0.001)}, pol) == {}   # < 60 obs → excluded → empty


def test_equal_weight_and_ex_ante_leverage():
    pol = AllocationPolicy(min_history_days=5, marginal_gate=False, max_leverage=1.5,
                           kelly_fraction=0.5, target_vol=0.12, kill_dd=0.99, reenter_dd=0.98)
    import numpy as np
    a = {f"2026-06-{d:02d}": (0.01 if d % 2 else -0.005) for d in range(1, 21)}
    b = {f"2026-06-{d:02d}": (-0.005 if d % 2 else 0.01) for d in range(1, 21)}
    w = compute_allocation({"a": a, "b": b}, pol)
    assert set(w) == {"a", "b"}
    assert abs(w["a"] - w["b"]) < 1e-9                       # equal base weights
    # K applied: weights sum to K (the leverage), each = K/2
    k = sum(w.values())
    assert 0 < k <= 1.5


def test_recent_vol_spike_does_not_change_K_antireactive():
    # THE load-bearing invariant: ex-ante vol uses the FULL history, so appending a recent spike must not
    # change the leverage on the earlier window.
    pol = AllocationPolicy(min_history_days=5, marginal_gate=False, kill_dd=0.99, reenter_dd=0.98)
    base = {f"2026-06-{d:02d}": (0.01 if d % 2 else -0.01) for d in range(1, 61)}
    w_before = compute_allocation({"a": dict(base), "b": dict(base)}, pol)
    spiked = dict(base); spiked["2026-06-30"] = 0.5           # a single huge recent day
    w_after = compute_allocation({"a": spiked, "b": dict(base)}, pol)
    # the full-history vol of "a" changes (it must — vol IS a full-history stat), so K adapts SLOWLY over the
    # whole record, NOT reactively to the last window. Assert K moved only via the full-history vol, i.e. the
    # book vol recomputed over ALL 60 days (not a trailing subset):
    from fxhnt.application.allocation_engine import full_history_vol
    from fxhnt.application.promotion_corr import deploy_book_returns
    vol_all = full_history_vol(deploy_book_returns({"a": spiked, "b": dict(base)}))
    k_after = sum(w_after.values())
    assert abs(k_after - min(pol.max_leverage, pol.kelly_fraction * pol.target_vol / vol_all)) < 1e-9


def test_killswitch_flattens():
    pol = AllocationPolicy(min_history_days=5, marginal_gate=False, kill_dd=0.20, reenter_dd=0.10)
    crash = {f"2026-06-{d:02d}": -0.04 for d in range(1, 11)}   # deep DD → killswitch
    w = compute_allocation({"a": crash, "b": crash}, pol)
    assert all(v == 0.0 for v in w.values())                   # flattened
```

- [ ] **Step 2: Run test to verify it fails**

Run: `uv run pytest tests/unit/test_compute_allocation.py -q`
Expected: FAIL with `ImportError: cannot import name 'compute_allocation'`.

- [ ] **Step 3: Implement `compute_allocation`**

Append to `src/fxhnt/application/allocation_engine.py`:

```python
def compute_allocation(returns_by_strategy: dict[str, dict[str, float]],
                       policy: AllocationPolicy) -> dict[str, float]:
    # 1. eligibility: a stable ex-ante vol needs enough history
    eligible = {s: r for s, r in returns_by_strategy.items() if len(r) >= policy.min_history_days}
    if not eligible:
        return {}
    # 2. marginal-Sharpe quality gate (equal-weight book), 3. equal-weight survivors
    survivors = equal_weight_marginal_prune(eligible) if policy.marginal_gate else set(eligible)
    n = len(survivors)
    base = {s: 1.0 / n for s in survivors}
    # per-strategy cap + renormalise (inert for pure equal weight, honours the policy guardrail)
    if any(w > policy.max_strategy_weight + 1e-12 for w in base.values()):
        capped = {s: min(w, policy.max_strategy_weight) for s, w in base.items()}
        tot = sum(capped.values())
        base = {s: w / tot for s, w in capped.items()} if tot > 0 else base
    # 5. rare latching aggregate killswitch → flatten
    book = deploy_book_returns({s: eligible[s] for s in survivors})
    if killswitch_active(book, policy.kill_dd, policy.reenter_dd):
        return {s: 0.0 for s in survivors}
    # 4. ex-ante vol-target leverage from the FULL-history book vol (never trailing)
    vol = full_history_vol(book)
    k = min(policy.max_leverage, policy.kelly_fraction * policy.target_vol / vol) if vol > 1e-12 else 0.0
    return {s: k * w for s, w in base.items()}
```

- [ ] **Step 4: Run tests → PASS. Ruff.**

Run: `uv run pytest tests/unit/test_compute_allocation.py -q` → PASS (4 passed).
Run: `uv run ruff check src/fxhnt/application/allocation_engine.py tests/unit/test_compute_allocation.py`

- [ ] **Step 5: Commit**

```bash
git add src/fxhnt/application/allocation_engine.py tests/unit/test_compute_allocation.py
git commit -F - <<'EOF'
feat(alloc): compute_allocation — ex-ante vol-target on full-history vol, marginal gate, killswitch (B2a)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
EOF
```

---

### Task 4: `target_capital` — the hard capital-ceiling scaling

**Files:**
- Modify: `src/fxhnt/application/allocation_engine.py` (add `target_capital`)
- Test: `tests/unit/test_target_capital.py`

**Interfaces:**
- Produces: `target_capital(target_weights: dict[str, float], equity: float, policy: AllocationPolicy) -> dict[str, float]` — per-strategy signed dollars; `sum(abs(dollars)) <= policy.capital_ceiling` always.

- [ ] **Step 1: Write the failing test**

```python
# tests/unit/test_target_capital.py
from fxhnt.application.allocation_policy import AllocationPolicy
from fxhnt.application.allocation_engine import target_capital


def test_ceiling_binds_and_caps_gross():
    pol = AllocationPolicy(capital_ceiling=10_000.0)
    # weights sum to 1.0 (gross), equity 50k → implied gross 50k > ceiling 10k → scaled to 10k
    w = {"a": 0.5, "b": 0.5}
    d = target_capital(w, equity=50_000.0, policy=pol)
    assert abs(sum(abs(x) for x in d.values()) - 10_000.0) < 1e-6      # gross capped at ceiling
    assert abs(d["a"] - d["b"]) < 1e-6                                 # equal split preserved


def test_ceiling_not_binding_uses_equity():
    pol = AllocationPolicy(capital_ceiling=100_000.0)
    d = target_capital({"a": 0.5, "b": 0.5}, equity=20_000.0, policy=pol)
    assert abs(sum(abs(x) for x in d.values()) - 20_000.0) < 1e-6      # gross = equity·1.0 < ceiling


def test_empty_or_zero_weights():
    pol = AllocationPolicy(capital_ceiling=10_000.0)
    assert target_capital({}, 50_000.0, pol) == {}
    assert target_capital({"a": 0.0, "b": 0.0}, 50_000.0, pol) == {"a": 0.0, "b": 0.0}
```

- [ ] **Step 2: Run test to verify it fails**

Run: `uv run pytest tests/unit/test_target_capital.py -q`
Expected: FAIL with `ImportError: cannot import name 'target_capital'`.

- [ ] **Step 3: Implement `target_capital`**

Append to `src/fxhnt/application/allocation_engine.py`:

```python
def target_capital(target_weights: dict[str, float], equity: float,
                   policy: AllocationPolicy) -> dict[str, float]:
    """Per-strategy signed dollars. Total deployed gross = min(equity·Σ|w|, capital_ceiling) — the HARD
    human gate: never deploy more than the ceiling (or more than equity·leverage). Signs preserved."""
    if not target_weights:
        return {}
    gross = sum(abs(w) for w in target_weights.values()) * equity
    if gross <= 0.0:
        return {s: 0.0 for s in target_weights}
    deployed = min(gross, policy.capital_ceiling)
    scale = deployed / gross
    return {s: w * equity * scale for s, w in target_weights.items()}
```

- [ ] **Step 4: Run tests → PASS. Ruff.**

Run: `uv run pytest tests/unit/test_target_capital.py -q` → PASS (3 passed).
Run: `uv run ruff check src/fxhnt/application/allocation_engine.py tests/unit/test_target_capital.py`

- [ ] **Step 5: Commit**

```bash
git add src/fxhnt/application/allocation_engine.py tests/unit/test_target_capital.py
git commit -F - <<'EOF'
feat(alloc): target_capital — hard capital_ceiling caps total deployed gross (B2a)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
EOF
```

---

### Task 5: `strategy_allocation` table + repo write/read

**Files:**
- Modify: `src/fxhnt/adapters/persistence/cockpit_models.py` (add `StrategyAllocationRow`)
- Modify: `src/fxhnt/adapters/persistence/forward_nav.py` (add `upsert_allocation`, `current_allocation`)
- Test: `tests/integration/test_strategy_allocation.py`

**Interfaces:**
- Produces:
  - `StrategyAllocationRow(strategy_id, target_weight, target_dollars, policy_version, policy_hash, as_of, computed_at)` — `strategy_id` PK (one current row per strategy, replaced each run).
  - `ForwardNavRepo.upsert_allocation(strategy_id, target_weight, target_dollars, policy_version, policy_hash, as_of, at)`
  - `ForwardNavRepo.current_allocation() -> dict[str, float]` — `{strategy_id: target_dollars}` for the latest snapshot.

- [ ] **Step 1: Write the failing test**

```python
# tests/integration/test_strategy_allocation.py
import datetime as dt

from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo


def test_allocation_round_trips(tmp_path):
    repo = ForwardNavRepo(f"sqlite:///{tmp_path}/a.db"); repo.migrate()
    at = dt.datetime(2026, 7, 9, 23, 30)
    repo.upsert_allocation("multistrat", target_weight=0.6, target_dollars=6000.0,
                           policy_version=1, policy_hash="h", as_of="2026-07-09", at=at)
    repo.upsert_allocation("bybit_4edge", target_weight=0.4, target_dollars=4000.0,
                           policy_version=1, policy_hash="h", as_of="2026-07-09", at=at)
    assert repo.current_allocation() == {"multistrat": 6000.0, "bybit_4edge": 4000.0}
    # re-run replaces (idempotent on strategy_id)
    repo.upsert_allocation("multistrat", target_weight=0.5, target_dollars=5000.0,
                           policy_version=1, policy_hash="h", as_of="2026-07-10", at=at)
    assert repo.current_allocation()["multistrat"] == 5000.0
```

- [ ] **Step 2: Run test to verify it fails**

Run: `uv run pytest tests/integration/test_strategy_allocation.py -q`
Expected: FAIL with `AttributeError: 'ForwardNavRepo' object has no attribute 'upsert_allocation'`.

- [ ] **Step 3: Add the ORM row**

In `cockpit_models.py`, append:

```python
class StrategyAllocationRow(CockpitBase):
    """The latest B2a strategy-level allocation snapshot — one row per strategy, replaced each run.
    target_weight = the ex-ante-sized weight; target_dollars = capital-ceiling-bounded dollars. Stamped with
    the allocation policy that produced it (audit). A pure derived cache of (policy + A's records + B1 status)."""
    __tablename__ = "strategy_allocation"
    strategy_id: Mapped[str] = mapped_column(String(64), primary_key=True)
    target_weight: Mapped[float] = mapped_column(Float)
    target_dollars: Mapped[float] = mapped_column(Float)
    policy_version: Mapped[int] = mapped_column(Integer)
    policy_hash: Mapped[str] = mapped_column(String(64))
    as_of: Mapped[str] = mapped_column(String(10))
    computed_at: Mapped[dt.datetime] = mapped_column(DateTime)
```

- [ ] **Step 4: Add the repo methods**

Add `StrategyAllocationRow` to the `cockpit_models` import in `forward_nav.py`, then add to `ForwardNavRepo`:

```python
    def upsert_allocation(self, strategy_id: str, target_weight: float, target_dollars: float,
                          policy_version: int, policy_hash: str, as_of: str, at: dt.datetime) -> None:
        with Session(self._engine) as s:
            s.merge(StrategyAllocationRow(
                strategy_id=strategy_id, target_weight=target_weight, target_dollars=target_dollars,
                policy_version=policy_version, policy_hash=policy_hash, as_of=as_of, computed_at=at))
            s.commit()

    def current_allocation(self) -> dict[str, float]:
        with Session(self._engine) as s:
            return {r.strategy_id: r.target_dollars for r in s.scalars(select(StrategyAllocationRow))}
```

- [ ] **Step 5: Run tests → PASS + ruff, then commit**

Run: `uv run pytest tests/integration/test_strategy_allocation.py -q` → PASS.
Run: `uv run ruff check src/fxhnt/adapters/persistence/cockpit_models.py src/fxhnt/adapters/persistence/forward_nav.py tests/integration/test_strategy_allocation.py`

```bash
git add src/fxhnt/adapters/persistence/cockpit_models.py src/fxhnt/adapters/persistence/forward_nav.py tests/integration/test_strategy_allocation.py
git commit -F - <<'EOF'
feat(alloc): strategy_allocation table + repo (record the B2a allocation snapshot) (B2a)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
EOF
```

---

### Task 6: nightly step — compute + record the allocation from A's records + B1's deploy set

**Files:**
- Create: `src/fxhnt/application/allocation_ingest.py` (`record_allocation(dsn, equity)`)
- Modify: `src/fxhnt/adapters/orchestration/assets.py` (`cockpit_forward` calls it after `evaluate_promotions`)
- Test: `tests/integration/test_allocation_ingest.py`

**Interfaces:**
- Consumes: `ForwardNavRepo.nav_history`, `promoted_strategy_ids`, `STRATEGY_REGISTRY` (`tier`), `compute_allocation`/`target_capital` (Tasks 3-4), `ALLOCATION_POLICY`/`ALLOCATION_POLICY_VERSION`/`allocation_policy_hash` (Task 1), `upsert_allocation` (Task 5).
- Produces: `record_allocation(dsn: str, equity: float) -> dict[str, float]` — computes the deploy-set allocation, records each strategy's snapshot, returns `{strategy_id: target_dollars}`.

- [ ] **Step 1: Write the failing test**

```python
# tests/integration/test_allocation_ingest.py
import datetime as dt

from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
from fxhnt.application.allocation_ingest import record_allocation
from fxhnt.application.forward_models import ForwardNavRow as NavRowDTO
from fxhnt import registry


def test_records_allocation_for_deploy_set(tmp_path, monkeypatch):
    dsn = f"sqlite:///{tmp_path}/ai.db"
    repo = ForwardNavRepo(dsn); repo.migrate()
    at = dt.datetime(2026, 7, 9, 23, 30)
    monkeypatch.setitem(registry.STRATEGY_REGISTRY, "d1",
                        {"display_name": "D1", "sleeve": "x", "gate_spec": {}, "tier": "deploy"})
    monkeypatch.setitem(registry.STRATEGY_REGISTRY, "d2",
                        {"display_name": "D2", "sleeve": "x", "gate_spec": {}, "tier": "deploy"})
    for sid, flip in (("d1", 1), ("d2", 0)):
        for i, dd in enumerate([f"2026-06-{x:02d}" for x in range(1, 21)] +
                               [f"2026-07-{x:02d}" for x in range(1, 41)]):
            repo.upsert_rows([NavRowDTO(strategy_id=sid, date=dd,
                                        ret=(0.01 if (i % 2) == flip else -0.008), nav=1.0)], at=at)
    out = record_allocation(dsn, equity=20_000.0)
    assert set(out) <= {"d1", "d2"}                              # only deploy-set considered
    assert sum(abs(v) for v in out.values()) <= 35_000.0 + 1e-6  # capped by the default ceiling
    assert repo.current_allocation() == out                     # recorded
```

- [ ] **Step 2: Run test to verify it fails**

Run: `uv run pytest tests/integration/test_allocation_ingest.py -q`
Expected: FAIL with `ModuleNotFoundError: No module named 'fxhnt.application.allocation_ingest'`.

- [ ] **Step 3: Implement `allocation_ingest.py`**

```python
# src/fxhnt/application/allocation_ingest.py
"""Compute + record the B2a strategy-level allocation from A's records + B1's deploy set. Runs in the nightly
AFTER B1's promotion (so the deploy-tier set is fresh). Pure allocation math (allocation_engine) + a recorded
snapshot; no capital is moved (that is sub-project C)."""
from __future__ import annotations

import datetime as dt
import logging

from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
from fxhnt.application.allocation_engine import compute_allocation, target_capital
from fxhnt.application.allocation_policy import (
    ALLOCATION_POLICY, ALLOCATION_POLICY_VERSION, allocation_policy_hash)
from fxhnt.registry import STRATEGY_REGISTRY

_log = logging.getLogger(__name__)


def record_allocation(dsn: str, equity: float) -> dict[str, float]:
    repo = ForwardNavRepo(dsn)
    repo.migrate()
    deploy_sids = {sid for sid, m in STRATEGY_REGISTRY.items() if m.get("tier") == "deploy"} \
        | repo.promoted_strategy_ids()
    returns = {sid: {r.date: r.ret for r in repo.nav_history(sid)} for sid in deploy_sids}
    weights = compute_allocation(returns, ALLOCATION_POLICY)
    dollars = target_capital(weights, equity, ALLOCATION_POLICY)
    at = dt.datetime.now(dt.timezone.utc).replace(tzinfo=None)
    phash = allocation_policy_hash(ALLOCATION_POLICY, ALLOCATION_POLICY_VERSION)
    as_of = at.strftime("%Y-%m-%d")
    for sid in weights:
        repo.upsert_allocation(sid, target_weight=weights[sid], target_dollars=dollars.get(sid, 0.0),
                               policy_version=ALLOCATION_POLICY_VERSION, policy_hash=phash, as_of=as_of, at=at)
    _log.info("record_allocation: %d strategies allocated (gross $%.0f, ceiling $%.0f)",
              len(weights), sum(abs(v) for v in dollars.values()), ALLOCATION_POLICY.capital_ceiling)
    return dollars
```

- [ ] **Step 4: Wire it into `cockpit_forward`**

In `src/fxhnt/adapters/orchestration/assets.py`, the `cockpit_forward` asset currently calls
`ingest_forward_state(dsn)` then `evaluate_promotions(dsn)`. Add the allocation record AFTER promotions (so
the deploy set is fresh), using the paper capital as the equity input:

```python
    from fxhnt.application.allocation_ingest import record_allocation
    # ... after: promoted = evaluate_promotions(dsn)
    allocated = record_allocation(dsn, equity=get_settings().paper_capital)
    context.log.info(f"cockpit_forward: allocated {len(allocated)} strategies (promoted: {promoted or 'none'})")
```
(Read the current `cockpit_forward` body first; keep the existing ingest + promotion calls + log line, add the allocation record after them.)

- [ ] **Step 5: Run tests + full suite + ruff, then commit**

Run: `uv run pytest tests/integration/test_allocation_ingest.py -q` → PASS.
Run the FULL suite ONCE (this touches the nightly `cockpit_forward` path): `uv run pytest -q` → PASS.
Run: `uv run ruff check src/fxhnt/application/allocation_ingest.py src/fxhnt/adapters/orchestration/assets.py tests/integration/test_allocation_ingest.py`

```bash
git add src/fxhnt/application/allocation_ingest.py src/fxhnt/adapters/orchestration/assets.py tests/integration/test_allocation_ingest.py
git commit -F - <<'EOF'
feat(alloc): nightly records the B2a allocation from A's records + B1 deploy set (B2a)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
EOF
```

---

## Self-Review notes

- **Spec coverage:** policy object (§1)→T1; engine helpers (§2)→T2; compute_allocation ex-ante composition (§2)→T3; capital ceiling (§3)→T4; recorded snapshot + audit stamp (§4)→T5+T6; where-human-stays (§5)→T6 (records targets, no capital moved); anti-reactive invariant (error-handling)→T3 test; killswitch→T2/T3; min_history exclusion→T3.
- **Anti-reactive invariant:** T3's `test_recent_vol_spike_does_not_change_K_antireactive` pins that `K` comes from the FULL-history book vol (not a trailing window) — the load-bearing guarantee.
- **No reactive path:** the plan reuses `promotion_corr.deploy_book_returns` + `per_period_sharpe` and reimplements the killswitch fold; it does NOT touch `book_allocator.combine_book` (the rejected trailing-vol leverage).
- **Type consistency:** `compute_allocation(returns_by_strategy: dict[str, dict[str,float]], policy)` → weights; `target_capital(weights, equity, policy)` → dollars; both consumed by `record_allocation` (T6). `deploy_book_returns` (B1) is the equal-weight aggregate used by both the vol + the marginal gate. `full_history_vol`/`equal_weight_marginal_prune`/`killswitch_active` signatures match T2 defs.
- **No capital moved:** T6 records dollar TARGETS only; execution (moving capital) is sub-project C.
