# Funding Revocation on Sustained Decay (B3) 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:** De-fund an effective-deploy edge whose reconciliation gate has shown sustained divergence from its backtest (hysteresis, no flicker), and re-fund it on sustained recovery — as a pure recompute over A's immutable forward record, narrowing B2a's allocation input.

**Architecture:** A machine-readable `category` on the gate verdict classifies each day (PASS / DIVERGENCE / IMMATURE / EXECUTION_GAP). A pure engine (`funding_status.py`) recomputes the reconciliation gate as-of each record-day and replays a hysteretic funding state machine from inception (no persisted mutable state). `record_allocation` filters the effective deploy set through it before B2a's `compute_allocation`; a `strategy_funding` snapshot is recorded for audit. B2a and promotion are untouched.

**Tech Stack:** Python 3.13, frozen dataclasses, SQLAlchemy 2.0 (SQLite in tests / Postgres+Timescale in prod), pytest, ruff.

## Global Constraints

- Signal is B1's reconciliation gate verdict category — **only `DIVERGENCE` de-funds; only `PASS` re-funds; `IMMATURE`/`EXECUTION_GAP` are neutral** (never de-fund/re-fund).
- **No new mutable state** — funding is a pure function of (policy + immutable forward record), replayed from inception each run (mirrors A/B2a; never a persisted latch).
- **Scope guard:** an edge's decay run advances **only after its first PASS** (`has_passed`); an edge that never PASSed is never de-funded.
- De-funding is **binary + structural** (sustained divergence over `defund_window`), never a continuous per-day throttle (honours [[pearl_fxhnt_reactive_risk_controllers_rejected]]). Re-funding is symmetric and slower (`refund_window > defund_window`).
- Funding-window knobs live in B1's versioned `GatePolicy`; any change bumps `POLICY_VERSION` and the content-derived `policy_hash` (self-check test forces the bump).
- Wiring is a **filter in front of B2a** (Approach A): a de-funded edge is *absent* from `compute_allocation`'s input → $0. B2a's `compute_allocation`/`target_capital` and promotion (`promoted_strategy_ids`) are **not modified**.
- No capital is moved (that is sub-project C). Cockpit visual surfacing is **deferred** (out of scope) — only the `strategy_funding` audit table is delivered here.
- `ForwardNavRow.nav` is cumulative NAV, 1.0-base → `total_return = nav − 1.0`. `nav_history(sid)` returns `NavRowDTO(strategy_id, date: ISO-str, ret: float, nav: float)` ascending by date.

---

### Task 1: Machine-readable gate-verdict category

**Files:**
- Modify: `src/fxhnt/application/forward_models.py` (add category constants + `GateVerdict.category`)
- Modify: `src/fxhnt/application/reconciliation_gate.py:99-144` (set category on every return)
- Modify: `src/fxhnt/application/gate.py:10-21` (set category on every return)
- Test: `tests/unit/test_gate_verdict_category.py`

**Interfaces:**
- Produces: `GateVerdict(status: str, reason: str, category: str = CAT_IMMATURE)`; constants `CAT_PASS="PASS"`, `CAT_DIVERGENCE="DIVERGENCE"`, `CAT_IMMATURE="IMMATURE"`, `CAT_EXECUTION_GAP="EXECUTION_GAP"` in `forward_models`.

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

Create `tests/unit/test_gate_verdict_category.py`:

