"""Dynamic, analysis-only asset allocation from confirmed goals and history.""" from collections import defaultdict from collections.abc import Callable from datetime import UTC, date, datetime from decimal import Decimal from typing import Any from app.core.advisor_allocation_contracts import AssetAllocationQuery from app.core.contracts import RequestContext from app.infrastructure.db import SessionFactory from app.repository.advisor_product_repository import AdvisorProductRepository from app.repository.portfolio_analysis_repository import PortfolioAnalysisRepository from app.service.authorization_service import AuthorizationService from app.service.dynamic_allocation_optimizer import ( ASSET_CLASSES, AssetClassMarketMetric, DynamicAllocationOptimizer, ) from app.service.investment_goal_service import InvestmentGoalService from app.service.product_governance_monitor_service import SALES_INSTITUTION from app.service.profile_governance_service import ProfileGovernanceService from app.service.suitability_service import SuitabilityService BASE_ALLOCATIONS = { "C1": {"cash_management_etf": 50, "bond_etf": 40, "equity_etf": 10}, "C2": {"cash_management_etf": 30, "bond_etf": 50, "equity_etf": 20}, "C3": {"cash_management_etf": 15, "bond_etf": 45, "equity_etf": 40}, "C4": {"cash_management_etf": 10, "bond_etf": 25, "equity_etf": 65}, "C5": {"cash_management_etf": 5, "bond_etf": 15, "equity_etf": 80}, } ASSET_LABELS = { "cash_management_etf": "现金管理类场内基金", "bond_etf": "债券类场内基金", "equity_etf": "权益类场内基金", } class AssetAllocationService: def __init__( self, *, session_factory: Callable[[], Any] = SessionFactory, enforce_profile_governance: bool = False, ) -> None: self.session_factory = session_factory self.enforce_profile_governance = enforce_profile_governance async def generate_for_agent( self, _arguments: AssetAllocationQuery, context: RequestContext ) -> dict[str, object]: await AuthorizationService.require(context, "asset-allocation:generate:self") if self.enforce_profile_governance: await ProfileGovernanceService().require_operable(int(context.user_id)) authority = await SuitabilityService().authority_for_customer(int(context.user_id)) if authority.customer_risk_level is None: return {"status": "profile_required"} goal = await InvestmentGoalService().current_for_agent(context) if goal is None: return {"status": "investment_goal_required"} horizon = goal.get("investment_horizon_months") if not isinstance(horizon, int): return {"status": "investment_goal_invalid"} risk = f"C{authority.customer_risk_level}" liquidity = str(goal["liquidity_requirement"]) strategic = self._strategic_weights( risk, horizon, liquidity, Decimal(str(goal["max_drawdown_pct"])) ) metrics = await self._market_metrics(authority.customer_risk_level) optimized = DynamicAllocationOptimizer.optimize( strategic, metrics, return_target_lower_pct=Decimal(str(goal["annualized_return_lower_pct"])), max_drawdown_pct=Decimal(str(goal["max_drawdown_pct"])), liquidity_requirement=liquidity, ) return { "status": "ready", "allocation": [ {"asset_class": key, "label": ASSET_LABELS[key], "target_pct": value} for key, value in optimized.weights.items() ], "optimization": { "method": "constrained_historical_multi_factor_v1", "dynamic": optimized.dynamic, "metric_coverage_pct": str(optimized.metric_coverage_pct.quantize(Decimal("0.01"))), "strategic_allocation": strategic, "factor_evidence": optimized.factors, }, "constraints": { "annualized_return_lower_pct": goal["annualized_return_lower_pct"], "max_drawdown_pct": goal["max_drawdown_pct"], "liquidity_requirement": liquidity, "investment_horizon_months": horizon, }, "analysis_only": True, } async def _market_metrics(self, customer_risk_level: int) -> list[AssetClassMarketMetric]: now = datetime.now(UTC).replace(tzinfo=None) async with self.session_factory() as session: candidates = await AdvisorProductRepository(session).authoritative_tradable_products( now, sales_institution=SALES_INSTITUTION, limit=50 ) candidates, _excluded = AdvisorProductRepository.hard_suitability_filter( candidates, customer_risk_level ) ids = tuple(item.product.id for item in candidates) repository = PortfolioAnalysisRepository(session) classifications = await repository.latest_asset_classifications(ids, date.today()) snapshots = await repository.latest_metrics(ids, date.today()) qualities = await repository.latest_quality(ids, date.today()) grouped: dict[str, list[AssetClassMarketMetric]] = defaultdict(list) for product_id in ids: classification = classifications.get(product_id) metric = snapshots.get(product_id) quality = qualities.get(product_id) if ( classification is None or classification.asset_class not in ASSET_CLASSES or quality is None or quality.status != "accepted" or metric is None or metric.trailing_120d_return_pct is None or metric.max_drawdown_pct is None or metric.average_daily_turnover_amount is None or metric.observation_count < 20 ): continue grouped[classification.asset_class].append( AssetClassMarketMetric( asset_class=classification.asset_class, trailing_120d_return_pct=metric.trailing_120d_return_pct, max_drawdown_pct=metric.max_drawdown_pct, average_daily_turnover_amount=metric.average_daily_turnover_amount, product_count=1, ) ) return [ AssetClassMarketMetric( asset_class=key, trailing_120d_return_pct=sum( (item.trailing_120d_return_pct for item in rows), Decimal() ) / len(rows), max_drawdown_pct=( sum((item.max_drawdown_pct for item in rows), Decimal()) / len(rows) ), average_daily_turnover_amount=sum( (item.average_daily_turnover_amount for item in rows), Decimal() ) / len(rows), product_count=len(rows), ) for key, rows in grouped.items() ] @staticmethod def _strategic_weights( risk: str, horizon: int, liquidity: str, max_drawdown: Decimal ) -> dict[str, int]: weights = dict(BASE_ALLOCATIONS[risk]) if horizon <= 12: AssetAllocationService._move(weights, "equity_etf", "cash_management_etf", 10) elif horizon >= 60 and risk != "C1": AssetAllocationService._move(weights, "bond_etf", "equity_etf", 5) if liquidity == "daily": AssetAllocationService._move(weights, "equity_etf", "cash_management_etf", 10) AssetAllocationService._move(weights, "bond_etf", "cash_management_etf", 5) if max_drawdown <= 10: AssetAllocationService._cap_equity(weights, 20) elif max_drawdown <= 20: AssetAllocationService._cap_equity(weights, 40) return weights @staticmethod def _move(weights: dict[str, int], source: str, target: str, amount: int) -> None: moved = min(weights[source], amount) weights[source] -= moved weights[target] += moved @staticmethod def _cap_equity(weights: dict[str, int], cap: int) -> None: excess = max(0, weights["equity_etf"] - cap) weights["equity_etf"] -= excess weights["bond_etf"] += excess async def asset_allocation_tool( arguments: AssetAllocationQuery, context: RequestContext ) -> dict[str, object]: return await AssetAllocationService(enforce_profile_governance=True).generate_for_agent( arguments, context )