118 lines
4.2 KiB
Python
118 lines
4.2 KiB
Python
"""Reproducible historical product metrics for advisory analysis."""
|
|
|
|
from collections.abc import Sequence
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, date, datetime
|
|
from decimal import ROUND_HALF_UP, Decimal
|
|
from math import sqrt
|
|
from typing import Protocol
|
|
|
|
from app.model.advisor_product import AdvisorProductMetricSnapshot
|
|
|
|
|
|
class HistoricalPrice(Protocol):
|
|
@property
|
|
def trade_date(self) -> date: ...
|
|
|
|
@property
|
|
def close_price(self) -> Decimal: ...
|
|
|
|
@property
|
|
def turnover_amount(self) -> Decimal | None: ...
|
|
|
|
_HUNDRED = Decimal("100")
|
|
_QUANTIZE = Decimal("0.0001")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CalculatedProductMetrics:
|
|
trailing_20d_return_pct: Decimal | None
|
|
trailing_120d_return_pct: Decimal | None
|
|
annualized_volatility_pct: Decimal | None
|
|
max_drawdown_pct: Decimal | None
|
|
average_daily_turnover_amount: Decimal | None
|
|
observation_count: int
|
|
|
|
|
|
class ProductMetricService:
|
|
CALCULATION_VERSION = "v1"
|
|
|
|
@classmethod
|
|
def calculate(cls, prices: Sequence[HistoricalPrice]) -> CalculatedProductMetrics:
|
|
ordered = sorted(prices, key=lambda item: item.trade_date)
|
|
closes = [item.close_price for item in ordered if item.close_price > 0]
|
|
returns = [
|
|
float(current / previous - 1)
|
|
for previous, current in zip(closes, closes[1:], strict=False)
|
|
if previous > 0
|
|
]
|
|
return CalculatedProductMetrics(
|
|
trailing_20d_return_pct=cls._period_return(closes, 20),
|
|
trailing_120d_return_pct=cls._period_return(closes, 120),
|
|
annualized_volatility_pct=cls._annualized_volatility(returns),
|
|
max_drawdown_pct=cls._max_drawdown(closes),
|
|
average_daily_turnover_amount=cls._average_turnover(ordered),
|
|
observation_count=len(closes),
|
|
)
|
|
|
|
@classmethod
|
|
def snapshot(
|
|
cls, product_id: int, prices: Sequence[HistoricalPrice]
|
|
) -> AdvisorProductMetricSnapshot | None:
|
|
if not prices:
|
|
return None
|
|
as_of_date = max(item.trade_date for item in prices)
|
|
calculated = cls.calculate(prices)
|
|
now = datetime.now(UTC).replace(tzinfo=None)
|
|
return AdvisorProductMetricSnapshot(
|
|
product_id=product_id,
|
|
as_of_date=as_of_date,
|
|
trailing_20d_return_pct=calculated.trailing_20d_return_pct,
|
|
trailing_120d_return_pct=calculated.trailing_120d_return_pct,
|
|
annualized_volatility_pct=calculated.annualized_volatility_pct,
|
|
max_drawdown_pct=calculated.max_drawdown_pct,
|
|
average_daily_turnover_amount=calculated.average_daily_turnover_amount,
|
|
observation_count=calculated.observation_count,
|
|
source="advisor_product_price_history_dual_read",
|
|
calculation_version=cls.CALCULATION_VERSION,
|
|
created_at=now,
|
|
)
|
|
|
|
@staticmethod
|
|
def _period_return(closes: list[Decimal], days: int) -> Decimal | None:
|
|
if len(closes) <= days:
|
|
return None
|
|
return ((closes[-1] / closes[-days - 1] - 1) * _HUNDRED).quantize(
|
|
_QUANTIZE, rounding=ROUND_HALF_UP
|
|
)
|
|
|
|
@staticmethod
|
|
def _annualized_volatility(returns: list[float]) -> Decimal | None:
|
|
if len(returns) < 20:
|
|
return None
|
|
mean = sum(returns) / len(returns)
|
|
variance = sum((value - mean) ** 2 for value in returns) / (len(returns) - 1)
|
|
return Decimal(str(sqrt(variance * 252) * 100)).quantize(
|
|
_QUANTIZE, rounding=ROUND_HALF_UP
|
|
)
|
|
|
|
@staticmethod
|
|
def _max_drawdown(closes: list[Decimal]) -> Decimal | None:
|
|
if len(closes) < 2:
|
|
return None
|
|
peak = closes[0]
|
|
drawdown = Decimal()
|
|
for close in closes:
|
|
peak = max(peak, close)
|
|
drawdown = min(drawdown, close / peak - 1)
|
|
return (drawdown * _HUNDRED).quantize(_QUANTIZE, rounding=ROUND_HALF_UP)
|
|
|
|
@staticmethod
|
|
def _average_turnover(prices: list[HistoricalPrice]) -> Decimal | None:
|
|
values = [item.turnover_amount for item in prices[-20:] if item.turnover_amount is not None]
|
|
if not values:
|
|
return None
|
|
return (sum(values, Decimal()) / len(values)).quantize(
|
|
Decimal("0.01"), rounding=ROUND_HALF_UP
|
|
)
|