refactor: credit ratio -> price & credit history filters

amalcaraz · amalcaraz · commit fb7235ccc748 · 2025-12-02T10:56:23.000+01:00
diff --git a/deployment/migrations/versions/0040_e1f2a3b4c5d6_rename_credit_history_ratio_to_price.py b/deployment/migrations/versions/0040_e1f2a3b4c5d6_rename_credit_history_ratio_to_price.py
@@ -0,0 +1,80 @@
+"""rename credit_history ratio column to price
+
+Revision ID: e1f2a3b4c5d6
+Revises: d0e1f2a3b4c5
+Create Date: 2025-12-01 00:00:00.000000
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = 'e1f2a3b4c5d6'
+down_revision = 'd0e1f2a3b4c5'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # First, add the new price column
+    op.add_column('credit_history', sa.Column('price', sa.DECIMAL(), nullable=True))
+
+    # Add the new bonus_amount column
+    op.add_column('credit_history', sa.Column('bonus_amount', sa.BigInteger(), nullable=True))
+
+    # Transform data: price calculation depends on payment token
+    # For ALEPH token: price = 1 / (ratio * 0.8)
+    # For other tokens: price = 1/ratio
+    # Only update rows where payment_method is NOT 'credit_expense' or 'credit_transfer'
+    # and where ratio is not null and not zero
+    connection = op.get_bind()
+    connection.execute(sa.text("""
+        UPDATE credit_history
+        SET price = CASE
+            WHEN token = 'ALEPH' THEN ROUND(1.0 / (ratio * 0.8), 18)
+            ELSE ROUND(1.0 / ratio, 18)
+        END
+        WHERE ratio IS NOT NULL
+        AND ratio != 0
+        AND (payment_method IS NULL
+             OR (payment_method != 'credit_expense' AND payment_method != 'credit_transfer'))
+    """))
+
+    # Update bonus_amount for ALEPH token records
+    connection.execute(sa.text("""
+        UPDATE credit_history
+        SET bonus_amount = TRUNC(amount * 0.20)
+        WHERE token = 'ALEPH'
+        AND amount IS NOT NULL
+    """))
+
+    # Drop the old ratio column
+    op.drop_column('credit_history', 'ratio')
+
+
+def downgrade() -> None:
+    # Add back the ratio column
+    op.add_column('credit_history', sa.Column('ratio', sa.DECIMAL(), nullable=True))
+
+    # Transform data back: reverse price calculation depends on payment token
+    # For ALEPH token: ratio = (1/price) / 0.8
+    # For other tokens: ratio = 1/price
+    connection = op.get_bind()
+    connection.execute(sa.text("""
+        UPDATE credit_history
+        SET ratio = CASE
+            WHEN token = 'ALEPH' THEN ROUND((1.0 / price) / 0.8, 18)
+            ELSE ROUND(1.0 / price, 18)
+        END
+        WHERE price IS NOT NULL
+        AND price != 0
+        AND (payment_method IS NULL
+             OR (payment_method != 'credit_expense' AND payment_method != 'credit_transfer'))
+    """))
+
+    # Drop the price column
+    op.drop_column('credit_history', 'price')
+
+    # Drop the bonus_amount column
+    op.drop_column('credit_history', 'bonus_amount')
diff --git a/src/aleph/db/accessors/balances.py b/src/aleph/db/accessors/balances.py
@@ -408,8 +408,8 @@ def _bulk_insert_credit_history(
     cursor = conn.cursor()
 
     # Column specification for credit history
-    # Include the new message_timestamp field
-    copy_columns = "address, amount, credit_ref, credit_index, message_timestamp, last_update, ratio, tx_hash, expiration_date, token, chain, origin, provider, origin_ref, payment_method"
+    # Include the new message_timestamp and bonus_amount fields
+    copy_columns = "address, amount, credit_ref, credit_index, message_timestamp, last_update, price, bonus_amount, tx_hash, expiration_date, token, chain, origin, provider, origin_ref, payment_method"
 
     csv_credit_history = StringIO("\n".join(csv_rows))
     cursor.copy_expert(
@@ -453,7 +453,7 @@ def update_credit_balances_distribution(
     """
     Updates credit balances for distribution messages (aleph_credit_distribution).
 
