60 lines
2.5 KiB
Python
60 lines
2.5 KiB
Python
"""Generic ForwardTracker: first-run freeze, forward-only booking, idempotency, extra round-trip."""
|
|
from __future__ import annotations
|
|
|
|
from fxhnt.application.forward_tracker import ForwardTracker
|
|
|
|
|
|
class _Recomputable:
|
|
"""Recomputable-series strategy: always returns the full known series."""
|
|
def __init__(self, series: list[tuple[str, float]]) -> None:
|
|
self._series = series
|
|
|
|
def advance(self, last_date, extra):
|
|
return list(self._series), extra
|
|
|
|
|
|
class _LiveBooking:
|
|
"""Live-booking strategy: books one row for `today`, carries a counter in extra."""
|
|
def __init__(self, today: str, ret: float) -> None:
|
|
self._today, self._ret = today, ret
|
|
|
|
def advance(self, last_date, extra):
|
|
n = int(extra.get("runs", 0)) + 1
|
|
return [(self._today, self._ret)], {"runs": n}
|
|
|
|
|
|
def test_first_run_freezes_and_books_nothing(tmp_path) -> None:
|
|
p = str(tmp_path / "s.json")
|
|
st = ForwardTracker(_Recomputable([("2026-01-01", 0.01), ("2026-01-02", 0.02)]), p).step()
|
|
assert st.forward_days == 0 # freeze: nothing booked on first run
|
|
assert st.inception == "2026-01-02" # inception = latest known date
|
|
assert st.last_date == "2026-01-02"
|
|
|
|
|
|
def test_second_run_books_only_forward_days(tmp_path) -> None:
|
|
p = str(tmp_path / "s.json")
|
|
strat = _Recomputable([("2026-01-01", 0.01), ("2026-01-02", 0.02)])
|
|
ForwardTracker(strat, p).step() # freeze at 2026-01-02
|
|
strat._series.append(("2026-01-03", 0.05)) # a new day appears
|
|
st = ForwardTracker(strat, p).step()
|
|
assert st.forward_days == 1 # only 2026-01-03 booked
|
|
assert abs(st.forward_return_pct - 5.0) < 1e-9
|
|
assert st.last_date == "2026-01-03"
|
|
|
|
|
|
def test_idempotent_rerun_books_nothing(tmp_path) -> None:
|
|
p = str(tmp_path / "s.json")
|
|
strat = _Recomputable([("2026-01-01", 0.01), ("2026-01-02", 0.02), ("2026-01-03", 0.05)])
|
|
ForwardTracker(strat, p).step() # freeze at 2026-01-03
|
|
st = ForwardTracker(strat, p).step() # nothing new
|
|
assert st.forward_days == 0
|
|
|
|
|
|
def test_extra_carry_state_round_trips(tmp_path) -> None:
|
|
p = str(tmp_path / "s.json")
|
|
ForwardTracker(_LiveBooking("2026-01-01", 0.0), p).step() # freeze, extra={"runs":1}
|
|
ForwardTracker(_LiveBooking("2026-01-02", 0.01), p).step() # books 2026-01-02
|
|
import json
|
|
extra = json.load(open(p))["extra"]
|
|
assert extra["runs"] == 2 # advance saw prior extra and incremented
|