feat(application): StableRotationForward live paper-track strategy (short-rich + decay kill)

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

src/fxhnt/application/stablecoin_strategy.py

@@ -0,0 +1,53 @@
+"""Live-booking stablecoin peg-reversion (short-rich) paper track — recomputable series.
+
+Reuses StableReversionRunner on a fresh Binance-spot daily-close panel; applies a PageHinkleyDecay kill
+incrementally over new days, carrying detector state in the tracker's `extra`. Short-rich only.
+Mirrors the ForwardStrategy contract: advance(last_date, extra) -> (rows, extra)."""
+from __future__ import annotations
+
+import datetime as dt
+from typing import Any, Callable
+
+from fxhnt.application.stablecoin_runner import StableReversionRunner
+from fxhnt.domain.edge_decay import PageHinkleyDecay
+
+_EPOCH = dt.date(1970, 1, 1)
+
+
+def _iso(epoch_day: int) -> str:
+    return (_EPOCH + dt.timedelta(days=int(epoch_day))).isoformat()
+
+
+class StableRotationForward:
+    def __init__(self, load_panel: Callable[[], dict[str, dict[int, float]]], *,
+                 thresh_bp: float = 50.0, cost_bps: float = 4.0, periods_per_year: int = 365,
+                 decay_lam: float = 0.20, decay_reenter_days: int = 10) -> None:
+        self._load_panel = load_panel
+        self._th = thresh_bp
+        self._cost = cost_bps
+        self._ppy = periods_per_year
+        self._lam = decay_lam
+        self._reenter = decay_reenter_days
+
+    def _series(self) -> tuple[list[int], list[float]]:
+        res = StableReversionRunner(self._load_panel(), thresh_bp=self._th, cost_bps=self._cost,
+                                    periods_per_year=self._ppy).run()
+        return [int(d) for d in res.dates], [float(r) for r in res.returns]
+
+    def advance(self, last_date: str | None,
+                extra: dict[str, Any]) -> tuple[list[tuple[str, float]], dict[str, Any]]:
+        dates, raw = self._series()
+        if not extra:
+            ph = PageHinkleyDecay(lam=self._lam, reenter_days=self._reenter)
+            rows: list[tuple[str, float]] = [(_iso(d), r) for d, r in zip(dates, raw)]
+            return rows, {"ph": ph.to_dict()}
+        ph = PageHinkleyDecay.from_dict(extra["ph"])
+        cutoff = last_date or ""
+        out: list[tuple[str, float]] = []
+        for d, r in zip(dates, raw):
+            iso = _iso(d)
+            if iso <= cutoff:
+                continue
+            alive = ph.update(r)
+            out.append((iso, r if alive else 0.0))
+        return out, {"ph": ph.to_dict()}

tests/integration/test_stablecoin_strategy.py

@@ -0,0 +1,61 @@
+import json
+
+from fxhnt.application.forward_tracker import ForwardTracker
+from fxhnt.application.stablecoin_runner import StableReversionRunner
+from fxhnt.application.stablecoin_strategy import StableRotationForward, _iso
+
+
+def _panel():
+    s0 = {d: 1.00 for d in range(160)}
+    s0[159] = 1.01
+    return {"S0": s0, "S1": {d: 1.00 for d in range(160)}}
+
+
+def test_inception_books_nothing(tmp_path):
+    strat = StableRotationForward(lambda: _panel())
+    path = str(tmp_path / "st.json")
+    status = ForwardTracker(strat, path).step()
+    assert status.forward_days == 0
+    state = json.load(open(path))
+    assert state["extra"]["ph"]["killed"] is False
+
+
+def test_subsequent_books_no_drift(tmp_path):
+    p1 = _panel()
+    holder = {"p": p1}
+    strat = StableRotationForward(lambda: holder["p"], decay_lam=1e9)
+    path = str(tmp_path / "st.json")
+    ForwardTracker(strat, path).step()
+    p2 = {s: dict(series) for s, series in p1.items()}
+    for s in p2:
+        p2[s][160] = 1.00
+    holder["p"] = p2
+    status = ForwardTracker(strat, path).step()
+    assert status.booked_today == 1
+    res = StableReversionRunner(p2, thresh_bp=50.0, cost_bps=4.0).run()
+    booked = next(d["ret"] for d in json.load(open(path))["days"] if d["date"] == _iso(int(res.dates[-1])))
+    assert booked == float(res.returns[-1])
+    assert booked > 0
+
+
+def test_killed_state_flattens(tmp_path):
+    p1 = _panel()
+    holder = {"p": p1}
+    strat = StableRotationForward(lambda: holder["p"])
+    path = str(tmp_path / "st.json")
+    ForwardTracker(strat, path).step()
+    state = json.load(open(path)); state["extra"]["ph"]["killed"] = True; state["extra"]["ph"]["recover"] = 0
+    json.dump(state, open(path, "w"))
+    p2 = {s: dict(series) for s, series in p1.items()}
+    for s in p2:
+        p2[s][160] = 1.00
+    holder["p"] = p2
+    ForwardTracker(strat, path).step()
+    assert json.load(open(path))["days"][-1]["ret"] == 0.0
+
+
+def test_extra_json_roundtrips(tmp_path):
+    strat = StableRotationForward(lambda: _panel())
+    path = str(tmp_path / "st.json")
+    ForwardTracker(strat, path).step()
+    json.dumps(json.load(open(path))["extra"])