```python
from fxhnt.application.forward_models import (
    CAT_DIVERGENCE, CAT_EXECUTION_GAP, CAT_IMMATURE, CAT_PASS, ForwardNavRow, ForwardSummary,
)
from fxhnt.application.gate import evaluate_gate
from fxhnt.application.gate_policy_resolve import ResolvedPolicy
from fxhnt.application.reconciliation_gate import BacktestReference, evaluate_reconciliation_gate

_RP = ResolvedPolicy(min_forward_days=3, recon_tolerance_frac=0.5, recon_min_band=0.05,
                     recon_min_corr=0.0, max_gap_days=1, min_days=3, min_total_return=0.0, min_sharpe=0.0)


def _rows(navs, start_day=1):
    # consecutive daily rows 2026-07-0X; ret unused by the band checks here
    return [ForwardNavRow(strategy_id="s", date=f"2026-07-{start_day+i:02d}", ret=0.0, nav=v)
            for i, v in enumerate(navs)]


def _summary(nav, days):
    return ForwardSummary(strategy_id="s", as_of="2026-07-09", days=days, nav=nav,
                          total_return=nav - 1.0, sharpe=0.0, maxdd=0.0)


def test_reconciliation_categories():
    ref = BacktestReference(expected_window_return=0.10, daily_returns=None)
    # building (days < min_forward_days) -> IMMATURE
    assert evaluate_reconciliation_gate(_summary(1.10, 2), _rows([1.05, 1.10]), ref, _RP).category == CAT_IMMATURE
    # no backtest reference -> IMMATURE
    assert evaluate_reconciliation_gate(_summary(1.10, 3), _rows([1.0, 1.05, 1.10]), None, _RP).category == CAT_IMMATURE
    # execution gap (date jump > max_gap_days) -> EXECUTION_GAP
    gap = [ForwardNavRow("s", "2026-07-01", 0.0, 1.0), ForwardNavRow("s", "2026-07-02", 0.0, 1.05),
           ForwardNavRow("s", "2026-07-05", 0.0, 1.10)]
    assert evaluate_reconciliation_gate(_summary(1.10, 3), gap, ref, _RP).category == CAT_EXECUTION_GAP
    # forward far below backtest expectation -> DIVERGENCE
    assert evaluate_reconciliation_gate(_summary(0.95, 3), _rows([1.0, 0.98, 0.95]), ref, _RP).category == CAT_DIVERGENCE
    # reconciles -> PASS
    assert evaluate_reconciliation_gate(_summary(1.10, 3), _rows([1.0, 1.05, 1.10]), ref, _RP).category == CAT_PASS


def test_absolute_gate_categories():
    # evaluate_gate: GO -> PASS, everything else -> IMMATURE (a non-reconciliation edge is never de-funded)
    assert evaluate_gate(_summary(1.10, 5), _RP).category == CAT_PASS
    assert evaluate_gate(_summary(1.0, 1), _RP).category == CAT_IMMATURE          # building -> IMMATURE
    neg = ResolvedPolicy(3, 0.5, 0.05, 0.0, 1, 3, 0.05, 0.0)                       # min_total_return=0.05
    assert evaluate_gate(_summary(1.0, 5), neg).category == CAT_IMMATURE          # NO_GO return floor -> IMMATURE
```

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

Run: `cd /home/jgrusewski/Work/fxhnt && source .venv/bin/activate && python -m pytest tests/unit/test_gate_verdict_category.py -q`
Expected: FAIL — `ImportError: cannot import name 'CAT_PASS'` (constants + field not defined yet).

- [ ] **Step 3: Add the constants + field in `forward_models.py`**

Replace the `GateVerdict` class (currently `forward_models.py:51-53`) with:

```python
CAT_PASS = "PASS"
CAT_DIVERGENCE = "DIVERGENCE"
CAT_IMMATURE = "IMMATURE"
CAT_EXECUTION_GAP = "EXECUTION_GAP"


@dataclass(frozen=True, slots=True)
class GateVerdict:
    status: str        # "WAIT" | "GO" | "NO_GO" | "PASS"
    reason: str
    # machine-readable funding-signal category (B3). Default IMMATURE = neutral (never de-funds) so any
    # verdict constructed without an explicit category is fail-safe.
    category: str = CAT_IMMATURE
```

- [ ] **Step 4: Set categories in `reconciliation_gate.py`**

Add to the imports (top of `reconciliation_gate.py`, the `from fxhnt.application.forward_models import` line):

```python
from fxhnt.application.forward_models import (
    CAT_DIVERGENCE, CAT_EXECUTION_GAP, CAT_IMMATURE, CAT_PASS,
    ForwardNavRow, ForwardSummary, GateVerdict,
)
```

Then set `category=` on each return in `evaluate_reconciliation_gate`:
- building (`return GateVerdict("WAIT", f"building ...")`) → add `, category=CAT_IMMATURE`
- execution gap (`return GateVerdict("WAIT", "execution gap ...")`) → add `, category=CAT_EXECUTION_GAP`
- no backtest reference (the `if backtest is None:` return) → add `, category=CAT_IMMATURE`
- cumulative-band divergence (`shortfall > band` return) → add `, category=CAT_DIVERGENCE`
- daily-corr divergence (the `corr < min_corr` return) → add `, category=CAT_DIVERGENCE`
- final PASS return → add `, category=CAT_PASS`

- [ ] **Step 5: Set categories in `gate.py`**

In `gate.py`, add `CAT_IMMATURE, CAT_PASS` to the `forward_models` import, then:
- `GateVerdict("WAIT", f"{summary.days}/{min_days} forward days")` → add `, category=CAT_IMMATURE`
- both `GateVerdict("NO_GO", ...)` returns → add `, category=CAT_IMMATURE`
- `GateVerdict("GO", ...)` return → add `, category=CAT_PASS`

