diff options
| -rw-r--r-- | generalresearch/managers/thl/ledger_manager/thl_ledger.py | 2 | ||||
| -rw-r--r-- | generalresearch/managers/thl/payout.py | 325 | ||||
| -rw-r--r-- | generalresearch/models/thl/payout.py | 51 | ||||
| -rw-r--r-- | generalresearch/thl_django/event/models.py | 12 |
4 files changed, 181 insertions, 209 deletions
diff --git a/generalresearch/managers/thl/ledger_manager/thl_ledger.py b/generalresearch/managers/thl/ledger_manager/thl_ledger.py index 2fd0a2a..3119a46 100644 --- a/generalresearch/managers/thl/ledger_manager/thl_ledger.py +++ b/generalresearch/managers/thl/ledger_manager/thl_ledger.py @@ -837,7 +837,7 @@ class ThlLedgerManager(LedgerManager): tmc.TX_TYPE: TransactionType.BP_PAYOUT, tmc.EVENT: payoutevent_uuid, } - # This tag might will uniquely identify this tx + # This tag will uniquely identify this tx tag = f"{self.currency.value}:bp_payout:{payoutevent_uuid}" cash_account = self.get_account_cash() bp_wallet_account = self.get_account_or_create_bp_wallet(product) diff --git a/generalresearch/managers/thl/payout.py b/generalresearch/managers/thl/payout.py index 22856f6..4b08be8 100644 --- a/generalresearch/managers/thl/payout.py +++ b/generalresearch/managers/thl/payout.py @@ -11,7 +11,7 @@ from uuid import UUID, uuid4 import numpy as np import pandas as pd -from psycopg import sql +from psycopg import sql, Cursor from pydantic import AwareDatetime, NonNegativeInt, PositiveInt from generalresearch.currency import USDCent @@ -104,7 +104,7 @@ class PayoutEventManager(PostgresManagerWithRedis): order_data = order_data if order_data is not None else payout_event.order_data payout_event.update(status=status, ext_ref_id=ext_ref_id, order_data=order_data) - d = payout_event.model_dump_mysql() + d = payout_event.model_dump_postgres() query = sql.SQL( """ UPDATE event_payout SET @@ -476,64 +476,6 @@ class BrokerageProductPayoutEventManager(PayoutEventManager): return True - def create( - self, - uuid: UUIDStr | None = None, - debit_account_uuid: UUIDStr | None = None, - created: AwareDatetimeISO = None, - amount: PositiveInt = None, - status: PayoutStatus | None = None, - ext_ref_id: str | None = None, - payout_type: PayoutType = None, - request_data: dict[str, Any] | None = None, - order_data: dict[str, Any] | CashMailOrderData | None = None, - # --- Support resources --- - account_product_mapping: dict[UUIDStr, UUIDStr] | None = None, - ) -> BrokerageProductPayoutEvent: - - if request_data is None: - request_data = dict() - - # This isn't really need for creation... but we're doing it so that - # it can return back a full BrokerageProductPayoutEvent instance - if account_product_mapping is None: - rc = self.redis_client - account_product_mapping: dict = rc.hgetall(name="pem:account_to_product") - assert isinstance(account_product_mapping, dict) - product_id = account_product_mapping[debit_account_uuid] - - bp_payout_event = BrokerageProductPayoutEvent( - uuid=uuid or uuid4().hex, - debit_account_uuid=debit_account_uuid, - cashout_method_uuid=self.CASHOUT_METHOD_UUID, - created=created or datetime.now(tz=timezone.utc), - amount=amount, - status=status, - ext_ref_id=ext_ref_id, - payout_type=payout_type, - request_data=request_data, - order_data=order_data, - product_id=product_id, - ) - d = bp_payout_event.model_dump_mysql() - - self.pg_config.execute_write( - query=""" - INSERT INTO event_payout ( - uuid, debit_account_uuid, created, cashout_method_uuid, amount, - status, ext_ref_id, payout_type, order_data, request_data - ) VALUES ( - %(uuid)s, %(debit_account_uuid)s, %(created)s, - %(cashout_method_uuid)s, %(amount)s, %(status)s, - %(ext_ref_id)s, %(payout_type)s, %(order_data)s, - %(request_data)s - ); - """, - params=d, - ) - - return bp_payout_event - def filter_by( self, reference_uuid: str | None = None, @@ -729,91 +671,33 @@ class BrokerageProductPayoutEventManager(PayoutEventManager): ): pass - def create_bp_payout_event( - self, - thl_ledger_manager: ThlLedgerManager, - product: Product, - amount: USDCent, - payout_type: PayoutType = PayoutType.ACH, - ext_ref_id: str | None = None, - created: AwareDatetime | None = None, - skip_wallet_balance_check: bool = False, - skip_one_per_day_check: bool = False, - ) -> BrokerageProductPayoutEvent: - """This should be called when a BP is paid out money from their - wallet. Typically, this is an ACH payment. This function creates - the PayoutEvent and the Ledger entries. - - :param thl_ledger_manager: - :param product: The BP being paid. Assuming we're paying them out - of the balance of their USD wallet account. - :param amount: We're assuming everything is in USD, and we're - paying out a USD currency account. We could theoretically also - pay, for e.g. a Bitcoin account with a bitcoin transfer, but - this is not supported for now. - :param payout_type: PayoutType. default ACH - :param cashout_method_uuid: The entry in the - accounting_cashoutmethod table that records payment method - details. By default, the generic ACH cashout method (that has - no actual banking details). - - :param ext_ref_id: This is a unique ID for the Supplier Payment. - Typically it'll be from JP Morgan Chase, but may also just be - random if we can retrieve anything - - :param created: - - :param skip_wallet_balance_check: By default, this will fail unless - the BP's wallet actually has the amount requested. - - :param skip_one_per_day_check: Safety mechanism, checks if there - has already been a payout to this wallet in the past 24 hours. - - :return: - """ - - assert isinstance(amount, USDCent), "Must provide a USDCent" - - if created: - # Try to do a quick dupe check first before we create the payout event - pes = self.filter_by( - reference_uuid=product.id, amount=amount, created=created - ) - if len(pes) > 0: - raise ValueError(f"Payout event already exists!: {pes}") - - if created is None: - created = datetime.now(tz=timezone.utc) - - # TODO: Explain why we're doing this. Why is it important to have - # Payout Events when the ledger has everything that should be - # needed. - bp_wallet = thl_ledger_manager.get_account_or_create_bp_wallet(product=product) - - bp_pe: BrokerageProductPayoutEvent = self.create( - debit_account_uuid=bp_wallet.uuid, - payout_type=payout_type, - amount=amount, - ext_ref_id=ext_ref_id, - created=created, - status=PayoutStatus.PENDING, - ) - return self._create_tx_bp_payout_from_payout_event( - thl_ledger_manager=thl_ledger_manager, - bp_pe=bp_pe, - product=product, - amount=amount, - created=created, - skip_one_per_day_check=skip_one_per_day_check, - skip_wallet_balance_check=skip_wallet_balance_check, - ) + # def create_bp_payout_event( + # self, + # thl_ledger_manager: ThlLedgerManager, + # product: Product, + # amount: USDCent, + # payout_type: PayoutType = PayoutType.ACH, + # ext_ref_id: str | None = None, + # created: AwareDatetime | None = None, + # skip_wallet_balance_check: bool = False, + # skip_one_per_day_check: bool = False, + # ) -> BrokerageProductPayoutEvent: + # + # return self._create_tx_bp_payout_from_payout_event( + # thl_ledger_manager=thl_ledger_manager, + # bp_pe=bp_pe, + # product=product, + # amount=amount, + # created=created, + # skip_one_per_day_check=skip_one_per_day_check, + # skip_wallet_balance_check=skip_wallet_balance_check, + # ) def _create_tx_bp_payout_from_payout_event( self, thl_ledger_manager: ThlLedgerManager, bp_pe: BrokerageProductPayoutEvent, product: Product, - amount: USDCent, created: AwareDatetime | None = None, skip_wallet_balance_check: bool = False, skip_one_per_day_check: bool = False, @@ -824,22 +708,22 @@ class BrokerageProductPayoutEventManager(PayoutEventManager): Handles exceptions: Check if the ledger tx actually exists or not, and set the payout event status accordingly. """ + created = created if created else bp_pe.created try: thl_ledger_manager.create_tx_bp_payout( product=product, - amount=amount, + amount=USDCent(bp_pe.amount), payoutevent_uuid=bp_pe.uuid, created=created, skip_wallet_balance_check=skip_wallet_balance_check, skip_one_per_day_check=skip_one_per_day_check, ) - except Exception as e: e.pe_uuid = bp_pe.uuid if self.check_for_ledger_tx( thl_ledger_manager=thl_ledger_manager, product_id=product.uuid, - amount=amount, + amount=USDCent(bp_pe.amount), payout_event=bp_pe, ): LOG.warning(f"Got exception {e} but ledger tx exists! Continuing ... ") @@ -954,37 +838,46 @@ class BusinessPayoutEventManager(BrokerageProductPayoutEventManager): with self.pg_config.make_connection() as conn: with conn.cursor() as c: # DELETE: tx_entry - c.execute(""" + c.execute( + """ DELETE FROM ledger_entry WHERE transaction_id = ANY(%s) AND id = ANY(%s) - """, [transaction_ids, tx_entry_ids]) + """, + [transaction_ids, tx_entry_ids], + ) # DELETE: tx_metadata - c.execute(""" + c.execute( + """ DELETE FROM ledger_transactionmetadata WHERE transaction_id = ANY(%s) AND id = ANY(%s) - """, [transaction_ids, list(tx_metadata_ids)], + """, + [transaction_ids, list(tx_metadata_ids)], ) # DELETE: transactions - c.execute(""" + c.execute( + """ DELETE FROM ledger_transaction WHERE id = ANY(%s) - """, [transaction_ids]) + """, + [transaction_ids], + ) # DELETE: event_payouts c.execute( - """ + """ DELETE FROM event_payout WHERE ext_ref_id = %s AND uuid = ANY(%s) - """, [ext_ref_id, event_payout_uuids], + """, + [ext_ref_id, event_payout_uuids], ) conn.commit() @@ -1147,30 +1040,62 @@ class BusinessPayoutEventManager(BrokerageProductPayoutEventManager): return allocation - def create_business_payout_event(self, bpe: BusinessPayoutEvent): + def create_business_payout_event( + self, + bpe: BusinessPayoutEvent, + ): + assert bpe.bp_payouts, "Must provide at least one BP Payout" + assert {bp_pe.status for bp_pe in bpe.bp_payouts} == { + PayoutStatus.PENDING + }, "All BP Payouts must be PENDING" + assert bpe.id is None, "Cannot create a BusinessPayoutEvent with an existing ID" with self.pg_config.make_connection() as conn: with conn.cursor() as c: - # Quick check to make sure it doesn't already exist - c.execute(""" - SELECT 1 - FROM supplier_payout - WHERE ext_ref_id = %(ext_ref_id)s - """, {'ext_ref_id': bpe.ext_ref_id}) - assert c.fetchone() is None - - c.execute(""" - """) - + # ext_ref_id has a unique constraint, so we don't need to even + # do an existence check first + c.execute( + """ + INSERT INTO supplier_payout ( + business_id, created, amount, + status, ext_ref_id, payout_type, + request_data, order_data + ) VALUES ( + %(business_id)s, %(created)s, %(amount)s, + %(status)s, %(ext_ref_id)s, %(payout_type)s, + %(request_data)s, %(order_data)s + ) RETURNING id; + """, + bpe.model_dump_postgres(), + ) + supplier_payout_pk = c.fetchone()["id"] + bpe.id = supplier_payout_pk + for bp_pe in bpe.bp_payouts: + c.execute( + """ + INSERT INTO event_payout ( + uuid, debit_account_uuid, created, cashout_method_uuid, + amount, status, ext_ref_id, payout_type, order_data, + request_data, supplier_payout_id + ) VALUES ( + %(uuid)s, %(debit_account_uuid)s, %(created)s, %(cashout_method_uuid)s, + %(amount)s, %(status)s, %(ext_ref_id)s, %(payout_type)s, %(order_data)s, + %(request_data)s, %(supplier_payout_id)s + ); + """, + bp_pe.model_dump_postgres() + | {"supplier_payout_id": supplier_payout_pk}, + ) + conn.commit() def create_from_ach_or_wire( self, business: Business, amount: USDCent, + transaction_id: str, pm: ProductManager, thl_lm: ThlLedgerManager, created: datetime | None = None, - transaction_id: str | None = None, ) -> BusinessPayoutEvent | None: """This records a single banking transfer to a supplier. Takes a specific Business that was paid out and how much. It then determines @@ -1192,7 +1117,7 @@ class BusinessPayoutEventManager(BrokerageProductPayoutEventManager): "the required Brokerage Product amounts." ) - assert amount > 100_00, "Must issue Supplier Payouts at least $100 minimum." + assert amount >= 100_00, "Must issue Supplier Payouts at least $100 minimum." LOG.warning(f"Paying out {business.name} {amount.to_usd_str()}") if created: @@ -1201,6 +1126,8 @@ class BusinessPayoutEventManager(BrokerageProductPayoutEventManager): assert created < datetime.now( tz=timezone.utc ), "created must be in the past" + else: + created = datetime.now(tz=timezone.utc) # Gather the total amount available balance from each and put into # a simple DF. We're using the available balance because we need it @@ -1240,38 +1167,74 @@ class BusinessPayoutEventManager(BrokerageProductPayoutEventManager): products = pm.get_by_uuids(product_uuids=list(amounts.keys())) product_lookup = {p.uuid: p for p in products} + # Bulk version of this ---v + # bp_wallet = thl_lm.get_account_or_create_bp_wallet(product=product) + qualified_names = [ + f"{thl_lm.currency.value}:bp_wallet:{bp.id}" for bp in products + ] + bp_wallets = thl_lm.get_accounts(qualified_names) + wallet_lookup = {bpw.reference_uuid: bpw.uuid for bpw in bp_wallets} + bpe = BusinessPayoutEvent( - uuid=uuid4().hex, + id=None, business_id=business.uuid, payout_type=PayoutType.ACH, amount=amount, created=created, ext_ref_id=transaction_id, # The ACH payment was sent! We haven't yet recorded it - # in the ledger, but it was sent by the bank. - status=PayoutStatus.COMPLETE, + # in the ledger, but it was sent by the bank. This + # is kind of ambiguous the meaning, we'll say it + # is not yet COMPLETE b/c the bp payouts + # haven't all been created yet. + status=PayoutStatus.APPROVED, ) bp_payouts: list[BrokerageProductPayoutEvent] = [] for product_id, item in amounts.items(): product = product_lookup[product_id] - bp_payouts.append(BrokerageProductPayoutEvent( - created=created, - payout_type=PayoutType.ACH, - status=PayoutStatus.PENDING, - uuid=uuid4().hex, - amount=USDCent(item["issue_amount"]), - ext_ref_id=transaction_id, - product_id=product.uuid, - # We will fill these in - debit_account_uuid=None, - cashout_method_uuid=None, - )) + bp_payouts.append( + BrokerageProductPayoutEvent( + created=created, + payout_type=PayoutType.ACH, + status=PayoutStatus.PENDING, + uuid=uuid4().hex, + amount=USDCent(item["issue_amount"]), + ext_ref_id=transaction_id, + product_id=product.uuid, + cashout_method_uuid=self.CASHOUT_METHOD_UUID, + debit_account_uuid=wallet_lookup[product_id], + ) + ) bpe.bp_payouts = bp_payouts + # The supplier_payout db row and all event_payout (BP rows) are all + # created in the same DB transaction. + self.create_business_payout_event(bpe=bpe) + assert bpe.id is not None, "Something failed creating BusinessPayoutEvent" + + # Now, go through each and create ledger txs. This is resumable + # from the BrokerageProductPayoutEvents + for bp_pe in bpe.bp_payouts: + product = product_lookup[bp_pe.product_id] + self._create_tx_bp_payout_from_payout_event( + thl_ledger_manager=thl_lm, + bp_pe=bp_pe, + product=product, + skip_one_per_day_check=True, + skip_wallet_balance_check=True, + ) + with self.connection() as conn: + with conn.cursor() as c: + c.execute( + """ + UPDATE supplier_payout + SET status = %(status)s + WHERE id = %(pk)s""", + {"pk": bpe.id, "status": PayoutStatus.COMPLETE}, + ) - - return BusinessPayoutEvent.model_validate({"bp_payouts": bp_payouts}) + return bpe # import duckdb diff --git a/generalresearch/models/thl/payout.py b/generalresearch/models/thl/payout.py index 5f51d01..3c89e3b 100644 --- a/generalresearch/models/thl/payout.py +++ b/generalresearch/models/thl/payout.py @@ -14,6 +14,7 @@ from pydantic import ( model_validator, ConfigDict, ) +from pydantic.json_schema import SkipJsonSchema from typing_extensions import Self from generalresearch.currency import USDCent @@ -54,15 +55,15 @@ class PayoutEvent(BaseModel): examples=["18298cb1583846fbb06e4747b5310693"], ) - cashout_method_uuid: UUIDStr | None= Field( + cashout_method_uuid: UUIDStr | None = Field( description="References a row in the account_cashoutmethod table. This " "is the cashout method that was used to request this " "payout. (A cashout is the same thing as a payout)", examples=["a6dc1fc1bf934557b952f253dee12813"], ) - created: AwareDatetimeISO | None = Field( - default = None + created: AwareDatetimeISO = Field( + default_factory=lambda: datetime.now(tz=timezone.utc) ) # In the smallest unit of the currency being transacted. For USD, this @@ -86,8 +87,8 @@ class PayoutEvent(BaseModel): description=PayoutType.as_openapi(), examples=[PayoutType.ACH] ) - request_data: dict = Field( - default_factory=dict, + request_data: dict | None = Field( + default=None, description="Stores payout-type-specific information that is used to " "request this payout from the external provider.", ) @@ -146,16 +147,15 @@ class PayoutEvent(BaseModel): # --- ORM --- - def model_dump_mysql(self, *args, **kwargs) -> dict: - d = self.model_dump(mode="json", *args, **kwargs) - - if "created" in d: - d["created"] = self.created.replace(tzinfo=None) + def model_dump_postgres(self) -> dict: + d = self.model_dump(mode="json", exclude={"request_data", "order_data"}) - if d.get("request_data") is not None: - d["request_data"] = json.dumps(self.request_data) - - if d.get("order_data") is not None: + d["request_data"] = ( + json.dumps(self.request_data) if self.request_data is not None else None + ) + if self.order_data is None: + d["order_data"] = None + else: if isinstance(self.order_data, dict): d["order_data"] = json.dumps(self.order_data) else: @@ -279,12 +279,10 @@ class BrokerageProductPayoutEvent(PayoutEvent): class BusinessPayoutEvent(BaseModel): """A single payout event to a supplier Business.""" + model_config = ConfigDict(validate_assignment=True) - uuid: UUIDStr = Field( - title="Supplier Payout Unique Identifier", - examples=["9453cd076713426cb68d05591c7145aa"], - ) + id: SkipJsonSchema[PositiveInt | None] = Field(exclude=True) # Used for holding a *unique*, external, payout-type-specific identifier. ext_ref_id: str = Field(title="Unique external reference ID") @@ -294,8 +292,8 @@ class BusinessPayoutEvent(BaseModel): examples=[uuid4().hex], ) - created: AwareDatetimeISO | None = Field( - default=None + created: AwareDatetimeISO = Field( + default_factory=lambda: datetime.now(tz=timezone.utc) ) # In the smallest unit of the currency being transacted. For USD, this @@ -393,3 +391,16 @@ class BusinessPayoutEvent(BaseModel): ) return self + + def model_dump_postgres(self): + d = self.model_dump( + mode="json", + exclude={"bp_payouts"}, + ) + d["request_data"] = ( + json.dumps(self.request_data) if self.request_data is not None else None + ) + d["order_data"] = ( + json.dumps(self.order_data) if self.order_data is not None else None + ) + return d diff --git a/generalresearch/thl_django/event/models.py b/generalresearch/thl_django/event/models.py index 7144bc4..6d92b20 100644 --- a/generalresearch/thl_django/event/models.py +++ b/generalresearch/thl_django/event/models.py @@ -51,7 +51,11 @@ class SupplierPayout(models.Model): to the Business. """ - uuid = models.UUIDField(default=uuid.uuid4, primary_key=True) + id = models.BigAutoField(primary_key=True) + + # Used for holding a unique, external, payouttype-specific identifier. + # For ACH, this is the ACH transaction id. + ext_ref_id = models.CharField(max_length=64, unique=True) # The Business receiving this payout business_id = models.UUIDField(null=True) @@ -64,10 +68,6 @@ class SupplierPayout(models.Model): # generalresearch/models/thl/payout.py:PayoutStatus status = models.CharField(max_length=20, null=True) - # Used for holding a unique, external, payouttype-specific identifier. - # For ACH, this is the ACH transaction id. - ext_ref_id = models.CharField(max_length=64, unique=True) - # The allowed values for `payout_type` are defined in generalresearch: # generalresearch/models/thl/payout.py:PayoutType payout_type = models.CharField(max_length=14) @@ -132,7 +132,6 @@ class Payout(models.Model): # payout transaction that this product-level split belongs to. supplier_payout = models.ForeignKey( SupplierPayout, - db_column="supplier_payout_uuid", null=True, on_delete=models.DO_NOTHING, ) @@ -144,5 +143,4 @@ class Payout(models.Model): models.Index(fields=["created"]), models.Index(fields=["debit_account_uuid"]), models.Index(fields=["ext_ref_id"]), - models.Index(fields=["supplier_payout"]), ] |