113 lines
4.2 KiB
Python
113 lines
4.2 KiB
Python
"""客户持仓记忆：fin_holdings 聚合摘要，供最终画像整合使用。
|
|
|
|
只输出聚合结果（总市值、产品类型分布、盈亏概览、前三大持仓），
|
|
不向 LLM 或接口消费方暴露原始持仓行，控制 token 并避免明细泄露。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from decimal import Decimal
|
|
from typing import Any
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from repositories.fin_holdings import FinHoldingsRepo
|
|
|
|
|
|
class CustomerHoldingsMemory:
|
|
"""读取并聚合客户当前持仓，输出画像整合所需的持仓摘要。"""
|
|
|
|
TOP_HOLDINGS_LIMIT = 3
|
|
|
|
def __init__(self, *, repository_factory=FinHoldingsRepo):
|
|
self.repository_factory = repository_factory
|
|
|
|
async def summary(
|
|
self, db: AsyncSession, customer_id: int
|
|
) -> tuple[dict[str, Any] | None, list[str]]:
|
|
"""聚合持有中持仓；无持仓返回 None，异常返回降级 warnings。"""
|
|
try:
|
|
rows = await self.repository_factory(db).list_with_products(
|
|
customer_id, include_closed=False
|
|
)
|
|
except Exception as exc: # noqa: BLE001 沿用记忆模块降级约定
|
|
return None, [f"holdings_recall_failed:{type(exc).__name__}"]
|
|
if not rows:
|
|
return None, []
|
|
|
|
total_value = Decimal("0")
|
|
total_cost = Decimal("0")
|
|
total_pnl = Decimal("0")
|
|
win_count = 0
|
|
lose_count = 0
|
|
mix: dict[str, Decimal] = {}
|
|
ranked: list[tuple[Decimal, str]] = []
|
|
for holding, product in rows:
|
|
value = self._to_decimal(holding.current_value)
|
|
total_value += value
|
|
total_cost += self._to_decimal(holding.cost_amount)
|
|
total_pnl += self._to_decimal(holding.profit_loss)
|
|
ratio = holding.profit_ratio
|
|
if ratio is not None:
|
|
if ratio > 0:
|
|
win_count += 1
|
|
elif ratio < 0:
|
|
lose_count += 1
|
|
product_type = (product.product_type if product else None) or "未知"
|
|
mix[product_type] = mix.get(product_type, Decimal("0")) + value
|
|
name = product.product_name if product else f"产品{holding.product_id}"
|
|
ranked.append((value, name))
|
|
|
|
payload: dict[str, Any] = {
|
|
"holding_count": len(rows),
|
|
"total_market_value": float(total_value),
|
|
"total_cost": float(total_cost),
|
|
"product_type_mix": self._ratio_map(mix, total_value),
|
|
"profit_summary": {
|
|
"total_pnl": float(total_pnl),
|
|
"total_pnl_ratio": (
|
|
round(float(total_pnl / total_cost), 4)
|
|
if total_cost > 0
|
|
else None
|
|
),
|
|
"win_count": win_count,
|
|
"lose_count": lose_count,
|
|
},
|
|
"top_holdings": self._top_holdings(ranked, total_value),
|
|
}
|
|
return payload, []
|
|
|
|
@staticmethod
|
|
def _to_decimal(value: Any) -> Decimal:
|
|
"""把 ORM Decimal / float / None 统一转换为 Decimal，防御脏数据类型。"""
|
|
if value is None:
|
|
return Decimal("0")
|
|
try:
|
|
return value if isinstance(value, Decimal) else Decimal(str(value))
|
|
except Exception: # noqa: BLE001
|
|
return Decimal("0")
|
|
|
|
@staticmethod
|
|
def _ratio_map(values: dict[str, Decimal], total: Decimal) -> dict[str, float]:
|
|
"""把各维度市值转换为相对总市值的占比，总量为 0 时返回空表。"""
|
|
if total <= 0:
|
|
return {}
|
|
return {
|
|
key: round(float(value / total), 4)
|
|
for key, value in sorted(values.items(), key=lambda item: -item[1])
|
|
}
|
|
|
|
def _top_holdings(
|
|
self, ranked: list[tuple[Decimal, str]], total: Decimal
|
|
) -> list[dict[str, Any]]:
|
|
"""按市值取前三大持仓，返回名称与占比。"""
|
|
if total <= 0:
|
|
return []
|
|
ranked.sort(key=lambda item: -item[0])
|
|
return [
|
|
{"name": name, "weight": round(float(value / total), 4)}
|
|
for value, name in ranked[: self.TOP_HOLDINGS_LIMIT]
|
|
]
|
|
|
|
|
|
__all__ = ["CustomerHoldingsMemory"]
|