- [ ] **Step 6: Run tests to verify they pass**

Run: `cd /home/jgrusewski/Work/fxhnt && source .venv/bin/activate && python -m pytest tests/unit/test_gate_verdict_category.py -q`
Expected: PASS (2 passed).

- [ ] **Step 7: Verify no existing gate tests broke + ruff**

Run: `python -m pytest tests/ -q -k "gate or reconcil or forward_ingest" && ruff check src/fxhnt/application/forward_models.py src/fxhnt/application/reconciliation_gate.py src/fxhnt/application/gate.py`
Expected: PASS, ruff clean. (Existing verdicts are byte-identical; only the new optional field is added.)

- [ ] **Step 8: Commit**

```bash
git add src/fxhnt/application/forward_models.py src/fxhnt/application/reconciliation_gate.py src/fxhnt/application/gate.py tests/unit/test_gate_verdict_category.py
git commit -m "feat(gate): machine-readable verdict category (PASS/DIVERGENCE/IMMATURE/EXECUTION_GAP) for B3

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

---

### Task 2: Funding windows in the versioned GatePolicy

**Files:**
- Modify: `src/fxhnt/application/gate_policy.py:16-34` (add fields + bump version)
- Modify: `tests/unit/test_gate_policy.py` (update defaults assertion + pinned hash)
- Test: `tests/unit/test_gate_policy.py`

**Interfaces:**
- Produces: `GatePolicy.defund_window: int = 10`, `GatePolicy.refund_window: int = 15`; `POLICY_VERSION = 2`.

- [ ] **Step 1: Update the failing test first (defaults + version)**

In `tests/unit/test_gate_policy.py`, extend `test_policy_defaults_match_todays_constants` by appending:

```python
    assert (POLICY.defund_window, POLICY.refund_window) == (10, 15)
    assert POLICY_VERSION == 2
```

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

Run: `cd /home/jgrusewski/Work/fxhnt && source .venv/bin/activate && python -m pytest tests/unit/test_gate_policy.py::test_policy_defaults_match_todays_constants -q`
Expected: FAIL — `AttributeError: 'GatePolicy' object has no attribute 'defund_window'`.

- [ ] **Step 3: Add the fields + bump the version**

In `gate_policy.py`, add to `GatePolicy` (after the promotion block, before the class ends):

```python
    # funding revocation on sustained decay (B3): consecutive DIVERGENCE days that de-fund, and consecutive
    # PASS days that re-fund. Asymmetric (quicker to de-risk, slower to re-arm capital). Debounce over an
    # already-cumulative gate signal.
    defund_window: int = 10
    refund_window: int = 15
```

Then change `POLICY_VERSION = 1` to `POLICY_VERSION = 2`.

- [ ] **Step 4: Recompute and pin the new hash**

Run: `cd /home/jgrusewski/Work/fxhnt && source .venv/bin/activate && python -c "from fxhnt.application.gate_policy import POLICY, POLICY_VERSION, policy_hash; print(policy_hash(POLICY, POLICY_VERSION))"`

Copy the printed hex into `tests/unit/test_gate_policy.py`, replacing the `_EXPECTED_HASH = "..."` literal with the new value.

- [ ] **Step 5: Run the gate-policy tests**

Run: `python -m pytest tests/unit/test_gate_policy.py -q`
Expected: PASS (3 passed) — the self-check now pins v2's hash, proving the version bump is deliberate.

- [ ] **Step 6: Fix any other tests that pinned policy version/hash**

Run: `python -m pytest tests/integration/test_forward_summary_policy_stamp.py tests/integration/test_forward_ingest.py tests/integration/test_strategy_allocation.py tests/integration/test_execute_multistrat_b2a.py tests/unit/test_allocation_policy.py -q`
Expected: PASS. If any assert `policy_version == 1` (gate policy) or pin the old gate `policy_hash`, update them to `POLICY_VERSION` / recomputed hash. (The B2a `ALLOCATION_POLICY_VERSION` is a SEPARATE policy — do not touch it; only the B1 gate `POLICY_VERSION` changed.)

- [ ] **Step 7: Commit**

```bash
git add src/fxhnt/application/gate_policy.py tests/unit/test_gate_policy.py
git commit -m "feat(policy): add defund/refund windows to GatePolicy, bump POLICY_VERSION=2 (B3)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>"
```
(Include any other test files you had to touch in Step 6 in the `git add`.)

---

### Task 3: The pure funding engine (`funding_status.py`)

**Files:**
- Create: `src/fxhnt/application/funding_status.py`
- Test: `tests/unit/test_funding_status.py`

**Interfaces:**
- Consumes: `GateVerdict.category` (Task 1); `ResolvedPolicy` (`gate_policy_resolve`); `evaluate_reconciliation_gate`, `BacktestReference` (`reconciliation_gate`); `ForwardSummary` (`forward_models`); rows with `.date: str`, `.ret: float`, `.nav: float` (A's `NavRowDTO`, ascending).
- Produces: `FundingStatus(state, label, decay_run, recovery_run, first_pass_date)`; `compute_funding(sid, rows, ref_for_days, resolved_policy, defund_window, refund_window) -> FundingStatus`, where `ref_for_days: Callable[[int], BacktestReference | None]`.

- [ ] **Step 1: Write the failing tests (the load-bearing state machine)**

Create `tests/unit/test_funding_status.py`:

```python
from fxhnt.application.forward_models import ForwardNavRow
from fxhnt.application.funding_status import compute_funding
from fxhnt.application.gate_policy_resolve import ResolvedPolicy
from fxhnt.application.reconciliation_gate import BacktestReference

