🎉 init
This commit is contained in:
@@ -0,0 +1,187 @@
|
||||
from datetime import UTC, datetime
|
||||
from decimal import Decimal
|
||||
from enum import StrEnum
|
||||
from typing import Annotated, 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 = "MISC_IN"
|
||||
|
||||
|
||||
class MiscOutTransaction(BaseTransaction):
|
||||
"""其他支出交易"""
|
||||
|
||||
txn_scene = "MISC_OUT"
|
||||
|
||||
|
||||
class PaymentTransaction(MiscOutTransaction):
|
||||
"""支付交易"""
|
||||
|
||||
txn_scene = "PAYMENT"
|
||||
merchant_name: Annotated[str | None, "商户名称", Field(default=None)]
|
||||
|
||||
|
||||
class EcomPaymentTransaction(PaymentTransaction):
|
||||
"""电子商务支付交易"""
|
||||
|
||||
txn_scene = "ECOM_PAYMENT"
|
||||
order_id: Annotated[str | None, "订单ID", Field(default=None)]
|
||||
|
||||
|
||||
class PosPaymentTransaction(PaymentTransaction):
|
||||
"""POS支付交易"""
|
||||
|
||||
txn_scene = "POS_PAYMENT"
|
||||
geo: Annotated[
|
||||
tuple[CountryAlpha2] | tuple[CountryAlpha2, str] | None, "地理信息", Field(default=None)
|
||||
]
|
||||
|
||||
|
||||
class ATMTransaction(BaseTransaction):
|
||||
"""ATM交易"""
|
||||
|
||||
txn_scene = "ATM"
|
||||
geo: Annotated[
|
||||
tuple[CountryAlpha2] | tuple[CountryAlpha2, str] | None, "地理信息", Field(default=None)
|
||||
]
|
||||
|
||||
|
||||
class TransferTransaction(BaseTransaction):
|
||||
"""转账交易"""
|
||||
|
||||
txn_scene = "TRANSFER"
|
||||
|
||||
|
||||
type Transaction = Annotated[
|
||||
MiscInTransaction
|
||||
| MiscOutTransaction
|
||||
| TransferTransaction
|
||||
| PaymentTransaction
|
||||
| EcomPaymentTransaction
|
||||
| PosPaymentTransaction
|
||||
| ATMTransaction,
|
||||
Field(discriminator="txn_scene"),
|
||||
]
|
||||
Reference in New Issue
Block a user