190 lines
7.2 KiB
Python
190 lines
7.2 KiB
Python
from datetime import UTC, datetime
|
|
from decimal import Decimal
|
|
from enum import StrEnum
|
|
from typing import Annotated, Literal, Self
|
|
|
|
from pydantic import BaseModel, Field, model_validator
|
|
from pydantic_extra_types.country import CountryAlpha2
|
|
from pydantic_extra_types.currency_code import Currency
|
|
from ulid import ULID
|
|
|
|
from .ref import ChannelRef, TransactionRef
|
|
|
|
|
|
class TransactionStatus(StrEnum):
|
|
"""交易状态"""
|
|
|
|
PENDING = "PENDING"
|
|
COMPLETED = "COMPLETED"
|
|
FAILED = "FAILED"
|
|
REFUNDED = "REFUNDED"
|
|
|
|
|
|
class DcFlag(StrEnum):
|
|
"""借贷方向"""
|
|
|
|
DEBIT = "DEBIT"
|
|
CREDIT = "CREDIT"
|
|
|
|
|
|
class FxRate(BaseModel):
|
|
"""汇率快照"""
|
|
|
|
from_ccy: Annotated[Currency, "源币种"]
|
|
to_ccy: Annotated[Currency, "目标币种"]
|
|
rate: Annotated[Decimal, "汇率"]
|
|
|
|
|
|
class BaseTransaction(BaseModel):
|
|
"""基础交易模型"""
|
|
|
|
id: ULID = Field(default_factory=ULID)
|
|
version: Annotated[int, "乐观锁版本号", Field(default=0)]
|
|
ref_transactions: Annotated[
|
|
list[TransactionRef], "关联的交易引用列表", Field(default_factory=list)
|
|
]
|
|
txn_date: Annotated[datetime, "交易日期"]
|
|
clearing_date: Annotated[datetime | None, "清算日期", Field(default=None)]
|
|
posting_date: Annotated[datetime | None, "入账日期", Field(default=None)]
|
|
created_at: Annotated[datetime, "创建时间", Field(default_factory=lambda: datetime.now(UTC))]
|
|
updated_at: Annotated[datetime, "更新时间", Field(default_factory=lambda: datetime.now(UTC))]
|
|
deleted_at: Annotated[datetime | None, "软删除时间（None=未删除）", Field(default=None)]
|
|
txn_amt: Annotated[Decimal, "交易金额"]
|
|
txn_ccy: Annotated[Currency, "交易币种"]
|
|
posting_amt: Annotated[Decimal | None, "入账金额", Field(default=None)]
|
|
posting_ccy: Annotated[Currency | None, "入账币种", Field(default=None)]
|
|
comm_amt: Annotated[Decimal | None, "手续费金额", Field(default=None)]
|
|
comm_ccy: Annotated[Currency | None, "手续费币种", Field(default=None)]
|
|
surcharge_amt: Annotated[Decimal | None, "服务费金额", Field(default=None)]
|
|
surcharge_ccy: Annotated[Currency | None, "服务费币种", Field(default=None)]
|
|
disc_amt: Annotated[Decimal | None, "折扣金额", Field(default=None)]
|
|
disc_ccy: Annotated[Currency | None, "折扣币种", Field(default=None)]
|
|
fx_rates: Annotated[list[FxRate] | None, "汇率快照列表", Field(default=None)]
|
|
dc_flag: Annotated[DcFlag, "借贷方向"]
|
|
ref_channel: Annotated[ChannelRef, "关联的渠道引用列表"]
|
|
cp: Annotated[str | ChannelRef, "对手方（字符串描述或渠道引用）"]
|
|
acq_inst: Annotated[str | None, "收单机构", Field(default=None)]
|
|
clearing_network: Annotated[str | None, "清算网络", Field(default=None)]
|
|
txn_sts: Annotated[TransactionStatus, "交易状态", Field(default=TransactionStatus.COMPLETED)]
|
|
description: Annotated[str | None, "交易描述", Field(default=None)]
|
|
memo: Annotated[str | None, "备注", Field(default=None)]
|
|
raw_description: Annotated[str | None, "原始交易描述", Field(default=None)]
|
|
raw_data: Annotated[dict | None, "原始数据", Field(default=None)]
|
|
|
|
txn_scene: Annotated[str | None, "交易场景", Field(default=None)]
|
|
|
|
@model_validator(mode="after")
|
|
def _validate_amt_ccy_pairs(self) -> Self:
|
|
pairs = [
|
|
("posting_amt", "posting_ccy"),
|
|
("comm_amt", "comm_ccy"),
|
|
("surcharge_amt", "surcharge_ccy"),
|
|
("disc_amt", "disc_ccy"),
|
|
]
|
|
for amt_field, ccy_field in pairs:
|
|
amt = getattr(self, amt_field)
|
|
ccy = getattr(self, ccy_field)
|
|
if (amt is None) != (ccy is None):
|
|
msg = f"{amt_field} and {ccy_field} must both be None or both have values"
|
|
raise ValueError(msg)
|
|
return self
|
|
|
|
@model_validator(mode="after")
|
|
def _validate_accounting_equation(self) -> Self:
|
|
"""
|
|
会计恒等式校验（同币种）
|
|
|
|
- CREDIT: posting_amt = txn_amt + surcharge_amt - comm_amt - disc_amt
|
|
- DEBIT: posting_amt = txn_amt + surcharge_amt + comm_amt - disc_amt
|
|
"""
|
|
if self.posting_amt is None or self.posting_ccy != self.txn_ccy:
|
|
return self
|
|
|
|
expected = self.txn_amt
|
|
if self.surcharge_amt is not None and self.surcharge_ccy == self.txn_ccy:
|
|
expected += self.surcharge_amt
|
|
if self.disc_amt is not None and self.disc_ccy == self.txn_ccy:
|
|
expected -= self.disc_amt
|
|
|
|
if self.dc_flag == DcFlag.CREDIT:
|
|
if self.comm_amt is not None and self.comm_ccy == self.txn_ccy:
|
|
expected -= self.comm_amt
|
|
elif self.dc_flag == DcFlag.DEBIT:
|
|
if self.comm_amt is not None and self.comm_ccy == self.txn_ccy:
|
|
expected += self.comm_amt
|
|
|
|
if self.posting_amt != expected:
|
|
msg = (
|
|
f"Accounting equation mismatch ({self.dc_flag.value}): "
|
|
f"posting_amt({self.posting_amt}) != "
|
|
f"txn_amt({self.txn_amt}) + surcharge_amt({self.surcharge_amt})"
|
|
f" {'+' if self.dc_flag == DcFlag.DEBIT else '-'} comm_amt({self.comm_amt})"
|
|
f" - disc_amt({self.disc_amt})"
|
|
)
|
|
raise ValueError(msg)
|
|
return self
|
|
|
|
|
|
class MiscInTransaction(BaseTransaction):
|
|
"""其他收入交易"""
|
|
|
|
txn_scene: Annotated[Literal["MISC_IN"], "交易场景"] = "MISC_IN"
|
|
dc_flag: Annotated[DcFlag, "借贷方向"] = DcFlag.CREDIT
|
|
|
|
|
|
class MiscOutTransaction(BaseTransaction):
|
|
"""其他支出交易"""
|
|
|
|
txn_scene: Annotated[Literal["MISC_OUT"], "交易场景"] = "MISC_OUT"
|
|
dc_flag: Annotated[DcFlag, "借贷方向"] = DcFlag.DEBIT
|
|
|
|
|
|
class PaymentTransaction(MiscOutTransaction):
|
|
"""支付交易"""
|
|
|
|
txn_scene: Annotated[Literal["PAYMENT"], "交易场景"] = "PAYMENT"
|
|
merchant_name: Annotated[str | None, "商户名称", Field(default=None)]
|
|
|
|
|
|
class EcomPaymentTransaction(PaymentTransaction):
|
|
"""电子商务支付交易"""
|
|
|
|
txn_scene: Annotated[Literal["ECOM_PAYMENT"], "交易场景"] = "ECOM_PAYMENT"
|
|
order_id: Annotated[str | None, "订单ID", Field(default=None)]
|
|
|
|
|
|
class PosPaymentTransaction(PaymentTransaction):
|
|
"""POS支付交易"""
|
|
|
|
txn_scene: Annotated[Literal["POS_PAYMENT"], "交易场景"] = "POS_PAYMENT"
|
|
geo: Annotated[
|
|
tuple[CountryAlpha2] | tuple[CountryAlpha2, str] | None, "地理信息", Field(default=None)
|
|
]
|
|
|
|
|
|
class ATMTransaction(BaseTransaction):
|
|
"""ATM交易"""
|
|
|
|
txn_scene: Annotated[Literal["ATM"], "交易场景"] = "ATM"
|
|
geo: Annotated[
|
|
tuple[CountryAlpha2] | tuple[CountryAlpha2, str] | None, "地理信息", Field(default=None)
|
|
]
|
|
|
|
|
|
class TransferTransaction(BaseTransaction):
|
|
"""转账交易"""
|
|
|
|
txn_scene: Annotated[Literal["TRANSFER"], "交易场景"] = "TRANSFER"
|
|
|
|
|
|
type Transaction = Annotated[
|
|
MiscInTransaction
|
|
| MiscOutTransaction
|
|
| TransferTransaction
|
|
| PaymentTransaction
|
|
| EcomPaymentTransaction
|
|
| PosPaymentTransaction
|
|
| ATMTransaction,
|
|
Field(discriminator="txn_scene"),
|
|
]
|