# min_forward_days=1 so every day yields a non-building verdict; tiny windows so sequences stay short.
_RP = ResolvedPolicy(min_forward_days=1, recon_tolerance_frac=0.5, recon_min_band=0.05,
                     recon_min_corr=0.0, max_gap_days=1, min_days=1, min_total_return=0.0, min_sharpe=0.0)
_REF = lambda _days: BacktestReference(expected_window_return=0.10, daily_returns=None)  # fixed expectation
_DEFUND, _REFUND = 2, 3


def _rows(navs):
    # consecutive dates so _has_gaps stays False (no EXECUTION_GAP unless we deliberately gap)
    return [ForwardNavRow("s", f"2026-07-{1+i:02d}", 0.0, v) for i, v in enumerate(navs)]


def _fund(navs):
    return compute_funding("s", _rows(navs), _REF, _RP, _DEFUND, _REFUND)


# nav 1.10 -> total_return 0.10 == expected -> PASS ; nav 0.95 -> -0.05, shortfall 0.15 > band 0.05 -> DIVERGENCE

def test_sustained_divergence_defunds_after_window():
    s = _fund([1.10, 0.95, 0.95])          # PASS, DIVERGENCE x2 (== defund_window)
    assert s.state == "de-funded"
    assert s.first_pass_date == "2026-07-01"


def test_transient_dip_does_not_flicker():
    s = _fund([1.10, 0.95, 1.10])          # PASS, DIVERGENCE(1), PASS -> decay run reset
    assert s.state == "funded"
    assert s.decay_run == 0


def test_recovery_refunds_after_window():
    s = _fund([1.10, 0.95, 0.95, 1.10, 1.10, 1.10])   # de-fund at idx2, then 3 PASS == refund_window
    assert s.state == "funded"


def test_never_passed_is_never_defunded():
    s = _fund([0.95, 0.95, 0.95, 0.95])    # all DIVERGENCE, no PASS -> decay run never advances
    assert s.state == "funded"
    assert s.first_pass_date is None


def test_execution_gap_resets_decay_run():
    rows = [ForwardNavRow("s", "2026-07-01", 0.0, 1.10),   # PASS
            ForwardNavRow("s", "2026-07-02", 0.0, 0.95),   # DIVERGENCE (decay_run 1)
            ForwardNavRow("s", "2026-07-04", 0.0, 0.95)]   # gap>1 -> whole record EXECUTION_GAP -> neutral
    s = compute_funding("s", rows, _REF, _RP, _DEFUND, _REFUND)
    assert s.state == "funded"


def test_deterministic():
    assert _fund([1.10, 0.95, 0.95]) == _fund([1.10, 0.95, 0.95])
```

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

Run: `cd /home/jgrusewski/Work/fxhnt && source .venv/bin/activate && python -m pytest tests/unit/test_funding_status.py -q`
Expected: FAIL — `ModuleNotFoundError: No module named 'fxhnt.application.funding_status'`.

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

Create `src/fxhnt/application/funding_status.py`:

```python
"""Sub-project B3 — funding revocation on sustained decay. Pure, recompute-from-record: replay B1's
reconciliation gate as-of each forward record-day and step a hysteretic funding state machine from
inception. No persisted mutable state (mirrors A/B2a). Only DIVERGENCE de-funds; only PASS re-funds;
IMMATURE/EXECUTION_GAP are neutral. An edge's decay run advances only after its first PASS."""
from __future__ import annotations

