497 lines
27 KiB
Python
497 lines
27 KiB
Python
"""Task 3 — `equity_allocation_inputs`: the equity analog of `honest_allocation_inputs`, wiring multistrat's
|
|
cost-netted `book_series` + real-ADV capacity curve into the allocation engine.
|
|
|
|
Step-1 finding (see module docstring in `equity_allocation_inputs.py`): `multistrat.book_series` returns an
|
|
ISO-date-string-keyed `list[tuple[str, float]]` (drop-in for `book_ret_by_day`), but `multistrat.target_weights`
|
|
returns ONLY the latest day's `{sym: weight}` snapshot — there is no weights HISTORY. `_weights_by_day`
|
|
reconstructs one by replaying `target_weights` on closes truncated day-by-day.
|
|
|
|
Also covers `first_crossing_ceiling` — the non-monotone-safe ceiling rule used for equity capacity (NOT
|
|
`allocation_engine.capacity_ceiling_from_curve`'s largest-viable-AUM rule, which would over-state the
|
|
ceiling on a curve that dips below `min_sharpe` and then ticks back up at extreme AUM).
|
|
|
|
TASK 2 (2026-07-16 UCITS re-base plan) additions: `regime="us"|"ucits"` sources each sleeve's ADV + spread
|
|
from the ACTUAL traded instrument per regime (see docs/superpowers/plans/2026-07-16-equity-ucits-rebase.md).
|
|
The pre-existing tests above are updated to pass `regime="us"` explicitly (they were written against the
|
|
US-only plumbing pre-regime; the FakeRepo fixtures they use only seed US-keyed volumes)."""
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
import math
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from fxhnt.application.equity_allocation_inputs import (
|
|
_tradeable_weight_fraction,
|
|
_weights_by_day,
|
|
equity_allocation_inputs,
|
|
first_crossing_ceiling,
|
|
)
|
|
from fxhnt.application.equity_honest import dollar_adv, equity_capacity_curve
|
|
from fxhnt.application.honest_gates import sharpe_of
|
|
from fxhnt.domain.strategies.multistrat import FUND_INSTRUMENTS, book_series
|
|
|
|
_BASE = dt.date(2026, 1, 1)
|
|
|
|
|
|
class _FakeRepo:
|
|
"""Minimal stand-in for `ForwardNavRepo` exposing the PIT reads the provider needs. TASK 2 adds
|
|
`ibkr_pit_volumes` as a SEPARATE store from `yahoo_pit_volumes` -- the `volume_source` flag switches
|
|
WHICH of the two tables the ucits branch reads, so the fake must expose two independently-seedable
|
|
tables (`volumes` for yahoo, `ibkr_volumes` for ibkr), not one shared dict."""
|
|
|
|
def __init__(self, closes: dict[str, dict[str, float]], volumes: dict[str, dict[str, float]],
|
|
ibkr_volumes: dict[str, dict[str, float]] | None = None) -> None:
|
|
self._closes = closes
|
|
self._volumes = volumes
|
|
self._ibkr_volumes = ibkr_volumes or {}
|
|
|
|
def yahoo_pit_closes(self, symbol: str) -> dict[str, float]:
|
|
return dict(self._closes.get(symbol, {}))
|
|
|
|
def yahoo_pit_volumes(self, symbol: str) -> dict[str, float]:
|
|
return dict(self._volumes.get(symbol, {}))
|
|
|
|
def ibkr_pit_volumes(self, symbol: str) -> dict[str, float]:
|
|
return dict(self._ibkr_volumes.get(symbol, {}))
|
|
|
|
|
|
def _synthetic_closes_and_volumes(days: int = 120, *, with_volumes: bool = True
|
|
) -> tuple[dict[str, dict[str, float]], dict[str, dict[str, float]]]:
|
|
"""Deterministic per-symbol price/volume paths, real enough (positive prices, sine-wave variation) that
|
|
multistrat's book_series/target_weights produce non-trivial output, and dollar-ADV huge relative to the
|
|
tested AUMs so capacity never binds (isolating the plumbing from capacity saturation). `with_volumes=False`
|
|
omits the volume snapshot entirely (simulating an empty/never-backfilled `yahoo_pit_volumes` store)."""
|
|
closes: dict[str, dict[str, float]] = {}
|
|
volumes: dict[str, dict[str, float]] = {}
|
|
for j, sym in enumerate(FUND_INSTRUMENTS):
|
|
c: dict[str, float] = {}
|
|
v: dict[str, float] = {}
|
|
price = 100.0 + 10.0 * j
|
|
for d in range(days):
|
|
date = (_BASE + dt.timedelta(days=d)).isoformat()
|
|
price *= 1.0 + 0.004 * math.sin(0.05 * d + j)
|
|
c[date] = price
|
|
if with_volumes:
|
|
v[date] = 5_000_000.0 + 200_000.0 * math.sin(0.07 * d + j)
|
|
closes[sym] = c
|
|
volumes[sym] = v
|
|
return closes, volumes
|
|
|
|
|
|
def _good_book_closes_and_volumes(days: int = 150, *, drift: float = 0.0004, noise: float = 0.002
|
|
) -> tuple[dict[str, dict[str, float]], dict[str, dict[str, float]]]:
|
|
"""A synthetic price path with a clear, low-noise positive drift so the resulting book_series/
|
|
target_weights book has an ANNUALIZED Sharpe well above the default `capacity_min_sharpe=1.0` (measured
|
|
~20 at these parameters) -- proving the ceiling math on a genuinely good book, not just isolating
|
|
plumbing from the modeled Sharpe's sign (`min_sharpe=-10` in the other tests here)."""
|
|
closes: dict[str, dict[str, float]] = {}
|
|
volumes: dict[str, dict[str, float]] = {}
|
|
for j, sym in enumerate(FUND_INSTRUMENTS):
|
|
c: dict[str, float] = {}
|
|
v: dict[str, float] = {}
|
|
price = 100.0 + 10.0 * j
|
|
for d in range(days):
|
|
date = (_BASE + dt.timedelta(days=d)).isoformat()
|
|
price *= 1.0 + drift + noise * math.sin(0.31 * d + j)
|
|
c[date] = price
|
|
v[date] = 5_000_000.0 + 200_000.0 * math.sin(0.07 * d + j)
|
|
closes[sym] = c
|
|
volumes[sym] = v
|
|
return closes, volumes
|
|
|
|
|
|
def test_equity_allocation_inputs_produces_series_and_finite_ceiling():
|
|
closes, volumes = _synthetic_closes_and_volumes(days=130)
|
|
repo = _FakeRepo(closes, volumes)
|
|
|
|
returns, ceilings = equity_allocation_inputs(
|
|
repo, {"multistrat"}, target_aum=100_000.0,
|
|
aums=[100_000, 350_000, 1_000_000, 3_000_000, 10_000_000],
|
|
min_sharpe=-10.0, # isolate plumbing from the modeled Sharpe's sign
|
|
regime="us", # this test's FakeRepo only seeds US-keyed volumes (written pre-regime)
|
|
)
|
|
|
|
assert "multistrat" in returns
|
|
series = returns["multistrat"]
|
|
assert len(series) >= 60
|
|
for k in series:
|
|
assert isinstance(k, str) and dt.date.fromisoformat(k)
|
|
|
|
assert ceilings["multistrat"] is not None
|
|
assert ceilings["multistrat"] == 10_000_000.0 # every grid AUM clears min_sharpe=-10 -> largest AUM
|
|
|
|
|
|
def test_equity_allocation_inputs_sizes_multistrat_under_the_default_min_sharpe():
|
|
"""Whole-branch-review fix: the ceiling curve used to be built with `per_period_sharpe` (raw daily
|
|
mean/std, NO annualization) and compared against `min_sharpe` on the ANNUALIZED (x sqrt(365)) scale --
|
|
~19x too small, so under the DEFAULT policy (`capacity_min_sharpe=1.0`) the curve was ALWAYS < 1.0,
|
|
`first_crossing_ceiling` broke on the very first grid AUM, and multistrat was capped to $0 regardless of
|
|
how good the book was. This test uses the DEFAULT threshold (not the -10 the other tests here use to
|
|
isolate plumbing) with a genuinely good synthetic book (measured annualized Sharpe ~20 well above 1.0)
|
|
and full ADV coverage -- proving the fixed units actually let multistrat size > $0."""
|
|
closes, volumes = _good_book_closes_and_volumes()
|
|
repo = _FakeRepo(closes, volumes)
|
|
|
|
returns, ceilings = equity_allocation_inputs(
|
|
repo, {"multistrat"}, target_aum=100_000.0,
|
|
aums=[100_000, 350_000, 1_000_000, 3_000_000, 10_000_000],
|
|
min_sharpe=1.0, # the production default (`AllocationPolicy.capacity_min_sharpe`)
|
|
regime="us", # this test's FakeRepo only seeds US-keyed volumes (written pre-regime)
|
|
)
|
|
|
|
assert "multistrat" in returns
|
|
assert len(returns["multistrat"]) > 0
|
|
# a non-empty, non-zero ceiling: either a finite AUM > 0, or None (== unbounded) -- NEVER 0.0, which
|
|
# would mean `target_capital` caps multistrat's dollars to $0 (the exact bug this test guards against).
|
|
ceiling = ceilings["multistrat"]
|
|
assert ceiling is None or ceiling > 0.0
|
|
|
|
|
|
def test_equity_allocation_inputs_insufficient_adv_coverage_falls_back():
|
|
"""Whole-branch-review BUG 2: before `backfill-etf-volume` runs (or if the nightly volume snapshot keeps
|
|
failing -- swallow-and-warn), `yahoo_pit_volumes` is empty. The provider's pre-fix guard only checked
|
|
`book_ret_by_day` (derived from closes, never volumes), so an empty ADV store still emitted a full-length
|
|
series where every day is scaled to zero capacity and books `-cost` -- a GARBAGE series that (being
|
|
non-empty) would be used INSTEAD of the nav fallback. The fixed provider must instead emit NO series (and
|
|
ceiling `None`), so `record_allocation`'s `honest_returns.get(sid) or nav_fallback` falls back to nav.
|
|
(Still true under the partial-sizing `_tradeable_weight_fraction` guard: with NO volumes at all, every
|
|
sleeve is dead, so tradeable weight is ~0.0 -- well below `_MIN_TRADEABLE_WEIGHT` -- a fully-degenerate
|
|
store, not a single-thin-sleeve partial-coverage case.)"""
|
|
closes, volumes = _synthetic_closes_and_volumes(days=130, with_volumes=False)
|
|
repo = _FakeRepo(closes, volumes)
|
|
|
|
returns, ceilings = equity_allocation_inputs(
|
|
repo, {"multistrat"}, target_aum=100_000.0,
|
|
aums=[100_000, 350_000, 1_000_000, 3_000_000, 10_000_000],
|
|
min_sharpe=1.0,
|
|
regime="us", # this test's FakeRepo only seeds US-keyed volumes (written pre-regime)
|
|
)
|
|
|
|
assert "multistrat" not in returns
|
|
assert ceilings["multistrat"] is None
|
|
|
|
|
|
def _ucits_map(*, half_spread_bps: float = 5.0, ticker_suffix: str = "_UCITS.L"
|
|
) -> dict[str, SimpleNamespace]:
|
|
"""A minimal stand-in for `settings.ucits.map` — only the two attributes
|
|
`equity_allocation_inputs` reads (`.yahoo_ticker`, `.half_spread_bps`)."""
|
|
return {sym: SimpleNamespace(yahoo_ticker=f"{sym}{ticker_suffix}", half_spread_bps=half_spread_bps)
|
|
for sym in FUND_INSTRUMENTS}
|
|
|
|
|
|
def test_equity_allocation_inputs_regime_us_is_byte_identical_to_pre_regime_output():
|
|
"""TASK 2 regression (Global Constraint: byte-identity of regime="us"): independently reconstructs the
|
|
EXACT pre-Task-2 computation (US-only ADV via `dollar_adv`/`equity_capacity_curve` with `spread_bps=None`
|
|
-> the module-constant `ETF_HALF_SPREAD_BPS` fallback -- the only code path that existed before this
|
|
task) in THIS test process, then asserts `equity_allocation_inputs(..., regime="us")` reproduces it
|
|
EXACTLY (`==`, not `math.isclose`). Deliberately NOT hardcoded literal floats: the cost loop's
|
|
`set(cur) | set(prev)` iterates in PYTHONHASHSEED-dependent order, so the exact float bits are only
|
|
reproducible WITHIN one process, not across separate interpreter invocations -- comparing two
|
|
same-process computations is the correct byte-identity check, a golden literal captured from a
|
|
different process is not."""
|
|
closes, volumes = _good_book_closes_and_volumes()
|
|
repo = _FakeRepo(closes, volumes)
|
|
aums = [100_000, 350_000, 1_000_000, 3_000_000, 10_000_000]
|
|
|
|
us_closes = {sym: repo.yahoo_pit_closes(sym) for sym in FUND_INSTRUMENTS}
|
|
us_volumes = {sym: repo.yahoo_pit_volumes(sym) for sym in FUND_INSTRUMENTS}
|
|
book_ret_by_day = dict(book_series(us_closes, FUND_INSTRUMENTS, drop_incomplete_today=False))
|
|
weights_by_day = _weights_by_day(us_closes, FUND_INSTRUMENTS)
|
|
adv = dollar_adv(us_closes, us_volumes)
|
|
all_aums = sorted({100_000.0, *(float(a) for a in aums)})
|
|
curve = equity_capacity_curve(book_ret_by_day, weights_by_day, adv, aums=all_aums, price_by_day=us_closes)
|
|
expected_series = curve[100_000.0]
|
|
expected_sharpe = {float(a): sharpe_of(curve[float(a)]) for a in aums}
|
|
expected_ceiling = first_crossing_ceiling(expected_sharpe, 1.0)
|
|
assert len(expected_series) >= 60 # sanity: this really is the good book, not an empty fixture
|
|
|
|
returns, ceilings = equity_allocation_inputs(
|
|
repo, {"multistrat"}, target_aum=100_000.0, aums=aums, min_sharpe=1.0, regime="us",
|
|
)
|
|
|
|
assert returns["multistrat"] == expected_series
|
|
assert ceilings["multistrat"] == expected_ceiling
|
|
|
|
|
|
def test_equity_allocation_inputs_ucits_ceiling_materially_lower_than_us_when_ucits_volume_thin():
|
|
"""TASK 2: with a HIGH US volume (deep US-ETF liquidity) but a LOW (thin, not absent) UCITS volume for
|
|
every sleeve, the `ucits` regime's capacity ceiling must be MATERIALLY lower than the `us` regime's --
|
|
proving the re-base actually re-sources capacity onto the thinner UCITS lines rather than silently
|
|
keeping the US ADV. Both regimes read the SAME closes/book -- only the ADV source differs."""
|
|
closes, us_volumes = _good_book_closes_and_volumes() # HIGH US volume (~5,000,000 shares/day)
|
|
ucits_map = _ucits_map(half_spread_bps=5.0)
|
|
# LOW UCITS volume: 1/10th of the US volume for every sleeve, every day -- thin but never absent (every
|
|
# day > 0, so the ADV-coverage guard still PASSES; see the thin/absent test below for the guard itself
|
|
# tripping). Measured: this alone drops the ceiling from the US grid's max ($10M) to $1M -- an order of
|
|
# magnitude, well within the plan's expected $0.5-3M UCITS-real band -- while a 1/500th cut collapses it
|
|
# all the way to $0 (still "materially lower" but less illustrative of a partial, non-degenerate ceiling).
|
|
repo_volumes = dict(us_volumes)
|
|
for sym in FUND_INSTRUMENTS:
|
|
repo_volumes[ucits_map[sym].yahoo_ticker] = {d: v / 10.0 for d, v in us_volumes[sym].items()}
|
|
repo = _FakeRepo(closes, repo_volumes)
|
|
|
|
aums = [100_000, 350_000, 1_000_000, 3_000_000, 10_000_000]
|
|
_, ceilings_us = equity_allocation_inputs(
|
|
repo, {"multistrat"}, target_aum=100_000.0, aums=aums, min_sharpe=1.0, regime="us")
|
|
_, ceilings_ucits = equity_allocation_inputs(
|
|
repo, {"multistrat"}, target_aum=100_000.0, aums=aums, min_sharpe=1.0,
|
|
regime="ucits", ucits_map=ucits_map)
|
|
|
|
ceiling_us = ceilings_us["multistrat"]
|
|
ceiling_ucits = ceilings_ucits["multistrat"]
|
|
assert ceiling_us == 10_000_000.0 # deep US liquidity clears min_sharpe at every grid AUM
|
|
assert ceiling_ucits is not None
|
|
# materially lower -- at least an order of magnitude below the US ceiling (proves the re-base bites,
|
|
# not just a marginal wobble from the different spread) -- AND non-zero, a genuine partial ceiling, not
|
|
# a full collapse (the coverage-guard test below covers the "no capacity at all" case).
|
|
assert 0.0 < ceiling_ucits <= ceiling_us / 10.0
|
|
|
|
|
|
def test_equity_allocation_inputs_ucits_volume_source_yahoo_is_byte_identical_to_direct_yahoo_read():
|
|
"""TASK 2 Global Constraint (byte-identity of `volume_source="yahoo"`, the default): independently
|
|
reconstructs the ucits-regime computation using `repo.yahoo_pit_volumes` directly (the ONLY code path
|
|
that existed before this task), then asserts `equity_allocation_inputs(..., volume_source="yahoo")`
|
|
reproduces it EXACTLY (`==`) -- proving the new flag is additive for the default, not a behavior change."""
|
|
closes, us_volumes = _good_book_closes_and_volumes()
|
|
ucits_map = _ucits_map(half_spread_bps=5.0)
|
|
repo_volumes = dict(us_volumes)
|
|
for sym in FUND_INSTRUMENTS:
|
|
repo_volumes[ucits_map[sym].yahoo_ticker] = {d: v / 10.0 for d, v in us_volumes[sym].items()}
|
|
repo = _FakeRepo(closes, repo_volumes)
|
|
aums = [100_000, 350_000, 1_000_000, 3_000_000, 10_000_000]
|
|
|
|
ucits_closes = {sym: repo.yahoo_pit_closes(sym) for sym in FUND_INSTRUMENTS}
|
|
book_ret_by_day = dict(book_series(ucits_closes, FUND_INSTRUMENTS, drop_incomplete_today=False))
|
|
weights_by_day = _weights_by_day(ucits_closes, FUND_INSTRUMENTS)
|
|
ucits_volumes = {sym: repo.yahoo_pit_volumes(ucits_map[sym].yahoo_ticker) for sym in FUND_INSTRUMENTS}
|
|
adv = dollar_adv(ucits_closes, ucits_volumes)
|
|
spread = {sym: ucits_map[sym].half_spread_bps for sym in FUND_INSTRUMENTS}
|
|
all_aums = sorted({100_000.0, *(float(a) for a in aums)})
|
|
curve = equity_capacity_curve(book_ret_by_day, weights_by_day, adv, aums=all_aums,
|
|
price_by_day=ucits_closes, spread_bps=spread)
|
|
expected_series = curve[100_000.0]
|
|
expected_sharpe = {float(a): sharpe_of(curve[float(a)]) for a in aums}
|
|
expected_ceiling = first_crossing_ceiling(expected_sharpe, 1.0)
|
|
assert len(expected_series) >= 60 # sanity: this really is the good book, not an empty fixture
|
|
|
|
returns, ceilings = equity_allocation_inputs(
|
|
repo, {"multistrat"}, target_aum=100_000.0, aums=aums, min_sharpe=1.0,
|
|
regime="ucits", ucits_map=ucits_map, volume_source="yahoo",
|
|
)
|
|
|
|
assert returns["multistrat"] == expected_series
|
|
assert ceilings["multistrat"] == expected_ceiling
|
|
|
|
|
|
def test_equity_allocation_inputs_ucits_volume_source_ibkr_reads_ibkr_pit_volumes_not_yahoo():
|
|
"""TASK 2: `volume_source="ibkr"` must read `repo.ibkr_pit_volumes`, NOT `yahoo_pit_volumes` -- seed the
|
|
fake repo with MATERIALLY DIFFERENT yahoo vs ibkr volume for the same UCITS ticker (yahoo 1/10th of US
|
|
volume -- thin but a real, non-collapsing ceiling per the test above; ibkr 1/500th -- thin enough to
|
|
collapse capacity to $0, per the measured ratios already established in the `us`-vs-`ucits` test above)
|
|
and assert the resulting ceiling actually DIFFERS by source, proving the flag switches which table is
|
|
read rather than being accepted and silently ignored."""
|
|
closes, us_volumes = _good_book_closes_and_volumes()
|
|
ucits_map = _ucits_map(half_spread_bps=5.0)
|
|
yahoo_volumes = dict(us_volumes)
|
|
for sym in FUND_INSTRUMENTS:
|
|
yahoo_volumes[ucits_map[sym].yahoo_ticker] = {d: v / 10.0 for d, v in us_volumes[sym].items()}
|
|
ibkr_volumes = {ucits_map[sym].yahoo_ticker: {d: v / 500.0 for d, v in us_volumes[sym].items()}
|
|
for sym in FUND_INSTRUMENTS}
|
|
repo = _FakeRepo(closes, yahoo_volumes, ibkr_volumes=ibkr_volumes)
|
|
|
|
aums = [100_000, 350_000, 1_000_000, 3_000_000, 10_000_000]
|
|
_, ceilings_yahoo = equity_allocation_inputs(
|
|
repo, {"multistrat"}, target_aum=100_000.0, aums=aums, min_sharpe=1.0,
|
|
regime="ucits", ucits_map=ucits_map, volume_source="yahoo")
|
|
_, ceilings_ibkr = equity_allocation_inputs(
|
|
repo, {"multistrat"}, target_aum=100_000.0, aums=aums, min_sharpe=1.0,
|
|
regime="ucits", ucits_map=ucits_map, volume_source="ibkr")
|
|
|
|
ceiling_yahoo = ceilings_yahoo["multistrat"]
|
|
ceiling_ibkr = ceilings_ibkr["multistrat"]
|
|
assert ceiling_yahoo is not None and ceiling_yahoo > 0.0
|
|
# ibkr's volume is 50x thinner than yahoo's here -> the ibkr-sourced ceiling must be materially lower
|
|
# (None or 0.0 both count as "lower" -- a full capacity collapse is still proof the switch bit).
|
|
assert ceiling_ibkr is None or ceiling_ibkr < ceiling_yahoo
|
|
|
|
|
|
def test_equity_allocation_inputs_unknown_volume_source_raises_instead_of_silently_defaulting():
|
|
"""`volume_source` is a plain string, so a typo must NOT silently fall into either read path -- fail
|
|
loud, mirroring the existing `regime` guard."""
|
|
closes, volumes = _synthetic_closes_and_volumes(days=120)
|
|
repo = _FakeRepo(closes, volumes)
|
|
|
|
with pytest.raises(ValueError, match="unknown ucits_volume_source"):
|
|
equity_allocation_inputs(repo, {"multistrat"}, target_aum=100_000.0, aums=[100_000],
|
|
regime="ucits", volume_source="bogus")
|
|
|
|
|
|
def test_equity_allocation_inputs_ucits_one_dead_sleeve_still_emits_partial_sizing():
|
|
"""PARTIAL-SIZING (weight-based guard): an ABSENT UCITS line for exactly ONE sleeve (e.g. DBMF.L -- Yahoo
|
|
lacks the new/small UCITS listing, or the LSE volume snapshot hasn't backfilled it yet) must NOT drop the
|
|
whole book. `dollar_adv` on an empty volume dict yields an empty ADV series for DBMF -> every traded day
|
|
holding DBMF has that ONE sleeve's weight untradeable, but the other four (most of the book's gross
|
|
weight) remain tradeable -> `_tradeable_weight_fraction` averages to ~0.8, comfortably ABOVE
|
|
`_MIN_TRADEABLE_WEIGHT` (0.5) -> the provider still EMITS a series (the dead DBMF sleeve gets scaled to
|
|
~0 by `equity_honest_returns`'s per-sleeve participation cap instead — "use as much as there is"), unlike
|
|
the prior all-or-nothing guard which would have zeroed the entire book over one dead sleeve."""
|
|
closes, us_volumes = _good_book_closes_and_volumes()
|
|
ucits_map = _ucits_map(half_spread_bps=30.0)
|
|
repo_volumes = dict(us_volumes)
|
|
for sym in FUND_INSTRUMENTS:
|
|
# every UCITS line present and liquid EXCEPT DBMF, which Yahoo has no data for at all (empty dict).
|
|
repo_volumes[ucits_map[sym].yahoo_ticker] = (
|
|
{} if sym == "DBMF" else {d: v / 10.0 for d, v in us_volumes[sym].items()}
|
|
)
|
|
repo = _FakeRepo(closes, repo_volumes)
|
|
|
|
returns, ceilings = equity_allocation_inputs(
|
|
repo, {"multistrat"}, target_aum=100_000.0,
|
|
aums=[100_000, 350_000, 1_000_000, 3_000_000, 10_000_000],
|
|
min_sharpe=1.0, regime="ucits", ucits_map=ucits_map,
|
|
)
|
|
|
|
assert "multistrat" in returns
|
|
assert len(returns["multistrat"]) > 0
|
|
assert ceilings["multistrat"] is None or ceilings["multistrat"] >= 0.0
|
|
|
|
|
|
def test_equity_allocation_inputs_ucits_all_sleeves_dead_falls_back_to_nav():
|
|
"""DEGENERATE case preserved: with EVERY UCITS line absent (an empty/pre-backfill volume store), the
|
|
tradeable-weight fraction is ~0.0 -- well below `_MIN_TRADEABLE_WEIGHT` -- so the provider must still emit
|
|
NO series (ceiling `None`, nav fallback), same as the pre-existing US-side insufficient-coverage test.
|
|
This is the genuinely degenerate case the guard exists for, distinct from the one-dead-sleeve partial
|
|
case above."""
|
|
closes, us_volumes = _good_book_closes_and_volumes()
|
|
ucits_map = _ucits_map(half_spread_bps=30.0)
|
|
repo_volumes = dict(us_volumes)
|
|
for sym in FUND_INSTRUMENTS:
|
|
repo_volumes[ucits_map[sym].yahoo_ticker] = {} # every UCITS line absent
|
|
repo = _FakeRepo(closes, repo_volumes)
|
|
|
|
returns, ceilings = equity_allocation_inputs(
|
|
repo, {"multistrat"}, target_aum=100_000.0,
|
|
aums=[100_000, 350_000, 1_000_000, 3_000_000, 10_000_000],
|
|
min_sharpe=1.0, regime="ucits", ucits_map=ucits_map,
|
|
)
|
|
|
|
assert "multistrat" not in returns
|
|
assert ceilings["multistrat"] is None
|
|
|
|
|
|
def test_tradeable_weight_fraction_one_dead_sleeve_is_about_point_eight():
|
|
"""Direct unit test of the weight-based guard's core metric: 5 equally-weighted (20% each) sleeves held
|
|
every day, ONE of which (E) has zero ADV every day it's held -- the tradeable fraction must be the
|
|
tradeable GROSS WEIGHT share (4/5 = 0.8), not 0 (the prior all-symbols-covered check would have failed
|
|
every day outright)."""
|
|
weights_by_day = {
|
|
"2026-01-01": {"A": 0.2, "B": 0.2, "C": 0.2, "D": 0.2, "E": 0.2},
|
|
"2026-01-02": {"A": 0.2, "B": 0.2, "C": 0.2, "D": 0.2, "E": 0.2},
|
|
}
|
|
adv = {
|
|
"A": {"2026-01-01": 1_000.0, "2026-01-02": 1_000.0},
|
|
"B": {"2026-01-01": 1_000.0, "2026-01-02": 1_000.0},
|
|
"C": {"2026-01-01": 1_000.0, "2026-01-02": 1_000.0},
|
|
"D": {"2026-01-01": 1_000.0, "2026-01-02": 1_000.0},
|
|
"E": {"2026-01-01": 0.0, "2026-01-02": 0.0}, # dead sleeve, every day
|
|
}
|
|
|
|
assert _tradeable_weight_fraction(weights_by_day, adv) == pytest.approx(0.8)
|
|
|
|
|
|
def test_tradeable_weight_fraction_skips_empty_weight_rows_and_handles_no_traded_days():
|
|
# A day with no weights (pre-warmup) is skipped, not counted as untradeable.
|
|
weights_by_day = {"2026-01-01": {}, "2026-01-02": {"A": 0.5, "B": 0.5}}
|
|
adv = {"A": {"2026-01-02": 1_000.0}, "B": {"2026-01-02": 1_000.0}}
|
|
assert _tradeable_weight_fraction(weights_by_day, adv) == pytest.approx(1.0)
|
|
|
|
# No traded days at all -> 0.0 (the safe default), not vacuously 1.0.
|
|
assert _tradeable_weight_fraction({}, {}) == 0.0
|
|
assert _tradeable_weight_fraction({"2026-01-01": {}}, {}) == 0.0
|
|
|
|
|
|
def test_equity_allocation_inputs_empty_deploy_sids_is_noop():
|
|
closes, volumes = _synthetic_closes_and_volumes(days=120)
|
|
repo = _FakeRepo(closes, volumes)
|
|
|
|
returns, ceilings = equity_allocation_inputs(repo, set(), target_aum=100_000.0, aums=[100_000, 1_000_000])
|
|
|
|
assert returns == {}
|
|
assert ceilings == {}
|
|
|
|
|
|
def test_equity_allocation_inputs_absent_pit_data_gives_no_series_and_none_ceiling():
|
|
repo = _FakeRepo({sym: {} for sym in FUND_INSTRUMENTS}, {sym: {} for sym in FUND_INSTRUMENTS})
|
|
|
|
returns, ceilings = equity_allocation_inputs(repo, {"multistrat"}, target_aum=100_000.0, aums=[100_000])
|
|
|
|
assert "multistrat" not in returns
|
|
assert ceilings["multistrat"] is None
|
|
|
|
|
|
def test_equity_allocation_inputs_short_pit_history_gives_no_series():
|
|
# Only 10 days -- well short of the ~66-day warmup book_series/target_weights need.
|
|
closes, volumes = _synthetic_closes_and_volumes(days=10)
|
|
repo = _FakeRepo(closes, volumes)
|
|
|
|
returns, ceilings = equity_allocation_inputs(repo, {"multistrat"}, target_aum=100_000.0, aums=[100_000])
|
|
|
|
assert "multistrat" not in returns
|
|
assert ceilings["multistrat"] is None
|
|
|
|
|
|
def test_equity_allocation_inputs_unknown_regime_raises_instead_of_silently_defaulting_to_us():
|
|
"""`regime` is a plain string, so a typo like `"US"` (wrong case) must NOT silently fall into the
|
|
aggressive `us` branch (deep US-liquidity capacity model) -- dangerous in the EU retail default. Fail
|
|
loud instead."""
|
|
closes, volumes = _synthetic_closes_and_volumes(days=120)
|
|
repo = _FakeRepo(closes, volumes)
|
|
|
|
with pytest.raises(ValueError, match="unknown equity_execution_regime"):
|
|
equity_allocation_inputs(repo, {"multistrat"}, target_aum=100_000.0, aums=[100_000], regime="US")
|
|
|
|
|
|
@pytest.mark.parametrize("regime", ["ucits", "us"])
|
|
def test_equity_allocation_inputs_valid_regimes_are_accepted(regime):
|
|
closes, volumes = _synthetic_closes_and_volumes(days=120)
|
|
repo = _FakeRepo(closes, volumes)
|
|
ucits_map = {
|
|
sym: SimpleNamespace(yahoo_ticker=sym, half_spread_bps=10.0) for sym in FUND_INSTRUMENTS
|
|
}
|
|
|
|
# Must not raise -- both real regime values are accepted.
|
|
equity_allocation_inputs(repo, {"multistrat"}, target_aum=100_000.0, aums=[100_000],
|
|
regime=regime, ucits_map=ucits_map)
|
|
|
|
|
|
def test_first_crossing_ceiling_empty_curve_is_none():
|
|
assert first_crossing_ceiling({}, 1.0) is None
|
|
|
|
|
|
def test_first_crossing_ceiling_all_viable_picks_largest():
|
|
curve = {100_000.0: 1.5, 350_000.0: 1.3, 1_000_000.0: 1.1}
|
|
assert first_crossing_ceiling(curve, 1.0) == 1_000_000.0
|
|
|
|
|
|
def test_first_crossing_ceiling_stops_at_first_dip_not_largest_viable():
|
|
"""The critical robustness fix: a curve that dips below min_sharpe and then ticks BACK UP at extreme AUM
|
|
(the equity scale_hit approximation's known non-monotone tail) must NOT report the later, spuriously
|
|
viable AUM as the ceiling -- it must stop at the AUM just before the first dip."""
|
|
curve = {
|
|
100_000.0: 1.5,
|
|
350_000.0: 1.2,
|
|
1_000_000.0: 0.5, # first dip below min_sharpe=1.0
|
|
3_000_000.0: 0.3,
|
|
10_000_000.0: 1.8, # ticks back up -- capacity_ceiling_from_curve would (wrongly) pick this
|
|
}
|
|
ceiling = first_crossing_ceiling(curve, 1.0)
|
|
assert ceiling == 350_000.0
|
|
assert ceiling != 10_000_000.0
|
|
|
|
|
|
def test_first_crossing_ceiling_none_viable_is_zero():
|
|
curve = {100_000.0: 0.2, 350_000.0: 0.1}
|
|
assert first_crossing_ceiling(curve, 1.0) == 0.0
|