"""场外基金申购赎回确定性规则。""" import re from dataclasses import dataclass from datetime import date, datetime from decimal import Decimal, InvalidOperation from app.core.offsite_fund_contracts import RuleResultStatus HUNDRED = Decimal("100") TEN_PERCENT = Decimal("0.10") TWENTY_PERCENT = Decimal("0.20") ONE_YUAN = Decimal("1") def display_decimal(value: Decimal) -> str: """把计算结果格式化成适合页面展示的十进制文本。""" text = format(value, "f") if "." in text: text = text.rstrip("0").rstrip(".") return text or "0" @dataclass(frozen=True) class RuleDecision: rule_code: str rule_name: str result: RuleResultStatus document_value: dict[str, object] database_value: dict[str, object] calculation: dict[str, object] def decimal_from(value: object) -> Decimal | None: if value is None or value == "": return None try: return Decimal(str(value).replace(",", "").strip()) except (InvalidOperation, ValueError): return None def normalize_amount_yuan(raw_value: object, unit: object) -> Decimal | None: raw_unit = str(unit or "元").strip().replace("人民币", "元") raw_text = str(raw_value or "").replace(",", "").replace("，", "").strip() if raw_unit and raw_text.endswith(raw_unit): raw_text = raw_text[: -len(raw_unit)].strip() # OCR 可能把币种前缀一起识别到金额字段，例如“人民币5,000.00元”。 # 只接受开头的常见币种标记，避免从任意业务文本中误抽数字。 raw_text = re.sub(r"^(?:人民币|RMB|CNY|¥|￥)\s*", "", raw_text, flags=re.IGNORECASE) amount = decimal_from(raw_text) if amount is None: return None if raw_unit == "万元": return amount * Decimal("10000") if raw_unit in {"元", "人民币", "CNY"}: return amount return None def parse_application_date(raw_value: object) -> date | None: text = str(raw_value or "").strip() for fmt in ("%Y-%m-%d", "%Y/%m/%d", "%Y年%m月%d日", "%Y%m%d"): try: return datetime.strptime(text, fmt).date() except ValueError: continue return None class OffsiteFundRuleEngine: """只做程序化计算，不读取数据库，也不调用模型。""" def check_subscription( self, *, amount_yuan: Decimal | None, nav: Decimal | None, total_fund_shares: Decimal | None, before_holding_shares: Decimal | None, ) -> list[RuleDecision]: decisions = [self._minimum_subscription(amount_yuan)] if (amount_yuan is None or nav is None or nav <= 0 or total_fund_shares is None or total_fund_shares == 0): return decisions + [ self._unknown("subscription_holding_ratio", "申购后单一投资者持有比例"), self._unknown("subscription_single_share_limit", "申购单笔份额上限"), ] current_shares = amount_yuan / nav before = before_holding_shares or Decimal("0") ratio = (before + current_shares) / total_fund_shares decisions.append(RuleDecision( rule_code="subscription_holding_ratio", rule_name="申购后单一投资者持有比例", result="异常" if ratio > TWENTY_PERCENT else "正常", document_value={"申购金额元": str(amount_yuan)}, database_value={"最新净值": str(nav), "基金最新总份额": str(total_fund_shares), "申请前持有份额": str(before)}, calculation={ "本次申购份额": str(current_shares), "申购后持有比例": str(ratio), "实际值": f"{display_decimal(ratio * HUNDRED)}%", "规则值": "≤ 20%", "比较": f"{display_decimal(ratio * HUNDRED)}% ≤ 20%", }, )) limit = total_fund_shares * TEN_PERCENT decisions.append(RuleDecision( rule_code="subscription_single_share_limit", rule_name="申购单笔份额上限", result="异常" if current_shares > limit else "正常", document_value={"申购金额元": str(amount_yuan)}, database_value={"最新净值": str(nav), "基金最新总份额": str(total_fund_shares)}, calculation={ "本次申购份额": str(current_shares), "份额上限": str(limit), "实际值": str(current_shares), "规则值": str(limit), "比较": f"{current_shares} ≤ {limit}", }, )) return decisions def check_redemption( self, *, redemption_shares: Decimal | None, total_fund_shares: Decimal | None, available_quantity: Decimal | None, ) -> list[RuleDecision]: if redemption_shares is None or total_fund_shares is None or total_fund_shares == 0: ratio = self._unknown("redemption_large_ratio", "赎回巨额比例") else: value = redemption_shares / total_fund_shares ratio = RuleDecision( rule_code="redemption_large_ratio", rule_name="赎回巨额比例", result="异常" if value > TWENTY_PERCENT else "正常", document_value={"赎回份额": str(redemption_shares)}, database_value={"产品最新总份额": str(total_fund_shares)}, calculation={ "赎回比例": str(value), "实际值": f"{display_decimal(value * HUNDRED)}%", "规则值": "≤ 20%", "比较": f"{display_decimal(value * HUNDRED)}% ≤ 20%", }, ) if redemption_shares is None or available_quantity is None: available = self._unknown("redemption_available_quantity", "账户可用份额") else: available = RuleDecision( rule_code="redemption_available_quantity", rule_name="账户可用份额", result="异常" if redemption_shares > available_quantity else "正常", document_value={"赎回份额": str(redemption_shares)}, database_value={"当前最新可用份额": str(available_quantity)}, calculation={ "是否超出可用份额": redemption_shares > available_quantity, "实际值": str(redemption_shares), "规则值": str(available_quantity), "比较": f"{redemption_shares} ≤ {available_quantity}", }, ) return [ratio, available] @staticmethod def _minimum_subscription(amount_yuan: Decimal | None) -> RuleDecision: result: RuleResultStatus = "无法判断" if amount_yuan is not None: result = "异常" if amount_yuan <= ONE_YUAN else "正常" return RuleDecision( rule_code="subscription_minimum_amount", rule_name="申购最低金额", result=result, document_value={"申购金额元": str(amount_yuan) if amount_yuan is not None else None}, database_value={}, calculation={ "判断口径": "标准化申购金额 <= 1 元为异常", "实际值": str(amount_yuan) if amount_yuan is not None else None, "规则值": "> 1 元", "比较": f"{amount_yuan} > 1 元" if amount_yuan is not None else None, }, ) @staticmethod def _unknown(rule_code: str, rule_name: str) -> RuleDecision: return RuleDecision( rule_code=rule_code, rule_name=rule_name, result="无法判断", document_value={}, database_value={}, calculation={"原因": "必要识别字段或查询结果缺失"}, )