from collections.abc import Callable, Sequence
from dataclasses import dataclass
from typing import Protocol

from fxhnt.application.forward_models import CAT_DIVERGENCE, CAT_PASS, ForwardSummary
from fxhnt.application.gate_policy_resolve import ResolvedPolicy
from fxhnt.application.reconciliation_gate import BacktestReference, evaluate_reconciliation_gate


class _Row(Protocol):
    date: str
    ret: float
    nav: float


@dataclass(frozen=True, slots=True)
class FundingStatus:
    state: str                    # "funded" | "de-funded" — the capital verdict
    label: str                    # "funded" | "decaying" | "de-funded" — display
    decay_run: int
    recovery_run: int
    first_pass_date: str | None


def compute_funding(sid: str, rows: Sequence[_Row],
                    ref_for_days: Callable[[int], BacktestReference | None],
                    resolved_policy: ResolvedPolicy,
                    defund_window: int, refund_window: int) -> FundingStatus:
    """Replay the funding state machine over `rows` (ascending). At each day, recompute the reconciliation
    gate as-of that day (as-of summary from the truncated record; backtest reference pro-rated to the day's
    forward length) and step the machine. Deterministic; no persisted state."""
    state = "funded"
    decay_run = recovery_run = 0
    has_passed = False
    first_pass_date: str | None = None

    for i in range(len(rows)):
        upto = rows[:i + 1]
        last = upto[-1]
        days = i + 1
        summary = ForwardSummary(strategy_id=sid, as_of=last.date, days=days, nav=last.nav,
                                 total_return=last.nav - 1.0, sharpe=0.0, maxdd=0.0)
        ref = ref_for_days(max(days, resolved_policy.min_forward_days))
        category = evaluate_reconciliation_gate(summary, list(upto), ref, resolved_policy).category

        if category == CAT_PASS and not has_passed:
            has_passed = True
            first_pass_date = last.date

        if state == "funded":
            decay_run = decay_run + 1 if (category == CAT_DIVERGENCE and has_passed) else 0
            if decay_run >= defund_window:
                state, decay_run, recovery_run = "de-funded", 0, 0
        else:  # de-funded
            recovery_run = recovery_run + 1 if category == CAT_PASS else 0
            if recovery_run >= refund_window:
                state, decay_run, recovery_run = "funded", 0, 0

    label = "de-funded" if state == "de-funded" else ("decaying" if decay_run > 0 else "funded")
    return FundingStatus(state, label, decay_run, recovery_run, first_pass_date)
```

- [ ] **Step 4: Run to verify pass**

Run: `cd /home/jgrusewski/Work/fxhnt && source .venv/bin/activate && python -m pytest tests/unit/test_funding_status.py -q`
Expected: PASS (6 passed).

- [ ] **Step 5: ruff**

Run: `ruff check src/fxhnt/application/funding_status.py tests/unit/test_funding_status.py`
Expected: clean.

- [ ] **Step 6: Commit**

```bash
git add src/fxhnt/application/funding_status.py tests/unit/test_funding_status.py
git commit -m "feat(funding): pure recompute-from-record funding state machine (B3)

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

---

### Task 4: `strategy_funding` audit table + repo

**Files:**
- Modify: `src/fxhnt/adapters/persistence/cockpit_models.py` (new `StrategyFundingRow`, after `StrategyAllocationRow`)
- Modify: `src/fxhnt/adapters/persistence/forward_nav.py` (import + `replace_funding` / `current_funding`)
- Test: `tests/integration/test_strategy_funding.py`

**Interfaces:**
- Produces: `StrategyFundingRow(strategy_id, state, label, decay_run, recovery_run, first_pass_date, policy_version, policy_hash, as_of, computed_at)`; `ForwardNavRepo.replace_funding(rows, at)`, `ForwardNavRepo.current_funding() -> dict[str, str]` (sid → state).

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

Create `tests/integration/test_strategy_funding.py`:

```python
import datetime as dt

from fxhnt.adapters.persistence.cockpit_models import StrategyFundingRow
from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo


def _repo(tmp_path):
    repo = ForwardNavRepo(f"sqlite:///{tmp_path/'c.db'}")
    repo.migrate()
    return repo


def _row(sid, state):
    return StrategyFundingRow(strategy_id=sid, state=state, label=state, decay_run=0, recovery_run=0,
                              first_pass_date=None, policy_version=2, policy_hash="h", as_of="2026-07-10",
                              computed_at=dt.datetime(2026, 7, 10))


def test_replace_and_read_funding(tmp_path):
    repo = _repo(tmp_path)
    at = dt.datetime(2026, 7, 10)
    repo.replace_funding([_row("a", "funded"), _row("b", "de-funded")], at)
    assert repo.current_funding() == {"a": "funded", "b": "de-funded"}


def test_replace_clears_stale(tmp_path):
    repo = _repo(tmp_path)
    at = dt.datetime(2026, 7, 10)
    repo.replace_funding([_row("a", "funded"), _row("b", "de-funded")], at)
    repo.replace_funding([_row("a", "funded")], at)          # 'b' dropped this run
    assert repo.current_funding() == {"a": "funded"}
```

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

