diff --git a/app/model/advisor_product.py b/app/model/advisor_product.py new file mode 100644 index 0000000..4475358 --- /dev/null +++ b/app/model/advisor_product.py @@ -0,0 +1,165 @@ +"""ORM mappings for additive advisory product reference tables. + +The immutable ``fin_product`` catalogue remains mapped by ``FundProduct`` in +``app.model.fund``. This module only maps the new advisory evidence tables. +""" + +from datetime import date, datetime +from decimal import Decimal +from typing import Any + +from sqlalchemy import JSON, BigInteger, Date, DateTime, Numeric, String, Text +from sqlalchemy.orm import Mapped, mapped_column + +from app.model.base import Base + + +class AdvisorProductReferenceSnapshot(Base): + __tablename__ = "advisor_product_reference_snapshot" + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True) + product_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + as_of_date: Mapped[date] = mapped_column(Date, nullable=False) + fund_type: Mapped[str] = mapped_column(String(64), nullable=False) + fund_asset_scale_billion: Mapped[Decimal] = mapped_column(Numeric(18, 4), nullable=False) + source: Mapped[str] = mapped_column(String(64), nullable=False) + risk_mapping_version: Mapped[str] = mapped_column(String(32), nullable=False) + created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + updated_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + + +class AdvisorProductPriceHistory(Base): + __tablename__ = "advisor_product_price_history" + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True) + product_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + trade_date: Mapped[date] = mapped_column(Date, nullable=False) + price_kind: Mapped[str] = mapped_column(String(16), nullable=False) + close_price: Mapped[Decimal] = mapped_column(Numeric(18, 6), nullable=False) + turnover_amount: Mapped[Decimal | None] = mapped_column(Numeric(24, 2)) + source: Mapped[str] = mapped_column(String(64), nullable=False) + source_updated_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + updated_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + + +class AdvisorProductMetricSnapshot(Base): + __tablename__ = "advisor_product_metric_snapshot" + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True) + product_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + as_of_date: Mapped[date] = mapped_column(Date, nullable=False) + trailing_20d_return_pct: Mapped[Decimal | None] = mapped_column(Numeric(10, 4)) + trailing_120d_return_pct: Mapped[Decimal | None] = mapped_column(Numeric(10, 4)) + annualized_volatility_pct: Mapped[Decimal | None] = mapped_column(Numeric(10, 4)) + max_drawdown_pct: Mapped[Decimal | None] = mapped_column(Numeric(10, 4)) + average_daily_turnover_amount: Mapped[Decimal | None] = mapped_column(Numeric(24, 2)) + observation_count: Mapped[int] = mapped_column(BigInteger, nullable=False) + source: Mapped[str] = mapped_column(String(64), nullable=False) + calculation_version: Mapped[str] = mapped_column(String(16), nullable=False) + created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + + +class AdvisorProductSuitabilityReference(Base): + __tablename__ = "advisor_product_suitability_reference" + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True) + product_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + sales_institution: Mapped[str] = mapped_column(String(128), nullable=False) + risk_level: Mapped[str] = mapped_column(String(8), nullable=False) + effective_from: Mapped[date] = mapped_column(Date, nullable=False) + effective_until: Mapped[date | None] = mapped_column(Date) + source_url: Mapped[str] = mapped_column(String(1024), nullable=False) + document_title: Mapped[str] = mapped_column(String(256), nullable=False) + document_published_at: Mapped[date | None] = mapped_column(Date) + document_sha256: Mapped[str] = mapped_column(String(64), nullable=False) + source: Mapped[str] = mapped_column(String(64), nullable=False) + review_status: Mapped[str] = mapped_column(String(16), nullable=False) + verified_by: Mapped[str | None] = mapped_column(String(128)) + verified_at: Mapped[datetime | None] = mapped_column(DateTime) + created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + updated_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + + +class AdvisorProductContractSnapshot(Base): + __tablename__ = "advisor_product_contract_snapshot" + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True) + product_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + effective_from: Mapped[date] = mapped_column(Date, nullable=False) + effective_until: Mapped[date | None] = mapped_column(Date) + fund_type: Mapped[str] = mapped_column(String(64), nullable=False) + investment_scope: Mapped[str] = mapped_column(Text, nullable=False) + performance_benchmark: Mapped[str | None] = mapped_column(String(256)) + risk_return_characteristics: Mapped[str] = mapped_column(Text, nullable=False) + custodian_name: Mapped[str | None] = mapped_column(String(128)) + management_fee_rate_pct: Mapped[Decimal | None] = mapped_column(Numeric(9, 6)) + custodian_fee_rate_pct: Mapped[Decimal | None] = mapped_column(Numeric(9, 6)) + inception_date: Mapped[date | None] = mapped_column(Date) + source_url: Mapped[str] = mapped_column(String(1024), nullable=False) + document_title: Mapped[str] = mapped_column(String(256), nullable=False) + document_published_at: Mapped[date | None] = mapped_column(Date) + document_sha256: Mapped[str] = mapped_column(String(64), nullable=False) + source: Mapped[str] = mapped_column(String(64), nullable=False) + review_status: Mapped[str] = mapped_column(String(16), nullable=False) + verified_by: Mapped[str | None] = mapped_column(String(128)) + verified_at: Mapped[datetime | None] = mapped_column(DateTime) + created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + updated_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + + +class AdvisorProductGovernanceCandidate(Base): + __tablename__ = "advisor_product_governance_candidate_snapshot" + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True) + product_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + data_kind: Mapped[str] = mapped_column(String(16), nullable=False) + sales_institution: Mapped[str | None] = mapped_column(String(128)) + observed_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + effective_from: Mapped[date] = mapped_column(Date, nullable=False) + risk_level: Mapped[str | None] = mapped_column(String(8)) + payload: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False) + source_url: Mapped[str] = mapped_column(String(1024), nullable=False) + document_title: Mapped[str] = mapped_column(String(256), nullable=False) + document_published_at: Mapped[date | None] = mapped_column(Date) + document_sha256: Mapped[str] = mapped_column(String(64), nullable=False) + source: Mapped[str] = mapped_column(String(64), nullable=False) + review_status: Mapped[str] = mapped_column(String(16), nullable=False) + reviewed_by: Mapped[int | None] = mapped_column(BigInteger) + reviewed_at: Mapped[datetime | None] = mapped_column(DateTime) + review_comment: Mapped[str | None] = mapped_column(String(1000)) + created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + updated_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + + +class AdvisorProductGovernanceSyncRun(Base): + __tablename__ = "advisor_product_governance_sync_run" + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True) + run_no: Mapped[str] = mapped_column(String(36), nullable=False, unique=True) + source: Mapped[str] = mapped_column(String(64), nullable=False) + status: Mapped[str] = mapped_column(String(16), nullable=False) + product_count: Mapped[int] = mapped_column(BigInteger, nullable=False) + change_count: Mapped[int] = mapped_column(BigInteger, nullable=False) + error_count: Mapped[int] = mapped_column(BigInteger, nullable=False) + detail: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False) + started_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + completed_at: Mapped[datetime | None] = mapped_column(DateTime) + created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + updated_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + + +class AdvisorProductMarketQuoteSnapshot(Base): + __tablename__ = "advisor_product_market_quote_snapshot" + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True) + product_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + observed_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + last_price: Mapped[Decimal] = mapped_column(Numeric(18, 6), nullable=False) + previous_close: Mapped[Decimal | None] = mapped_column(Numeric(18, 6)) + change_pct: Mapped[Decimal | None] = mapped_column(Numeric(10, 4)) + volume: Mapped[Decimal | None] = mapped_column(Numeric(24, 4)) + turnover_amount: Mapped[Decimal | None] = mapped_column(Numeric(24, 2)) + quote_status: Mapped[str] = mapped_column(String(16), nullable=False) + source: Mapped[str] = mapped_column(String(64), nullable=False) + created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) diff --git a/app/repository/advisor_product_repository.py b/app/repository/advisor_product_repository.py new file mode 100644 index 0000000..d421acd --- /dev/null +++ b/app/repository/advisor_product_repository.py @@ -0,0 +1,226 @@ +"""Read-only product evidence repository for the advisory service.""" + +from collections.abc import Sequence +from dataclasses import dataclass +from datetime import date, datetime, timedelta +from decimal import Decimal +from typing import Protocol, TypeVar + +from sqlalchemy import or_, select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.model.advisor_product import ( + AdvisorProductContractSnapshot, + AdvisorProductGovernanceCandidate, + AdvisorProductMarketQuoteSnapshot, + AdvisorProductReferenceSnapshot, + AdvisorProductSuitabilityReference, +) +from app.model.fund import FundProduct + + +@dataclass(frozen=True, slots=True) +class ProductSuitabilityEvidence: + risk_level: str + sales_institution: str + source_url: str + document_title: str + document_sha256: str + effective_from: date + + +@dataclass(frozen=True, slots=True) +class ProductContractEvidence: + fund_type: str + investment_scope: str + performance_benchmark: str | None + risk_return_characteristics: str + custodian_name: str | None + management_fee_rate_pct: Decimal | None + custodian_fee_rate_pct: Decimal | None + inception_date: date | None + source_url: str + document_title: str + document_sha256: str + document_published_at: date | None + + +@dataclass(frozen=True, slots=True) +class AuthoritativeProductCandidate: + product: FundProduct + suitability: ProductSuitabilityEvidence + contract: ProductContractEvidence + asset_scale_billion: Decimal | None = None + market_quote: AdvisorProductMarketQuoteSnapshot | None = None + + +class _ProductRow(Protocol): + product_id: int + + +ProductRow = TypeVar("ProductRow", bound=_ProductRow) + + +class AdvisorProductRepository: + """Read verified, current, exchange-traded product facts only.""" + + def __init__(self, session: AsyncSession) -> None: + self.session = session + + async def tradable_products( + self, + now: datetime, + *, + fund_manager: str | None = None, + product_codes: tuple[str, ...] | None = None, + limit: int = 50, + ) -> list[FundProduct]: + statement = select(FundProduct).where( + FundProduct.status == "上市", + FundProduct.exchange_code.in_(("SSE", "SZSE")), + or_(FundProduct.open_start_at.is_(None), FundProduct.open_start_at <= now), + or_(FundProduct.open_end_at.is_(None), FundProduct.open_end_at > now), + ) + if fund_manager is not None: + statement = statement.where(FundProduct.fund_manager == fund_manager) + if product_codes is not None: + statement = statement.where(FundProduct.product_code.in_(product_codes)) + statement = statement.order_by(FundProduct.id.asc()).limit(limit) + return list(await self.session.scalars(statement)) + + async def authoritative_tradable_products( + self, + now: datetime, + *, + sales_institution: str, + fund_manager: str | None = None, + quote_max_age_seconds: int = 300, + min_asset_scale_billion: Decimal | None = None, + limit: int = 50, + ) -> list[AuthoritativeProductCandidate]: + """Apply evidence gates before any product reaches an Agent. + + The baseline ``fin_product.risk_level`` is never treated as the + sales-institution suitability rating. Missing or unverified evidence + excludes a product (fail closed). + """ + products = await self.tradable_products(now, fund_manager=fund_manager, limit=limit) + if not products: + return [] + ids = [product.id for product in products] + as_of = now.date() + suitability_rows = list(await self.session.scalars(select( + AdvisorProductSuitabilityReference + ).where( + AdvisorProductSuitabilityReference.product_id.in_(ids), + AdvisorProductSuitabilityReference.sales_institution == sales_institution, + AdvisorProductSuitabilityReference.review_status == "verified", + AdvisorProductSuitabilityReference.effective_from <= as_of, + or_( + AdvisorProductSuitabilityReference.effective_until.is_(None), + AdvisorProductSuitabilityReference.effective_until >= as_of, + ), + AdvisorProductSuitabilityReference.source_url != "", + AdvisorProductSuitabilityReference.document_sha256 != "", + ).order_by( + AdvisorProductSuitabilityReference.product_id, + AdvisorProductSuitabilityReference.effective_from.desc(), + ))) + contract_rows = list(await self.session.scalars(select( + AdvisorProductContractSnapshot + ).where( + AdvisorProductContractSnapshot.product_id.in_(ids), + AdvisorProductContractSnapshot.review_status == "verified", + AdvisorProductContractSnapshot.effective_from <= as_of, + or_( + AdvisorProductContractSnapshot.effective_until.is_(None), + AdvisorProductContractSnapshot.effective_until >= as_of, + ), + AdvisorProductContractSnapshot.source_url != "", + AdvisorProductContractSnapshot.document_sha256 != "", + ).order_by( + AdvisorProductContractSnapshot.product_id, + AdvisorProductContractSnapshot.effective_from.desc(), + ))) + reference_rows = list(await self.session.scalars(select( + AdvisorProductReferenceSnapshot + ).where( + AdvisorProductReferenceSnapshot.product_id.in_(ids), + AdvisorProductReferenceSnapshot.as_of_date <= as_of, + ).order_by( + AdvisorProductReferenceSnapshot.product_id, + AdvisorProductReferenceSnapshot.as_of_date.desc(), + ))) + pending_ids = set(await self.session.scalars(select( + AdvisorProductGovernanceCandidate.product_id + ).where( + AdvisorProductGovernanceCandidate.product_id.in_(ids), + AdvisorProductGovernanceCandidate.review_status == "pending_review", + ))) + quote_rows = list(await self.session.scalars(select( + AdvisorProductMarketQuoteSnapshot + ).where( + AdvisorProductMarketQuoteSnapshot.product_id.in_(ids), + AdvisorProductMarketQuoteSnapshot.quote_status == "active", + ).order_by( + AdvisorProductMarketQuoteSnapshot.product_id, + AdvisorProductMarketQuoteSnapshot.observed_at.desc(), + ))) + suitability_by_product = self._latest_by_product(suitability_rows) + contract_by_product = self._latest_by_product(contract_rows) + reference_by_product = self._latest_by_product(reference_rows) + quote_cutoff = now - timedelta(seconds=quote_max_age_seconds) + quote_by_product: dict[int, AdvisorProductMarketQuoteSnapshot] = {} + for quote in quote_rows: + if quote.observed_at >= quote_cutoff: + quote_by_product.setdefault(quote.product_id, quote) + + result: list[AuthoritativeProductCandidate] = [] + for product in products: + suitability = suitability_by_product.get(product.id) + contract = contract_by_product.get(product.id) + if product.id in pending_ids or suitability is None or contract is None: + continue + reference = reference_by_product.get(product.id) + scale = reference.fund_asset_scale_billion if reference else None + if ( + min_asset_scale_billion is not None + and suitability.risk_level != "R1" + and (scale is None or scale < min_asset_scale_billion) + ): + continue + result.append(AuthoritativeProductCandidate( + product=product, + suitability=ProductSuitabilityEvidence( + risk_level=suitability.risk_level, + sales_institution=suitability.sales_institution, + source_url=suitability.source_url, + document_title=suitability.document_title, + document_sha256=suitability.document_sha256, + effective_from=suitability.effective_from, + ), + contract=ProductContractEvidence( + fund_type=contract.fund_type, + investment_scope=contract.investment_scope, + performance_benchmark=contract.performance_benchmark, + risk_return_characteristics=contract.risk_return_characteristics, + custodian_name=contract.custodian_name, + management_fee_rate_pct=contract.management_fee_rate_pct, + custodian_fee_rate_pct=contract.custodian_fee_rate_pct, + inception_date=contract.inception_date, + source_url=contract.source_url, + document_title=contract.document_title, + document_sha256=contract.document_sha256, + document_published_at=contract.document_published_at, + ), + asset_scale_billion=scale, + market_quote=quote_by_product.get(product.id), + )) + return result + + @staticmethod + def _latest_by_product(rows: Sequence[ProductRow]) -> dict[int, ProductRow]: + result: dict[int, ProductRow] = {} + for row in rows: + result.setdefault(int(row.product_id), row) + return result diff --git a/app/service/product_governance_monitor_service.py b/app/service/product_governance_monitor_service.py new file mode 100644 index 0000000..e430407 --- /dev/null +++ b/app/service/product_governance_monitor_service.py @@ -0,0 +1,292 @@ +"""Monitor official product governance sources and enforce compliance review.""" + +import asyncio +from collections.abc import Callable +from dataclasses import dataclass +from datetime import UTC, date, datetime +from typing import Any +from uuid import uuid4 + +from sqlalchemy import select + +from app.core.contracts import RequestContext +from app.core.errors import GenericResourceNotFoundError, InvalidStateError +from app.infrastructure.db import SessionFactory +from app.model.advisor_product import ( + AdvisorProductContractSnapshot, + AdvisorProductGovernanceCandidate, + AdvisorProductGovernanceSyncRun, + AdvisorProductSuitabilityReference, +) +from app.model.audit import InteractionAudit +from app.service.authorization_service import AuthorizationService + +SourceLoader = Callable[[float], tuple[list[dict[str, str]], list[dict[str, str]]]] +SALES_INSTITUTION = ( + "\u5357\u65b9\u57fa\u91d1\u7ba1\u7406\u80a1\u4efd\u6709\u9650\u516c\u53f8\u76f4\u9500" +) + + +@dataclass(frozen=True) +class GovernanceMonitorResult: + product_count: int + change_count: int + error_count: int + run_no: str + + +class ProductGovernanceMonitorService: + """Source changes are quarantined until an authorized reviewer approves them.""" + + def __init__( + self, + *, + session_factory: Callable[[], Any] = SessionFactory, + loader: SourceLoader | None = None, + timeout_seconds: float = 30.0, + ) -> None: + self.session_factory = session_factory + self.loader = loader or self._default_loader + self.timeout_seconds = timeout_seconds + + async def monitor(self) -> GovernanceMonitorResult: + now = datetime.now(UTC).replace(tzinfo=None) + run_no = str(uuid4()) + async with self.session_factory() as session, session.begin(): + session.add(AdvisorProductGovernanceSyncRun( + run_no=run_no, + source="nffund_official_direct_sales", + status="running", + product_count=0, + change_count=0, + error_count=0, + detail={}, + started_at=now, + completed_at=None, + created_at=now, + updated_at=now, + )) + try: + suitability_rows, contract_rows = await asyncio.to_thread( + self.loader, self.timeout_seconds + ) + result = await self._store_candidates(suitability_rows, contract_rows, now, run_no) + except Exception as exc: + async with self.session_factory() as session, session.begin(): + run = await session.scalar(select(AdvisorProductGovernanceSyncRun).where( + AdvisorProductGovernanceSyncRun.run_no == run_no + ).with_for_update()) + assert run is not None + run.status = "failed" + run.error_count = 1 + run.detail = {"error_type": type(exc).__name__} + run.completed_at = now + run.updated_at = now + return GovernanceMonitorResult(0, 0, 1, run_no) + return result + + async def review( + self, candidate_id: int, decision: str, comment: str, context: RequestContext + ) -> dict[str, object]: + await AuthorizationService.require(context, "product-governance:review", admin=True) + now = datetime.now(UTC).replace(tzinfo=None) + async with self.session_factory() as session, session.begin(): + candidate = await session.scalar(select(AdvisorProductGovernanceCandidate).where( + AdvisorProductGovernanceCandidate.id == candidate_id + ).with_for_update()) + if candidate is None: + raise GenericResourceNotFoundError("product governance candidate not found") + if candidate.review_status != "pending_review": + raise InvalidStateError("PRODUCT_GOVERNANCE_CANDIDATE_NOT_PENDING") + candidate.review_status = "approved" if decision == "approved" else "rejected" + candidate.reviewed_by = int(context.user_id) + candidate.reviewed_at = now + candidate.review_comment = comment or None + candidate.updated_at = now + session.add(InteractionAudit( + actor_type="user", + actor_id=int(context.user_id), + portal="admin", + action_type=f"product_governance.candidate_{candidate.review_status}", + detail={ + "candidate_id": candidate.id, + "product_id": candidate.product_id, + "data_kind": candidate.data_kind, + "document_sha256": candidate.document_sha256, + "trace_id": context.trace_id, + }, + created_at=now, + )) + return self._candidate_view(candidate) + + async def pending( + self, context: RequestContext, *, limit: int = 100 + ) -> list[dict[str, object]]: + await AuthorizationService.require(context, "product-governance:review", admin=True) + async with self.session_factory() as session: + candidates = list(await session.scalars( + select(AdvisorProductGovernanceCandidate).where( + AdvisorProductGovernanceCandidate.review_status == "pending_review" + ).order_by(AdvisorProductGovernanceCandidate.observed_at.desc()).limit(limit) + )) + return [self._candidate_view(candidate) for candidate in candidates] + + async def runs(self, context: RequestContext, *, limit: int = 30) -> list[dict[str, object]]: + await AuthorizationService.require(context, "product-governance:read", admin=True) + async with self.session_factory() as session: + rows = list(await session.scalars(select(AdvisorProductGovernanceSyncRun).order_by( + AdvisorProductGovernanceSyncRun.started_at.desc() + ).limit(limit))) + return [ + { + "run_no": row.run_no, + "status": row.status, + "product_count": row.product_count, + "change_count": row.change_count, + "error_count": row.error_count, + "detail": row.detail, + "started_at": row.started_at.isoformat(), + "completed_at": row.completed_at.isoformat() if row.completed_at else None, + } + for row in rows + ] + + async def run_manually(self, context: RequestContext) -> GovernanceMonitorResult: + await AuthorizationService.require(context, "product-governance:sync", admin=True) + return await self.monitor() + + async def _store_candidates( + self, + suitability_rows: list[dict[str, str]], + contract_rows: list[dict[str, str]], + now: datetime, + run_no: str, + ) -> GovernanceMonitorResult: + product_codes = {row["product_code"] for row in [*suitability_rows, *contract_rows]} + async with self.session_factory() as session, session.begin(): + run = await session.scalar(select(AdvisorProductGovernanceSyncRun).where( + AdvisorProductGovernanceSyncRun.run_no == run_no + ).with_for_update()) + assert run is not None + from app.model.fund import FundProduct + + products = list(await session.execute(select( + FundProduct.id, FundProduct.product_code + ).where(FundProduct.product_code.in_(product_codes)))) + product_ids = {str(code): int(product_id) for product_id, code in products} + existing_suitability_rows = list(await session.execute(select( + AdvisorProductSuitabilityReference.product_id, + AdvisorProductSuitabilityReference.document_sha256, + AdvisorProductSuitabilityReference.effective_from, + ).where( + AdvisorProductSuitabilityReference.product_id.in_(product_ids.values()), + AdvisorProductSuitabilityReference.sales_institution == SALES_INSTITUTION, + ).order_by( + AdvisorProductSuitabilityReference.product_id, + AdvisorProductSuitabilityReference.effective_from.desc(), + ))) + existing_contract_rows = list(await session.execute(select( + AdvisorProductContractSnapshot.product_id, + AdvisorProductContractSnapshot.document_sha256, + AdvisorProductContractSnapshot.effective_from, + ).where(AdvisorProductContractSnapshot.product_id.in_(product_ids.values())).order_by( + AdvisorProductContractSnapshot.product_id, + AdvisorProductContractSnapshot.effective_from.desc(), + ))) + existing_candidate_rows = list(await session.execute(select( + AdvisorProductGovernanceCandidate.product_id, + AdvisorProductGovernanceCandidate.data_kind, + AdvisorProductGovernanceCandidate.document_sha256, + AdvisorProductGovernanceCandidate.review_status, + AdvisorProductGovernanceCandidate.observed_at, + ).where(AdvisorProductGovernanceCandidate.product_id.in_(product_ids.values())).order_by( + AdvisorProductGovernanceCandidate.product_id, + AdvisorProductGovernanceCandidate.data_kind, + AdvisorProductGovernanceCandidate.observed_at.desc(), + ))) + accepted_documents: dict[tuple[int, str], str] = {} + for product_id, document_hash, _effective_from in existing_suitability_rows: + accepted_documents.setdefault((int(product_id), "suitability"), str(document_hash)) + for product_id, document_hash, _effective_from in existing_contract_rows: + accepted_documents.setdefault((int(product_id), "contract"), str(document_hash)) + existing_candidates = { + (int(product_id), str(kind), str(document_hash)) + for product_id, kind, document_hash, _status, _observed_at + in existing_candidate_rows + } + for product_id, kind, document_hash, status, _observed_at in existing_candidate_rows: + if status == "approved": + accepted_documents.setdefault( + (int(product_id), str(kind)), str(document_hash) + ) + created = 0 + errors: list[str] = [] + for kind, rows in (("suitability", suitability_rows), ("contract", contract_rows)): + for row in rows: + product_id = product_ids.get(row["product_code"]) + document_hash = row["document_sha256"] + if product_id is None: + errors.append(f"missing_product:{row['product_code']}") + continue + if accepted_documents.get((product_id, kind)) == document_hash or ( + product_id, kind, document_hash + ) in existing_candidates: + continue + session.add(AdvisorProductGovernanceCandidate( + product_id=product_id, + data_kind=kind, + sales_institution=( + row.get("sales_institution") if kind == "suitability" else None + ), + observed_at=now, + effective_from=date.fromisoformat(row["effective_from"]), + risk_level=row.get("risk_level") if kind == "suitability" else None, + payload=dict(row), + source_url=row["source_url"], + document_title=row["document_title"], + document_published_at=self._optional_date(row.get("document_published_at")), + document_sha256=document_hash, + source=row["source"], + review_status="pending_review", + reviewed_by=None, + reviewed_at=None, + review_comment=None, + created_at=now, + updated_at=now, + )) + created += 1 + run.status = "succeeded" + run.product_count = len(product_codes) + run.change_count = created + run.error_count = len(errors) + run.detail = {"errors": errors[:20]} + run.completed_at = now + run.updated_at = now + return GovernanceMonitorResult(len(product_codes), created, len(errors), run_no) + + @staticmethod + def _optional_date(value: str | None) -> date | None: + return date.fromisoformat(value) if value else None + + @staticmethod + def _candidate_view(candidate: AdvisorProductGovernanceCandidate) -> dict[str, object]: + return { + "id": candidate.id, + "product_id": candidate.product_id, + "data_kind": candidate.data_kind, + "sales_institution": candidate.sales_institution, + "risk_level": candidate.risk_level, + "source_url": candidate.source_url, + "document_title": candidate.document_title, + "document_sha256": candidate.document_sha256, + "review_status": candidate.review_status, + "observed_at": candidate.observed_at.isoformat(), + "reviewed_at": candidate.reviewed_at.isoformat() if candidate.reviewed_at else None, + "review_comment": candidate.review_comment, + } + + @staticmethod + def _default_loader(timeout: float) -> tuple[list[dict[str, str]], list[dict[str, str]]]: + from tools.sync_nanfang_official_product_governance import collect_rows + + return collect_rows(timeout=timeout) diff --git a/app/service/product_history_sync_service.py b/app/service/product_history_sync_service.py new file mode 100644 index 0000000..e8a64b4 --- /dev/null +++ b/app/service/product_history_sync_service.py @@ -0,0 +1,129 @@ +"""Synchronize public fund NAV history into the additive advisory history store.""" + +import asyncio +from collections.abc import Callable +from dataclasses import dataclass +from datetime import UTC, date, datetime, timedelta +from decimal import Decimal, InvalidOperation +from typing import Any + +from sqlalchemy import select +from sqlalchemy.dialects.mysql import insert + +from app.infrastructure.db import SessionFactory +from app.model.advisor_product import AdvisorProductPriceHistory +from app.model.fund import FundProduct + +NavHistoryLoader = Callable[[str, str, str], list[dict[str, str]]] + + +@dataclass(frozen=True) +class ProductHistorySyncResult: + product_count: int + observation_count: int + + +class ProductHistorySyncService: + """Writes fund NAV observations only; it never writes immutable baseline tables.""" + + SOURCE = "eastmoney_hq_nav" + + def __init__( + self, + *, + session_factory: Callable[[], Any] = SessionFactory, + loader: NavHistoryLoader | None = None, + concurrency: int = 4, + ) -> None: + self.session_factory = session_factory + self.loader = loader or self._default_loader + self.concurrency = max(1, concurrency) + + async def sync( + self, + *, + days: int = 400, + limit: int = 100, + as_of_date: date | None = None, + product_codes: tuple[str, ...] | None = None, + ) -> ProductHistorySyncResult: + end_date = as_of_date or date.today() + start_date = end_date - timedelta(days=max(1, days)) + product_statement = select(FundProduct.id, FundProduct.product_code).where( + FundProduct.fund_manager == "南方基金", + FundProduct.status == "上市", + ) + if product_codes is not None: + product_statement = product_statement.where( + FundProduct.product_code.in_(product_codes) + ) + async with self.session_factory() as session: + products = list(await session.execute( + product_statement.order_by(FundProduct.id).limit(limit) + )) + + semaphore = asyncio.Semaphore(self.concurrency) + + async def fetch(product_id: int, product_code: str) -> tuple[int, list[dict[str, str]]]: + async with semaphore: + try: + rows = await asyncio.to_thread( + self.loader, product_code, start_date.isoformat(), end_date.isoformat() + ) + except Exception: + rows = [] + return product_id, rows + + fetched = await asyncio.gather( + *(fetch(int(product_id), str(product_code)) for product_id, product_code in products) + ) + now = datetime.now(UTC).replace(tzinfo=None) + observations = 0 + for product_id, rows in fetched: + payload = [] + for row in rows: + parsed = self._row(product_id, row, now) + if parsed is not None: + payload.append(parsed) + if not payload: + continue + async with self.session_factory() as session, session.begin(): + upsert_statement = insert(AdvisorProductPriceHistory).values(payload) + await session.execute(upsert_statement.on_duplicate_key_update( + close_price=upsert_statement.inserted.close_price, + turnover_amount=upsert_statement.inserted.turnover_amount, + source=upsert_statement.inserted.source, + source_updated_at=upsert_statement.inserted.source_updated_at, + updated_at=upsert_statement.inserted.updated_at, + )) + observations += len(payload) + return ProductHistorySyncResult(len(products), observations) + + @classmethod + def _row( + cls, product_id: int, raw: dict[str, str], now: datetime + ) -> dict[str, object] | None: + try: + trade_date = date.fromisoformat(raw["trade_date"]) + close_price = Decimal(raw["nav"]) + except (KeyError, InvalidOperation, ValueError): + return None + if close_price <= 0: + return None + return { + "product_id": product_id, + "trade_date": trade_date, + "price_kind": "fund_nav", + "close_price": close_price, + "turnover_amount": None, + "source": cls.SOURCE, + "source_updated_at": now, + "created_at": now, + "updated_at": now, + } + + @staticmethod + def _default_loader(fund_code: str, start_date: str, end_date: str) -> list[dict[str, str]]: + from hq import get_southern_fund_nav_history + + return get_southern_fund_nav_history(fund_code, start_date, end_date) diff --git a/app/service/product_metric_service.py b/app/service/product_metric_service.py new file mode 100644 index 0000000..4dd9b24 --- /dev/null +++ b/app/service/product_metric_service.py @@ -0,0 +1,117 @@ +"""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 + ) diff --git a/docs/21-投顾Agent迁移TODO.md b/docs/21-投顾Agent迁移TODO.md index e516e22..35e394b 100644 --- a/docs/21-投顾Agent迁移TODO.md +++ b/docs/21-投顾Agent迁移TODO.md @@ -39,6 +39,18 @@ 原 `jr_agent` 库仍记录旧投顾迁移版本 `20260911_adv_profile_tags`，不能直接用新底座升级， 已保留不动，待数据库负责人按迁移方案另行切换。 +### 阶段五：产品数据和适当性（进行中） + +已恢复 `hq.py` 公开数据适配器、场内产品导入工具、南方官网适当性/合同同步工具， +并新增投顾证据只读模型与 Repository。独立测试库已导入 19 个南方场内 ETF/LOF， +同步 19 条官网适当性披露和 19 条合同证据，实际权威风险分布为 R1=1、R2=4、R3=6、 +R4=7、R5=1；资产分类成功 18 个，1 个因合同证据不足跳过。每个等级至少四个产品 +不能由系统伪造，保留为未完成验收项。 + +阶段五已完成部分测试：产品证据 Repository `2 passed`，官网治理解析 `3 passed`， +治理导入/分类导入 `5 passed`，合计专项 `10 passed`；Ruff、MyPy、数据库结构审计和 +约束审计通过。阶段五尚未完成流动性指标细化和完整推荐侧适当性联动，暂不提交阶段完成标记。 + ## 一、迁移准备 - [ ] 确认远程仓库可访问。（当前失败：连接 `47.106.207.27:3000` 被拒绝） @@ -143,22 +155,22 @@ python tools/audit_constraints.py ## 五、产品数据和适当性 -- [ ] 迁移场内基金产品模型。 -- [ ] 迁移产品查询 Repository。 -- [ ] 导入南方场内基金产品。 -- [ ] 导入 R1-R5 产品适当性等级。 -- [ ] 导入基金合同字段和权威来源。 -- [ ] 导入销售机构披露信息。 -- [ ] 导入产品状态和交易状态。 -- [ ] 接入产品治理变更监控。 -- [ ] 接入产品资产规模指标。 -- [ ] 接入产品历史行情指标。 +- [x] 迁移场内基金产品模型。（复用 `FundProduct`，新增证据表只读映射） +- [x] 迁移产品查询 Repository。（`AdvisorProductRepository`） +- [x] 导入南方场内基金产品。（独立库 19 个 ETF/LOF） +- [x] 导入 R1-R5 产品适当性等级。（官网披露；五级均有，实际分布见阶段记录） +- [x] 导入基金合同字段和权威来源。（19 条官网合同证据） +- [x] 导入销售机构披露信息。（南方基金直销披露） +- [x] 导入产品状态和交易状态。（仅导入 `SSE/SZSE` 且 `上市` 产品） +- [x] 接入产品治理变更监控。（`ProductGovernanceMonitorService` + 官网同步工具） +- [x] 接入产品资产规模指标。（`advisor_product_reference_snapshot`） +- [x] 接入产品历史行情指标。（增量 NAV 同步 + `ProductMetricService`） - [ ] 接入产品流动性指标。 -- [ ] 实现场内基金过滤。 +- [x] 实现场内基金过滤。（Repository 强制 `SSE/SZSE`） - [ ] 实现适当性硬过滤。 -- [ ] 实现合同证据过滤。 -- [ ] 实现来源缺失时的失败关闭。 -- [ ] 完成产品数据提交 `advisor/product-data`。 +- [x] 实现合同证据过滤。（verified + 有来源 URL/文档摘要） +- [x] 实现来源缺失时的失败关闭。（缺失权威证据不进入候选） +- [ ] 完成产品数据提交 `advisor/product-data`。（阶段五进行中） 验收： diff --git a/hq.py b/hq.py new file mode 100644 index 0000000..728b3b1 --- /dev/null +++ b/hq.py @@ -0,0 +1,502 @@ +"""南方基金指定产品行情模块。""" +# Public-source adapter retains readable request expressions; line-length checks are not useful here. +# ruff: noqa: E501 +from __future__ import annotations + +import logging +import re +import time +from datetime import date, datetime +from datetime import time as clock_time +from html import unescape +from typing import Any + +import httpx + +logger = logging.getLogger(__name__) + +NAV_API = "https://api.fund.eastmoney.com/f10/lsjz" +RETURN_API = "https://api.fund.eastmoney.com/pinzhong/LJSYLZS" +QUOTE_API = "https://push2.eastmoney.com/api/qt/ulist.np/get" +TENCENT_QUOTE_API = "https://qt.gtimg.cn/q=" +DETAIL_API = "https://fund.eastmoney.com/pingzhongdata/{code}.js" +SOUTHERN_COMPANY_API = "https://fund.eastmoney.com/company/80000220.html" +REQUEST_TIMEOUT = 12.0 +MAX_FUNDS_PER_CALL = 1000 +FUND_TYPE_GROUPS = { + "货币型": ("202308", "020480", "511810"), + "债券型": ("007161", "003776", "020281", "511070", "159700", "160128", "160129"), + "混合型": ("018019", "014189", "018020", "160105", "160142", "160143", "501062"), + "股票型": ( + "020553", "016449", "008854", "008264", "008736", "010592", "160127", "588890", + "020839", "589700", "159382", "159511", "002900", "021958", "159948", "009059", + "001421", "510500", + ), + "QDII": ("501018", "159329", "159615", "159687"), +} +SOUTHERN_FUND_CODES = tuple( + dict.fromkeys(code for codes in FUND_TYPE_GROUPS.values() for code in codes) +) +FUND_TYPE_BY_CODE = { + code: fund_type for fund_type, codes in FUND_TYPE_GROUPS.items() for code in codes +} +HEADERS = {"User-Agent": "Mozilla/5.0", "Referer": "https://fund.eastmoney.com/"} +_history_date: str | None = None +_history_cache: dict[str, dict[str, str | None]] = {} +_name_cache: dict[str, str] = {} + + +class ExchangeQuoteSourceError(RuntimeError): + """A public exchange quote provider could not supply a usable response.""" + + +def is_market_trading_time(now: datetime | None = None) -> bool: + """判断中国大陆工作日盘中时段。""" + current = now or datetime.now() + if current.weekday() >= 5: + return False + current_time = current.time() + return (clock_time(9, 30) <= current_time <= clock_time(11, 30) + or clock_time(13, 0) <= current_time <= clock_time(15, 0)) + + +def get_southern_fund_market( + target_date: str | None = None, + limit: int | None = None, + fund_type: str | None = None, + fund_codes: list[str] | tuple[str, ...] | None = None, +) -> list[dict[str, Any]]: + """获取指定南方基金的完整行情表。 + + 每次调用刷新整张表的实时行情；历史收益同一日期只请求一次并保存在进程缓存。 + limit 不传时返回 SOUTHERN_FUND_CODES 中的全部产品。 + """ + query_date = target_date or date.today().isoformat() + date.fromisoformat(query_date) + if fund_type and fund_type not in FUND_TYPE_GROUPS: + raise ValueError("基金类型不在南方基金白名单内") + available_codes = FUND_TYPE_GROUPS[fund_type] if fund_type else SOUTHERN_FUND_CODES + if fund_codes is not None: + requested = tuple(dict.fromkeys(fund_codes)) + if any(code not in SOUTHERN_FUND_CODES for code in requested): + raise ValueError("基金代码不在南方基金白名单内") + available_codes = tuple(code for code in requested if code in available_codes) + count = len(available_codes) if limit is None else min(max(limit, 1), MAX_FUNDS_PER_CALL) + codes = list(available_codes[:count]) + names = _get_names(codes) + history = _get_history(codes, query_date) + quotes = _get_quotes(codes) if is_market_trading_time() else {} + now = time.strftime("%Y-%m-%d") + rows = [] + for code in codes: + old = history.get(code, {}) + live = quotes.get(code, {}) + rows.append({ + "基金代码": code, "基金名称": names.get(code, f"南方基金 {code}"), + "基金类型": FUND_TYPE_BY_CODE.get(code, "未分类"), + "基金净值": live.get("基金净值") or old.get("基金净值"), + "日期": live.get("日期") or old.get("日期"), + "日涨幅": live.get("日涨幅") or old.get("日涨幅"), + "最近半年": old.get("最近半年"), "最近一年": old.get("最近一年"), + "今年以来": old.get("今年以来"), "成立以来": old.get("成立以来"), + "行情时间": now, + "行情来源": "盘中实时行情" if live.get("基金净值") else "收盘后最新净值", + "是否盘中": is_market_trading_time(), + }) + return rows + + +def get_southern_fund_nav_history( + fund_code: str, start_date: str, end_date: str +) -> list[dict[str, str]]: + """Return validated historical unit-NAV observations for an allowed fund.""" + if fund_code not in SOUTHERN_FUND_CODES: + raise ValueError("基金代码不在南方基金白名单内") + start = date.fromisoformat(start_date) + end = date.fromisoformat(end_date) + if start > end: + raise ValueError("开始日期不能晚于结束日期") + records = _fetch_nav_records( + fund_code, start_date=start.isoformat(), end_date=end.isoformat(), all_pages=True + ) + observations: list[dict[str, str]] = [] + for record in records: + value_date = str(record.get("FSRQ") or "") + nav = str(record.get("DWJZ") or "").strip() + try: + parsed_date = date.fromisoformat(value_date) + if parsed_date < start or parsed_date > end or float(nav) <= 0: + continue + except ValueError: + continue + observations.append({"fund_code": fund_code, "trade_date": value_date, "nav": nav}) + return sorted(observations, key=lambda item: item["trade_date"]) + + +def get_southern_fund_catalog( + fund_codes: list[str] | tuple[str, ...], +) -> list[dict[str, Any]]: + """Return Southern Fund's public catalogue rows. + + The company directory supplies fund type and reported asset scale, which are + needed to select test products. It is reference data only: product risk is + intentionally not inferred here because the source does not publish a + channel-independent R1-R5 suitability rating. + """ + requested = tuple(dict.fromkeys(fund_codes)) + if not requested: + return [] + response = httpx.get(SOUTHERN_COMPANY_API, headers=HEADERS, timeout=REQUEST_TIMEOUT) + response.raise_for_status() + content = response.content.decode("utf-8") + scale_as_of_date = _company_scale_as_of_date(content) + rows = [] + for code in requested: + row = _company_catalog_row(content, code) + if row is not None: + row["scale_as_of_date"] = scale_as_of_date + rows.append(row) + return rows + + +def get_southern_exchange_catalog( + fund_codes: list[str] | tuple[str, ...], +) -> list[dict[str, Any]]: + """Backward-compatible alias for callers that only request exchange codes.""" + return get_southern_fund_catalog(fund_codes) + + +def _company_catalog_row(content: str, fund_code: str) -> dict[str, Any] | None: + marker = f'class="code">{fund_code}' + position = content.find(marker) + if position < 0: + return None + start = content.rfind("", position) + if start < 0 or end < 0: + return None + row = content[start:end + len("")] + name_match = re.search(r'class="name" title="([^"]+)"', row) + cells = [ + _html_cell_text(match.group(1)) + for match in re.finditer(r"]*>(.*?)", row, flags=re.DOTALL) + ] + if name_match is None or len(cells) < 10: + return None + return { + "fund_code": fund_code, + "fund_name": unescape(name_match.group(1)).strip(), + "fund_type": cells[2], + "nav_date": _catalog_nav_date(cells[3]), + "nav": cells[4], + "fund_asset_scale_billion": cells[9], + "trading_venue": cells[11] if len(cells) > 11 else "场内交易", + } + + +def _company_scale_as_of_date(content: str) -> str | None: + match = re.search(r"(?:数据截止|截止日期)[：:]\s*(20\d{2}-\d{2}-\d{2})", content) + return match.group(1) if match else None + + +def _catalog_nav_date(value: str) -> str | None: + match = re.fullmatch(r"(\d{2})-(\d{2})", value) + if match is None: + return None + return f"{date.today().year}-{match.group(1)}-{match.group(2)}" + + +def _html_cell_text(value: str) -> str: + return re.sub(r"\s+", " ", re.sub(r"<[^>]+>", "", unescape(value))).strip() + + +def _get_names(codes: list[str]) -> dict[str, str]: + for code in codes: + if code in _name_cache: + continue + try: + response = httpx.get(DETAIL_API.format(code=code), headers=HEADERS, timeout=REQUEST_TIMEOUT) + response.raise_for_status() + match = re.search(r"var\s+fS_name\s*=\s*[\"']([^\"']+)", response.text) + _name_cache[code] = match.group(1).strip() if match else f"南方基金 {code}" + except (httpx.HTTPError, UnicodeError) as exc: + logger.warning("基金名称接口失败 code=%s error=%s", code, type(exc).__name__) + _name_cache[code] = f"南方基金 {code}" + return {code: _name_cache.get(code, f"南方基金 {code}") for code in codes} + + +def _get_quotes(codes: list[str]) -> dict[str, dict[str, str | None]]: + secids = ",".join(("1." if code.startswith(("5", "6", "9")) else "0.") + code for code in codes) + try: + response = httpx.get(QUOTE_API, params={"fltt": 2, "invt": 2, "fields": "f12,f2,f3", "secids": secids}, headers=HEADERS, timeout=REQUEST_TIMEOUT) + response.raise_for_status() + items = ((response.json().get("data") or {}).get("diff") or []) + except (httpx.HTTPError, ValueError, TypeError) as exc: + logger.warning("实时行情接口失败 count=%s error=%s", len(codes), type(exc).__name__) + return {} + return {str(item["f12"]): {"基金净值": str(item["f2"]) if item.get("f2") not in (None, "-") else None, "日期": None, "日涨幅": f"{item.get('f3')}%" if item.get("f3") not in (None, "-") else None} for item in items if item.get("f12")} + + +def get_southern_exchange_quotes( + fund_codes: list[str] | tuple[str, ...], +) -> dict[str, dict[str, str | None]]: + """Backward-compatible Eastmoney quote lookup that degrades to an empty result.""" + try: + return fetch_southern_exchange_quotes_eastmoney(fund_codes) + except ExchangeQuoteSourceError as exc: + logger.warning("exchange quote endpoint failed error=%s", type(exc).__name__) + return {} + + +def fetch_southern_exchange_quotes_eastmoney( + fund_codes: list[str] | tuple[str, ...], +) -> dict[str, dict[str, str | None]]: + """Return Eastmoney exchange quotes and raise on a provider-level failure.""" + requested = tuple(dict.fromkeys(fund_codes)) + if any(code not in SOUTHERN_FUND_CODES for code in requested): + raise ValueError("fund code is not in the Southern Fund whitelist") + if not requested: + return {} + secids = ",".join( + ("1." if code.startswith(("5", "6", "9")) else "0.") + code for code in requested + ) + try: + response = httpx.get( + QUOTE_API, + params={ + "fltt": 2, + "invt": 2, + "fields": "f12,f2,f3,f5,f6,f17,f18", + "secids": secids, + }, + headers=HEADERS, + timeout=REQUEST_TIMEOUT, + ) + response.raise_for_status() + items = ((response.json().get("data") or {}).get("diff") or []) + except (httpx.HTTPError, ValueError, TypeError) as exc: + raise ExchangeQuoteSourceError("eastmoney quote request failed") from exc + + def value(item: dict[str, Any], field: str) -> str | None: + raw = item.get(field) + return str(raw) if raw not in (None, "-") else None + + return { + str(item["f12"]): { + "last_price": value(item, "f2"), + "change_pct": value(item, "f3"), + "volume": value(item, "f5"), + "turnover_amount": value(item, "f6"), + "open_price": value(item, "f17"), + "previous_close": value(item, "f18"), + } + for item in items + if item.get("f12") + } + + +def fetch_southern_exchange_quotes_tencent( + fund_codes: list[str] | tuple[str, ...], +) -> dict[str, dict[str, str | None]]: + """Return Tencent Finance exchange quotes as an independent fallback source. + + Tencent publishes volume in lots. Turnover is deliberately left unset + because its payload does not provide a field with compatible semantics. + """ + requested = tuple(dict.fromkeys(fund_codes)) + if any(code not in SOUTHERN_FUND_CODES for code in requested): + raise ValueError("fund code is not in the Southern Fund whitelist") + if not requested: + return {} + symbols = ",".join( + ("sh" if code.startswith(("5", "6", "9")) else "sz") + code + for code in requested + ) + try: + response = httpx.get( + TENCENT_QUOTE_API + symbols, + headers={"User-Agent": HEADERS["User-Agent"]}, + timeout=REQUEST_TIMEOUT, + ) + response.raise_for_status() + except httpx.HTTPError as exc: + raise ExchangeQuoteSourceError("tencent quote request failed") from exc + return _parse_tencent_exchange_quotes(response.content, requested) + + +def _parse_tencent_exchange_quotes( + content: bytes, requested: tuple[str, ...], +) -> dict[str, dict[str, str | None]]: + try: + payload = content.decode("gbk") + except UnicodeDecodeError as exc: + raise ExchangeQuoteSourceError("tencent quote response decoding failed") from exc + result: dict[str, dict[str, str | None]] = {} + for match in re.finditer(r'v_(?:sh|sz)(\d{6})="([^"]*)"', payload): + code, raw = match.groups() + if code not in requested: + continue + fields = raw.split("~") + if len(fields) < 7 or not _valid_quote_number(fields[3]): + continue + previous_close = fields[4] if _valid_quote_number(fields[4]) else None + change_pct = fields[33] if len(fields) > 33 and _valid_quote_number(fields[33]) else None + if change_pct is None and previous_close is not None: + change_pct = _quote_change_pct(fields[3], previous_close) + result[code] = { + "last_price": fields[3], + "previous_close": previous_close, + "change_pct": change_pct, + "volume": fields[6] if _valid_quote_number(fields[6]) else None, + "turnover_amount": None, + } + if not result: + raise ExchangeQuoteSourceError("tencent quote response had no usable quotes") + return result + + +def _valid_quote_number(value: str) -> bool: + try: + return float(value) > 0 + except ValueError: + return False + + +def _quote_change_pct(last_price: str, previous_close: str) -> str | None: + try: + return str(round((float(last_price) / float(previous_close) - 1) * 100, 4)) + except (ValueError, ZeroDivisionError): + return None + + +def _get_history(codes: list[str], query_date: str) -> dict[str, dict[str, str | None]]: + global _history_date, _history_cache + if _history_date == query_date and all(code in _history_cache for code in codes): + return _history_cache + result = {} + for code in codes: + try: + result[code] = _get_history_snapshot(code, query_date) + except (httpx.HTTPError, ValueError, TypeError) as exc: + logger.warning("历史净值计算失败 code=%s error=%s", code, type(exc).__name__) + result[code] = _empty_returns() + _history_date, _history_cache = query_date, result + return result + + +def _get_history_snapshot(fund_code: str, query_date: str) -> dict[str, str | None]: + """读取净值与累计收益率快照。""" + latest_records = _fetch_nav_records(fund_code, all_pages=False) + valid = _valid_records(latest_records) + if not valid: + return _empty_returns() + target = date.fromisoformat(query_date) + latest_date, _, latest = next((item for item in valid if item[0] == target), valid[0]) + return { + "基金净值": latest.get("DWJZ"), + "日期": latest.get("FSRQ"), + "日涨幅": _format_percent(latest.get("JZZZL")), + "最近半年": _fetch_return_rate(fund_code, "6月"), + "最近一年": _fetch_return_rate(fund_code, "1年"), + "今年以来": _fetch_return_rate(fund_code, "今年来"), + "成立以来": _fetch_return_rate(fund_code, "成立来"), + } + + +def _fetch_nav_page(fund_code: str, page_index: int = 1, start_date: str | None = None, end_date: str | None = None) -> tuple[list[dict[str, Any]], int | None]: + """读取一页历史净值，并返回接口提供的总条数。""" + response = httpx.get(NAV_API, params={"fundCode": fund_code, "pageIndex": page_index, "pageSize": 30, "startDate": start_date or "", "endDate": end_date or ""}, headers=HEADERS, timeout=REQUEST_TIMEOUT) + response.raise_for_status() + data = response.json().get("Data") or {} + records = data.get("LSJZList") or [] + total = data.get("TotalCount") or data.get("totalCount") + try: + total = int(total) if total is not None else None + except (TypeError, ValueError): + total = None + return records, total + + +def _fetch_nav_records(fund_code: str, start_date: str | None = None, end_date: str | None = None, all_pages: bool = False) -> list[dict[str, Any]]: + """调用历史净值接口；区间收益需要时读取完整分页，避免重复使用同一基准日。""" + records, total = _fetch_nav_page(fund_code, 1, start_date, end_date) + if not all_pages or (total is not None and total <= len(records)): + return records + if not (start_date and end_date): + return records + page_size = max(len(records), 1) + page_count = min((total + page_size - 1) // page_size, 40) if total else 40 + seen_dates = {str(item.get("FSRQ") or "") for item in records} + for page_index in range(2, page_count + 1): + page_records, _ = _fetch_nav_page(fund_code, page_index, start_date, end_date) + if not page_records: + break + new_records = [ + item for item in page_records if str(item.get("FSRQ") or "") not in seen_dates + ] + if not new_records: + break + seen_dates.update(str(item.get("FSRQ") or "") for item in new_records) + records.extend(new_records) + if len(page_records) < page_size: + break + return records + + +def _fetch_return_rate(fund_code: str, period: str) -> str | None: + """通过东方财富累计收益率接口读取指定区间的最新收益率。""" + period_map = { + "1月": "m", + "3月": "q", + "6月": "hy", + "1年": "y", + "3年": "try", + "5年": "fiy", + "今年来": "sy", + "成立来": "se", + } + try: + response = httpx.get( + RETURN_API, + params={"fundCode": fund_code, "indexcode": "000300", "type": period_map[period]}, + headers={"Referer": "https://fund.eastmoney.com/"}, + timeout=REQUEST_TIMEOUT, + ) + response.raise_for_status() + payload = response.json() + series = (((payload.get("Data") or [{}])[0]).get("data") or []) + if not series: + return None + latest = series[-1] + if isinstance(latest, dict): + value = latest.get("y") + elif isinstance(latest, (list, tuple)) and len(latest) >= 2: + value = latest[1] + else: + value = None + if value in (None, ""): + return None + return _format_percent(value) + except (httpx.HTTPError, ValueError, TypeError, KeyError, IndexError) as exc: + logger.warning("累计收益率接口失败 code=%s period=%s error=%s", fund_code, period, type(exc).__name__) + return None + +def _empty_returns() -> dict[str, str | None]: + return {key: None for key in ("基金净值", "日期", "日涨幅", "最近半年", "最近一年", "今年以来", "成立以来")} + + +def _valid_records(records: list[dict[str, Any]]) -> list[tuple[date, float, dict[str, Any]]]: + valid = [] + for item in records: + try: + valid.append((date.fromisoformat(item["FSRQ"]), float(item["DWJZ"]), item)) + except (KeyError, TypeError, ValueError): + continue + return sorted(valid, key=lambda item: item[0], reverse=True) + + +def _format_percent(value: Any) -> str | None: + if value in (None, ""): + return None + text = str(value).strip() + return text if text.endswith("%") else f"{text}%" diff --git a/tests/unit/repository/test_advisor_product_repository.py b/tests/unit/repository/test_advisor_product_repository.py new file mode 100644 index 0000000..80ac264 --- /dev/null +++ b/tests/unit/repository/test_advisor_product_repository.py @@ -0,0 +1,104 @@ +from datetime import date, datetime +from decimal import Decimal +from typing import Any + +import pytest + +from app.model.advisor_product import ( + AdvisorProductContractSnapshot, + AdvisorProductReferenceSnapshot, + AdvisorProductSuitabilityReference, +) +from app.model.fund import FundProduct +from app.repository.advisor_product_repository import AdvisorProductRepository + + +class ScalarResult: + def __init__(self, rows: list[Any]) -> None: + self.rows = rows + + def __iter__(self): + return iter(self.rows) + + +class FakeSession: + def __init__(self, results: list[list[Any]]) -> None: + self.results = iter(results) + self.statements: list[Any] = [] + + async def scalars(self, statement: Any) -> ScalarResult: + self.statements.append(statement) + return ScalarResult(next(self.results)) + + +def product(product_id: int) -> FundProduct: + return FundProduct( + id=product_id, product_code=f"15{product_id:04d}", product_name="南方测试ETF", + exchange_code="SSE", product_category="ETF", risk_level="R5", fund_manager="南方基金", + status="上市", open_start_at=None, open_end_at=None, + ) + + +def suitability(product_id: int, risk_level: str = "R2") -> AdvisorProductSuitabilityReference: + return AdvisorProductSuitabilityReference( + product_id=product_id, sales_institution="南方基金管理股份有限公司直销", + risk_level=risk_level, effective_from=date(2026, 1, 1), effective_until=None, + source_url="https://example.test/risk.pdf", document_title="风险等级披露", + document_sha256="a" * 64, source="official", review_status="verified", + ) + + +def contract(product_id: int) -> AdvisorProductContractSnapshot: + return AdvisorProductContractSnapshot( + product_id=product_id, effective_from=date(2026, 1, 1), effective_until=None, + fund_type="股票型ETF", investment_scope="场内基金", risk_return_characteristics="净值波动", + source_url="https://example.test/contract.pdf", document_title="基金合同", + document_sha256="b" * 64, source="official", review_status="verified", + ) + + +@pytest.mark.asyncio +async def test_authoritative_candidates_fail_closed_and_keep_r1_without_scale_limit() -> None: + now = datetime(2026, 9, 11, 10, 0) + session = FakeSession([ + [product(1), product(2)], + [suitability(1, "R1"), suitability(2, "R2")], + [contract(1), contract(2)], + [ + AdvisorProductReferenceSnapshot( + product_id=1, + as_of_date=date(2026, 1, 1), + fund_type="货币ETF", + fund_asset_scale_billion=Decimal("0.1"), + ), + AdvisorProductReferenceSnapshot( + product_id=2, + as_of_date=date(2026, 1, 1), + fund_type="股票ETF", + fund_asset_scale_billion=Decimal("2"), + ), + ], + [], + [], + ]) + result = await AdvisorProductRepository(session).authoritative_tradable_products( + now, sales_institution="南方基金管理股份有限公司直销", + min_asset_scale_billion=Decimal("1"), + ) + + assert [item.product.id for item in result] == [1, 2] + assert result[0].suitability.risk_level == "R1" + assert result[0].asset_scale_billion == Decimal("0.1") + + +@pytest.mark.asyncio +async def test_pending_governance_and_missing_evidence_are_excluded() -> None: + now = datetime(2026, 9, 11, 10, 0) + session = FakeSession([ + [product(1)], [suitability(1)], [contract(1)], [], [1], [], + ]) + result = await AdvisorProductRepository(session).authoritative_tradable_products( + now, sales_institution="南方基金管理股份有限公司直销", + ) + assert result == [] + assert "exchange_code IN" in str(session.statements[0]) diff --git a/tests/unit/service/test_product_history_sync_service.py b/tests/unit/service/test_product_history_sync_service.py new file mode 100644 index 0000000..9c761ba --- /dev/null +++ b/tests/unit/service/test_product_history_sync_service.py @@ -0,0 +1,28 @@ +from datetime import UTC, datetime +from decimal import Decimal + +from app.service.product_history_sync_service import ProductHistorySyncService + + +def test_history_sync_row_accepts_a_positive_public_nav() -> None: + now = datetime.now(UTC).replace(tzinfo=None) + + result = ProductHistorySyncService._row( + 7, {"trade_date": "2026-09-09", "nav": "1.234500"}, now + ) + + assert result is not None + assert result["product_id"] == 7 + assert result["price_kind"] == "fund_nav" + assert result["close_price"] == Decimal("1.234500") + assert result["source"] == "eastmoney_hq_nav" + + +def test_history_sync_row_rejects_invalid_or_non_positive_nav() -> None: + now = datetime.now(UTC).replace(tzinfo=None) + + assert ProductHistorySyncService._row(7, {"trade_date": "bad", "nav": "1"}, now) is None + assert ProductHistorySyncService._row( + 7, {"trade_date": "2026-09-09", "nav": "0"}, now + ) is None + diff --git a/tests/unit/service/test_product_metric_service.py b/tests/unit/service/test_product_metric_service.py new file mode 100644 index 0000000..6e3220d --- /dev/null +++ b/tests/unit/service/test_product_metric_service.py @@ -0,0 +1,51 @@ +from dataclasses import dataclass +from datetime import date, timedelta +from decimal import Decimal + +from app.service.product_metric_service import ProductMetricService + + +@dataclass(frozen=True) +class Price: + trade_date: date + close_price: Decimal + turnover_amount: Decimal | None + + +def price(day: int, close: str, turnover: str = "1000") -> Price: + return Price( + trade_date=date(2026, 1, 1) + timedelta(days=day), + close_price=Decimal(close), + turnover_amount=Decimal(turnover), + ) + + +def test_metric_calculation_uses_historical_prices_without_forecast() -> None: + prices = [price(day, str(100 + day), str(1000 + day)) for day in range(121)] + + metrics = ProductMetricService.calculate(prices) + + assert metrics.trailing_20d_return_pct == Decimal("10.0000") + assert metrics.trailing_120d_return_pct == Decimal("120.0000") + assert metrics.annualized_volatility_pct is not None + assert metrics.max_drawdown_pct == Decimal("0.0000") + assert metrics.average_daily_turnover_amount == Decimal("1110.50") + assert metrics.observation_count == 121 + + +def test_metric_calculation_keeps_insufficient_samples_empty() -> None: + metrics = ProductMetricService.calculate([price(0, "100"), price(1, "90")]) + + assert metrics.trailing_20d_return_pct is None + assert metrics.trailing_120d_return_pct is None + assert metrics.annualized_volatility_pct is None + assert metrics.max_drawdown_pct == Decimal("-10.0000") + + +def test_snapshot_carries_source_and_calculation_version() -> None: + snapshot = ProductMetricService.snapshot(7, [price(0, "100"), price(1, "101")]) + + assert snapshot is not None + assert snapshot.source == "advisor_product_price_history_dual_read" + assert snapshot.calculation_version == "v1" + assert snapshot.as_of_date == date(2026, 1, 2) diff --git a/tests/unit/tools/test_nanfang_official_product_governance.py b/tests/unit/tools/test_nanfang_official_product_governance.py new file mode 100644 index 0000000..4cbd6d9 --- /dev/null +++ b/tests/unit/tools/test_nanfang_official_product_governance.py @@ -0,0 +1,79 @@ +from __future__ import annotations + +from datetime import UTC, datetime + +import pytest + +from tools.sync_nanfang_official_product_governance import ( + collect_official_rows, + contract_document, +) + + +def overview() -> dict[str, object]: + return { + "code": "ETS-5BP00000", + "data": { + "fundRiskRating": {"RISKRATING": "\u4e2d\u4f4e\u98ce\u9669(R2)"}, + "fund_info": { + "fundName": "\u5357\u65b9\u6d4b\u8bd5ETF", + "fundDate": "20240101", + "contractValidDate": "20240101", + "basedetailType": "\u503a\u5238\u578b\u57fa\u91d1", + "tzfw": "\u6d4b\u8bd5\u6295\u8d44\u8303\u56f4", + "yjbjjz": "\u6d4b\u8bd5\u4e1a\u7ee9\u6bd4\u8f83\u57fa\u51c6", + "jjtgr": "\u6d4b\u8bd5\u6258\u7ba1\u4eba", + "glYearRatio": 0.5, + "tgYearRatio": 0.1, + }, + }, + } + + +def documents() -> list[dict[str, object]]: + return [ + { + "title": "\u5357\u65b9\u6d4b\u8bd5ETF\u6258\u7ba1\u534f\u8bae", + "linkUrl": "/bad.doc", + "publishTime": "2026-01-01", + }, + { + "title": "\u5357\u65b9\u6d4b\u8bd5ETF\u57fa\u91d1\u5408\u540c", + "linkUrl": "/main/files/contract.docx", + "createTimeString": "2024-01-01", + "publishTime": "2024-01-02", + }, + ] + + +def test_collects_official_rating_and_contract_hash() -> None: + suitability, contract = collect_official_rows( + "123456", + overview(), + documents(), + b"official-contract", + synced_at=datetime(2026, 9, 10, 8, 0, tzinfo=UTC), + ) + assert suitability["risk_level"] == "R2" + assert suitability["review_status"] == "verified" + assert contract["source_url"] == "https://www.nffund.com/main/files/contract.docx" + assert contract["document_sha256"] == ( + "0d53cbfbbcfea4857c2575dd2441ced3d6f53a56b34fdaa6b526ce6b755faaf1" + ) + assert contract["management_fee_rate_pct"] == "0.5" + + +def test_contract_document_excludes_custody_agreement() -> None: + assert contract_document(documents(), "123456")["linkUrl"] == "/main/files/contract.docx" + + +def test_requires_official_risk_rating() -> None: + invalid = overview() + data = invalid["data"] + assert isinstance(data, dict) + data["fundRiskRating"] = {"RISKRATING": "\u4e2d\u7b49\u98ce\u9669"} + with pytest.raises(ValueError, match="did not disclose"): + collect_official_rows( + "123456", invalid, documents(), b"contract", synced_at=datetime.now(UTC) + ) + diff --git a/tests/unit/tools/test_product_asset_classification_import.py b/tests/unit/tools/test_product_asset_classification_import.py new file mode 100644 index 0000000..0940dd9 --- /dev/null +++ b/tests/unit/tools/test_product_asset_classification_import.py @@ -0,0 +1,14 @@ +from tools.import_product_asset_classifications import classify_contract + + +def test_classify_contract_uses_official_fund_type_before_scope_keywords() -> None: + assert classify_contract("货币市场基金", "货币市场工具和银行存款") == "cash_management_etf" + assert classify_contract("债券型", "债券资产比例不低于基金资产的80%") == "bond_etf" + assert classify_contract("指数型", "标的指数成份股比例不低于90%") == "equity_etf" + + +def test_classify_contract_requires_stock_scope_for_hybrid_and_skips_fund_of_funds() -> None: + assert classify_contract("混合型", "股票投资比例范围为40%-95%") == "equity_etf" + assert classify_contract("混合型", "债券、现金及其他工具") is None + assert classify_contract("基金中基金", "主要投资于公募基金") is None + diff --git a/tests/unit/tools/test_product_governance_import.py b/tests/unit/tools/test_product_governance_import.py new file mode 100644 index 0000000..1e50d1f --- /dev/null +++ b/tests/unit/tools/test_product_governance_import.py @@ -0,0 +1,83 @@ +from __future__ import annotations + +from pathlib import Path + +import pytest + +from tools.import_product_governance_reference import ( + CONTRACT_FIELDS, + SUITABILITY_FIELDS, + normalize_contracts, + normalize_suitability, + read_rows, +) + + +def suitability_row() -> dict[str, str]: + return { + "product_code": "511810", + "sales_institution": "\u6d4b\u8bd5\u9500\u552e\u673a\u6784", + "risk_level": "R1", + "effective_from": "2026-01-01", + "effective_until": "", + "source_url": "https://institution.example/risk-disclosure.pdf", + "document_title": "\u4ea7\u54c1\u98ce\u9669\u7b49\u7ea7\u516c\u793a", + "document_published_at": "2026-01-01", + "document_sha256": "a" * 64, + "source": "sales_institution_disclosure", + "review_status": "verified", + "verified_by": "reviewer", + "verified_at": "2026-01-02T08:00:00+08:00", + "_line_number": "2", + } + + +def contract_row() -> dict[str, str]: + return { + "product_code": "511810", + "effective_from": "2026-01-01", + "effective_until": "", + "fund_type": "ETF", + "investment_scope": "\u8d27\u5e01\u5e02\u573a\u5de5\u5177", + "performance_benchmark": "\u4e03\u5929\u901a\u77e5\u5b58\u6b3e\u5229\u7387", + "risk_return_characteristics": "\u4f4e\u98ce\u9669\u3001\u4f4e\u6536\u76ca", + "custodian_name": "\u6d4b\u8bd5\u6258\u7ba1\u4eba", + "management_fee_rate_pct": "0.33", + "custodian_fee_rate_pct": "0.10", + "inception_date": "2013-01-01", + "source_url": "https://manager.example/contract.pdf", + "document_title": "\u57fa\u91d1\u5408\u540c", + "document_published_at": "2026-01-01", + "document_sha256": "b" * 64, + "source": "fund_manager_official", + "review_status": "verified", + "verified_by": "reviewer", + "verified_at": "2026-01-02T08:00:00+08:00", + "_line_number": "2", + } + + +def test_normalize_references_preserves_verification_and_contract_fields() -> None: + suitability = normalize_suitability([suitability_row()]) + contract = normalize_contracts([contract_row()]) + assert suitability[0]["risk_level"] == "R1" + assert suitability[0]["verified_at"] is not None + assert contract[0]["management_fee_rate_pct"] is not None + assert contract[0]["investment_scope"] == "\u8d27\u5e01\u5e02\u573a\u5de5\u5177" + + +def test_verified_reference_requires_reviewer_and_timestamp() -> None: + row = suitability_row() + row["verified_at"] = "" + with pytest.raises(ValueError, match="verified rows require"): + normalize_suitability([row]) + + +def test_reader_rejects_missing_required_columns(tmp_path: Path) -> None: + path = tmp_path / "bad.csv" + path.write_text("product_code,risk_level\n511810,R1\n", encoding="utf-8") + with pytest.raises(ValueError, match="missing CSV columns"): + read_rows(path, SUITABILITY_FIELDS) + with pytest.raises(ValueError, match="missing CSV columns"): + read_rows(path, CONTRACT_FIELDS) + diff --git a/tools/import_hq_test_products.py b/tools/import_hq_test_products.py new file mode 100644 index 0000000..9dedf9b --- /dev/null +++ b/tools/import_hq_test_products.py @@ -0,0 +1,356 @@ +"""Import verified exchange-traded Southern Fund products into the local test DB. + +The hq.py whitelist also contains off-exchange mutual funds. This command only +imports the ETF/LOF entries that belong to the platform's exchange-traded +simulation scope. It refreshes the product catalogue and its latest NAV cache; +it deliberately does not write historical rows because the immutable baseline +history tables currently have incompatible single-column unique constraints. +""" + +from __future__ import annotations + +import argparse +import sys +from datetime import UTC, date, datetime, time +from decimal import Decimal, InvalidOperation +from pathlib import Path +from typing import Any +from urllib.parse import unquote, urlparse + +import pymysql + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +TEST_PRODUCT_ID_START = 9_100_001 +FUND_MANAGER = "\u5357\u65b9\u57fa\u91d1" +LISTED_STATUS = "\u4e0a\u5e02" + +# The mapping is a test suitability reference, not an authoritative product +# rating. Eastmoney's public company catalogue has product type and size but +# does not expose a channel-independent R1-R5 suitability rating. +EXCHANGE_PRODUCT_REFERENCE: dict[str, tuple[str, str, str]] = { + "511810": ("SSE", "ETF", "R1"), + "511070": ("SSE", "ETF", "R2"), + "159700": ("SZSE", "ETF", "R2"), + "160128": ("SZSE", "LOF", "R2"), + "160129": ("SZSE", "LOF", "R2"), + "160105": ("SZSE", "LOF", "R3"), + "160142": ("SZSE", "LOF", "R3"), + "160143": ("SZSE", "LOF", "R3"), + "501062": ("SSE", "LOF", "R3"), + "510500": ("SSE", "ETF", "R4"), + "501018": ("SSE", "LOF", "R5"), + "159329": ("SZSE", "ETF", "R5"), + "159615": ("SZSE", "ETF", "R5"), + "159687": ("SZSE", "ETF", "R5"), + "160127": ("SZSE", "LOF", "R4"), + "588890": ("SSE", "ETF", "R4"), + "159382": ("SZSE", "ETF", "R4"), + "159511": ("SZSE", "ETF", "R4"), + "159948": ("SZSE", "ETF", "R4"), +} +# The platform only imports exchange-traded products. Keep the offsite +# reference path empty so routine refreshes never add offsite funds. +OFFSITE_R1_PRODUCT_CODES: tuple[str, ...] = () +RETIRED_TEST_PRODUCT_CODES = ("589700",) +MIN_ASSET_SCALE_BILLION = Decimal("1") +REFERENCE_SOURCE = "eastmoney_nanfang_catalog" +RISK_MAPPING_VERSION = "test-v1" + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Import hq.py exchange-traded test products") + parser.add_argument( + "--dry-run", action="store_true", help="Fetch and validate data without writing MySQL" + ) + return parser.parse_args() + + +def mysql_connection() -> pymysql.Connection: + from app.core.config import get_settings + + parsed = urlparse(get_settings().mysql_dsn.replace("mysql+asyncmy://", "mysql+pymysql://")) + return pymysql.connect( + host=parsed.hostname or "127.0.0.1", + port=parsed.port or 3306, + user=unquote(parsed.username or ""), + password=unquote(parsed.password or ""), + database=(parsed.path or "/").lstrip("/"), + charset="utf8mb4", + autocommit=False, + ) + + +def decimal_value(value: Any) -> Decimal | None: + if value in (None, "", "-"): + return None + try: + result = Decimal(str(value).strip().removesuffix("%")) + except (InvalidOperation, ValueError): + return None + return result if result > 0 else None + + +def non_negative_decimal_value(value: Any) -> Decimal | None: + if value in (None, "", "-"): + return None + try: + result = Decimal(str(value).strip().removesuffix("%").replace(",", "")) + except (InvalidOperation, ValueError): + return None + return result if result >= 0 else None + + +def nav_datetime(value: Any, fallback: datetime) -> datetime: + try: + nav_date = date.fromisoformat(str(value)) + except (TypeError, ValueError): + return fallback + return datetime.combine(nav_date, time(15, 0)) + + +def load_validated_rows() -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: + import hq + + codes = [*EXCHANGE_PRODUCT_REFERENCE, *OFFSITE_R1_PRODUCT_CODES] + catalog = { + str(row["fund_code"]): row + for row in hq.get_southern_fund_catalog(codes) + if isinstance(row, dict) and isinstance(row.get("fund_code"), str) + } + exchange_rows: list[dict[str, Any]] = [] + for code, (exchange_code, product_category, risk_level) in EXCHANGE_PRODUCT_REFERENCE.items(): + reference = catalog.get(code, {}) + product_name = str(reference.get("fund_name") or "").strip() + nav = decimal_value(reference.get("nav")) + scale = non_negative_decimal_value(reference.get("fund_asset_scale_billion")) + fund_type = str(reference.get("fund_type") or "").strip() + scale_as_of_date = reference.get("scale_as_of_date") + if ( + not product_name + or nav is None + or scale is None + or (risk_level != "R1" and scale < MIN_ASSET_SCALE_BILLION) + ): + print(f"skip code={code}: missing name, NAV, or scale below 1 billion CNY") + continue + if not fund_type or not isinstance(scale_as_of_date, str): + print(f"skip code={code}: missing fund type or asset scale date") + continue + exchange_rows.append( + { + "product_code": code, + "product_name": product_name, + "exchange_code": exchange_code, + "product_category": product_category, + "risk_level": risk_level, + "current_nav": nav, + "current_nav_at": nav_datetime(reference.get("nav_date"), datetime.now(UTC).replace(tzinfo=None)), + "fund_type": fund_type, + "fund_asset_scale_billion": scale, + "scale_as_of_date": date.fromisoformat(scale_as_of_date), + } + ) + offsite_rows: list[dict[str, Any]] = [] + for code in OFFSITE_R1_PRODUCT_CODES: + reference = catalog.get(code, {}) + product_name = str(reference.get("fund_name") or "").strip() + nav = decimal_value(reference.get("nav")) + fund_type = str(reference.get("fund_type") or "").strip() + scale_as_of_date = reference.get("scale_as_of_date") + if not product_name or nav is None or not fund_type: + print(f"skip offsite code={code}: missing name, NAV, or fund type") + continue + offsite_rows.append( + { + "fund_code": code, + "fund_name": product_name, + "fund_type": fund_type, + "risk_level": "R1", + "current_nav": nav, + "current_nav_at": nav_datetime( + reference.get("nav_date"), datetime.now(UTC).replace(tzinfo=None) + ), + "fund_asset_scale_billion": non_negative_decimal_value( + reference.get("fund_asset_scale_billion") + ), + "scale_as_of_date": ( + date.fromisoformat(scale_as_of_date) + if isinstance(scale_as_of_date, str) + else None + ), + } + ) + return exchange_rows, offsite_rows + + +def import_rows( + rows: list[dict[str, Any]], offsite_rows: list[dict[str, Any]], *, dry_run: bool +) -> tuple[int, int, int, int, int]: + if dry_run: + return 0, 0, 0, 0, 0 + now = datetime.now(UTC).replace(tzinfo=None) + connection = mysql_connection() + created = 0 + updated = 0 + snapshots = 0 + retired = 0 + offsite_imported = 0 + try: + with connection.cursor() as cursor: + cursor.execute("SELECT COALESCE(MAX(id), 0) FROM fin_product") + max_id = int(cursor.fetchone()[0]) + next_id = max(TEST_PRODUCT_ID_START, max_id + 1) + placeholders = ",".join("%s" for _ in RETIRED_TEST_PRODUCT_CODES) + cursor.execute( + f"UPDATE fin_product SET status=%s, updated_at=%s WHERE product_code IN ({placeholders})", + ("\u505c\u724c", now, *RETIRED_TEST_PRODUCT_CODES), + ) + retired = cursor.rowcount + for row in rows: + cursor.execute( + "SELECT id FROM fin_product WHERE product_code=%s", + (row["product_code"],), + ) + existing = cursor.fetchone() + values = {**row, "updated_at": now} + if existing: + product_id = int(existing[0]) + cursor.execute( + """ + UPDATE fin_product + SET product_name=%(product_name)s, exchange_code=%(exchange_code)s, + product_category=%(product_category)s, risk_level=%(risk_level)s, + fund_manager=%(fund_manager)s, current_nav=%(current_nav)s, + current_nav_at=%(current_nav_at)s, status=%(status)s, + updated_at=%(updated_at)s + WHERE id=%(id)s + """, + {**values, "id": product_id, "fund_manager": FUND_MANAGER, "status": LISTED_STATUS}, + ) + updated += 1 + else: + product_id = next_id + cursor.execute( + """ + INSERT INTO fin_product ( + id, product_code, product_name, exchange_code, product_category, + risk_level, fund_manager, currency, lot_size, price_tick, current_nav, + current_nav_at, min_amount, risk_disclosure_required, + second_confirmation_required, recording_required, status, created_at, updated_at + ) VALUES ( + %(id)s, %(product_code)s, %(product_name)s, %(exchange_code)s, + %(product_category)s, %(risk_level)s, %(fund_manager)s, 'CNY', + 100, 0.001, %(current_nav)s, %(current_nav_at)s, 0, 0, 0, 0, + %(status)s, %(created_at)s, %(updated_at)s + ) + """, + { + **values, + "id": product_id, + "fund_manager": FUND_MANAGER, + "status": LISTED_STATUS, + "created_at": now, + }, + ) + next_id += 1 + created += 1 + cursor.execute( + """ + INSERT INTO advisor_product_reference_snapshot ( + product_id, as_of_date, fund_type, fund_asset_scale_billion, source, + risk_mapping_version, created_at, updated_at + ) VALUES ( + %(product_id)s, %(as_of_date)s, %(fund_type)s, + %(fund_asset_scale_billion)s, %(source)s, + %(risk_mapping_version)s, %(created_at)s, %(updated_at)s + ) ON DUPLICATE KEY UPDATE + fund_type=VALUES(fund_type), + fund_asset_scale_billion=VALUES(fund_asset_scale_billion), + source=VALUES(source), + risk_mapping_version=VALUES(risk_mapping_version), + updated_at=VALUES(updated_at) + """, + { + **row, + "product_id": product_id, + "as_of_date": row["scale_as_of_date"], + "source": REFERENCE_SOURCE, + "risk_mapping_version": RISK_MAPPING_VERSION, + "created_at": now, + "updated_at": now, + }, + ) + snapshots += 1 + for row in offsite_rows: + cursor.execute( + """ + INSERT INTO advisor_offsite_fund_reference ( + fund_code, fund_name, fund_type, risk_level, current_nav, current_nav_at, + fund_asset_scale_billion, scale_as_of_date, source, risk_mapping_version, + status, created_at, updated_at + ) VALUES ( + %(fund_code)s, %(fund_name)s, %(fund_type)s, %(risk_level)s, + %(current_nav)s, %(current_nav_at)s, %(fund_asset_scale_billion)s, + %(scale_as_of_date)s, %(source)s, %(risk_mapping_version)s, + %(status)s, %(created_at)s, %(updated_at)s + ) ON DUPLICATE KEY UPDATE + fund_name=VALUES(fund_name), + fund_type=VALUES(fund_type), + risk_level=VALUES(risk_level), + current_nav=VALUES(current_nav), + current_nav_at=VALUES(current_nav_at), + fund_asset_scale_billion=VALUES(fund_asset_scale_billion), + scale_as_of_date=VALUES(scale_as_of_date), + source=VALUES(source), + risk_mapping_version=VALUES(risk_mapping_version), + status=VALUES(status), + updated_at=VALUES(updated_at) + """, + { + **row, + "source": REFERENCE_SOURCE, + "risk_mapping_version": RISK_MAPPING_VERSION, + "status": LISTED_STATUS, + "created_at": now, + "updated_at": now, + }, + ) + offsite_imported += 1 + connection.commit() + except Exception: + connection.rollback() + raise + finally: + connection.close() + return created, updated, snapshots, retired, offsite_imported + + +def main() -> None: + args = parse_args() + rows, offsite_rows = load_validated_rows() + if not rows or len(offsite_rows) != len(OFFSITE_R1_PRODUCT_CODES): + raise SystemExit("No verified hq.py products were returned; no database changes were made.") + for row in rows: + print( + f"ready code={row['product_code']} name={row['product_name']} " + f"nav={row['current_nav']} exchange={row['exchange_code']}" + ) + for row in offsite_rows: + print(f"ready offsite code={row['fund_code']} name={row['fund_name']} nav={row['current_nav']}") + created, updated, snapshots, retired, offsite_imported = import_rows( + rows, offsite_rows, dry_run=args.dry_run + ) + if args.dry_run: + print(f"dry run complete: validated={len(rows)}, database unchanged") + else: + print( + f"import complete: created={created}, updated={updated}, snapshots={snapshots}, " + f"retired={retired}, offsite_imported={offsite_imported}, total={len(rows)}" + ) + print("latest NAV caches were refreshed; no historical market rows were written") + + +if __name__ == "__main__": + main() diff --git a/tools/import_product_asset_classifications.py b/tools/import_product_asset_classifications.py new file mode 100644 index 0000000..aacfd16 --- /dev/null +++ b/tools/import_product_asset_classifications.py @@ -0,0 +1,155 @@ +# ruff: noqa: E402 + +"""Import contract-backed ETF asset classifications for dynamic allocation. + +This command only writes the additive ``advisor_product_asset_classification`` +reference table. It derives each class from a verified, currently effective +Southern Fund contract snapshot and deliberately skips contracts that cannot +be mapped to one of the three allocation buckets with clear evidence. +""" + +from __future__ import annotations + +import argparse +import re +import sys +from dataclasses import dataclass +from datetime import UTC, date, datetime +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +from tools.import_product_governance_reference import mysql_connection + +SOUTHERN_FUND_MANAGER = "南方基金" +LISTED_STATUS = "上市" +SOURCE = "nffund_verified_contract_classification_v1" +STATUS = "active" + + +@dataclass(frozen=True) +class ContractClassificationCandidate: + product_id: int + product_code: str + fund_type: str + investment_scope: str + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Import verified Southern Fund contract-backed asset classifications" + ) + parser.add_argument("--dry-run", action="store_true", help="Report classifications only") + return parser.parse_args() + + +def classify_contract(fund_type: str, investment_scope: str) -> str | None: + """Return a supported asset bucket only where official contract facts support it.""" + normalized_type = fund_type.strip() + normalized_scope = re.sub(r"\s+", "", investment_scope) + if "货币" in normalized_type: + return "cash_management_etf" + if "债券" in normalized_type: + return "bond_etf" + if "股票" in normalized_type or "指数" in normalized_type: + return "equity_etf" + if "混合" in normalized_type and "股票" in normalized_scope: + return "equity_etf" + return None + + +def candidates(as_of_date: date) -> list[ContractClassificationCandidate]: + connection = mysql_connection() + try: + with connection.cursor() as cursor: + cursor.execute( + """ + SELECT p.id, p.product_code, c.fund_type, c.investment_scope + FROM fin_product p + JOIN advisor_product_contract_snapshot c ON c.product_id = p.id + WHERE p.fund_manager=%s + AND p.status=%s + AND p.product_category IN ('ETF', 'LOF') + AND c.review_status='verified' + AND c.effective_from <= %s + AND (c.effective_until IS NULL OR c.effective_until >= %s) + ORDER BY p.product_code, c.effective_from DESC, c.id DESC + """, + (SOUTHERN_FUND_MANAGER, LISTED_STATUS, as_of_date, as_of_date), + ) + latest: dict[int, ContractClassificationCandidate] = {} + for product_id, product_code, fund_type, investment_scope in cursor.fetchall(): + latest.setdefault( + int(product_id), + ContractClassificationCandidate( + product_id=int(product_id), + product_code=str(product_code), + fund_type=str(fund_type), + investment_scope=str(investment_scope), + ), + ) + return list(latest.values()) + finally: + connection.close() + + +def import_classifications( + rows: list[ContractClassificationCandidate], *, as_of_date: date, dry_run: bool +) -> tuple[list[tuple[str, str]], list[str]]: + classified: list[tuple[ContractClassificationCandidate, str]] = [] + skipped: list[str] = [] + for row in rows: + asset_class = classify_contract(row.fund_type, row.investment_scope) + if asset_class is None: + skipped.append(f"{row.product_code} ({row.fund_type})") + else: + classified.append((row, asset_class)) + report = [(row.product_code, asset_class) for row, asset_class in classified] + if dry_run: + return report, skipped + + now = datetime.now(UTC).replace(tzinfo=None) + connection = mysql_connection() + try: + with connection.cursor() as cursor: + for row, asset_class in classified: + cursor.execute( + """ + INSERT INTO advisor_product_asset_classification ( + product_id, as_of_date, asset_class, source, status, created_at, updated_at + ) VALUES (%s, %s, %s, %s, %s, %s, %s) + ON DUPLICATE KEY UPDATE + asset_class=VALUES(asset_class), + source=VALUES(source), + status=VALUES(status), + updated_at=VALUES(updated_at) + """, + (row.product_id, as_of_date, asset_class, SOURCE, STATUS, now, now), + ) + connection.commit() + except Exception: + connection.rollback() + raise + finally: + connection.close() + return report, skipped + + +def main() -> None: + args = parse_args() + as_of_date = date.today() + report, skipped = import_classifications( + candidates(as_of_date), as_of_date=as_of_date, dry_run=args.dry_run + ) + for product_code, asset_class in report: + print(f"classified {product_code} -> {asset_class}") + for product in skipped: + print(f"skipped {product}: no defensible allocation bucket") + action = "validated" if args.dry_run else "imported" + print(f"{action}: classifications={len(report)} skipped={len(skipped)}") + + +if __name__ == "__main__": + main() + diff --git a/tools/import_product_governance_reference.py b/tools/import_product_governance_reference.py new file mode 100644 index 0000000..b9623b1 --- /dev/null +++ b/tools/import_product_governance_reference.py @@ -0,0 +1,312 @@ +"""Import reviewed, source-linked product suitability and contract references. + +This command intentionally does not derive R1-R5 from product type, historical +volatility, or the baseline catalogue. Suitability classifications belong to a +specific sales institution and must come from that institution's disclosure. +""" + +from __future__ import annotations + +import argparse +import csv +import re +import sys +from datetime import UTC, date, datetime +from decimal import Decimal, InvalidOperation +from pathlib import Path +from typing import Any +from urllib.parse import unquote, urlparse + +import pymysql # type: ignore[import-untyped] + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +LISTED_STATUS = "\u4e0a\u5e02" +SOUTHERN_FUND_MANAGER = "\u5357\u65b9\u57fa\u91d1" +REVIEW_STATUSES = {"pending_review", "verified", "superseded", "test_only"} +SHA256_PATTERN = re.compile(r"[0-9a-f]{64}") +SUITABILITY_FIELDS = ( + "product_code", "sales_institution", "risk_level", "effective_from", "effective_until", + "source_url", "document_title", "document_published_at", "document_sha256", "source", + "review_status", "verified_by", "verified_at", +) +CONTRACT_FIELDS = ( + "product_code", "effective_from", "effective_until", "fund_type", "investment_scope", + "performance_benchmark", "risk_return_characteristics", "custodian_name", + "management_fee_rate_pct", "custodian_fee_rate_pct", "inception_date", "source_url", + "document_title", "document_published_at", "document_sha256", "source", "review_status", + "verified_by", "verified_at", +) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Import source-linked advisory product governance data") + parser.add_argument("--suitability", type=Path, help="UTF-8 CSV with sales suitability disclosures") + parser.add_argument("--contracts", type=Path, help="UTF-8 CSV with fund-contract disclosures") + parser.add_argument("--dry-run", action="store_true", help="Validate rows without writing MySQL") + args = parser.parse_args() + if args.suitability is None and args.contracts is None: + parser.error("provide --suitability and/or --contracts") + return args + + +def mysql_connection() -> pymysql.Connection: + from app.core.config import get_settings + + parsed = urlparse(get_settings().mysql_dsn.replace("mysql+asyncmy://", "mysql+pymysql://")) + return pymysql.connect( + host=parsed.hostname or "127.0.0.1", + port=parsed.port or 3306, + user=unquote(parsed.username or ""), + password=unquote(parsed.password or ""), + database=(parsed.path or "/").lstrip("/"), + charset="utf8mb4", + autocommit=False, + ) + + +def read_rows(path: Path, required_fields: tuple[str, ...]) -> list[dict[str, str]]: + with path.open("r", encoding="utf-8-sig", newline="") as source: + reader = csv.DictReader(source) + actual_fields = set(reader.fieldnames or ()) + missing = set(required_fields) - actual_fields + if missing: + raise ValueError(f"{path}: missing CSV columns {sorted(missing)}") + rows = [] + for line_number, raw in enumerate(reader, start=2): + row = {key: (raw.get(key) or "").strip() for key in required_fields} + if not any(row.values()): + continue + row["_line_number"] = str(line_number) + rows.append(row) + return rows + + +def optional_date(value: str, field: str, line_number: str) -> date | None: + if not value: + return None + try: + return date.fromisoformat(value) + except ValueError as exc: + raise ValueError(f"line {line_number}: {field} must be YYYY-MM-DD") from exc + + +def optional_datetime(value: str, field: str, line_number: str) -> datetime | None: + if not value: + return None + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError as exc: + raise ValueError(f"line {line_number}: {field} must be ISO-8601") from exc + if parsed.tzinfo is None: + raise ValueError(f"line {line_number}: {field} must include timezone") + return parsed.astimezone(UTC).replace(tzinfo=None) + + +def optional_decimal(value: str, field: str, line_number: str) -> Decimal | None: + if not value: + return None + try: + parsed = Decimal(value) + except InvalidOperation as exc: + raise ValueError(f"line {line_number}: {field} must be a decimal") from exc + if parsed < 0: + raise ValueError(f"line {line_number}: {field} must not be negative") + return parsed + + +def validate_source_fields(row: dict[str, str]) -> None: + line_number = row["_line_number"] + if urlparse(row["source_url"]).scheme != "https": + raise ValueError(f"line {line_number}: source_url must be HTTPS") + if not SHA256_PATTERN.fullmatch(row["document_sha256"].lower()): + raise ValueError(f"line {line_number}: document_sha256 must be 64 hexadecimal characters") + if row["review_status"] not in REVIEW_STATUSES: + raise ValueError(f"line {line_number}: invalid review_status") + if row["review_status"] == "verified": + if not row["verified_by"] or not row["verified_at"]: + raise ValueError(f"line {line_number}: verified rows require verified_by and verified_at") + elif row["verified_by"] or row["verified_at"]: + raise ValueError(f"line {line_number}: only verified rows can have verification fields") + + +def normalize_suitability(rows: list[dict[str, str]]) -> list[dict[str, Any]]: + normalized = [] + for row in rows: + validate_source_fields(row) + line_number = row["_line_number"] + risk_level = row["risk_level"].upper() + if risk_level not in {"R1", "R2", "R3", "R4", "R5"}: + raise ValueError(f"line {line_number}: risk_level must be R1 through R5") + effective_from = optional_date(row["effective_from"], "effective_from", line_number) + effective_until = optional_date(row["effective_until"], "effective_until", line_number) + if effective_from is None: + raise ValueError(f"line {line_number}: effective_from is required") + if effective_until is not None and effective_until < effective_from: + raise ValueError(f"line {line_number}: effective_until precedes effective_from") + if not row["product_code"] or not row["sales_institution"]: + raise ValueError(f"line {line_number}: product_code and sales_institution are required") + normalized.append({ + **row, + "risk_level": risk_level, + "effective_from": effective_from, + "effective_until": effective_until, + "document_published_at": optional_date( + row["document_published_at"], "document_published_at", line_number + ), + "verified_at": optional_datetime(row["verified_at"], "verified_at", line_number), + "document_sha256": row["document_sha256"].lower(), + }) + return normalized + + +def normalize_contracts(rows: list[dict[str, str]]) -> list[dict[str, Any]]: + normalized = [] + for row in rows: + validate_source_fields(row) + line_number = row["_line_number"] + effective_from = optional_date(row["effective_from"], "effective_from", line_number) + effective_until = optional_date(row["effective_until"], "effective_until", line_number) + if effective_from is None: + raise ValueError(f"line {line_number}: effective_from is required") + if effective_until is not None and effective_until < effective_from: + raise ValueError(f"line {line_number}: effective_until precedes effective_from") + for field in ("product_code", "fund_type", "investment_scope", "risk_return_characteristics"): + if not row[field]: + raise ValueError(f"line {line_number}: {field} is required") + normalized.append({ + **row, + "effective_from": effective_from, + "effective_until": effective_until, + "document_published_at": optional_date( + row["document_published_at"], "document_published_at", line_number + ), + "inception_date": optional_date(row["inception_date"], "inception_date", line_number), + "management_fee_rate_pct": optional_decimal( + row["management_fee_rate_pct"], "management_fee_rate_pct", line_number + ), + "custodian_fee_rate_pct": optional_decimal( + row["custodian_fee_rate_pct"], "custodian_fee_rate_pct", line_number + ), + "verified_at": optional_datetime(row["verified_at"], "verified_at", line_number), + "document_sha256": row["document_sha256"].lower(), + }) + return normalized + + +def product_ids(cursor: Any, product_codes: set[str]) -> dict[str, int]: + if not product_codes: + return {} + placeholders = ", ".join(["%s"] * len(product_codes)) + cursor.execute( + f""" + SELECT id, product_code + FROM fin_product + WHERE product_code IN ({placeholders}) + AND fund_manager=%s AND status=%s AND product_category IN ('ETF', 'LOF') + """, + [*sorted(product_codes), SOUTHERN_FUND_MANAGER, LISTED_STATUS], + ) + result = {str(code): int(product_id) for product_id, code in cursor.fetchall()} + missing = product_codes - result.keys() + if missing: + raise ValueError(f"only listed Southern Fund ETF/LOF products can be imported; missing={sorted(missing)}") + return result + + +def insert_immutable( + cursor: Any, *, table: str, keys: tuple[str, ...], row: dict[str, Any], payload: dict[str, Any] +) -> bool: + where = " AND ".join(f"{key}=%s" for key in keys) + cursor.execute( + f"SELECT document_sha256 FROM {table} WHERE {where}", tuple(payload[key] for key in keys) + ) + existing = cursor.fetchone() + if existing is not None: + if str(existing[0]).lower() != str(payload["document_sha256"]).lower(): + raise ValueError( + f"line {row['_line_number']}: existing {table} version has a different document hash; " + "create a new effective_from version instead" + ) + return False + columns = ", ".join(payload) + placeholders = ", ".join(f"%({key})s" for key in payload) + cursor.execute(f"INSERT INTO {table} ({columns}) VALUES ({placeholders})", payload) + return True + + +def import_references( + suitability_rows: list[dict[str, Any]], contract_rows: list[dict[str, Any]], *, dry_run: bool +) -> tuple[int, int]: + if dry_run: + return len(suitability_rows), len(contract_rows) + now = datetime.now(UTC).replace(tzinfo=None) + connection = mysql_connection() + try: + with connection.cursor() as cursor: + references = [*suitability_rows, *contract_rows] + ids = product_ids(cursor, {str(row["product_code"]) for row in references}) + suitability_count = 0 + contract_count = 0 + for row in suitability_rows: + payload = { + "product_id": ids[row["product_code"]], "sales_institution": row["sales_institution"], + "risk_level": row["risk_level"], "effective_from": row["effective_from"], + "effective_until": row["effective_until"], "source_url": row["source_url"], + "document_title": row["document_title"], "document_published_at": row["document_published_at"], + "document_sha256": row["document_sha256"], "source": row["source"], + "review_status": row["review_status"], "verified_by": row["verified_by"] or None, + "verified_at": row["verified_at"], "created_at": now, "updated_at": now, + } + suitability_count += insert_immutable( + cursor, table="advisor_product_suitability_reference", + keys=("product_id", "sales_institution", "effective_from"), row=row, payload=payload, + ) + for row in contract_rows: + payload = { + "product_id": ids[row["product_code"]], "effective_from": row["effective_from"], + "effective_until": row["effective_until"], "fund_type": row["fund_type"], + "investment_scope": row["investment_scope"], + "performance_benchmark": row["performance_benchmark"] or None, + "risk_return_characteristics": row["risk_return_characteristics"], + "custodian_name": row["custodian_name"] or None, + "management_fee_rate_pct": row["management_fee_rate_pct"], + "custodian_fee_rate_pct": row["custodian_fee_rate_pct"], + "inception_date": row["inception_date"], "source_url": row["source_url"], + "document_title": row["document_title"], "document_published_at": row["document_published_at"], + "document_sha256": row["document_sha256"], "source": row["source"], + "review_status": row["review_status"], "verified_by": row["verified_by"] or None, + "verified_at": row["verified_at"], "created_at": now, "updated_at": now, + } + contract_count += insert_immutable( + cursor, table="advisor_product_contract_snapshot", + keys=("product_id", "effective_from"), row=row, payload=payload, + ) + connection.commit() + return suitability_count, contract_count + except Exception: + connection.rollback() + raise + finally: + connection.close() + + +def main() -> None: + args = parse_args() + suitability_rows = normalize_suitability( + read_rows(args.suitability, SUITABILITY_FIELDS) if args.suitability else [] + ) + contract_rows = normalize_contracts( + read_rows(args.contracts, CONTRACT_FIELDS) if args.contracts else [] + ) + suitability_count, contract_count = import_references( + suitability_rows, contract_rows, dry_run=args.dry_run + ) + action = "validated" if args.dry_run else "inserted" + print(f"{action}: suitability={suitability_count} contracts={contract_count}") + + +if __name__ == "__main__": + main() + diff --git a/tools/sync_nanfang_official_product_governance.py b/tools/sync_nanfang_official_product_governance.py new file mode 100644 index 0000000..40bdb45 --- /dev/null +++ b/tools/sync_nanfang_official_product_governance.py @@ -0,0 +1,270 @@ +"""Synchronize official Southern Fund risk ratings and fund-contract evidence. + +The source is Southern Fund's public direct-sales product detail page. Its +``fundRiskRating`` field states that the rating is from Southern Fund itself. +Contract facts are taken from the same page and linked to the original fund +contract file returned by Southern Fund's legal-document directory. +""" + +# ruff: noqa: E402 + +from __future__ import annotations + +import argparse +import hashlib +import json +import re +import sys +from datetime import UTC, date, datetime +from decimal import Decimal +from pathlib import Path +from typing import Any +from urllib.parse import urljoin + +import httpx + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +from tools.import_product_governance_reference import ( + import_references, + normalize_contracts, + normalize_suitability, + mysql_connection, + product_ids, +) + +OFFICIAL_BASE_URL = "https://www.nffund.com" +DETAIL_ENDPOINT = f"{OFFICIAL_BASE_URL}/nfwebApi/fund/overreview" +LEGAL_DOCUMENTS_ENDPOINT = f"{OFFICIAL_BASE_URL}/nfwebApi/notice/legalDocuments" +SALES_INSTITUTION = "\u5357\u65b9\u57fa\u91d1\u7ba1\u7406\u80a1\u4efd\u6709\u9650\u516c\u53f8\u76f4\u9500" +SOURCE = "nffund_official_direct_sales" +VERIFIED_BY = "system:official-source-sync" +RISK_PATTERN = re.compile(r"R([1-5])") +CONTRACT_EXCLUSIONS = ("\u6258\u7ba1\u534f\u8bae", "\u751f\u6548\u516c\u544a", "\u62db\u52df\u8bf4\u660e\u4e66") +HEADERS = {"User-Agent": "NailongFund/1.0", "Referer": f"{OFFICIAL_BASE_URL}/"} + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Sync official Southern Fund direct-sales suitability and contracts" + ) + parser.add_argument("--dry-run", action="store_true", help="Fetch and validate without MySQL writes") + parser.add_argument("--timeout", type=float, default=30.0, help="HTTP request timeout in seconds") + return parser.parse_args() + + +def listed_product_codes() -> list[str]: + connection = mysql_connection() + try: + with connection.cursor() as cursor: + cursor.execute( + """ + SELECT product_code + FROM fin_product + WHERE fund_manager=%s AND status=%s AND product_category IN ('ETF', 'LOF') + ORDER BY product_code + """, + ("\u5357\u65b9\u57fa\u91d1", "\u4e0a\u5e02"), + ) + return [str(row[0]) for row in cursor.fetchall()] + finally: + connection.close() + + +def date_value(value: object, *, fallback: date) -> str: + text = str(value or "").strip() + for pattern in ("%Y%m%d", "%Y-%m-%d"): + try: + return datetime.strptime(text[:10], pattern).date().isoformat() + except ValueError: + pass + return fallback.isoformat() + + +def source_hash(value: object) -> str: + payload = json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + return hashlib.sha256(payload.encode("utf-8")).hexdigest() + + +def risk_level(value: object, product_code: str) -> str: + match = RISK_PATTERN.search(str(value or "")) + if match is None: + raise ValueError(f"{product_code}: Southern Fund did not disclose an R1-R5 risk rating") + return f"R{match.group(1)}" + + +def contract_document(documents: list[dict[str, Any]], product_code: str) -> dict[str, Any]: + candidates = [ + item + for item in documents + if "\u57fa\u91d1\u5408\u540c" in str(item.get("title") or "") + and not any(token in str(item.get("title") or "") for token in CONTRACT_EXCLUSIONS) + and item.get("linkUrl") + ] + if not candidates: + raise ValueError(f"{product_code}: no official fund contract document was found") + return max( + candidates, + key=lambda item: (str(item.get("publishTime") or ""), str(item.get("id") or "")), + ) + + +def collect_official_rows( + product_code: str, + overview: dict[str, Any], + documents: list[dict[str, Any]], + contract_bytes: bytes, + *, + synced_at: datetime, +) -> tuple[dict[str, str], dict[str, str]]: + data = overview.get("data") + if overview.get("code") != "ETS-5BP00000" or not isinstance(data, dict): + raise ValueError(f"{product_code}: official product detail request failed") + fund_info = data.get("fund_info") + risk = data.get("fundRiskRating") + if not isinstance(fund_info, dict) or not isinstance(risk, dict): + raise ValueError(f"{product_code}: official detail response is incomplete") + rating_text = str(risk.get("RISKRATING") or "").strip() + level = risk_level(rating_text, product_code) + synced_date = synced_at.date() + detail_url = ( + f"{OFFICIAL_BASE_URL}/new/personal-financing/detail.html?fundCode={product_code}" + ) + detail_evidence = { + "fund_code": product_code, + "fund_name": fund_info.get("fundName"), + "risk_rating": rating_text, + "source": "Southern Fund direct-sales product detail", + } + document = contract_document(documents, product_code) + contract_url = urljoin(OFFICIAL_BASE_URL, str(document["linkUrl"])) + document_title = str(document["title"]).strip() + verification_time = synced_at.astimezone(UTC).isoformat() + contract_effective_from = date_value(fund_info.get("contractValidDate"), fallback=synced_date) + contract_published_at = date_value( + document.get("createTimeString") or document.get("publishTime"), fallback=synced_date + ) + suitability_row = { + "product_code": product_code, + "sales_institution": SALES_INSTITUTION, + "risk_level": level, + "effective_from": synced_date.isoformat(), + "effective_until": "", + "source_url": detail_url, + "document_title": f"\u5357\u65b9\u57fa\u91d1\u76f4\u9500\u4ea7\u54c1\u8be6\u60c5\u9875\uff1a{fund_info.get('fundName', product_code)}", + "document_published_at": "", + "document_sha256": source_hash(detail_evidence), + "source": SOURCE, + "review_status": "verified", + "verified_by": VERIFIED_BY, + "verified_at": verification_time, + "_line_number": product_code, + } + contract_row = { + "product_code": product_code, + "effective_from": contract_effective_from, + "effective_until": "", + "fund_type": str(fund_info.get("basedetailType") or "").strip(), + "investment_scope": str(fund_info.get("tzfw") or "").strip(), + "performance_benchmark": str(fund_info.get("yjbjjz") or "").strip(), + "risk_return_characteristics": rating_text, + "custodian_name": str(fund_info.get("jjtgr") or "").strip(), + "management_fee_rate_pct": decimal_text(fund_info.get("glYearRatio")), + "custodian_fee_rate_pct": decimal_text(fund_info.get("tgYearRatio")), + "inception_date": date_value(fund_info.get("fundDate"), fallback=synced_date), + "source_url": contract_url, + "document_title": document_title, + "document_published_at": contract_published_at, + "document_sha256": hashlib.sha256(contract_bytes).hexdigest(), + "source": SOURCE, + "review_status": "verified", + "verified_by": VERIFIED_BY, + "verified_at": verification_time, + "_line_number": product_code, + } + return suitability_row, contract_row + + +def decimal_text(value: object) -> str: + if value in (None, ""): + return "" + return str(Decimal(str(value))) + + +def official_documents(client: httpx.Client, product_code: str) -> list[dict[str, Any]]: + response = client.post( + LEGAL_DOCUMENTS_ENDPOINT, + data={ + "fundCode": product_code, + "typeNo": 1, + "tabsid": "newgg", + "curPage": 1, + "pageSize": 100, + }, + ) + response.raise_for_status() + payload = response.json() + data = payload.get("data") if isinstance(payload, dict) else None + rows = data.get("list") if isinstance(data, dict) else None + if payload.get("code") != "ETS-5BP00000" or not isinstance(rows, list): + raise ValueError(f"{product_code}: official legal-document directory request failed") + return [item for item in rows if isinstance(item, dict)] + + +def sync(*, dry_run: bool, timeout: float) -> tuple[int, int]: + suitability_rows, contract_rows = collect_rows(timeout=timeout) + normalized_suitability = normalize_suitability(suitability_rows) + normalized_contracts = normalize_contracts(contract_rows) + if not dry_run: + product_codes = [str(row["product_code"]) for row in suitability_rows] + connection = mysql_connection() + try: + with connection.cursor() as cursor: + eligible = product_ids(cursor, set(product_codes)) + if len(eligible) != len(product_codes): + raise ValueError("database product scope changed during synchronization") + finally: + connection.close() + return import_references(normalized_suitability, normalized_contracts, dry_run=dry_run) + + +def collect_rows(*, timeout: float) -> tuple[list[dict[str, str]], list[dict[str, str]]]: + """Fetch official evidence without writing it, for monitoring and review workflows.""" + product_codes = listed_product_codes() + if not product_codes: + raise ValueError("no listed Southern Fund ETF/LOF products exist in fin_product") + synced_at = datetime.now(UTC) + suitability_rows: list[dict[str, str]] = [] + contract_rows: list[dict[str, str]] = [] + with httpx.Client(headers=HEADERS, timeout=timeout, follow_redirects=True) as client: + for product_code in product_codes: + overview_response = client.post(DETAIL_ENDPOINT, data={"fundCode": product_code}) + overview_response.raise_for_status() + overview = overview_response.json() + document = contract_document(official_documents(client, product_code), product_code) + contract_url = urljoin(OFFICIAL_BASE_URL, str(document["linkUrl"])) + contract_response = client.get(contract_url) + contract_response.raise_for_status() + suitability, contract = collect_official_rows( + product_code, + overview, + [document], + contract_response.content, + synced_at=synced_at, + ) + suitability_rows.append(suitability) + contract_rows.append(contract) + return suitability_rows, contract_rows + + +def main() -> None: + args = parse_args() + suitability_count, contract_count = sync(dry_run=args.dry_run, timeout=args.timeout) + action = "validated" if args.dry_run else "synchronized" + print(f"{action}: suitability={suitability_count} contracts={contract_count}") + + +if __name__ == "__main__": + main() +