"""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) ]