Run: `cd /home/jgrusewski/Work/fxhnt && source .venv/bin/activate && python -m pytest tests/integration/test_strategy_funding.py -q`
Expected: FAIL — `ImportError: cannot import name 'StrategyFundingRow'`.

- [ ] **Step 3: Add the model**

In `cockpit_models.py`, after the `StrategyAllocationRow` class, add:

```python
class StrategyFundingRow(CockpitBase):
    """B3 funding-status snapshot — one row per effective-deploy strategy, replaced each run. A derived cache
    of (gate policy + A's immutable record). `state` gates B2a's allocation input; `label` is the display."""
    __tablename__ = "strategy_funding"
    strategy_id: Mapped[str] = mapped_column(String(64), primary_key=True)
    state: Mapped[str] = mapped_column(String(16))              # funded | de-funded
    label: Mapped[str] = mapped_column(String(16))             # funded | decaying | de-funded
    decay_run: Mapped[int] = mapped_column(Integer)
    recovery_run: Mapped[int] = mapped_column(Integer)
    first_pass_date: Mapped[str | None] = mapped_column(String(10), nullable=True, default=None)
    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**

In `forward_nav.py`, add `StrategyFundingRow` to the `cockpit_models` import block (beside `StrategyAllocationRow`), then add after `current_allocation` (near line 272):

```python
    def replace_funding(self, rows: list[StrategyFundingRow], at: dt.datetime) -> None:
        """Replace the ENTIRE strategy_funding snapshot (delete-then-insert) so an edge that leaves the
        effective-deploy set leaves no stale funding row behind."""
        with Session(self._engine) as s:
            s.query(StrategyFundingRow).delete()
            for r in rows:
                r.computed_at = at
                s.add(r)
            s.commit()

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

`migrate()` already calls `CockpitBase.metadata.create_all` — it creates the new `strategy_funding` table on both SQLite and prod Postgres (create_all adds missing *tables*); no `_ensure` column-patch is needed for a brand-new table.

- [ ] **Step 5: Run to verify pass**

Run: `cd /home/jgrusewski/Work/fxhnt && source .venv/bin/activate && python -m pytest tests/integration/test_strategy_funding.py -q`
Expected: PASS (2 passed).

- [ ] **Step 6: ruff**

Run: `ruff check src/fxhnt/adapters/persistence/cockpit_models.py src/fxhnt/adapters/persistence/forward_nav.py tests/integration/test_strategy_funding.py`
Expected: clean.

- [ ] **Step 7: Commit**

```bash
git add src/fxhnt/adapters/persistence/cockpit_models.py src/fxhnt/adapters/persistence/forward_nav.py tests/integration/test_strategy_funding.py
git commit -m "feat(persistence): strategy_funding audit table + repo (B3)

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

---

### Task 5: Wire the funding filter into `record_allocation`

**Files:**
- Modify: `src/fxhnt/application/allocation_ingest.py`
- Test: `tests/integration/test_allocation_funding_filter.py`

**Interfaces:**
- Consumes: `compute_funding` (Task 3), `StrategyFundingRow` + `replace_funding` (Task 4), `CockpitBacktestRefProvider` (`forward_ingest`), `resolve_policy` + `POLICY`/`POLICY_VERSION`/`policy_hash` (`gate_policy*`).
- Produces: unchanged `record_allocation(dsn, equity) -> dict[str, float]` signature (a de-funded edge is absent from the returned dollars); a recorded `strategy_funding` snapshot over ALL effective-deploy edges.

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

Create `tests/integration/test_allocation_funding_filter.py`. It seeds the real `bybit_4edge` (a registry `tier=="deploy"`, reconciliation-gated edge) two ways in two tests: (1) healthy, ≥60 days → stays funded and receives dollars; (2) PASSes then sustainedly diverges → de-funded and absent from the allocation. The dsn passed to `record_allocation` is the same sqlite file the repo uses. Seed values use **wide margins** so the gate verdicts are unambiguous — adjust only if a local threshold shifts them.

```python
import datetime as dt