-    Distribution messages include all fields like ratio, tx_hash, provider,
+    Distribution messages include all fields like price, tx_hash, provider,
     payment_method, token, chain, and expiration_date.
     """
 
@@ -463,7 +463,7 @@ def update_credit_balances_distribution(
     for index, credit_entry in enumerate(credits_list):
         address = credit_entry["address"]
         amount = abs(int(credit_entry["amount"]))
-        ratio = Decimal(credit_entry["ratio"])
+        price = Decimal(credit_entry["price"])
         tx_hash = credit_entry["tx_hash"]
         provider = credit_entry["provider"]
 
@@ -472,6 +472,7 @@ def update_credit_balances_distribution(
         origin = credit_entry.get("origin", "")
         origin_ref = credit_entry.get("ref", "")
         payment_method = credit_entry.get("payment_method", "")
+        bonus_amount = credit_entry.get("bonus_amount", "")
 
         # Convert expiration timestamp to datetime
 
@@ -482,7 +483,7 @@ def update_credit_balances_distribution(
         )
 
         csv_rows.append(
-            f"{address};{amount};{message_hash};{index};{message_timestamp};{last_update};{ratio};{tx_hash};{expiration_date or ''};{token};{chain};{origin};{provider};{origin_ref};{payment_method}"
+            f"{address};{amount};{message_hash};{index};{message_timestamp};{last_update};{price};{bonus_amount or ''};{tx_hash};{expiration_date or ''};{token};{chain};{origin};{provider};{origin_ref};{payment_method}"
         )
 
     _bulk_insert_credit_history(session, csv_rows)
@@ -500,7 +501,7 @@ def update_credit_balances_expense(
     Expense messages have negative amounts and can include:
     - execution_id (mapped to origin)
     - node_id (mapped to tx_hash)
-    - price (mapped to ratio)
+    - price (mapped to price)
     - time (skipped for now)
     - ref (mapped to origin_ref)
     """
