119 lines
6.4 KiB
Python
119 lines
6.4 KiB
Python
"""Task 5 — `record_allocation` honest-reference sizing + per-strategy capacity cap.
|
|
|
|
Two paths are exercised:
|
|
(a) `store=None` → today's behavior exactly (nav_history sizing, no crash);
|
|
(b) `store=<fake>` → sizing comes from the honest backtest series (long, so the allocation is non-empty even
|
|
though the forward nav is SHORTER than `min_history_days`) and each `StrategyAllocationRow`'s dollars are
|
|
capped at the per-strategy capacity ceiling.
|
|
|
|
The full honest-reference engine (`honest_allocation_inputs`) is monkeypatched to a controlled
|
|
`(returns_by_strategy, ceilings)` so the test is fast and deterministic and does NOT need a live warehouse —
|
|
the store is a plain sentinel; what matters is that `record_allocation` calls the provider when a store is
|
|
present, sizes from its series, and applies its ceiling."""
|
|
import datetime as dt
|
|
|
|
from fxhnt import registry
|
|
from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
|
|
from fxhnt.application import allocation_ingest
|
|
from fxhnt.application.allocation_ingest import record_allocation
|
|
from fxhnt.application.forward_models import ForwardNavRow as NavRowDTO
|
|
|
|
|
|
def _seed_deploy(repo, sid, ndays, at, base=dt.date(2026, 6, 1)):
|
|
"""Seed a deploy-tier strategy (gate_spec {} → non-reconciliation → always funded) with `ndays` of nav."""
|
|
registry_row = {"display_name": sid.upper(), "sleeve": "x", "gate_spec": {}, "tier": "deploy"}
|
|
return registry_row, [
|
|
NavRowDTO(strategy_id=sid, date=(base + dt.timedelta(days=i)).isoformat(),
|
|
ret=(0.01 if (i % 2) else -0.008), nav=1.0)
|
|
for i in range(ndays)
|
|
]
|
|
|
|
|
|
def test_store_none_matches_nav_history_path(tmp_path, monkeypatch):
|
|
"""(a) store=None → nav_history sizing, no crash, and the allocation is recorded (existing behavior)."""
|
|
dsn = f"sqlite:///{tmp_path}/none.db"
|
|
repo = ForwardNavRepo(dsn)
|
|
repo.migrate()
|
|
at = dt.datetime(2026, 7, 13, 23, 30)
|
|
for sid, flip in (("d1", 1), ("d2", 0)):
|
|
row, _ = _seed_deploy(repo, sid, 60, at)
|
|
monkeypatch.setitem(registry.STRATEGY_REGISTRY, sid, row)
|
|
for i in range(60):
|
|
d = (dt.date(2026, 6, 1) + dt.timedelta(days=i)).isoformat()
|
|
repo.upsert_rows([NavRowDTO(strategy_id=sid, date=d,
|
|
ret=(0.01 if (i % 2) == flip else -0.008), nav=1.0)], at=at)
|
|
out = record_allocation(dsn, equity=20_000.0) # store omitted → nav_history fallback (today's behavior)
|
|
assert set(out) <= {"d1", "d2"}
|
|
assert repo.current_allocation() == out # recorded, no crash
|
|
|
|
|
|
def test_store_sizes_from_honest_series_and_caps_at_capacity(tmp_path, monkeypatch):
|
|
"""(b) with a store, sizing comes from the honest series and dollars are capped at the capacity ceiling.
|
|
|
|
The forward nav is only 10 days (< min_history_days=60), so a nav_history-sized book would be EMPTY; the
|
|
non-empty allocation proves the sizing came from the (400-day) honest series. equity=$1M so the fund
|
|
ceiling ($1M paper) does not bind — the $100k per-strategy capacity ceiling is what caps the strategy."""
|
|
sid = "cap_crypto"
|
|
dsn = f"sqlite:///{tmp_path}/honest.db"
|
|
repo = ForwardNavRepo(dsn)
|
|
repo.migrate()
|
|
at = dt.datetime(2026, 7, 13, 23, 30)
|
|
row, _ = _seed_deploy(repo, sid, 10, at) # SHORT forward nav (< 60) on purpose
|
|
monkeypatch.setitem(registry.STRATEGY_REGISTRY, sid, row)
|
|
for i in range(10):
|
|
d = (dt.date(2026, 6, 1) + dt.timedelta(days=i)).isoformat()
|
|
repo.upsert_rows([NavRowDTO(strategy_id=sid, date=d, ret=(0.002 if i % 2 else -0.002), nav=1.0)], at=at)
|
|
|
|
long_series = {str(i): (0.002 if i % 2 else -0.002) for i in range(400)} # long, stable honest backtest
|
|
captured = {}
|
|
|
|
def _fake_inputs(store, spread_bps, unlock_events, *, deploy_sids, target_aum, aums, **kw):
|
|
captured["store"] = store
|
|
captured["deploy_sids"] = set(deploy_sids)
|
|
captured["target_aum"] = target_aum
|
|
return {sid: long_series}, {sid: 100_000.0}
|
|
|
|
monkeypatch.setattr(allocation_ingest, "honest_allocation_inputs", _fake_inputs)
|
|
|
|
sentinel_store = object()
|
|
out = record_allocation(dsn, equity=1_000_000.0, store=sentinel_store)
|
|
|
|
assert captured["store"] is sentinel_store # provider called with the injected store
|
|
assert sid in captured["deploy_sids"] # funded deploy set passed through
|
|
assert captured["target_aum"] == 1_000_000.0
|
|
# sizing came from the honest series: non-empty despite the 10-day forward nav being < min_history_days.
|
|
assert out.get(sid, 0.0) > 0.0
|
|
# capacity cap binds: the recorded dollars respect the $100k per-strategy ceiling.
|
|
assert 0.0 < out[sid] <= 100_000.0 + 1e-6
|
|
rows = repo.all_allocations()
|
|
assert rows and all(abs(r.target_dollars) <= 100_000.0 + 1e-6 for r in rows if r.strategy_id == sid)
|
|
|
|
|
|
def test_killswitch_runs_on_forward_nav_not_honest_backtest(tmp_path, monkeypatch):
|
|
"""Fix A end-to-end: the honest backtest sizing series carries a deep (>kill_dd=0.25) drawdown that WOULD
|
|
latch the killswitch and flatten the allocation to $0. But the killswitch is LIVE protection, so
|
|
`record_allocation` runs it on the FORWARD nav record (short + benign) — so the allocation stays NON-zero.
|
|
Without the fix (killswitch on the honest series) this allocation would be all-zero."""
|
|
sid = "ks_crypto"
|
|
dsn = f"sqlite:///{tmp_path}/ks.db"
|
|
repo = ForwardNavRepo(dsn)
|
|
repo.migrate()
|
|
at = dt.datetime(2026, 7, 13, 23, 30)
|
|
row, _ = _seed_deploy(repo, sid, 8, at) # SHORT + benign forward nav (no drawdown)
|
|
monkeypatch.setitem(registry.STRATEGY_REGISTRY, sid, row)
|
|
for i in range(8):
|
|
d = (dt.date(2026, 6, 1) + dt.timedelta(days=i)).isoformat()
|
|
repo.upsert_rows([NavRowDTO(strategy_id=sid, date=d, ret=(0.002 if i % 2 else -0.001), nav=1.0)], at=at)
|
|
|
|
# honest sizing series: a deep early crash (12x -0.05 -> ~46% drawdown, trips kill_dd=0.25) then benign.
|
|
crash_honest = {str(i): (-0.05 if i < 12 else (0.002 if i % 2 else -0.002)) for i in range(400)}
|
|
|
|
def _fake_inputs(store, spread_bps, unlock_events, *, deploy_sids, target_aum, aums, **kw):
|
|
return {sid: crash_honest}, {sid: None}
|
|
|
|
monkeypatch.setattr(allocation_ingest, "honest_allocation_inputs", _fake_inputs)
|
|
|
|
out = record_allocation(dsn, equity=1_000_000.0, store=object())
|
|
# killswitch ran on the benign forward nav (not the crashy honest series) -> NOT latched -> non-zero.
|
|
assert out.get(sid, 0.0) != 0.0
|