import pytest
from sqlalchemy.orm import Session

from fxhnt.adapters.persistence.cockpit_models import BacktestSummaryRow, ForwardNavRow, ForwardSummaryRow
from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
from fxhnt.application.allocation_ingest import record_allocation
from fxhnt.registry import STRATEGY_REGISTRY

_DSN = None
_SID = "bybit_4edge"   # registry tier==deploy, gate_type reconciliation, min_forward_days 21


def _mk_repo(tmp_path):
    global _DSN
    _DSN = f"sqlite:///{tmp_path/'c.db'}"
    repo = ForwardNavRepo(_DSN)
    repo.migrate()
    return repo


def _seed(repo, sid, navs, cagr, at, start="2026-01-01"):
    """Seed a forward record (consecutive daily rows), a summary, and a backtest_summary (the recon ref)."""
    d0 = dt.date.fromisoformat(start)
    with Session(repo._engine) as s:
        for i, v in enumerate(navs):
            ret = (navs[i] / navs[i - 1] - 1.0) if i else (v - 1.0)
            s.add(ForwardNavRow(strategy_id=sid, date=d0 + dt.timedelta(days=i), ret=ret, nav=v, src_updated_at=at))
        last = (d0 + dt.timedelta(days=len(navs) - 1)).isoformat()
        s.add(ForwardSummaryRow(strategy_id=sid, as_of=last, days=len(navs), nav=navs[-1],
                                total_return=navs[-1] - 1.0, sharpe=1.0, maxdd=-0.05,
                                gate_status="WAIT", gate_reason="", src_updated_at=at))
        s.add(BacktestSummaryRow(strategy_id=sid, as_of=last, cagr=cagr, ann_vol=0.2, sharpe=1.0,
                                 max_drawdown=-0.1, passed=True, dsr=0.9, is_sharpe=1.0, oos_sharpe=1.0,
                                 pvalue=0.01, src_updated_at=at))
        s.commit()


@pytest.mark.skipif(_SID not in STRATEGY_REGISTRY or STRATEGY_REGISTRY.get(_SID, {}).get("tier") != "deploy",
                    reason="needs bybit_4edge as a tier==deploy reconciliation edge")
def test_healthy_deploy_edge_stays_funded_and_allocated(tmp_path):
    repo = _mk_repo(tmp_path)
    at = dt.datetime(2026, 7, 10)
    # 65 days steadily climbing (>= B2a min_history_days=60, positive Sharpe), modest backtest cagr -> reconciles.
    _seed(repo, _SID, [1.0 + 0.0015 * i for i in range(65)], cagr=0.6, at=at)
    dollars = record_allocation(_DSN, equity=35_000.0)
    assert repo.current_funding().get(_SID) == "funded"
    assert dollars.get(_SID, 0.0) > 0.0


@pytest.mark.skipif(_SID not in STRATEGY_REGISTRY or STRATEGY_REGISTRY.get(_SID, {}).get("tier") != "deploy",
                    reason="needs bybit_4edge as a tier==deploy reconciliation edge")
def test_decayed_deploy_edge_is_defunded_and_absent(tmp_path):
    repo = _mk_repo(tmp_path)
    at = dt.datetime(2026, 7, 10)
    # Days 0..22 climb to +7.7% (first PASS once days >= min_forward_days 21), then 11 days flat at -15%
    # (sustained DIVERGENCE >= defund_window 10 -> de-funded). cagr 2.0 keeps the backtest expectation clearly
    # positive so the -15% forward is an unambiguous divergence.
    navs = [1.0 + 0.0035 * i for i in range(23)] + [0.85] * 11
    _seed(repo, _SID, navs, cagr=2.0, at=at)
    dollars = record_allocation(_DSN, equity=35_000.0)
    assert repo.current_funding().get(_SID) == "de-funded"
    assert _SID not in dollars                          # de-funded -> absent from B2a's input -> no dollars
```

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

Run: `cd /home/jgrusewski/Work/fxhnt && source .venv/bin/activate && python -m pytest tests/integration/test_allocation_funding_filter.py -q`
Expected: FAIL — `current_funding` empty / `record_allocation` does not yet record funding.

- [ ] **Step 3: Wire the filter + snapshot into `record_allocation`**

Rewrite `allocation_ingest.py`'s imports and `record_allocation` body:

```python
from __future__ import annotations

import datetime as dt
import functools
import logging

