feat: add dynamic advisor asset allocation
This commit is contained in:
Windows
@@ -0,0 +1,160 @@
|
||||
"""Explainable constraint-first allocation optimizer."""
|
||||
|
||||
from dataclasses import dataclass
|
||||
from decimal import ROUND_HALF_UP, Decimal
|
||||
|
||||
HUNDRED = Decimal("100")
|
||||
ASSET_CLASSES = ("cash_management_etf", "bond_etf", "equity_etf")
|
||||
|
||||
|
||||
@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
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class DynamicAllocationResult:
|
||||
weights: dict[str, int]
|
||||
dynamic: bool
|
||||
metric_coverage_pct: Decimal
|
||||
factors: list[dict[str, object]]
|
||||
|
||||
|
||||
class DynamicAllocationOptimizer:
|
||||
@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:
|
||||
available = {
|
||||
item.asset_class: item for item in metrics if item.asset_class in ASSET_CLASSES
|
||||
}
|
||||
coverage = Decimal(len(available)) / Decimal(len(ASSET_CLASSES)) * HUNDRED
|
||||
if len(available) < 2:
|
||||
return DynamicAllocationResult(
|
||||
dict(strategic_weights), False, coverage, cls._factors(metrics, {})
|
||||
)
|
||||
scores = {
|
||||
key: cls._score(
|
||||
value, metrics, return_target_lower_pct, max_drawdown_pct, liquidity_requirement
|
||||
)
|
||||
for key, value in available.items()
|
||||
}
|
||||
weights = {key: Decimal(value) for key, value in strategic_weights.items()}
|
||||
mean = sum(scores.values(), Decimal()) / Decimal(len(scores))
|
||||
winners = {key: score - mean for key, score in scores.items() if score > mean}
|
||||
losers = {key: mean - score for key, score in scores.items() if score < mean}
|
||||
if winners and losers:
|
||||
tilt = min(Decimal("15"), sum(weights.get(key, Decimal()) for key in losers))
|
||||
for key, value in winners.items():
|
||||
weights[key] = weights.get(key, Decimal()) + tilt * value / sum(winners.values())
|
||||
for key, value in losers.items():
|
||||
weights[key] = max(
|
||||
Decimal(), weights.get(key, Decimal()) - tilt * value / sum(losers.values())
|
||||
)
|
||||
equity_cap = (
|
||||
Decimal("20")
|
||||
if max_drawdown_pct <= 10
|
||||
else Decimal("40")
|
||||
if max_drawdown_pct <= 20
|
||||
else Decimal("85")
|
||||
)
|
||||
excess = max(Decimal(), weights["equity_etf"] - equity_cap)
|
||||
weights["equity_etf"] -= excess
|
||||
weights["bond_etf"] += excess
|
||||
cash_min = {
|
||||
"daily": Decimal("25"),
|
||||
"within_7_days": Decimal("15"),
|
||||
"within_30_days": Decimal("5"),
|
||||
"over_30_days": Decimal(),
|
||||
}[liquidity_requirement]
|
||||
shortfall = max(Decimal(), cash_min - weights["cash_management_etf"])
|
||||
for key in ("bond_etf", "equity_etf"):
|
||||
moved = min(shortfall, weights[key])
|
||||
weights[key] -= moved
|
||||
weights["cash_management_etf"] += moved
|
||||
shortfall -= moved
|
||||
rounded = {
|
||||
key: int(value.quantize(Decimal("1"), rounding=ROUND_HALF_UP))
|
||||
for key, value in weights.items()
|
||||
}
|
||||
largest = max(rounded, key=lambda key: rounded[key])
|
||||
rounded[largest] += 100 - sum(rounded.values())
|
||||
return DynamicAllocationResult(rounded, True, coverage, cls._factors(metrics, scores))
|
||||
|
||||
@staticmethod
|
||||
def _score(
|
||||
metric: AssetClassMarketMetric,
|
||||
peers: list[AssetClassMarketMetric],
|
||||
target: Decimal,
|
||||
drawdown_limit: Decimal,
|
||||
liquidity_requirement: str,
|
||||
) -> Decimal:
|
||||
returns = [item.trailing_120d_return_pct for item in peers]
|
||||
liquidities = [item.average_daily_turnover_amount for item in peers]
|
||||
return_fit = (
|
||||
min(
|
||||
Decimal("1"),
|
||||
max(Decimal(), metric.trailing_120d_return_pct * Decimal("2.1") / target),
|
||||
)
|
||||
if target > 0
|
||||
else Decimal("0.5")
|
||||
)
|
||||
return_rank = DynamicAllocationOptimizer._rank(metric.trailing_120d_return_pct, returns)
|
||||
drawdown = abs(metric.max_drawdown_pct)
|
||||
drawdown_fit = (
|
||||
min(Decimal("1"), drawdown_limit / drawdown) if drawdown > 0 else Decimal("1")
|
||||
)
|
||||
liquidity_floor = {
|
||||
"daily": Decimal("10000000"),
|
||||
"within_7_days": Decimal("3000000"),
|
||||
"within_30_days": Decimal("500000"),
|
||||
"over_30_days": Decimal(),
|
||||
}[liquidity_requirement]
|
||||
liquidity_fit = (
|
||||
min(Decimal("1"), metric.average_daily_turnover_amount / liquidity_floor)
|
||||
if liquidity_floor
|
||||
else Decimal("0.5")
|
||||
)
|
||||
liquidity_rank = DynamicAllocationOptimizer._rank(
|
||||
metric.average_daily_turnover_amount, liquidities
|
||||
)
|
||||
return (
|
||||
Decimal("0.45") * (return_fit + return_rank) / 2
|
||||
+ Decimal("0.35") * drawdown_fit
|
||||
+ Decimal("0.20") * (liquidity_fit + liquidity_rank) / 2
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _rank(value: Decimal, peers: list[Decimal]) -> Decimal:
|
||||
low, high = min(peers), max(peers)
|
||||
return Decimal("0.5") if low == high else (value - low) / (high - low)
|
||||
|
||||
@staticmethod
|
||||
def _factors(
|
||||
metrics: list[AssetClassMarketMetric], scores: dict[str, Decimal]
|
||||
) -> list[dict[str, object]]:
|
||||
return [
|
||||
{
|
||||
"asset_class": item.asset_class,
|
||||
"trailing_120d_return_pct": str(item.trailing_120d_return_pct),
|
||||
"max_drawdown_pct": str(item.max_drawdown_pct),
|
||||
"average_daily_turnover_amount": str(item.average_daily_turnover_amount),
|
||||
"product_count": item.product_count,
|
||||
"composite_score": (
|
||||
str(scores[item.asset_class].quantize(Decimal("0.0001")))
|
||||
if item.asset_class in scores
|
||||
else None
|
||||
),
|
||||
}
|
||||
for item in sorted(metrics, key=lambda value: value.asset_class)
|
||||
]
|
||||
Reference in New Issue
Block a user