119 lines
5.5 KiB
Python
119 lines
5.5 KiB
Python
"""Evidence-based comparison for listed Southern Fund products."""
|
|
|
|
from collections.abc import Callable
|
|
from datetime import UTC, date, datetime
|
|
from typing import Any
|
|
|
|
from app.core.contracts import RequestContext
|
|
from app.core.product_comparison_contracts import ProductComparisonQuery
|
|
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.product_governance_monitor_service import SALES_INSTITUTION
|
|
|
|
|
|
class ProductComparisonService:
|
|
def __init__(self, *, session_factory: Callable[[], Any] = SessionFactory) -> None:
|
|
self.session_factory = session_factory
|
|
|
|
async def compare(
|
|
self, payload: ProductComparisonQuery, context: RequestContext
|
|
) -> dict[str, object]:
|
|
await AuthorizationService.require(context, "product-comparison:read:self")
|
|
codes = tuple(dict.fromkeys(payload.product_codes))
|
|
if len(codes) < 2:
|
|
return {"status": "products_required", "message": "至少需要两个不同的基金代码。"}
|
|
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, fund_manager="南方基金", limit=50
|
|
)
|
|
selected = [item for item in candidates if item.product.product_code in codes]
|
|
by_code = {item.product.product_code: item for item in selected}
|
|
missing = [code for code in codes if code not in by_code]
|
|
ids = tuple(item.product.id for item in selected)
|
|
exposures = await PortfolioAnalysisRepository(session).latest_industry_exposures(
|
|
ids, date.today()
|
|
)
|
|
if missing:
|
|
return {
|
|
"status": "evidence_required",
|
|
"missing_product_codes": missing,
|
|
"message": "部分产品缺少当前有效的权威适当性或合同证据，无法完成对比。",
|
|
}
|
|
rows = [
|
|
self._product_view(by_code[code], exposures.get(by_code[code].product.id, []))
|
|
for code in codes
|
|
]
|
|
industry_sets: list[set[str]] = []
|
|
for row in rows:
|
|
raw_exposure = row.get("industry_exposure")
|
|
industry_sets.append({
|
|
str(item["industry_name"])
|
|
for item in raw_exposure
|
|
if isinstance(item, dict) and "industry_name" in item
|
|
} if isinstance(raw_exposure, list) else set())
|
|
common_industries = sorted(set.intersection(*industry_sets)) if industry_sets else []
|
|
return {
|
|
"status": "ready",
|
|
"products": rows,
|
|
"common_industries": common_industries,
|
|
"differences": self._differences(rows),
|
|
"analysis_only": True,
|
|
"disclaimer": "对比结果仅供分析参考，不构成交易指令。",
|
|
}
|
|
|
|
@staticmethod
|
|
def _product_view(candidate: Any, exposures: list[Any]) -> dict[str, object]:
|
|
product = candidate.product
|
|
contract = candidate.contract
|
|
liquidity = candidate.liquidity
|
|
return {
|
|
"product_code": product.product_code,
|
|
"product_name": product.product_name,
|
|
"product_category": product.product_category,
|
|
"suitability_risk_level": candidate.suitability.risk_level,
|
|
"fund_type": contract.fund_type,
|
|
"investment_scope": contract.investment_scope,
|
|
"performance_benchmark": contract.performance_benchmark,
|
|
"management_fee_rate_pct": str(contract.management_fee_rate_pct)
|
|
if contract.management_fee_rate_pct is not None else None,
|
|
"custodian_fee_rate_pct": str(contract.custodian_fee_rate_pct)
|
|
if contract.custodian_fee_rate_pct is not None else None,
|
|
"asset_scale_billion": str(candidate.asset_scale_billion)
|
|
if candidate.asset_scale_billion is not None else None,
|
|
"liquidity": {
|
|
"status": liquidity.status if liquidity else "unknown",
|
|
"average_daily_turnover_amount": str(liquidity.average_daily_turnover_amount)
|
|
if liquidity and liquidity.average_daily_turnover_amount is not None else None,
|
|
},
|
|
"industry_exposure": [
|
|
{"industry_name": item.industry_name,
|
|
"exposure_weight_pct": str(item.exposure_weight_pct)}
|
|
for item in exposures
|
|
],
|
|
"evidence": {
|
|
"suitability_source_url": candidate.suitability.source_url,
|
|
"contract_source_url": contract.source_url,
|
|
},
|
|
}
|
|
|
|
@staticmethod
|
|
def _differences(rows: list[dict[str, object]]) -> dict[str, list[object]]:
|
|
fields = (
|
|
"product_category", "suitability_risk_level", "fund_type",
|
|
"performance_benchmark", "management_fee_rate_pct", "custodian_fee_rate_pct",
|
|
)
|
|
return {
|
|
field: [row[field] for row in rows]
|
|
for field in fields
|
|
if len({str(row[field]) for row in rows}) > 1
|
|
}
|
|
|
|
|
|
async def product_comparison_tool(
|
|
arguments: ProductComparisonQuery, context: RequestContext
|
|
) -> dict[str, object]:
|
|
return await ProductComparisonService().compare(arguments, context)
|