347 lines
15 KiB
Python
347 lines
15 KiB
Python
"""Historical, analysis-only validation of static and dynamic allocations."""
|
|
|
|
from collections import defaultdict
|
|
from collections.abc import Callable, Sequence
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, date, datetime
|
|
from decimal import Decimal
|
|
from typing import Any
|
|
from uuid import uuid4
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.advisor_backtest_contracts import AllocationBacktestQuery
|
|
from app.core.contracts import RequestContext
|
|
from app.infrastructure.db import SessionFactory
|
|
from app.model.advisor_product import (
|
|
AdvisorAllocationBacktestRun,
|
|
AdvisorProductPriceHistory,
|
|
)
|
|
from app.model.audit import InteractionAudit
|
|
from app.repository.advisor_product_repository import AdvisorProductRepository
|
|
from app.repository.portfolio_analysis_repository import PortfolioAnalysisRepository
|
|
from app.service.asset_allocation_service import AssetAllocationService
|
|
from app.service.authorization_service import AuthorizationService
|
|
from app.service.dynamic_allocation_optimizer import (
|
|
ASSET_CLASSES,
|
|
AssetClassMarketMetric,
|
|
DynamicAllocationOptimizer,
|
|
)
|
|
from app.service.product_governance_monitor_service import SALES_INSTITUTION
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class BacktestObservation:
|
|
trade_date: date
|
|
returns_pct: dict[str, Decimal]
|
|
liquidity_observed: bool
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Performance:
|
|
total_return_pct: Decimal
|
|
max_drawdown_pct: Decimal
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class BacktestResult:
|
|
status: str
|
|
observation_count: int
|
|
static: Performance | None
|
|
dynamic: Performance | None
|
|
dynamic_rebalance_count: int
|
|
liquidity_history_coverage_pct: Decimal
|
|
limitations: tuple[str, ...]
|
|
|
|
|
|
class AllocationBacktestEngine:
|
|
@staticmethod
|
|
def run(
|
|
observations: Sequence[BacktestObservation],
|
|
static_weights: dict[str, int],
|
|
dynamic_weights: dict[date, dict[str, int]],
|
|
) -> BacktestResult:
|
|
ordered = sorted(observations, key=lambda item: item.trade_date)
|
|
liquidity_count = sum(item.liquidity_observed for item in ordered)
|
|
liquidity_coverage = (
|
|
Decimal(liquidity_count) / Decimal(len(ordered)) * Decimal("100")
|
|
if ordered
|
|
else Decimal()
|
|
)
|
|
limitations: list[str] = []
|
|
if len(ordered) < 120:
|
|
limitations.append("历史观察不足 120 个交易日，动态优化无法覆盖完整窗口。")
|
|
if liquidity_coverage < 80:
|
|
limitations.append("流动性历史字段覆盖率低于 80%，流动性结论受限。")
|
|
status = "ready"
|
|
if len(ordered) < 20:
|
|
status = "insufficient_history"
|
|
elif liquidity_coverage < 80:
|
|
status = "partial"
|
|
return BacktestResult(
|
|
status=status,
|
|
observation_count=len(ordered),
|
|
static=(
|
|
AllocationBacktestEngine._performance(ordered, static_weights) if ordered else None
|
|
),
|
|
dynamic=(
|
|
AllocationBacktestEngine._performance_by_date(
|
|
ordered, dynamic_weights, static_weights
|
|
)
|
|
if ordered
|
|
else None
|
|
),
|
|
dynamic_rebalance_count=sum(
|
|
1 for item in ordered if item.trade_date in dynamic_weights
|
|
),
|
|
liquidity_history_coverage_pct=liquidity_coverage,
|
|
limitations=tuple(limitations),
|
|
)
|
|
|
|
@staticmethod
|
|
def _performance(
|
|
observations: Sequence[BacktestObservation], weights: dict[str, int]
|
|
) -> Performance:
|
|
return AllocationBacktestEngine._performance_by_date(observations, {}, weights)
|
|
|
|
@staticmethod
|
|
def _performance_by_date(
|
|
observations: Sequence[BacktestObservation],
|
|
weights_by_date: dict[date, dict[str, int]],
|
|
fallback: dict[str, int],
|
|
) -> Performance:
|
|
value = Decimal("1")
|
|
peak = value
|
|
max_drawdown = Decimal()
|
|
for observation in observations:
|
|
weights = weights_by_date.get(observation.trade_date, fallback)
|
|
daily_return = sum(
|
|
Decimal(weight)
|
|
/ Decimal("100")
|
|
* observation.returns_pct.get(asset_class, Decimal())
|
|
/ Decimal("100")
|
|
for asset_class, weight in weights.items()
|
|
)
|
|
value *= Decimal("1") + daily_return
|
|
peak = max(peak, value)
|
|
if peak > 0:
|
|
max_drawdown = min(max_drawdown, value / peak - Decimal("1"))
|
|
return Performance(
|
|
total_return_pct=((value - Decimal("1")) * Decimal("100")).quantize(Decimal("0.0001")),
|
|
max_drawdown_pct=(max_drawdown * Decimal("100")).quantize(Decimal("0.0001")),
|
|
)
|
|
|
|
|
|
class AllocationBacktestService:
|
|
def __init__(self, *, session_factory: Callable[[], Any] = SessionFactory) -> None:
|
|
self.session_factory = session_factory
|
|
|
|
async def run(
|
|
self, payload: AllocationBacktestQuery, context: RequestContext, key: str | None = None
|
|
) -> dict[str, object]:
|
|
await AuthorizationService.require(context, "asset-allocation:backtest", admin=True)
|
|
result = await self._calculate(payload)
|
|
|
|
async def operation(session: AsyncSession) -> dict[str, Any]:
|
|
row = AdvisorAllocationBacktestRun(
|
|
backtest_no=f"AB-{uuid4().hex[:24]}",
|
|
started_on=payload.started_on,
|
|
ended_on=payload.ended_on,
|
|
profile_risk_level=f"C{payload.profile_risk_level}",
|
|
return_target_lower_pct=Decimal(payload.return_target_lower_pct),
|
|
max_drawdown_pct=Decimal(payload.max_drawdown_pct),
|
|
liquidity_requirement=payload.liquidity_requirement,
|
|
status=result.status,
|
|
observation_count=result.observation_count,
|
|
static_total_return_pct=result.static.total_return_pct if result.static else None,
|
|
dynamic_total_return_pct=result.dynamic.total_return_pct
|
|
if result.dynamic
|
|
else None,
|
|
static_max_drawdown_pct=result.static.max_drawdown_pct if result.static else None,
|
|
dynamic_max_drawdown_pct=result.dynamic.max_drawdown_pct
|
|
if result.dynamic
|
|
else None,
|
|
dynamic_rebalance_count=result.dynamic_rebalance_count,
|
|
liquidity_history_coverage_pct=result.liquidity_history_coverage_pct,
|
|
limitations=list(result.limitations),
|
|
strategy_version="constrained_historical_multi_factor_v1",
|
|
created_at=datetime.now(UTC).replace(tzinfo=None),
|
|
)
|
|
session.add(row)
|
|
session.add(
|
|
InteractionAudit(
|
|
actor_type="user",
|
|
actor_id=int(context.user_id),
|
|
portal=context.portal,
|
|
action_type="advisor.allocation_backtest_created",
|
|
detail={"backtest_no": row.backtest_no, "trace_id": context.trace_id},
|
|
created_at=datetime.now(UTC).replace(tzinfo=None),
|
|
)
|
|
)
|
|
await session.flush()
|
|
return {"data": self._view(row), "meta": {"trace_id": context.trace_id}}
|
|
|
|
from app.service.api_transaction_service import ApiTransactionService
|
|
|
|
return await ApiTransactionService().execute(
|
|
context,
|
|
f"advisor:allocation-backtests:{payload.started_on}:{payload.ended_on}",
|
|
key,
|
|
payload.model_dump(mode="json"),
|
|
operation,
|
|
)
|
|
|
|
async def _calculate(self, payload: AllocationBacktestQuery) -> BacktestResult:
|
|
end_at = datetime.combine(payload.ended_on, datetime.max.time())
|
|
async with self.session_factory() as session:
|
|
candidates = await AdvisorProductRepository(session).authoritative_tradable_products(
|
|
end_at, sales_institution=SALES_INSTITUTION, limit=100
|
|
)
|
|
candidates, _ = AdvisorProductRepository.hard_suitability_filter(
|
|
candidates, payload.profile_risk_level
|
|
)
|
|
product_ids = tuple(item.product.id for item in candidates)
|
|
repository = PortfolioAnalysisRepository(session)
|
|
classifications = await repository.latest_asset_classifications(
|
|
product_ids, payload.ended_on
|
|
)
|
|
qualities = await repository.latest_quality(product_ids, payload.ended_on)
|
|
rows = list(
|
|
await session.scalars(
|
|
select(AdvisorProductPriceHistory)
|
|
.where(
|
|
AdvisorProductPriceHistory.product_id.in_(product_ids),
|
|
AdvisorProductPriceHistory.trade_date >= payload.started_on,
|
|
AdvisorProductPriceHistory.trade_date <= payload.ended_on,
|
|
AdvisorProductPriceHistory.price_kind == "fund_nav",
|
|
)
|
|
.order_by(
|
|
AdvisorProductPriceHistory.product_id, AdvisorProductPriceHistory.trade_date
|
|
)
|
|
)
|
|
)
|
|
eligible = {
|
|
product_id: classification.asset_class
|
|
for product_id, classification in classifications.items()
|
|
if classification.asset_class in ASSET_CLASSES
|
|
and qualities.get(product_id) is not None
|
|
and qualities[product_id].status == "accepted"
|
|
}
|
|
return self._calculate_from_rows(rows, eligible, payload)
|
|
|
|
@staticmethod
|
|
def _calculate_from_rows(
|
|
rows: Sequence[AdvisorProductPriceHistory],
|
|
eligible: dict[int, str],
|
|
payload: AllocationBacktestQuery,
|
|
) -> BacktestResult:
|
|
per_product: dict[int, list[AdvisorProductPriceHistory]] = defaultdict(list)
|
|
for row in rows:
|
|
if row.product_id in eligible:
|
|
per_product[row.product_id].append(row)
|
|
daily_returns: dict[date, dict[str, list[Decimal]]] = defaultdict(lambda: defaultdict(list))
|
|
daily_liquidity: dict[date, list[bool]] = defaultdict(list)
|
|
for product_id, history in per_product.items():
|
|
for previous, current in zip(history, history[1:], strict=False):
|
|
if previous.close_price <= 0:
|
|
continue
|
|
daily_returns[current.trade_date][eligible[product_id]].append(
|
|
(current.close_price / previous.close_price - Decimal("1")) * Decimal("100")
|
|
)
|
|
daily_liquidity[current.trade_date].append(current.turnover_amount is not None)
|
|
observations = [
|
|
BacktestObservation(
|
|
trade_date=trade_date,
|
|
returns_pct={
|
|
asset_class: sum(values, Decimal()) / len(values)
|
|
for asset_class, values in returns.items()
|
|
},
|
|
liquidity_observed=all(daily_liquidity[trade_date]),
|
|
)
|
|
for trade_date, returns in sorted(daily_returns.items())
|
|
if returns
|
|
]
|
|
static = AssetAllocationService._strategic_weights(
|
|
f"C{payload.profile_risk_level}",
|
|
max(13, (payload.ended_on - payload.started_on).days // 30),
|
|
payload.liquidity_requirement,
|
|
Decimal(payload.max_drawdown_pct),
|
|
)
|
|
dynamic: dict[date, dict[str, int]] = {}
|
|
for index in range(119, len(observations), 20):
|
|
metrics = AllocationBacktestService._rolling_metrics(
|
|
observations[index - 119 : index + 1]
|
|
)
|
|
optimized = DynamicAllocationOptimizer.optimize(
|
|
static,
|
|
metrics,
|
|
return_target_lower_pct=Decimal(payload.return_target_lower_pct),
|
|
max_drawdown_pct=Decimal(payload.max_drawdown_pct),
|
|
liquidity_requirement=payload.liquidity_requirement,
|
|
)
|
|
dynamic[observations[index].trade_date] = optimized.weights
|
|
return AllocationBacktestEngine.run(observations, static, dynamic)
|
|
|
|
@staticmethod
|
|
def _rolling_metrics(
|
|
observations: Sequence[BacktestObservation],
|
|
) -> list[AssetClassMarketMetric]:
|
|
grouped: dict[str, list[Decimal]] = defaultdict(list)
|
|
liquidity: dict[str, list[Decimal]] = defaultdict(list)
|
|
for observation in observations:
|
|
for asset_class, value in observation.returns_pct.items():
|
|
grouped[asset_class].append(value)
|
|
if observation.liquidity_observed:
|
|
liquidity[asset_class].append(Decimal("1"))
|
|
result: list[AssetClassMarketMetric] = []
|
|
for asset_class, returns in grouped.items():
|
|
value = Decimal("1")
|
|
peak = value
|
|
drawdown = Decimal()
|
|
for item in returns:
|
|
value *= Decimal("1") + item / Decimal("100")
|
|
peak = max(peak, value)
|
|
drawdown = min(drawdown, value / peak - Decimal("1"))
|
|
result.append(
|
|
AssetClassMarketMetric(
|
|
asset_class=asset_class,
|
|
trailing_120d_return_pct=(value - Decimal("1")) * Decimal("100"),
|
|
max_drawdown_pct=drawdown * Decimal("100"),
|
|
average_daily_turnover_amount=Decimal("10000000")
|
|
* (Decimal(len(liquidity[asset_class])) / Decimal(len(returns))),
|
|
product_count=len(returns),
|
|
)
|
|
)
|
|
return result
|
|
|
|
@staticmethod
|
|
def _view(row: AdvisorAllocationBacktestRun) -> dict[str, object]:
|
|
return {
|
|
"backtest_no": row.backtest_no,
|
|
"started_on": row.started_on.isoformat(),
|
|
"ended_on": row.ended_on.isoformat(),
|
|
"profile_risk_level": row.profile_risk_level,
|
|
"return_target_lower_pct": str(row.return_target_lower_pct),
|
|
"max_drawdown_pct": str(row.max_drawdown_pct),
|
|
"liquidity_requirement": row.liquidity_requirement,
|
|
"status": row.status,
|
|
"observation_count": row.observation_count,
|
|
"static_total_return_pct": str(row.static_total_return_pct)
|
|
if row.static_total_return_pct is not None
|
|
else None,
|
|
"dynamic_total_return_pct": str(row.dynamic_total_return_pct)
|
|
if row.dynamic_total_return_pct is not None
|
|
else None,
|
|
"static_max_drawdown_pct": str(row.static_max_drawdown_pct)
|
|
if row.static_max_drawdown_pct is not None
|
|
else None,
|
|
"dynamic_max_drawdown_pct": str(row.dynamic_max_drawdown_pct)
|
|
if row.dynamic_max_drawdown_pct is not None
|
|
else None,
|
|
"dynamic_rebalance_count": row.dynamic_rebalance_count,
|
|
"liquidity_history_coverage_pct": str(row.liquidity_history_coverage_pct),
|
|
"limitations": row.limitations,
|
|
"strategy_version": row.strategy_version,
|
|
}
|