@@ -516,11 +517,11 @@ def update_credit_balances_expense(
         # Map new fields
         origin = credit_entry.get("execution_id", "")
         tx_hash = credit_entry.get("node_id", "")
-        ratio = credit_entry.get("price", "")
+        price = credit_entry.get("price", "")
         # Skip time field for now
 
         csv_rows.append(
-            f"{address};{amount};{message_hash};{index};{message_timestamp};{last_update};{ratio};{tx_hash};;;;{origin};ALEPH;{origin_ref};credit_expense"
+            f"{address};{amount};{message_hash};{index};{message_timestamp};{last_update};{price};;{tx_hash};;;;{origin};ALEPH;{origin_ref};credit_expense"
         )
 
     _bulk_insert_credit_history(session, csv_rows)
@@ -562,15 +563,15 @@ def update_credit_balances_transfer(
 
         # Add positive entry for recipient (origin = sender, provider = ALEPH, payment_method = credit_transfer)
         csv_rows.append(
-            f"{recipient_address};{amount};{message_hash};{index};{message_timestamp};{last_update};;;{expiration_date or ''};;;{sender_address};ALEPH;;credit_transfer"
+            f"{recipient_address};{amount};{message_hash};{index};{message_timestamp};{last_update};;;;{expiration_date or ''};;;{sender_address};ALEPH;;credit_transfer"
         )
         index += 1
 
         # Add negative entry for sender (unless sender is in whitelisted addresses)
         # (origin = recipient, provider = ALEPH, payment_method = credit_transfer)
         if sender_address not in whitelisted_addresses:
             csv_rows.append(
-                f"{sender_address};{-amount};{message_hash};{index};{message_timestamp};{last_update};;;;;;{recipient_address};ALEPH;;credit_transfer"
+                f"{sender_address};{-amount};{message_hash};{index};{message_timestamp};{last_update};;;;;;;{recipient_address};ALEPH;;credit_transfer"
             )
             index += 1
 
@@ -602,6 +603,13 @@ def get_address_credit_history(
     address: str,
     page: int = 1,
     pagination: int = 0,
+    tx_hash: Optional[str] = None,
+    token: Optional[str] = None,
+    chain: Optional[str] = None,
+    provider: Optional[str] = None,
+    origin: Optional[str] = None,
+    origin_ref: Optional[str] = None,
+    payment_method: Optional[str] = None,
 ) -> Sequence[AlephCreditHistoryDb]:
     """
     Get paginated credit history entries for a specific address, ordered from newest to oldest.
@@ -611,6 +619,13 @@ def get_address_credit_history(
         address: Address to get credit history for
         page: Page number (starts at 1)
         pagination: Number of entries per page (0 for all entries)
+        tx_hash: Filter by transaction hash
+        token: Filter by token
+        chain: Filter by chain
+        provider: Filter by provider
+        origin: Filter by origin
+        origin_ref: Filter by origin reference
+        payment_method: Filter by payment method
 
     Returns:
         List of credit history entries ordered by message_timestamp desc
@@ -621,6 +636,22 @@ def get_address_credit_history(
         .order_by(AlephCreditHistoryDb.message_timestamp.desc())
     )
 
+    # Apply filters
+    if tx_hash is not None:
+        query = query.where(AlephCreditHistoryDb.tx_hash == tx_hash)
+    if token is not None:
+        query = query.where(AlephCreditHistoryDb.token == token)
+    if chain is not None:
+        query = query.where(AlephCreditHistoryDb.chain == chain)
+    if provider is not None:
+        query = query.where(AlephCreditHistoryDb.provider == provider)
+    if origin is not None:
+        query = query.where(AlephCreditHistoryDb.origin == origin)
+    if origin_ref is not None:
+        query = query.where(AlephCreditHistoryDb.origin_ref == origin_ref)
+    if payment_method is not None:
+        query = query.where(AlephCreditHistoryDb.payment_method == payment_method)
+
     if pagination > 0:
         query = query.offset((page - 1) * pagination).limit(pagination)
 
@@ -630,21 +661,51 @@ def get_address_credit_history(
 def count_address_credit_history(
     session: DbSession,
     address: str,
+    tx_hash: Optional[str] = None,
+    token: Optional[str] = None,
+    chain: Optional[str] = None,
+    provider: Optional[str] = None,
+    origin: Optional[str] = None,
+    origin_ref: Optional[str] = None,
+    payment_method: Optional[str] = None,
 ) -> int:
     """
-    Count total credit history entries for a specific address.
+    Count total credit history entries for a specific address with optional filters.
 
     Args:
         session: Database session
         address: Address to count credit history for
+        tx_hash: Filter by transaction hash
+        token: Filter by token
+        chain: Filter by chain
+        provider: Filter by provider
+        origin: Filter by origin
+        origin_ref: Filter by origin reference
+        payment_method: Filter by payment method
 
     Returns:
-        Total number of credit history entries for the address
+        Total number of credit history entries for the address matching the filters
     """
     query = select(func.count(AlephCreditHistoryDb.credit_ref)).where(
         AlephCreditHistoryDb.address == address
     )
 
+    # Apply filters
+    if tx_hash is not None:
+        query = query.where(AlephCreditHistoryDb.tx_hash == tx_hash)
+    if token is not None:
+        query = query.where(AlephCreditHistoryDb.token == token)
+    if chain is not None:
+        query = query.where(AlephCreditHistoryDb.chain == chain)
+    if provider is not None:
+        query = query.where(AlephCreditHistoryDb.provider == provider)
+    if origin is not None:
+        query = query.where(AlephCreditHistoryDb.origin == origin)
+    if origin_ref is not None:
+        query = query.where(AlephCreditHistoryDb.origin_ref == origin_ref)
+    if payment_method is not None:
+        query = query.where(AlephCreditHistoryDb.payment_method == payment_method)
+
     return session.execute(query).scalar_one()
 
 
diff --git a/src/aleph/db/models/balances.py b/src/aleph/db/models/balances.py
@@ -49,7 +49,8 @@ class AlephCreditHistoryDb(Base):
 
     address: str = Column(String, nullable=False, index=True)
     amount: int = Column(BigInteger, nullable=False)
-    ratio: Optional[Decimal] = Column(DECIMAL, nullable=True)
+    price: Optional[Decimal] = Column(DECIMAL, nullable=True)
+    bonus_amount: Optional[int] = Column(BigInteger, nullable=True)
     tx_hash: Optional[str] = Column(String, nullable=True)
     token: Optional[str] = Column(String, nullable=True)
     chain: Optional[str] = Column(String, nullable=True)
diff --git a/src/aleph/schemas/api/accounts.py b/src/aleph/schemas/api/accounts.py
@@ -123,13 +123,34 @@ class GetAccountCreditHistoryQueryParams(BaseModel):
     page: int = Field(
         default=DEFAULT_PAGE, ge=1, description="Offset in pages. Starts at 1."
     )
+    tx_hash: Optional[str] = Field(
+        default=None, description="Filter by transaction hash"
+    )
+    token: Optional[str] = Field(
+        default=None, description="Filter by token"
+    )
+    chain: Optional[str] = Field(
+        default=None, description="Filter by chain"
+    )
+    provider: Optional[str] = Field(
+        default=None, description="Filter by provider"
+    )
+    origin: Optional[str] = Field(
+        default=None, description="Filter by origin"
+    )
+    origin_ref: Optional[str] = Field(
+        default=None, description="Filter by origin reference"
+    )
+    payment_method: Optional[str] = Field(
+        default=None, description="Filter by payment method"
+    )
 
 
 class CreditHistoryResponseItem(BaseModel):
     model_config = ConfigDict(from_attributes=True)
 
     amount: int
-    ratio: Optional[Decimal] = None
+    price: Optional[Decimal] = None
     tx_hash: Optional[str] = None
     token: Optional[str] = None
     chain: Optional[str] = None
diff --git a/src/aleph/web/controllers/accounts.py b/src/aleph/web/controllers/accounts.py
@@ -234,19 +234,36 @@ async def get_account_credit_history(request: web.Request) -> web.Response:
             address=address,
             page=query_params.page,
             pagination=query_params.pagination,
+            tx_hash=query_params.tx_hash,
+            token=query_params.token,
+            chain=query_params.chain,
+            provider=query_params.provider,
+            origin=query_params.origin,
+            origin_ref=query_params.origin_ref,
+            payment_method=query_params.payment_method,
         )
 
         if not credit_history_entries:
             raise web.HTTPNotFound(text="No credit history found for this address")
 
-        total_entries = count_address_credit_history(session=session, address=address)
+        total_entries = count_address_credit_history(
+            session=session,
+            address=address,
+            tx_hash=query_params.tx_hash,
+            token=query_params.token,
+            chain=query_params.chain,
+            provider=query_params.provider,
+            origin=query_params.origin,
+            origin_ref=query_params.origin_ref,
+            payment_method=query_params.payment_method,
+        )
 
         # Convert to response items
         history_adapter = TypeAdapter(list[CreditHistoryResponseItem])
         credit_history_list = [
             {
                 "amount": entry.amount,
-                "ratio": entry.ratio,
+                "price": entry.price,
                 "tx_hash": entry.tx_hash,
                 "token": entry.token,
                 "chain": entry.chain,
diff --git a/tests/db/test_credit_balances.py b/tests/db/test_credit_balances.py
