feat(application): TsTrendForward live paper-track strategy (overlay + decay kill)

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

src/fxhnt/application/ts_trend_strategy.py

@@ -0,0 +1,88 @@
+"""Live-booking crypto TS-trend (long/flat TSMOM) paper track — recomputable series.
+
+Reuses the validated TrendRunner unchanged on the existing crypto_pit panel, then applies a CAUSAL
+portfolio vol-target overlay (the configurable book-vol knob) and a binary PageHinkleyDecay kill
+incrementally over new days, carrying overlay-buffer + detector state in the tracker's `extra`.
+Mirrors the ForwardStrategy contract: advance(last_date, extra) -> (rows, extra)."""
+from __future__ import annotations
+
+import datetime as dt
+import math
+from typing import Any, Callable
+
+import numpy as np
+
+from fxhnt.application.trend_runner import TrendRunner
+from fxhnt.domain.edge_decay import PageHinkleyDecay
+
+_EPOCH = dt.date(1970, 1, 1)
+
+
+def _today() -> str:
+    return dt.date.today().isoformat()
+
+
+def _iso(epoch_day: int) -> str:
+    return (_EPOCH + dt.timedelta(days=int(epoch_day))).isoformat()
+
+
+class TsTrendForward:
+    def __init__(self, load_panel: Callable[[], dict[str, dict[int, float]]], *,
+                 lookbacks: tuple[int, ...] = (20, 60, 120), vol_window: int = 30,
+                 internal_target_vol: float = 0.10, gross_cap: float = 1.0,
+                 cost_bps: float = 10.0, periods_per_year: int = 365,
+                 book_target_vol: float = 0.15, lev_cap: float = 3.0, vol_buf_window: int = 30,
+                 decay_lam: float = 0.05, decay_reenter_days: int = 10,
+                 clock: Callable[[], str] = _today) -> None:
+        self._load_panel = load_panel
+        self._lb = tuple(lookbacks)
+        self._vw = vol_window
+        self._itv = internal_target_vol
+        self._gcap = gross_cap
+        self._cost = cost_bps
+        self._ppy = periods_per_year
+        self._btv = book_target_vol
+        self._levcap = lev_cap
+        self._bufw = vol_buf_window
+        self._lam = decay_lam
+        self._reenter = decay_reenter_days
+        self._clock = clock
+
+    def _raw_series(self) -> tuple[list[int], list[float]]:
+        panel = self._load_panel()
+        res = TrendRunner(panel, lookbacks=self._lb, vol_window=self._vw, target_vol=self._itv,
+                          gross_cap=self._gcap, long_short=False, 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._raw_series()
+
+        if not extra:                                       # INCEPTION: warm overlay, fresh PH, book nothing
+            vol_buf = raw[-self._bufw:] if raw else []
+            ph = PageHinkleyDecay(lam=self._lam, reenter_days=self._reenter)
+            rows = [(_iso(d), r) for d, r in zip(dates, raw)]   # tracker freezes inception at latest, books none
+            new_extra = {"ph": ph.to_dict(), "vol_buf": vol_buf,
+                         "last_day": (int(dates[-1]) if dates else None)}
+            return rows, new_extra
+
+        ph = PageHinkleyDecay.from_dict(extra["ph"])
+        vol_buf = [float(x) for x in extra.get("vol_buf", [])]
+        cutoff = last_date or ""
+        btv_daily = self._btv / math.sqrt(self._ppy)
+        rows: list[tuple[str, float]] = []
+        for d, r in zip(dates, raw):
+            iso = _iso(d)
+            if iso <= cutoff:                               # already booked / pre-inception
+                continue
+            tv = float(np.std(vol_buf)) if len(vol_buf) >= 2 else 0.0
+            lev = min(self._levcap, btv_daily / tv) if tv > 1e-12 else 1.0
+            alive = ph.update(r)                            # PH consumes shadow raw return
+            rows.append((iso, lev * r if alive else 0.0))
+            vol_buf.append(r)                               # shadow buffer always advances
+            if len(vol_buf) > self._bufw:
+                vol_buf = vol_buf[-self._bufw:]
+        new_extra = {"ph": ph.to_dict(), "vol_buf": vol_buf,
+                     "last_day": (int(dates[-1]) if dates else extra.get("last_day"))}
+        return rows, new_extra

tests/integration/test_tstrend_strategy.py

@@ -0,0 +1,94 @@
+import json
+import math
+
+import numpy as np
+
+from fxhnt.application.forward_tracker import ForwardTracker
+from fxhnt.application.trend_runner import TrendRunner
+from fxhnt.application.ts_trend_strategy import TsTrendForward, _iso
+
+
+def _coin(n, start_day, drift, seed):
+    rng = np.random.default_rng(seed)
+    rets = drift + 0.01 * rng.standard_normal(n)
+    close = 100.0 * np.cumprod(1.0 + rets)
+    return {int(start_day + i): float(close[i]) for i in range(n)}
+
+
+def _panel(n=160):
+    return {f"C{j}": _coin(n, 0, 0.001, seed=j) for j in range(6)}
+
+
+def _trend_returns(panel):
+    return TrendRunner(panel, lookbacks=(20, 60, 120), vol_window=30, target_vol=0.10,
+                       gross_cap=1.0, long_short=False, cost_bps=10.0,
+                       periods_per_year=365).run()
+
+
+def test_inception_books_nothing_and_warms_state(tmp_path):
+    panel = _panel()
+    strat = TsTrendForward(lambda: panel, book_target_vol=0.15)
+    path = str(tmp_path / "st.json")
+    status = ForwardTracker(strat, path).step()
+    assert status.forward_days == 0                         # inception books nothing
+    state = json.load(open(path))
+    assert state["extra"]["ph"]["killed"] is False          # fresh, alive (history can't pre-kill)
+    assert len(state["extra"]["vol_buf"]) > 0               # overlay warm-started
+    assert state["inception"] == state["last_date"]
+
+
+def test_subsequent_books_lev_times_raw_no_drift(tmp_path):
+    panel1 = _panel()
+    holder = {"p": panel1}
+    # decay_lam high -> never kills in this test; isolates the overlay + no-drift property
+    strat = TsTrendForward(lambda: holder["p"], book_target_vol=0.15, lev_cap=3.0, decay_lam=1e9)
+    path = str(tmp_path / "st.json")
+    ForwardTracker(strat, path).step()                      # inception
+    vb = json.load(open(path))["extra"]["vol_buf"]
+    # extend the panel by exactly one day per coin
+    last = max(next(iter(panel1.values())).keys())
+    panel2 = {s: dict(series) for s, series in panel1.items()}
+    for s in panel2:
+        panel2[s][last + 1] = panel2[s][last] * 1.002
+    holder["p"] = panel2
+    status = ForwardTracker(strat, path).step()
+    assert status.booked_today == 1
+    res = _trend_returns(panel2)
+    raw_last = float(res.returns[-1])
+    date_last = _iso(int(res.dates[-1]))
+    tv = float(np.std(vb))
+    lev = min(3.0, (0.15 / math.sqrt(365)) / tv) if tv > 1e-12 else 1.0
+    booked = next(d["ret"] for d in json.load(open(path))["days"] if d["date"] == date_last)
+    assert booked == lev * raw_last                         # booked == lev*raw AND raw == TrendRunner (no drift)
+
+
+def test_killed_state_flattens_booking(tmp_path):
+    panel1 = _panel()
+    holder = {"p": panel1}
+    strat = TsTrendForward(lambda: holder["p"], book_target_vol=0.15)
+    path = str(tmp_path / "st.json")
+    ForwardTracker(strat, path).step()                      # inception
+    # force the carried PH into a killed, non-recovering state
+    state = json.load(open(path))
+    state["extra"]["ph"]["killed"] = True
+    state["extra"]["ph"]["recover"] = 0
+    json.dump(state, open(path, "w"))
+    last = max(next(iter(panel1.values())).keys())
+    panel2 = {s: dict(series) for s, series in panel1.items()}
+    for s in panel2:
+        panel2[s][last + 1] = panel2[s][last] * 0.99        # a down day (keeps it killed)
+    holder["p"] = panel2
+    status = ForwardTracker(strat, path).step()
+    assert status.booked_today == 1
+    state2 = json.load(open(path))
+    assert state2["days"][-1]["ret"] == 0.0                 # flattened while killed
+    assert len(state2["extra"]["vol_buf"]) > 0              # shadow buffer still advances
+
+
+def test_extra_json_roundtrips(tmp_path):
+    panel = _panel()
+    strat = TsTrendForward(lambda: panel, book_target_vol=0.15)
+    path = str(tmp_path / "st.json")
+    ForwardTracker(strat, path).step()
+    state = json.load(open(path))
+    json.dumps(state["extra"])                              # must not raise