from fxhnt.adapters.persistence.cockpit_models import StrategyAllocationRow, StrategyFundingRow
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.application.forward_ingest import CockpitBacktestRefProvider
from fxhnt.application.funding_status import FundingStatus, compute_funding
from fxhnt.application.gate_policy import POLICY, POLICY_VERSION, policy_hash
from fxhnt.application.gate_policy_resolve import resolve_policy
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()
    at = dt.datetime.now(dt.UTC).replace(tzinfo=None)
    as_of = at.strftime("%Y-%m-%d")

    deploy_sids = {sid for sid, m in STRATEGY_REGISTRY.items() if m.get("tier") == "deploy"} \
        | repo.promoted_strategy_ids()
    nav_rows = {sid: repo.nav_history(sid) for sid in deploy_sids}

    # B3 — funding revocation on sustained decay: recompute each edge's funding from its immutable record.
    provider = CockpitBacktestRefProvider(repo)
    gphash = policy_hash(POLICY, POLICY_VERSION)
    funding = {}
    for sid in deploy_sids:
        gate_spec = STRATEGY_REGISTRY.get(sid, {}).get("gate_spec", {})
        if gate_spec.get("gate_type") == "reconciliation":
            rp = resolve_policy(POLICY, gate_spec)
            ref_for_days = functools.partial(provider.reference, sid)
            funding[sid] = compute_funding(sid, nav_rows[sid], ref_for_days, rp,
                                           POLICY.defund_window, POLICY.refund_window)
        else:
            # a non-reconciliation edge has no divergence signal to decay against -> never de-funded.
            funding[sid] = FundingStatus("funded", "funded", 0, 0, None)
    funded_sids = {sid for sid in deploy_sids if funding[sid].state != "de-funded"}
    defunded = sorted(deploy_sids - funded_sids)
    if defunded:
        _log.warning("record_allocation: de-funded (sustained decay): %s", defunded)

    returns = {sid: {r.date: r.ret for r in nav_rows[sid]} for sid in funded_sids}
    weights = compute_allocation(returns, ALLOCATION_POLICY)
    dollars = target_capital(weights, equity, ALLOCATION_POLICY)

    phash = allocation_policy_hash(ALLOCATION_POLICY, ALLOCATION_POLICY_VERSION)
    alloc_rows = [StrategyAllocationRow(strategy_id=sid, target_weight=weights[sid],
                                        target_dollars=dollars.get(sid, 0.0),
                                        policy_version=ALLOCATION_POLICY_VERSION, policy_hash=phash,
                                        as_of=as_of, computed_at=at)
                  for sid in weights]
    repo.replace_allocations(alloc_rows, at)

    fund_rows = [StrategyFundingRow(strategy_id=sid, state=f.state, label=f.label, decay_run=f.decay_run,
                                    recovery_run=f.recovery_run, first_pass_date=f.first_pass_date,
                                    policy_version=POLICY_VERSION, policy_hash=gphash,
                                    as_of=as_of, computed_at=at)
                 for sid, f in funding.items()]
    repo.replace_funding(fund_rows, at)

    _log.info("record_allocation: %d/%d strategies funded+allocated (gross $%.0f, ceiling $%.0f)",
              len(weights), len(deploy_sids), sum(abs(v) for v in dollars.values()),
              ALLOCATION_POLICY.capital_ceiling)
    return dollars
```

- [ ] **Step 4: Run the new test + the existing allocation test**

Run: `cd /home/jgrusewski/Work/fxhnt && source .venv/bin/activate && python -m pytest tests/integration/test_allocation_funding_filter.py tests/integration/test_strategy_allocation.py -q`
Expected: PASS. (The existing `test_strategy_allocation` must still pass — a fully-healthy set means `funded_sids == deploy_sids`, so the allocation is unchanged.)

- [ ] **Step 5: ruff**

Run: `ruff check src/fxhnt/application/allocation_ingest.py tests/integration/test_allocation_funding_filter.py`
Expected: clean.

- [ ] **Step 6: Commit**

```bash
git add src/fxhnt/application/allocation_ingest.py tests/integration/test_allocation_funding_filter.py
git commit -m "feat(allocation): filter B2a input through B3 funding status + record snapshot

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

---

## Final verification (after all tasks)

- [ ] Run the full suite: `cd /home/jgrusewski/Work/fxhnt && source .venv/bin/activate && python -m pytest -q -p no:cacheprovider`
  Expected: all green (prior baseline 1915 + the new B3 tests). Investigate any failure that pins the old gate `POLICY_VERSION`/hash (Task 2 Step 6 should have caught these).
- [ ] `ruff check src/fxhnt` — no new violations from B3 files.
