195 lines
7.9 KiB
Python
195 lines
7.9 KiB
Python
"""场外基金申购赎回确定性规则。"""
|
||
|
||
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={"原因": "必要识别字段或查询结果缺失"},
|
||
)
|