group_fqcd_jr/app/service/dynamic_allocation_optimizer.py

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