211 lines
8.1 KiB
Python
211 lines
8.1 KiB
Python
"""Explainable, constrained portfolio tilts from historical product metrics."""
|
|
|
|
from dataclasses import dataclass
|
|
from decimal import ROUND_HALF_UP, Decimal
|
|
|
|
_HUNDRED = Decimal("100")
|
|
_ZERO = Decimal()
|
|
_ASSET_CLASSES = ("cash_management_etf", "bond_etf", "equity_etf")
|
|
_LIQUIDITY_FLOORS = {
|
|
"daily": Decimal("10000000"),
|
|
"within_7_days": Decimal("3000000"),
|
|
"within_30_days": Decimal("500000"),
|
|
"over_30_days": Decimal(),
|
|
}
|
|
_CASH_MINIMUMS = {
|
|
"daily": Decimal("25"),
|
|
"within_7_days": Decimal("15"),
|
|
"within_30_days": Decimal("5"),
|
|
"over_30_days": Decimal(),
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AssetClassMarketMetric:
|
|
asset_class: str
|
|
trailing_120d_return_pct: Decimal
|
|
max_drawdown_pct: Decimal
|
|
average_daily_turnover_amount: Decimal
|
|
product_count: int
|
|
liquidity_source: str = "historical_turnover"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DynamicAllocationResult:
|
|
weights: dict[str, int]
|
|
dynamic: bool
|
|
metric_coverage_pct: Decimal
|
|
factors: list[dict[str, object]]
|
|
|
|
|
|
class DynamicAllocationOptimizer:
|
|
"""Tilts strategic weights with historical evidence, never return forecasts."""
|
|
|
|
MAX_DYNAMIC_TILT_PCT = Decimal("15")
|
|
|
|
@classmethod
|
|
def optimize(
|
|
cls,
|
|
strategic_weights: dict[str, int],
|
|
metrics: list[AssetClassMarketMetric],
|
|
*,
|
|
return_target_lower_pct: Decimal,
|
|
max_drawdown_pct: Decimal,
|
|
liquidity_requirement: str,
|
|
) -> DynamicAllocationResult:
|
|
metrics_by_class = {metric.asset_class: metric for metric in metrics}
|
|
coverage = Decimal(len(metrics_by_class)) / Decimal(len(_ASSET_CLASSES)) * _HUNDRED
|
|
if len(metrics_by_class) < 2:
|
|
return DynamicAllocationResult(
|
|
weights=dict(strategic_weights),
|
|
dynamic=False,
|
|
metric_coverage_pct=coverage,
|
|
factors=cls._factor_views(metrics, {}),
|
|
)
|
|
scores = {
|
|
asset_class: cls._score(
|
|
metric,
|
|
metrics,
|
|
return_target_lower_pct=return_target_lower_pct,
|
|
max_drawdown_pct=max_drawdown_pct,
|
|
liquidity_requirement=liquidity_requirement,
|
|
)
|
|
for asset_class, metric in metrics_by_class.items()
|
|
}
|
|
tilted = {asset_class: Decimal(value) for asset_class, value in strategic_weights.items()}
|
|
cls._apply_tilt(tilted, scores)
|
|
cls._apply_constraints(tilted, max_drawdown_pct, liquidity_requirement, scores)
|
|
return DynamicAllocationResult(
|
|
weights=cls._whole_percentages(tilted),
|
|
dynamic=True,
|
|
metric_coverage_pct=coverage,
|
|
factors=cls._factor_views(metrics, scores),
|
|
)
|
|
|
|
@classmethod
|
|
def _score(
|
|
cls,
|
|
metric: AssetClassMarketMetric,
|
|
peers: list[AssetClassMarketMetric],
|
|
*,
|
|
return_target_lower_pct: Decimal,
|
|
max_drawdown_pct: Decimal,
|
|
liquidity_requirement: str,
|
|
) -> Decimal:
|
|
annualized_proxy = metric.trailing_120d_return_pct * Decimal("2.1")
|
|
return_fit = cls._ratio(annualized_proxy, return_target_lower_pct)
|
|
return_rank = cls._rank(metric.trailing_120d_return_pct, [
|
|
item.trailing_120d_return_pct for item in peers
|
|
])
|
|
drawdown = abs(metric.max_drawdown_pct)
|
|
drawdown_fit = cls._inverse_ratio(drawdown, max_drawdown_pct)
|
|
drawdown_rank = cls._rank(-drawdown, [-abs(item.max_drawdown_pct) for item in peers])
|
|
liquidity_floor = _LIQUIDITY_FLOORS[liquidity_requirement]
|
|
liquidity_fit = cls._ratio(metric.average_daily_turnover_amount, liquidity_floor)
|
|
liquidity_rank = cls._rank(metric.average_daily_turnover_amount, [
|
|
item.average_daily_turnover_amount for item in peers
|
|
])
|
|
return (
|
|
Decimal("0.45") * (return_fit + return_rank) / 2
|
|
+ Decimal("0.35") * (drawdown_fit + drawdown_rank) / 2
|
|
+ Decimal("0.20") * (liquidity_fit + liquidity_rank) / 2
|
|
)
|
|
|
|
@classmethod
|
|
def _apply_tilt(cls, weights: dict[str, Decimal], scores: dict[str, Decimal]) -> None:
|
|
mean = sum(scores.values(), _ZERO) / Decimal(len(scores))
|
|
positive = {key: value - mean for key, value in scores.items() if value > mean}
|
|
negative = {key: mean - value for key, value in scores.items() if value < mean}
|
|
if not positive or not negative:
|
|
return
|
|
tilt = min(cls.MAX_DYNAMIC_TILT_PCT, sum(weights.get(key, _ZERO) for key in negative))
|
|
positive_total = sum(positive.values(), _ZERO)
|
|
negative_total = sum(negative.values(), _ZERO)
|
|
for asset_class, value in positive.items():
|
|
weights[asset_class] = weights.get(asset_class, _ZERO) + tilt * value / positive_total
|
|
for asset_class, value in negative.items():
|
|
weights[asset_class] = max(
|
|
_ZERO,
|
|
weights.get(asset_class, _ZERO) - tilt * value / negative_total,
|
|
)
|
|
|
|
@classmethod
|
|
def _apply_constraints(
|
|
cls,
|
|
weights: dict[str, Decimal],
|
|
max_drawdown_pct: Decimal,
|
|
liquidity_requirement: str,
|
|
scores: dict[str, Decimal],
|
|
) -> None:
|
|
equity_cap = Decimal("20") if max_drawdown_pct <= Decimal("10") else (
|
|
Decimal("40") if max_drawdown_pct <= Decimal("20") else Decimal("85")
|
|
)
|
|
equity_excess = max(_ZERO, weights["equity_etf"] - equity_cap)
|
|
weights["equity_etf"] -= equity_excess
|
|
weights["bond_etf"] += equity_excess
|
|
cash_minimum = _CASH_MINIMUMS[liquidity_requirement]
|
|
cash_shortfall = max(_ZERO, cash_minimum - weights["cash_management_etf"])
|
|
if cash_shortfall:
|
|
sources = sorted(
|
|
("bond_etf", "equity_etf"), key=lambda key: scores.get(key, _ZERO)
|
|
)
|
|
for source in sources:
|
|
moved = min(cash_shortfall, weights[source])
|
|
weights[source] -= moved
|
|
weights["cash_management_etf"] += moved
|
|
cash_shortfall -= moved
|
|
if cash_shortfall == 0:
|
|
break
|
|
difference = _HUNDRED - sum(weights.values(), _ZERO)
|
|
recipient = max(weights, key=lambda key: scores.get(key, _ZERO))
|
|
weights[recipient] += difference
|
|
|
|
@staticmethod
|
|
def _ratio(value: Decimal, target: Decimal) -> Decimal:
|
|
if target <= 0:
|
|
return Decimal("0.5")
|
|
return min(Decimal("1"), max(_ZERO, value / target))
|
|
|
|
@staticmethod
|
|
def _inverse_ratio(value: Decimal, target: Decimal) -> Decimal:
|
|
if target <= 0:
|
|
return Decimal("0.5")
|
|
if value <= 0:
|
|
return Decimal("1")
|
|
return min(Decimal("1"), target / value)
|
|
|
|
@staticmethod
|
|
def _rank(value: Decimal, peers: list[Decimal]) -> Decimal:
|
|
lower, upper = min(peers), max(peers)
|
|
if lower == upper:
|
|
return Decimal("0.5")
|
|
return (value - lower) / (upper - lower)
|
|
|
|
@staticmethod
|
|
def _whole_percentages(weights: dict[str, Decimal]) -> dict[str, int]:
|
|
rounded = {
|
|
key: int(value.quantize(Decimal("1"), rounding=ROUND_HALF_UP))
|
|
for key, value in weights.items()
|
|
}
|
|
difference = 100 - sum(rounded.values())
|
|
rounded[max(weights, key=lambda key: weights[key])] += difference
|
|
return rounded
|
|
|
|
@staticmethod
|
|
def _factor_views(
|
|
metrics: list[AssetClassMarketMetric], scores: dict[str, Decimal]
|
|
) -> list[dict[str, object]]:
|
|
return [
|
|
{
|
|
"asset_class": metric.asset_class,
|
|
"trailing_120d_return_pct": str(metric.trailing_120d_return_pct),
|
|
"max_drawdown_pct": str(metric.max_drawdown_pct),
|
|
"average_daily_turnover_amount": str(metric.average_daily_turnover_amount),
|
|
"liquidity_source": metric.liquidity_source,
|
|
"product_count": metric.product_count,
|
|
"composite_score": str(scores[metric.asset_class].quantize(Decimal("0.0001")))
|
|
if metric.asset_class in scores else None,
|
|
}
|
|
for metric in sorted(metrics, key=lambda item: item.asset_class)
|
|
]
|